www.digitalmars.com         C & C++   DMDScript  

digitalmars.D.learn - Simplest multithreading example

reply Brian <brian.patrick.forbes gmail.com> writes:
Hello, I am trying to get the most trivial example of 
multithreading working, but can't seem to figure it out.
I want to split a task across threads, and wait for all those 
tasks to finish before moving to the next line of code.

The following 2 attempts have failed :

-----------------------------------------------------
Trial 1 :
-----------------------------------------------------

auto I = std.range.iota(0,500);
int [] X; // output
foreach (i; parallel(I) )
     X ~= i;
core.thread.thread_joinAll(); // Apparently no applicable here ?
writeln(X); // some random subset of indices

------------------------------------------------
Trial 2 : (closer to Java)
------------------------------------------------
class DerivedThread : Thread
{
     int [] X;
     int i;
     this(int [] X, int i){
         this.X = X;
	this.i = i;
         super(&run);
     }

     private:
         void run(){
             X ~= i;
         }
}

void main(){
	auto I = std.range.iota(0,500);
	int [] X; // output
	Thread [] threads;
	foreach (i; I )
		threads ~= new DerivedThread( X,i);
	foreach( thread; threads)
		thread.start();
	foreach( thread; threads)
		thread.join(); // does not seem to do anything
	core.thread.thread_joinAll(); // also not doing anything

	writeln(X); // X contains nothing at all
}

How can I get the program to wait until all threads have finished 
before moving to the next line of code ?

Thank you !
Aug 31
next sibling parent Michael Coulombe <kirsybuu gmail.com> writes:
On Friday, 1 September 2017 at 01:59:07 UTC, Brian wrote:
 Hello, I am trying to get the most trivial example of 
 multithreading working, but can't seem to figure it out.
 I want to split a task across threads, and wait for all those 
 tasks to finish before moving to the next line of code.

 The following 2 attempts have failed :

 -----------------------------------------------------
 Trial 1 :
 -----------------------------------------------------

 auto I = std.range.iota(0,500);
 int [] X; // output
 foreach (i; parallel(I) )
     X ~= i;
 core.thread.thread_joinAll(); // Apparently no applicable here ?
 writeln(X); // some random subset of indices

 ------------------------------------------------
 Trial 2 : (closer to Java)
 ------------------------------------------------
 class DerivedThread : Thread
 {
     int [] X;
     int i;
     this(int [] X, int i){
         this.X = X;
 	this.i = i;
         super(&run);
     }

     private:
         void run(){
             X ~= i;
         }
 }

 void main(){
 	auto I = std.range.iota(0,500);
 	int [] X; // output
 	Thread [] threads;
 	foreach (i; I )
 		threads ~= new DerivedThread( X,i);
 	foreach( thread; threads)
 		thread.start();
 	foreach( thread; threads)
 		thread.join(); // does not seem to do anything
 	core.thread.thread_joinAll(); // also not doing anything

 	writeln(X); // X contains nothing at all
 }

 How can I get the program to wait until all threads have 
 finished before moving to the next line of code ?

 Thank you !
Just like a sequential loop, when you do "foreach (i; parallel(I) ) { ... }", execution will not continue past the foreach loop until all the tasks associated with each element of I have finished. Your particular example of "X ~= i" in the body of the loop is not thread-safe, so if that is the code you really intend to run, you should protect X with a Mutex or something comparable.
Aug 31
prev sibling parent reply =?UTF-8?Q?Ali_=c3=87ehreli?= <acehreli yahoo.com> writes:
On 08/31/2017 06:59 PM, Brian wrote:
 Hello, I am trying to get the most trivial example of multithreading
 working, but can't seem to figure it out.
 I want to split a task across threads, and wait for all those tasks to
 finish before moving to the next line of code.

 The following 2 attempts have failed :

 -----------------------------------------------------
 Trial 1 :
 -----------------------------------------------------

 auto I = std.range.iota(0,500);
 int [] X; // output
 foreach (i; parallel(I) )
     X ~= i;
 core.thread.thread_joinAll(); // Apparently no applicable here ?
As Michael Coulombe said, parallel() does that implicitly. If the problem is to generate numbers in parallel, I restructured the code by letting each thread touch only its element of a results array that has already been resized for all the results (so that there is no race condition): import std.stdio; import std.parallelism; import std.range; void main() { auto arrs = new int[][](totalCPUs); const perWorker = 10; foreach (i, arr; parallel(arrs)) { const beg = cast(int)i * perWorker; const end = beg + perWorker; arrs[i] = std.range.iota(beg,end).array; } writeln(arrs); } If needed, std.algorithm.joiner can be used to make it a single sequence of ints: import std.algorithm; writeln(arrs.joiner); Ali
Aug 31
parent reply Brian <brian.patrick.forbes gmail.com> writes:
On Friday, 1 September 2017 at 04:43:29 UTC, Ali Çehreli wrote:
 On 08/31/2017 06:59 PM, Brian wrote:
 Hello, I am trying to get the most trivial example of
multithreading
 working, but can't seem to figure it out.
 I want to split a task across threads, and wait for all those
tasks to
 finish before moving to the next line of code.

 The following 2 attempts have failed :

 -----------------------------------------------------
 Trial 1 :
 -----------------------------------------------------

 auto I = std.range.iota(0,500);
 int [] X; // output
 foreach (i; parallel(I) )
     X ~= i;
 core.thread.thread_joinAll(); // Apparently no applicable
here ? As Michael Coulombe said, parallel() does that implicitly. If the problem is to generate numbers in parallel, I restructured the code by letting each thread touch only its element of a results array that has already been resized for all the results (so that there is no race condition): import std.stdio; import std.parallelism; import std.range; void main() { auto arrs = new int[][](totalCPUs); const perWorker = 10; foreach (i, arr; parallel(arrs)) { const beg = cast(int)i * perWorker; const end = beg + perWorker; arrs[i] = std.range.iota(beg,end).array; } writeln(arrs); } If needed, std.algorithm.joiner can be used to make it a single sequence of ints: import std.algorithm; writeln(arrs.joiner); Ali
Hello, thank you very much for your quick replies ! I was trying to make a trivial example, but the 'real' problem is trying to split a huge calculation to different threads. Schematically : double [] hugeCalc(int i){ // Code that takes a long time } so if I do double[][int] _hugeCalcCache; foreach(i ; I) _hugeCalcCache[i] = hugeCalc(i); of course the required time is I.length * (a long time), so I wanted to shorten this by splitting to different threads : foreach(i ; parallel(I) ) _hugeCalcCache[i] = hugeCalc(i); but as you can guess, it doesn't work that easily. Very interesting approach about letting only the thread touch a particular element, I will try that. FYI I did manage to make the following work, but not sure if this is really still multi-threaded ? int [] I; foreach (i; 0 .. 500) I ~= i; int [] X; // output class DerivedThread : Thread { private int [] i; this(int [] i){ this.i = i; super(&run); } private void run(){ synchronized{ // Need synchronization here ! foreach( i0; i) X ~= i0; } } } Thread [] threads; foreach (i; std.range.chunks( I, 50 ) ) threads ~= new DerivedThread( i); foreach( thread; threads) thread.start(); core.thread.thread_joinAll(); // Does in fact seem to 'join all' threads writeln(X);
Aug 31
next sibling parent =?UTF-8?Q?Ali_=c3=87ehreli?= <acehreli yahoo.com> writes:
On 08/31/2017 10:27 PM, Brian wrote:

 the 'real' problem is trying
 to split a huge calculation to different threads.
I still think you can take advantage of std.parallelism: https://dlang.org/phobos/std_parallelism.html Unfortunately, its features like asyncBuf, map, and amap do not stand out in the documentation. Here's my interpretation of them: http://ddili.org/ders/d.en/parallelism.html Ali
Sep 01
prev sibling parent reply ag0aep6g <anonymous example.com> writes:
On 09/01/2017 07:27 AM, Brian wrote:
 double [] hugeCalc(int i){
      // Code that takes a long time
 }
 
 so if I do
 
 
 double[][int] _hugeCalcCache;
 foreach(i ; I)
     _hugeCalcCache[i] = hugeCalc(i);
 
 of course the required time is I.length * (a long time), so I wanted to 
 shorten this by splitting to different threads :
 
 foreach(i ; parallel(I) )
     _hugeCalcCache[i] = hugeCalc(i);
 
 but as you can guess, it doesn't work that easily.
Works pretty well for me: ---- double [] hugeCalc(int i) { // Code that takes a long time import core.thread: Thread; import std.datetime: seconds; Thread.sleep(1.seconds); return [i]; } void main() { static import std.range; import std.parallelism: parallel; auto I = std.range.iota(0, 10); double[][int] _hugeCalcCache; foreach(i ; parallel(I)) _hugeCalcCache[i] = hugeCalc(i); } ---- That runs in about 3 seconds here. The serial version would of course take about 10 seconds. So, parallelism achieved! Though I don't know if it's safe to access an associative array concurrently like that. I'd use a normal dynamic array instead and initialize it before going parallel: ---- auto _hugeCalcCache = new double[][](10); ----
Sep 01
parent reply Brian <brian.patrick.forbes gmail.com> writes:
On Friday, 1 September 2017 at 20:02:23 UTC, ag0aep6g wrote:
 On 09/01/2017 07:27 AM, Brian wrote:
 double [] hugeCalc(int i){
      // Code that takes a long time
 }
 
 so if I do
 
 
 double[][int] _hugeCalcCache;
 foreach(i ; I)
     _hugeCalcCache[i] = hugeCalc(i);
 
 of course the required time is I.length * (a long time), so I 
 wanted to shorten this by splitting to different threads :
 
 foreach(i ; parallel(I) )
     _hugeCalcCache[i] = hugeCalc(i);
 
 but as you can guess, it doesn't work that easily.
Works pretty well for me: ---- double [] hugeCalc(int i) { // Code that takes a long time import core.thread: Thread; import std.datetime: seconds; Thread.sleep(1.seconds); return [i]; } void main() { static import std.range; import std.parallelism: parallel; auto I = std.range.iota(0, 10); double[][int] _hugeCalcCache; foreach(i ; parallel(I)) _hugeCalcCache[i] = hugeCalc(i); } ---- That runs in about 3 seconds here. The serial version would of course take about 10 seconds. So, parallelism achieved! Though I don't know if it's safe to access an associative array concurrently like that. I'd use a normal dynamic array instead and initialize it before going parallel: ---- auto _hugeCalcCache = new double[][](10); ----
Thanks very much for your help, I finally had time to try your suggestions. The initial example you showed does indeed have the same problem of not iterating over all values : double [] hugeCalc(int i){ // Code that takes a long time import core.thread: Thread; import std.datetime: seconds; Thread.sleep(1.seconds); return [i]; } static import std.range; import std.parallelism: parallel; auto I = std.range.iota(0, 100); double[][int] _hugeCalcCache; foreach(i ; parallel(I)) _hugeCalcCache[i] = hugeCalc(i); writeln( _hugeCalcCache.keys ); // this is some random subset of (0,100) but this does seem to work using your other method of initialization : auto _hugeCalcCache = new double[][](100); foreach(i ; parallel(I)) _hugeCalcCache[i] = hugeCalc(i); foreach( double[] x ; _hugeCalcCache) writeln( x ); // this now contains all values so I guess initializing the whole array at compile time makes it thread safe ? (The second case runs in 16 seconds on my computer.) Anyways it seems to work, thanks again !
Sep 04
parent ag0aep6g <anonymous example.com> writes:
On 09/05/2017 03:15 AM, Brian wrote:
 Thanks very much for your help, I finally had time to try your 
 suggestions. The initial example you showed does indeed have the same 
 problem of not iterating over all values :
 
 
      double [] hugeCalc(int i){
      // Code that takes a long time
      import core.thread: Thread;
      import std.datetime: seconds;
      Thread.sleep(1.seconds);
      return [i];
      }
 
      static import std.range;
      import std.parallelism: parallel;
      auto I = std.range.iota(0, 100);
      double[][int] _hugeCalcCache;
      foreach(i ; parallel(I))
          _hugeCalcCache[i] = hugeCalc(i);
 
 
       writeln( _hugeCalcCache.keys ); // this is some random subset of 
 (0,100)
Yeah. As expected, associative array accesses are apparently not thread-safe. A simple writeln is a terrible way to figure that out, though. I'd suggest sorting the keys and comparing that to `I`: ---- import std.algorithm: equal, sort; auto sortedKeys = _hugeCalcCache.keys.sort; assert(sortedKeys.equal(I)); ----
 but this does seem to work using your other method of initialization :
 
 
      auto _hugeCalcCache = new double[][](100);
      foreach(i ; parallel(I))
          _hugeCalcCache[i] = hugeCalc(i);
 
      foreach( double[] x ; _hugeCalcCache)
      writeln( x ); // this now contains all values
 
 
 so I guess initializing the whole array at compile time makes it thread 
 safe ?
There's nothing compile-timey about the code. The initialization is done at run-time, but before the parallel stuff starts. Note that the type of `_hugeCalcCache` here is different from above. Here it's `double[][]`, i.e. a dynamic array. Above it's `double[][int]`, i.e. an associative array. Those types are quite different, despite their similar names. You can prepare an associative array in a similar way, before doing the parallel stuff. Then it might be thread-safe (not sure): ---- double[][int] _hugeCalcCache; /* associative array */ /* First initialize the elements serially: */ foreach(i; I) _hugeCalcCache[i] = []; /* Then do the huge calculations in parallel: */ foreach(i; parallel(I)) _hugeCalcCache[i] = hugeCalc(i); ---- But if your keys are consecutive numbers, I see no point in using an associative array.
Sep 05