D Programmierung - Parallelität

Parallelität führt dazu, dass ein Programm auf mehreren Threads gleichzeitig ausgeführt wird. Ein Beispiel für ein gleichzeitiges Programm ist ein Webserver, der gleichzeitig auf viele Clients reagiert. Parallelität ist bei der Nachrichtenübermittlung einfach, aber sehr schwer zu schreiben, wenn sie auf Datenaustausch basieren.

Daten, die zwischen Threads übertragen werden, werden als Nachrichten bezeichnet. Nachrichten können aus einem beliebigen Typ und einer beliebigen Anzahl von Variablen bestehen. Jeder Thread hat eine ID, mit der Empfänger von Nachrichten angegeben werden. Jeder Thread, der einen anderen Thread startet, wird als Eigentümer des neuen Threads bezeichnet.

Threads in D initiieren

Die Funktion spawn () verwendet einen Zeiger als Parameter und startet von dieser Funktion aus einen neuen Thread. Alle Operationen, die von dieser Funktion ausgeführt werden, einschließlich anderer Funktionen, die sie möglicherweise aufruft, werden auf dem neuen Thread ausgeführt. Der Eigentümer und der Arbeiter beginnen beide getrennt auszuführen, als wären sie unabhängige Programme.

Beispiel

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.");  
}

Wenn der obige Code kompiliert und ausgeführt wird, liest er die im vorherigen Abschnitt erstellte Datei und erzeugt das folgende Ergebnis:

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-IDs in D.

Die auf Modulebene global verfügbare Variable thisTid ist immer die ID des aktuellen Threads. Sie können die threadId auch erhalten, wenn spawn aufgerufen wird. Ein Beispiel ist unten gezeigt.

Beispiel

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); 
}

Wenn der obige Code kompiliert und ausgeführt wird, liest er die im vorherigen Abschnitt erstellte Datei und erzeugt das folgende Ergebnis:

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

Nachrichtenübermittlung in D.

Die Funktion send () sendet Nachrichten und die Funktion receiveOnly () wartet auf eine Nachricht eines bestimmten Typs. Es gibt andere Funktionen mit den Namen prioritySend (), receive () und receiveTimeout (), die später erläutert werden.

Der Eigentümer im folgenden Programm sendet seinem Worker eine Nachricht vom Typ int und wartet auf eine Nachricht vom Worker vom Typ double. Die Threads senden weiterhin Nachrichten hin und her, bis der Eigentümer ein negatives int sendet. Ein Beispiel ist unten gezeigt.

Beispiel

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); 
}

Wenn der obige Code kompiliert und ausgeführt wird, liest er die im vorherigen Abschnitt erstellte Datei und erzeugt das folgende Ergebnis:

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

Nachrichtenübergabe mit Wartezeit in D.

Ein einfaches Beispiel für die Nachricht mit Wartezeit wird unten gezeigt.

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"); 
      }
   } 
}

Wenn der obige Code kompiliert und ausgeführt wird, liest er die im vorherigen Abschnitt erstellte Datei und erzeugt das folgende Ergebnis:

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