OmniThreadLibrary 1.05

As there were no error reports related to OmniThreadLibrary 1.05 RC, I’ve released final 1.05 version just few moments ago. There are almost no changes between the RC and final release – one demo was added and Parallel.Join code was tweaked a little.

You can download OTL 1.05 from the Google Code. Alternatively, you can update SVN trunk (checkout instructions) or checkout the release-1.05 tag.

Support is available on the web discussion forum.

Big rename

Many internal classes and interfaces was renamed. This should not affect most of the users.

• TOmniBaseStack –> TOmniBaseBoundedStack
• TOmniStack –> TOmniBoundedStack
• TOmniBaseQueue –> TOmniBaseBoundedQueue
• TOmniQueue –> TOmniBoundedQueue
• IInterfaceDictionary –> IOmniInterfaceDictionary
• IInterfaceDictionaryEnumerator -> IOmniInterfaceDictionaryEnumerator,
• TInterfaceDictionaryPair –> TOmniInterfaceDictionaryPair

I’m sorry for that. Some names are badly chosen and some did not follow the OTL naming conventions.

Dynamic lock-free queue

Implemented dynamically allocated, O(1) enqueue and dequeue, threadsafe,  lock-free queue. Class TOmniBaseQueue contains base implementation while TOmniQueue adds observer support. Both classes live in the OtlContainers unit.

Read more about the TOmniQueue: Dynamic lock-free queue – doing it right.

Inverse semaphore

Implemented resource counter with empty state signalling TOmniResourceCount (unit  OtlSync).

Read more: Three steps to the blocking collection: [1] Inverse semaphore.

Blocking collection

New unit OtlCollection which contains blocking collection implementation  TOmniBlockingCollection.

Read more: Three steps to the blocking collection: [3] Blocking collection

Parallel

New high-level parallelism support (unit OtlParallel). Requires at least Delphi 2009.

Two parallel control structures are supported: for each (with optional aggregator) and join.

The demo for Parallel.ForEach can be found in project 35_ParallelFor. The same code is reprinted near the end of the Three steps to the blocking collection: [3] Blocking collection post.

Parallel.ForEach.Aggregate was described in Parallel.ForEach.Aggreate post and is demoed in project 36_ParallelAggregate.

At the moment ForEach is fairly limited. It can iterate over a range of numbers or over a collection supporting the IOmniValueEnumerable interface (TOmniBlockingCollection, for example). The second limitation will be removed in the future. The plan is to support any collection that implements IEnumerable.

Parallel.Join is very simple code that executes multiple tasks and waits for their completion. It was designed to execute simple tasks that don’t require communication with the owner. It is demoed in project 37_ParallelJoin.

Environment

Unit OtlCommon contains new interface IOmniEnvironment and function Environment that returns singleton of this type. Environment can be used to query some basic information on system, process and thread. Some information (for example process and thread affinity) can also be modified using the same interface.

  IOmniAffinity = interface
    property AsString: string;
    property Count: integer;
    property Mask: DWORD;
  end;

  IOmniProcessEnvironment = interface
    property Affinity: IOmniAffinity;
    property Memory: TOmniProcessMemoryCounters;
    property PriorityClass: TOmniProcessPriorityClass;
    property Times: TOmniProcessTimes;
  end;

  IOmniSystemEnvironment = interface
    property Affinity: IOmniAffinity;
  end;

  IOmniThreadEnvironment = interface
    property Affinity: IOmniAffinity;
    property ID: cardinal;
  end;

  IOmniEnvironment = interface
    property Process: IOmniProcessEnvironment;
    property System: IOmniSystemEnvironment;
    property Thread: IOmniThreadEnvironment;
  end;

Newer demos are using some parts of the Environment interface. For example, in demo 33_BlockingCollection, process affinity is set with

  Environment.Process.Affinity.Count := inpNumCPU.Value; 

while the demo 35_ParallelFor uses following code fragment to query process affinity

  numTasks := Environment.Process.Affinity.Count; 

Cancellation token

New interface IOmniCancellationToken is used in the Parallel.ForLoop (see post Three steps to the blocking collection: [3] Blocking collection for the example) and in IOmniTaskControl.TerminateWhen.

IOmniTaskControl and IOmniTask implement CancellationToken: IOmniCancellationToken  property which can be used by the task and task controller.

IOmniCancellationToken is just a simple wrapper around the Win32 event primitive and is defined in the OtlSync unit.

  IOmniCancellationToken = interface
    procedure Clear;
    function  IsSignaled: boolean;
    procedure Signal;
    property Handle: THandle;
  end; { IOmniCancellationToken }

Message dispatcher

IOmniTaskControl now implements message dispatching setter in form OnMessage(msgID, handler). Use it to route specific message IDs to specific functions when global TOmniEventMonitor is not used.

An example from one of my applications:

  spmDatabaseConn := CreateTask(
      TSttdbPlaylistDatabaseWorker.Create(), 
      'Playlist Monitor Database Connection')
    .SetParameters([serverAddress, serverPort, username, password])
    .SetTimer(15*1000, @TSttdbPlaylistDatabaseWorker.CheckDBVersion)
    .OnMessage(MSG_DB_ERROR,   HandleError)
    .OnMessage(MSG_DB_STATUS,  HandleDatabaseStatus)
    .OnMessage(MSG_DB_VERSION, HandleDatabaseVersion)
    .Run; 

UserData[]

Implemented IOmniTaskControl.UserData[]. The application can store any values in this array. It can be accessed via the integer or string index. This storage are can only be access from the task controller side. Access is not thread-safe so you should use it only from one thread or create your own protection mechanism.

Small changes

• IOmniTask implements Implementor property which points back to the worker instance  (but only if worker is TOmniWorker-based).
• Refactored and enhanced TOmniValueContainer.
• TOmniTaskFunction now takes 'const' parameter. 
  TOmniTaskFunction = reference to procedure(const task: IOmniTask).
• Implemented TOmniValue.IsInteger.

Bugs fixed

• TOmniEventMonitor.OnTaskUndeliveredMessage was missing 'message' parameter.
• Set package names and designtime/runtime type in D2009/D2010 packages.

New demos

• 32_Queue: Stress test for new TOmniBaseQueue and TOmniQueue.
• 33_BlockingCollection: Stress test for new TOmniBlockingCollection, also demoes  the use of Environment to set process affinity.
• 34_TreeScan: Parallel tree scan using TOmniBlockingCollection.
• 35_ParallelFor: Parallel tree scan using Parallel.ForEach (Delphi 2009 and newer).
• 36_ParallelAggregate: Parallel calculations using Parallel.ForEach.Aggregate  (Delphi 2009 and newer).