A Simple Thread Pool for Smalltalk




Object subclass: #ThreadWorker
    instanceVariableNames: 'block expires value lock error'
    classVariableNames: ''
    poolDictionaries: ''
    category: 'ThreadPool'


ThreadWorker class>>on: aBlock
    ^ self new
        block: aBlock;

ThreadWorker class>>on: aBlock expires: aTime
    ^ self new
        block: aBlock;
        expires: aTime;

On the instance side let's initialize the instance and set up the necessary accessors for the constructors above.

    super initialize.
    lock := Semaphore new 

ThreadWorker>>block: aBlock
    block := aBlock 

ThreadWorker>>expires: aTime
    expires := aTime


    DateAndTime now < expires ifTrue: 
        [ [ value := block value ] 
            on: Error
    lock signal


    lock isSignaled ifFalse: [ lock wait ].
    "rethrow any error from worker thread on calling thread"
    error ifNotNil: 
        [ error
            privHandlerContext: thisContext;
            signal ].
    ^ value


    ^ lock isSignaled


Object subclass: #ThreadPool
    instanceVariableNames: ''
    classVariableNames: 'MaxPoolSize MinPoolSize 
        PoolManager QueueWorkers WorkQueue'
    poolDictionaries: ''
    category: 'ThreadPool'

ThreadPool class>>initialize
    "self initialize"
    WorkQueue := SharedQueue2 new.
    QueueWorkers := OrderedCollection new.
    MinPoolSize := 5.
    MaxPoolSize := 15.
    Smalltalk addToStartUpList: self 

ThreadPool class>>maxPoolSize: aSize
    MaxPoolSize := aSize 

ThreadPool class>>minPoolSize: aSize
    MinPoolSize := aSize


ThreadPool class>>isPoolTooBig
    ^ QueueWorkers size > MinPoolSize 
        and: [ WorkQueue size < QueueWorkers size ]

ThreadPool class>>isPoolTooSmall
    ^ QueueWorkers size < MinPoolSize 
        or: [ WorkQueue size > QueueWorkers size 
            and: [ QueueWorkers size < MaxPoolSize ] ]


ThreadPool class>>processQueueElement
    [ WorkQueue next evaluate ] 
        on: Error
        do: [  ]


ThreadPool class>>startWorker
    QueueWorkers add: ([ [ self processQueueElement ] repeat ] 
            forkAt: Processor systemBackgroundPriority
            named: 'pool worker')


ThreadPool class>>queueWorkItem: aBlock expiresAt: aTimestamp 
    | worker |
    worker := ThreadWorker on: aBlock expires: aTimestamp.
    WorkQueue nextPut: worker.
    ^ worker 

ThreadPool class>>queueWorkItem: aBlock expiresAt: aTimestamp 
    session: aSession 
    | worker |
    "a special method for Seaside2.8 so the worker threads 
    still have access to the current session"
    worker := ThreadWorker 
            [ WACurrentSession 
                use: aSession
                during: aBlock ]
        expires: aTimestamp.
    WorkQueue nextPut: worker.
    ^ worker 

BlockClosure>>queueWorkAndExpireIn: aDuration
    ^ ThreadPool 
        queueWorkItem: self
        expiresAt: DateAndTime now + aDuration

BlockClosure>>queueWorkAndExpireIn: aDuration session: aSession 
    "a special method for Seaside2.8 so the worker threads 
     still have access to the current session"
    ^ ThreadPool 
        queueWorkItem: self
        expiresAt: DateAndTime now + aDuration
        session: aSession

And now we're able to queue a job to kill a thread, making sure to double-check at time of actual execution that the pool is still too big and the thread still needs to die.

ThreadPool class>>killWorker
    "just queue a task that kill the activeProcess, 
    which will be the worker that picks it up"

    [ self isPoolTooBig ifTrue: 
       [ (QueueWorkers remove: Processor activeProcess) terminate ] ] 
        queueWorkAndExpireIn: 10 minutes


ThreadPool class>>adjustThreadPoolSize
    "starting up processes too fast is dangerous 
     and wasteful, ensure a reasonable delay"
    1 second asDelay wait.
    self isPoolTooSmall 
        ifTrue: [ self startWorker ]
        ifFalse: [ self isPoolTooBig ifTrue: [ self killWorker ] ]


ThreadPool class>>startUp
    "self startUp"
    self shutDown.
    MinPoolSize timesRepeat: [ self startWorker ].
    PoolManager := [ [ self adjustThreadPoolSize ] repeat ] 
        forkAt: Processor systemBackgroundPriority
        named: 'pool manager'

And clean up everything on shutdown so every time the image starts up we're starting from a clean slate.

ThreadPool class>>shutDown
    "self shutDown"
    WorkQueue := SharedQueue2 new.
    PoolManager ifNotNil: [ PoolManager terminate ].
    QueueWorkers do: [ :each | each terminate ].
    QueueWorkers removeAll.


[ self someTaskToDo ] fork

I just do...

[ self someTaskToDo ] queueWorkAndExpireIn: 25 seconds

Or in Seaside...

[ self someTaskToDo ] queueWorkAndExpireIn: 25 seconds session: self session



Comments (automatically disabled after 1 year)

Vadim Tsushko 3875 days ago

Nice to see you again, Ramon. And with such a real good post, too ;)

Paul DeBruicker 3875 days ago

Shoud this:

ThreadPool class&gt;&gt;isPoolTooBig
    ^ QueueWorkers size &gt; MinPoolSize 
        and: [ WorkQueue size &lt; QueueWorkers size ]

be this:

ThreadPool class&gt;&gt;isPoolTooBig
    ^ QueueWorkers size &gt; MaxPoolSize 
        and: [ WorkQueue size &lt; QueueWorkers size ]
Ramon Leon 3875 days ago

No, it's correct. Read it as "if there are at least the minimum number of workers but there is less work to do than there are workers, the pool is too big."

isPoolTooSmall checks the max and prevents the queue from ever growing larger than it.

Michael van der Gulik 3875 days ago

Personally, I think it would be worth improving the scheduler in Squeak. Thread pools are a hack around an inefficient threading system.

Squeak's processes are very light-weight; they're just one object and comparable in overhead to a stack frame. If the scheduler was improved, you could fork like a drunken sailer who has no shortage of wenches.

Ramon Leon 3875 days ago

That's a big if. Thread pools are a hack, but they work and it solves my problem quickly with a library rather than a change to a core part of the image that I'd then have to try and get accepted. Plus I don't know enough about the process scheduler to even know why I can't fork a thousand threads, let alone fix it.

This was also something I did a while back, I just need a topic for a post because I felt like writing something again. I just did a file out and started writing around the code.

Colin 3872 days ago

Great to see you back!

As a newcomer to smalltalk your blog is brilliant reading. Ever thought about writing a book in a similar vein?


Ramon Leon 3872 days ago

Nah, a blog is the modern version of a self published book. Sitting down to write an actual books is a labor of love, a lot of work for very little return, especially in a market as small as Smalltalk programmers.

Tim Ziebart 3788 days ago

Ramon, is this ready to go for Seaside 3.0? I noticed a possible change needed in "queueWorkItem: aBlock expiresAt: aTimestamp session: aSession" from WACurrentSession to WACurrentRequestContext. If I do this I then get the error in the thread worker 'key not found'. This on the block closure on: do:. Are there other changes needed? I cannot seem to get this working on 3. Thank you.

Ramon Leon 3787 days ago

Two things, no, it's not ready for 3.0, but I can fix that today. Secondly, load the latest code from the ThreadPool package on SqueakSource. I'll get it updated for 3.0 later today or tonight. I'm still using 2.8 so I just haven't gotten around to it.

Tim Ziebart 3787 days ago

Thanks Ramon, I loaded this 2 days ago so, I assume it is the latest. Look forward to update.

about me|good books|popular posts|atom|rss