Fiber and Thread Communication

[Ali Çehreli via Digitalmars-d-learn]

On 04/08/2016 02:42 PM, Dicebot wrote:

 >> Thanks Dicebot. I don't think the included
 >> std.concurrency.FiberScheduler has support for message passing because
 >> FiberScheduler.spawn does not return a Tid. If so, I don't see how
 >> it's possible to send messages between fibers.
 >>
 >> Ali
 >
 > Looks like a (funny) oversight.

Sorry, I misled you. :)

 > Note that you get it for get fiber via
 > 
https://github.com/D-Programming-Language/phobos/blob/master/std/concurrency.d#L1337 

 > (and FiberScheduler specifically extends Fiber to add ThreadInfo to it)
 > but there is no clear way to pass that info to spawn host. I have a
 > feeling that if that code is patched to simply provide Tid, message
 > passing will just magically work. Needs to be checked though.

It turns out, instead of calling scheduler.spawn() directly, the program 
sets the __gshared 'scheduler' variable first and then calls spawn() as 
usual, which does return that fiber's Tid:

import std.stdio;
import std.concurrency;
import std.range;
import std.algorithm;

struct Done {
}

void workerTask(int id) {
     writefln("workerTask %s started", id);

     bool done = false;
     while (!done) {
         receive(
             (int message) {
                 writefln("workerTask %s received %s", id, message);
                 ownerTid.send(message * id);
             },
             (Done message) {
                 writefln("workerTask %s received Done", id);
                 done = true;
             });

         // Seems not to be needed:
         // scheduler.yield();
     }

     writefln("workerTask %s exiting", id);
}

void mainTask() {
     enum workerCount = 5;
     enum loopCount = 3;

     writeln("mainTask started");

     auto workers = iota(workerCount)
                    .map!(id => spawn(&workerTask, id))
                    .array;

     foreach (i; 0 .. loopCount) {
         foreach (id, worker; workers) {
             worker.send(i);
             auto response = receiveOnly!int();
             assert(response == i * id);
             writefln("mainTask received %s", response);
         }
     }

     writeln("mainTask sending Done messages");

     foreach (worker; workers) {
         worker.send(Done());
     }

     writeln("mainTask exiting");
}

void main() {
     scheduler = new FiberScheduler;
     scheduler.start({
         mainTask();
     });
}

Ali