std.concurrency

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

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.

struct Tid;

An opaque type used to represent a logical local process.

@property Tid thisTid();

Returns the caller's Tid.

Tid spawn(T...)(void function(T) fn, T args);

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:

fn	The function to execute.
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;

 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(T...)(void function(T) fn, T args);

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:

fn	The function to execute.
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;

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

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...)(long ms, T ops);

Deprecated. It will be removed in August 2012. Please use the version which takes a core.time.Duration instead.

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;

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.

struct List(T);