raku - 共享电源可以同时运行多个分接头 block 吗？

Question

考虑此代码，其中点击需要一段时间才能完成。所有块同时运行(立即输出)然后休眠。大多数没有完成，因为程序结束得比他们做的要早:

my $supply = Supply.interval(0.2);
my $tap = $supply.tap: { say "1 $^a"; sleep 5;  };
sleep 5;

输出(省略)有 25 行(每 5 秒内 0.2 的每个刻度线):

1. 0
1. 1
...
1. 24

然后我将该供应更改为 .share :

my $supply = Supply.interval(0.2).share;
my $tap = $supply.tap: { say "1. $^a"; sleep 5 };
sleep 5;

我只看到一行输入，但我期望得到相同的输出:

1. 1

.share使得多次点击可以获得相同的值成为可能。

my $supply = Supply.interval(0.2).share;
my $tap  = $supply.tap: { say "1. $^a"; sleep 5 };
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

输出仍然只有第一个抽头的输出并且仍然只有一行。我预计每个有 25 行:

1. 1

Answer 1

Supply的基本规则是:

在没有明确要求的情况下不引入并发

通过发送方支付模型的背压

一条消息在下一条消息之前被完整处理(因此 .map({ ...something with state... }) 可以信任不会导致状态冲突)

规则 3 并不真正适用于 share因为在那之后有单独的下游操作链，但规则 1 和 2 确实如此。 share的目的是为了允许发布/订阅，并且还允许多个下游消息处理器重用处理块。引入并行消息处理是一个单独的问题。

这是各种选择。一种是将并行处理的消息放入 Channel .这明确地引入了一个消息缓冲的地方(好吧，直到你用完内存......这正是 Supply 带有发送者支付背压模型的原因)。胁迫 Channel回到 Supply获取从 Channel 中提取的值并在 Supply 上发出在池线程上。这种方式看起来像:

my $supply = Supply.interval(0.2).share;
my $tap  = $supply.Channel.Supply.tap: { say "1. $^a"; sleep 5 };
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

请注意，由于 whenever自动强制要求它对 Supply 使用react的事物，然后看起来像 whenever $supply.Channel { } ，这使它成为一个非常简短的解决方案 - 但同时也非常明确，因为它表明了正常的背压机制是如何被回避的。该解决方案的另一个特性是它保留了消息的顺序，并且仍然在 Channel 的下游提供一次一次的处理。 .

另一种方法是通过启动一些异步工作来处理每个消息来对每个消息使用react。 start对 Supply 的操作为收到的每条消息调度它传递的块以在线程池上运行，从而不阻止下一条消息的到达。结果是 Supply的 Supply .这迫使一个人点击每个内部 Supply实际上让任何事情发生，起初这似乎有点违反直觉，但实际上是为了程序员的利益:它清楚地表明有一些额外的异步工作需要跟踪。我强烈建议将此与 react 结合使用。/ whenever语法，自动进行订阅管理和错误传播。题中代码最直接的改造是:

my $supply = Supply.interval(0.2).share;
my $tap  = supply { whenever $supply.start({ say "1. $^a"; sleep 5 }) { whenever $_ {} } }.tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

尽管也可以将其写为:

my $supply = Supply.interval(0.2).share;
my $tap  = supply { whenever $supply -> $a { whenever start { say "1. $a"; sleep 5 } {} } }.tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

这指出了写 parallelize 的可能性 Supply组合器:

my $supply = Supply.interval(0.2).share;
my $tap  = parallelize($supply, { say "1. $^a"; sleep 5 }).tap;
my $tap2 = $supply.tap: { say "2. $^a";  };
sleep 5;

sub parallelize(Supply $messages, &operation) {
    supply {
        whenever $messages -> $value {
            whenever start operation($value) {
                emit $_;
            }
        }
     }
}

这种方法的输出与 Channel 有很大不同。一，因为一旦消息进来，操作就全部开始了。而且它不保留消息顺序。仍然有一个隐式队列(与 Channel 方法的显式队列不同)，只是现在是线程池调度程序的工作队列和操作系统调度程序必须跟踪正在进行的工作。同样，没有背压，但请注意，通过跟踪未完成的 Promises 完全有可能实现这一点。并使用 await Promise.anyof(@outstanding) 阻止进一步传入的消息.

最后，我会注意到对 hyper whenever 有一些考虑。和 race whenever构造以提供一些语言级别的机制来处理 Supply 的并行处理消息。然而这样的语义，以及它们是如何发挥作用的 supply -block 设计目标和安全特性，代表了重大的设计挑战。