On Wednesday, 25 May 2022 at 21:35:07 UTC, Christian Köstlin wrote:
Is there also a way to get the "real"

I'm using that functions inside threads:

core.sys.windows.winbase.GetCurrentThreadId on Windows
core.sys.posix.pthread.pthread_self on Unix (implied pthreads are used)Thanks, I will give those a try!

On 5/25/22 14:35, Christian Köstlin wrote:

I think this is where thread-local storage comes in handy. As the D runtime does for dmd's -profile command line switch, you can have a module-level Appender, which collects data for that thread without worrying about race conditions.
Nice idea. Similar to the fast singleton here: https://p0nce.github.io/d-idioms/#Leveraging-TLS-for-a-fast-thread-safe-singleton

You need the mutex only when each thread is exiting to combine their results. Each static ~this() could grab the mutex and add its result to the one global Appender. And that "global" Appender would be 'shared' or __gshared; doesn't matter.
Thanks for the explanation. From my understanding __gshared does not put the shared type attribute the the profiler. If I use shared I would also have to make all methods I call on the shared profiler shared and also would have to have a shared appender (or cast shared away)?

On 5/26/22 12:54, Christian Köstlin wrote:

I am not sure without testing but I am under the impression that mutexes can be very slow especially if the data collection is happening very frequently. The threads may not have breathing room to perform their tasks. Since dumping would be much slower compared to the performance penalty of grabbing a mutex that it can be measured as zero-cost in the dumping time. :) So, you may still want to collect per-thread and combine incrementally as you dump? Perhaps a premature optimization... Please test. :)

If you want to combine before dumping, then collector registration may be needed so that the dumper can contact each collector and say "please add your data".

Correct.
thanks!

Possibly. :/

Ali

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

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);
}

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

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);
}