工人在向中间工人发送消息时未完成

Question

我正在使用"programming in D"学习D语言。我写了一个简单的程序来生成一个 worker 并向它发送一个数字以将其方块作为字符串接收。工人 1 得到数字的平方并发送给工人 2（一个不同的函数）以转换为字符串，该字符串被 returned 给工人 1，因此它 returns 到主函数调用.我可以在一个线程中编写整个内容。我写它是为了更好地了解工人。我使用 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，直接把字符串传回去。但是当我 return 到 worker 1 时，它被击中并且不再继续。我怎样才能让一个线程从主线程和从线程接收输入。关于线程的另外几个问题：

如果我想在不退出的情况下将 -1 作为数据发送给我的工作人员。我该怎么做？
在整个过程中使用一个 worker 是否可以，或者我可以像在 foreach 循环中那样使用多个 worker？
本书使用了以下代码。为什么它的值 >= 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给worker线程也没有问题，只有在明确询问时才会退出线程。

这是您尝试的修改版本：

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
}

工人在向中间工人发送消息时未完成

Worker is not finished while sending message to intermediate worker

concurrency

multithreading

d