"shared" status

[Ali Çehreli]

On 04/16/2012 03:57 AM, Zardoz wrote:

 > So, if I need to share a array of 0x10000 elements between 3 or more
 > threads, how should do it ?

1) The following program starts four threads to fill different parts of 
a shared array:

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

void numberFiller(shared(int)[] area, int fillValue)
{
     foreach (ref number; area) {
         number = fillValue;
     }
}

void main()
{
     enum totalNumbers = 0x10;
     auto numbers = new shared(int)[totalNumbers];

     enum totalThreads = 4;
     enum numbersPerThread = totalNumbers / totalThreads;

     foreach (i; 0 .. totalThreads) {
         immutable start = i * numbersPerThread;
         immutable fillValue = i;

         spawn(&numberFiller,
               numbers[start .. start + numbersPerThread],
               cast(int)fillValue);
     }

     thread_joinAll();

     writeln(numbers);
}

The output:

[0, 0, 0, 0, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3]


2) The program above is being careful to limit the threads to different 
parts of the array. In other cases lock-based multi-threading can be 
used. The following program allows four thread append to a single array 
as they get a hold of the slice:

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

class Job
{
     int[] * slice;
     size_t count;

     this(ref int[] slice, size_t count)
     {
         this.slice = &slice;
         this.count = count;
     }
}

void numberAppender(shared(Job) job, int appendValue)
{
     foreach (i; 0 .. job.count) {
         synchronized (job) {
             *job.slice ~= appendValue;
         }

         Thread.sleep(dur!"msecs"(uniform(1,100)));
     }
}

void main()
{
     enum totalNumbers = 0x10;
     int[] numbers;

     enum totalThreads = 4;
     enum numbersPerThread = totalNumbers / totalThreads;

     auto job = new shared(Job)(numbers, numbersPerThread);

     foreach (i; 0 .. totalThreads) {
         int appendValue = i;

         spawn(&numberAppender, job, appendValue);
     }

     thread_joinAll();

     writeln(numbers);
}

The output should be similar to this:

[0, 1, 3, 2, 1, 1, 2, 0, 3, 1, 3, 0, 2, 3, 0, 2]

(Note: I wish there were 'ref' variables in D. That's why Job.slice 
above had to be a pointer.)


3) Better than the two approaches above may be to use message passing 
and have the threads produce separate results to be either combined 
later or simply used separately:

import std.stdio;
import std.concurrency;

void arrayMaker(Tid owner, int count, int value)
{
     immutable(int)[] result;

     foreach (i; 0 .. count) {
         result ~= value;
     }

     owner.send(result);
}

void main()
{
     enum totalNumbers = 0x10;
     enum totalThreads = 4;
     enum numbersPerThread = totalNumbers / totalThreads;

     foreach (i; 0 .. totalThreads) {
         int value = i;
         spawn(&arrayMaker, thisTid, numbersPerThread, value);
     }

     immutable(int[])[] results;

     foreach (i; 0 .. totalThreads) {
         auto result = receiveOnly!(immutable(int)[])();
         results ~= result;
     }

     writeln(results);
}

The output should be similar to this:

[[0, 0, 0, 0], [1, 1, 1, 1], [3, 3, 3, 3], [2, 2, 2, 2]]

(Note: I could not pass the results as shared int slices so I went back 
to immutable.)

Ali