關(guān)于perl6并發(fā)的問題

aef25u

簡介：
1、開啟各15個線程的1個生產(chǎn)者與2個消費(fèi)者A、B（AB同為耗時任務(wù)）
2、生產(chǎn)者按條件通過2個Channel向A與B分別發(fā)送數(shù)據(jù)（數(shù)據(jù)不需按原始順序發(fā)送；按業(yè)務(wù)邏輯會發(fā)送不同數(shù)據(jù)，暫時按發(fā)送同一數(shù)據(jù)模擬）
3、數(shù)組@share為共享數(shù)據(jù)，消費(fèi)者A、B會向其讀取數(shù)據(jù)（不需寫入或改變元素）
4、消費(fèi)者A處理接收的數(shù)據(jù)，不符合業(yè)務(wù)邏輯的數(shù)據(jù)會繼續(xù)傳入B進(jìn)行處理  if $v==999 {$supplierB.emit($v);}
問題：
1、這樣組織代碼有沒有不合理的地方？
2、共享的@share數(shù)組安不安全？
3、假如在消費(fèi)者A、B中使用以下cached 函數(shù)，在各自不同線程下能起作用不？

     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);}#向消費(fèi)者B發(fā)送數(shù)據(jù)，真實情況是發(fā)送$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任務(wù)比A任務(wù)耗時
    
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;

復(fù)制代碼

自已發(fā)現(xiàn)了個問題，要保證A處理完不符合業(yè)務(wù)邏輯的數(shù)據(jù)能傳到B不能這樣寫：

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

復(fù)制代碼

aef25u

自已發(fā)現(xiàn)個問題，要保證A處理完不符業(yè)務(wù)邏輯的數(shù)據(jù)能傳到B，不能這書寫

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

復(fù)制代碼

而要這樣寫

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

復(fù)制代碼

第一種寫法可能出現(xiàn)A還沒處理完$channelB就被關(guān)閉了。

aef25u

通過自已上網(wǎng)搜索與親自測試試驗結(jié)果，初步可以回答之前提的問題，如果有不正確的地方，希望大神們指正。
1、這樣組織代碼有沒有不合理的地方？
     這樣組織可以滿足我的任務(wù)，但之前認(rèn)為@pA與@pB內(nèi)不需要sleep是錯誤的。
     供應(yīng)者需 sleep 0；消費(fèi)者@pA與@pB分別為 sleep 0.2 sleep 0.1，這樣有利于各個線程跑的任務(wù)相對均衡（因我使用的是Windows操作系統(tǒng)，CPU競爭策略屬于搶占式）。當(dāng)然 @pA設(shè)成與@pB一樣也沒問題，但 @pAsleep時間不能太大，否則可能出現(xiàn)B先結(jié)束，而@pA仍向關(guān)閉的通道B發(fā)送數(shù)據(jù)，但發(fā)送不了，任務(wù)永遠(yuǎn)不會結(jié)束。


2、共享的@share數(shù)組安不安全？
    共享的資源@share的讀寫均是安全的。

3、假如在消費(fèi)者A、B中使用以下cached 函數(shù)，在各自不同線程下能起作用不？
     use experimental :cached;
     sub fun-name( $val1, $val2) is cached {...}
    這個對我的任務(wù)作用不大，而我的任務(wù)要求并發(fā)數(shù)大（在我的任務(wù)中我使用了100個并發(fā)線程），而分配到的各個線程的任務(wù)數(shù)小，再加上任務(wù)是隨機(jī)分到各線程的，需使用cached的值的機(jī)會很小

最后，給出更完善的代碼：

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. #查看各線程最終分配執(zhí)行的任務(wù)數(shù)
    
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;

復(fù)制代碼