Parallel Programming with OmniThreadLibrary

Primož Gabrijelčič
This book is for sale at [Link]

This version was published on 2019-03-01

This is a Leanpub book. Leanpub empowers authors and publishers with the Lean Publishing process. Lean
Publishing is the act of publishing an in-progress ebook using lightweight tools and many iterations to get
reader feedback, pivot until you have the right book and build traction once you do.

© 2012 - 2019 Primož Gabrijelčič

Tweet This Book!
Please help Primož Gabrijelčič by spreading the word about this book on Twitter!
The suggested hashtag for this book is #omnithreadlibrary.
Find out what other people are saying about the book by clicking on this link to search for this hashtag on
Twitter:
#omnithreadlibrary
Contents

Sample Book . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . i

About me . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ii

Credits . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . iii

Introduction . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . iv
Formatting conventions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . iv
Learn more . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . iv

Release notes . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . vi

1. Introduction to OmniThreadLibrary . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1
1.1 Requirements . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1
1.2 License . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 1
1.3 Installation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 2
1.3.1 Installing with GetIt . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 3
1.3.2 Installing with Delphinus . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4
1.3.3 Installing design package . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 5
1.4 Why use OmniThreadLibrary? . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 6
1.5 Tasks vs. threads . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 6
1.6 Locking vs. messaging . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 7
1.7 Message loop required . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 8
1.7.1 OmniThreadLibrary and console . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 9
1.7.2 OmniThreadLibrary task started from another task . . . . . . . . . . . . . . . . . . 10
1.7.3 OmniThreadLibrary task started from a TThread . . . . . . . . . . . . . . . . . . . . 12
1.8 TOmniValue . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 14
1.8.1 Data access . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 15
1.8.2 Type testing . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 17
1.8.3 Clearing the content . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 18
1.8.4 Operators . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 18
1.8.5 Using with generic types . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 19
1.8.6 Array access . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 20
1.8.7 Handling records . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 21
1.8.8 Object ownership . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 21
1.8.9 Working with TValue . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 22
1.8.10 Low-level methods . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 22
CONTENTS

1.9 TOmniValueObj . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 22
1.10 Fluent interfaces . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 23

2. High-level multi-threading . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 26
2.1 Async . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 27
2.1.1 Handling exceptions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 29
2.2 Async/Await . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 30
2.3 ForEach . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 31
2.3.1 Cooperation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 32
2.3.2 Iterating over … . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 32
[Link] … Number ranges . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 32
[Link] … Enumerable collections . . . . . . . . . . . . . . . . . . . . . . . . . . . 33
[Link] … Thread-safe enumerable collections . . . . . . . . . . . . . . . . . . . 35
[Link] … Blocking collections . . . . . . . . . . . . . . . . . . . . . . . . . . . . 35
[Link] … Anything . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 35
2.3.3 Providing external input . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 36
2.3.4 IOmniParallelLoop interface . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 38
2.3.5 Preserving output order . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 41
2.3.6 Aggregation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 43
2.3.7 Cancellation . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 45
2.3.8 Task initialization and finalization . . . . . . . . . . . . . . . . . . . . . . . . . . . . 46
2.3.9 Handling exceptions . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 47
2.3.10 Examples . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 47

3. Low-level multi-threading . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 48
3.1 Low-level for the impatient . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 48
3.2 Four ways to create a task . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 49
3.3 IOmniTaskControl and IOmniTask interfaces . . . . . . . . . . . . . . . . . . . . . . . . . . . 50
3.4 Task controller needs an owner . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 52
3.5 Communication subsystem . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 53
3.6 Processor groups and NUMA nodes . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 55
3.7 Lock-free collections . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 56
3.7.1 Bounded Stack . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 56
3.7.2 Bounded queue . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 58
3.7.3 Message queue . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 59
3.7.4 Dynamic queue . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 61
3.7.5 Observing lock-free collections . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 62
[Link] Examples . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 63
3.7.6 Benchmarks . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 64

4. Synchronization . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 67
4.1 Critical sections . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 67
4.1.1 IOmniCriticalSection . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 67
4.1.2 TOmniCS . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 68
4.1.3 Locked<T> . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 70
[Link] Why not use TMonitor? . . . . . . . . . . . . . . . . . . . . . . . . . . . . 72
CONTENTS

4.2 TWaitFor . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 73
4.3 TOmniCounter . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 74
4.4 TOmniAlignedInt32 and TOmniAlignedInt64 . . . . . . . . . . . . . . . . . . . . . . . . . . . 75

5. How-to . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 77
5.1 Parallel data production . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 79
5.2 QuickSort and parallel max . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 82
5.2.1 QuickSort . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 82
5.2.2 Parallel max . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 84

6. B. Demo applications . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 86

7. C. Examples . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 88

8. D. Hooking into OmniThreadLibrary . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 89

8.1 Exception notifications . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 89
8.2 Thread notifications . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 89
8.3 Pool notifications . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 90
Sample Book
This is just the sample of the real book. It includes selected parts from some chapters.
You can buy the book on the LeanPub¹.
¹[Link]
About me
I started programming on 8-bit micros in 1980s which gave me a never fully satisfied need to push as much
as possible from available hardware. That inheritance brought me quite organically to the current job where
I program high-availability server applications used in the broadcasting industry.
As a side result of this focus my open source projects also tend to deal with performance. I wrote a source-
instrumenting profiler for Delphi, GpProfile, which is now obsolete and no longer used. Another project, a
parallel programming library for Delphi called OmniThreadLibrary quickly became a favourite in the Delphi
community and is now used by many thousand users all over the world.
I enjoy teaching almost as much as coding. In my younger years I wrote for leading Slovenian technical
magazine ‘Monitor’², for topical ‘The Delphi Magazine’ and later for ‘Blaise Pascal Magazine’³. Most of my
articles written for ‘The Delphi Magazine’ are reprinted with permission on my blog ‘The Delphi Geek’⁴.
Lately my focus has shifted from writing for magazines to presenting on Delphi conferences, organizing
workshops (in real life or online), consulting and writing books. Besides the book you are reading, I also
wrote ‘Delphi High Performance’⁵ and ‘Hands-On Design Patterns with Delphi‘⁶, both published by ‘Packt
Publishing’⁷.
Find more about me at [Link]
²[Link]
³[Link]
⁴[Link]
⁵[Link]
⁶[Link]
⁷[Link]
Credits
I would like to thank Gorazd Jernejc (also known as GJ) for his contribution to the OmniThreadLibrary
project. Gorazd, OTL wouldn’t be the same without you!
OmniThreadLibrary would not be the same without the contributors (listed in alphabetical order): ajasja, Alex
Egorov, Alexander Alexeev, andi, Anton Alisov, arioch, Dean Hill, dottor_jeckill, geskill, Hallvard Vassbotn,
HHasenack, Jamie, Lee Nover, M.J. Brandt, Mason Wheeler, Mayjest, meishier, morlic, Passella, Qmodem,
[Link], Steve Maughan, Unspoken, Uwe Raabe, VyPu, Zarko.
Great thanks go to Pierre le Riche who wrote beautiful FastMM⁸ memory manager and allowed me to include
his fine work in the distribution.
I would like to thank Joe C. Hecht for reading my book from start from finish and providing me with many
suggestions that helped make this book even better.
Great thanks go also to the platform which allows me to self-publish my books - LeanPub⁹. This book would
never happen without the hard-working people running the LeanPub.
The ProWritingAid¹⁰ tool was indispensable in fixing style and grammar errors.
Cover page was photographed by © Dave Gingrich¹¹.
⁸[Link]
⁹[Link]
¹⁰[Link]
¹¹[Link]
Introduction
This is a book about OmniThreadLibrary¹², a multi-threading library for the Embarcadero Delphi rapid
development environment.
To follow the book, the reader should have some understanding about the multi-threading programming. If
you are new to multi-threading, I recommend reading the ‘Multithreading - The Delphi Way’¹³ by Martin
Harvey. That book is an oldie, but goldie.
A more up-to-date overview of Delphi multi-threading capabilities was published in the ‘Delphi XE2
Foundations, Part 3’¹⁴ by Chris Rolliston, available on Amazon¹⁵.

Formatting conventions
This book covers the latest official OmniThreadLibrary release – 3.07.7.
When a part of the book covers a different version, a [version tag] in superscript will show relevant version or
versions. Version numbers (f.i. 2.1) are used for older releases and SVN revision numbers (f.i. r1184) are used
for a functionality that was added after the last official release.
A single version or revision number (f.i. [r1184] ) shows that the topic in question was introduced in this version
and that it is still supported in the current release.
A range of two versions (f.i. [1.0-1.1] ) shows that the topic was introduced in the first version (1.0 in this
example) and that it was supported up to the second version (1.1). After that, support for this topic was
removed, or it was changed so much that an additional section was added to describe the new functionality.

Learn more
A good way to learn more about the OmniThreadLibrary is to go through the included demos. They are
part of the standard OmniThreadLibrary distribution and you should have them on the disk already if
you have installed OmniThreadLibrary. Each demo deals with a very limited subset of OmniThreadLibrary
functionality so they are fairly easy to understand.
From time to time I present OmniThreadLibrary in live webinars. Recordings are available on Gumroad¹⁶.
I frequently post about the OmniThreadLibrary on my blog where the articles relevant to OmniThreadLibrary
are specifically tagged with the ‘OmniThreadLibrary’ tag¹⁷.
¹²[Link]
¹³[Link]
¹⁴[Link]
¹⁵[Link]
¹⁶[Link]
¹⁷[Link]
Introduction v

Support is also available on StackOverflow¹⁸ (tag the question with ‘omnithreadlibrary’) and on the
OmniThreadLibrary forum¹⁹.
¹⁸[Link]
¹⁹[Link]
Release notes
2018-02-28

• Version 1.0 of the book released. It covers OmniThreadLibrary version 3.07.5.

2019-03-01

• Version 1.01 of the book released. It covers OmniThreadLibrary version 3.07.7.

• Numerous style and grammar errors were fixed.
1. Introduction to OmniThreadLibrary
OmniThreadLibrary¹ is a multi-threading library for Delphi, written mostly by the author of this book (see
Credits for a full list of contributors). OmniThreadLibrary can be roughly divided into three parts. Firstly,
there are building blocks that can be used either with the OmniThreadLibrary threading helpers or with any
other threading approach (f.i. with Delphi’s TThread² or with AsyncCalls³). Most of these building blocks
are described in chapter Miscellaneous, while some parts are covered elsewhere in the book (Lock-free
Collections, Blocking collection, Synchronization).
Secondly, OmniThreadLibrary brings low-level multithreading framework, which can be thought of as a
scaffolding that wraps the TThread class. This framework simplifies passing messages to and from the
background threads, starting background tasks, using thread pools and more. In some ways it is similar to
ITask⁴ which was introduced in Delphi XE7 except that OmniThreadLibrary’s implementation offers more
rounded feature set.
Thirdly, OmniThreadLibrary introduces high-level multithreading concept. High-level framework contains
multiple pre-packaged solutions (so-called abstractions) which can be used in your code. The idea is that the
user should just choose appropriate abstraction (Future, Pipeline, Map …) and write the worker code, while
the OmniThreadLibrary provides the framework that implements the tricky multi-threaded parts, takes care
of synchronisation and handles other menial tasks.

1.1 Requirements
OmniThreadLibrary requires at least Delphi 2007 and doesn’t work with FreePascal. The reason for this is
that most parts of OmniThreadLibrary use language constructs that are not yet supported by the FreePascal
compiler.
High-level multithreading framework requires at least Delphi 2009. Delphi XE or newer is recommended as
some parts of the framework aren’t supported in Delphi 2009 and 2010 due to compiler bugs.
OmniThreadLibrary only works in Windows applications. Both 32-bit and 64-bit platform are supported.
Applications can be compiled with the VCL library, as a service or as a console application. FireMonkey is
currently not supported.

1.2 License
OmniThreadLibrary is an open-sourced library with the OpenBSD license.
¹[Link]
²[Link]
³[Link]
⁴[Link]
Introduction to OmniThreadLibrary 2

This software is distributed under the BSD license.

Copyright (c) Primoz Gabrijelcic
All rights reserved.
Redistribution and use in source and binary forms, with or without modification, are permitted
provided that the following conditions are met:

• Redistributions of source code must retain the above copyright notice, this list of conditions
and the following disclaimer.
• Redistributions in binary form must reproduce the above copyright notice, this list of
conditions and the following disclaimer in the documentation and/or other materials
provided with the distribution.
• The name of the Primoz Gabrijelcic may not be used to endorse or promote products derived
from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

“AS IS” AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PAR-
TICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

In simpler words, the license says:

1. You can use the library in any project, free, open source or commercial, without having to mention my
name or the name of the library anywhere in your project, documentation or on the website.
2. You can change the source for your own use. You can also put a modified version on the web, but you
must not remove my name or the license from the source code.
3. I’m not guilty if the software blows in your face. Remember, you got OmniThreadLibrary for free.

In case your company would like to get a support contract for the OmniThreadLibrary, please contact me.

1.3 Installation
1. Download the last stable edition (download link is available at the OmniThreadLibrary site⁵), or
download the latest state from the repository⁶. Typically, it is safe to follow the repository trunk as
only tested code is committed.
⁵[Link]
⁶[Link]
Introduction to OmniThreadLibrary 3

2. If you have downloaded the last stable edition, unpack it to a folder.

3. Add the folder where you [unpacked last stable edition/checked out the SVN trunk] to the Delphi’s
Library path. Also add the src subfolder to the Library path. In case you are already using units from
my GpDelphiUnits⁷ project, you can ignore the copies in the src folder and use GpDelphiUnits version.
4. Add necessary units to the uses statement and start using the library!

1.3.1 Installing with GetIt

Delphi/RAD Studio XE8 and newer come with an integrated package manager called GetIt.
With GetIt you can install OmniThreadLibrary with just a few clicks. Start Delphi and open Tools, GetIt
Package Manager. Enter ‘omnithreadlibrary’ into the search bar. Then click on the INSTALL button under
the OmniThreadLibrary graphics.

GetIt will download OmniThreadLibrary from Embarcadero’s servers, add source path to the Library path
and compile and install the design package.
To find the demos, look at the Library path. It will contain something like this at the end: $(BDSCata-
logRepository)\OmniThreadLibrary_3.07.1-Tokyo\src\. To find the true path, look into Tools, Options,
Environment Options, Environment Variables where BDSCatalogRepository is defined.
⁷[Link]
Introduction to OmniThreadLibrary 4

Removing OmniThreadLibrary from Delphi is equally simple. Just open GetIt, select the Installed category
and click the UNINSTALL button under the OmniThreadLibrary graphics.

1.3.2 Installing with Delphinus

Delphinus is a 3ʳ party package manager for Delphi XE and newer, similar to Embarcadero’s own GetIt
package manager.
Unlike GetIt, which comes integrated into Delphi, you have to install Delphinus manually. Recommended
installation procedure is:

1. Stop Delphi/RAD Studio.

2. Install Delphinus with the web installer⁸.
a. It is recommended to right-click on the installer and select ‘Run as administrator’. Otherwise
Delphinus may not be able to create the installation folder.
3. Start Delphi/RAD Studio.
4. (Optional, but highly recommended) Open Tools, Delphinus, click the Settings icon and enter OAuth-
Token. This will prevent frequent GitHub ‘rate limitation’ errors during operation. Instructions for
generating the token can be found on the Delphinus wiki⁹.

Once Delphinus is installed, click the Tools, Delphinus and in the Delphinus window click the green Refresh
icon. When the package list is refreshed, enter ‘omnithreadlibrary’ into the search bar and press <Enter>. Click
on the OmniThreadLibrary list item to get additional description in the panel on the right.

To install OmniThreadLibrary, click the Install icon (blue arrow pointing downwards to the disk drive). Be
patient as Delphinus may take some time without displaying any progress on the screen.
⁸[Link]
⁹[Link]
Introduction to OmniThreadLibrary 5

When installation is complete, click the Show log button and in the log find the path where Om-
niThreadLibrary was installed (look for Adding libpathes message). Inside that folder you’ll also find all
OmniThreadLibrary demos.
K> You can find this path in Delphi’s Library path configuration setting.
Delphinus will compile and install appropriate package so everything is set up for you.
Removing OmniThreadLibrary from Delphi is equally simple. Just open Delphinus, select the Installed
category, select OmniThreadLibrary and click the Remove icon (red circle with white X).

1.3.3 Installing design package

OmniThreadLibrary includes one design-time component (TOmniEventMonitor) which may be used to receive
messages sent from the background tasks and to monitor thread creation/destruction. It is used in some demo
applications.
If you installed OmniThreadLibrary with GetIt or Delphinus the installation process has already installed the
design package and you may omit this step.
To compile and install the package containing this component, follow these steps:

• From Delphi, open packages subfolder of the OmniThreadLibrary installation and select file Om-
[Link] from the appropriate folder.
• In the Project Manager window you’ll find two projects – OmniThreadLibraryRuntime{VER}.bpl and
OmniThreadLibraryDesigntime{VER}.bpl (where {VER} is package version¹⁰ of your Delphi). If the
Project Manager window is not visible, select View, Project Manager from the menu.

• Right-click on the OmniThreadLibraryRuntime{VER}.bpl and select Build from the pop-up menu.
• Right-click on the OmniThreadLibraryDesigntime{VER}.bpl and select Build from the pop-up menu.
• Right-click again on the OmniThreadLibraryDesigntime{VER}.bpl and select Install from the pop-up
menu.
• Delphi will report that the TOmniEventMonitor component was installed.
¹⁰[Link]
Introduction to OmniThreadLibrary 6

• Close the project group with File, Close All. If Delphi asks you whether to save modified files, choose
No.

You should repeat these steps whenever the OmniThreadLibrary installation is updated.

1.4 Why use OmniThreadLibrary?

OmniThreadLibrary approaches the threading problem from a different perspective than TThread. While
the Delphi’s native approach is oriented towards creating and managing threads on a very low level, the
main design guideline behind OmniThreadLibrary is: “Enable the programmer to work with threads in as
fluent way as possible.” The code should ideally relieve you from all burdens commonly associated with
multi-threading.
OmniThreadLibrary was designed to become a “VCL for multi-threading” – a library that will make typical
multi-threading tasks really simple but still allow you to dig deeper and mess with the multi-threading code
at the operating system level. While still allowing this low-level tinkering, OmniThreadLibrary enables you
to work on a higher level of abstraction most of the time.
There are two important points of distinction between TThread and OmniThreadLibrary, both explained
further in this chapter. One is that OmniThreadLibrary focuses on tasks, not threads and another is that in
OmniThreadLibrary messaging tries to replace locking whenever possible.
By moving most of the critical multi-threaded code into reusable components (classes and high-level
abstractions), OmniThreadLibrary allows you to write better multithreaded code faster.

1.5 Tasks vs. threads

In OmniThreadLibrary you don’t create threads but tasks. A task can be executed in a new thread or in an
existing thread, taken from a thread pool.
A task is created using CreateTask function, which takes as a parameter a global procedure, a method, an
instance of a TOmniWorker class (or, usually, a descendant of that class) or an anonymous method (in Delphi
Introduction to OmniThreadLibrary 7

2009 and newer). CreateTask returns an IOmniTaskControl interface, which can be used to control the task.
A task is always created in a suspended state and you have to call Run to activate it (or Schedule to run it in
a thread pool).

The task has access to the IOmniTask interface and can use it to communicate with the owner (the part of the
program that started the task). Both interfaces are explained in detail in chapter Low-level multithreading.
The distinction between the task and the thread can be summarized in few simple words.

Task is part of code that has to be executed.

Thread is the execution environment.

You take care of the task, OmniThreadLibrary takes care of the thread.

1.6 Locking vs. messaging

I believe that locking is evil. It leads to slow code and deadlocks and is one of the main reasons for almost-
working multi-threaded code (especially when you use shared data and forget to lock it up). Because of that,
OmniThreadLibrary tries to move as much away from the shared data approach as possible. Cooperation
between threads is rather achieved with messaging.
If we compare shared data approach with messaging, both have good and bad sides. On the good side, shared
data approach is fast because it doesn’t move data around and is less memory intensive as the data is kept
only in one copy. On the bad side, locking must be used to access data which leads to bad scaling (slowdowns
when many threads are accessing the data), deadlocks and livelocks.

To see how easy it is to run into problems with locking, play the The Deadlock Empire¹¹ game.

¹¹[Link]
Introduction to OmniThreadLibrary 8

The situation is almost reversed for messaging. There’s no shared data so no locking, which makes the
program faster, more scalable and less prone to fall in the deadlocking trap. (Livelocking is still possible,
though.) On the bad side, it uses more memory, requires copying data around (which may be a problem if
shared data is large) and may lead to complicated and hard to understand algorithms.
OmniThreadLibrary uses custom lock-free structures to transfer data between the task and its owner (or
directly between two tasks). The system is tuned for high data rates and can transfer more than million
messages per second. However, in some situations shared data approach is necessary, and that’s why
OmniThreadLibrary adds significant support for synchronisation.

Some would disagree with the OmniThreadLibrary communication structures being called lock-free. In
reality, there are no communication mechanisms that would correctly work in a multi-threaded world
without using locking. The name lock-free only implies that no operating system locking primitives are being
used. Instead, OmniThreadLibrary achieves the thread-safeness by using bus locking, a special processor
instruction prefix that achieves atomic operation in a multiprocessor system. Bus-locked operations are
slower than the normal assembler code, especially as they may stop other cores for the time of the operation
execution, but then they are also faster than the operating system locking.

Lock-free (or micro-locked) structures in OmniThreadLibrary encompass:

• bounded (size-limited) stack

• bounded (size-limited) queue
• message queue
• dynamic (growing) queue
• blocking collection

OmniThreadLibrary automatically inserts two bounded queues between the task owner (IOmniTaskControl)
and the task (IOmniTask) so that the messages can flow in both directions.

1.7 Message loop required

Because of implementation details, OmniThreadLibrary requires that each thread owner maintains and
processes a message queue. This condition is automatically satisfied in VCL and service applications, but
running OmniThreadLibrary thread from a console application requires more work. Additional work is also
required if you are creating OmniThreadLibrary threads from background threads.
 Introduction to OmniThreadLibrary 9

1.7.1 OmniThreadLibrary and console

To correctly use OmniThreadLibrary in a console application, said application must process Windows
messages. This is demonstrated in the 62_console demo which is reproduced below.

1 program app_62_console;
2
3 {$APPTYPE CONSOLE}
4
5 uses
6 Windows, Messages, SysUtils,
7 OtlComm, OtlTask, OtlTaskControl, OtlParallel;
8
9 const
10 MSG_STATUS = WM_USER;
11
12 procedure ProcessMessages;
13 var
14 Msg: TMsg;
15 begin
16 while integer(PeekMessage(Msg, 0, 0, 0, PM_REMOVE)) <> 0 do begin
17 TranslateMessage(Msg);
18 DispatchMessage(Msg);
19 end;
20 end;
21
22 function DoTheCalculation(const task: IOmniTask): integer;
23 var
24 i: integer;
25 begin
26 for i := 1 to 5 do begin
27 [Link](MSG_STATUS, '... still calculating');
28 Sleep(1000);
29 end;
30 Result := 42;
31 end;
32
33 var
34 calc: IOmniFuture<integer>;
35
36 begin
37 try
38 calc := [Link]<integer>(DoTheCalculation,
39 [Link](MSG_STATUS,
40 procedure(const task: IOmniTaskControl; const msg: TOmniMessage)
41 begin
 Introduction to OmniThreadLibrary 10

42 Writeln([Link]);
43 end));
44
45 Writeln('Background thread is calculating ...');
46 while not [Link] do
47 ProcessMessages;
48 Writeln('And the answer is: ', [Link]);
49
50 if DebugHook <> 0 then
51 Readln;
52 except
53 on E: Exception do
54 Writeln([Link], ': ', [Link]);
55 end;
56 end.

The main program creates a Future which runs a function DoTheCalculation and then waits for it to return
a value (while not [Link]).
The future waits five seconds, and each second sends a message back to the owner ([Link](MSG_-
STATUS, '... still calculating')). Main thread processes these messages in the MSG_STATUS handler.

If you run the program, you’ll see that the message “... still calculating” is displayed five times with one
second delay between two messages. After that, “And the answer is: 42” is displayed.
The critical part of this program are two lines:

1 while not [Link] do

2 ProcessMessages;

If you comment them out, the program will write “Background thread is calculating ...”, then nothing
will happen (visibly) for five seconds and then all messages will be displayed at once.
In this example it is not critical to process messages. The program would continue to function correctly even
when message processing is removed. In other cases, however, all kinds of weird behaviour can occur if
messages are not processed. OmniThreadLibrary occasionally uses messages for internal purpose and if you
prevent processing of these messages, applications may misbehave. The best approach is to always include
periodic calls to ProcessMessages in a console application.

1.7.2 OmniThreadLibrary task started from another task

Similar considerations take order when an OmniThreadLibrary task is started from another OmniThreadLi-
brary task. The intermediate task (the task which starts another task) must process messages. The easiest way
to achieve that is by using the MsgWait qualifier when creating a task.
In the 66_ThreadsInThreads demo, the click on the OTL from an OTL task button creates a task that will
create a Future.
 Introduction to OmniThreadLibrary 11

1 FOwnerTask := CreateTask([Link](), 'OTL owner')

2 .OnMessage(Self)
3 .OnTerminated(TaskTerminated)
4 .MsgWait // critical, this allows OTL task to process messages
5 .Run;

The important part of this demo is the call to MsgWait which causes internal loop in the task to process
Windows messages. Without this MsgWait the program would stop working.
The worker does all the work in its Initialization method.

1 function [Link]: boolean;

2 begin
3 Result := inherited Initialize;
4 if Result then begin
5 [Link](WM_LOG, Format('[%d] Starting a Future', [GetCurrentThreadID]));
6 FCalc := [Link]<integer>(Asy_DoTheCalculation,
7 [Link]
8 .OnMessage(MSG_STATUS,
9 procedure(const workerTask: IOmniTaskControl; const msg: TOmniMessage)
10 begin
11 // workerTask = task controller for [Link] worker thread
12 // Task = [Link] = interface of the TWorker task
13 [Link](WM_LOG, Format('[%d] Future sent a message: %s',
14 [GetCurrentThreadID, [Link]]));
15 end)
16 .OnTerminated(
17 procedure
18 begin
19 [Link](WM_LOG, Format('[%d] Future terminated, result = %d',
20 [GetCurrentThreadID, [Link]]));
21 FCalc := nil;
22 [Link](WM_LOG, Format('[%d] Terminating worker',
23 [GetCurrentThreadID]));
24 // Terminate TWorker
25 [Link];
26 end));
27 end;
28 end;

This code executes in a background worker thread. It may look complicated, however, the code simply
creates a Future calculation (FCalc := [Link]<integer>) and sets up event handlers that will
process messages sent from the future (.OnMessage) and handle the completion of the future calculation
(.OnTerminated).
OnMessagetakes a message that was sent from the future, adds some text and current thread ID and sends
new message to the form where it is logged in the WMLog method (not shown here).
 Introduction to OmniThreadLibrary 12

OnTerminated also logs the event, clears the future interface and terminates self ([Link]). After that,
form’s TaskTerminated method (not shown here) is called and cleans the task controller interface.
The future itself does nothing special, it simply sends five messages with one second delay between them and
then returns a value.

1 function TWorker.Asy_DoTheCalculation(const task: IOmniTask): integer;

2 var
3 i: integer;
4 begin
5 for i := 1 to 5 do begin
6 [Link](MSG_STATUS, Format('[%d] ... still calculating',
7 [GetCurrentThreadID]));
8 Sleep(1000);
9 end;
10 Result := 42;
11 end;

When you run the program and click on the button, following text will be displayed (thread IDs – numbers
in brackets – will be different in your case).

We can see that Future messages were generated in thread 18420, then passed through the parent thread 6088
and ended in the main thread 11968.

1.7.3 OmniThreadLibrary task started from a TThread

Enhancing a basic Delphi TThread is easy with the OmniThreadLibrary and takes only a few simple steps. We
have to make sure that any thread messages are periodically processed by calling the DSiProcessThreadMessages
function from the DSiWin32 unit (or a similar code that calls PeekMessage / TranslateMessage / DispatchMessage).
The 66_ThreadsInThreads demo contains an example.
 Introduction to OmniThreadLibrary 13

1 FThread := [Link](true);
2 [Link] := ThreadTerminated;
3 [Link] := true;
4 [Link];

The main thread method firstly creates a Future and sets up an .OnMessage handler which just resends
messages to the main thread.

1 procedure [Link];
2 var
3 awaited: DWORD;
4 calc : IOmniFuture<integer>;
5 handles: array [0..0] of THandle;
6 begin
7 Log('Starting a Future');
8
9 calc := [Link]<integer>(Asy_DoTheCalculation,
10 [Link]
11 .OnMessage(MSG_STATUS,
12 procedure(const workerTask: IOmniTaskControl; const msg: TOmniMessage)
13 begin
14 Log('Future sent a message: ' + [Link]);
15 end));
16
17 repeat
18 awaited := MsgWaitForMultipleObjects(0, handles, false, INFINITE, QS_ALLPOSTMESSAGE);
19 if awaited = WAIT_OBJECT_0 + 0 {handle count} then
20 DSiProcessThreadMessages;
21 until [Link];
22
23 Log('Future terminated, result = ' + IntToStr([Link]));
24 calc := nil;
25 Log('Terminating worker');
26 end;

Then it enters a repeat .. until loop in which it waits for a Windows message (MsgWaitForMultipleObjects),
processes all waiting messages (DSiProcessThreadMessages) and checks whether the calculation has com-
pleted ([Link]).
At the end it cleans up the future and exits. That destroys the TWorkerThread thread.
Calling MsgWaitForMultipleObjects¹² is not strictly necessary. You could just call DSiProcessThreadMessages
from time to time. It does, however, improve the performance as the code uses no CPU time in such wait.
If you are already using some other kind of wait-and-dispach mechanism in your thread (WaitForSingleObject,
WaitForSingleObjectEx, WaitForMultipleObjects, WaitForMultipleObjectsEx) then they are easy to convert
to MsgWaitForMultipleObjects or MsgWaitForMultipleObjectsEx.
¹²[Link]
Introduction to OmniThreadLibrary 14

Running the program shows a similar behaviour as in the previous example.

1.8 TOmniValue
A TOmniValue (part of the OtlCommon unit) is a data type central to the whole OmniThreadLibrary. It is
used in all parts of the code (for example in a communication subsystem) when type of the data that is to be
stored/passed around is not known in advance.
It is implemented as a smart record (a record with functions and operators) which functions similarly to a
Variant¹³ or TValue¹⁴ but is faster¹⁵. It can store following data types:

• simple values (byte, integer, char, double, …)

• strings (Ansi, Unicode)
• Variant
• objects
• interfaces
• records (in D2009 and newer)

In all cases ownership of reference-counted data types (strings, interfaces) is managed correctly so no memory
leaks can occur when such a type is stored in a TOmniValue variable.

Two kinds of floating-point numbers can be stored in a TOmniValue – double and extended. Former
are stored directly in the TOmniValue record while latter are wrapped in an TOmniExtendedData
record, which increases memory usage and decreases performance. The use of double floating-point
numbers is therefore recommended.

The TOmniValue type is too large to be shown in one piece so I’ll show various parts of its interface throughout
this chapter.
¹³[Link]
¹⁴[Link]
¹⁵[Link]
 Introduction to OmniThreadLibrary 15

1.8.1 Data access

The content of a TOmniValue record can be accessed in many ways, the simplest (and in most cases the most
useful) being through the AsXXX properties.

1 property AsAnsiString: AnsiString;

2 property AsBoolean: boolean;
3 property AsCardinal: cardinal;
4 property AsDouble: Double;
5 property AsDateTime: TDateTime;
6 property AsException: Exception;
7 property AsExtended: Extended;
8 property AsInt64: int64 read;
9 property AsInteger: integer;
10 property AsInterface: IInterface;
11 property AsObject: TObject;
12 property AsOwnedObject: TObject;
13 property AsPointer: pointer;
14 property AsString: string;
15 property AsVariant: Variant;
16 property AsWideString: WideString;

Exceptions can be stored through the AsObject property, but there’s also a special support for
Exception data type with its own data access property AsException. It is extensively used in the
Pipeline abstraction.

While the setters for those properties are pretty straightforward, getters all have a special logic built in which
tries to convert data from any reasonable source type to the requested type. If that cannot be done, an
exception is raised.
For example, the getter for the AsString property is called CastToString. Internally it calls TryCastToString,
which is a public function of TOmniValue.

1 function [Link]: string;

2 begin
3 if not TryCastToString(Result) then
4 raise [Link]('TOmniValue cannot be converted to string');
5 end;
6
7 function [Link](var value: string): boolean;
8 begin
9 Result := true;
10 case ovType of
11 ovtNull: value := '';
12 ovtBoolean: value := BoolToStr(AsBoolean, true);
 Introduction to OmniThreadLibrary 16

13 ovtInteger: value := IntToStr(ovData);

14 ovtDouble,
15 ovtDateTime,
16 ovtExtended: value := FloatToStr(AsExtended);
17 ovtAnsiString: value := string((ovIntf as IOmniAnsiStringData).Value);
18 ovtString: value := (ovIntf as IOmniStringData).Value;
19 ovtWideString: value := (ovIntf as IOmniWideStringData).Value;
20 ovtVariant: value := string(AsVariant);
21 else Result := false;
22 end;
23 end;

When you don’t know the data type stored in a TOmniValue variable and you don’t want to raise an exception
if compatible data is not available, you can use the TryCastToXXX family of functions directly.

1 function TryCastToAnsiString(var value: AnsiString): boolean;

2 function TryCastToBoolean(var value: boolean): boolean;
3 function TryCastToCardinal(var value: cardinal): boolean;
4 function TryCastToDouble(var value: Double): boolean;
5 function TryCastToDateTime(var value: TDateTime): boolean;
6 function TryCastToException(var value: Exception): boolean;
7 function TryCastToExtended(var value: Extended): boolean;
8 function TryCastToInt64(var value: int64): boolean;
9 function TryCastToInteger(var value: integer): boolean;
10 function TryCastToInterface(var value: IInterface): boolean;
11 function TryCastToObject(var value: TObject): boolean;
12 function TryCastToPointer(var value: pointer): boolean;
13 function TryCastToString(var value: string): boolean;
14 function TryCastToVariant(var value: Variant): boolean;
15 function TryCastToWideString(var value: WideString): boolean;

Alternatively, you can use CastToXXXDef functions which return a default value if current value of the
TOmniValue cannot be converted into required data type.

1 function CastToAnsiStringDef(const defValue: AnsiString): AnsiString;

2 function CastToBooleanDef(defValue: boolean): boolean;
3 function CastToCardinalDef(defValue: cardinal): cardinal;
4 function CastToDoubleDef(defValue: Double): Double;
5 function CastToDateTimeDef(defValue: TDateTime): TDateTime;
6 function CastToExceptionDef(defValue: Exception): Exception;
7 function CastToExtendedDef(defValue: Extended): Extended;
8 function CastToInt64Def(defValue: int64): int64;
9 function CastToIntegerDef(defValue: integer): integer;
10 function CastToInterfaceDef(const defValue: IInterface): IInterface;
11 function CastToObjectDef(defValue: TObject): TObject;
12 function CastToPointerDef(defValue: pointer): pointer;
 Introduction to OmniThreadLibrary 17

13 function CastToStringDef(const defValue: string): string;

14 function CastToVariantDef(defValue: Variant): Variant;
15 function CastToWideStringDef(defValue: WideString): WideString;

They are all implemented in the same manner, similar to the CastToObjectDef below.

1 function [Link](defValue: TObject): TObject;

2 begin
3 if not TryCastToObject(Result) then
4 Result := defValue;
5 end;

Function LogValue [3.07.6] returns a string containing both the type of the stored data and stored value.

1 function LogValue: string;

This function is useful for data logging and debugging. See the source code for details.

1.8.2 Type testing

For situations where you would like to determine the type of data stored inside the TOmniValue, there is the
IsXXX family of functions.

1 function IsAnsiString: boolean;

2 function IsArray: boolean;
3 function IsBoolean: boolean;
4 function IsEmpty: boolean;
5 function IsException: boolean;
6 function IsFloating: boolean;
7 function IsDateTime: boolean;
8 function IsInteger: boolean;
9 function IsInterface: boolean;
10 function IsInterfacedType: boolean;
11 function IsObject: boolean;
12 function IsOwnedObject: boolean;
13 function IsPointer: boolean;
14 function IsRecord: boolean;
15 function IsString: boolean;
16 function IsVariant: boolean;
17 function IsWideString: boolean;

Alternatively, you can use the DataType property.

 Introduction to OmniThreadLibrary 18

1 type
2 TOmniValueDataType = (ovtNull, ovtBoolean, ovtInteger, ovtDouble, ovtObject,
3 ovtPointer, ovtDateTime, ovtException, ovtExtended, ovtString, ovtInterface,
4 ovtVariant, ovtWideString, ovtArray, ovtRecord, ovtAnsiString, ovtOwnedObject);
5
6 property DataType: TOmniValueDataType;

1.8.3 Clearing the content

There are two ways to clear a TOmniValue – you can either call its Clearmethod, or you can assign to it a
[Link].

1 procedure Clear;
2 class function Null: TOmniValue; static;

An example:

1 var
2 ov: TOmniValue;
3
4 [Link];
5 // or
6 ov := [Link];

Calling Clear is slightly faster.

1.8.4 Operators
TOmniValue also implements several Implicit operators which help with automatic conversion to and from
different data types. Internally, they are implemented as an assignment to/from the AsXXX property.

1 class operator Equal(const a: TOmniValue; i: integer): boolean;

2 class operator Equal(const a: TOmniValue; const s: string): boolean;
3 class operator Implicit(const a: AnsiString): TOmniValue;
4 class operator Implicit(const a: boolean): TOmniValue;
5 class operator Implicit(const a: Double): TOmniValue;
6 class operator Implicit(const a: Extended): TOmniValue;
7 class operator Implicit(const a: integer): TOmniValue;
8 class operator Implicit(const a: int64): TOmniValue;
9 class operator Implicit(const a: pointer): TOmniValue;
10 class operator Implicit(const a: string): TOmniValue;
11 class operator Implicit(const a: IInterface): TOmniValue;
12 class operator Implicit(const a: TObject): TOmniValue;
13 class operator Implicit(const a: Exception): TOmniValue;
 Introduction to OmniThreadLibrary 19

14 class operator Implicit(const a: TOmniValue): AnsiString;

15 class operator Implicit(const a: TOmniValue): int64;
16 class operator Implicit(const a: TOmniValue): TObject;
17 class operator Implicit(const a: TOmniValue): Double;
18 class operator Implicit(const a: TOmniValue): Exception;
19 class operator Implicit(const a: TOmniValue): Extended;
20 class operator Implicit(const a: TOmniValue): string;
21 class operator Implicit(const a: TOmniValue): integer;
22 class operator Implicit(const a: TOmniValue): pointer;
23 class operator Implicit(const a: TOmniValue): WideString;
24 class operator Implicit(const a: TOmniValue): boolean;
25 class operator Implicit(const a: TOmniValue): IInterface;
26 class operator Implicit(const a: WideString): TOmniValue;
27 class operator Implicit(const a: Variant): TOmniValue;
28 class operator Implicit(const a: TDateTime): TOmniValue;
29 class operator Implicit(const a: TOmniValue): TDateTime;

Implicit conversion to/from TDateTime is supported only in Delphi XE and newer.

Two Equal operators simplify comparing TOmniValue to an integer and string data.

1.8.5 Using with generic types

Few methods simplify using TOmniValue with class and record data.

1 class function CastFrom<T>(const value: T): TOmniValue; static;

2 function CastTo<T>: T;
3 function CastToObject<T: class>: T; overload;
4 function ToObject<T: class>: T;
5 class function Wrap<T>(const value: T): TOmniValue; static;
6 function Unwrap<T>: T;

CastFrom<T> converts any type into a TOmniValue. In Delphi 2009, this function is severely limited as only
simple types (integer, object) are supported. Starting with Delphi 2010, TValue type is used to facilitate the
conversion and all data types supported by the TOmniValue can be converted.
CastTo<T> converts TOmniValue into any other type. In Delphi 2009 same limitations apply as for CastFrom<T>.
Since [3.07.7] CastTo<T> supports casting into an interface on Delphi XE3 and newer.
CastToObject<T> (available in Delphi 2010 and newer) performs a hard cast with no type checking. It is
equivalent to using T([Link])
ToObject<T> (available in Delphi 2010 and newer) casts the object to type T with type checking. It is equivalent
to using [Link] as T.
Wrap<T> [3.06]
wraps any data type in an instance of TOmniRecordWrapper<T> and stores this value in a
TOmniValue variable.
 Introduction to OmniThreadLibrary 20

Unwrap<T> [3.06] unwraps a TOmniValue holding a TOmniRecordWrapper<T> and returns owned value of type T.
It has to be called in this form: [Link]<T>(). Trailing () is required.
Wrap and Unwrap are especially useful as they allow you to store TMethod data (event handlers) in a TOmniValue
variable.

1.8.6 Array access

Each TOmniValue can contain an array of other TOmniValues. Internally, they are stored in a TOmniValueContainer
object. This object can be accessed directly by reading the AsArray property.

1 property AsArray: TOmniValueContainer read GetAsArray;

IsArray can be used to test whether a TOmniValue contains an array of values.

Arrays can be accessed by an integer indexes (starting with 0), or by string indexes (named access). Integer-
indexed arrays are created by calling [Link] and string-indexed arrays are created by calling
[Link].

1 constructor Create(const values: array of const);

2 constructor CreateNamed(const values: array of const);

In the latter case, elements of the values parameter must alternate between names (string indexes) and values.

1 ov := [Link](
2 ['Key1', 'Value of ov[''Key1'']',
3 'Key2', 'Value of ov[''Key2'']'
4 ]);

In the example above, both ov[0] and ov['Key1'] would return the same string, namely 'Value of
ov[''Key1'']'.

Array elements can be accessed with the AsArrayItem property, by using an integer index (for integer-indexed
arrays), a string index (for string-indexed arrays), or a TOmniValue index. In the last case, the type of data
stored inside the TOmniValue index parameter will determine how the array element is accessed. This last
form is not available in Delphi 2007, where AsArrayItemOV should be used instead.
All forms of AsArrayItem allow extending an array. If you write data into an index which doesn’t already
exist, the array will automatically grow to accomodate the new value.

1 property AsArrayItem[idx: integer]: TOmniValue; default;

2 property AsArrayItem[const name: string]: TOmniValue; default;
3 property AsArrayItem[const param: TOmniValue]: TOmniValue; default;
4 property AsArrayItemOV[const param: TOmniValue]: TOmniValue;

If you want to test whether an array element exists, use the HasArrayItem function.
 Introduction to OmniThreadLibrary 21

1 function HasArrayItem(idx: integer): boolean; overload;

2 function HasArrayItem(const name: string): boolean; overload;
3 function HasArrayItem(const param: TOmniValue): boolean; overload;

Starting with Delphi 2010 TOmniValue also implements functions for converting data to and from TArray<T>
for any supported type. CastFrom<T> and CastTo<T> functions are used internally to do the conversion.

1 class function FromArray<T>(const values: TArray<T>): TOmniValue; static;

2 function ToArray<T>: TArray<T>;

1.8.7 Handling records

A record T can be stored inside a TOmniValue by calling the FromRecord<T> function. To extract the data back
into a record, use the ToRecord<T> function.

1 class function FromRecord<T: record>(const value: T): TOmniValue; static;

2 function ToRecord<T>: T;

An example:

1 var
2 ts: TTimeStamp;
3 ov: TOmniValue;
4
5 ov := [Link]<TTimeStamp>(ts);
6 ts := [Link]<TTimeStamp>;

TOmniValue jumps through quite some hoops to store a record. It is first converted into a TOmniRecordWrapper
which is then wrapped inside an IOmniAutoDestroyObject interface to provide a reference-counted lifetime
management.
Because of that convoluted process, storing records inside TOmniValue is not that fast.

1 class function [Link]<T>(const value: T): TOmniValue;

2 begin
3 [Link](
4 CreateAutoDestroyObject(
5 TOmniRecordWrapper<T>.Create(value)));
6 end;

1.8.8 Object ownership

TOmniValue can take an ownership of a TObject-type data. To achieve that, you can either assign an object to
the AsOwnedObject property or set the OwnsObject property to True.
 Introduction to OmniThreadLibrary 22

1 property AsOwnedObject: TObject;

2 function IsOwnedObject: boolean;
3 property OwnsObject: boolean;

When an object-owning TOmniValue goes out of scope, it automatically destroys the owned object.
You can change the ownership status at any time by setting the OwnsObject property.

1.8.9 Working with TValue

Starting with Delphi 2010 TOmniValue provides an AsTValue property and corresponding Implicit operator
so you can easily convert a TValue data into a TOmniValue and back.

1 class operator Implicit(const a: TValue): TOmniValue; inline;

2 class operator Implicit(const a: TOmniValue): TValue; inline;
3 property AsTValue: TValue;

1.8.10 Low-level methods

For programmers with special requirements (and for internal OmniThreadLibrary use), TOmniValue exposes
following public methods.

1 procedure _AddRef;
2 procedure _Release;
3 procedure _ReleaseAndClear;
4 function RawData: PInt64;
5 procedure RawZero;

_AddRef increments reference counter of stored data if TOmniValue contains such data.
_Release decrements reference counter of stored data if TOmniValue contains such data.
_ReleaseAndClear is just a shorthand for calling a _Release followed by a call to RawZero.
RawData returns a pointer to the data stored in the TOmniValue.
RawZero clears the stored data without decrementing the reference counter.

1.9 TOmniValueObj
The OtlCommon unit implements a simple object which can wrap a TOmniValue for situations where you
would like to store it inside a data structure that only supports object types.
 Introduction to OmniThreadLibrary 23

1 TOmniValueObj = class
2 constructor Create(const value: TOmniValue);
3 property Value: TOmniValue read FValue;
4 end;

1.10 Fluent interfaces

OmniThreadLibrary heavily uses fluent interface¹⁶ approach. Most of the functions in OmniThreadLibrary
interfaces are returning Self as the result. Take for example this declaration of the Pipeline abstraction,
slightly edited for brevity.

1 IOmniPipeline = interface
2 procedure Cancel;
3 function From(const queue: IOmniBlockingCollection): IOmniPipeline;
4 function HandleExceptions: IOmniPipeline;
5 function NumTasks(numTasks: integer): IOmniPipeline;
6 function OnStop(const stopCode: TProc): IOmniPipeline;
7 function Run: IOmniPipeline;
8 function Stage(
9 pipelineStage: TPipelineSimpleStageDelegate;
10 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
11 function Stage(
12 pipelineStage: TPipelineStageDelegate;
13 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
14 function Stage(
15 pipelineStage: TPipelineStageDelegateEx;
16 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
17 function Stages(
18 const pipelineStages: array of TPipelineSimpleStageDelegate;
19 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
20 function Stages(
21 const pipelineStages: array of TPipelineStageDelegate;
22 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
23 function Stages(
24 const pipelineStages: array of TPipelineStageDelegateEx;
25 taskConfig: IOmniTaskConfig = nil): IOmniPipeline; overload;
26 function Throttle(numEntries: integer; unblockAtCount: integer = 0):
27 IOmniPipeline;
28 function WaitFor(timeout_ms: cardinal): boolean;
29 end;

As you can see, most of the functions return the IOmniPipeline interface. In code, this is implemented by
returning Self.
¹⁶[Link]
 Introduction to OmniThreadLibrary 24

1 function [Link](
2 const queue: IOmniBlockingCollection): IOmniPipeline;
3 begin
4 opInput := queue;
5 Result := Self;
6 end;

This allows calls to such interfaces to be chained. For example, the following code from the Pipeline section
of the book shows how to use [Link] without ever storing the resulting interface in a variable.

1 var
2 sum: integer;
3
4 sum := [Link]
5 .Stage(
6 procedure (const input, output: IOmniBlockingCollection)
7 var
8 i: integer;
9 begin
10 for i := 1 to 1000000 do
11 [Link](i);
12 end)
13 .Stage(
14 procedure (const input: TOmniValue; var output: TOmniValue)
15 begin
16 output := [Link] * 3;
17 end)
18 .Stage(
19 procedure (const input, output: IOmniBlockingCollection)
20 var
21 sum: integer;
22 value: TOmniValue;
23 begin
24 sum := 0;
25 for value in input do
26 Inc(sum, value);
27 [Link](sum);
28 end)
29 .[Link];

If you don’t like fluent interface approach, don’t worry. OmniThreadLibrary can be used without it. You can
always call a function as if it is a procedure and compiler will just throw away the result.
The example above could be rewritten as such:
 Introduction to OmniThreadLibrary 25

1 var
2 sum: integer;
3 pipe: IOmniPipeline;
4
5 pipe := [Link];
6 [Link](
7 procedure (const input, output: IOmniBlockingCollection)
8 var
9 i: integer;
10 begin
11 for i := 1 to 1000000 do
12 [Link](i);
13 end);
14 [Link](
15 procedure (const input: TOmniValue; var output: TOmniValue)
16 begin
17 output := [Link] * 3;
18 end);
19 [Link](
20 procedure (const input, output: IOmniBlockingCollection)
21 var
22 sum: integer;
23 value: TOmniValue;
24 begin
25 sum := 0;
26 for value in input do
27 Inc(sum, value);
28 [Link](sum);
29 end);
30 [Link];
31 sum := [Link];
2. High-level multi-threading
Face it – multi-threading programming is hard. It is hard to design a multi-threaded program, it is hard to
write and test it and it is insanely hard to debug it. To ease this problem, OmniThreadLibrary introduces
several pre-packaged multi-threading solutions; so-called abstractions.
The idea behind the high-level abstractions is that the user should just choose appropriate abstraction and
write the worker code, while the OmniThreadLibrary provides the framework that implements the tricky
multi-threaded parts, takes care of synchronisation and so on.
 High-level multi-threading 27

2.1 Async
Async is the simplest of high-level abstractions and is typically used for fire and forget scenarios. To create
an Async task, call [Link].

When you call [Link], code is started in a new thread (indicated by the bold vertical line) and both
main and background threads continue execution. After some time, background task completes execution and
disappears.

See also demo 46_Async.

Example:

1 [Link](
2 procedure
3 begin
4 MessageBeep($FFFFFFFF);
5 end);

This simple program creates a background task with a sole purpose to make some noise. The task is coded as
an anonymous method but you can also use a normal method or a normal procedure for the task code.
The Parallel class defines two Async overloads. The first accepts a parameter-less background task and an
optional task configuration block and the second accepts a background task with an IOmniTask parameter
and an optional task configuration block.
 High-level multi-threading 28

1 type
2 TOmniTaskDelegate = reference to procedure(const task: IOmniTask);
3
4 Parallel = class
5 class procedure Async(task: TProc;
6 taskConfig: IOmniTaskConfig = nil); overload;
7 class procedure Async(task: TOmniTaskDelegate;
8 taskConfig: IOmniTaskConfig = nil); overload;
9 ...
10 end;

The second form is useful if the background code needs access to the IOmniTask interface, for example, to
send messages to the owner or to execute code in the owner thread (typically that will be the main thread).
The example below uses Async task to fetch the contents of a web page (by calling a mysterious function
HttpGet) and then uses Invoke to execute a code that logs the length of the result in the main thread.

1 [Link](
2 procedure (const task: IOmniTask)
3 var
4 page: string;
5 begin
6 HttpGet('[Link]', 80, '[Link]', page, '');
7 [Link](
8 procedure
9 begin
10 [Link](Format('Async GET: %d ms; page length = %d',
11 [time, Length(page)]))
12 end);
13 end);

The same result could be achieved by sending a message from the background thread to the main thread. In
the example below, TaskConfig block is used to configure message handler.

1 const
2 WM_RESULT = WM_USER;
3
4 procedure LogResult(const task: IOmniTaskControl; const msg: TOmniMessage);
5 begin
6 [Link](Format('Async GET: %d ms; page length = %d',
7 [time, Length(page)]))
8 end;
9
10 [Link](
11 procedure (const task: IOmniTask)
12 var
 High-level multi-threading 29

13 page: string;
14 begin
15 HttpGet('[Link]', 80, '[Link]', page, '');
16 [Link](WM_RESULT, page);
17 end,
18 [Link](WM_RESULT, LogResult)
19 );

Let me warn you that in cases where you want to return a result from a background task, Async abstraction
is not the most appropriate. You would be better off using a Future.

2.1.1 Handling exceptions

If the background code raises an unhandled exception, OmniThreadLibrary catches this exception and re-
raises it in the OnTerminated handler. This way the exception travels from the background thread to the
owner thread where it can be processed.
As the OnTerminated handler executes at an unspecified moment when Windows are processing window
messages, there is no good way to catch this message with a try..except block. The caller must install its own
OnTerminated handler instead and handle exception there.

The following example uses OnTerminated handler to detach fatal exception from the task, log the exception
details and destroy the exception object.

1 [Link](
2 procedure
3 begin
4 Sleep(1000);
5 raise [Link]('Exception in Async');
6 end,
7 [Link](
8 procedure (const task: IOmniTaskControl)
9 var
10 excp: Exception;
11 begin
12 if assigned([Link]) then begin
13 excp := [Link];
14 Log('Caught async exception %s:%s',[[Link], [Link]]);
15 FreeAndNil(excp);
16 end;
17 end
18 ));

If you don’t install an OnTerminated handler, an exception will be handled by the application-level filter,
which will by default cause a message box to appear.
See also demo 48_OtlParallelExceptions.
 High-level multi-threading 30

2.2 Async/Await
Async/Await is a simplified version of the Async abstraction which mimics the .NET Async/Await¹
mechanism.²
Async/Await accepts two parameter less anonymous methods. The first one is executed in a background
thread and the second one is executed in the main thread after the background thread has completed its
work.
See also demo 53_AsyncAwait.
Using Async/Await you can, for example, create a background operation which is triggered by a click and
which re-enables button after the background job has been completed.

1 procedure TForm1.Button1Click(Sender: TObject);

2 var
3 button: TButton;
4 begin
5 button := Sender as TButton;
6 [Link] := 'Working ...';
7 [Link] := false;
8 Async(
9 // executed in a background thread
10 procedure begin
11 Sleep(5000);
12 end).
13 Await(
14 // executed in the main thread after
15 // the anonymous method passed to
16 // Async has completed its work
17 procedure begin
18 [Link] := true;
19 [Link] := 'Done!';
20 end);
21 end;

Keep in mind that Async is invoked by calling [Link] and Async/Await by calling Async.

Exceptions in the Async part are currently not handled by the OmniThreadLibrary.

¹[Link]
²Read more at [Link]
High-level multi-threading 31

2.3 ForEach
ForEach abstraction creates a parallel for loop that iterates over a range of data (number range, list, queue,
dataset …) in multiple threads. To create a ForEach abstraction, call [Link].

When you use [Link], OmniThreadLibrary starts multiple background tasks and connects them to
the source through a serialization mechanism. Output is optionally sorted in the order of the input. By default,
ForEach waits for all background threads to complete before the control is returned to the caller.

See also demos 35_ParallelFor and 36_OrderedFor.

OmmiThreadLibrary also implements Parallel for abstraction, which is more limited than ForEach
but runs faster.
 High-level multi-threading 32

Example:

1 [Link] := 0;
2 [Link](1, 1000000).Execute(
3 procedure (const value: integer)
4 begin
5 if IsPrime(value) then
6 [Link];
7 end;
8 end);

This simple program calculates the number of prime numbers in the range from one to one million. The
PrimeCount object must be capable of atomic increment (a thread-safe increment), which is simple to achieve
with the use of the TOmniAlignedInt32 record. The ForEach task is coded as an anonymous method but you
can also use a normal method or a normal procedure for the task code.

2.3.1 Cooperation

The main point of the ForEach abstraction is cooperation between the parallel tasks. ForEach goes to great
lengths to minimize potential clashes between threads when they access the source data. Except in special
occasions (number range, IOmniBlockingCollection), source data is not thread-safe and locking must be used
to synchronize access.
To minimize this locking, source data is allocated to worker tasks in blocks. ForEach creates a source provider
object which accesses the source data in a thread-safe manner. This source provider makes sure to always
return an appropriately sized block of source data (size will depend on the number of tasks, type of the source
data and other factors) when a task runs out of data to process.
Because the source data is allocated in blocks, it is possible that one task runs out of work while other tasks
are still busy. In this case, a task will steal data from one of the other tasks. This approach makes all tasks as
busy as possible while minimizing the contention.
The details of this process are further discussed in section Internals below.

2.3.2 Iterating over …

The Parallel class defines many ForEach overloads, each supporting different container type. We will look
at them in more detail in the following sections.

[Link] … Number ranges

To iterate over a range, pass first and last index to the ForEach call. Optionally, you can pass a step
parameter, which defaults to 1. ForEach will then iterate from first to last with a step increment.
 High-level multi-threading 33

1 class function ForEach(low, high: integer; step: integer = 1):

2 IOmniParallelLoop<integer>; overload;

The pseudo-code for numeric ForEach could be written as

1 i := low;
2 while ((step > 0) and (i <= high)) or
3 ((step < 0) and (i >= high)) do
4 begin
5 // process 'i' in parallel
6 if low < high then Inc(i, step)
7 else Dec(i, step);
8 end;

[Link] … Enumerable collections

If you want to iterate over a collection (say, a TStringList), you have two possibilities.
One is to use an equivalent of for i := 0 to [Link]-1 do Something(sl[i]).

1 [Link](0, [Link]-1).Execute(
2 procedure (const value: integer)
3 begin
4 Something(sl[value]);
5 end);

Another is to use an equivalent of for s in sl do Something(s).

1 [Link](sl).Execute(
2 procedure (const value: TOmniValue)
3 begin
4 Something(value);
5 end);

In the second example, value is passed to the task function as a TOmniValue parameter. In the example above,
it will be automatically converted into a string, but sometimes you’ll have to do it manually, by calling
[Link] (or use appropriate casting function when iterating over a different container).

A variation of the second approach is to tell the ForEach that the container contains strings. OmniThreadLi-
brary will then do the conversion for you.
 High-level multi-threading 34

1 [Link]<string>(sl).Execute(
2 procedure (const value: string)
3 begin
4 Something(value);
5 end);

You may wonder which of those approaches is better. The answer depends on whether you can simultaneously
access different items in the container from different threads at the same time. In other words, you have
to know whether the container is thread-safe for reading. Luckily, all important Delphi containers (TList,
TObjectList, TStringList) fall into this category.

If the container is thread-safe for reading, then the numeric approach (ForEach(0, [Link]-1)) is much
faster than the for..in approach (ForEach(sl)). The speed difference comes from the locking - in the former
example ForEach never locks anything and in the latter example locking is used to synchronize access to the
container.
However, if the container is not thread-safe for reading, you have to use the latter approach.
There are three ways to iterate over enumerable containers. You can provide the ForEach call with an
IEnumerable interface, with an IEnumerator interface or with an enumerable collection itself. In the latter case,
OmniThreadLibrary will use RTTI to access the enumerator for the collection. For this to work, enumerator
itself must be implemented as an object, not as a record or interface. Luckily, most if not all the VCL
enumerators are implemented in this way.

1 class function ForEach(const enumerable: IEnumerable):

2 IOmniParallelLoop; overload;
3 class function ForEach(const enum: IEnumerator):
4 IOmniParallelLoop; overload;
5 class function ForEach(const enumerable: TObject):
6 IOmniParallelLoop; overload;
7 class function ForEach<T>(const enumerable: IEnumerable):
8 IOmniParallelLoop<T>; overload;
9 class function ForEach<T>(const enum: IEnumerator):
10 IOmniParallelLoop<T>; overload;
11 class function ForEach<T>(const enumerable: TEnumerable<T>):
12 IOmniParallelLoop<T>; overload;
13 class function ForEach<T>(const enum: TEnumerator<T>):
14 IOmniParallelLoop<T>; overload;
15 class function ForEach<T>(const enumerable: TObject):
16 IOmniParallelLoop<T>; overload;
17 class function ForEach<T>(const enum: IEnumerator<T>):
18 IOmniParallelLoop<T>; overload;
19 class function ForEach<T>(const enumerable: IEnumerable<T>):
20 IOmniParallelLoop<T>;

Support for enumerating over IEnumerator<T> and IEnumerable<T> was added in version [3.07.7] .
 High-level multi-threading 35

[Link] … Thread-safe enumerable collections

Collection enumeration uses locking to synchronize access to the collection enumerator, which slows down
the enumeration process. In some special cases, collection may be enumerable without the locking. To support
enumeration, such collection must implement IOmniValueEnumerable and IOmniValueEnumerator interfaces,
which are defined in the OtlCommon unit.

1 class function ForEach(const enumerable: IOmniValueEnumerable):

2 IOmniParallelLoop; overload;
3 class function ForEach(const enum: IOmniValueEnumerator):
4 IOmniParallelLoop; overload;
5 class function ForEach<T>(const enumerable: IOmniValueEnumerable):
6 IOmniParallelLoop<T>; overload;
7 class function ForEach<T>(const enum: IOmniValueEnumerator):
8 IOmniParallelLoop<T>; overload;

[Link] … Blocking collections

To simplify enumerating over blocking collections, the Parallel class implements two ForEach overloads
accepting a blocking collection. Internally, blocking collection is enumerated with the IOmniValueEnumerable
interface.

1 class function ForEach(const source: IOmniBlockingCollection):

2 IOmniParallelLoop; overload;
3 class function ForEach<T>(const source: IOmniBlockingCollection):
4 IOmniParallelLoop<T>; overload;

[Link] … Anything

As a last resort, the Parallel class implements three ForEach overloads that will (with some help from the
programmer) iterate over any data.
The TOmniSourceProvider way is powerful, but complicated.

1 class function ForEach(const sourceProvider: TOmniSourceProvider):

2 IOmniParallelLoop; overload;

You must implement a descendant of the TOmniSourceProvider class. All methods must be thread-safe. For
more information about the source providers, see the Internals section, below.
 High-level multi-threading 36

1 TOmniSourceProvider = class abstract

2 public
3 function Count: int64; virtual; abstract;
4 function CreateDataPackage: TOmniDataPackage; virtual; abstract;
5 function GetCapabilities: TOmniSourceProviderCapabilities;
6 virtual; abstract;
7 function GetPackage(dataCount: integer; package: TOmniDataPackage):
8 boolean; virtual; abstract;
9 function GetPackageSizeLimit: integer; virtual; abstract;
10 end;

As this approach is not for the faint of heart, OmniThreadLibrary provides a slower but much simpler version.

1 class function ForEach(enumerator: TEnumeratorDelegate):

2 IOmniParallelLoop; overload;
3 class function ForEach<T>(enumerator: TEnumeratorDelegate<T>):
4 IOmniParallelLoop<T>; overload;

Here, you must provide a function that will return next data whenever the ForEach asks for it.

1 TEnumeratorDelegate = reference to function(var next: TOmniValue): boolean;

2 TEnumeratorDelegate<T> = reference to function(var next: T): boolean;

OmniThreadLibrary will provide the synchronisation (locking) so you can be sure this method will only be
called from one thread at a time. As you may expect, this will slow things down, but parallelization may
still give you a reasonable performance increase if ForEach payload is substantial (i.e. if the method you are
executing in the ForEach loop takes non-trivial time to execute).
The TEnumeratorDelegate function can also be used as a generator; that is it can calculate the values that will
then be processed in the parallel for loop.

2.3.3 Providing external input

Sometimes, especially when you are dealing with datasets, synchronized access to the container will not be
enough. When you are dealing with database connections, datasets etc you can easily run into thread affinity
problems - that is the inability of some component to work correctly if it is called from a different thread
than the one that it was created in.

Always initialize database connections and datasets in the thread that will use them. You code
may work without that precaution but unless you have extensively tested database components
in multiple threads, you should not assume that they will work correctly unless that condition
(initialization and use in the same thread) is met.

In such case, the best way is to provide the input directly from the main thread. There are few different ways
to achieve that.
 High-level multi-threading 37

1. Repackage data into another collection that can be easily consumed in ForEach (TObjectList, TStringList,
TOmniBlockingCollection).
2. Run the ForEach in NoWait mode, then write the data into the input queue and when you run out of data,
wait for the ForEach loop to terminate. This approach is also useful when you want to push ForEach
into background and provide it with data from some asynchronous event handler.

An example of the second approach will help clarify the idea.

1 uses
2 OtlCommon,
3 OtlCollections,
4 OtlParallel;
5
6 procedure Test;
7 var
8 i : integer;
9 input: IOmniBlockingCollection;
10 loop : IOmniParallelLoop<integer>;
11 wait : IOmniWaitableValue;
12 begin
13 // create the container
14 input := [Link];
15 // create the 'end of work' signal
16 wait := CreateWaitableValue;
17 loop := [Link]<integer>(input);
18 // set up the termination method which will signal 'end of work'
19 [Link](
20 procedure
21 begin
22 [Link];
23 end);
24 // start the parallel for loop in NoWait mode
25 [Link](
26 procedure (const value: integer)
27 begin
28 // do something with the input value
29 OutputDebugString(PChar(Format('%d', [value])));
30 end
31 );
32 // provide the data to the parallel for loop
33 for i := 1 to 1000 do
34 [Link](i);
35 // signal to the parallel for loop that there's no more data to process
36 [Link];
37 // wait for the parallel for loop to stop
38 [Link];
 High-level multi-threading 38

39 // destroy the parallel for loop

40 loop := nil;
41 end;

2.3.4 IOmniParallelLoop interface

The [Link] returns an IOmniParallelLoop interface which is used to configure and run the parallel
for loop.

1 IOmniParallelLoop = interface
2 function Aggregate(defaultAggregateValue: TOmniValue;
3 aggregator: TOmniAggregatorDelegate): IOmniParallelAggregatorLoop;
4 function AggregateSum: IOmniParallelAggregatorLoop;
5 procedure Execute(loopBody: TOmniIteratorDelegate); overload;
6 procedure Execute(loopBody: TOmniIteratorTaskDelegate); overload;
7 function CancelWith(const token: IOmniCancellationToken):
8 IOmniParallelLoop;
9 function Initialize(taskInitializer: TOmniTaskInitializerDelegate):
10 IOmniParallelInitializedLoop;
11 function Into(const queue: IOmniBlockingCollection):
12 IOmniParallelIntoLoop; overload;
13 function NoWait: IOmniParallelLoop;
14 function NumTasks(taskCount : integer): IOmniParallelLoop;
15 function OnMessage(eventDispatcher: TObject):
16 IOmniParallelLoop; overload; deprecated 'use TaskConfig';
17 function OnMessage(msgID: word; eventHandler: TOmniTaskMessageEvent):
18 IOmniParallelLoop; overload; deprecated 'use TaskConfig';
19 function OnMessage(msgID: word; eventHandler: TOmniOnMessageFunction):
20 IOmniParallelLoop; overload; deprecated 'use TaskConfig';
21 function OnTaskCreate(taskCreateDelegate: TOmniTaskCreateDelegate):
22 IOmniParallelLoop; overload;
23 function OnTaskCreate(taskCreateDelegate:
24 TOmniTaskControlCreateDelegate): IOmniParallelLoop; overload;
25 function OnStop(stopCode: TProc): IOmniParallelLoop;
26 function OnStop(stopCode: TOmniTaskStopDelegate): IOmniParallelLoop; overload;
27 function OnStopInvoke(stopCode: TProc): IOmniParallelLoop;
28 function PreserveOrder: IOmniParallelLoop;
29 function TaskConfig(const config: IOmniTaskConfig): IOmniParallelLoop;
30 end;

ForEach<T> returns an IOmniParallelLoop<T> interface, which is exactly the same as the IOmniParallelLoop
except that each method returns the appropriate <T> version of the interface.
Aggregate and AggregateSum are used to implement aggregation. See the Aggregation section, below.
 High-level multi-threading 39

Execute accepts the block of code to be executed for each value in the input container. Two method signatures
are supported, both having the <T> variant. One accepts only the iteration value parameter, and another
accepts an additional IOmniTask parameter.

1 TOmniIteratorDelegate = reference to procedure(const value: TOmniValue);

2 TOmniIteratorDelegate<T> = reference to procedure(const value: T);
3 TOmniIteratorTaskDelegate =
4 reference to procedure(const task: IOmniTask; const value: TOmniValue);
5 TOmniIteratorTaskDelegate<T> =
6 reference to procedure(const task: IOmniTask; const value: T);

CancelWith enables the cancellation mechanism.

With Initialize and OnTaskCreate, you can initialize per-task data before the task begins execution. See the
Task initialization section, below.
Into sets up the output queue, see Preserving output order.
If you call the NoWait function, parallel for will start in the background and control will be returned to the
main thread immediately. If NoWait is not called, Execute will only return after all tasks have stopped working.
By calling NumTasks you can set up the number of worker tasks. By default, the number of tasks is set to
[number of cores available to the process] - 1 if NoWait or PreserveOrder modifiers are used and to [number
of cores available to the process] in all other cases.
If NumTasks receives a positive parameter (> 0), the number of worker tasks is set to that number. For example,
NumTasks(16) starts 16 worker tasks, even if that is more than number of available cores.

If NumTasks receives a negative parameter (< 0), it specifies the number of cores that should be reserved for
other use. The number of worker tasks is then set to <number of available cores> - <number of reserved
cores>. If, for example, current process can use 16 cores and NumTasks(-4) is used, only 12 (16-4) worker tasks
will be started.
Value 0 is not allowed and results in an exception.
OnMessage functions are deprecated, use TaskConfig instead.
OnStop sets up a termination handler which will be called after all parallel for tasks will have completed their
work. If NoWait function was called, OnStop will be called from one of the worker threads. If, however, NoWait
function was not called, OnStop will be called from the thread that created the ForEach abstraction. This
behaviour makes it hard to execute VCL code from the OnStop so release [3.02] introduced another variation
accepting a delegate with an IOmniTask parameter.

1 TOmniTaskStopDelegate = reference to procedure (const task: IOmniTask);

2 IOmniParallelLoop = interface
3 function OnStop(stopCode: TOmniTaskStopDelegate): IOmniParallelLoop;
4 overload;
5 end;

Using this version of OnStop, the termination handler can use [Link] to execute code in the main thread.
This, however, requires the ForEach abstraction to stay alive until the Invoked code is executed so you must
 High-level multi-threading 40

store the result of the ForEach method in a global variable (form field, for example) and destroy it only in the
termination handler.

1 var
2 loop: IOmniParallelLoop<integer>;
3
4 loop := [Link](1, N).NoWait;
5 [Link](
6 procedure (const task: IOmniTask)
7 begin
8 [Link](
9 procedure
10 begin
11 // do anything
12 loop := nil;
13 end);
14 end);
15 [Link](
16 procedure (const value: integer)
17 begin
18 ...
19 end);

Release [3.07.2] introduced method OnStopInvoke which works like OnStop except that the termination handler
is automatically executed in the context of the owner thread via implicit Invoke.
The code fragment above can be rewritten using OnStopInvoke as follows.

1 var
2 loop: IOmniParallelLoop<integer>;
3
4 loop := [Link](1, N).NoWait;
5 [Link](
6 procedure
7 begin
8 // do anything
9 loop := nil;
10 end);
11 [Link](
12 procedure (const value: integer)
13 begin
14 ...
15 end);

PreserveOrder modifies the Parallel for behaviour so that output values are generated in the order of the
corresponding input values. See the Preserving output order section, below.
 High-level multi-threading 41

TaskConfig sets up a task configuration block. Same task configuration block will be applied to all worker
tasks.
The following example uses TaskConfig to set up a message handler which will receive messages sent from
ForEach worker tasks.

1 FParallel := [Link](1, 17)

2 .TaskConfig([Link](Self))
3 .NoWait
4 .OnStop(procedure begin FParallel := nil; end);
5
6 FParallel
7 .Execute(
8 procedure (const task: IOmniTask; const value: integer)
9 begin
10 [Link](WM_LOG, value);
11 end);

Messages sent from the worker task are received and dispatched by the IOmniParallelLoop interface. This
requires the ForEach abstraction to stay alive until the messages are processed so you must store the result of
the ForEach method in a global variable (form field, for example) and destroy it only in the OnStop handler.
Some functions return a different interface. Typically, it only implements the Execute function accepting a dif-
ferent parameter than the ‘normal’ Execute. For example, Aggregate returns the IOmniParallelAggregatorLoop
interface.

1 TOmniIteratorIntoDelegate =
2 reference to procedure(const value: TOmniValue; var result: TOmniValue);
3
4 IOmniParallelAggregatorLoop = interface
5 function Execute(loopBody: TOmniIteratorIntoDelegate): TOmniValue;
6 end;

These variants of the IOmniParallelLoop interface will be described in following sections.

2.3.5 Preserving output order

When you run a ForEach loop, you can’t tell in advance in which order elements from the input collection
will be processed in. For example, the code below will generate all primes from 1 to CMaxPrime and write
them into the output queue (primeQueue) in a nondeterministic order.
 High-level multi-threading 42

1 primeQueue := [Link];
2 [Link](1, CMaxPrime).Execute(
3 procedure (const value: integer)
4 begin
5 if IsPrime(value) then begin
6 [Link](value);
7 end;
8 end);

Sometimes this will represent a big problem and you’ll have to write a sorting function that will re-sort
the output before it can be processed further. To ease the problem, IOmniParallelLoop implements the
PreserveOrder modifier. When used, ForEach internally sorts the results produced in the worker task method.

Using PreserveOrder also forces you to use the Into method which returns the IOmniParallelIntoLoop
interface. (As you may expect, there’s also the <T> version of that interface.)

1 TOmniIteratorIntoDelegate =
2 reference to procedure(const value: TOmniValue; var result: TOmniValue);
3 TOmniIteratorIntoTaskDelegate =
4 reference to procedure(const task: IOmniTask; const value: TOmniValue;
5 var result: TOmniValue);
6
7 IOmniParallelIntoLoop = interface
8 procedure Execute(loopBody: TOmniIteratorIntoDelegate); overload;
9 procedure Execute(loopBody: TOmniIteratorIntoTaskDelegate); overload;
10 end;

As you can see, the Execute method in IOmniParallelIntoLoop takes a different parameter than the ‘normal’
Execute. Because of that, you’ll have to change a code that is passed to the Execute to return a result.

1 primeQueue := [Link];
2 [Link](1, CMaxPrime)
3 .PreserveOrder
4 .Into(primeQueue)
5 .Execute(
6 procedure (const value: integer; var res: TOmniValue)
7 begin
8 if IsPrime(value) then
9 res := value;
10 end);

When using PreserveOrder and Into, ForEach calls your worker code for each input value. If the worker code
sets output parameter (res) to any value, it will be inserted into a temporary buffer. Then the magic happens
(see the Internals section, below) and as soon as the appropriate (sorted) value is available in the temporary
buffer, it is inserted into the output queue (the one passed to the Into parameter).
You can also use Into without the PreserveOrder. This will give you queue management but no ordering.
 High-level multi-threading 43

2.3.6 Aggregation
Aggregation allows you to collect data from ForEach tasks and calculate one number that is returned to the
user.
Let’s start with an example - intentionally a terrible one! The following code fragment tries to calculate the
number of prime numbers between 1 and CMaxPrime.

1 numPrimes := 0;
2 [Link](1, CMaxPrime).Execute(
3 procedure (const value: integer)
4 begin
5 if IsPrime(value) then
6 Inc(numPrimes);
7 end);

Let’s say it out loud - this code is wrong! Access to the shared variable is not synchronized between threads
and that will make the result indeterminable. One way to solve the problem is to wrap the Inc(numPrimes)
with locking and another is to use InterlockedIncrement instead of Inc, but both will slow down the
execution a lot.
A solution to this problem is to use the Aggregate function.

1 procedure SumPrimes(var aggregate: TOmniValue; const value: TOmniValue)

2 begin
3 aggregate := aggregate.AsInt64 + value.AsInt64;
4 end;
5
6 procedure CheckPrime(const value: integer; var result: TOmniValue)
7 begin
8 if IsPrime(value) then
9 Result := 1;
10 end;
11
12 numPrimes :=
13 [Link](1, CMaxPrime)
14 .Aggregate(0, SumPrimes)
15 .Execute(CheckPrime);

Aggregate takes two parameters - the first is the initial value for the aggregate and the second is an aggregation
function - a piece of code that will take the current aggregate value and update it with the value returned
from the worker task.
When using Aggregate, worker task (the code passed to the Execute function) has the same signature as when
used with Into. It takes the current iteration value and optionally produces a result.
We could approximate the code above with the following for loop which works the same, but uses only one
thread.
 High-level multi-threading 44

1 agg := 0;
2 [Link];
3 for value := 1 to CMaxPrime do begin
4 CheckPrime(value, result);
5 if not [Link] then begin
6 SumPrimes(agg, result);
7 [Link];
8 end;
9 end;
10 numPrimes := agg;

ForEach executes the aggregation in two stages. While the worker task is running, it will aggregate data into
a local variable. When it runs out of work, it will call the same aggregation method to aggregate this local
variable into a global result. In this second stage, however, locking will be used to protect the access to the
global result.

Because the summation is the most common usage of aggregation, IOmniParallelLoop implements function
AggregateSum, which works exactly the same as the SumPrimes above.
 High-level multi-threading 45

1 numPrimes :=
2 [Link](1, CMaxPrime)
3 .AggregateSum
4 .Execute(
5 procedure (const value: integer; var result: TOmniValue)
6 begin
7 if IsPrime(value) then
8 Result := 1;
9 end
10 );

Aggregation function can do something else but the summation. The following code segment uses aggregation
to find the length of the longest line in a file.

1 function GetLongestLineInFile(const fileName: string): integer;

2 var
3 maxLength: TOmniValue;
4 sl : TStringList;
5 begin
6 sl := [Link];
7 try
8 [Link](fileName);
9 maxLength := [Link]<string>(sl)
10 .Aggregate(0,
11 procedure(var aggregate: TOmniValue; const value: TOmniValue)
12 begin
13 if [Link] > [Link] then
14 aggregate := [Link];
15 end)
16 .Execute(
17 procedure(const value: string; var result: TOmniValue)
18 begin
19 result := Length(value);
20 end);
21 Result := maxLength;
22 finally FreeAndNil(sl); end;
23 end;

2.3.7 Cancellation

ForEach has a built-in cancellation mechanism. To use it, create a cancellation token and pass it to the
CancelWith function. When a cancellation token gets signalled, all worker loops will complete the current
iteration and then stop.
An example of using cancellation token can be found in the chapter Parallel search in a tree.
 High-level multi-threading 46

2.3.8 Task initialization and finalization

ForEach implements a mechanism that can be used by worker tasks to initialize and destroy task-specific
structures. In such cases, you have to call the Initialize function.

1 TOmniTaskInitializerDelegate =
2 reference to procedure(var taskState: TOmniValue);
3 TOmniTaskFinalizerDelegate =
4 reference to procedure(const taskState: TOmniValue);
5 TOmniIteratorStateDelegate =
6 reference to procedure(const value: TOmniValue; var taskState: TOmniValue);
7
8 IOmniParallelInitializedLoop = interface
9 function Finalize(taskFinalizer: TOmniTaskFinalizerDelegate):
10 IOmniParallelInitializedLoop;
11 procedure Execute(loopBody: TOmniIteratorStateDelegate);
12 end;
13
14 IOmniParallelLoop = interface
15 ...
16 function Initialize(taskInitializer: TOmniTaskInitializerDelegate):
17 IOmniParallelInitializedLoop;
18 end;

You provide Initialize with task initializer, a procedure that will be called in each worker task when it is
created and before it starts enumerating values. This procedure can initialize the taskState parameter with
any value.
Initialize returns an IOmniParallelInitializedLoop interface which implements two functions - Finalize
and Execute. Call Finalize to set up task finalizer, a procedure that gets called after all values have been
enumerated and before the worker task ends its job.
Execute accepts a worker method with two parameters - the first one is the usual value from the enumerated
container and the second contains the shared task state.
All these functions and interfaces are implemented in the <T> version, too.
The following example shows how to calculate the number of primes from 1 to CHighPrime by using initializers
and finalizers.
 High-level multi-threading 47

1 var
2 lockNum : TOmniCS;
3 numPrimes: integer;
4 begin
5 numPrimes := 0;
6 [Link](1, CHighPrime)
7 .Initialize(
8 procedure (var taskState: TOmniValue)
9 begin
10 [Link] := 0;
11 end)
12 .Finalize(
13 procedure (const taskState: TOmniValue)
14 begin
15 [Link];
16 try
17 numPrimes := numPrimes + [Link];
18 finally [Link]; end;
19 end)
20 .Execute(
21 procedure (const value: integer; var taskState: TOmniValue)
22 begin
23 if IsPrime(value) then
24 [Link] := [Link] + 1;
25 end
26 );
27 end;

2.3.9 Handling exceptions

ForEach abstraction does not yet implement any exception handling. You should always wrap task method
(code passed to the Execute) in try..except if you expect the code to raise exceptions.

2.3.10 Examples

Practical example of ForEach usage can be found in chapters Parallel for with synchronized output and
Parallel search in a tree.
 3. Low-level multi-threading
The low-level OmniThreadLibrary layer focuses on the task concept. In most aspects this is similar to the
Delphi’s TThread approach except that OmniThreadLibrary focuses on the code (a.k.a. task) and interaction
with the code, while the Delphi focuses on the operating system primitive required for executing additional
threads.
A task is created using the CreateTask function, which takes as a parameter a global procedure, a method,
an instance of the TOmniWorker class (or, usually, a descendant of that class) or an anonymous procedure (in
Delphi 2009 and newer). CreateTask will also accept an optional second parameter, a task name, which will
be displayed in the Delphi’s Thread view on the thread running the task.

1 type
2 TOmniTaskProcedure = procedure(const task: IOmniTask);
3 TOmniTaskMethod = procedure(const task: IOmniTask) of object;
4 TOmniTaskDelegate = reference to procedure(const task: IOmniTask);
5
6 function CreateTask(worker: TOmniTaskProcedure; const taskName: string = ''):
7 IOmniTaskControl; overload;
8 function CreateTask(worker: TOmniTaskMethod; const taskName: string = ''):
9 IOmniTaskControl; overload;
10 function CreateTask(worker: TOmniTaskDelegate; const taskName: string = ''):
11 IOmniTaskControl; overload;
12 function CreateTask(const worker: IOmniWorker; const taskName: string = ''):
13 IOmniTaskControl; overload;

CreateTask returns a feature-full interface IOmniTaskControl which we will explore in this chapter. The most
important function in this interface, Run, creates a new thread and starts your task in it.

3.1 Low-level for the impatient

The following code represents the simplest low-level OmniThreadLibrary example. It executes the Beep
function in a background thread. The Beep function merely beeps and exits. By exiting from the task function,
the Windows thread running the task is also terminated.
 Low-level multi-threading 49

1 procedure [Link](const task: IOmniTask);

2 begin
3 //Executed in a background thread
4 MessageBeep(MB_ICONEXCLAMATION);
5 end;
6
7 CreateTask(Beep, 'Beep').Run;

Another way to start a task is to call a Schedule function which starts it in a thread allocated from a thread
pool. This is covered in the Thread pooling chapter.

3.2 Four ways to create a task

Let’s examine all four ways of creating a task. The simplest way (demoed in application 2_TwoWayHello) is
to pass a name of a global procedure to the CreateTask. This global procedure must accept one parameter of
type IOmniTask .

1 procedure RunHelloWorld(const task: IOmniTask);

2 begin
3 //
4 end;
5
6 CreateTask(RunHelloWorld, 'HelloWorld').Run;

A variation on the theme is passing a name of a method to the CreateTask. This approach is used in the demo
application 1_HelloWorld. The interesting point here is that you can declare this method in the same class
from which the CreateTask is called. That way you can access all class fields and methods from the threaded
code. Just keep in mind you’ll be doing this from another thread so make sure you protect shared access with
locking!

1 procedure [Link](const task: IOmniTask);

2 begin
3 //
4 end;
5
6 procedure [Link];
7 begin
8 CreateTask(RunHelloWorld, 'HelloWorld').Run;
9 end;

In Delphi 2009 and newer you can also write the task code as an anonymous function.
 Low-level multi-threading 50

1 CreateTask(
2 procedure (const task: IOmniTask)
3 begin
4 //
5 end,
6 'HellowWorld').Run;

For all except the simplest tasks, you’ll use the fourth approach as it will give you access to the true
OmniThreadLibrary power (namely internal wait loop and message dispatching). To use it, you have to
create a worker object deriving from the TOmniWorker class.

1 type
2 THelloWorker = class(TOmniWorker)
3 end;
4
5 procedure [Link](Sender: TObject);
6 begin
7 FHelloTask :=
8 CreateTask([Link](), 'Hello').
9 Run;
10 end;

3.3 IOmniTaskControl and IOmniTask interfaces

When you create a low-level task, OmniThreadLibrary returns a task controller interface IOmniTaskControl.
This interface, which is defined in the OtlTaskControl unit, can be used to control the task from the owner’s
side. The task code, on the other hand, has access to another interface, IOmniTask (defined in the OtlTask
unit), which can be used to communicate with the owner and manipulate the task itself. A picture in the
Tasks vs. threads chapter shows the relationship between these interfaces.
This chapter deals mainly with these two interfaces. For the reference reasons, the IOmniTaskControl is
reprinted here in full. In the rest of the chapter I’ll just show relevant interface parts.
The IOmniTask interface is described at the end of this chapter.

1 type
2 IOmniTaskControl = interface
3 function Alertable: IOmniTaskControl;
4 function CancelWith(const token: IOmniCancellationToken): IOmniTaskControl;
5 function ChainTo(const task: IOmniTaskControl;
6 ignoreErrors: boolean = false): IOmniTaskControl;
7 function ClearTimer(timerID: integer): IOmniTaskControl;
8 function DetachException: Exception;
9 function Enforced(forceExecution: boolean = true): IOmniTaskControl;
10 function GetFatalException: Exception;
 Low-level multi-threading 51

11 function GetParam: TOmniValueContainer;

12 function Invoke(const msgMethod: pointer): IOmniTaskControl; overload;
13 function Invoke(const msgMethod: pointer;
14 msgData: array of const): IOmniTaskControl; overload;
15 function Invoke(const msgMethod: pointer;
16 msgData: TOmniValue): IOmniTaskControl; overload;
17 function Invoke(const msgName: string): IOmniTaskControl; overload;
18 function Invoke(const msgName: string;
19 msgData: array of const): IOmniTaskControl; overload;
20 function Invoke(const msgName: string;
21 msgData: TOmniValue): IOmniTaskControl; overload;
22 function Invoke(remoteFunc: TOmniTaskControlInvokeFunction):
23 IOmniTaskControl; overload;
24 function Invoke(remoteFunc: TOmniTaskControlInvokeFunctionEx):
25 IOmniTaskControl; overload;
26 function Join(const group: IOmniTaskGroup): IOmniTaskControl;
27 function Leave(const group: IOmniTaskGroup): IOmniTaskControl;
28 function MonitorWith(const monitor: IOmniTaskControlMonitor):
29 IOmniTaskControl;
30 function MsgWait(wakeMask: DWORD = QS_ALLEVENTS): IOmniTaskControl;
31 function NUMANode(numaNodeNumber: integer): IOmniTaskControl;
32 function OnMessage(eventDispatcher: TObject): IOmniTaskControl; overload;
33 function OnMessage(eventHandler: TOmniTaskMessageEvent): IOmniTaskControl; overload;
34 function OnMessage(msgID: word; eventHandler: TOmniTaskMessageEvent):
35 IOmniTaskControl; overload;
36 function OnMessage(msgID: word; eventHandler: TOmniMessageExec):
37 IOmniTaskControl; overload;
38 function OnMessage(eventHandler: TOmniOnMessageFunction):
39 IOmniTaskControl; overload;
40 function OnMessage(msgID: word; eventHandler: TOmniOnMessageFunction):
41 IOmniTaskControl; overload;
42 function OnTerminated(eventHandler: TOmniOnTerminatedFunction):
43 IOmniTaskControl; overload;
44 function OnTerminated(eventHandler: TOmniOnTerminatedFunctionSimple):
45 IOmniTaskControl; overload;
46 function OnTerminated(eventHandler: TOmniTaskTerminatedEvent):
47 IOmniTaskControl; overload;
48 function ProcessorGroup(procGroupNumber: integer): IOmniTaskControl;
49 function RemoveMonitor: IOmniTaskControl;
50 function Run: IOmniTaskControl;
51 function Schedule(const threadPool: IOmniThreadPool = nil {default pool}):
52 IOmniTaskControl;
53 function SetMonitor(hWindow: THandle): IOmniTaskControl;
54 function SetParameter(const paramName: string;
55 const paramValue: TOmniValue): IOmniTaskControl; overload;
56 function SetParameter(const paramValue: TOmniValue):
 Low-level multi-threading 52

57 IOmniTaskControl; overload;
58 function SetParameters(const parameters: array of TOmniValue):
59 IOmniTaskControl;
60 function SetPriority(threadPriority: TOTLThreadPriority): IOmniTaskControl;
61 function SetQueueSize(numMessages: integer): IOmniTaskControl;
62 function SetTimer(timerID: integer; interval_ms: cardinal;
63 const timerMessage: TOmniMessageID): IOmniTaskControl; overload;
64 procedure SetTimer(timerID: integer; interval_ms: cardinal;
65 const timerMessage: TProc); overload;
66 procedure SetTimer(timerID: integer; interval_ms: cardinal;
67 const timerMessage: TProc<integer>); overload;
68 function SetUserData(const idxData: TOmniValue;
69 const value: TOmniValue): IOmniTaskControl;
70 procedure Stop;
71 function Terminate(maxWait_ms: cardinal = INFINITE): boolean;
72 function TerminateWhen(event: THandle): IOmniTaskControl; overload;
73 function TerminateWhen(token: IOmniCancellationToken):
74 IOmniTaskControl; overload;
75 function Unobserved: IOmniTaskControl;
76 function WaitFor(maxWait_ms: cardinal): boolean;
77 function WaitForInit: boolean;
78 function WithCounter(const counter: IOmniCounter): IOmniTaskControl;
79 function WithLock(const lock: TSynchroObject;
80 autoDestroyLock: boolean = true): IOmniTaskControl; overload;
81 function WithLock(const lock: IOmniCriticalSection):
82 IOmniTaskControl; overload;
83 //
84 property CancellationToken: IOmniCancellationToken
85 read GetCancellationToken;
86 property Comm: IOmniCommunicationEndpoint read GetComm;
87 property ExitCode: integer read GetExitCode;
88 property ExitMessage: string read GetExitMessage;
89 property FatalException: Exception read GetFatalException;
90 property Lock: TSynchroObject read GetLock;
91 property Name: string read GetName;
92 property Param: TOmniValueContainer read GetParam;
93 property UniqueID: int64 read GetUniqueID;
94 property UserData[const idxData: TOmniValue]: TOmniValue
95 read GetUserDataVal write SetUserDataVal;
96 end;

3.4 Task controller needs an owner

The IOmniTaskController interface returned from the CreateTask must always be stored in a variable/field
with a scope that exceeds the lifetime of the background task. In other words, don’t store a long-term
 Low-level multi-threading 53

background task interface in a local variable.

The simplest example of the wrong approach can be written in one line:

1 CreateTask(MyWorker).Run;

This code looks fine, but it doesn’t work. In this case, the IOmniTaskController interface is stored in a hidden
temporary variable which is destroyed at the end of the current method. This then causes the task controller
to be destroyed which in turn causes the background task to be destroyed. Running this code would therefore
just create and then destroy the task.
A common solution is to just store the interface in some field.

1 FTaskControl := CreateTask(MyWorker).Run;

When you don’t need background worker anymore, you should terminate the task and free the task controller.

1 [Link];
2 FTaskControl := nil;

Another solution is to provide the task with an implicit owner. You can, for example, use the event monitor
to monitor tasks lifetime or messages sent from the task and that will make the task owned by the monitor.
The following code is therefore valid:

1 CreateTask(MyWorker).MonitorWith(eventMonitor).Run;

Yet another possibility is to call the Unobserved before the Run. This method makes the task being observed
by an internal monitor.

1 CreateTask(MyWorker).[Link];

When you use a thread pool to run a task, the thread pool acts as a task owner so there’s no need for an
additional explicit owner.

1 procedure Beep(const task: IOmniTask);

2 begin
3 MessageBeep(MB_ICONEXCLAMATION);
4 end;
5
6 CreateTask(Beep, 'Beep').Schedule;

3.5 Communication subsystem

As it is explained in the Locking vs. messaging section, OmniThreadLibrary automatically creates a
communication channel between the task controller and the task and exposes it through the Comm property.
The communication channel is not exclusive to the OmniThreadLibrary; you could use it equally well from
a TThread-based multi-threading code.
 Low-level multi-threading 54

1 property Comm: IOmniCommunicationEndpoint read GetComm;

The IOmniCommunicationEndpoint interface exposes a simple interface for sending and receiving messages.

1 type
2 TOmniMessage = record
3 MsgID : word;
4 MsgData: TOmniValue;
5 constructor Create(aMsgID: word; aMsgData: TOmniValue); overload;
6 constructor Create(aMsgID: word); overload;
7 end;
8
9 IOmniCommunicationEndpoint = interface
10 function Receive(var msg: TOmniMessage): boolean; overload;
11 function Receive(var msgID: word; var msgData: TOmniValue): boolean; overload;
12 function ReceiveWait(var msg: TOmniMessage; timeout_ms: cardinal): boolean; overload;
13 function ReceiveWait(var msgID: word; var msgData: TOmniValue;
14 timeout_ms: cardinal): boolean; overload;
15 procedure Send(const msg: TOmniMessage); overload;
16 procedure Send(msgID: word); overload;
17 procedure Send(msgID: word; msgData: array of const); overload;
18 procedure Send(msgID: word; msgData: TOmniValue); overload;
19 function SendWait(msgID: word;
20 timeout_ms: cardinal = CMaxSendWaitTime_ms): boolean; overload;
21 function SendWait(msgID: word; msgData: TOmniValue;
22 timeout_ms: cardinal = CMaxSendWaitTime_ms): boolean; overload;
23 property NewMessageEvent: THandle read GetNewMessageEvent;
24 property OtherEndpoint: IOmniCommunicationEndpoint read GetOtherEndpoint;
25 property Reader: TOmniMessageQueue read GetReader;
26 property Writer: TOmniMessageQueue read GetWriter;
27 end;

• Receive
Both variants of Receive return the first message from the message queue, either as a TOmniMessage
record or as a (message ID, message data) pair. Data is always passed as a TOmniValue record.
The function returns True if a message was returned, False if the message queue is empty.
• ReceiveWait
These two variations of the Receive allow you to specify the maximum timeout (in milliseconds) you
are willing to wait for the next message. Timeout of 0 milliseconds makes the function behave just like
the Receive. Special timeout value INFINITE (defined in the Windows unit) will make the function wait
until a message is available.
The function returns True if a message was returned, False if the message queue is still empty after the
timeout.
Low-level multi-threading 55

• Send
Four overloaded versions of Send all write a message to the message queue and raise an exception if
the queue is full. [Message queue size defaults to 1000 elements and can be increased by calling the
[Link] before the communication channel is used for the first time.]
The Send(msgID: word) version sends an empty message data ([Link]).
The Send(msgID: word; msgData: array of const) version packs the data array into one TOmniValue
value by calling [Link](msgData).
• SendWait
These two variations of the Send method allow you to specify the maximum timeout (in milliseconds)
you are willing to wait if a message queue is full and there’s no place for the messages. The timeout of 0
ms makes the function behave just like the Send. A timeout of INFINITE milliseconds is also supported.
The function returns True if the message was successfully sent, False if the message queue is still full
after the timeout.
• NewMessageEvent
This property returns Windows event which is signalled every time new data is inserted in the queue.
This event is not created until the code accesses the NewMessageEvent property for the first time.
• OtherEndpoint
Returns the other end of the communication channel (task’s end if accessed through the [Link]
and task controller’s end if accessed through the [Link] interface).
• Reader
Returns the input queue associated with this endpoint.
• Writer
Returns the output queue associated with this endpoint.

In versions up to [3.04a] , both SendWait and ReceiveWait were designed to be used from only one
thread at a time. Since OmniThreadLibrary [3.04b] they are both fully thread-safe and can be used
from multiple producers and consumers at the same time.

For practical examples on a communication channel usage, see the Communication subsection of simple tasks
and TOmniWorker tasks sections.

Communication message queue is implemented using the Bounded Queue structure.

3.6 Processor groups and NUMA nodes

On a system with multiple processor groups you can use ProcessorGroup [3.06] function to specify a processor
group this task should run on.
Low-level multi-threading 56

On a system with multiple NUMA nodes you can use NUMANode [3.06] function to specify a NUMA node this
task should run on.
When a task is not started directly (Run) but executed via thread pool (Schedule), [Link]
and [Link] should be used instead.
An information about existing processor groups and NUMA nodes can be accessed through the Environment
object.
Demo 64_ProcessorGroups_NUMA demonstrates the use of ProcessorGroup and NUMANode functions.

3.7 Lock-free collections

OmniThreadLibrary implements three lock-free data structures suitable for low-level usage – bounded stack,
bounded queue and dynamic queue. Bounded queue is used inside the OmniThreadLibrary for messaging
while dynamic queue is used as a basis of the blocking collection.
All three data structures are fully thread-safe. They support multiple simultaneous readers and writers. They
are implemented in the OtlContainers unit.
Another lock-free data structure, a message queue, is defined in the OtlComm unit and is mostly intended
for internal operation (such as sending messages to and from a thread) although it can also be used for other
tasks. An example of such usage is shown in the Using message queue with a TThread worker chapter.

The term lock-free is not well defined (and not even universally accepted). In the context of this
book lock-free means that the synchronisation between threads is not achieved with the user-
or kernel-level synchronisation primitives such as critical sections, but with bus-locking CPU
instructions. With modern CPU architectures this approach is much faster than locking on the
operating system level.

See also demos 10_Containers and 32_Queue.

3.7.1 Bounded Stack

The bounded stack¹ structure is a very fast stack with limited length. The core of the implementation is stored
in the TOmniBaseBoundedStack class.
Derived class TOmniBoundedStack adds support for external observers. Both classes implement the same
interface – IOmniStack – so you can code against the class or against the interface.

¹[Link]
 Low-level multi-threading 57

1 type
2 IOmniStack = interface
3 procedure Empty;
4 procedure Initialize(numElements, elementSize: integer);
5 function IsEmpty: boolean;
6 function IsFull: boolean;
7 function Pop(var value): boolean;
8 function Push(const value): boolean;
9 end;
10
11 TOmniBaseBoundedStack = class(TInterfacedObject, IOmniStack)
12 public
13 destructor Destroy; override;
14 procedure Empty;
15 procedure Initialize(numElements, elementSize: integer); virtual;
16 function IsEmpty: boolean; inline;
17 function IsFull: boolean; inline;
18 function Pop(var value): boolean;
19 function Push(const value): boolean;
20 property ElementSize: integer read obsElementSize;
21 property NumElements: integer read obsNumElements;
22 end;
23
24 TOmniBoundedStack = class(TOmniBaseBoundedStack)
25 public
26 constructor Create(numElements, elementSize: integer;
27 partlyEmptyLoadFactor: real = CPartlyEmptyLoadFactor;
28 almostFullLoadFactor: real = CAlmostFullLoadFactor);
29 destructor Destroy; override;
30 function Pop(var value): boolean;
31 function Push(const value): boolean;
32 property ContainerSubject: TOmniContainerSubject read osContainerSubject;
33 end;

• Empty
Empties the stack.
• Initialize
Initializes the stack for maximum numElements elements of size elementSize.
• IsEmpty
Returns True when the stack is empty.
• IsFull
Returns True when the stack is full.
• Pop
Takes one value from the stack and returns True if the stack was not empty before the operation.
 Low-level multi-threading 58

• Push
Puts one value on the stack and returns True if there was a place for the value (the stack was not full
before the operation).
• ElementSize
Returns the size of the stack element as set in the Initialize call.
• NumElements
Returns the maximum number of elements in the stack as set in the Initialize call.
• ContainerSubject
Provides a point for attaching external observers as described in the Observing lock-free collections
section.

3.7.2 Bounded queue

The bounded queue² structure is a very fast queue with limited length.
The core of the implementation is stored in the TOmniBaseBoundedQueue class. Derived class TOmniBoundedQueue
adds support for external observers. Both classes implement the same interface – IOmniQueue – so you can
code against the class or against the interface.

1 type
2 IOmniQueue = interface
3 function Dequeue(var value): boolean;
4 procedure Empty;
5 function Enqueue(const value): boolean;
6 procedure Initialize(numElements, elementSize: integer);
7 function IsEmpty: boolean;
8 function IsFull: boolean;
9 end;
10
11 TOmniBaseBoundedQueue = class(TInterfacedObject, IOmniQueue)
12 public
13 destructor Destroy; override;
14 function Dequeue(var value): boolean;
15 procedure Empty;
16 function Enqueue(const value): boolean;
17 procedure Initialize(numElements, elementSize: integer); virtual;
18 function IsEmpty: boolean;
19 function IsFull: boolean;
20 property ElementSize: integer read obqElementSize;
21 property NumElements: integer read obqNumElements;
22 end;
23
24 TOmniBoundedQueue = class(TOmniBaseBoundedQueue)
25 public
²[Link]
 Low-level multi-threading 59

26 constructor Create(numElements, elementSize: integer;

27 partlyEmptyLoadFactor: real = CPartlyEmptyLoadFactor;
28 almostFullLoadFactor: real = CAlmostFullLoadFactor);
29 destructor Destroy; override;
30 function Dequeue(var value): boolean;
31 function Enqueue(const value): boolean;
32 property ContainerSubject: TOmniContainerSubject read oqContainerSubject;
33 end;

• Empty
Empties the stack.
• Dequeue
Takes one value from the queue’s head and returns True if the queue was not empty before the operation.
• Enqueue
Inserts one value on the queue’s tail and returns True if there was place for the value (the queue was
not full before the operation).
• Initialize
Initializes the queue for maximum numElements elements of size elementSize.
• IsEmpty
Returns True when the queue is empty.
• IsFull
Returns True when the queue is full.
• ElementSize
Returns the size of the queue element as set in the Initialize call.
• NumElements
Returns the maximum number of elements in the queue as set in the Initialize call.
• ContainerSubject
Provides a point for attaching external observers as described in the Observing lock-free collections
section.

3.7.3 Message queue

The TOmniMessageQueue is just a thin wrapper around the bounded queue data structure. An element of this
queue is a (message ID, message data) pair, stored in a TOmniMessage record.
This class greatly simplifies creating and attaching event and window observers.
 Low-level multi-threading 60

1 type
2 TOmniMessage = record
3 MsgID : word;
4 MsgData: TOmniValue;
5 constructor Create(aMsgID: word; aMsgData: TOmniValue); overload;
6 constructor Create(aMsgID: word); overload;
7 end;
8
9 TOmniContainerWindowsEventObserver = class(TOmniContainerObserver)
10 public
11 function GetEvent: THandle; virtual; abstract;
12 end;
13
14 TOmniMessageQueueMessageEvent =
15 procedure(Sender: TObject; const msg: TOmniMessage) of object;
16
17 TOmniMessageQueue = class(TOmniBoundedQueue)
18 public
19 constructor Create(numMessages: integer;
20 createEventObserver: boolean = true); reintroduce;
21 destructor Destroy; override;
22 function Dequeue: TOmniMessage; reintroduce;
23 function Enqueue(const value: TOmniMessage): boolean; reintroduce;
24 procedure Empty;
25 function GetNewMessageEvent: THandle;
26 function TryDequeue(var msg: TOmniMessage): boolean; reintroduce;
27 property EventObserver: TOmniContainerWindowsEventObserver
28 read mqWinEventObserver;
29 property OnMessage: TOmniMessageQueueMessageEvent
30 read [Link] write SetOnMessage;
31 end;

[Link] creates an event observer unless the second parameter (createEventObserver) is

set to False. It is created with the coiNotifyOnAllInserts interest meaning that an event (accessible through
the GetNewMessageEvent function) is signalled each time an element (a message) is added to the queue. The
observer itself is accessible through the EventObserver property.
You can also easily create a window message observer by attaching an event handler to the OnMessage
property. This observer is also created with the coiNotifyOnAllInserts interest which causes the OnMessage
event handler to be called each time an element (a message) is added to the queue. You can destroy this
observer at any time by assigning a nil value to the OnMessage event.
For an example, see chapter Using message queue with a TThread worker.
 Low-level multi-threading 61

3.7.4 Dynamic queue

The dynamic queue³ is a fast queue with unlimited length. It can grow as required as the data used to store
elements is dynamically allocated.
The core of the implementation is stored in the TOmniBaseQueue class. Derived class TOmniQueue adds support
for external observers. Both structures store TOmniValue elements.

1 type
2 TOmniBaseQueue = class
3 ...
4 public
5 constructor Create(blockSize: integer = 65536; numCachedBlocks: integer = 4);
6 destructor Destroy; override;
7 function Dequeue: TOmniValue;
8 procedure Enqueue(const value: TOmniValue);
9 function IsEmpty: boolean;
10 function TryDequeue(var value: TOmniValue): boolean;
11 end;
12
13 TOmniQueue = class(TOmniBaseQueue)
14 ...
15 public
16 function Dequeue: TOmniValue;
17 procedure Enqueue(const value: TOmniValue);
18 function TryDequeue(var value: TOmniValue): boolean;
19 property ContainerSubject: TOmniContainerSubject read ocContainerSubject;
20 end;

• Create
Creates a queue object with a specified page size (blockSize) where numCachedBlocks are always
preserved for future use. Defaults (65536 and 4) should be appropriate for most scenarios.
• Dequeue
Takes one element from queue’s head and returns it. If the queue is empty, an exception is raised.
• Enqueue
Inserts an element on the queue’s tail.
• IsEmpty
Returns True when the queue is empty.
• TryDequeue
Takes one element from queue’s head and returns it in the value parameter. Returns True if an element
was returned (the queue was not empty before the operation).
• ContainerSubject
Provides a point for attaching external observers as described in the Observing lock-free collections
section.
³[Link]
 Low-level multi-threading 62

3.7.5 Observing lock-free collections

OmniThreadLibrary data structures support the observer⁴ design pattern. Each structure can be observed by
multiple observers at the same time. Supporting code and two observer implementations are stored in the
OtlContainerObserver unit.
Current architecture supports four different kinds of events that can be observed:

1 type
2 ///<summary>All possible actions observer can take interest in.</summary>
3 TOmniContainerObserverInterest = (
4 //Interests with permanent subscription:
5 coiNotifyOnAllInserts, coiNotifyOnAllRemoves,
6 //Interests with one-shot subscription:
7 coiNotifyOnPartlyEmpty, coiNotifyOnAlmostFull
8 );

• coiNotifyOnAllInserts
Observer is notified whenever a data element is inserted into the structure.
• coiNotifyOnAllRemoves
Observer is notified whenever a data element is removed from the structure.
• coiNotifyOnPartlyEmpty
Observer is notified whenever a data usage drops below the partlyEmptyLoadFactor (parameter of the
data structure constructor, 80% by default). This event is only supported for bounded structures.
This event can only be observed once. After that you should destroy the observer and (if required)
create another one and attach it to the data structure.
• coiNotifyOnAlmostFull
Observer is notified whenever a data usage rises above the almostFullLoadFactor (parameter of the
data structure constructor, 90% by default). This event is only supported for bounded structures.
This event can only be observed once. After that you should destroy the observer and (if required)
create another one and attach it to the data structure.

The OtlContainerObserver unit implements event and message observers.

⁴[Link]
 Low-level multi-threading 63

1 TOmniContainerWindowsEventObserver = class(TOmniContainerObserver)
2 public
3 function GetEvent: THandle; virtual; abstract;
4 end;
5
6 TOmniContainerWindowsMessageObserver = class(TOmniContainerObserver)
7 strict protected
8 function GetHandle: THandle; virtual; abstract;
9 public
10 procedure Send(aMessage: cardinal; wParam, lParam: integer);
11 virtual; abstract;
12 property Handle: THandle read GetHandle;
13 end;
14
15 function CreateContainerWindowsEventObserver(externalEvent: THandle = 0):
16 TOmniContainerWindowsEventObserver;
17
18 function CreateContainerWindowsMessageObserver(hWindow: THandle;
19 msg: cardinal; wParam, lParam: integer):
20 TOmniContainerWindowsMessageObserver;

The event observer TOmniContainerWindowsEventObserver raises an event every time the observed event
occurs.
The message observer TOmniContainerWindowsMessageObserver sends a message to a window every time the
observed event occurs.

[Link] Examples

Create and attach the event observer:

1 FObserver := CreateContainerWindowsEventObserver;
2 [Link](FObserver, coiNotifyOnAllInserts);

Access the observer event so you can wait on it:

1 FEvent := [Link];

Detach and destroy the observer:

1 [Link](FObserver, coiNotifyOnAllInserts);
2 FreeAndNil(FObserver);

Create and attach the message observer:

 Low-level multi-threading 64

1 FWindow := DSiAllocateHWnd(ObserverWndProc);
2 FObserver := CreateContainerWindowsMessageObserver(
3 FWindow, MSG_ITEM_INSERTED, 0, 0);
4 [Link](FObserver, coiNotifyOnAllInserts);

Process observer messages:

1 procedure ObserverWndProc(var message: TMessage);

2 var
3 ovWorkItem: TOmniValue;
4 workItem : IOmniWorkItem;
5 begin
6 if [Link] = MSG_ITEM_INSERTED then begin
7 //...
8 [Link] := Ord(true);
9 end
10 else
11 [Link] := DefWindowProc(FWindow, [Link],
12 [Link], [Link]);
13 end;

Detach and destroy the observer:

1 [Link](FObserver, coiNotifyOnAllInserts);
2 FreeAndNil(FObserver);
3 DSiDeallocateHWnd(FWindow);

3.7.6 Benchmarks

OmniThreadLibrary contains two demos that can be used to measure the performance of the lock-
free structures. Bounded structures are benchmarked in the 10_Containers demo and dynamic queue is
benchmarked in the 32_Queue demo.
Following results were measured on 4-core i7-2630QM running at 2 GHz. As you can see, lock-free structures
can transfer from 2,5 to 5 million messages per second.
Low-level multi-threading 65
Low-level multi-threading 66
4. Synchronization
Although the OmniThreadLibrary treats communication as a superior approach to locking, there are still
times when using “standard” synchronization primitives such as a critical section are unavoidable. As the
standard Delphi/Windows approach to locking is low-level, OmniThreadLibrary builds on it and improves
it in some significant ways. All these improvements are collected in the OtlSync unit and are described in
the following sections. The only exception is the waitable value class/interface, which is declared in the
OtlCommon unit.
This part of the book assumes that you have a basic understanding of locking. If you are new to the topic,
you should first read the appropriate chapters from one of the books mentioned in the introduction.

4.1 Critical sections

The most useful synchronisation primitive for multi-threaded programming is indubitably the critical
section¹²
OmniThreadLibrary simplifies sharing critical sections between a task owner and a task with the use of the
WithLock method. High-level tasks can access this method through the task configuration block.

I was always holding the opinion that locks should be as granular as possible. Putting many small locks around
many unrelated pieces of code is better than using one giant lock for everything. However, programmers
frequently use one or few locks because managing many critical sections can be a bother.
To help you with writing a better code, OmniThreadLibrary implements three extensions to the Delphi’s
TCriticalSection class - IOmniCriticalSection, TOmniCS and Locked<T>.

4.1.1 IOmniCriticalSection

Delphi implements critical section support with a TCriticalSection class which must be created and
destroyed in the code. (There is also a TRTLCriticalSection record, but it is only supported on Windows.)
OmniThreadLibrary extends this implementation with an IOmniCriticalSection interface, which you only
have to create. The compiler will make sure that it is destroyed automatically at the appropriate place.

¹[Link]
²[Link]
 Synchronization 68

1 type
2 IOmniCriticalSection = interface
3 procedure Acquire;
4 procedure Release;
5 function GetSyncObj: TSynchroObject;
6 property LockCount: integer read GetLockCount;
7 end;
8
9 function CreateOmniCriticalSection: IOmniCriticalSection;

IOmniCriticalSection uses TCriticalSection internally³. It acts just as a proxy that calls TCriticalSection
functions. Besides that, it provides an additional functionality by counting the number of times a critical
section has been acquired, which can help a lot while debugging. This counter can be read through the
LockCount property.

A critical section can be acquired multiple times from one thread. For example, the following code
is perfectly valid:

1 cSec := CreateOmniCriticalSection; //LockCount = 0

2 [Link]; //LockCount = 1
3 [Link]; //LockCount = 2
4 [Link]; //LockCount = 1
5 [Link]; //LockCount = 0

4.1.2 TOmniCS

Another TCriticalSection extension found in the OmniThreadLibrary is the TOmniCS record. It allows you
to use a critical section by declaring a record in an appropriate place.
Using TOmniCS, locking can be as simple as this:

1 uses
2 GpLists,
3 OtlSync;
4
5 procedure ProcessList(const intf: IGpIntegerList);
6 begin
7 //...
8 end;
9
10 var
11 lock: TOmniCS;
³IOmniCriticalSection doesn’t use TCriticalSection directly, but wraps it into a larger object as suggested by Eric Grange.
 Synchronization 69

12 intf: IGpIntegerList;
13
14 procedure Test1;
15 begin
16 intf := [Link];
17 //...
18 [Link];
19 try
20 ProcessList(intf);
21 finally [Link]; end;
22 end;

TOmniCS is implemented as a record with one private field holding the IOmniCriticalSection interface.

1 type
2 TOmniCS = record
3 strict private
4 ocsSync: IOmniCriticalSection;
5 private
6 function GetLockCount: integer; inline;
7 function GetSyncObj: TSynchroObject; inline;
8 public
9 procedure Initialize;
10 procedure Acquire; inline;
11 procedure Release; inline;
12 property LockCount: integer read GetLockCount;
13 property SyncObj: TSynchroObject read GetSyncObj;
14 end;

The Release method merely calls the Release method on the internal interface, while the Acquire method is
more tricky as it has to initialize the ocsSync field first.

1 procedure [Link];
2 begin
3 Initialize;
4 [Link];
5 end;
6
7 procedure [Link];
8 begin
9 [Link];
10 end;

The initialization uses a global critical section to synchronize access to the code that should not be executed
from two threads at once.
 Synchronization 70

1 procedure [Link];
2 begin
3 if not assigned(ocsSync) then begin
4 [Link];
5 try
6 if not assigned(ocsSync) then
7 ocsSync := CreateOmniCriticalSection;
8 finally [Link]; end;
9 end;
10 end;

4.1.3 Locked<T>

TOmniCS is a great simplification of the critical section concept, but it still requires you to declare a separate
locking entity. If this locking entity is only used to synchronize access to a specific instance (being that an
object, record, interface or even a simple type) it is often better to declare a variable/field of type Locked<T>
which combines any type with a critical section.
Using Locked<T>, the example from the TOmniCS section can be rewritten as follows.

1 uses
2 GpLists,
3 OtlSync;
4
5 procedure ProcessList(const intf: IGpIntegerList);
6 begin
7 //...
8 end;
9
10 var
11 lockedIntf: Locked<IGpIntegerList>;
12
13 procedure Test2;
14 begin
15 lockedIntf := [Link];
16 //...
17 [Link];
18 try
19 ProcessList(lockedIntf);
20 finally [Link]; end;
21 end;

The interesting fact to notice is although the lockedIntf is declared as a variable of type Locked<IGpIntegerList>,
it can be initialized and used as if it is of type IGpIntegerList. This is accomplished by providing Implicit
operators for conversion from Locked<T> to T and back. Delphi compiler is (sadly) not smart enough to use this
 Synchronization 71

conversion operator in some cases so you would still sometimes have to use the provided Value property. For
example, you’d have to do it to release wrapped object. (In the example above we have wrapped an interface
and the compiler itself handled the destruction.)

1 procedure ProcessObjList(obj: TGpIntegerList);

2 begin
3 //...
4 end;
5
6 var
7 lockedObj: Locked<TGpIntegerList>;
8
9 procedure Test3;
10 begin
11 lockedObj := [Link];
12 try
13 //...
14 [Link];
15 try
16 ProcessObjList(lockedObj);
17 finally [Link]; end;
18 //...
19 finally [Link]; end;
20 end;

Besides the standard Acquire/Release methods, Locked<T> also implements methods used for pessimistic
locking, which is described later in this chapter, and two almost identical methods called Locked which allow
you to execute a code segment (a procedure, a method or an anonymous method) while the critical section is
acquired. (In other words, you can be assured that the code passed to the Locked method is always executed
only once provided that all code in the program properly locks access to the shared variable.)

1 type
2 Locked<T> = record
3 public
4 type TFactory = reference to function: T;
5 type TProcT = reference to procedure(const value: T);
6 constructor Create(const value: T; ownsObject: boolean = true);
7 class operator Implicit(const value: Locked<T>): T; inline;
8 class operator Implicit(const value: T): Locked<T>; inline;
9 function Initialize(factory: TFactory): T; overload;
10 {$IFDEF OTL_ERTTI}
11 function Initialize: T; overload;
12 {$ENDIF OTL_ERTTI}
13 procedure Acquire; inline;
14 procedure Locked(proc: TProc); overload; inline;
15 procedure Locked(proc: TProcT); overload; inline;
 Synchronization 72

16 procedure Release; inline;

17 procedure Free; inline;
18 property Value: T read GetValue;
19 end;
20
21 procedure Locked<T>.Locked(proc: TProc);
22 begin
23 Acquire;
24 try
25 proc;
26 finally Release; end;
27 end;
28
29 procedure Locked<T>.Locked(proc: TProcT);
30 begin
31 Acquire;
32 try
33 proc(Value);
34 finally Release; end;
35 end;

[Link] Why not use TMonitor?

There is an alternative built into Delphi since 2009 which provides functionality similar to the Locked<T>
– TMonitor. In modern Delphis, every object can be locked by using [Link] function and
unlocked by using [Link]. The example above could be rewritten to use the TMonitor with
little work.

1 var
2 obj: TGpIntegerList;
3
4 procedure Test4;
5 begin
6 obj := [Link];
7 try
8 //...
9 [Link](obj);
10 try
11 ProcessObjList(obj);
12 finally [Link](obj); end;
13 //...
14 finally FreeAndNil(obj); end;
15 end;

A reasonable question to ask is, therefore, why implementing Locked<T>. Why is TMonitor not good enough?
There are plenty of reasons for that.
 Synchronization 73

• TMonitor was buggy since its inception⁴,⁵ (although that was fixed few years later).
• Using TMonitor doesn’t convey your intentions. Just by looking at the variable/field declaration you
wouldn’t know that the entity is supposed to be used in a thread-safe manner. Using Locked<T>,
however, explicitly declares your intent.
• [Link]/Exit doesn’t work with interfaces, records and primitive types. Locked<T> does.

On the positive size, TMonitor is faster than a critical section.

4.2 TWaitFor
A common scenario in parallel programming is that the program has to wait for something to happen. The
occurrence of that something is usually signalled with an event⁶.
On Windows, this is usually accomplished by calling one of the functions from the WaitForMultipleObjects⁷
family. While they are powerful and quite simple to use, they also have a big limitation – one can only wait
for up to 64 events at the same time.
Windows also offers a RegisterWaitForSingleObject⁸ API call which can be used to circumvent this
limitation. Its use is, however, quite complicated to use. To simplify programmer’s life, OmniThreadLibrary
introduces a TWaitFor class which allows the code to wait for any number of events.

1 type
2 TWaitFor = class
3 public type
4 TWaitResult = (
5 waAwaited, // WAIT_OBJECT_0 .. WAIT_OBJECT_n
6 waTimeout, // WAIT_TIMEOUT
7 waFailed, // WAIT_FAILED
8 waIOCompletion // WAIT_IO_COMPLETION
9 );
10 THandleInfo = record
11 Index: integer;
12 end;
13 THandles = array of THandleInfo;
14
15 constructor Create; overload;
16 constructor Create(const handles: array of THandle); overload;
17 destructor Destroy; override;
18 function MsgWaitAny(timeout_ms, wakeMask, flags: cardinal): TWaitResult;
19 procedure SetHandles(const handles: array of THandle);
20 function WaitAll(timeout_ms: cardinal): TWaitResult;
21 function WaitAny(timeout_ms: cardinal; alertable: boolean = false): TWaitResult;
⁴[Link]
⁵[Link]
⁶[Link]
⁷[Link]
⁸[Link]
 Synchronization 74

22 property Signalled: THandles read FSignalledHandles;

23 end;

To use TWaitFor, create an instance of this class and pass it an array of handles either as a constructor
parameter or by calling the SetHandles method. All handles must be created with the CreateEvent Windows
function.
You can then wait for any (WaitAny) or all (WaitAll) events to become signalled. In both cases the Signalled
array is filled with information about signalled (set) events. The Signalled property is an array of THandleInfo
records, each of which only contains one field - an index (into the handles array) of the signalled event.
For example, if you want to wait for two events and then react to them, use the following approach:

1 var
2 wf: TWaitFor;
3 info: THandleInfo;
4
5 wf := [Link]([handle1, handle2]);
6 try
7 if [Link](INFINITE) = waAwaited then begin
8 for info in [Link] do
9 if [Link] = 0 then
10 // handle1 is signalled - do something
11 else if [Link] = 1 then
12 // handle2 is signalled - do something
13 end;
14 finally FreeAndNil(wf); end;

You don’t have to recreate TWaitFor for each wait operation; it is perfectly ok to call WaitXXX functions
repeatedly on the same object. It is also fine to change the array of handles between two WaitXXX calls by
calling the SetHandles method.
The WaitAny method also comes in a variant which processes Windows messages, I/O completion routines
and APC calls (MsgWaitAny). It’s wakeMask and flags parameters are the same as the corresponding parameters
to the MsgWaitForMultipleObjectsEx⁹ API.
The use of the TWaitFor is shown in demo 59_TWaitFor.

4.3 TOmniCounter
The CreateCounter (OtlCommon unit) function creates a counter with an atomic increment and decrement
operations. Such counter can be used from multiple threads at the same time with no locking. Accessing the
counter’s value is also thread-safe.
The counter is returned as an IOmniCounter interface. It is implemented by the TOmniCounter class, which
you can use in your code directly if you’d rather deal with objects than interfaces.
⁹[Link]
 Synchronization 75

1 type
2 IOmniCounter = interface
3 function Increment: integer;
4 function Decrement: integer;
5 function Take(count: integer): integer; overload;
6 function Take(count: integer; var taken: integer): boolean; overload;
7 property Value: integer read GetValue write SetValue;
8 end;
9
10 TOmniCounter = record
11 procedure Initialize;
12 function Increment: integer;
13 function Decrement: integer;
14 function Take(count: integer): integer; overload;
15 function Take(count: integer; var taken: integer): boolean; overload;
16 property Value: integer read GetValue write SetValue;
17 end;
18
19 function CreateCounter(initialValue: integer = 0): IOmniCounter;

The counter part of the TOmniCounter record is automatically initialized on the first use. If you want, you can
call Initialize in advance, although that is not required.
Take is a special operation which tries to decrement the counter by count but stops at 0. It returns the number
that could be taken from the counter (basically, Min(count, [Link])). Its effect is the same as the
following code (except that the real implementation of Take is thread-safe).

1 Result := Min(counter, count);

2 counter := counter - Result;

Take is used in demo Parallel Data Production.

4.4 TOmniAlignedInt32 and TOmniAlignedInt64

Those two records hold 4-byte (32 bit) and 8-byte (64 bit) values, respectively. These values are suitably aligned
so they can be read from and written to in an atomic operation. They also implement atomic Increment,
Decrement, Add, and Substract operations.

These two types were added in version [3.06] . Previously, OmniThreadLibrary used functionally
equivalent types TGp4AlignedInt and TGp8AlignedInt64 from the GpStuff unit.

Reading and writing values stored in the record (through the Value property or by using a supplied Implicit
operator) is also atomic on the Win64 platform.
 Synchronization 76

1 type
2 TOmniAlignedInt32 = record
3 public
4 procedure Initialize; inline;
5 function Add(value: integer): integer; inline;
6 function Addr: PInteger; inline;
7 function CAS(oldValue, newValue: integer): boolean;
8 function Decrement: integer; overload; inline;
9 function Decrement(value: integer): integer; overload; inline;
10 function Increment: integer; overload; inline;
11 function Increment(value: integer): integer; overload; inline;
12 function Subtract(value: integer): integer; inline;
13 class operator Add(const ai: TOmniAlignedInt32; i: integer): cardinal; inline;
14 class operator Equal(const ai: TOmniAlignedInt32; i: integer): boolean; inline;
15 class operator GreaterThan(const ai: TOmniAlignedInt32; i: integer): boolean; inline;
16 class operator GreaterThanOrEqual(const ai: TOmniAlignedInt32; i: integer): boolean;
17 inline;
18 class operator Implicit(const ai: TOmniAlignedInt32): integer; inline;
19 class operator Implicit(const ai: TOmniAlignedInt32): cardinal; inline;
20 class operator Implicit(const ai: TOmniAlignedInt32): PInteger; inline;
21 class operator LessThan(const ai: TOmniAlignedInt32; i: integer): boolean; inline;
22 class operator LessThanOrEqual(const ai: TOmniAlignedInt32; i: integer): boolean;
23 inline;
24 class operator NotEqual(const ai: TOmniAlignedInt32; i: integer): boolean; inline;
25 class operator Subtract(ai: TOmniAlignedInt32; i: integer): cardinal; inline;
26 property Value: integer read GetValue write SetValue;
27 end;
28
29 TOmniAlignedInt64 = record
30 public
31 procedure Initialize; inline;
32 function Add(value: int64): int64; inline;
33 function Addr: PInt64; inline;
34 function CAS(oldValue, newValue: int64): boolean;
35 function Decrement: int64; overload; inline;
36 function Decrement(value: int64): int64; overload; inline;
37 function Increment: int64; overload; inline;
38 function Increment(value: int64): int64; overload; inline;
39 function Subtract(value: int64): int64; inline;
40 property Value: int64 read GetValue write SetValue;
41 end;
5. How-to
This part of the book contains practical examples of OmniThreadLibrary usage. Each of them starts with a
question that introduces the problem and continues with the discussion of the solution.
Following topics are covered:

• Background file scanning

Scanning folders and files in a background thread.

• Web download and database storage

Multiple workers downloading data and storing it in a single database.

• Parallel for with synchronized output

Redirecting output from a parallel for loop into a structure that doesn’t support multi-threaded access.

• Using taskIndex and task initializer in parallel for

Using taskIndex property and task initializer delegate to provide a per-task data storage in Parallel for.

• Background worker and list partitioning

Writing server-like background processing.

• Parallel data production

Multiple workers generating data and writing it into a single file.

• Building a connection pool

Using OmniThreadLibrary to create a pool of database connections.

• QuickSort and parallel max

How to sort an array and how to process an array using multiple threads.

• Parallel search in a tree

How-to 78

Finding data in a tree.

• Multiple workers with multiple frames

Graphical user interface containing multiple frames where each frame is working as a front end for a
background task.

• OmniThreadLibrary and databases

Using databases from OmniThreadLibrary.

• OmniThreadLibrary and COM/OLE

Using COM/OLE from OmniThreadLibrary.

• Using a message queue with a TThread worker

Using OmniThreadLibrary’s TOmniMessageQueue to communicate with a TThread worker.

 How-to 79

5.1 Parallel data production

This question comes from StackOverflow¹. It is reproduced here in a slightly shortened form.

I am looking into generating a file (750 MB) full of random bytes. The problem is that it takes ages until the
process completes. Any ideas for a faster approach?

This solution uses Parallel Task abstraction.

The algorithm works as follows:

• do in parallel:
– repeat
* find out how many bytes to process in this iteration
· if there’s no more work to do, exit the loop
* prepare the buffer
* send it to the output queue

The tricky part is implementing the third item – ‘find out how many bytes to process in this iteration’ – in
a lock-free fashion. What we need is a thread-safe equivalent of the following (completely thread-unsafe)
fragment.

1 if fileSize > CBlockSize then

2 numBytes := CBlockSize
3 else
4 numBytes := fileSize;
5 fileSize := fileSize - numBytes;

OmniThreadLibrary implements a thread-safe version of this pattern in [Link]. If you have

TOmniCounter initialized with some value (say, fileSize) and you call [Link](numBytes), the
code will behave exactly the same as the fragment above except that it will work correctly if Take is called
from multiple threads at the same time. In addition to that, the new value of the fileSize will be stored in
the TOmniCounter’s counter and returned as a function result.
There’s another version of Take which returns the result in a var parameter and sets its result to True if value
returned is larger than zero.

¹[Link]
 How-to 80

1 function [Link](count: integer;

2 var taken: integer): boolean;
3 begin
4 taken := Take(count);
5 Result := (taken > 0);
6 end; { [Link] }

This version of Take allows you to write elegant iteration code which also works when multiple tasks are
accessing the same counter instance.

1 counter := CreateCounter(numBytes);
2 while [Link](blockSize, blockBytes) do begin
3 // process blockBytes bytes
4 end;

The solution creates a counter which holds the number of bytes to be generated (unwritten) and a queue
(outQueue) that will hold generated data buffers until they are written to a file. Then it starts a ParallelTask
abstraction on all available cores. While the abstraction is running in the background (because NoWait is
used), the main thread continues with the CreateRandomFile execution, reads the data from the outQueue and
writes blocks to the file.

1 procedure CreateRandomFile(fileSize: integer; output: TStream);

2 const
3 CBlockSize = 1 * 1024 * 1024 {1 MB};
4 var
5 buffer : TOmniValue;
6 memStr : TMemoryStream;
7 outQueue : IOmniBlockingCollection;
8 unwritten: IOmniCounter;
9 begin
10 outQueue := [Link];
11 unwritten := CreateCounter(fileSize);
12 [Link]
13 .NumTasks([Link])
14 .OnStop([Link](outQueue))
15 .Execute(
16 procedure
17 var
18 buffer : TMemoryStream;
19 bytesToWrite: integer;
20 randomGen : TGpRandom;
21 begin
22 randomGen := [Link];
23 try
24 while [Link](CBlockSize, bytesToWrite) do begin
25 buffer := [Link];
 How-to 81

26 [Link] := bytesToWrite;
27 FillBuffer([Link], bytesToWrite, randomGen);
28 [Link](buffer);
29 end;
30 finally FreeAndNil(randomGen); end;
31 end
32 );
33 for buffer in outQueue do begin
34 memStr := [Link] as TMemoryStream;
35 [Link](memStr, 0);
36 FreeAndNil(memStr);
37 end;
38 end;

The parallel part firstly creates a random generator in each task. Because the random generator code is not
thread-safe, it cannot be shared between the tasks. Next it uses the above-mentioned Take pattern to grab a
bunch of work, generates that much random data (inside the FillBuffer which is not shown here) and adds
the buffer to the outQueue.
You may be asking yourself how will this code stop? When the unwritten counter drops to zero, Take will
fail in every task and anonymous method running inside the task will exit. When this happens in all tasks,
OnStop handler will be called automatically.

The code above passes [Link] to the OnStop. This is a special helper which creates a delegate
that calls CompleteAdding on its parameter. Therefore, OnStop handler will call [Link],
which will cause the for loop in CreateRandomFile to exit after all data is processed.
 How-to 82

5.2 QuickSort and parallel max

I would like to sort a big array of data, but my comparison function is quite convoluted and sorting takes a
long time. Can I use OmniThreadLibrary to speed up sorting?
On a similar topic – sometimes I’d also like to find a maximum data element in this big array, without doing
the sorting. How would I approach this problem?

The answer to both parts of the problem is the same – use the Fork/Join abstraction.

5.2.1 QuickSort
The first part of this how-to implements a well-known quicksort² algorithm in a parallel way (see demo
application 44_Fork-Join QuickSort for the full code).
Let’s start with a non-optimized single-threaded sorter. This simple implementation is easy to convert to the
multi-threaded form.

1 procedure [Link](left, right: integer);

2 var
3 pivotIndex: integer;
4 begin
5 if right > left then begin
6 if (right - left) <= CSortThreshold then
7 InsertionSort(left, right)
8 else begin
9 pivotIndex := Partition(left, right, (left + right) div 2);
10 QuickSort(left, pivotIndex - 1);
11 QuickSort(pivotIndex + 1, right);
12 end;
13 end;
14 end;

As you can see, the code switches to an insertion sort when the dimension of the array drops below some
threshold. This is not important for the single-threaded version (it only brings a small speedup) but it will
help immensely with the multi-threaded version.
Converting this quicksort to a multi-threaded version is simple.
Firstly, we have to create a Fork/Join computation pool. In this example, it is stored in a global field.

1 FForkJoin := [Link];

Secondly, we have to adapt the QuickSort method.

²[Link]
 How-to 83

1 procedure [Link](left, right: integer);

2 var
3 pivotIndex: integer;
4 sortLeft : IOmniCompute;
5 sortRight : IOmniCompute;
6 begin
7 if right > left then begin
8 if (right - left) <= CSortThreshold then
9 InsertionSort(left, right)
10 else begin
11 pivotIndex := Partition(left, right, (left + right) div 2);
12 sortLeft := [Link](
13 procedure
14 begin
15 QuickSort(left, pivotIndex - 1);
16 end);
17 sortRight := [Link](
18 procedure
19 begin
20 QuickSort(pivotIndex + 1, right);
21 end);
22 [Link];
23 [Link];
24 end;
25 end;
26 end;

The code looks much longer but changes are simple. Each recursive call to QuickSort is replaced with the call
to Compute …

1 sortLeft := [Link](
2 procedure
3 begin
4 QuickSort(left, pivotIndex - 1);
5 end);

… and the code Awaits on both subtasks.

Instead of calling QuickSort directly, parallel version creates IOmniCompute interface by calling [Link].
This creates a subtask wrapping the anonymous function which was passed to the Compute and puts this
subtask into the Fork/Join computation pool.
The subtask is later read from this pool by one of the Fork/Join workers and is processed in the background
thread.
Calling Await checks if the subtask has finished its work. In that case, Await returns and the code can proceed.
Otherwise (subtask is still working), Await tries to get one subtask from the computation pool, executes it,
 How-to 84

and then repeats from the beginning (by checking if the subtask has finished its work). This way, all threads
are always busy either with executing their own code or a subtask from the computation pool.
Because two IOmniCompute interfaces are stored on the stack in each QuickSort call, this code uses more stack
space than the single-threaded version. That is the main reason the parallel execution is stopped at some level
and simple sequential version is used to sort remaining fields.

5.2.2 Parallel max

The second part of this how-to finds a maximum element of an array in a parallel way (see demo application
45_Fork-Join max for the full code).

The parallel solution is similar to the quicksort example above with few important differences related to the
fact that the code must return a value (the quicksort code merely sorted the array returning nothing).
This directly affects the interface usage – instead of working with IOmniForkJoin and IOmniCompute the
code uses IOmniForkJoin<T> and IOmniCompute<T>. As our example array contains integers, the parallel code
creates IOmniForkJoin<integer> and passes it to the ParallelMax function.

1 max := ParallelMax([Link]<integer>, Low(FData), High(FData));

In this example Fork/Join computation pool is passed as a parameter. This approach is more flexible but is
also slightly slower and – more importantly – uses more stack space.

1 function ParallelMax(
2 const forkJoin: IOmniForkJoin<integer>;
3 left, right: integer): integer;
4
5 var
6 computeLeft : IOmniCompute<integer>;
7 computeRight: IOmniCompute<integer>;
8 mid : integer;
9
10 function Compute(left, right: integer): IOmniCompute<integer>;
11 begin
12 Result := [Link](
13 function: integer
14 begin
15 Result := ParallelMax(forkJoin, left, right);
16 end
17 );
18 end;
19
20 begin
21 if (right - left) < CSeqThreshold then
22 Result := SequentialMax(left, right)
23 else begin
 How-to 85

24 mid := (left + right) div 2;

25 computeLeft := Compute(left, mid);
26 computeRight := Compute(mid + 1, right);
27 Result := Max([Link], [Link]);
28 end;
29 end;

When the array subrange is small enough, ParallelMax calls the sequential (single threaded) version – just
as the parallel QuickSort did, and because of the same reason – not to run out of stack space.
With a big subrange, the code creates two IOmniCompute<integer> subtasks each wrapping a function
returning an integer. This function calls back ParallelMax (but with a smaller range). To get the result of the
anonymous function wrapped by the Compute, the code calls the Value function. Just as with the Await, Value
either returns a result (if it was already computed) or executes other Fork/Join subtasks from the computation
pool.

While creating Fork/Join programs, keep in mind this anti-pattern. The following code fragment is
wrong!

1 Result := Max(Compute(left, mid).Value,

2 Compute(mid + 1, right).Value);

You must always create all subtasks before calling Await or Value! Otherwise, your code will not
execute in parallel at all – it will all be processed by a single thread.
6. B. Demo applications
OmniThreadLibrary distribution includes plenty of demo applications that will help you get started. They
are stored in the tests subfolder. This chapter lists all tests.

• 0_Beep: The simplest possible OmniThreadLibrary threading code.

• 1_HelloWorld: Threaded “Hello, World” with TOmniEventMonitor component created in run-time.
• 2_TwoWayHello : “Hello, World” with bidirectional communication; TOmniEventMonitor created in run-
time.
• 3_HelloWorld_with_package: Threaded “Hello, World” with TOmniEventMonitor component on the
form.
• 4_TwoWayHello_with_package: Hello, World with bidirectional communication; TOmniEventMonitor
component on the form.
• 5_TwoWayHello_without_loop : Hello, World with bidirectional communication, the OTL way.
• 6_TwoWayHello_with_object_worker: Obsolete, almost the same as demo 5_TwoWayHello_without_-
loop.
• 7_InitTest: Demonstrates WaitForInit, ExitCode, ExitMessage, and SetPriority.
• 8_RegisterComm: Demonstrates creation of additional communication channels.
• 9_Communications: Simple communication subsystem tester.
• 10_Containers: Full-blown communication subsystem tester. Used to verify correctness of the lock-free
code.
• 11_ThreadPool: Thread pool demo.
• 12_Lock: Demonstrates WithLock.
• 13_Exceptions: Demonstrates exception catching.
• 14_TerminateWhen: Demonstrates TerminateWhen and WithCounter.
• 15_TaskGroup: Task group demo.
• 16_ChainTo: Demonstrates ChainTo.
• 17_MsgWait: Demonstrates MsgWait and Windows message processing inside tasks.
• 18_StringMsgDispatch: Calling task methods by name and address.
• 19_StringMsgBenchmark: Benchmarks various ways of task method invocation.
• 20_QuickSort: Parallel QuickSort demo.
• 21_Anonymous_methods: Demonstrates the use of anonymous methods as task workers in Delphi 2009.
• 22_Termination: Tests for Terminate and Terminated.
• 23_BackgroundFileSearch: Demonstrates file scanning in a background thread.
• 24_ConnectionPool: Demonstrates how to create a connection pool with OmniThreadLibrary.
• 25_WaitableComm: Demo for ReceiveWait and SendWait.
• 26_MultiEventMonitor: How to run multiple event monitors in parallel.
• 27_RecursiveTree: Parallel tree processing.
• 28_Hooks: Demo for the new hook system.
• 29_ImplicitEventMonitor: Demo for OnMessage and OnTerminated, named method approach.
B. Demo applications 87

• 30_AnonymousEventMonitor: Demo for OnMessage and OnTerminated, anonymous method approach.

• 31_WaitableObjects: Demo for the RegisterWaitObject/UnregisterWaitObject API.
• 32_Queue: Stress test for TOmniBaseQueue and TOmniQueue.
• 33_BlockingCollection: Stress test for the TOmniBlockingCollection, also shows the use of Environ-
ment to set process affinity.
• 34_TreeScan: Parallel tree scan using TOmniBlockingCollection.
• 35_ParallelFor: Parallel tree scan using ForEach (Delphi 2009 and newer).
• 37_ParallelJoin: ParallelJoin: Join demo.
• 38_OrderedFor: Ordered ForEach loops.
• 39_Future: Futures.
• 40_Mandelbrot: Very simple parallel graphics demo.
• 41_Pipeline: Multistage parallel processes using Pipeline.
• 42_MessageQueue: Stress test for TOmniMessageQueue.
• 43_InvokeAnonymous: Demo for [Link].
• 44_Fork-Join QuickSort: QuickSort implemented using Fork/Join.
• 45_Fork-Join max: Max(array) implemented using Fork/Join.
• 46_Async: Demo for Async abstraction.
• 47_TaskConfig: Demo for task configuration with [Link].
• 48_OtlParallelExceptions: Exception handling in high-level OTL constructs.
• 49_FramedWorkers: Multiple frames, each communicating with its own worker task.
• 50_OmniValueArray: Wrapping arrays, hashes and records stored in TOmniValue.
• 51_PipelineStressTest: Pipeline stress test by [Anton Alisov].
• 52_BackgroundWorker: Demo for the Background worker abstraction.
• 53_AsyncAwait: Demo for the Async/Await abstraction.
• 54_LockManager: Lock manager (IOmniLockManager<K>) demo.
• 55_ForEachProgress: Demonstrates progress bar updating from a ForEach loop.
• 56_RunInvoke: Simplified ‘run & invoke’ low-level API.
• 57_For: Simple and fast Parallel for.
• 58_ForVsForEach: Speed comparison between ForEach, Parallel for, and [Link]¹ (XE7+).
• 59_TWaitFor: Demo for the TWaitFor class.
• 60_Map: Demonstrates the Map abstraction.
• 61_CollectionToArray: Demonstrates the [Link] method.
• 62_Console: Demonstrates how to use OmniThreadLibrary from a console application.
• 63_Service: Demonstrates how to use OmniThreadLibrary from a service application.
• 64_ProcessorGroups_NUMA: Demonstrates how to work with processor groups and NUMA nodes.
• 65_TimedTask: Demonstrates the TimedTask abstraction.
• 66_ThreadsInThreads: Demonstrates how to start OmniThreadLibrary threads from background threads.
• 67_ArrayToCollection: Demonstrates the use of [Link] and FromRange.

¹[Link]
7. C. Examples
OmniThreadLibrary distribution includes some complex examples, stored in the examples subfolder. This
chapter lists all examples. Many are also explained in the How-to chapter.

• checkVat
OmniThreadLibrary and COM/OLE
Using COM/OLE from OmniThreadLibrary.
• forEach output
Parallel for with synchronized output
Redirecting output from a parallel ForEach loop into a structure that doesn’t support multi-threaded
access.
• report generator
Simulation of a report generator, which uses multiple Background workers to generate reports; one
worker per client.
• stringlist parser
Background worker and list partitioning
Writing server-like background processing.
• TThread communication
Using a message queue with a TThread worker
Using TOmniMessageQueue to communicate with a TThread-based worker.
• twofish
OmniThreadLibrary and databases
Using databases from OmniThreadLibrary.
 8. D. Hooking into OmniThreadLibrary
The OtlHooks unit allows your code to hook into internal OmniThreadLibrary processes. Currently, you can
register notification methods which are called when a thread is created/destroyed, a pool is created/destroyed,
or an unhandled exception ‘escapes’ from a task.

8.1 Exception notifications

Exception filter allows your code to be notified when an unhandled exception in a task occurs. You can also
prevent exception from being stored in the [Link] property.

1 type
2 TExceptionFilterProc = procedure(var e: Exception; var continueProcessing: boolean);
3 TExceptionFilterMeth = procedure(var e: Exception; var continueProcessing: boolean)
4 of object;
5
6 procedure RegisterExceptionFilter(filterProc: TExceptionFilterProc); overload;
7 procedure RegisterExceptionFilter(filterMethod: TExceptionFilterMeth); overload;
8 procedure UnregisterExceptionFilter(filterProc: TExceptionFilterProc); overload;
9 procedure UnregisterExceptionFilter(filterMethod: TExceptionFilterMeth); overload;

Call RegisterExceptionFilter to register a custom exception filter.

Call UnregisterExceptionFilter to remove custom exception filter.
Exception filter can use application-specific logging code to log detailed information about application state.
It can also free the exception object e and set it to nil, which will prevent this exception to be stored in the
FatalException property.

If the filter sets continueProcessing to false, further custom exception filters won’t be called. Filters are
always called in the order in which they were registered.

8.2 Thread notifications

Thread notifications allow your code to be notified when a thread is created or destroyed inside the
OmniThreadLibrary. This allows OmniThreadLibrary to cooperate with application-specific exception-
logging code.
 D. Hooking into OmniThreadLibrary 90

1 type
2 TThreadNotificationType = (tntCreate, tntDestroy);
3 TThreadNotificationProc = procedure(notifyType: TThreadNotificationType;
4 const threadName: string);
5 TThreadNotificationMeth = procedure(notifyType: TThreadNotificationType;
6 const threadName: string) of object;
7
8 procedure RegisterThreadNotification(notifyProc: TThreadNotificationProc); overload;
9 procedure RegisterThreadNotification(notifyMethod: TThreadNotificationMeth); overload;
10 procedure UnregisterThreadNotification(notifyProc: TThreadNotificationProc); overload;
11 procedure UnregisterThreadNotification(notifyMethod: TThreadNotificationMeth); overload;

Call RegisterThreadNotification to register a thread notification method.

Call UnregisterThreadNotification to unregister such method.
Notification method is always called in the context of the thread being created/destroyed.
For example, the following code fragment registers/unregisters OmniThreadLibrary threads with an appli-
cation-specific thread logger.

1 procedure OtlThreadNotify(notifyType: TThreadNotificationType; const threadName: string);

2 var
3 name: string;
4 begin
5 case notifyType of
6 tntCreate:
7 begin
8 if threadName <> '' then
9 name := threadName
10 else
11 name := 'unnamed OTL thread';
12 LoggerRegisterThread(name);
13 end;
14 tntDestroy:
15 LoggerUnregisterThread;
16 else
17 raise [Link]('OtlThreadNotify: Unexpected notification type');
18 end;
19 end;
20
21 [Link](OtlThreadNotify);

8.3 Pool notifications

Pool notifications allow your code to be notified when a thread pool is being created or destroyed. This allows
the application to modify pool parameters on the fly.
 D. Hooking into OmniThreadLibrary 91

1 type
2 TPoolNotificationType = (pntCreate, pntDestroy);
3 TPoolNotificationProc = procedure(notifyType: TPoolNotificationType;
4 const pool: IOmniThreadPool);
5 TPoolNotificationMeth = procedure(notifyType: TPoolNotificationType;
6 const pool: IOmniThreadPool) of object;
7
8 procedure RegisterPoolNotification(notifyProc: TPoolNotificationProc); overload;
9 procedure RegisterPoolNotification(notifyMethod: TPoolNotificationMeth); overload;
10 procedure UnregisterPoolNotification(notifyProc: TPoolNotificationProc); overload;
11 procedure UnregisterPoolNotification(notifyMethod: TPoolNotificationMeth); overload;

Call RegisterPoolNotification to register a pool notification method.

Call UnregisterPoolNotification to unregister such method.
You can, for example, use pool notification mechanism to set Asy_OnUnhandledWorkerException property
whenever a thread pool is created.

1 procedure OtlPoolNotify(notifyType: TPoolNotificationType; const pool: IOmniThreadPool);

2 begin
3 case notifyType of
4 pntCreate: pool.Asy_OnUnhandledWorkerException := Asy_LogUnhandledOtlWorkerException;
5 pntDestroy: pool.Asy_OnUnhandledWorkerException := nil;
6 else raise [Link]('OtlPoolNotify: Unexpected notification type');
7 end;
8 end;
9
10 [Link](OtlPoolNotify);