PowerShellで 非同期処理 での ref変数へのデータ追加が無視される事があるのですが 確実にref変数へのデータ追加は出来ないのでしょうか?

Question

前提・実現したいこと

非同期で実行した処理結果を参照渡しの変数に追加

発生している問題

データの追加がされていないことがある上に
エラーも発生しない

問題 確認用 サンプルソースコード

無駄に処理を分けてますが確認のためです...
時間が掛かりすぎる場合は$Aryの行数を減らしてください。

<#
    情報源
        guitarrapc_tech 様
            タイトル : PowerShell による同期処理、非同期処理、並列処理 を考えてみる
            URL : http://tech.guitarrapc.com/entry/2013/10/29/100946

        *** 様
            タイトル : [properties] Powershellクラスはget setプロパティを実装します
            URL : https://code-examples.net/ja/q/25e096e
#>
using namespace  System.Management.Automation.Runspaces;
using namespace System.Collections;
using namespace System.Threading;

# クラスを定義します
class _MultiThread
{

   # ノーマル プロパティ
   [array] $RunspaceCollection

   # 非表示 プロパティ
   hidden [RunspacePool] $runspacePool

   # 完了 プロパティ
   hidden [bool]$_AllCompleted = $($this | Add-Member ScriptProperty -Name 'AllCompleted' -Value {
            $($this.runspaceCollection.RunSpace | sort IsCompleted -Unique).IsCompleted -eq $true
        })

   # コンストラクター
   _MultiThread ($MyHost)
   {
        # create Runspace
        [initialsessionstate] $sessionstate = [initialsessionstate]::CreateDefault()
        [int] $minPoolSize = [int] $maxPoolSize = 1000
        $this.runspacePool = [runspacefactory]::CreateRunspacePool($minPoolSize, $maxPoolSize,  $sessionstate, $MyHost) # create Runspace Pool
        $this.runspacePool.ApartmentState = "STA"
        $this.runspacePool.Open() # open pool
   }
   
   # スレッドの追加
   [void]AddThread([scriptblock]$ScriptBlock)
   {
        # Main Invokation
        [powershell] $powershell = [PowerShell]::Create().AddScript($ScriptBlock)
        $powershell.RunspacePool = $this.runspacePool
        $this.RunspaceCollection += New-Object -TypeName PSObject -Property @{
            Runspace = $powershell.BeginInvoke();
            powershell = $powershell
        }
        $powershell.Dispose()
   }

   # スレッドの追加
   [void]AddThread([scriptblock]$ScriptBlock,[psobject]$Parameters)
   {
        # Main Invokation
        [powershell] $powershell = [PowerShell]::Create().AddScript($ScriptBlock).AddParameters($Parameters)
        $powershell.RunspacePool = $this.runspacePool
        $this.RunspaceCollection += New-Object -TypeName PSObject -Property $(
            [ordered]@{
                Runspace = $powershell.BeginInvoke();
                powershell = $powershell
            }
        )
   }


   # 結果の取得
   [object]GetResult()
   {
        # get process result and end powershell session
        return $(foreach ($runspace in $this.runspaceCollection)
        {
            # get reuslt
            $runspace.powershell.EndInvoke($runspace.Runspace)
        })
   }

   # デストラクタの代わり手動で実行
   [void]Dispose()
   {
        if (!($this.runspacePool.runspacePool.IsDisposed))
        {
            foreach ($Runspace in $this.RunspaceCollection)
            {
                $Runspace.powershell.Dispose()
            }

            # Dispose Runspace
            $this.runspacePool.Dispose()
        }
   }
}

function _New_MultiThread()
{
[OutputType([_MultiThread])]
    $Thread = New-Object _MultiThread($host)
    return $Thread
}


#検索対象データの読み込み
$_Sarch_List_Read = {

    Param
    (
        # 読み込む二次元配列
        [Parameter(Mandatory,
                    Position=0)]
        [ref]
        $Ary,
            
        # 読み込み開始位置
        [Parameter(Mandatory,
                    Position=1)]
        [int]
        $Start,
            
        # 読み込み終了位置
        [Parameter(Mandatory,
                    Position=2)]
        [int]
        $End,

        # 取得データの格納変数
        [Parameter(Mandatory,
                    Position=3)]
        [ref]
        $CSV
    )
    for ($i = $Start; $i -le $End; $i++)
    { 
        $CSV_Data = New-Object psobject | select -Property No,Category,Type,PutName,ListPrice,Purchase,PrimeCost,SellingPrice,Maker,DisplayOrderNo,EquipmentName,CharacterSymbol
        $CSV_Data.No = $Ary.Value[$i,1]
        $CSV_Data.Category = $Ary.Value[$i,2]
        $CSV_Data.Type = $Ary.Value[$i,3]
        $CSV_Data.PutName = $Ary.Value[$i,4]
        $CSV_Data.ListPrice = $Ary.Value[$i,5]
        $CSV_Data.Purchase = $Ary.Value[$i,6]
        $CSV_Data.PrimeCost = $Ary.Value[$i,7]
        $CSV_Data.SellingPrice = $Ary.Value[$i,8]
        $CSV_Data.Maker = $Ary.Value[$i,9]
        $CSV_Data.DisplayOrderNo = $Ary.Value[$i,10]
        $CSV_Data.EquipmentName = $Ary.Value[$i,11]
        $CSV_Data.CharacterSymbol = $Ary.Value[$i,12]

        $CSV.Value += $CSV_Data

    }
}

# 20000 x 12 の配列を作成 元がExcelから持ってきてるので 添え字は1から始まりと仮定する
$Ary = New-Object -TypeName 'System.Object[,]' -ArgumentList 20001,13
$i = 0

# 適当にデータ代入
foreach ($item in $Ary)
{
    $_ = $i
    $i++
}


    $MultiThread = _New_MultiThread
    
Write-Host "検索テーブルを読み込んでいます..."
$Search = @()
$i = 0
1..$Ary.GetUpperBound(0) | foreach {
    $i++
    $MultiThread.AddThread($_Sarch_List_Read,[ordered]@{
            Ary = [ref]$Ary
            Start = $_
            End = $_
            CSV = [ref]$Search
    })
}
Write-host "処理を実行した回数 = $i"

while (!($MultiThread.AllCompleted)){}
$MultiThread.Dispose()
Write-host "登録されたデータ数 = $($Search.Count)"

試したこと

上記 サンプルコードを作成、および実行して データの内容は関係ないことの確認
実行するたびに登録されたデータ数が変わる...

ひとまず refつかっての受け取りは諦めて
処理完了後にシリアル処理で戻り値を順に変数に格納~~しようと思っています。~~しました。

function _New_MultiThread()
{
[OutputType([_MultiThread])]
    $Thread = New-Object _MultiThread($host)
    return $Thread
}


#検索対象データの読み込み
$_Sarch_List_Read = {

    Param
    (
        # 読み込む二次元配列
        [Parameter(Mandatory,
                    Position=0)]
        $Ary,
            
        # 読み込み開始位置
        [Parameter(Mandatory,
                    Position=1)]
        [int]
        $Start,
            
        # 読み込み終了位置
        [Parameter(Mandatory,
                    Position=2)]
        [int]
        $End
    )
    for ($i = $Start; $i -le $End; $i++)
    { 
        $CSV_Data = New-Object psobject | select -Property No,Category,Type,PutName,ListPrice,Purchase,PrimeCost,SellingPrice,Maker,DisplayOrderNo,EquipmentName,CharacterSymbol
        $CSV_Data.No = $Ary[$i,1]
        $CSV_Data.Category = $Ary[$i,2]
        $CSV_Data.Type = $Ary[$i,3]
        $CSV_Data.PutName = $Ary[$i,4]
        $CSV_Data.ListPrice = $Ary[$i,5]
        $CSV_Data.Purchase = $Ary[$i,6]
        $CSV_Data.PrimeCost = $Ary[$i,7]
        $CSV_Data.SellingPrice = $Ary[$i,8]
        $CSV_Data.Maker = $Ary[$i,9]
        $CSV_Data.DisplayOrderNo = $Ary[$i,10]
        $CSV_Data.EquipmentName = $Ary[$i,11]
        $CSV_Data.CharacterSymbol = $Ary[$i,12]

        $CSV_Data

    }
}

# 20000 x 12 の配列を作成 元がExcelから持ってきてるので 添え字は1から始まりと仮定する
$Ary = New-Object -TypeName 'System.Object[,]' -ArgumentList 20001,13
$i = 0

# 適当にデータ代入
foreach ($item in $Ary)
{
    $_ = $i
    $i++
}


    $MultiThread = _New_MultiThread
    
Write-Host "検索テーブルを読み込んでいます..."
$Max = $Ary.GetUpperBound(0)
Write-host "対象データ行数 = $($Max)"
$Step = 2000
for ($i = 1; $i -le $Max; $i+=$Step)
{ 
    if ($($i + $Step - 1) -le $Max)
    {
        $End = $($i + $Step - 1)
    }
    else
    {
        $End = $Max
    }
    $MultiThread.AddThread($_Sarch_List_Read,[ordered]@{
        Ary = $Ary
        Start = $i
        End = $End
    })
}

while (!($MultiThread.AllCompleted)){}
$Search = @($MultiThread.GetResult() | foreach {$_})
$MultiThread.Dispose()
Write-host "登録されたデータ数 = $($Search.Count)"

補足情報（FW/ツールのバージョンなど）

Win10

Answer 1

PowerShellの配列に対する+=演算子は、新しく配列を作り直す処理のはずなので、作り直す間に他の処理が入ると追加に失敗する、と予想しました。

非同期処理なら…ということで
$Search = @()を
$Search = New-Object -TypeName Concurrent.ConcurrentQueue[psobject]
へ変更し、
$_Sarch_List_Read内の$CSV.Value += $CSV_Dataを
while( -not $CSV.Value.TryAdd($CSV_Data) ) { }
へ変更したところ、数が一致するようになりました。
非同期処理は詳しくないので、もっと良い方法があるかもしれませんが。

また、この場合[ref]にせず直接キューの参照を渡しても、「登録されたデータ数」の値は「処理を実行した回数」と一致しました。

Comments

[kamikazelight]

ありがとうございます。
ちゃんとマルチスレッド処理等用のオブジェクトがあったのですね。
全く知りませんでした。
今の作業がひと段落してから試してみたいと思います。

分からない事があるのですが「直接キューの参照を渡す」には どうしたらいいのでしょうか？
そもそも「キュー」自体 どっかの書籍でデータを入れた順番で取り出せる変数? 的な内容が書いてあったのを見たことがあるだけで 使ったことがあるのかさえ分かりません。
PowerShellで使えるのでしょうか?　是非 教えて頂きたいです。
...別の質問で再度投稿したほうがいいでしょうか?

[imihito]

「キュー」に関しては私の表記ぶれです、すみません。
ConcurrentQueue→スレッドセーフなキュー
回答中の「ConcurrentQueue」と「キュー」は同じものを示しています。
また、スレッドセーフなコレクションであればなんでも良かったので、その中から「ConcurrentQueue」を選択した理由は特にありません。
https://docs.microsoft.com/ja-jp/dotnet/standard/collections/thread-safe/

「ConcurrentQueue」などはクラス=参照型のため、[ref]を付けずに関数に渡しても、オブジェクトのコピーではなく、そのオブジェクトへの参照（データ本体ではなくデータへのリンクのイメージ）が渡されます。
他の関数に渡しても、渡しているのはリンクで元のオブジェクトは一つのため、`TryAdd`などのメソッドでオブジェクトの中身を変更できるということになります。

# 配列で[ref]が必要な理由
VBAと異なり、配列も参照型（正確にはVBAの挙動が特殊例）ですが、配列に対して+=演算子を使うと新しく配列を作り直します。
元の配列とは別のものになるため、引数としてもらったリンクそのものを書き換える必要があるため[ref]にしないといけません。

[kamikazelight]

おかげさまでイメージが湧きました。
分かりやすい回答ありがとうございました。