Phobos

Github Page

The D Programming Language

This is a low-level messaging API upon which more structured or restrictive APIs may be built. The general idea is that every messageable entity is represented by a common handle type (called a Cid in this implementation), which allows messages to be sent to in-process threads, on-host processes, and foreign-host processes using the same interface. This is an important aspect of scalability because it allows the components of a program to be spread across available resources with few to no changes to the actual implementation.

Right now, only in-process threads are supported and referenced by a more specialized handle called a Tid. It is effectively a subclass of Cid, with additional features specific to in-process messaging.

Synposis:
import std.stdio;
import std.concurrency;

void spawnedFunc(Tid tid)
{
    // Receive a message from the owner thread.

    receive(
        (int i) { writeln("Received the number ", i);}
    );

    // Send a message back to the owner thread

    // indicating success.

    send(tid, true);
}

void main()
{
    // Start spawnedFunc in a new thread.

    auto tid = spawn(&spawnedFunc, thisTid);

    // Send the number 42 to this new thread.

    send(tid, 42);

    // Receive the result code.

    auto wasSuccessful = receiveOnly!(bool);
    assert(wasSuccessful);
    writeln("Successfully printed number.");
}
License
Boost License 1.0.
Authors
Sean Kelly, Alex Rønne Petersen
Source:
std/concurrency.d

class  MessageMismatch: object.Exception;

Thrown on calls to receiveOnly if a message other than the type the receiving thread expected is sent.

class  OwnerTerminated: object.Exception;

Thrown on calls to receive if the thread that spawned the receiving thread has terminated and no more messages exist.

class  LinkTerminated: object.Exception;

Thrown if a linked thread has terminated.

class  PriorityMessageException: object.Exception;

Thrown if a message was sent to a thread via std.concurrency.prioritySend and the receiver does not have a handler for a message of this type.

Variant  message;

The  message that was sent.

class  MailboxFull: object.Exception;

Thrown on mailbox crowding if the mailbox is configured with OnCrowding.throwException.

class  TidMissingException: object.Exception;

Thrown when a Tid is missing, e.g. when ownerTid doesn't find an owner thread.

struct  Tid;

An opaque type used to represent a logical local process.

@property Tid  thisTid();

Returns the caller's Tid.

@property Tid  ownerTid();

Return the Tid of the thread which spawned the caller's thread.

Throws
A TidMissingException exception if there is no owner thread.

Tid  spawn(F, T...)(F fn, T args) if (isSpawnable!(F, T));

Executes the supplied function in a new context represented by Tid. The calling context is designated as the owner of the new context. When the owner context terminated an OwnerTerminated message will be sent to the new context, causing an OwnerTerminated exception to be thrown on receive().

Parameters
F fn The function to execute.
T args Arguments to the function.
Returns
A Tid representing the new context.
Notes:
args must not have unshared aliasing. In other words, all arguments to fn must either be shared or immutable or have no pointer indirection. This is necessary for enforcing isolation among threads.
Example:
import std.stdio, std.concurrency;

void f1(string str)
{
    writeln(str);
}

void f2(char[] str)
{
    writeln(str);
}

void main()
{
    auto str = "Hello, world";

    // Works:  string is immutable.

    auto tid1 = spawn(&f1, str);

    // Fails:  char[] has mutable aliasing.

    auto tid2 = spawn(&f2, str.dup);
}

Tid  spawnLinked(F, T...)(F fn, T args) if (isSpawnable!(F, T));

Executes the supplied function in a new context represented by Tid. This new context is linked to the calling context so that if either it or the calling context terminates a LinkTerminated message will be sent to the other, causing a LinkTerminated exception to be thrown on receive(). The owner relationship from spawn() is preserved as well, so if the link between threads is broken, owner termination will still result in an OwnerTerminated exception to be thrown on receive().

Parameters
F fn The function to execute.
T args Arguments to the function.
Returns
A Tid representing the new context.

void  send(T...)(Tid tid, T vals);

Sends the supplied value to the context represented by tid. As with std.concurrency.spawn, T must not have unshared aliasing.

void  prioritySend(T...)(Tid tid, T vals);

Send a message to tid but place it at the front of tid's message queue instead of at the back. This function is typically used for out-of-band communication, to signal exceptional conditions, etc.

void  receive(T...)(T ops);

Receive a message from another thread, or block if no messages of the specified types are available. This function works by pattern matching a message against a set of delegates and executing the first match found.

If a delegate that accepts a std.variant.Variant is included as the last argument to  receive, it will match any message that was not matched by an earlier delegate. If more than one argument is sent, the Variant will contain a std.typecons.Tuple of all values sent.

Example:
import std.stdio;
import std.variant;
import std.concurrency;

void spawnedFunction()
{
    receive(
        (int i) { writeln("Received an int."); },
        (float f) { writeln("Received a float."); },
        (Variant v) { writeln("Received some other type."); }
    );
}

void main()
{
     auto tid = spawn(&spawnedFunction);
     send(tid, 42);
}

receiveOnlyRet!T  receiveOnly(T...)();

Receives only messages with arguments of types T.

Throws
MessageMismatch if a message of types other than T is received.
Returns
The received message. If T.length is greater than one, the message will be packed into a std.typecons.Tuple.
Example:
import std.concurrency;

void spawnedFunc()
{
    auto msg = receiveOnly!(int, string)();
    assert(msg[0] == 42);
    assert(msg[1] == "42");
}

void main()
{
    auto tid = spawn(&spawnedFunc);
    send(tid, 42, "42");
}

bool  receiveTimeout(T...)(Duration duration, T ops);

Same as receive except that rather than wait forever for a message, it waits until either it receives a message or the given core.time.Duration has passed. It returns true if it received a message and false if it timed out waiting for one.

enum  OnCrowding: int;

These behaviors may be specified when a mailbox is full.

 block

Wait until room is available.

 throwException

Throw a MailboxFull exception.

 ignore

Abort the send and return.

void  setMaxMailboxSize(Tid tid, size_t messages, OnCrowding doThis);

Sets a limit on the maximum number of user messages allowed in the mailbox. If this limit is reached, the caller attempting to add a new message will execute the behavior specified by doThis. If messages is zero, the mailbox is unbounded.

Parameters
Tid tid The Tid of the thread for which this limit should be set.
size_t messages The maximum number of messages or zero if no limit.
OnCrowding doThis The behavior executed when a message is sent to a full mailbox.

void  setMaxMailboxSize(Tid tid, size_t messages, bool function(Tid) onCrowdingDoThis);

Sets a limit on the maximum number of user messages allowed in the mailbox. If this limit is reached, the caller attempting to add a new message will execute onCrowdingDoThis. If messages is zero, the mailbox is unbounded.

Parameters
Tid tid The Tid of the thread for which this limit should be set.
size_t messages The maximum number of messages or zero if no limit.
bool function(Tid) onCrowdingDoThis The routine called when a message is sent to a full mailbox.

bool  register(string name, Tid tid);

Associates name with tid in a process-local map. When the thread represented by tid termiantes, any names associated with it will be automatically unregistered.

Parameters
string name The name to associate with tid.
Tid tid The tid  register by name.
Returns
true if the name is available and tid is not known to represent a defunct thread.

bool  unregister(string name);

Removes the registered name associated with a tid.

Parameters
string name The name to  unregister.
Returns
true if the name is registered, false if not.

Tid  locate(string name);

Gets the Tid associated with name.

Parameters
string name The name to  locate within the registry.
Returns
The associated Tid or Tid.init if name is not registered.