D Programming Language Tutorial
Ali Çehreli



İngilizce Kaynaklar

Diğer



Message Passing Concurrency

Concurrency is similar to but different from the topic of the previous chapter, parallelism. As these two concepts both involve executing programs on threads, and as parallelism is based on concurrency, they are sometimes confused with each other.

The following are the differences between parallelism and concurrency:

D supports both models of concurrency: message passing and data sharing. We will cover message passing in this chapter and data sharing in the next chapter.

Concepts

Thread: Operating systems execute programs as work units called threads. D programs start executing with main() on a thread that has been assigned to that program by the operating system. All of the operations of the program are normally executed on that thread. The program is free to start other threads to be able to work on multiple tasks at the same time. In fact, tasks that have been covered in the previous chapter are based on threads that are started automatically by std.parallelism.

The operating system can pause threads at unpredictable times for unpredictable durations. As a result, even operations as simple as incrementing a variable may be paused mid operation:

    ++i;

The operation above involves three steps: Reading the value of the variable, incrementing the value, and assigning the new value back to the variable. The thread may be paused at any point between these steps to be continued after an unpredictable time.

Message: Data that is passed between threads are called messages. Messages may be composed of any type and any number of variables.

Thread identifier: Every thread has an id, which is used for specifying recipients of messages.

Owner: Any thread that starts another thread is called the owner of the new thread.

Worker: Any thread that is started by an owner is called a worker.

Starting threads

spawn() takes a function pointer as a parameter and starts a new thread from that function. Any operations that are carried out by that function, including other functions that it may call, would be executed on the new thread. As soon as a new thread is started, the owner and the worker start executing separately as if they were independent programs:

import std.stdio;
import std.concurrency;
import core.thread;

void worker()
{
    foreach (i; 0 .. 5) {
        Thread.sleep(500.msecs);
        writeln(i, " (worker)");
    }
}

void main()
{
    spawn(&worker);

    foreach (i; 0 .. 5) {
        Thread.sleep(300.msecs);
        writeln(i, " (main)");
    }

    writeln("main is done.");
}

The examples in this chapter call Thread.sleep to slow down threads to demonstrate that they run at the same time. The output of the program shows that the two threads, one that runs main() and the other that has been started by spawn(), execute independently at the same time:

0 (main)
0 (worker)
1 (main)
2 (main)
1 (worker)
3 (main)
2 (worker)
4 (main)
main is done.
3 (worker)
4 (worker)

The program automatically waits for all of the threads to finish executing. We can see this in the output above by the fact that worker() continues executing even after main() exits after printing "main is done."

The parameters that the thread function takes are passed to spawn() as its second and later arguments. The two worker threads in the following program print four numbers each. They take the starting number as the thread function parameter:

import std.stdio;
import std.concurrency;
import core.thread;

void worker(int firstNumber)
{
    foreach (i; 0 .. 4) {
        Thread.sleep(500.msecs);
        writeln(firstNumber + i);
    }
}

void main()
{
    foreach (i; 1 .. 3) {
        spawn(&worker, i * 10);
    }
}

The output of one of the threads is indicated in blue:

10
20
11
21
12
22
13
23

The lines of the output may be different at different times depending on how the threads are paused and resumed by the operating system.

Thread identifiers

The thisTid variable that is available globally at the module level is always the id of the current thread.

import std.stdio;
import std.concurrency;

void printTid(string tag)
{
    writefln("%s: %s, address: %s", tag, thisTid, &thisTid);
}

void worker()
{
    printTid("Worker");
}

void main()
{
    spawn(&worker);
    printTid("Owner ");
}

The type of thisTid is Tid, but its value has no significance for the program. Further, both threads report it to be surprisingly at the same address:

Owner : Tid(std.concurrency.MessageBox), address: 809C360
Worker: Tid(std.concurrency.MessageBox), address: 809C360

Even so, the two thisTid variables above are independent from each other. This is because in D data is thread-local by default.

The owner of a worker thread is the ownerTid variable.

The return value of spawn(), which I have been ignoring until this point is the id of the worker thread:

    Tid myWorker = spawn(&worker);

In summary, the owner is identified by ownerTid and the worker is identified by the return value of spawn().

Message Passing

send() sends messages and receiveOnly() waits for a message of a particular type. (There is also prioritySend(), receive(), and receiveTimeout(), which will be explained later below.)

The owner in the following program sends its worker a message of type int and waits for a message from the worker of type double. The threads continue sending messages back and forth until the owner sends a negative int. This is the owner thread:

void main()
{
    Tid worker = spawn(&workerFunc);

    foreach (value; 1 .. 5) {
        worker.send(value);
        double result = receiveOnly!double();
        writefln("sent: %s, received: %s", value, result);
    }

    /* Sending a negative value to the worker so that it
     * terminates.  */
    worker.send(-1);
}

main() stores the return value of spawn() under the name worker and uses that variable when sending messages to the worker.

On the other side, the worker receives the message that it needs as an int, uses that value in a calculation, and sends the result as type double to its owner:

void workerFunc()
{
    int value = 0;

    while (value >= 0) {
        value = receiveOnly!int();
        double result = to!double(value) / 5;
        ownerTid.send(result);
    }
}

The main thread reports the messages that it sends and the messages that it receives:

sent: 1, received: 0.2
sent: 2, received: 0.4
sent: 3, received: 0.6
sent: 4, received: 0.8

It is possible to send more than one value as a part of the same message. The following message consists of three parts:

    ownerTid.send(thisTid, 42, 1.5);

Values that are passed as parts of a single message appear as a tuple on the receiver's side. In such cases the template parameters of receiveOnly() must match the types of the tuple members:

    /* Wait for a message composed of Tid, int, and double. */
    auto message = receiveOnly!(Tid, int, double)();

    auto sender   = message[0];    // of type Tid
    auto integer  = message[1];    // of type int
    auto floating = message[2];    // of type double

If the types do not match, a MessageMismatch exception is thrown:

import std.concurrency;

void workerFunc()
{
    // Sending string
    ownerTid.send("hello");
}

void main()
{
    spawn(&workerFunc);

    // Expecting double
    auto message = receiveOnly!double();
}

The output:

std.concurrency.MessageMismatch@std/concurrency.d(235):
Unexpected message type: expected 'double', got 'immutable(char)[]'

The exceptions that the worker may throw cannot be caught by the owner. One solution is to have the worker catch the exception to be sent as a message. We will see this below.

Example

Let's use what we have seen so far in a simulation program.

The following program simulates robots moving randomly and indepently in a two dimensional space. The movement of each robot is handled by a separate thread, which takes four pieces of information when started:

Last three pieces of this information can be stored in the following Job struct:

struct Job
{
    size_t robotId;
    Position origin;
    Duration restDuration;
}

The thread function that moves each robot continuously sends the number of the robot and its movement to the owner thread:

void robotMover(Job job)
{
    Position from = job.origin;

    while (true) {
        Thread.sleep(job.restDuration);

        Position to = randomNeighbor(from);
        Movement movement = Movement(from, to);
        from = to;

        ownerTid.send(MovementMessage(job.robotId, movement));
    }
}

The owner simply waits for these messages in an infinite loop. It identifies the robots by the robot number that is sent as a part of the message and simply prints each movement:

    while (true) {
        auto message = receiveOnly!MovementMessage();

        writefln("%s %s",
                 robots[message.robotId], message.movement);
    }

All of the messages in this simple program go from the worker to the owner. Message passing would normally be more complicated in many other types of programs.

Here is the complete program:

import std.stdio;
import std.random;
import std.string;
import std.concurrency;
import core.thread;

struct Position
{
    int line;
    int column;

    string toString()
    {
        return format("%s,%s", line, column);
    }
}

struct Movement
{
    Position from;
    Position to;

    string toString()
    {
        return ((from == to)
                ? format("%s (idle)", from)
                : format("%s -> %s", from, to));
    }
}

class Robot
{
    string image;
    Duration restDuration;

    this(string image, Duration restDuration)
    {
        this.image = image;
        this.restDuration = restDuration;
    }

    override string toString()
    {
        return format("%s(%s)", image, restDuration);
    }
}

/* Returns a random position around 0,0. */
Position randomPosition()
{
    return Position(uniform!"[]"(-10, 10),
                    uniform!"[]"(-10, 10));
}

/* Returns at most one step from the specified coordinate. */
int randomStep(int current)
{
    return current + uniform!"[]"(-1, 1);
}

/* Returns a neighbor of the specified Position. It may be one
 * of the neighbors at eight directions, or the specified
 * position itself. */
Position randomNeighbor(Position position)
{
    return Position(randomStep(position.line),
                    randomStep(position.column));
}

struct Job
{
    size_t robotId;
    Position origin;
    Duration restDuration;
}

struct MovementMessage
{
    size_t robotId;
    Movement movement;
}

void robotMover(Job job)
{
    Position from = job.origin;

    while (true) {
        Thread.sleep(job.restDuration);

        Position to = randomNeighbor(from);
        Movement movement = Movement(from, to);
        from = to;

        ownerTid.send(MovementMessage(job.robotId, movement));
    }
}

void main()
{
    /* Robots with various restDurations. */
    Robot[] robots = [ new Robot("A",  600.msecs),
                       new Robot("B", 2000.msecs),
                       new Robot("C", 5000.msecs) ];

    /* Start a mover thread for each robot. */
    foreach (robotId, robot; robots) {
        spawn(&robotMover, Job(robotId,
                               randomPosition(),
                               robot.restDuration));
    }

    /* Ready to collect information about the movements of the
     * robots. */
    while (true) {
        auto message = receiveOnly!MovementMessage();

        /* Print the movement of this robot. */
        writefln("%s %s",
                 robots[message.robotId], message.movement);
    }
}

The program prints each movement until terminated:

A(600 ms) 6,2 -> 7,3
A(600 ms) 7,3 -> 8,3
A(600 ms) 8,3 -> 7,3
B(2 secs) -7,-4 -> -6,-3
A(600 ms) 7,3 -> 6,2
A(600 ms) 6,2 -> 7,1
A(600 ms) 7,1 (idle)
B(2 secs) -6,-3 (idle)
A(600 ms) 7,1 -> 7,2
A(600 ms) 7,2 -> 7,3
C(5 secs) -4,-4 -> -3,-5
A(600 ms) 7,3 -> 6,4
...

The power of concurrency by message passing is demonstrated in this program. Movements of each robot are calculated independently by separate threads. It is the owner thread that serializes the printing of the movements by printing them in the order that it receives the messages.

Expecting different types of messages

receiveOnly() can expect only one type of message. receive() on the other hand can wait for more than one type of message. It dispatches messages to message handling delegates. When a message arrives, it is compared to the message type of each delegate. The delegate that matches the type of the particular message handles it.

For example, the following receive() call specifies two message handlers that would handle messages of types int and string, respectively:

void workerFunc()
{
    bool isDone = false;

    while (!isDone) {
        void intHandler(int message)
        {
            writeln("handling int message: ", message);

            if (message == -1) {
                writeln("exiting");
                isDone = true;
            }
        }

        void stringHandler(string message)
        {
            writeln("handling string message: ", message);
        }

        receive(&intHandler, &stringHandler);
    }
}

Messages of type int would match intHandler() and messages of type string would match stringHandler(). The worker thread above can be tested by the following program:

import std.stdio;
import std.concurrency;

// ...

void main()
{
    auto worker = spawn(&workerFunc);

    worker.send(10);
    worker.send(42);
    worker.send("hello");
    worker.send(-1);        // ← to terminate the worker
}

The output of the program indicates that the messages are handled by matching functions on the receiver's side:

handling int message: 10
handling int message: 42
handling string message: hello
handling int message: -1
exiting

Lambda functions and objects of types that define the opCall() member function can also be passed to receive() as message handlers. The following worker handles messages by lambda functions. The following program also defines a special type named Exit used for communicating to the thread that it is time for it to exit. Using such a specific type is more expressive than sending the arbitrary value of -1.

There are three anonymous functions below that are passed to receive() as message handlers. Their curly brackets are highlighted:

import std.stdio;
import std.concurrency;

struct Exit
{}

void workerFunc()
{
    bool isDone = false;

    while (!isDone) {
        receive(
            (int message) {
                writeln("int message: ", message);
            },

            (string message) {
                writeln("string message: ", message);
            },

            (Exit message) {
                writeln("exiting");
                isDone = true;
            });
    }
}

void main()
{
    auto worker = spawn(&workerFunc);

    worker.send(10);
    worker.send(42);
    worker.send("hello");
    worker.send(Exit());
}
Receiving any type of message

std.variant.Variant is a type that can encapsulate any type of data. Messages that do not match the handlers that are specified earlier in the argument list always match a Variant handler:

import std.stdio;
import std.concurrency;

void workerFunc()
{
    receive(
        (int message) { /* ... */ },

        (double message) { /* ... */ },

        (Variant message) {
            writeln("Unexpected message: ", message);
        });
}

struct SpecialMessage
{
    /* ... */
}

void main()
{
    auto worker = spawn(&workerFunc);
    worker.send(SpecialMessage());
}

The output:

Unexpected message: SpecialMessage()

The details of Variant are outside of the scope of this chapter.

Waiting for messages up to a certain time

It may not make sense to wait for messages beyond a certain time. The sender may have been busy temporarily or may have terminated with an exception. receiveTimeout() prevents blocking the receiving thread indefinitely.

The first parameter of receiveTimeout() determines how long the message should be waited for. Its return value is true if a message has been received within that time, false otherwise.

import std.stdio;
import std.concurrency;
import core.thread;

void workerFunc()
{
    Thread.sleep(3.seconds);
    ownerTid.send("hello");
}

void main()
{
    spawn(&workerFunc);

    writeln("Waiting for a message");
    bool received = false;
    while (!received) {
        received = receiveTimeout(600.msecs,
                                  (string message) {
                                      writeln("received: ", message);
                                });

        if (!received) {
            writeln("... no message yet");

            /* ... other operations may be executed here ... */
        }
    }
}

The owner above waits for a message for up to 600 milliseconds. It can continue working on other things if a message does not arrive within that time:

Waiting for a message
... no message yet
... no message yet
... no message yet
... no message yet
received: hello
Exceptions during the execution of the worker

As we have seen in the previous chapter, the facilities of the std.parallelism module automatically catch exceptions that have been thrown during the execution of tasks and rethrow them in the context of the owner. This allows the owner to catch such exceptions:

    try {
        theTask.yieldForce();

    } catch (Exception exc) {
        writefln("Detected an error in the task: '%s'",
                 exc.msg);
    }

std.concurrency does not provide such a convenience for general exception types. The exceptions can be caught and sent explicitly by the worker. As we will see below, it is also possible to receive OwnerTerminated and LinkTerminated exceptions as messages.

The calculate() function below receives string messages, converts them to double, adds 0.5, and sends the result back as a message:

void calculate()
{
    while (true) {
        auto message = receiveOnly!string();
        ownerTid.send(to!double(message) + 0.5);
    }
}

The to!double() call above would throw an exception if the string cannot be converted to a double value. Because such an exception would terminate the worker thread right away, the owner in the following program can receive a response only for the first message:

import std.stdio;
import std.concurrency;
import std.conv;

/* ... */

void main()
{
    Tid calculator = spawn(&calculate);

    calculator.send("1.2");
    calculator.send("hello");  // ← incorrect input
    calculator.send("3.4");

    foreach (i; 0 .. 3) {
        auto message = receiveOnly!double();
        writefln("result %s: %s", i, message);
    }
}

The owner receives the response for "1.2" as 1.7 but because the worker has been terminated, the owner would be blocked waiting for a message that would never arrive:

result 0: 1.7
                 ← waiting for a message that will never arrive

One thing that the worker can do is to catch the exception explicitly and to send it as a special error message. The following program sends the reason of the failure as a CalculationFailure message. Additionally, this program takes advantage of a special message type to signal to the worker when it is time to exit:

import std.stdio;
import std.concurrency;
import std.conv;

struct CalculationFailure
{
    string reason;
}

struct Exit
{}

void calculate()
{
    bool isDone = false;

    while (!isDone) {
        receive(
            (string message) {
                try {
                    ownerTid.send(to!double(message) + 0.5);

                } catch (Exception exc) {
                    ownerTid.send(CalculationFailure(exc.msg));
                }
            },

            (Exit message) {
                isDone = true;
            });
    }
}

void main()
{
    Tid calculator = spawn(&calculate);

    calculator.send("1.2");
    calculator.send("hello");  // ← incorrect input
    calculator.send("3.4");
    calculator.send(Exit());

    foreach (i; 0 .. 3) {
        writef("result %s: ", i);

        receive(
            (double message) {
                writeln(message);
            },

            (CalculationFailure message) {
                writefln("ERROR! '%s'", message.reason);
            });
    }
}

This time the reason of the failure is printed by the owner:

result 0: 1.7
result 1: ERROR! 'no digits seen'
result 2: 3.9

Another method would be to send the actual exception object itself to the owner. The owner can use the exception object or simply rethrow it:

// ... at the worker ...
                try {
                    // ...

                } catch (shared(Exception) exc) {
                    ownerTid.send(exc);
                }},

// ... at the owner ...
        receive(
            // ...

            (shared(Exception) exc) {
                throw exc;
            });

The reason why the shared specifiers are necessary is explained in the next chapter.

Detecting thread termination

Threads can detect that the receiver of a message has terminated.

OwnerTerminated exception

This exception is thrown when receiving a message from the owner if the owner has been terminated. The intermediate owner thread below simply exits after sending two messages to its worker. This causes an OwnerTerminated exception to be thrown at the worker thread:

import std.stdio;
import std.concurrency;

void main()
{
    spawn(&intermediaryFunc);
}

void intermediaryFunc()
{
    auto worker = spawn(&workerFunc);
    worker.send(1);
    worker.send(2);
}  // ← Terminates after sending two messages

void workerFunc()
{
    while (true) {
        auto m = receiveOnly!int(); // ← An exception is
                                    //   thrown if the owner
                                    //   has terminated.
        writeln("Message: ", m);
    }
}

The output:

Message: 1
Message: 2
std.concurrency.OwnerTerminated@std/concurrency.d(248):
Owner terminated

The worker can catch that exception to exit gracefully:

void workerFunc()
{
    bool isDone = false;

    while (!isDone) {
        try {
            auto m = receiveOnly!int();
            writeln("Message: ", m);

        } catch (OwnerTerminated exc) {
            writeln("The owner has terminated.");
            isDone = true;
        }
    }
}

The output:

Message: 1
Message: 2
The owner has terminated.

We will see below that this exception can be received as a message as well.

LinkTerminated exception

spawnLinked() is used in the same way as spawn(). When a worker that has been started by spawnLinked() terminates, a LinkTerminated exception is thrown at the owner:

import std.stdio;
import std.concurrency;

void main()
{
    auto worker = spawnLinked(&workerFunc);

    while (true) {
        auto m = receiveOnly!int(); // ← An exception is
                                    //   thrown if the worker
                                    //   has terminated.
        writeln("Message: ", m);
    }
}

void workerFunc()
{
    ownerTid.send(10);
    ownerTid.send(20);
}  // ← Terminates after sending two messages

The worker above terminates after sending two messages. Since the worker has been started by spawnLinked(), the owner is notified of the worker's termination by a LinkTerminated exception:

Message: 10
Message: 20
std.concurrency.LinkTerminated@std/concurrency.d(263):
Link terminated

The owner can catch the exception to do something special like terminating gracefully:

    bool isDone = false;

    while (!isDone) {
        try {
            auto m = receiveOnly!int();
            writeln("Message: ", m);

        } catch (LinkTerminated exc) {
            writeln("The worker has terminated.");
            isDone = true;
        }
    }

The output:

Message: 10
Message: 20
The worker has terminated.

This exception can be received as a message as well.

Receiving exceptions as messages

The OwnerTerminated and LinkTerminated exceptions can be received as messages as well. The following code demonstrates this for the OwnerTerminated exception:

    bool isDone = false;

    while (!isDone) {
        receive(
            (int message)
            {
                writeln("Message: ", message);
            },

            (OwnerTerminated exc)
            {
                writeln("The owner has terminated; exiting.");
                isDone = true;
            }
        );
    }
Mailbox management

Every thread has a private mailbox that holds the messages that are sent to that thread. The number of messages in a mailbox may increase or decrease depending on how long it takes for the thread to receive and respond to each message. A continuously growing mailbox puts stress on the entire system and may point to a design flaw in the program. It may also mean that the thread may never get to the most recent messages.

setMaxMailboxSize() is used for limiting the number of messages that a mailbox can hold. Its three parameters specify the mailbox, the maximum number of messages that it can hold, and what should happen when the mailbox is full, in that order. There are four choices for the last parameter:

Before seeing an example of setMaxMailboxSize(), let's first cause a mailbox to grow continuously. The worker in the following program sends messages back to back but the owner spends some time for each message:

/* WARNING: Your system may become unresponsive when this
 *          program is running. */
import std.concurrency;
import core.thread;

void workerFunc()
{
    while (true) {
        ownerTid.send(42);    // ← Produces messages continuously
    }
}

void main()
{
    spawn(&workerFunc);

    while (true) {
        receive(
            (int message) {
                // Spends time for each message
                Thread.sleep(1.seconds);
            });
    }
}

Because the consumer is slower than the producer, the memory that the program above uses would grow continuously. To prevent that, the owner may limit the size of its mailbox before starting the worker:

void main()
{
    setMaxMailboxSize(thisTid, 1000, OnCrowding.block);

    spawn(&workerFunc);
// ...
}

The setMaxMailboxSize() call above sets the main thread's mailbox size to 1000. OnCrowding.block causes the sender to wait until there is room in the mailbox.

The following example uses OnCrowding.throwException, which causes a MailboxFull exception to be thrown when sending a message to a mailbox that is full:

import std.concurrency;
import core.thread;

void workerFunc()
{
    while (true) {
        try {
            ownerTid.send(42);

        } catch (MailboxFull exc) {
            /* Failed to send; will try again later. */
            Thread.sleep(1.msecs);
        }
    }
}

void main()
{
    setMaxMailboxSize(thisTid, 1000, OnCrowding.throwException);

    spawn(&workerFunc);

    while (true) {
        receive(
            (int message) {
                Thread.sleep(1.seconds);
            });
    }
}
Priority messages

Messages can be sent with higher priority than regular messages by prioritySend(). These messages are handled before the other messages that are already in the mailbox:

    prioritySend(ownerTid, ImportantMessage(100));

If the receiver does not have a message handler that matches the type of the priority message, then a PriorityMessageException is thrown:

std.concurrency.PriorityMessageException@std/concurrency.d(280):
Priority message
Thread names

In the simple programs that we have used above, it was easy to pass the thread ids of owners and workers. Passing thread ids from thread to thread may be overly complicated in programs that use more than a couple of threads. To reduce this complexity, it is possible to assign names to threads, which are globally accessible from any thread.

The following three functions define an interface to an associative array that every thread has access to:

The following program starts two threads that find each other by their names. These threads continuously send messages to each other until instructed to terminate by an Exit message:

import std.stdio;
import std.concurrency;
import core.thread;

struct Exit
{}

void main()
{
    // A thread whose partner is named "second"
    auto first = spawn(&player, "second");
    register("first", first);
    scope(exit) unregister("first");

    // A thread whose partner is named "first"
    auto second = spawn(&player, "first");
    register("second", second);
    scope(exit) unregister("second");

    Thread.sleep(2.seconds);

    prioritySend(first, Exit());
    prioritySend(second, Exit());

    // For the unregister() calls to succeed, main() must wait
    // until the workers terminate.
    thread_joinAll();
}

void player(string nameOfPartner)
{
    Tid partner;

    while (partner == Tid.init) {
        Thread.sleep(1.msecs);
        partner = locate(nameOfPartner);
    }

    bool isDone = false;

    while (!isDone) {
        partner.send("hello " ~ nameOfPartner);
        receive(
            (string message) {
                writeln("Message: ", message);
                Thread.sleep(500.msecs);
            },

            (Exit message) {
                writefln("%s, I am exiting.", nameOfPartner);
                isDone = true;
            });
    }
}

The output:

Message: hello second
Message: hello first
Message: hello second
Message: hello first
Message: hello first
Message: hello second
Message: hello first
Message: hello second
second, I am exiting.
first, I am exiting.
Summary