D Programming - Concurrency

Concurrency is making a program run on multiple threads at a time. An example of a concurrent program is a web server responding many clients at the same time. Concurrency is easy with message passing but very difficult to write if they are based on data sharing.

Data that is passed between threads are called messages. Messages may be composed of any type and any number of variables. Every thread has an id, which is used for specifying recipients of messages. Any thread that starts another thread is called the owner of the new thread.

Initiating Threads in D

The function spawn() takes a 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. The owner and the worker both start executing separately as if they were independent programs.

Example

import std.stdio; 
import std.stdio; 
import std.concurrency; 
import core.thread;
  
void worker(int a) { 
   foreach (i; 0 .. 4) { 
      Thread.sleep(1); 
      writeln("Worker Thread ",a + i); 
   } 
}

void main() { 
   foreach (i; 1 .. 4) { 
      Thread.sleep(2); 
      writeln("Main Thread ",i); 
      spawn(≈worker, i * 5); 
   }
   
   writeln("main is done.");  
}

When the above code is compiled and executed, it reads the file created in previous section and produces the following result −

Main Thread 1 
Worker Thread 5 
Main Thread 2 
Worker Thread 6 
Worker Thread 10 
Main Thread 3 
main is done. 
Worker Thread 7 
Worker Thread 11 
Worker Thread 15 
Worker Thread 8 
Worker Thread 12 
Worker Thread 16 
Worker Thread 13
Worker Thread 17 
Worker Thread 18

Thread Identifiers in D

The thisTid variable available globally at the module level is always the id of the current thread. Also you can receive the threadId when spawn is called. An example is shown below.

Example

import std.stdio; 
import std.concurrency;  

void printTid(string tag) { 
   writefln("%s: %s, address: %s", tag, thisTid, &thisTid); 
} 
 
void worker() { 
   printTid("Worker"); 
}
  
void main() { 
   Tid myWorker = spawn(&worker); 
   
   printTid("Owner "); 
   
   writeln(myWorker); 
}

When the above code is compiled and executed, it reads the file created in previous section and produces the following result −

Owner : Tid(std.concurrency.MessageBox), address: 10C71A59C 
Worker: Tid(std.concurrency.MessageBox), address: 10C71A59C 
Tid(std.concurrency.MessageBox)

Message Passing in D

The function send() sends messages and the function receiveOnly() waits for a message of a particular type. There are other functions named prioritySend(), receive(), and receiveTimeout(), which are explained later.

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. An example is shown below.

Example

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

void workerFunc(Tid tid) { 
   int value = 0;  
   while (value >= 0) { 
      value = receiveOnly!int(); 
      auto result = to!double(value) * 5; tid.send(result);
   }
} 
 
void main() { 
   Tid worker = spawn(&workerFunc,thisTid); 
    
   foreach (value; 5 .. 10) { 
      worker.send(value); 
      auto result = receiveOnly!double(); 
      writefln("sent: %s, received: %s", value, result); 
   }
   
   worker.send(-1); 
} 

When the above code is compiled and executed, it reads the file created in previous section and produces the following result −

sent: 5, received: 25 
sent: 6, received: 30 
sent: 7, received: 35 
sent: 8, received: 40 
sent: 9, received: 45 

Message Passing with Wait in D

A simple example with the message passing with wait is shown below.

import std.stdio; 
import std.concurrency; 
import core.thread; 
import std.conv; 
 
void workerFunc(Tid tid) { 
   Thread.sleep(dur!("msecs")( 500 ),); 
   tid.send("hello"); 
}
  
void main() { 
   spawn(&workerFunc,thisTid);  
   writeln("Waiting for a message");  
   bool received = false;
   
   while (!received) { 
      received = receiveTimeout(dur!("msecs")( 100 ), (string message) { 
         writeln("received: ", message); 
      });

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

When the above code is compiled and executed, it reads the file created in previous section and produces the following result −

Waiting for a message 
... no message yet 
... no message yet 
... no message yet 
... no message yet 
received: hello