[squeak-dev] Problem with fork

On Wed, Jul 1, 2009 at 4:22 PM, Igor Stasenko <[hidden email]> wrote:

2009/7/1 Mariano Martinez Peck <[hidden email]>:

> Hi folks! I am doing some benchmarks and I having problems with fork (I
> think).
>
> I have this method collect:
>
> collect
> |semaphores tr|
> semaphores := Array new: 10.
> tr := ThreadSafeTranscript new.
> tr open.
> 1 to: 10 do: [ :index | semaphores at: index put: Semaphore
> forMutualExclusion ].
>
>     1 to: 10 do: [:i |
>         [
>         tr show: 'one fork'; cr.
>         (semaphores at: i) signal.
>         ] fork
>     ].
>
>     semaphores do: [:each | each wait ].
>     tr show: 'all forks proccesed'; cr.
>
>
> What I want is that the method collect returns when ALL of the forks created
> inside, are finished. Obviously in my real code, I don't do a transcript
> show: but another thing.
>
> I would expect something like this:
>
> one fork
> one fork
> one fork
> one fork
> one fork
> one fork
> one fork
> one fork
> one fork
> one fork
> all forks proccesed
>
> But the output is:
>
>
> all forks proccesed
> one fork
> one fork
> one fork
> one forkone fork
> one fork
> one fork
> one forkone fork
> one fork
> one fork
> one forkone fork
>
> So, I guess it isn't working :(   I don't know too much about forks so any
> help is welcome!
>
> can be a problem with  ThreadSafeTranscript  ? If so, how could I test if my
> code is working as expected ?
>

Mariano, sorry it way my mistake. In example that i gave you on IRC ,
i put Semaphore forMutualExclusion,
while obviously it should be simply: Semaphore new.

> Thanks,
>
> Mariano
>
>
>
>
>



--
Best regards,
Igor Stasenko AKA sig.

On Wed, Jul 1, 2009 at 4:34 PM, Igor Stasenko <[hidden email]> wrote:

2009/7/1 Mariano Martinez Peck <[hidden email]>:
>
>
> On Wed, Jul 1, 2009 at 4:22 PM, Igor Stasenko <[hidden email]> wrote:
>>
>> 2009/7/1 Mariano Martinez Peck <[hidden email]>:
>> > Hi folks! I am doing some benchmarks and I having problems with fork (I
>> > think).
>> >
>> > I have this method collect:
>> >
>> > collect
>> > |semaphores tr|
>> > semaphores := Array new: 10.
>> > tr := ThreadSafeTranscript new.
>> > tr open.
>> > 1 to: 10 do: [ :index | semaphores at: index put: Semaphore
>> > forMutualExclusion ].
>> >
>> >     1 to: 10 do: [:i |
>> >         [
>> >         tr show: 'one fork'; cr.
>> >         (semaphores at: i) signal.
>> >         ] fork
>> >     ].
>> >
>> >     semaphores do: [:each | each wait ].
>> >     tr show: 'all forks proccesed'; cr.
>> >
>> >
>> > What I want is that the method collect returns when ALL of the forks
>> > created
>> > inside, are finished. Obviously in my real code, I don't do a transcript
>> > show: but another thing.
>> >
>> > I would expect something like this:
>> >
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > one fork
>> > all forks proccesed
>> >
>> > But the output is:
>> >
>> >
>> > all forks proccesed
>> > one fork
>> > one fork
>> > one fork
>> > one forkone fork
>> > one fork
>> > one fork
>> > one forkone fork
>> > one fork
>> > one fork
>> > one forkone fork
>> >
>> > So, I guess it isn't working :(   I don't know too much about forks so
>> > any
>> > help is welcome!
>> >
>> > can be a problem with  ThreadSafeTranscript  ? If so, how could I test
>> > if my
>> > code is working as expected ?
>> >
>>
>> Mariano, sorry it way my mistake. In example that i gave you on IRC ,
>> i put Semaphore forMutualExclusion,
>> while obviously it should be simply: Semaphore new.
>
> Sorry, I was having lunch.
>
> Igor: I didn't know that sig__ where you :)

see my footer, down below :)


--

Best regards,
Igor Stasenko AKA sig.

On Wed, Jul 1, 2009 at 3:10 PM, Eliot Miranda <[hidden email]> wrote:

Hi Mariano,

    Smalltalk's scheduler is preemptive across priorities and co-operative within priorities, which is a well-chosen real-time scheduling policy giving precise control over processes.  Because all your processes have the same priority they don't preempt the first process and so they don't get to run until after the doit has executed and the system has returned to some idle loop state.  If instead you change your code to either use a higher priority, e.g. [...] forkAt: Processor activePriority + 1, or to use yield, e.g. [....] fork. Processor yield, then you'll be able to get your desired behaviour. 

On Wed, Jul 1, 2009 at 7:41 AM, Mariano Martinez Peck <[hidden email]> wrote:

Hi folks! I am doing some benchmarks and I having problems with fork (I think).

I have this method collect:

collect
|semaphores tr|
semaphores := Array new: 10.
tr := ThreadSafeTranscript new.
tr open.
1 to: 10 do: [ :index | semaphores at: index put: Semaphore forMutualExclusion ].

    1 to: 10 do: [:i |
        [
        tr show: 'one fork'; cr.
        (semaphores at: i) signal.
        ] fork     
    ].

    semaphores do: [:each | each wait ].
    tr show: 'all forks proccesed'; cr.
   

What I want is that the method collect returns when ALL of the forks created inside, are finished. Obviously in my real code, I don't do a transcript show: but another thing.

I would expect something like this:

one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
all forks proccesed

But the output is:


all forks proccesed
one fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork

So, I guess it isn't working :(   I don't know too much about forks so any help is welcome!

can be a problem with  ThreadSafeTranscript  ? If so, how could I test if my code is working as expected ?

Thanks,

Mariano

On Wed, Jul 1, 2009 at 10:55 AM, Mariano Martinez Peck <[hidden email]> wrote:

On Wed, Jul 1, 2009 at 3:10 PM, Eliot Miranda <[hidden email]> wrote:

Hi Mariano,

    Smalltalk's scheduler is preemptive across priorities and co-operative within priorities, which is a well-chosen real-time scheduling policy giving precise control over processes.  Because all your processes have the same priority they don't preempt the first process and so they don't get to run until after the doit has executed and the system has returned to some idle loop state.  If instead you change your code to either use a higher priority, e.g. [...] forkAt: Processor activePriority + 1, or to use yield, e.g. [....] fork. Processor yield, then you'll be able to get your desired behaviour. 

Eliot: Thanks for the reply. I think I am understanding a bit more. Let me explain myself a bit more what I am actually trying to do. I have SqueakDBX which uses FFI to call external functions of OpenDBX and do queries to databases. If you use SqueakDBX in a webapplication, like seaside, you will have like several forks using SqueakDBX. And that's exactly what I am trying to simulate: multiples forks (in a webapp they would be users connected). I am doing some benchmarks as I have some things to define (timeouts) and see how they affect this situation.

My real code (of the benchmark) is something like this:

collect: aBlock

|semaphores|
semaphores := Array new: self amountOfForks.
1 to: self amountOfForks do: [ :index | semaphores at: index put: Semaphore new ].

    1 to: self amountOfForks do: [:i |
        [
        mutex critical: [ times add: (MessageTally time: aBlock) ].
        (semaphores at: i) signal.
        ] forkAt: self benchPriority       

    ].

    semaphores do: [:each | each wait ].

In this case, aBlock is a piece of code that does a big select.

During all the methods which are called in aBlock (SqueakDBX methods) there are for example, FFI calls and several things more. Suppose all the forks are with the same priority. My question is now when would the scheduler take another one?  As OpenDBX supports Asynchronous queries I just ask using FFI is the query if done or not and the function returns immediately. But, in order to let another connection to do another query, what I do, in case the query is not finished is:

(Delay forMilliseconds: 1000) wait

This piece of code will free that process and pass to another? if not, I must add a yield ?

Thanks a lot for the help!

Mariano

 

On Wed, Jul 1, 2009 at 7:41 AM, Mariano Martinez Peck <[hidden email]> wrote:

Hi folks! I am doing some benchmarks and I having problems with fork (I think).

I have this method collect:

collect
|semaphores tr|
semaphores := Array new: 10.
tr := ThreadSafeTranscript new.
tr open.
1 to: 10 do: [ :index | semaphores at: index put: Semaphore forMutualExclusion ].

    1 to: 10 do: [:i |
        [
        tr show: 'one fork'; cr.
        (semaphores at: i) signal.
        ] fork     
    ].

    semaphores do: [:each | each wait ].
    tr show: 'all forks proccesed'; cr.
   

What I want is that the method collect returns when ALL of the forks created inside, are finished. Obviously in my real code, I don't do a transcript show: but another thing.

I would expect something like this:

one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
all forks proccesed

But the output is:


all forks proccesed
one fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork

So, I guess it isn't working :(   I don't know too much about forks so any help is welcome!

can be a problem with  ThreadSafeTranscript  ? If so, how could I test if my code is working as expected ?

Thanks,

Mariano

On Wed, Jul 1, 2009 at 5:40 PM, Eliot Miranda <[hidden email]> wrote:

On Wed, Jul 1, 2009 at 10:55 AM, Mariano Martinez Peck <[hidden email]> wrote:

On Wed, Jul 1, 2009 at 3:10 PM, Eliot Miranda <[hidden email]> wrote:

Hi Mariano,

    Smalltalk's scheduler is preemptive across priorities and co-operative within priorities, which is a well-chosen real-time scheduling policy giving precise control over processes.  Because all your processes have the same priority they don't preempt the first process and so they don't get to run until after the doit has executed and the system has returned to some idle loop state.  If instead you change your code to either use a higher priority, e.g. [...] forkAt: Processor activePriority + 1, or to use yield, e.g. [....] fork. Processor yield, then you'll be able to get your desired behaviour. 

Eliot: Thanks for the reply. I think I am understanding a bit more. Let me explain myself a bit more what I am actually trying to do. I have SqueakDBX which uses FFI to call external functions of OpenDBX and do queries to databases. If you use SqueakDBX in a webapplication, like seaside, you will have like several forks using SqueakDBX. And that's exactly what I am trying to simulate: multiples forks (in a webapp they would be users connected). I am doing some benchmarks as I have some things to define (timeouts) and see how they affect this situation.

My real code (of the benchmark) is something like this:

collect: aBlock

|semaphores|
semaphores := Array new: self amountOfForks.
1 to: self amountOfForks do: [ :index | semaphores at: index put: Semaphore new ].

    1 to: self amountOfForks do: [:i |
        [
        mutex critical: [ times add: (MessageTally time: aBlock) ].
        (semaphores at: i) signal.
        ] forkAt: self benchPriority       

    ].

    semaphores do: [:each | each wait ].

In this case, aBlock is a piece of code that does a big select.

During all the methods which are called in aBlock (SqueakDBX methods) there are for example, FFI calls and several things more. Suppose all the forks are with the same priority. My question is now when would the scheduler take another one?  As OpenDBX supports Asynchronous queries I just ask using FFI is the query if done or not and the function returns immediately. But, in order to let another connection to do another query, what I do, in case the query is not finished is:

(Delay forMilliseconds: 1000) wait

This piece of code will free that process and pass to another? if not, I must add a yield ?

Yes it will pass to another and no you don't need a yield.  Internal to a delay there is a semaphore and when the program does aDelay wait the active process waits on that semaphore.  The scheduler then switches to the first highest priority process.  The Delay class maintains a process that uses a VM primitive primSignal:atMilliseconds: to start a timer in the VM that will signal the semaphore supplied to the primitive when the timer expires.  The class side process then wakes (preempting the current process if the delay process is of higher priority, which it very likely is), checks the set of waiting delays and signals the semaphores of those that should be woken up, which will make those processes runnable.

You only need to do a yield when a process wants to allow a process at the same priority a chance to run but the yielding process has work to do.  If instead the process waits on a semaphore (in a Delay or explicitly) it will effectively yield.  Also note that the yield primitive is only a space efficient version of the following Smalltalk code:

     oldSlowLetsCreateLotsOfObjectsButNotRequireVMModificationsYield

        | sem |

        sem := Semaphore new.

        [sem signal] fork.

        sem wait

which is a real waste if there are no processes at the current priority level waiting to run, hence the VM primitive which avoids all this work by simply inspecting teh current priority level's runnable process queue.

I was going to suggest that you read the class comments of ProcessorScheduler, Process, Semaphore, Mutex and Delay, but they're pretty simplistic.  Better to read the blue book chapters on the scheduler either here http://web.archive.org/web/20060615014524/http://users.ipa.net/~dwighth/smalltalk/bluebook/bluebook_chapter29.html#ControlPrimitives29 or here: http://stephane.ducasse.free.fr/FreeBooks/BlueBook/

I'm sure others will have better reference material to suggest.

cheers

e.

Thanks a lot for the help!

Mariano

 

On Wed, Jul 1, 2009 at 7:41 AM, Mariano Martinez Peck <[hidden email]> wrote:

Hi folks! I am doing some benchmarks and I having problems with fork (I think).

I have this method collect:

collect
|semaphores tr|
semaphores := Array new: 10.
tr := ThreadSafeTranscript new.
tr open.
1 to: 10 do: [ :index | semaphores at: index put: Semaphore forMutualExclusion ].

    1 to: 10 do: [:i |
        [
        tr show: 'one fork'; cr.
        (semaphores at: i) signal.
        ] fork     
    ].

    semaphores do: [:each | each wait ].
    tr show: 'all forks proccesed'; cr.
   

What I want is that the method collect returns when ALL of the forks created inside, are finished. Obviously in my real code, I don't do a transcript show: but another thing.

I would expect something like this:

one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
one fork
all forks proccesed

But the output is:


all forks proccesed
one fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork
one fork
one fork
one forkone fork

So, I guess it isn't working :(   I don't know too much about forks so any help is welcome!

can be a problem with  ThreadSafeTranscript  ? If so, how could I test if my code is working as expected ?

Thanks,

Mariano

Hi all,
  this is my first post here, just joined this group so let me do a
quick introduction.
I have 10+ years of experience doing full-time development with
VisualWorks (creating trading platforms for the European energy
exchanges).
Also I have near-zero experience with using Squeak or how its
development process works. I'll slowly cut my teeth, starting by
interacting on this mailing list.


Multi threading is something I have spent a lot of quality time with,
so I want to share some thoughts on the following:
1 to: 10 do: [ :index | semaphores at: index put: Semaphore new ].
        tr nextPutAll: i printString, ' fork '; cr.
I have seen this pattern often (allocating a semaphore for every
forked process), I usually interpret this as a signal that such code
is still in its first 'make it work/make it right' stages.
What a lot of people don't realize is that at its heart a semapore is
a thread-safe counter/register (and if you look at the hierarchy it is
implemented on you wouldn't guess that either, since the hierarchy
stresses the implementation part that manages waiting processes rather
than the counter aspect).

So trying to take the code snippet toward 'make it abstract' territory
this could be refactored to lean more on the counter aspect of
semaphores and use only a single semaphore:


|count sem tr|
count := 10.
sem := Semaphore new.

1 to: count do: [:i |
       [       tr nextPutAll: (i printString, ' fork\') withCRs.
               sem signal.
       ] fork].

count timesRepeat: [sem wait].
Now the above is about as far as you can go with the current Squeak
and VisualWorks implementations so you can take it as a simple
refactoring advise.




However I want to press on a bit more (and go a bit off-topic for this
list ;-) because I feel it still has a big problem: we need to
maintain a 'count' and pass that between the two loops in the above
example.
In the current example this is not much of a problem but in more
complex applications where the forking is done by yet other forked
processes we will need to make 'count' thread-safe as well -- I find
this very ugly, because you will need an extra semaphore just to make
the original semaphore work as required.
Furthermore you cannot add new forked processes once the second loop
has started running.

So here is an experiment I did a couple of years ago with VisualWorks:
I altered the VM (just one line of its source ;-) so it would react
properly to semaphores that have negative values in the
'excessSignals' instance variable, and I added a method #unsignal to
Semaphore that would decrease the value of that ivar.

In my experiments that yielded many opportunities to simplify
multiprocessing code (not only for thread synchronization but also for
passing around counts in a thread-safe register!).

In the above code that would allow us to 'pre-load' the semaphore at
the place where the threads are created with as result that the
'count' variable can be removed and the bottom loop can be removed too:


|count sem tr|
sem := Semaphore forMutualExclusion. "We need one excessSignal to
balance the #wait below"
       sem unsignal. "outside the forked code"
       [       tr nextPutAll: (i printString, ' fork\') withCRs.
               sem signal. "balance the unsignal"
       ] fork].

sem wait. "no loop, no need to know the count!"
Above was just a simple refactoring, but look at how I needed it:

|sem tr|
sem := Semaphore new. "no excessSignal this time"

"set up a monitoring system first(!)"
[       sem wait.
] fork.

"then create jobs (in my case I had only a single first job that would
recursively create more jobs, not shown here)"
       sem unsignal.
       [       tr nextPutAll: (i printString, ' fork\') withCRs.
               sem signal.
       ] fork].
"Now that we are sure at least one job is entered balance the #wait we
started out with"
sem signal.


Since we elided 'count' I can move the code that relied on it up in
front of the thread creation code, I very much like this flavor of
decoupling.


I guess this illustrates that Semaphore is stuck in the 'make it work/
make it right' phase for thirty years now, and that moving it into
'make it abstract' territory will make lots of hairy multi-threading
code much simpler to express...

(And for those thinking this through: yes I did implement a thread-
safe #add: and #valueWithReset on semaphore too ;-)


I hope I didn't bore y'all and stray to far off-topic, but I did want
to share this bit of insight I gained by tinkering with the semaphore
implementation: semaphores are thread-safe counters at their heart.



Cheers,

Reinout
-------

PS: big congrats with the license cleaning milestone, this is what
finally pulled me into this project :-)