关于perl6并发的问题

aef25u

简介：
1、开启各15个线程的1个生产者与2个消费者A、B（AB同为耗时任务）
2、生产者按条件通过2个Channel向A与B分别发送数据（数据不需按原始顺序发送；按业务逻辑会发送不同数据，暂时按发送同一数据模拟）
3、数组@share为共享数据，消费者A、B会向其读取数据（不需写入或改变元素）
4、消费者A处理接收的数据，不符合业务逻辑的数据会继续传入B进行处理  if $v==999 {$supplierB.emit($v);}
问题：
1、这样组织代码有没有不合理的地方？
2、共享的@share数组安不安全？
3、假如在消费者A、B中使用以下cached 函数，在各自不同线程下能起作用不？

     use experimental :cached;
     sub fun-name( $val1, $val2) is cached {...}

1. my $TIME = now;
   
2. my $supplierA = Supplier.new;
   
3. my $channelA = $supplierA.Supply.Channel;
   
4. my $supplierB = Supplier.new;
   
5. my $channelB = $supplierB.Supply.Channel;
   
6. my @share=("C","D");
   
7. my $threads=15;
   
8. my (@pA,@pB);
   
9. for 1 .. $threads {
   
10. 
    
11.   @pA.push: start {
    
12.       react  {
    
13.             whenever $channelA -> $v {
    
14.               if $v==999 {$supplierB.emit(1000);}#向消费者B发送数据，真实情况是发送$v
    
15.               say "channelA shareArr {@share[0]}:Thread {$*THREAD.id} got $v";
    
16. 
    
17.           }
    
18.      }
    
19.   }
    
20.   @pB.push: start {
    
21.       react  {
    
22.             whenever $channelB -> $v {
    
23.               sleep 0.5;#真实代码不需这一行，模拟单个B任务比A任务耗时
    
24.              say "channelB shareArr {@share[1]}:Thread {$*THREAD.id} got $v";
    
25.           }
    
26.      }
    
27.   }
    
28. }
    
29. 
    
30. my @promises;
    
31. for ^1000 -> $r {
    
32.     push @promises, start {
    
33.           #sleep $r*0.001;
    
34.           sleep rand;
    
35.           $supplierA.emit($r);
    
36.           $supplierB.emit($r);
    
37.     };
    
38.     if @promises == 15 {
    
39.          await Promise.anyof(@promises);
    
40.          @promises .= grep({ !$_ });
    
41.     }
    
42. }
    
43. 
    
44. await @promises;
    
45. 
    
46. $supplierA.done;
    
47. await @pA;
    
48. $supplierB.done;
    
49. await @pB;
    
50. $TIME = now - $TIME;
    
51. say $TIME;

复制代码

自已发现了个问题，要保证A处理完不符合业务逻辑的数据能传到B不能这样写：

1. $supplierA.done;
   
2. $supplierB.done;
   
3. await @pA,@pB;

复制代码

aef25u

自已发现个问题，要保证A处理完不符业务逻辑的数据能传到B，不能这书写

1. $supplierA.done;
   
2. $supplierB.done;
   
3. await @pA,@pB;

复制代码

而要这样写

1. $supplierA.done;
   
2. await @pA;
   
3. $supplierB.done;
   
4. await @pB;

复制代码

第一种写法可能出现A还没处理完$channelB就被关闭了。

aef25u

通过自已上网搜索与亲自测试试验结果，初步可以回答之前提的问题，如果有不正确的地方，希望大神们指正。
1、这样组织代码有没有不合理的地方？
     这样组织可以满足我的任务，但之前认为@pA与@pB内不需要sleep是错误的。
     供应者需 sleep 0；消费者@pA与@pB分别为 sleep 0.2 sleep 0.1，这样有利于各个线程跑的任务相对均衡（因我使用的是Windows操作系统，CPU竞争策略属于抢占式）。当然 @pA设成与@pB一样也没问题，但 @pAsleep时间不能太大，否则可能出现B先结束，而@pA仍向关闭的通道B发送数据，但发送不了，任务永远不会结束。


2、共享的@share数组安不安全？
    共享的资源@share的读写均是安全的。

3、假如在消费者A、B中使用以下cached 函数，在各自不同线程下能起作用不？
     use experimental :cached;
     sub fun-name( $val1, $val2) is cached {...}
    这个对我的任务作用不大，而我的任务要求并发数大（在我的任务中我使用了100个并发线程），而分配到的各个线程的任务数小，再加上任务是随机分到各线程的，需使用cached的值的机会很小

最后，给出更完善的代码：

1. my $TIME = now;
   
2. my $supplierA = Supplier.new;
   
3. my $channelA = $supplierA.Supply.Channel;
   
4. my $supplierB = Supplier.new;
   
5. my $channelB = $supplierB.Supply.Channel;
   
6. my @share=("C","D");
   
7. my $threads=15;
   
8. my (@pA,@pB);
   
9. for 1 .. $threads {
   
10. 
    
11.   @pA.push: start {
    
12.     my @aFir;
    
13.     react  {
    
14.         whenever $channelA -> $v {
    
15.           sleep 0.2;
    
16.           if $v % 2 {$supplierB.emit(1000);}
    
17.           say "channelA shareArr {@share[0]}:Thread {$*THREAD.id} got $v";
    
18.           #my $h={"Val"=>$v};
    
19.           @aFir.push: $v*2;
    
20.         }
    
21.     }
    
22.     @aFir;
    
23.   }
    
24.   @pB.push: start {
    
25. 
    
26.     my @aSec;
    
27.     react  {
    
28.         whenever $channelB -> $v {
    
29.           sleep 0.1;
    
30.           say "channelB shareArr {@share[1]}:Thread {$*THREAD.id} got $v";
    
31.           #my $h={"Val"=>$v};
    
32.           @aSec.push:  $v*2;
    
33.         }
    
34.     }
    
35.     @aSec;
    
36.   }
    
37. }
    
38. sleep 1;
    
39. my @promises;
    
40. for ^1000 -> $r {
    
41.   push @promises, start {
    
42.         sleep 0;
    
43. 
    
44.         $supplierA.emit($r);
    
45.         $supplierB.emit($r);
    
46.     };
    
47.     if @promises == $threads {
    
48.        await Promise.anyof(@promises);
    
49.        @promises .= grep({ !$_ });
    
50.     }
    
51. }
    
52. 
    
53. await @promises;
    
54. 
    
55. $supplierA.done;
    
56. my @retAdrgFIR=await @pA;
    
57. $supplierB.done;
    
58. my @retAdrgSEC=await @pB;
    
59. 
    
60. #@retAdrgFIR .=grep({ $_});
    
61. #@retAdrgSEC .=grep({ $_});
    
62. dd @retAdrgFIR;
    
63. dd @retAdrgSEC;
    
64. 
    
65. #查看各线程最终分配执行的任务数
    
66. my @nFIR=@retAdrgFIR.map( -> $n {$n.List.elems});
    
67. my @nSEC=@retAdrgSEC.map( -> $n {$n.List.elems});
    
68. dd @nFIR;
    
69. dd @nSEC;
    
70. 
    
71. $TIME = now - $TIME;
    
72. say $TIME;

复制代码