simplified SemaphoreGroup by switching a semaphore array to a semaphore list.

added some copy methods to Array
This commit is contained in:
hyunghwan.chung 2017-10-08 15:40:32 +00:00
parent 1fd9ce285c
commit 32a77bc6db
6 changed files with 206 additions and 98 deletions

View File

@ -64,17 +64,54 @@ class(#pointer) Array(Collection)
method last method last
{ {
^self at: (self basicSize - 1). ^self at: (self size - 1).
} }
method do: aBlock method do: aBlock
{ {
0 priorTo: (self basicSize) do: [:i | aBlock value: (self at: i)]. 0 priorTo: (self size) do: [:i | aBlock value: (self at: i)].
} }
method copy: anArray method copy: anArray
{ {
0 priorTo: (anArray basicSize) do: [:i | self at: i put: (anArray at: i) ]. 0 priorTo: (anArray size) do: [:i | self at: i put: (anArray at: i) ].
}
method copy: anArray from: start to: end
{
## copy elements from an array 'anArray' starting from
## the index 'start' to the index 'end'.
| s i ss |
(*
s := anArray size.
if (start < 0) { start := 0 }
elsif (start >= s) { start := s - 1 }.
if (end < 0) { end := 0 }
elsif (end >= s) { end := s - 1 }.
*)
i := 0.
ss := self size.
while (start <= end)
{
if (i >= ss) { break }.
self at: i put: (anArray at: start).
i := i + 1.
start := start + 1.
}.
}
method copyFrom: start to: end
{
## returns a copy of the receiver starting from the element
## at index 'start' to the element at index 'end'.
| newsz |
newsz := end - start + 1.
^(self class new: newsz) copy: self from: start to: end
} }
method = anArray method = anArray

View File

@ -60,8 +60,11 @@ class Semaphore(Object)
ioHandle := nil, ioHandle := nil,
ioMask := 0. ioMask := 0.
var (#get,#set) signalAction := nil. var(#get,#set) signalAction := nil.
var(#get,#set) _group := nil.
var(#get,#set) _group := nil,
_grm_next := nil,
_grm_prev := nil.
## ================================================================== ## ==================================================================
@ -198,69 +201,30 @@ class SemaphoreGroup(Object)
{ {
var waiting_head := nil, var waiting_head := nil,
waiting_tail := nil, waiting_tail := nil,
size := 0, first_sem := nil,
pos := 0, last_sem := nil,
semarr := nil. first_sigsem := nil,
last_sigsem := nil.
(* TODO: good idea to a shortcut way to prohibit a certain method in the heirarchy chain? (* TODO: good idea to a shortcut way to prohibit a certain method in the heirarchy chain?
method(#class,#prohibited) new. method(#class,#prohibited) new.
method(#class,#prohibited) new: size. method(#class,#prohibited) new: size.
method(#class,#abstract) xxx. => method(#class) xxx { self subclassResponsibility: #xxxx } method(#class,#abstract) xxx. => method(#class) xxx { self subclassResponsibility: #xxxx }
*) *)
(*
method(#class) new { self messageProhibited: #new } method(#class) new { self messageProhibited: #new }
method(#class) new: size { self messageProhibited: #new: } method(#class) new: size { self messageProhibited: #new: }
*)
method(#class,#variadic) with() method(#primitive) _addSemaphore: sem.
{ method(#primitive) _removeSemaphore: sem.
| i x arr sem |
i := 0.
x := thisContext vargCount.
arr := Array new: x.
while (i < x)
{
sem := thisContext vargAt: i.
if (sem _group notNil)
{
System.Exception signal: 'Cannot add a semaphore in a group to another group'
}.
arr at: i put: sem.
i := i + 1.
}.
^self basicNew initialize: arr.
}
method initialize
{
self.semarr := Array new: 10.
}
method initialize: arr
{
| i sem |
self.size := arr size.
self.semarr := arr.
i := 0.
while (i < self.size)
{
sem := self.semarr at: i.
sem _group: self.
i := i + 1.
}
}
method(#primitive) _wait. method(#primitive) _wait.
method wait method wait
{ {
| r | | r |
r := self wait. r := self _wait.
if (r signalAction notNil) { r signalAction value: r }. if (r signalAction notNil) { r signalAction value: r }.
^r ^r
} }
@ -274,7 +238,8 @@ method(#class,#abstract) xxx. => method(#class) xxx { self subclassResponsibilit
## grant the partial membership to the internal semaphore. ## grant the partial membership to the internal semaphore.
## it's partial because it's not added to self.semarr. ## it's partial because it's not added to self.semarr.
s _group: self. ##s _group: self.
self _addSemaphore: s.
## arrange the processor to notify upon timeout. ## arrange the processor to notify upon timeout.
Processor signal: s after: seconds. Processor signal: s after: seconds.
@ -288,7 +253,8 @@ method(#class,#abstract) xxx. => method(#class) xxx { self subclassResponsibilit
elsif (r signalAction notNil) { r signalAction value: r }. elsif (r signalAction notNil) { r signalAction value: r }.
## nullify the membership ## nullify the membership
s _group: nil. ##s _group: nil.
self _removeSemaphore: s.
## cancel the notification arrangement in case it didn't time out. ## cancel the notification arrangement in case it didn't time out.
Processor unsignal: s. Processor unsignal: s.

View File

@ -5531,22 +5531,35 @@ static int compile_while_expression (moo_t* moo) /* or compile_until_expression
postcondpos = moo->c->mth.code.len; postcondpos = moo->c->mth.code.len;
if (precondpos + 1 == postcondpos) if (precondpos + 1 == postcondpos)
{ {
moo_uint8_t inst1, inst2;
if (is_until_loop)
{
inst1 = BCODE_PUSH_FALSE;
inst2 = BCODE_PUSH_TRUE;
}
else
{
inst1 = BCODE_PUSH_TRUE;
inst2 = BCODE_PUSH_FALSE;
}
/* simple optimization - /* simple optimization -
* if the conditional is known to be true, emit the absolute jump instruction. * if the conditional is known to be true, emit the absolute jump instruction.
* if it is known to be false, kill all generated instructions. */ * if it is known to be false, kill all generated instructions. */
if (moo->c->mth.code.ptr[precondpos] == (is_until_loop? BCODE_PUSH_FALSE: BCODE_PUSH_TRUE)) if (moo->c->mth.code.ptr[precondpos] == inst1)
{ {
/* the conditional is always true for while, or false for until*/ /* the conditional is always true for while, or false for until*/
cond_style = 1; cond_style = 1;
eliminate_instructions (moo, precondpos, moo->c->mth.code.len - 1); eliminate_instructions (moo, precondpos, moo->c->mth.code.len - 1);
postcondpos = precondpos; postcondpos = precondpos;
} }
else if (moo->c->mth.code.ptr[precondpos] == (is_until_loop? BCODE_PUSH_TRUE: BCODE_PUSH_FALSE)) else if (moo->c->mth.code.ptr[precondpos] == inst2)
{ {
/* the conditional is always false for while, or false for until */ /* the conditional is always false for while, or false for until */
cond_style = -1; cond_style = -1;
} }
/* TODO: at least check for some literals for optimization. /* TODO: at least check some other literals for optimization.
* most literal values must be evaluate to true. */ * most literal values must be evaluate to true. */
} }

View File

@ -737,13 +737,12 @@ static moo_oop_process_t signal_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
{ {
moo_oop_process_t proc; moo_oop_process_t proc;
moo_ooi_t count; moo_ooi_t count;
if ((moo_oop_t)sem->group != moo->_nil)
{
/* the semaphore belongs to a semaphore group */
moo_oop_semaphore_group_t semgrp; moo_oop_semaphore_group_t semgrp;
semgrp = sem->group; semgrp = sem->group;
if ((moo_oop_t)semgrp != moo->_nil)
{
/* the semaphore belongs to a semaphore group */
if ((moo_oop_t)semgrp->waiting.first != moo->_nil) if ((moo_oop_t)semgrp->waiting.first != moo->_nil)
{ {
/* there is a process waiting on the process group */ /* there is a process waiting on the process group */
@ -752,7 +751,9 @@ static moo_oop_process_t signal_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
unchain_from_semaphore (moo, proc); unchain_from_semaphore (moo, proc);
resume_process (moo, proc); resume_process (moo, proc);
/* the waiting process has been suspended after a waiting /* [IMPORTANT] RETURN VALUE of SemaphoreGroup's wait.
* ------------------------------------------------------------
* the waiting process has been suspended after a waiting
* primitive function in Semaphore or SemaphoreGroup. * primitive function in Semaphore or SemaphoreGroup.
* the top of the stack of the process must hold the temporary * the top of the stack of the process must hold the temporary
* return value set by await_semaphore() or await_semaphore_group(). * return value set by await_semaphore() or await_semaphore_group().
@ -773,13 +774,20 @@ static moo_oop_process_t signal_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
* *
* TODO: implement a fair scheduling policy. or do i simply have to disallow individual wait on a semaphore belonging to a group? * TODO: implement a fair scheduling policy. or do i simply have to disallow individual wait on a semaphore belonging to a group?
* *
* if it doesn't belong to a sempahore group, i'm free from the ambiguity * if it doesn't belong to a sempahore group, i'm free from the starvation issue.
* issue.
*/ */
if ((moo_oop_t)sem->waiting.first == moo->_nil) if ((moo_oop_t)sem->waiting.first == moo->_nil)
{ {
/* no process is waiting on this semaphore */ /* no process is waiting on this semaphore */
count = MOO_OOP_TO_SMOOI(sem->count); count = MOO_OOP_TO_SMOOI(sem->count);
MOO_ASSERT (moo, count >= 0);
if (count == 0 && (moo_oop_t)semgrp != moo->_nil)
{
/* move the semaphore from the unsignaled list to the signaled list
* if the semaphore count has changed from 0 to 1 and it belongs a group. */
MOO_DELETE_FROM_OOP_LIST (moo, &semgrp->sems, sem, grm);
MOO_APPEND_TO_OOP_LIST (moo, &semgrp->sigsems, moo_oop_semaphore_t, sem, grm);
}
count++; count++;
sem->count = MOO_SMOOI_TO_OOP(count); sem->count = MOO_SMOOI_TO_OOP(count);
@ -806,13 +814,14 @@ static moo_oop_process_t signal_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
static MOO_INLINE void await_semaphore (moo_t* moo, moo_oop_semaphore_t sem) static MOO_INLINE void await_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
{ {
/* TODO: support timeout */
moo_oop_process_t proc; moo_oop_process_t proc;
moo_ooi_t count; moo_ooi_t count;
moo_oop_semaphore_group_t semgrp;
semgrp = sem->group;
#if 0 #if 0
/* TODO: do i have to disallow?? */ /* TODO: do i have to disallow?? */
if ((moo_oop_t)sem->group != moo->_nil) if ((moo_oop_t)semgrp != moo->_nil)
{ {
/* disallow a semaphore in a semaphore group to be waited on */ /* disallow a semaphore in a semaphore group to be waited on */
moo_seterrnum (moo, MOO_EPERM); moo_seterrnum (moo, MOO_EPERM);
@ -826,6 +835,14 @@ static MOO_INLINE void await_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
/* it's already signalled */ /* it's already signalled */
count--; count--;
sem->count = MOO_SMOOI_TO_OOP(count); sem->count = MOO_SMOOI_TO_OOP(count);
if ((moo_oop_t)semgrp != moo->_nil && count == 0)
{
/* TODO: if i disallow individual wait on a semaphore in a group,
* this membership manipulation is redundant */
MOO_DELETE_FROM_OOP_LIST (moo, &semgrp->sigsems, sem, grm);
MOO_APPEND_TO_OOP_LIST (moo, &semgrp->sems, moo_oop_semaphore_t, sem, grm);
}
} }
else else
{ {
@ -850,47 +867,43 @@ static MOO_INLINE void await_semaphore (moo_t* moo, moo_oop_semaphore_t sem)
#endif #endif
} }
static MOO_INLINE moo_oop_t await_semaphore_group (moo_t* moo, moo_oop_semaphore_group_t semgrp, const moo_ntime_t* tmout) static MOO_INLINE moo_oop_t await_semaphore_group (moo_t* moo, moo_oop_semaphore_group_t semgrp)
{ {
/* TODO: support timeout and wait all */ /* TODO: support timeout and wait all */
/* wait for one of semaphores in the group to be signaled */ /* wait for one of semaphores in the group to be signaled */
moo_oop_process_t proc; moo_oop_process_t proc;
moo_oop_semaphore_t sem; moo_oop_semaphore_t sem;
moo_ooi_t numsems, sempos, i, count;
MOO_ASSERT (moo, moo_iskindof(moo, (moo_oop_t)semgrp, moo->_semaphore_group)); MOO_ASSERT (moo, moo_iskindof(moo, (moo_oop_t)semgrp, moo->_semaphore_group));
/* check if there is a signaled semaphore in the group */ sem = semgrp->sigsems.first;
numsems = MOO_OOP_TO_SMOOI(semgrp->size); if ((moo_oop_t)sem != moo->_nil)
sempos = MOO_OOP_TO_SMOOI(semgrp->pos);
for (i = 0; i < numsems; i++)
{ {
sem = (moo_oop_semaphore_t)((moo_oop_oop_t)semgrp->semarr)->slot[sempos]; moo_ooi_t count;
sempos = (sempos + 1) % numsems;
count = MOO_OOP_TO_SMOOI(sem->count); count = MOO_OOP_TO_SMOOI(sem->count);
if (count > 0) MOO_ASSERT (moo, count > 0);
{
count--; count--;
sem->count = MOO_SMOOI_TO_OOP(count); sem->count = MOO_SMOOI_TO_OOP(count);
semgrp->pos = MOO_SMOOI_TO_OOP(sempos); /* position of the last inspected semaphore */
return (moo_oop_t)sem; MOO_DELETE_FROM_OOP_LIST (moo, &semgrp->sigsems, sem, grm);
if (count > 0)
{
/* move the item to the back of signaled semaphore list */
MOO_APPEND_TO_OOP_LIST (moo, &semgrp->sigsems, moo_oop_semaphore_t, sem, grm);
} }
else
{
/* move the semaphore to the unsigned semaphore list */
MOO_APPEND_TO_OOP_LIST (moo, &semgrp->sems, moo_oop_semaphore_t, sem, grm);
}
return (moo_oop_t)sem;
} }
/* no semaphores have been signaled. suspend the current process /* no semaphores have been signaled. suspend the current process
* until the at least one of them is signaled */ * until the at least one of them is signaled */
#if 0
if (tmout)
{
/* create an internal semaphore for timeout signaling */
/* TODO: */
if (add_to_sem_heap (moo, tmout_sem) <= -1) return MOO_PF_HARD_FAILURE;
}
#endif
proc = moo->processor->active; proc = moo->processor->active;
/* suspend the active process */ /* suspend the active process */
@ -2550,6 +2563,72 @@ static moo_pfrc_t pf_semaphore_wait (moo_t* moo, moo_ooi_t nargs)
return MOO_PF_SUCCESS; return MOO_PF_SUCCESS;
} }
static moo_pfrc_t pf_semaphore_group_add_semaphore (moo_t* moo, moo_ooi_t nargs)
{
moo_oop_semaphore_group_t rcv;
moo_oop_semaphore_t sem;
rcv = (moo_oop_semaphore_group_t)MOO_STACK_GETRCV(moo, nargs);
MOO_PF_CHECK_RCV (moo, moo_iskindof(moo, (moo_oop_t)rcv, moo->_semaphore_group));
sem = (moo_oop_semaphore_t)MOO_STACK_GETARG (moo, nargs, 0);
MOO_PF_CHECK_ARGS (moo, nargs, moo_iskindof(moo, (moo_oop_t)sem, moo->_semaphore));
if ((moo_oop_t)sem->group == moo->_nil)
{
if (MOO_OOP_TO_SMOOI(sem->count) > 0)
{
MOO_APPEND_TO_OOP_LIST (moo, &rcv->sigsems, moo_oop_semaphore_t, sem, grm);
}
else
{
MOO_APPEND_TO_OOP_LIST (moo, &rcv->sems, moo_oop_semaphore_t, sem, grm);
}
sem->group = rcv;
MOO_STACK_SETRETTORCV (moo, nargs);
}
else
{
/* the semaphore belongs to a group already */
MOO_STACK_SETRETTOERROR (moo, nargs, MOO_EPERM);
}
return MOO_PF_SUCCESS;
}
static moo_pfrc_t pf_semaphore_group_remove_semaphore (moo_t* moo, moo_ooi_t nargs)
{
moo_oop_semaphore_group_t rcv;
moo_oop_semaphore_t sem;
rcv = (moo_oop_semaphore_group_t)MOO_STACK_GETRCV(moo, nargs);
MOO_PF_CHECK_RCV (moo, moo_iskindof(moo, (moo_oop_t)rcv, moo->_semaphore_group));
sem = (moo_oop_semaphore_t)MOO_STACK_GETARG (moo, nargs, 0);
MOO_PF_CHECK_ARGS (moo, nargs, moo_iskindof(moo, (moo_oop_t)sem, moo->_semaphore));
if ((moo_oop_t)sem->group == moo->_nil)
{
/* it doesn't belong to a group */
MOO_STACK_SETRETTOERROR (moo, nargs, MOO_EPERM);
}
else
{
if (MOO_OOP_TO_SMOOI(sem->count) > 0)
{
MOO_DELETE_FROM_OOP_LIST (moo, &rcv->sigsems, sem, grm);
}
else
{
MOO_DELETE_FROM_OOP_LIST (moo, &rcv->sems, sem, grm);
}
sem->group = (moo_oop_semaphore_group_t)moo->_nil;
MOO_STACK_SETRETTORCV (moo, nargs);
}
return MOO_PF_SUCCESS;
}
static moo_pfrc_t pf_semaphore_group_wait (moo_t* moo, moo_ooi_t nargs) static moo_pfrc_t pf_semaphore_group_wait (moo_t* moo, moo_ooi_t nargs)
{ {
moo_oop_t rcv, sem; moo_oop_t rcv, sem;
@ -2566,7 +2645,7 @@ static moo_pfrc_t pf_semaphore_group_wait (moo_t* moo, moo_ooi_t nargs)
* the stack from this moment on. */ * the stack from this moment on. */
MOO_STACK_SETRETTORCV (moo, nargs); MOO_STACK_SETRETTORCV (moo, nargs);
sem = await_semaphore_group (moo, (moo_oop_semaphore_group_t)rcv, MOO_NULL); sem = await_semaphore_group (moo, (moo_oop_semaphore_group_t)rcv);
if (sem != moo->_nil) if (sem != moo->_nil)
{ {
/* there was a singaled semaphore. the active process won't get /* there was a singaled semaphore. the active process won't get
@ -4244,6 +4323,8 @@ static pf_t pftab[] =
{ "Semaphore_signal", { pf_semaphore_signal, 0, 0 } }, { "Semaphore_signal", { pf_semaphore_signal, 0, 0 } },
{ "Semaphore__wait", { pf_semaphore_wait, 0, 0 } }, { "Semaphore__wait", { pf_semaphore_wait, 0, 0 } },
{ "SemaphoreGroup__addSemaphore:", { pf_semaphore_group_add_semaphore, 1, 1 } },
{ "SemaphoreGroup__removeSemaphore:", { pf_semaphore_group_remove_semaphore, 1, 1 } },
{ "SemaphoreGroup__wait", { pf_semaphore_group_wait, 0, 0 } }, { "SemaphoreGroup__wait", { pf_semaphore_group_wait, 0, 0 } },
{ "SmallInteger_asCharacter", { pf_smooi_as_character, 0, 0 } }, { "SmallInteger_asCharacter", { pf_smooi_as_character, 0, 0 } },

View File

@ -1771,7 +1771,7 @@ static void vm_muxwait (moo_t* moo, const moo_ntime_t* dur, moo_vmprim_muxwait_c
#elif defined(USE_SELECT) #elif defined(USE_SELECT)
revents = xtn->ev.buf[n].events; revents = xtn->ev.buf[n].events;
#else #else
revents = 0; /* TODO: fake. unsupported */ revents = 0; /* TODO: fake. unsupported but to compile on such an unsupported system.*/
#endif #endif
mask = 0; mask = 0;
@ -1790,8 +1790,6 @@ static void vm_muxwait (moo_t* moo, const moo_ntime_t* dur, moo_vmprim_muxwait_c
muxwcb (moo, mask, (void*)xtn->epd.ptr[xtn->ev.buf[n].fd]); muxwcb (moo, mask, (void*)xtn->epd.ptr[xtn->ev.buf[n].fd]);
#elif defined(USE_SELECT) #elif defined(USE_SELECT)
muxwcb (moo, mask, (void*)xtn->epd.data[xtn->ev.buf[n].fd]); muxwcb (moo, mask, (void*)xtn->epd.data[xtn->ev.buf[n].fd]);
#else
# error UNSUPPORTED
#endif #endif
} }

View File

@ -748,11 +748,11 @@ struct moo_context_t
typedef struct moo_process_t moo_process_t; typedef struct moo_process_t moo_process_t;
typedef struct moo_process_t* moo_oop_process_t; typedef struct moo_process_t* moo_oop_process_t;
#define MOO_SEMAPHORE_NAMED_INSTVARS 11 #define MOO_SEMAPHORE_NAMED_INSTVARS 13
typedef struct moo_semaphore_t moo_semaphore_t; typedef struct moo_semaphore_t moo_semaphore_t;
typedef struct moo_semaphore_t* moo_oop_semaphore_t; typedef struct moo_semaphore_t* moo_oop_semaphore_t;
#define MOO_SEMAPHORE_GROUP_NAMED_INSTVARS 5 #define MOO_SEMAPHORE_GROUP_NAMED_INSTVARS 6
typedef struct moo_semaphore_group_t moo_semaphore_group_t; typedef struct moo_semaphore_group_t moo_semaphore_group_t;
typedef struct moo_semaphore_group_t* moo_oop_semaphore_group_t; typedef struct moo_semaphore_group_t* moo_oop_semaphore_group_t;
@ -816,6 +816,11 @@ struct moo_semaphore_t
moo_oop_t signal_action; moo_oop_t signal_action;
moo_oop_semaphore_group_t group; /* nil or belonging semaphore group */ moo_oop_semaphore_group_t group; /* nil or belonging semaphore group */
struct
{
moo_oop_semaphore_t prev;
moo_oop_semaphore_t next;
} grm; /* group membership chain */
}; };
struct moo_semaphore_group_t struct moo_semaphore_group_t
@ -831,9 +836,17 @@ struct moo_semaphore_group_t
} waiting; } waiting;
/* [END IMPORTANT] */ /* [END IMPORTANT] */
moo_oop_t size; /* SmallInteger */ struct
moo_oop_t pos; /* current processing position */ {
moo_oop_oop_t semarr; /* Array of Semaphores */ moo_oop_semaphore_t first;
moo_oop_semaphore_t last;
} sems;
struct
{
moo_oop_semaphore_t first;
moo_oop_semaphore_t last;
} sigsems;
}; };
#define MOO_PROCESS_SCHEDULER_NAMED_INSTVARS 9 #define MOO_PROCESS_SCHEDULER_NAMED_INSTVARS 9