YOUSEFUL / DELPHI / SAMPLE PROJECTS

Threaded asynchronous logging

Console program Self-checking Product: Atlas Log · Components: TAtlasLogger, TAtlasAsyncSink, TAtlasFileSink, TAtlasCallbackSink

A console program that goes deeper than the first logger: several threads, a background writer with a bounded queue, and what happens when the target cannot keep up. Every result is checked.

What it demonstrates

Part 1, four threads and one JSON file:

Four worker threads log 500 debug events and one warning each through one TAtlasLogger, each inside its own BeginScope.
A TAtlasAsyncSink (Capacity 256, Overflow := ofBlock) queues the events and writes them on its background thread to a TAtlasFileSink with Layout := loJson. The file sink is only the Target; it is not attached to the logger itself.
A TAtlasCallbackSink with MinLevel := llWarning whose OnEvent handler runs on the thread that logged.
A custom sink, TLevelCounter, written by overriding TAtlasLogSink.DoWrite.
Log.Flush waits until the queue is empty and the file is flushed. The program then parses all 2004 lines with System.JSON and checks that nothing was dropped, sequence numbers are unique, and each thread's steps are in order under that thread's scope.

Part 2, a target that is held back:

A second TAtlasAsyncSink with Capacity 10 writes to a callback sink whose handler waits on an event, so the queue fills.
ofDropNewest discards what does not fit (19 or 20 of 30 events, depending on whether the background thread had already taken the first one) and counts them in Dropped; the oldest events are written.
ofDropOldest discards from the front and keeps the newest event.
A target that raises (a TAtlasFileSink without a FileName) is counted in TargetErrors; nothing is raised into the program and the logger's SinkErrors stays 0.

Build and run

Open ThreadedAsyncLogging.dpr in Delphi 13; the IDE creates the project file. Add the Atlas Log src folder to the project's search path, then build and run for Win32 or Win64. No packages need to be installed. The program needs no input. It checks its own results and stops with an error if any of them differs from the output below.

It writes to a folder under %TEMP% and removes it at the end.

Expected output

The thread id and timestamp differ, and the two drop counts may be 19 or 20:

Part 1: four threads, one asynchronous JSON file
  ok: Flush emptied the queue and ofBlock dropped nothing
  ok: the callback sink saw one warning per worker
  ok: the custom sink counted every event by level
  ok: 2004 JSON lines in workers.jsonl
  ok: each thread's events stay in order and carry that thread's scope only
    {"ts":"2026-09-22T01:37:46.320Z","seq":2004,"level":"debug","category":"worker","thread":25760,"scope":"worker 2","msg":"Worker 2 step 500","template":"Worker {worker} step {step}","props":{"worker":"2","step":"500"}}
Part 2: a stuck target and the overflow policies
  ok: ofDropNewest: 19 of 30 events dropped while the target was stuck
  ok: the rest were written, oldest first
  ok: ofDropOldest: 19 dropped from the front, the newest event kept
  ok: a failing target is counted in TargetErrors and never raised
All checks passed.

Exit code 0 means every check passed; a failed check prints Exception: Check failed: ... and sets exit code 1.

Adapting it

Keep Part 1's wiring for a real application: the file sink as Target, the asynchronous sink attached to the logger. Choose the overflow policy by what matters more: ofBlock (the default) never loses an event but can slow the logging threads when the disk stalls; ofDropNewest and ofDropOldest never block and report losses in Dropped. Call Flush (or free the asynchronous sink) before the program ends so the queue reaches the file.

Project files

Download all 34 projects

The ZIP contains every sample project with its README. See getting started for the search path to add.

Source

ThreadedAsyncLogging.dpr

program ThreadedAsyncLogging;

// Atlas Log in depth. Four threads log through one logger. An asynchronous
// sink with a bounded queue writes their events to a JSON-lines file on a
// background thread; a callback sink watches for warnings on the logging
// threads; a custom sink counts events per level. A second part holds a slow
// target back to show the drop-newest and drop-oldest overflow policies and
// the Dropped and TargetErrors counters. Exit code 0 means every check passed.

{$APPTYPE CONSOLE}

uses
  System.SysUtils, System.Classes, System.SyncObjs, System.IOUtils, System.JSON,
  System.Generics.Collections, Atlas.Log, Atlas.LogSinks;

const
  Workers = 4;
  Steps = 500;

type
  // A custom sink. DoWrite runs under the sink's own lock, so it needs no locking.
  TLevelCounter = class(TAtlasLogSink)
  protected
    procedure DoWrite(const Event: TAtlasLogEvent); override;
  public
    Counts: array[TAtlasLogLevel] of Integer;
  end;

  // Handlers for the callback sinks.
  THandlers = class
  public
    Warnings: Integer;
    Gate: TEvent;
    Written: TList<string>;
    constructor Create;
    destructor Destroy; override;
    procedure WarningSeen(Sender: TObject; const Event: TAtlasLogEvent);
    procedure SlowWrite(Sender: TObject; const Event: TAtlasLogEvent);
  end;

procedure TLevelCounter.DoWrite(const Event: TAtlasLogEvent);
begin
  Inc(Counts[Event.Level]);
end;

constructor THandlers.Create;
begin
  inherited;
  Gate := TEvent.Create(nil, True, True, '');
  Written := TList<string>.Create;
end;

destructor THandlers.Destroy;
begin
  Written.Free;
  Gate.Free;
  inherited;
end;

procedure THandlers.WarningSeen(Sender: TObject; const Event: TAtlasLogEvent);
begin
  // Runs on the thread that logged; calls into one sink never overlap.
  Inc(Warnings);
end;

procedure THandlers.SlowWrite(Sender: TObject; const Event: TAtlasLogEvent);
begin
  // Runs on the asynchronous sink's background thread; waits while the gate is closed.
  Gate.WaitFor(INFINITE);
  Written.Add(Event.Message);
end;

procedure Check(Condition: Boolean; const What: string);
begin
  if not Condition then raise Exception.Create('Check failed: ' + What);
  Writeln('  ok: ', What);
end;

function ReadLines(const FileName: string): TArray<string>;
var Stream: TFileStream; Bytes: TBytes;
begin
  Stream := TFileStream.Create(FileName, fmOpenRead or fmShareDenyNone);
  try
    SetLength(Bytes, Stream.Size);
    if Length(Bytes) > 0 then Stream.ReadBuffer(Bytes[0], Length(Bytes));
  finally
    Stream.Free;
  end;
  Result := TEncoding.UTF8.GetString(Bytes).Split([sLineBreak], TStringSplitOptions.ExcludeEmpty);
end;

// A factory, so each thread captures its own number rather than the loop variable.
function Worker(Log: TAtlasLogger; Number: Integer): TProc;
begin
  Result :=
    procedure
    var Step: Integer; Scope: IInterface; Channel: TAtlasLogChannel;
    begin
      Channel := Log.Channel('worker');
      Scope := Log.BeginScope('worker ' + IntToStr(Number));
      for Step := 1 to Steps do
      begin
        Channel.Debug('Worker {worker} step {step}', [Number, Step]);
        if Step = Steps div 2 then Channel.Warning('Worker {worker} is half way', [Number]);
      end;
    end;
end;

procedure ThreadsIntoAsyncFile(const Folder: string; Handlers: THandlers);
var
  Log: TAtlasLogger; Json: TAtlasFileSink; Async: TAtlasAsyncSink; Alerts: TAtlasCallbackSink; Counter: TLevelCounter;
  Threads: TArray<TThread>; T, Step, Previous: Integer; Line: string; Lines: TArray<string>; Parsed: TJSONValue;
  LastStep: TDictionary<string, Integer>; Sequences: TDictionary<Int64, Boolean>; Ordered: Boolean;
begin
  Log := TAtlasLogger.Create(nil);
  Json := TAtlasFileSink.Create(nil);
  Async := TAtlasAsyncSink.Create(nil);
  Alerts := TAtlasCallbackSink.Create(nil);
  Counter := TLevelCounter.Create(nil);
  LastStep := TDictionary<string, Integer>.Create;
  Sequences := TDictionary<Int64, Boolean>.Create;
  try
    Log.MinLevel := llDebug;
    // The file sink is the async sink's target; it is not attached to the logger itself.
    Json.FileName := TPath.Combine(Folder, 'workers.jsonl');
    Json.Layout := loJson;
    Async.Target := Json;
    Async.Capacity := 256;
    Async.Overflow := ofBlock;
    Async.Logger := Log;
    Alerts.MinLevel := llWarning;
    Alerts.OnEvent := Handlers.WarningSeen;
    Alerts.Logger := Log;
    Counter.Logger := Log;

    SetLength(Threads, Workers);
    for T := 0 to High(Threads) do
    begin
      Threads[T] := TThread.CreateAnonymousThread(Worker(Log, T + 1));
      Threads[T].FreeOnTerminate := False;
      Threads[T].Start;
    end;
    for T := 0 to High(Threads) do
    begin
      Threads[T].WaitFor;
      FreeAndNil(Threads[T]);
    end;
    // Waits until the background thread has written the whole queue, then flushes the file.
    Log.Flush;

    Check((Async.Pending = 0) and (Async.Dropped = 0) and (Async.TargetErrors = 0), 'Flush emptied the queue and ofBlock dropped nothing');
    Check(Handlers.Warnings = Workers, 'the callback sink saw one warning per worker');
    Check((Counter.Counts[llDebug] = Workers * Steps) and (Counter.Counts[llWarning] = Workers), 'the custom sink counted every event by level');
    Lines := ReadLines(Json.ActiveFileName);
    Check(Length(Lines) = Workers * (Steps + 1), Format('%d JSON lines in %s', [Length(Lines), ExtractFileName(Json.ActiveFileName)]));
    Ordered := True;
    for Line in Lines do
    begin
      Parsed := TJSONObject.ParseJSONValue(Line);
      try
        if not (Parsed is TJSONObject) then Check(False, 'every line parses as a JSON object');
        if Sequences.ContainsKey(Parsed.GetValue<Int64>('seq')) then Check(False, 'sequence numbers are unique');
        Sequences.Add(Parsed.GetValue<Int64>('seq'), True);
        if Parsed.GetValue<string>('level') = 'debug' then
        begin
          Step := StrToInt(Parsed.GetValue<string>('props.step'));
          if LastStep.TryGetValue(Parsed.GetValue<string>('scope'), Previous) and (Step <> Previous + 1) then Ordered := False;
          LastStep.AddOrSetValue(Parsed.GetValue<string>('scope'), Step);
        end;
      finally
        Parsed.Free;
      end;
    end;
    Check(Ordered and (LastStep.Count = Workers), 'each thread''s events stay in order and carry that thread''s scope only');
    Writeln('    ', Lines[High(Lines)]);
  finally
    Sequences.Free;
    LastStep.Free;
    // The async sink first: destroying it writes anything still queued to its target.
    Async.Free;
    Json.Free;
    Alerts.Free;
    Counter.Free;
    Log.Free;
  end;
end;

procedure OverflowPolicies(Handlers: THandlers);
var Log: TAtlasLogger; Slow: TAtlasCallbackSink; Async: TAtlasAsyncSink; Broken: TAtlasFileSink; I: Integer; Before: Int64;
begin
  Log := TAtlasLogger.Create(nil);
  Slow := TAtlasCallbackSink.Create(nil);
  Async := TAtlasAsyncSink.Create(nil);
  Broken := TAtlasFileSink.Create(nil);
  try
    Slow.OnEvent := Handlers.SlowWrite;
    Async.Target := Slow;
    Async.Capacity := 10;
    Async.Overflow := ofDropNewest;
    Async.Logger := Log;

    // Hold the target back: the queue fills and the overflow policy decides.
    Handlers.Gate.ResetEvent;
    for I := 1 to 30 do Log.Info('n{i}', [I]);
    Check((Async.Dropped >= 19) and (Async.Dropped <= 20), Format('ofDropNewest: %d of 30 events dropped while the target was stuck', [Async.Dropped]));
    Handlers.Gate.SetEvent;
    Log.Flush;
    Check((Handlers.Written.Count + Async.Dropped = 30) and (Handlers.Written[0] = 'n1'), 'the rest were written, oldest first');

    Handlers.Written.Clear;
    Handlers.Gate.ResetEvent;
    Async.Overflow := ofDropOldest;
    Before := Async.Dropped;
    for I := 1 to 30 do Log.Info('o{i}', [I]);
    Handlers.Gate.SetEvent;
    Log.Flush;
    Check((Async.Dropped > Before) and (Handlers.Written.Count + (Async.Dropped - Before) = 30) and (Handlers.Written.Last = 'o30'),
      Format('ofDropOldest: %d dropped from the front, the newest event kept', [Async.Dropped - Before]));

    // A target that fails: a file sink without a FileName raises in DoWrite.
    Async.Target := Broken;
    Log.Info('Nowhere to go');
    Log.Flush;
    Check((Async.TargetErrors = 1) and (Log.SinkErrors = 0), 'a failing target is counted in TargetErrors and never raised');
  finally
    // Never leave the background thread waiting on a closed gate.
    Handlers.Gate.SetEvent;
    Async.Free;
    Broken.Free;
    Slow.Free;
    Log.Free;
  end;
end;

var
  Folder: string;
  Handlers: THandlers;
begin
  Folder := TPath.Combine(TPath.GetTempPath, 'AtlasLogThreadedAsync');
  Handlers := THandlers.Create;
  try
    try
      if TDirectory.Exists(Folder) then TDirectory.Delete(Folder, True);
      Writeln('Part 1: four threads, one asynchronous JSON file');
      ThreadsIntoAsyncFile(Folder, Handlers);
      Writeln('Part 2: a stuck target and the overflow policies');
      OverflowPolicies(Handlers);
      Writeln('All checks passed.');
    finally
      if TDirectory.Exists(Folder) then TDirectory.Delete(Folder, True);
      Handlers.Free;
    end;
  except
    on E: Exception do
    begin
      Writeln(E.ClassName, ': ', E.Message);
      ExitCode := 1;
    end;
  end;
end.