Tracing/Profiling D Applications

[Ali Çehreli]

On 5/27/22 06:55, Christian Köstlin wrote:

 > I wonder how I can synchronize the "dumping" and the
 > collection of the threads. Would be cool to have an efficient lockless
 > implementation of appender ...

That turned out to be nontrivial.

The following is a draft I played with. Collector collects and Dumper 
dumps. They use a SpinLock, an unpublished feature of core.internal for 
locking. The implementation of spinlock (e.g. at 
/usr/include/dlang/dmd/core/internal/spinlock.d) has a reference to 
"test and test-and-set (TTAS)":

   https://en.wikipedia.org/wiki/Test_and_test-and-set

I learned about TTAS from Rikki Cattermole yesterday at TeaConf. :)

The code is attached and works on my system.

Ali

import std;
import std.datetime.stopwatch;
import core.thread;
import core.atomic;
import core.internal.spinlock;

enum workerCount = 8;
enum threadRunTime = 4.seconds;
enum mainRunTime = threadRunTime + 1.seconds;

shared struct ScopeLock {
   @disable this(this);
   @disable void opAssign(ref const(typeof(this)));

   SpinLock * lock;

   this(shared(SpinLock) * lock) {
     this.lock = lock;
     lock.lock();
   }

   ~this() {
     lock.unlock();
   }
}

struct Collector {
   long[] data;

   shared(SpinLock) lock;

   auto scopeLock() shared {
     return ScopeLock(&lock);
   }

   // Adds a data point to this collector.
   void add(long i) shared {
     auto sl = scopeLock();

     /// Some crazy way of adding data points. Real code should
     // make more sense.
     data ~= i;
   }

   // Adds the data of this collector to the specified array
   // array. Again, real code should use a more sophisticated
   // method.
   void aggregate(ref long[] where) shared {
     auto sl = scopeLock();

     where ~= data.sum;
     data.length = 0;
     (cast(long[])data).assumeSafeAppend();
   }
}

// A variable to help us trust the code. We will print this at
// the end of main.
long allThatHasBeenDumped = 0;
// Used only for validating the code.
shared long allCollectedByThreads;

synchronized class Dumper {
private:
   shared(Collector)*[] collectors;

   void register(shared(Collector) * collector) shared {
     writeln("registering ", collector);
     collectors ~= collector;
   }

   // Dumps current results.
   void dump(File output) shared {
     long[] data;

     foreach (collector; collectors) {
       collector.aggregate(data);
     }

     const allData = data.sum;

     if (allData != 0) {
       stdout.writefln!"Just collected:%-(\n  %,s%)"(data);
       allThatHasBeenDumped += allData;
     }
   }
}

shared(Dumper) dumper;

shared static this() {
   writeln("Making a Dumper");
   dumper = new Dumper();
}

shared(Collector) * collector;

static this() {
   writeln("Making a Collector");
   collector = new shared(Collector)();
   dumper.register(cast(shared)collector);
}

// Main thread function
void doWork() {
   try {
     doWorkImpl();

   } catch (Throwable exc) {
     stderr.writeln("Caught Throwable: ", exc.msg);
   }
}

// The implementation of each thread.
void doWorkImpl() {
   auto sw = StopWatch();
   sw.start();

   long i = 0;
   while (sw.peek < threadRunTime) {
     (cast(shared)collector).add(i);
     ++i;
   }

   --i;
   auto total = i * (i + 1) / 2;
   writefln("Thread collected %,s items equaling %,s with %s",
            i, total, collector);

   atomicOp!"+="(allCollectedByThreads, total);
}

void main() {
   writeln("main started");
   iota(workerCount).each!(_ => spawn(&doWork));

   auto sw = StopWatch();
   sw.start();

   while (sw.peek < mainRunTime) {
     dumper.dump(stdout);
     Thread.sleep(100.msecs);
   }

   // One final collection (and dump):
   dumper.dump(stdout);

   assert(allThatHasBeenDumped == allCollectedByThreads);
}