關于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;

復制代碼