1. Computing

Threaded Delphi Tasks In A Thread Pool - OTL (OmniThreadLibrary) Example

Running Tasks In Threads - Thread Pool Implementation In OTL

By

Delphi Thread Pool - OTL
Delphi Thread Pool - OTL
I've been working on my threaded string parser - a test project to see how I could speed up one of the tasks in a Delphi application I am working on.

The problem I had to solve is how to make file processing as fast as possible. Having a bunch of files (from 500 to 2000+) I have to process them (let's call this "scanning"). The sequential approach (one after another) proven to be too slow. Machines have more cores today and the idea to have this file scanning running in multiple threads is something that must speed up the entire process.

The initial idea to have each file being scanned in a separate thread would mean creating a thread for each file and have all the threads work their task at the same time - resulting in having 500-2000+ threads running in the background.

This is not a good idea - Windows does not like having an application start (and run) too many threads at once: Pushing the Limits of Windows: Processes and Threads

The solution would be to use a thread pool.

Thread Pools

In simple words, a thread pool is a collection of threads that can be used to perform a number of tasks in the background (not in the main application thread). You send a task to be executes in the background and the thread pool ensures there's only a limited number of threads running. The thread pool is typically implemented as a queue of tasks. As soon as one thread in a pool finishes with a task it is served with the next task to process.

While Delphi does allow you to easily create a separate thread (by extending the TThread class) it does not provide a thread pool implementation out of the box.

Luckily there are a few free (even open source) thread pool implementations a Delphi developer can use. One such library is the OTL - Omni Thread Library.

Let's see how to do threaded tasks in a thread pool using OTL

Thread Pool Implementation In OTL

The OTL developed by Primoz Gabrijelcic is a simple to use threading library for Delphi. The "is simple" means that Gabriel has successfully hidden from you the complexity of running tasks in threads where thread synchronization and critical sections for data protection are nicely handled by the OTL.

While OTL is easy to use, one needs to invest time to figure out all the possibilities OTL exposes as (mostly is the case) there's no good documentation but lots and lots of examples to help you get to speed with OTL.

Enough theory, let's see OTL in action: background tasks and thread pool implementation:

For the sake of simplicity the task I want to be executed in the background (a thread) is a simple Sleep procedure call - the easiest way to mimic some workload that would be done by the real-world task you would like to be executed in a separate thread. Sleep procedure is (from my point of view) so useless but so great at the same time: Sleep Sort in OTL.

The OTL thread pool implementation is based around the "GlobalOmniThreadPool" singleton instance of IOmniThreadPool. You can create your own IOmniThreadPool, but GlobalOmniThreadPool would be enough most of the time.

GlobalOmniThreadPool can be used (among lots of other things) to specify how many threads should process your tasks (the "MaxExecuting" property). Also, the CountExecuting and CountQueued would return the number of tasks (threads) being executed at a time and also the number of tasks waiting in the queue.

You could set all the required GlobalOmniThreadPool properties in the OnCreate event of your main form, for example:

procedure TMainOTLTestForm.FormCreate(Sender: TObject);
begin
  GlobalOmniThreadPool.MaxExecuting := 2 * System.CPUCount;
end;
The above sets the number of threads in the GlobalOmniThreadPool to be 2 * number of processors (cores) on a machine.

Here's the real work using, what is called in OTL, "low level" approach:

procedure TMainOTLTestForm.Button1Click(Sender: TObject);
const
  MSG_START = 1;
var
  iTask, numTasks, delay_ms : integer;
begin
  numTasks := 100;

  delay_ms := Random(100);

  for iTask := 1 to numTasks do
  begin
    Application.ProcessMessages;

    CreateTask(
      procedure(const task: IOmniTask)
      var
        sleepTime : integer;
      begin
        task.Comm.Send(MSG_START);

        sleepTime := task.Param['SleepTime'].AsInteger;

        //some task workload - this is where your background processing goes
	Sleep(sleepTime);
      end)
      .OnMessage(
        procedure(const task: IOmniTaskControl; const msg: TOmniMessage)
        var
          taskNumber : integer;
        begin
          if msg.MsgID = MSG_START then
          begin
            taskNumber := task.Param['TaskNumber'].AsInteger;

            LogPoolStatus(Format('task %d / %d start', [taskNumber, task.UniqueID]));
          end;
        end)
      .OnTerminated(
        procedure(const task: IOmniTaskControl)
        begin
          LogPoolStatus(Format('task id %d terminated', [task.UniqueID]));
        end)
      .SetParameter('TaskNumber', iTask)
      .SetParameter('SleepTime', delay_ms)
      .Unobserved
      .Schedule;
  end; (* for loop *)

  //wait all finished
  while GlobalOmniThreadPool.CountExecuting + GlobalOmniThreadPool.CountQueued > 0 do
    Application.ProcessMessages;

  //all task completed
  LogPoolStatus('ALL DONE');
end;

Let's first deal with the less complex sections of the above example: the LogPoolStatus is a form (main UI) level procedure used to print out (add a line in a Memo control) the status of the thread pool and the task being executed.

procedure TMainOTLTestForm.LogPoolStatus(const msg : string);
begin
  Memo1.Lines.Add(Format('MSG: %s. Pool: %d executing / %d queued',
    [msg, GlobalOmniThreadPool.CountExecuting, GlobalOmniThreadPool.CountQueued]));
end;

Here's what happens in the above code:

  1. 100 (numTasks variable) tasks should be executed out of the main thread (in the background).
  2. For each task we create an IOmniTask(Control) instance using the CreateTask procedure.
  3. The CreateTask uses an anonymous procedure to specify what the background task should be.
  4. The task we execute in the background is a simple Sleep call.
  5. When the tasks starts (is grabbed by the pool to start its execution in a background thread) we send a message using task.Comm.Send that will be handled by the OnMessage procedure.
  6. The OnMessage is executed in the main thread - this is where you can safely update your user interface. Think of OnMessage as the Synhronize procedure as implemented in Delphi's TThread.
  7. When the task is done executing (Sleep function has finished) the OnTerminated can be used to again safely update the main UI.
  8. The task would certainly need some input parameters - this is what SetParameter can be used for. We send to the background task two parameters: SleepTime and TaskNumber.
  9. The Unobserved tells to the poll that we are not using TOmniEventMonitor instance to control freeing of the created task, rather the task should free itself when terminated.
  10. Finally, Schedule adds the task to the thread pool.

Note: there are different approaches in OTL to do the same. I love anonymous methods and how all the code (the task) to be executed in a separate thread is at the same place.

In my real-world threaded task I needed to allow the user to either wait for all the tasks to be competed or cancel the tasks that are waiting in the pool.

To wait for all the task to be finished:

//wait all finished
while GlobalOmniThreadPool.CountExecuting + GlobalOmniThreadPool.CountQueued > 0 do
  Application.ProcessMessages;
To cancel the tasks in the pool queue:
  GlobalOmniThreadPool.CancelAll;

Grab the full source code from here: Thread Pool OTL Sample and explore / upgrade if you like.

AsyncCalls by Andreas Hausladen

AsyncCalls is another Delphi threading library implementing a thread pool you can use when in need to execute multiple functions at the same time and synchronize them at every point in the function or method that started them. Read: Delphi Thread Pool Example Using AsyncCalls.
  1. About.com
  2. Computing
  3. Delphi
  4. Coding Delphi Applications
  5. Threading in Delphi
  6. Threaded Delphi Tasks In A Thread Pool - OTL (OmniThreadLibrary) Example

©2014 About.com. All rights reserved.