FireMonkey, Android, Windows and PostMessage

FireMonkey framework (FMX for short) is definitely able to use custom [Windows] messages much like we have always done with the VCL. And this is also true when using FireMonkey to build Android applications.

Both Windows and Android support a messaging system. It is well known by Windows developers who use it with PostMessage, GetMessage, PeekMessage and similar Windows API call. It is much less known by Android developers. Android has a “looper” API which has the same purpose as Windows own messaging system although it is implemented differently and has somewhat more features.

Often, we use FireMonkey framework to build multi-platform applications. Thanks to Delphi XE5, we can build an application for different targets such as Win32, Win64, Android, iOS and MAC OSx. If correctly written, the same application source code can be recompiled for different target and run unchanged. Embarcadero made a lot of efforts to hide differences between the supported platforms.

Speaking about the messaging system, it must admit that Embarcadero forgot to write the abstraction layer required for the platforms. They made some work but it is incomplete and undocumented. This is why I wrote it. At least for Win32, Win64 and Android which are the 3 platforms I currently use.

The layer I wrote is made of a single class I named “TMessagingSystem”. I made two different implementations: one for Android and one for Win 32/64. TMessagingSystem class allows you to register any number of custom messages to a form and associate a custom message handler. Of course it also allows you to call PostMessage to put a message into the message queue.

At the application level, you use the exact same code for Windows or Android. You just have to make use of one of the implementations. You’ll do that using a conditional compilation.

Before showing the implementation details, I will present a demo application. That you can target for Windows or Android without changing a single line.

Demo application for Windows and Android

I built a simple application to emphasize how to use TMessagingSystem. Actually it does not do anything very interesting. It is made of a single form having a button and a memo. When you click on the button, it starts a new thread which will periodically PostMessage a custom message to the main form. You can click many times on the button to start many threads. Each thread will do the same.

The image above shows on the left a screen dump of the application running under Win7 and on the right, the same application running on my Nexus7.

All you see is a memo with messages. Nevertheless, this is really one of the main usages of a messaging system: organize asynchronous operation between threads.

Each line looks like this:

8380] Thread=2 Count=8 ThreadID=7528

“8380” is the thread ID of the thread doing the display. This is always the same and is the main thread ID. “Thread=2” is the sequential thread number having generated the message, “Count=8” is the number of messages generated by this thread and finally, “ThreadID=7528” is the thread ID of the thread generating the message. The later change according to each started thread.

Demo application source code

unit FmxMultiplatformPostMessageDemoMain;

interface

uses
    System.SysUtils, System.Types, System.UITypes, System.Classes,
    Generics.Collections,
    FMX.Types, FMX.Controls, FMX.Forms, FMX.Graphics, FMX.Dialogs,
    FMX.StdCtrls, FMX.Layouts, FMX.Memo,
    FMX.Overbyte.MessageHandling;

const
    WM_SHOW_MESSAGE = WM_USER + 1;

type
    TWorkerThread = class(TThread)
    public
        MsgSys : TMessagingSystem;
        Id     : Integer;
        procedure Execute; override;
    end;

    TForm1 = class(TForm)
        RunThreadButton: TButton;
        DisplayMemo : TMemo;
        ToolPanel: TPanel;
        procedure RunThreadButtonClick(Sender: TObject);
    private
        FMsgSys      : TMessagingSystem;
        FThreadCount : Integer;
        procedure Display(const Msg: String);
        procedure WorkerThreadTerminate(Sender: TObject);
        procedure WMShowMessage(var Msg: TMessage);
    protected
        procedure CreateHandle; override;
        procedure DestroyHandle; override;
    end;

var
  Form1: TForm1;

implementation

{$R *.fmx}

{ TForm1 }

procedure TForm1.CreateHandle;
begin
    inherited CreateHandle;
    FMsgSys := TMessagingSystem.Create(Self);
    FMsgSys.RegisterMessageHandler(WM_SHOW_MESSAGE, WMShowMessage);
end;

procedure TForm1.DestroyHandle;
begin
    FreeAndNil(FMsgSys);
    inherited DestroyHandle;
end;
 
procedure TForm1.RunThreadButtonClick(Sender: TObject);
var
    WorkerThread : TWorkerThread;
begin
    Inc(FThreadCount);
    Display('Start thread ' + IntToStr(FThreadCount));
    WorkerThread                 := TWorkerThread.Create(TRUE);
    WorkerThread.MsgSys          := FMsgSys;
    WorkerThread.Id              := FThreadCount;
    WorkerThread.FreeOnTerminate := TRUE;
    WorkerThread.OnTerminate     := WorkerThreadTerminate;
    WorkerThread.Start;
end;

procedure TForm1.WorkerThreadTerminate(Sender: TObject);
begin
    Display('Thread ' +
            IntToStr((Sender as TWorkerThread).Id) +
            ' terminated');
end;

procedure TForm1.WMShowMessage(var Msg: TMessage);
var
    Buffer : PChar;
begin
    Buffer := PChar(Msg.LParam);
    Display(Buffer);
    FreeMem(Buffer);
end;

procedure TForm1.Display(const Msg: String);
begin
    Displaymemo.Lines.Add(IntToStr(GetCurrentThreadID) + '] ' + Msg);
end;

{ TWorkerThread }

procedure TWorkerThread.Execute;
var
    I      : Integer;
    Buffer : PChar;
const
    MaxLen = 100;
begin
    // For demo, let's do it 10 times
    for I := 1 to 10 do begin
        // Simulate some processing time by sleeping
        Sleep(1000);

        // Allocate memory to hold a message, take care of the ending nul char
        GetMem(Buffer, SizeOf(Char) * (MaxLen + 1));
        // Copy message to allocated memory, protecting overflow
        StrLCopy(Buffer,
                 PChar('Thread=' + IntToStr(Id) +
                       ' Count=' + IntToStr(I) +
                       ' ThreadID=' + IntToStr(GetCurrentThreadID)),
                 MaxLen);
        // Force a nul char at the end of buffer
        Buffer[MaxLen] := #0;
        // Post a message to the main thread which will display
        // the message and then free memory
        MsgSys.PostMessage(WM_SHOW_MESSAGE, I, LParam(Buffer));
    end;
end;

end.

This source code is really simple, isn’t it? The beauty is that it can be compiled for Win32, Win64 and Android targets without changing anything.

All the code depending on the platform has been moved to “FMX.Overbyte.MessageHandling” unit. That one takes care of calling the correct API function according to the compiler used. This is the power of OOP.

There is nothing special in the demo application except one thing: The worker thread generates messages to be displayed by the main thread. We have to take care of what happens with the storage used for the message. We cannot simply pass a string because messages are limited to two parameters of type WParam and LParam, both mapped to NativeInt. We can neither pass a reference to a string variable because it is possible a new message is generated before the previous is consumed (This happens if the main thread is heavily busy while the worker thread runs at full speed). We have to dynamically allocate storage for the message and pass the reference thru one of the message parameters. I’ve chosen to use a simple memory block allocated by GetMem and freed by FreeMem. The pointer is then passed thru the LParam parameter. The thread allocates the memory and the main thread frees it. The same allocation size is always used regardless of the message length. It is better for the memory allocator, limiting memory fragmentation.

How to use it?

TMessagingSystem class must be instantiated when the form is allocated a handle. It must be freed when the form’s handle is destroyed. After instantiation, or at any point in time, RegisterMessageHandler must be called for each custom message. That’s all!

Single unit, multiple platforms

We have seen in the demo code that the same unit to “FMX.Overbyte.MessageHandling” is used whatever the target platform is. The magic is in that unit. Here is very short source code:

unit FMX.Overbyte.MessageHandling;
{$DEFINE OVERBYTE_INCLUDE_MODE}
{$IFDEF ANDROID}
    {$I FMX.Overbyte.Android.MessageHandling.pas}
{$ENDIF}
{$IFDEF MSWINDOWS}
    {$I FMX.Overbyte.Windows.MessageHandling.pas}
{$ENDIF}

The magic is into the conditional compilation. Symbols ANDROID and MSWINDOWS are automatically defined by the compiler according to the target platform you compile for. So that small unit actually includes the Android or the Windows specific unit depending on the compiler target platform.

The two included units are just normal unit, well almost. You cannot include a unit into another one without having a problem with the “unit” line. You cannot have two such lines. This is why the symbol “OVERBYTE_INCLUDE_MODE” is defined. In the two included units, this symbol is used to conditionally compile the “unit” line.

Implementation for Android

Messaging system on Android platform is hidden in the “Looper” API. Basically, the idea is simple: Android monitors a list of handle for data availability. The list of handles is maintained by the API. You can add a new handle using ALooper_addFd API function. Each handle is associated with a callback function that Android calls when data is available.

As a handle, I use the read side of a pipe. A pipe, under Android as well as other operating systems, is like a first-in first-out queue. It has two ends identified by two handles. One is the writing end; the other is the reading end. What you write at one end is available for reading at the other end. Between both ends is a buffer. Reads and writes are asynchronous. If writing is faster than reading, the buffer is filled and nothing is lost.

This pipe is used here is the message queue. When PostMessage is called, a record with the parameters is written to the pipe. When data is available for reading, the looper API will call the LooperCallBack function we registered. From this callback, we read the pipe to remove one record at a time. When a record is read, the message number written in it is used to fetch the message handler to be executed.

{$IFNDEF OVERBYTE_INCLUDE_MODE}
unit FMX.Overbyte.Android.MessageHandling;
{$ENDIF}

interface

uses
    System.SysUtils, System.Types, System.Classes, System.SyncObjs,
    Generics.Collections,
    FMX.Platform.Android,
    Androidapi.AppGlue, Androidapi.Looper,
    Posix.UniStd, Posix.Errno, Posix.StrOpts, Posix.PThread;

const
    WM_USER         = 1024;

type
    LPARAM  = NativeInt;
    WPARAM  = NativeInt;
    LRESULT = NativeInt;

    TMessage = record
        Msg    : NativeInt;
        WParam : WPARAM;
        LParam : LPARAM;
        Result : LRESULT;
    end;
    TMessageHandler = procedure (var Msg: TMessage) of object;

    TMessagingSystem = class(TComponent)
    protected
        FPipeFD    : TPipeDescriptors;
        FData      : Byte;
        FHandlers  : TDictionary;
        FLastError : String;
        FCritSect  : TCriticalSection;
        procedure HandleMessage(var Msg : TMessage);
        function  CreatePipe: Integer;
        procedure ClosePipe;
        procedure InstallEventHandler;
        procedure UninstallEventHandler;
    public
        constructor Create(AOwner : TComponent); override;
        destructor  Destroy; override;
        function RegisterMessageHandler(uMsg    : NativeInt;
                                        Handler : TMessageHandler) : Boolean;
        function PostMessage(uMsg   : NativeInt;
                             WParam : WPARAM;
                             LParam : LPARAM) : Boolean;
        property LastError : String read FLastError;
    end;

    HWND   = TMessagingSystem;

function GetCurrentThreadID : TThreadID;

implementation

function LooperCallback(
    FileDescriptor : Integer;
    Events         : Integer;
    Data           : Pointer): Integer; cdecl;
var
    Len : Integer;
    Msg : TMessage;
    Obj : TMessagingSystem;
begin
    Result := 1;
    // Data contains a reference to our class
    if Data = nil then
        Exit;
    // Ready to cast to our class
    Obj := TMessagingSystem(Data);
    // Check if it's our ReadDes
    Obj.FCritSect.Enter;
    try
        if FileDescriptor <> Obj.FPipeFD.ReadDes then
            Exit;
    finally
        Obj.FCritSect.Leave;
    end;

    while TRUE do begin
        Len := __read(FileDescriptor, @Msg, SizeOf(Msg));
        if Len <= 0 then
            break;
        Obj.HandleMessage(Msg);
    end;
end;

{ TMessagingSystem }

constructor TMessagingSystem.Create(AOwner: TComponent);
begin
    inherited Create(AOwner);
    FCritSect  := TCriticalSection.Create;
    FHandlers  := TDictionary.Create;
    CreatePipe;
    InstallEventHandler;
end;

destructor TMessagingSystem.Destroy;
begin
    UninstallEventHandler;
    ClosePipe;
    FreeAndNil(FCritSect);
    inherited Destroy;
end;

function TMessagingSystem.CreatePipe: Integer;
var
    Status  : Integer;
    Val     : Integer;
const
    FIONBIO = $5421;
begin
    FCritSect.Enter;
    try
        if (FPipeFD.ReadDes <> 0) or (FPipeFD.WriteDes <> 0) then begin
            FLastError := 'Pipe already created';
            Result := -1;
            Exit;
        end;
        Status := Pipe(FPipeFD);
        if Status = -1 then begin
            Result := errno;
            FLastError := 'Pipe() failed. Error #' + IntToStr(Result);
        end
        else begin
            Result := 0;
            Val := 1;
            if ioctl(FPipeFD.ReadDes, FIONBIO, @Val) = -1 then begin
                Result := errno;
                FLastError := 'ioctl(FIONBIO) failed. Error #' + IntToStr(Result);
                Exit;
            end;
        end;
    finally
        FCritSect.Leave;
    end;
end;

procedure TMessagingSystem.ClosePipe;
begin
    FCritSect.Enter;
    try
        if FPipeFD.ReadDes <> 0 then begin
            __close(FPipeFD.ReadDes);
            FPipeFD.ReadDes  := 0;
        end;
        if FPipeFD.WriteDes <> 0 then begin
            __close(FPipeFD.WriteDes);
            FPipeFD.WriteDes := 0;
        end;
    finally
        FCritSect.Leave;
    end;
end;

procedure TMessagingSystem.InstallEventHandler;
var
    AndroidApp : PAndroid_app;
    Data       : Pointer;
const
    LOOPER_ID_MESSAGE_OVERBYTE = LOOPER_ID_USER;
begin
    AndroidApp := GetAndroidApp;

    Data := Self;
    ALooper_addFd(AndroidApp.looper,
                  FPipeFD.ReadDes,
                  LOOPER_ID_MESSAGE_OVERBYTE,
                  ALOOPER_EVENT_INPUT,
                  LooperCallback,
                  Data);
end;

procedure TMessagingSystem.UninstallEventHandler;
var
    AndroidApp : PAndroid_app;
begin
    FCritSect.Enter;
    try
        if FPipeFD.ReadDes <> 0 then begin
            AndroidApp := GetAndroidApp;
            ALooper_removeFd(AndroidApp.looper, FPipeFD.ReadDes);
        end;
    finally
        FCritSect.Leave;
    end;
end;

function TMessagingSystem.RegisterMessageHandler(
    uMsg    : NativeInt;
    Handler : TMessageHandler): Boolean;
begin
    FCritSect.Enter;
    try
        FHandlers.AddOrSetValue(uMsg, Handler);
    finally
        FCritSect.Leave;
    end;
    Result := TRUE;
end;

function TMessagingSystem.PostMessage(
    uMsg   : NativeInt;
    WParam : WParam;
    LParam : LParam): Boolean;
var
    Msg : TMessage;
begin
    Result := FALSE;
    FCritSect.Enter;
    try
        if FPipeFD.WriteDes = 0 then begin
            FLastError := 'Pipe is not open';
            Exit;
        end;
        Msg.Msg    := uMsg;
        Msg.WParam := WParam;
        Msg.LParam := LParam;
        Msg.Result := 0;

        if __write(FPipeFD.WriteDes, @Msg, SizeOf(Msg)) = -1 then begin
            FLastError := 'write() failed. ErrCode=' + IntToStr(errno);
            Exit;
        end;
    finally
        FCritSect.Leave;
    end;
    Result := TRUE;
end;

procedure TMessagingSystem.HandleMessage(var Msg: TMessage);
var
    Handler : TMessageHandler;
    Status  : Boolean;
begin
    FCritSect.Enter;
    try
        Status := FHandlers.TryGetValue(Msg.Msg, Handler);
    finally
        FCritSect.Leave;
    end;
    if Status then
        Handler(Msg);
end;

function GetCurrentThreadID : TThreadID;
begin
    Result := Posix.PThread.GetCurrentThreadID;
end;

end.

In that code, you’ll find a few data types frequently used in Windows applications. I used the same data types for compatibility with existing code.

TMessagingSystem class is very simple. Basically, it registers a pipe read handle with the looper API with an associated callback function. It also maintains a dictionary of message handlers. The key is the message number. The looper API also carries one pointer for you. It will give it back as an argument of the callback function. Here the pointer is used as a reference to the class instance, making is available when the callback function is called.

A critical section is used to avoid problems accessing the class data from several threads at the same time. Using this critical section makes the class fully thread safe.

Implementation for Windows

The Windows implementation makes obviously use of Windows own messaging API. There is no queue in the class because Windows queue is used.

FireMonkey forms does not provide any support for custom messages. This is not really a problem because a FireMonkey forms are just a Windows window. As any window, a FireMonkey form running on Windows has a HWND (Handle of WiNDow) and a window procedure handling all messages for the window.

To hook into this system, we must use standard Windows programming. By standard I mean it has always existed as far as I remember. What we need is to “subclass” the window. And surprisingly, this is very easy!

Windows internally maintain a structure for each window. In that structure you have all informations required for Windows to handle the window. This includes the pointer to the window procedure.

And Windows provides a function to access his internal structure. Our problem is just to get the current pointer to the window procedure and replace it with a pointer to our own procedure. From our own procedure, we will call the original procedure, or not. Our own window procedure has access to all messages sent/posted to the window, including those we add.

We have just one small problem: Windows does not know anything about a Delphi class instance. A window procedure is a simple procedure, not an object method. The problem is to get hand on our TMessagingSystem class instance from our own window procedure.

Fortunately Windows is incredibly well designed. We, as developer, can associate with any window a small piece of data called an “Atom” in Windows terminology. Once an “Atom” is created (It just has a name), you can associate the atom with any window along with a piece of data. That piece of data will be the reference to our TMessagingSystem class instance.

When called by Windows, our window procedure receives the handle of the window. We use it to fetch the piece of data we associated using the atom. From there we have access to TMessagingSystem class instance and check for the message to handle. if it is one of our registered messages, we just call the handler. If not one of our messages, the the original window procedure is called.

Here is the source code:

{$IFNDEF OVERBYTE_INCLUDE_MODE}
unit FMX.Overbyte.Windows.MessageHandling;
{$ENDIF}

interface

uses
    WinApi.Windows, WinApi.Messages,
    System.Classes, System.SysUtils, System.SyncObjs,
    Generics.Collections,
    FMX.Forms, FMX.Platform.Win;

const
    WM_USER = WinApi.Messages.WM_USER;

type
    TMessage        = WinApi.Messages.TMessage;
    WPARAM          = WinApi.Windows.WPARAM;
    LPARAM          = WinApi.Windows.LPARAM;
    TMessageHandler = procedure (var Msg: TMessage) of object;
    TWndProc        = function (hwnd   : HWND;
                                uMsg   : UINT;
                                wParam : WPARAM;
                                lParam : LPARAM): LRESULT; stdcall;

    TMessagingSystem = class(TComponent)
    protected
        FHWnd             : HWND;
        FHandlers         : TDictionary;
        FOriginalWndProc  : TWndProc;
        FLastError        : String;
        FCritSect         : TCriticalSection;
    public
        constructor Create(AOwner : TComponent); override;
        destructor  Destroy; override;
        function RegisterMessageHandler(uMsg    : NativeInt;
                                        Handler : TMessageHandler) : Boolean;
        function PostMessage(uMsg   : NativeInt;
                             WParam : WPARAM;
                             LParam : LPARAM) : Boolean;
        property LastError : String read FLastError;
    end;

function GetCurrentThreadId: DWORD; stdcall;

implementation

var
  MsgSysAtom       : TAtom;
  MsgSysAtomString : String;


function WndProc(hwnd: HWND; uMsg: UINT; wParam: WPARAM; lParam: LPARAM): LRESULT; stdcall;
var
    Msg     : TMessage;
    MsgSys  : TMessagingSystem;
    Handler : TMessageHandler;
    Status  : Boolean;
begin
    // Search if the window handle is associated with TMessageingInstance
    // We know this because we registered an atom for that purpose
    if GlobalFindAtomW(PChar(MsgSysAtomString)) <> MsgSysAtom then begin
        // Not found, just do default processing
        Result := DefWindowProc(hwnd, uMsg, wParam, lParam);
        Exit;
    end;
    // Fetch the atom property and cast it to a TMessagingSystem class
    MsgSys := TMessagingSystem(GetProp(hwnd, MakeIntAtom(MsgSysAtom)));

    // Now use the dictionary to see if the message is one we'll handle
    MsgSys.FCritSect.Enter;
    try
        Status := MsgSys.FHandlers.TryGetValue(uMsg, Handler);
    finally
        MsgSys.FCritSect.Leave;
    end;
    if Status then begin
        // Found the message and his message handler. Call it using
        // the TMessage record to hold the values
        Msg.Msg    := uMsg;
        Msg.WParam := wParam;
        Msg.LParam := lParam;
        Msg.Result := 0;
        Handler(Msg);
        Result := Msg.Result;
    end
    else begin
        // Not one of our messages, just execute original window procedure
        Result := MsgSys.FOriginalWndProc(hwnd, uMsg, wParam, lParam);
    end;
end;

{ TMessagingSystem }

constructor TMessagingSystem.Create(AOwner: TComponent);
begin
    if not (AOwner is TCommonCustomForm) then
        raise Exception.Create('TMessagingSystem.Create failed. Invalid owner');
    inherited Create(AOwner);
    FCritSect  := TCriticalSection.Create;
    FHandlers  := TDictionary.Create;

    // Find window handle corresponding to the owner form
    FHWnd := WindowHandleToPlatform(TCommonCustomForm(AOwner).Handle).Wnd;

    // If not already done, register the atom we'll use to associate
    // our messaging system with the window handle
    if MsgSysAtom = 0 then begin
        MsgSysAtomString := 'OverbyteMessagingSystem' +
                                     IntToHex(GetCurrentProcessID, 8);
        MsgSysAtom       := GlobalAddAtomW(PChar(MsgSysAtomString));
    end;

    // Associate our messaging system with the window handle
    SetProp(FHWnd, MakeIntAtom(MsgSysAtom), THandle(Self));

    // Subclass the form. That is change his handling procedure
    FOriginalWndProc := TWndProc(GetWindowLongPtr(FHWnd, GWLP_WNDPROC));
    SetWindowLongPtr(FHWnd, GWLP_WNDPROC, NativeInt(@WndProc));
end;

destructor TMessagingSystem.Destroy;
begin
    if Assigned(FOriginalWndProc) then begin
        SetWindowLongPtr(FHWnd, GWLP_WNDPROC, NativeInt(@FOriginalWndProc));
        FOriginalWndProc := nil;
    end;
    FreeAndNil(FHandlers);
    FreeAndNil(FCritSect);
    inherited Destroy;
end;

function TMessagingSystem.RegisterMessageHandler(
    uMsg    : NativeInt;
    Handler : TMessageHandler): Boolean;
begin
    FCritSect.Enter;
    try
        FHandlers.AddOrSetValue(uMsg, Handler);
    finally
        FCritSect.Leave;
    end;
    Result := TRUE;
end;

function TMessagingSystem.PostMessage(
    uMsg   : NativeInt;
    WParam : WPARAM;
    LParam : LPARAM): Boolean;
begin
    Result := WinApi.Windows.PostMessage(FHWnd, uMsg, WParam, LParam);
end;

function GetCurrentThreadId: DWORD; stdcall;
begin
    Result := WinApi.Windows.GetCurrentThreadId;
end;

end.

All the code is shown above. If you are interested by the complete project as source code, just drop me a private email.


Follow me on Twitter
Follow me on LinkedIn
Follow me on Google+
Visit my website: http://www.overbyte.be
This article is available from http://francois-piette.blogspot.be

Inter Process Communication Using Pipes

A pipe is a communication channel between two ends. It is mostly used to communicate between processes running within a computer. As such it is an Inter Process Communication (IPC) mechanism.

The concept of pipe is well known in the Linux (Unix) world. It is used on the command line to direct the output of a command as input of the next command. The so called “pipe character” is used as a syntax gadget for that purpose. Windows is using the same syntax.

More generally, a pipe is an operating system object which is close to a file. An application opens a pipe, then read and writes data and finally closes it. The difference between a pipe and a file is that data is kept in memory and the pipe has two ends. The same or different processes may open the same pipe. What is written at one end becomes readable at the other end. This is why it is named a “pipe”: it fact as a pipe or tunnel between the two ends, data flows from one end and the other. Unlike pipes in the real world, computer pipes are bi-directional. One computer pipe is actually made of two distinct sub-pipes: one for each direction.

We can think of pipes as client/server architecture. The server side opens a pipe and waits for data to be available. The client side opens the pipe and then data may flow into the pipe in both directions independently. If several clients open the same pipe, the server sees it as several independent pipes. At server side, data may be written to individual client’s pipes or broadcasted to all clients.

As seen from Windows API, a pipe can be created with or without a name. Giving a name to a pipe allows processes to share the same pipe easily. When using anonymous pipes, a process knows about it because it has inherited his handle.

The process that creates a pipe is the pipe server. A process that connects to a pipe is a pipe client. One process writes data to the pipe, and then the other process reads the data from the pipe. This is bidirectional: each process can write data that will be read by the other. Read and writes are asynchronous. It means a lot of data may be written before it is read at the other end. Data is stored by the operating system into shared memory. The size of the shared memory area is managed by the operating system using advisory values passed to the pipe’s creation function.

Windows pipe API is relatively complex. It has been encapsulated into Delphi classes by Russell Libby in 2003. Although Russell website is no more available, his code is still around in the internet and has been enhanced by several peoples. I have updated his code for Delphi XE3 and made it available from my website at http://www.overbyte.be/eng/blog_source_code.html

Russell wrote a unit named “pipes.pas” containing 3 components:
- TPipeClient: client component
- TPipeServer: Server component
- TPipConsole: console pipe redirection component

For your convenience, I created two packages: one runtime package and one design time package. The components are installed in the component palette under “Pipes” tab.
Using the components is fairly simple. I made two simple demo programs: PipeClient and PipeServer. You run PipeServer first and then one or more PipeClient instances. PipeClient has an edit box and a “Send” button. When you click on “Send”, the edit box content is sent to the server which in turn displays it in a memo. At server side, the “send” button is replaced by “broadcast” to send the same message to all connected client, if any. Of course you may also send to a specific client.

To send the messages, I actually send strings, taking care of Unicode. When sending (Write or broadcast method), you have to pass a pointer to the first byte to be sent and a count of bytes. At receiver side, data is read from a stream which is also made of bytes.

Warning: a pipe is a byte oriented communication channel. There is no guarantee that each single write at one end will correspond to a single read at the other end. Sending strings has I’ve done in my demo is good for interactive application where a user clicks on a button. In a real world application, you have to design a protocol, just like you do when using socket, so that the reader knows exactly what the writer sent. For example, you send a message made of a byte count and the then actual bytes. Or you design a “line oriented” protocol much like most TCP/IP protocol are. Your messages are made you line of text terminated by a CRLF pair. The reader knows he received a complete message because he received the CRLF pair.

To say it in other words, a pipe is like a TFileStream. The read doesn’t know how the stream has been written. He knows how to read it, for example he knows the stream is made of text lines.

UPDATE 2013-10-04: arno.garrels@gmx.de added 64-bit support and fixed code to compile with Delphi 7 to XE5 (earlier versions may compile however untested).

Follow me on Twitter
Follow me on LinkedIn
Follow me on Google+
Visit my website: http://www.overbyte.be
This article is available from http://francois-piette.blogspot.be
Download source code from my website at
      http://www.overbyte.be/eng/blog_source_code.html

High speed generic queue class

This article presents a generic class implementing a high speed queue. It has been specially designed for communication use but can be used for anything else.

Reading this article, you'll learn:

• How to design a generic class
• How to use nested data type
• How to implement an enumerator
• How to make a class thread safe
• How to use variant record
• How to use methods in a record

The code itself is quite short! Thanks to the efficient Delphi language.


High Performance

About high speed: Initially, I had an issue with performance in a communication system where a process was producing messages to be handled and another process was processing the messages. The two processes are totally asynchronous. I use the term "process" not to refer to different programs, but to different units of a single program.

The bottle neck was coming from memory allocation. Record where allocated by the producer process and freed by the processing processes. This resulted in a large amount of memory being allocated freed.

I designed this class to avoid as much as possible memory allocation. The design makes use of a several doubly linked lists of "buffers". The buffers are actually the data type specified by the generic type. Usually it is a record with a structure specific to the messages exchanged between the two process. If the application requires several types of buffers, then you should use a variant record to hold all types. We will see an example later.

High speed comes also from the fact that you don't copy data except when really forced to. To avoid copying data, we use pointers. Data stay where it is, we access it thru his address. The compiler takes care of the details for us.


Doubly linked list as a queue

A queue is a data structure in which you can append items (aka "node") at one end and remove items at the other end. This scheme is frequently named "First in - First out" or FIFO for short. You have another variant which append and remove items at the same end. That one is named LIFO which stands for "Last in - First out". In the application I described above, I need a FIFO queue.

FIFO queues can be implemented in a lot of different ways. Here I selected a doubly linked list because in my context this is the fastest way doing it.

A doubly linked list is a very classic data structure. Basically a doubly linked list is a variable number of items (or nodes) of any data type. One item is linked to the next using a pointer and linked to the previous using another pointer. There are two more separate pointers: one point to the first item and one point to the last item. To append items, we use the pointer to the last item and to remove items, we use the pointer to the first item.

You can find a lot of articles about it everywhere on the internet. So I will only describe what make my implementation specific.


High performance revisited

Now that we have a doubly linked list in hand, we will use it to achieve high performance. We will use 3 linked lists to avoid memory allocation:

• Linked list of active items
• Linked list of inactive items
• Linked list of acquired items

Active items are those referring to the items added to the queue and not removed yet from the queue. They are actively waiting to be processed.

Inactive items are those having already be processed. Instead of freeing the processed items, we put it in the inactive queue where they are waiting for reuse. When a new active item has to be created, before allocating one new item, the inactive linked list is checked and if not empty, an item is removed from the list and reused, that is moved to the end of the active list. If the inactive list is empty, then a new item is allocated. This has a tremendous impact on the performance: adding a large number of items take 30 mS the first time when they need to be allocated. Later, when they can be reused, it only takes 17 mS!

Finally, the acquired items linked list is used when there are several processes consuming the same queue. For example when we have a multithread application and several threads are fetching messages from the same active queue. The third list is used to store one item extracted from the active list while it is processed so that another thread won't take the same item. After it has been processed, the "acquired" item is moved to the inactive list using a method I named Release.

Generic class use

A generic class is a class which has a variable part not known when designing the class. In our case, the variable part is a record. I named the generic class TCommQueue.

To use TCommQueue, you have to declare a record type and a variable. Let's assume the record type is:

  TCommFrame = packed record
    Name : array [0..50] of Char;
    Size : Integer;
  end;

This is a totally arbitrary record made simple for this article. In a real application, this record will be much more complex with all the members required for your application.

The variable representing the queue is declared as follow:

    FQueue : TCommQueue<TCommFrame>;

It is probably a member variable of a form in your application or in a component, data module or whatever you need. It doesn't matters in fact. TCommQueue inherit from TObject so before use, you must create it and after use you must free it. In most applications, you will do that in the constructor and destructor of your form, data module or component. A more complete source code is as follow:

type
    TForm1 = class(TForm)   
    protected
        FQueue : TCommQueue<TCommFrame>;
    public
        constructor Create(OAwner: TComponent); override;
        destructor Destroy; override
    end;

implementation

constructor TForm1.Create(AOwner : TComponent);
begin
    inherited;
    FQueue := TCommQueue<TCommFrame>.Create;
end;

destructor TForm1.Destroy;
begin
    FreeAndNil(FQueue);
    inherited;
end;

You see that using a generic type is very easy: you use the generic type name (TCommQueue here) and append your own data type in angle brackets ( here).

The compiler will compile the generic class by replacing the parameter type with the one you supplied.


Generic class declared

Now it's time to see the declaration of TCommQueue generic class. Remember we are using doubly linked list which are defined here as nested data type.

Nested data types have been introduced with Delphi XE3. See online help at http://docwiki.embarcadero.com/RADStudio/XE3/en/Nested_Type_Declarations

Nested data types are not related to generics, but are very handy in that case because nested data types within a generic class may make use on the parameter data type.

type
    // Forward declarations
    TCommQueue<T : record>     = class;
    TCommQueueEnum<T : record> = class;

    TInitCallBack = procedure (Sender : TObject;
                               Item   : Pointer) of object;

    TCommQueue<T : record> = class(TObject)
    // This nested type actually defines a doubly linked list of <T>
    type
        TItemPtr = ^T;          // Pointer to <T>
        PLLNode  = ^TLLNode;    // Pointer to a linked list node
        TLLNode  = record       // Doubly linked list cell
            Item : T;           // Item /must/ be the first node member
            Next : PLLNode;     // Pointer to the next node
            Prev : PLLNode;     // Pointer to the previous node
        end;
        TLinkedList = record    // Doubly linked list
            First : PLLNode;    // Pointer to the first node in the list
            Last  : PLLNode;    // Pointer to the last node in the list
            Count : Integer;    // Number of nodes in the list
            procedure Add(Item : PLLNode);
            function  ExtractLast : PLLNode;
            procedure Extract(Item : PLLNode);
            procedure FreeAllItems; inline;
        end;
    protected
        FActiveList   : TLinkedList;      // List of active items
        FInactiveList : TLinkedList;      // List of inactive items
        FAcquiredList : TLinkedList;      // List of acquired items
        FCurrent      : TItemPtr;         // The current item
        FCritSection  : TCriticalSection; // For multithread support
        // Enum is called back from the enumerator class
        // Give an item pointer, it returns the next item pointer or nil
        function  Enum(Item : Pointer) : Pointer;
        // GetItem is called back from the enumerator class
        // Get an item from the list. The item is COPYED while all other
        // functions make use of pointers.
        function  GetItem(ItemPt : Pointer) : T;
    public
        constructor Create;
        destructor Destroy; override;
        // Required method to supprt for..in contruct (not thread safe)
        function  GetEnumerator : TCommQueueEnum;
        // Add a new item, reusing an old one of any and return a pointer to it
        function  Add : TItemPtr;
        function  Add(InitCallBack : TInitCallBack) : TItemPtr; overload;
        // Remove an item and return pointer to the next
        // Current is not affected unless it is the one removed
        function  Remove(Item : TItemPtr) : TItemPtr; overload;
        // Remove current item and return a pointer to the next
        // The current becomes the next
        function  Remove : TItemPtr; overload;
        // Return pointer to first item. Update current.
        function  First : TItemPtr;
        // Return pointer to last item. Update current.
        function  Last : TItemPtr;
        // Return pointer to next item. Update current.
        function  Next : TItemPtr; overload;
        // Return a pointer to the item after a given one. Update current.
        function  Next(Item : TItemPtr) : TItemPtr; overload;
        // Return pointer to previous item. Update current.
        function  Previous : TItemPtr; overload;
        // Return a pointer to the item before a given one. Update current.
        function  Previous(Item : TItemPtr) : TItemPtr; overload;
        // Acquire the first item, if any and remove it from the list
        // it must be later be released by calling ReleaseItem
        // Acquire Item doesn't affect current unless it is the one acquired
        function  AcquireItem : TItemPtr;
        // Release an item previously acuired. The item is moved to the
        // inactive list for later reuse
        procedure ReleaseItem(Item : TItemPtr);
        // Free all items in the inactive List
        procedure FreeAllInactiveItems;

        property Current       : TItemPtr      read FCurrent;
        property Count         : Integer       read FActiveList.Count;
        property CountInactive : Integer       read FInactiveList.Count;
        property CountAcquired : Integer       read FAcquiredList.Count;
    end;

    // Class to support for..in construct with the main TCommQueue<T> class
    // Warning: for..in is not really thread safe
    TCommQueueEnum<T : Record> = class
    protected
        FQueue : TCommQueue<T>;
        FIndex : Pointer;
    public
        constructor Create(AQueue : TCommQueue<T>);
        function GetCurrent: T;
        function MoveNext: Boolean;
        property Current: T read GetCurrent;
    end;

The generic class TCommQueue takes one parameter which is the data type you want to use for the items. Instead of I could have used. The difference is that the later allow any data type to be used for T while specifying "record" impose the constraint of have a non nullable data type, mostly a record but also a scalar type such as integer, char and the likes. You cannot use string, array, class nor interface. I imposed such restriction because memory is allocated/freed using new/dispose.

The doubly linked list is declared as a nested data type at line 19. It is a record named TLinkedList. It contains two pointers to each end of the linked list, and a count of items in the list. I added a count because counting elements in a linked list is expensive: you have to iterate thru entire list to count how many node it has. Maintaining a counter is a useful optimization which doesn't cost much.

The nodes or items of the linked list are made of a record named TLLNode defined. As you can see at line 15, it makes use of the parameter type to build the record. Remember, in a doubly linked list, a node is made of the actual data you want in the list plus two pointers to the next and previous node in the list.

Lines 23 to 26 are method declarations for the record. Their implementation handles adding (append) or extracting (remove) nodes. This is only a matter of manipulation the pointers, no data is actually allocated, moved nor freed. This is really fast!

At line 29, 30 and 31 are declared the 3 doubly linked lists we talked before. Since TLinkedList are records, there is no need to allocate/free memory.

Line 31 declares FCurrent a pointer to an item. It is used in conjunction with the methods First, Next, Previous and Last that I implemented to iterate thru the list. Please note that those are NOT thread safe.

For the fun, I implemented an enumerator for the queue. Please refer to my article "Writing an iterator for a container" for details: http://francois-piette.blogspot.be/2012/12/writing-iterator-for-container.html
The enumerator is NOT thread safe and copies the data which is bad for performance.

The real interesting stuff is in methods Add, AcquireItem and ReleaseItem. They are all fully thread safe.

Add will add an item to the queue (Active linked list), reusing an inactive one or allocating a new item. Add returns a pointer to the item. There are two overloaded versions of method add. One without argument and one with a method pointer argument.

The version with the method pointer argument (line 48) is required to be fully thread safe. Allocating a new item in the queue may require initialization of the item. But initialization must take place before the item is visible in the queue otherwise another thread could remove it before it is initialized. The method pointer passed as argument is used as a callback (a kind of event if you like) which will be called within add method just after the new item is allocated or extracted from inactive list and just before it is actually appended to the active list.

A last note about multithreading: we handle linked list pointers in the implementation. To be thread safe, the handling must make sure only one thread at a time can update any of the linked lists. This is why I used a critical section declared at line 33.


Implementation

The implementation is rather boring. It is actually very basic Delphi programming. I give the source below to be complete. Drop a message if you need some more explanations.
Complete source code is available at http://www.overbyte.be/eng/blog_source_code.html


Follow me on Twitter
Follow me on LinkedIn
Follow me on Google+
Visit my website: http://www.overbyte.be

The article is available at:
     http://francois-piette.blogspot.com/2013/02/high-speed-generic-queue-class.html