向中间工作人员发送消息时，工作人员未完成

Question

我正在使用"D编程"来学习D语言。我写了一个简单的程序，生成一个工人并发送一个数字以接收其正方形作为字符串。工作线程 1 获取数字的平方并发送给工作线程 2（不同的函数）以转换为字符串，该字符串返回到工作线程 1，因此它将其返回到主函数调用。我可以在一个线程中编写整个内容。我写它是为了更好地了解工人。我使用 receive 让工人 1 按照输入进行操作。程序如下

import std.stdio;
import std.concurrency;
import std.conv;
import core.thread;
void main() {
  foreach (int num; 1..100) {
    auto square_tid = spawn(&square);
    square_tid.send(num);
    auto square = receiveOnly!string();
    writeln(square);
  }
}

void square() {
  static i = 0;
  receive (
       (int num) {
         auto square = num * num;
         writeln("sqaure : Comes in with " , num , " for " , ++i , " time");
         auto stringWorker = spawn(&stringConverter);
         stringWorker.send(thisTid, square, ownerTid);
       },
       (Tid tid, string str) {
         writeln("comes in string");
         send(tid, "hello");
       });
}
void stringConverter() {
  static i = 0;
  auto params = receiveOnly!(Tid, int, Tid)();
  auto stringified = to!string(params[1]); // Stringify the square
  writeln("string : Comes in with " , params[1], " for " , ++i , " time");
  params[0].send(params[2], stringified); // params[0] - square function tid, params[2] - main function tid
}

我可以接收主函数 tid 并直接将字符串发回。但是当我回到工人 1 时，它被击中并且没有继续前进。如何使线程接收来自主线程和从线程的输入。关于线程的另一个问题：

• 如果我想将 -1 作为数据发送给我的工作线程而不退出它。我该怎么做？
• 是否可以在整个过程中使用单个工作线程，或者我可以像在 foreach 循环中那样使用多个工作线程？
• 本书使用以下代码。为什么它在代码中具有值>= 0 作为显而易见的值。

    import std.stdio;
    import std.concurrency;
    import std.conv;
    void main() {
      Tid worker = spawn(&workerFunc);
      foreach (value; 1 .. 5) {
        worker.send(value);
        double result = receiveOnly!double();
        writefln("sent: %s, received: %s", value, result);
      }
      /* Sending a negative value to the worker so that it
       * terminates. */
      worker.send(-1);
    }
    void workerFunc() {
      int value = 0;
      while (value >= 0) {
        value = receiveOnly!int();
        double result = to!double(value) / 5;
        ownerTid.send(result);
      }
    }

如果我在任何术语中错了，请纠正我。

Answer 1

对于这种任务最好使用 std.parallelism

import std.stdio;
import std.parallelism;
void main() {
    auto squares = new long[100];
    foreach(i, ref elem; parallel(squares)) {
        elem = i * i;
    }
    writeln(squares);
}

并且向工作线程发送 -1 没有问题，它不会仅在明确询问时才退出线程。

这是您尝试的修改版本：

import std.stdio;
import std.concurrency;
import std.conv;
void main() {
    foreach (int num; 1..100) {
        auto square_tid = spawn(&square);
        square_tid.send(num);
        auto square = receiveOnly!string();
        writeln(square);
    }
}

void square() {
    static shared i = 0;
    receive (
        (int num) {
        int square = num * num;
        writeln("sqaure : Comes in with " , num , " for " , ++i , " time");
        auto stringWorker = spawn(&stringConverter);
        stringWorker.send(thisTid, square, ownerTid);
        receive ((Tid tid, string str) { writeln("comes in string"); send(tid, "hello");});
    });
}
void stringConverter() {
    static shared i = 0;
    auto params = receiveOnly!(Tid, int, Tid)();
    auto stringified = to!string(params[1]); // Stringify the square
    writeln("string : Comes in with " , params[1], " for " , ++i , " time");
    params[0].send(params[2], stringified); // params[0] - square function tid, params[2] - main function tid
}

更新说明

代码中的 square 函数在 receive 之后结束。因此，它永远不会尝试(Tid tid, string str)部分的下一个块。这就是为什么我把它放在receive的第一部分.

每次调用spawn都会创建新线程。由于 D 默认使用 TLS，因此 static 关键字在您的示例中毫无用处。因为在每个新线程中i都会0.这就是我使用shared关键字的原因。

更新 2

这是一个可以解释更多工作原理的版本：

import std.stdio;
import std.concurrency;
import std.conv;
void main() {
    foreach (int num; 1..100) {
        auto square_tid = spawn(&square);
        square_tid.send(num);
        auto square = receiveOnly!string();
        writeln(square);
    }
}

void square() {
    shared static i = 0;
    bool end = false;
    while(!end) receive (
        (int num) {
        auto square = num * num;
        writeln("sqaure : Comes in with " , num , " for " , ++i , " time");
        auto stringWorker = spawn(&stringConverter);
        stringWorker.send(square);
    },
    (string str) {
        writeln("comes in string");
        ownerTid.send(str);
        end = true;
    });
}
void stringConverter() {
    shared static i = 0;
    auto params = receiveOnly!(int)();
    auto stringified = to!string(params); // Stringify the square
    writeln("string : Comes in with " , params, " for " , ++i , " time");
    ownerTid.send(stringified); // params[0] - square function tid, params[2] - main function tid
}