More than 3 years have passed since last update.

UnityAdvent Calendar 2021

JobParallelForの改良案

Last updated at 2021-12-15Posted at 2021-12-06

はじめに

FixedUpdateのタイミングに合わせてJobSystemのIJobParallelForで並列処理をする書き方を説明しようと思います。
処理速度を追求した内容ではなく使いやすさを優先で実装した例となります。
実装方法については下記を工夫しましたので最後まで読んで頂ければと思います。

Jobのnewは最初の１回で良さそう。
JobのNativeArrayへの参照も１回設定すれば良さそう。
なので、NativeArrayのメモリはAllocator.Persistentを使って毎回newで確保しない。
更新しないデータはNativeArrayに入れたまま使用する。
NativeArray、Arrayの配列へ要素を追加する場合は末尾に追加する。削除する場合は末尾と入れ替えてから削除する。
外からの要素の追加と削除はbufferを設けてJob実行中にも受けとれるようにする。
データの受け渡しはinterfaceを使用して疎結合にする。
ついでにinterfaceがnullか生存確認をして削除する。

前提知識

この記事はJobsystemのIJobParallelForについての詳細は説明しないので使ったことがある方を対象とします。
JobParallelForについてしか扱わないのでJobと省略して書いています。
バージョンはUnity2021.2.4f1とBurst 1.6.3で動作確認をしています。

全体のおおまかな流れ

GameObjetからinterfaceをシングルトンのJobManagerに渡します。
JobManagerはinterfaceを受け取ったら一旦バッファに追加します。
バッファはJobの実行前のタイミングにメモリへ追加または削除します。
メモリの構成は列ごとに１つのオブジェクトのパラメータとして管理します。
データ追加時は最後の列に追加し、データ削除時は使用メモリの最後の列と入れ替えてから削除します。
Jobに必要な値をinterface経由で受け取り、更新してからJobを実行します。
JobはNativeArrayを参照していて列ごとに並列に処理します。
Jobの実行後の結果はinterface経由で返します。

図にしてみましたので参考にしてください。

最初にJobの実装内容を決める

まずは下記の2つを決めます。

Jobで処理する内容と変数を決めます。
Jobの実行と完了のタイミングを決めます。

この記事では一番簡単そうなtransformをJobで更新する場合を実装していきます。
こんな感じにオブジェクトはなんでもいいですが、大量に回りながら落下していくのを想定してます。

1. Jobの実行内容と変数

Transformは非Blittable型なのでNativeArrayで直接扱えません。
そこでVector3とQuaternionはNativeArrayで扱えるので、位置をNativeArray<Vector3>で回転をNativeArray<Quaternion>で受け取ってJobで更新して結果を返すようにします。
他には移動速度をNativeArray<Vector3>に回転速度をNativeArray<Quaternion>にします。
今回のようなTransformを更新するだけの場合はIJobParallelForTransformを使ったほうが良いですが、ここではあえてIJobParallelForで実装します。

JobSystemParallelForTemplate.cs

//Jobを定義
[BurstCompile]
struct MyParallelForJob : IJobParallelFor
{
    public NativeArray<Vector3> positions;
    public NativeArray<Quaternion> rotations;

    [Unity.Collections.ReadOnly]
    public NativeArray<Vector3> moveSpeeds;
    [Unity.Collections.ReadOnly]
    public NativeArray<Quaternion> rotationSpeeds;

    void IJobParallelFor.Execute(int i)
    {
        positions[i] += moveSpeeds[i];
        rotations[i] *= rotationSpeeds[i];

        //高さが-50m以下であれば100m上に移動させる
        if (positions[i].y < -100.0f / 2.0f)
        {
            positions[i] = new Vector3(positions[i].x, positions[i].y + 100.0f, positions[i].z);
        }
    }
}

2. Jobの実行と完了のタイミング

Jobの実行タイミングはなるべく空いている時間に実行して後で結果を受け取るのが理想です。
適切なタイミングをUnity公式のスクリプトライフサイクル等で確認します。

今回は物理演算の後のタイミングにJobを実行し、結果をFixedUpdateの最初のタイミングで受け取ることにします。

Jobの実行タイミング
UnityにLateFixedUpdateのイベントが用意されていれば良かったのですが無いのでコルーチンのWaitForFixedUpdateのタイミングを使用します。
new WaitForFixedUpdate()は毎回newせずにキャッシュして再利用します。
Jobの完了タイミング
FixedUpdateのタイミングにします。
ただし、他のFixedUpdateが実行されるよりも先に計算結果を得たいのでクラス定義の前に[DefaultExecutionOrder(-20)]を追加して実行タイミングを先になるように調整します。

JobSystemParallelForTemplate.cs

[DefaultExecutionOrder(-20)]//実行タイミングを調整する
public class JobSystemParallelForTemplate : MonoBehaviour
{
    WaitForFixedUpdate waitForFixedUpdate = new WaitForFixedUpdate();//キャッシュ
    //実行開始
    void OnEnable()
    {
        //コルーチン開始
        StartCoroutine(LateFixedUpdate());
    }

    void OnDisable()
    {
        //コルーチン停止
        StopCoroutine(LateFixedUpdate());

        //jobが実行中だった場合を考慮
        jobHandle.Complete();
        isJobRunning = false;
    }

    //LateFixedUpdateが無いので代わり
    IEnumerator LateFixedUpdate()
    {
        while (true)
        {
            yield return waitForFixedUpdate;
            ArrayUpdate();//追加依頼と削除依頼のバッファを処理する

            if (ArrayUseLength == 0) continue;//要素がないから終了
            DataUpdate();//データを更新する。
            JobRun();//Jobの実行を開始する
        }
    }

    void FixedUpdate()
    {
        if (ArrayUseLength == 0) return;
        if (isJobRunning == false) return;
        jobHandle.Complete();
        isJobRunning = false;

        ResultReturn();//Jobの結果を返す
    }
}

あとは順番に実装していきます

Jobの実装内容が決まれば残りはほぼ似たような書き方になります。

データを一か所に集めて処理します

一か所に集める方法はなんでもいいですが、jobの実行と結果を受け取るタイミングも利用するのでここではMonoBehaviourをシングルトンで実装します。

このシングルトンの書き方は呼び出したシーンと一緒に削除されます。
マルチシーンで運用する場合はDontDestroyOnLoadを追加したり、先に削除されないようにしてください。

JobSystemParallelForTemplate.cs

//シングルトン
private static JobSystemTemplate instance;
private static bool isDestroy = false;

public static JobSystemParallelForTemplate Instance
{
    get
    {
        if (instance == null)
        {
            if (isDestroy == false)
            {
                instance = FindObjectOfType<JobSystemParallelForTemplate>();
                if (instance == null)
                {
                    instance = new GameObject(typeof(JobSystemParallelForTemplate).Name).AddComponent<JobSystemParallelForTemplate>();
                }
            }
        }
        return instance;
    }
}

外部オブジェクトから登録と削除の依頼をするためのinterfaceを定義します

Jobを実行する側にinterfaceを渡して、渡されたinterface経由で必要なデータを受け渡します。
Jobからみれば外部の実装を考慮しなくてよくなり疎結合にできます。
Jobの結果で直接値を更新しないので後の処理を実装側が柔軟に変更できます。
ここでのinterfaceが持つ役目は下記になります。

外部とのデータの受け渡しの疎結合化
外部のインスタンスの生存確認
追加、削除タイミングの調整
Job実行直前の更新データの取得
Jobの結果を返す

IJobConnector

//インターフェイスを定義
public interface IJobConnector
{
    Vector3 GetPosition();
    Quaternion GetRotation();
    Vector3 GetMoveSpeed();
    Quaternion GetRotationSpeed();
    void SetResult(Vector3 position, Quaternion rotation);
    int MyArrayIndex { get; set; }
}

メンバ変数を定義します

確保するメモリの配列は毎回更新するデータと更新しないデータで分けて、無駄な更新をしないように扱いを分けます。
毎回更新するデータはArrayに入れてからCopyFromを使用してNativeArrayに渡します。
NativeArrayは各要素へのアクセスが遅いので少し手間をかけてCopyFromやCopyToでArray経由で渡した方が速いためです。
更新しないデータはNativeArrayに直接入れてそのまま利用します。

interfaceのIJobConnectorはArrayで保持します。
位置と回転は更新するのでArrayとNativeArrayの2つを定義します。
回転速度は今回は変更しないのでNativeArrayに保持します。
追加と削除用のBufferをListで定義します。
他にもメモリ管理やJobの実行に必要なものを定義します。
myParallelForJob もwaitForFixedUpdateもここでnewでインスタンス化します。

JobSystemParallelForTemplate.cs

//メモリ管理
const int InitMemoryLength = 256;//初期化と追加でメモリを確保する時の配列サイズ
const int JobBatchCount = 0;//Jobのバッチ実行時の分割数(0はコア数に自動設定)
int memoryLength = 0;
int ArrayUseLength = 0;//配列を使用中のlength

//Jobの実行
JobHandle jobHandle;
bool isJobRunning = false;

//WaitForFixedUpdateのキャッシュ
WaitForFixedUpdate waitForFixedUpdate = new WaitForFixedUpdate();

//追加と削除のBuffer
List<IJobConnector> addBuffer = new List<IJobConnector>();
List<IJobConnector> removeBuffer = new List<IJobConnector>();

//Interfaceの参照の保持
IJobConnector[] connectorArray = new IJobConnector[InitMemoryLength];

//NativeArrayにコピーするときに使う普通の配列
Vector3[] positionArray = new Vector3[InitMemoryLength];
Quaternion[] rotationArray = new Quaternion[InitMemoryLength];
Quaternion[] rotationSpeedArray = new Quaternion[InitMemoryLength];

//Jobの計算に使うNativeArrayの配列
NativeArray<Vector3> positions;
NativeArray<Quaternion> rotations;
NativeArray<Vector3> moveSpeeds;
NativeArray<Quaternion> rotationSpeeds;

//Jobを作成
MyParallelForJob myParallelForJob = new MyParallelForJob();

必要な関数を実装します

関数をリストにしてみたら結構多かったですが順番に説明していきます。

InitMemory()
Dispose()
Resize()
Add(IJobConnector item)
Remove(IJobConnector item)
Remove(int index)
ArrayUpdate()
DataUpdate()
ResultReturn()

InitMemory

Awakeのタイミングでメモリを確保します。
確保時のAllocator.Persistentを使用してこのオブジェクトが存在する限り確保し続けます。
Allocator.Persistentはメモリ割り当てと解放が遅いと書かれていますが、毎回Jobを実行する度にメモリを確保して開放をするよりは速いです。
ずっと確保したままなのでJobへの参照もここで設定しておきます。

JobSystemParallelForTemplate.cs

private void Awake()
{
    //メモリを確保
    InitMemory();
}

//Jobで使用するメモリを初期化
void InitMemory()
{
    this.memoryLength = InitMemoryLength;

    //初期設定でメモリを確保
    positions = new NativeArray<Vector3>(this.memoryLength, Allocator.Persistent);
    rotations = new NativeArray<Quaternion>(this.memoryLength, Allocator.Persistent);
    moveSpeeds = new NativeArray<Vector3>(this.memoryLength, Allocator.Persistent);
    rotationSpeeds = new NativeArray<Quaternion>(this.memoryLength, Allocator.Persistent);

    //Jobの参照を設定
    myParallelForJob.positions = positions;
    myParallelForJob.rotations = rotations;
    myParallelForJob.moveSpeeds = moveSpeeds;
    myParallelForJob.rotationSpeeds = rotationSpeeds;
}

Dispose

OnDestroyのタイミングでメモリを解放します。
参照型を入れているArrayもnullにします。

JobSystemParallelForTemplate.cs

void OnDestroy()
{
    //Jobの完了を待つ
    jobHandle.Complete();
    isJobRunning = false;

    //メモリを解放
    DisposeMemory();
    instance = null;
}

//使用メモリの解放
void DisposeMemory()
{
    //NativeArrayの開放
    positions.Dispose();
    rotations.Dispose();
    moveSpeeds.Dispose();
    rotationSpeeds.Dispose();

    //参照を持つArrayをnullにする
    connectorArray = null;
}

Resize

要素を追加するときにサイズが足りなかった場合拡張します。
本来はなるべくリサイズを行わないように設計するのが理想だと思いつつとりあえず実装します。
ArrayはResize()で簡単にリサイズできるのですがNativeArrayにResize関数は無いです。
なので同様に一行で書きたかったので関数を自作しました。
リサイズ後は関数内でNativeArrayをDispose()しているので忘れずにJobの参照も設定しておきます。

JobSystemParallelForTemplate.cs

//使用メモリのサイズ変更
void Resize(int memoryLength)
{
    this.memoryLength = memoryLength;

    //要素のコピーが必要ない場合はfalseを追加
    ResizeNativeArray(ref positions, memoryLength, false);
    ResizeNativeArray(ref rotations, memoryLength, false);
    ResizeNativeArray(ref moveSpeeds, memoryLength, true);
    ResizeNativeArray(ref rotationSpeeds, memoryLength, false);

    //Jobの参照を設定
    myParallelForJob.positions = positions;
    myParallelForJob.rotations = rotations;
    myParallelForJob.moveSpeeds = moveSpeeds;
    myParallelForJob.rotationSpeeds = rotationSpeeds;

    //配列のサイズを変更
    Array.Resize(ref connectorArray, memoryLength);
    Array.Resize(ref positionArray, memoryLength);
    Array.Resize(ref rotationArray, memoryLength);
    Array.Resize(ref rotationSpeedArray, memoryLength);
}

//NativeArrayのリサイズ用関数
void ResizeNativeArray<T>(ref NativeArray<T> nativeArray, int length, bool dataCopy = true) where T : struct
{
    if (dataCopy == true)
    {
        //データをコピーしてリサイズ
        var temp = new T[nativeArray.Length];
        nativeArray.CopyTo(temp);
        nativeArray.Dispose();
        nativeArray = new NativeArray<T>(length, Allocator.Persistent);
        Array.Resize(ref temp, length);//tempのサイズを変更
        nativeArray.CopyFrom(temp);
    }
    else
    {
        //データをコピーしないでリサイズ
        nativeArray.Dispose();
        nativeArray = new NativeArray<T>(length, Allocator.Persistent);
    }
}

AddRequest、RemoveRequest、Add、Remove

AddRequest、RemoveRequestを外部から呼び出してBufferに追加します。
Bufferに追加された要素はWaitForFixedUpdateのタイミングに次で説明するArrayUpdate内でAdd、Removeを呼び出して実際にメモリを更新します。
Bufferを用意する理由はJobの実行中にNativeArrayを変更しないようにするためです。

JobSystemParallelForTemplate.cs

//Job利用の追加依頼をバッファに貯めておく
public void AddRequest(IJobConnector item)
{
    addBuffer.Add(item);
}

//Job利用の削除依頼をバッファに貯めておく
public void RemoveRequest(IJobConnector item)
{
    removeBuffer.Add(item);
}

void Add(IJobConnector item)
{
    //足りなければメモリサイズを拡張
    if (memoryLength < ArrayUseLength + 1)
    {
        //メモリサイズを+1では無くInitMemoryLengthずつ加算で増やす。
        memoryLength += InitMemoryLength;
        Resize(memoryLength);
    }

    //結果を返すために相手のInterfaceを保持する
    connectorArray[ArrayUseLength] = item;

    //毎フレーム更新しないデータはここで代入しておく。
    moveSpeeds[ArrayUseLength] = item.GetMoveSpeed();
    
    item.MyArrayIndex = ArrayUseLength;
    ++ArrayUseLength;
}

void Remove(int index)
{
    if (index != -1)
    {
        --ArrayUseLength;

        //参照するデータは配列の最後と入れ替えてからnullにする。
        connectorArray[index] = connectorArray[ArrayUseLength];
        connectorArray[index].MyArrayIndex = index;//入れ替えた側のIndexを更新
        connectorArray[ArrayUseLength] = null;

        //更新しないデータはindexがずれるので配列の最後と入れ替える。
        moveSpeeds[index] = moveSpeeds[ArrayUseLength];
    }
}

ArrayUpdate、DataUpdate、JobRun

WaitForFixedUpdateから順番にArrayUpdate、DataUpdate、JobRunを呼びます。
呼び出し元のLateFixedUpdateも再掲載しておきます。

ArrayUpdate
外から登録されたBufferでAdd、Removeを呼び出してArrayを更新します。
interfaceがnullになっている場合もここで削除します。
DataUpdate
毎回更新が必要なデータだけを更新します。
NativeArray.CopyFrom()を使用した方が速いのでこういう実装になってます。
更新が必要ないデータはAddの時に代入しておくのでここでは更新しません。
JobRun
Jobを実行します。
JobのScheduleで実行する配列のサイズにArrayUseLengthの値を指定しています。
Jobを実行したらisJobRunningをtrueにします。

JobSystemParallelForTemplate.cs

//LateFixedUpdateが無いので代わり
IEnumerator LateFixedUpdate()
{
    while (true)
    {
        yield return waitForFixedUpdate;
        ArrayUpdate();//追加依頼と削除依頼のバッファを処理する

        if (ArrayUseLength == 0) continue;//要素がないから終了
        DataUpdate();//データを更新する。
        JobRun();//Jobの実行を開始する
    }
}

//配列を更新
void ArrayUpdate()
{
    //connectorArrayがnullになっていた場合は削除(forで逆順処理)
    for (int i = ArrayUseLength - 1; 0 <= i; i--)
    {
        if (connectorArray[i].Equals(null))
        {
            Remove(i);
        }
    }

    //追加バッファを処理
    foreach (IJobConnector item in addBuffer)
    {
        if (item.Equals(null)) continue;
        Add(item);
    }
    addBuffer.Clear();

    //削除バッファを処理
    foreach (IJobConnector item in removeBuffer)
    {
        if (item.Equals(null)) continue;
        //indexと中身が違っていたらエラー
        if (connectorArray[item.MyArrayIndex].Equals(item) == false) { Debug.LogError("No target in the index"); continue; }
        
        Remove(item.MyArrayIndex);
        item.MyArrayIndex = -1;
    }
    removeBuffer.Clear();
}

//データを更新
void DataUpdate()
{
    for (int i = 0; i < ArrayUseLength; i++)
    {
        //毎フレーム更新するデータをArrayに代入
        positionArray[i] = connectorArray[i].GetPosition();
        rotationArray[i] = connectorArray[i].GetRotation();
        rotationSpeedArray[i] = connectorArray[i].GetRotationSpeed();
    }

    //NativeArrayの各要素へのアクセスは遅い
    //Arrayを経由してCopyFromでコピーした方が速い
    positions.CopyFrom(positionArray);
    rotations.CopyFrom(rotationArray);
    rotationSpeeds.CopyFrom(rotationSpeedArray);
}

void JobRun()
{
    //Jobの実行順番を整理
    jobHandle = myParallelForJob.Schedule(ArrayUseLength, JobBatchCount);

    //Jobを実行
    JobHandle.ScheduleBatchedJobs();
    isJobRunning = true;
}

ResultReturn

FixedUpdateのタイミングにJobの完了を待って、Jobの結果をinterfaceで返します。
Jobが完了したらisJobRunningをfalseにします。
isJobRunningの判定でJobの実行前に結果を返さないようにしています。
呼び出し元のFixedUpdateも再掲載しておきます。

JobSystemParallelForTemplate.cs

void FixedUpdate()
{
    if (ArrayUseLength == 0) return;
    if (isJobRunning == false) return;
    jobHandle.Complete();
    isJobRunning = false;

    ResultReturn();//Jobの結果を返す
}

//結果を返す
void ResultReturn()
{
    //結果の反映
    positions.CopyTo(positionArray);
    rotations.CopyTo(rotationArray);

    for (int i = 0; i < ArrayUseLength; i++)
    {
        if (connectorArray[i].Equals(null)) continue;
        connectorArray[i].SetResult(positionArray[i], rotationArray[i]);
    }
}

Jobを使う側のスクリプト

interfaceを継承してパラメータを渡す関数と結果を受け取る関数を実装します。
OnEnableにJobSystemTemplate.Instance?.AddRequest(this)で追加して
OnDisableにJobSystemTemplate.Instance?.RemoveRequest(this)で削除します。
このスクリプトをGameObjectに追加して使います。
もし弾などに使う場合はJobの結果を実装側の都合でRigidbody.MovePositon(position)などに書き換えたりします。

UseJobTemplate.cs

using UnityEngine;
using A_rosuko.JobSystemParallelForTemplate;

public class UseJobTemplate : MonoBehaviour, IJobConnector
{
    Transform myTransform;

    [SerializeField]
    float moveSpeedRange = -0.03f;
    [SerializeField]
    float rotationSpeedRange = 10f;

    Vector3 moveSpeed;
    Quaternion rotationSpeed;

    void Awake()
    {
        myTransform = transform;
        rotationSpeed = Quaternion.Euler(0, Random.value * rotationSpeedRange, 0);
        moveSpeed = new Vector3(0, moveSpeedRange, 0);
    }

    void OnEnable()
    {
        JobSystemParallelForTemplate.Instance?.AddRequest(this);
    }

    void OnDisable()
    {
        JobSystemParallelForTemplate.Instance?.RemoveRequest(this);
    }

    public Vector3 GetPosition()
    {
        return myTransform.position;
    }

    public Quaternion GetRotation()
    {
        return myTransform.rotation;
    }

    public Quaternion GetRotationSpeed()
    {
        return rotationSpeed;
    }

    public Vector3 GetMoveSpeed()
    {
        return moveSpeed;
    }

    public void SetResult(Vector3 position, Quaternion rotation)
    {
        //Jobの結果を反映
        myTransform.SetPositionAndRotation(position, rotation);
    }
    
    public int MyArrayIndex { get; set; } = -1;
}

全スクリプト

JobSystemParallelForTemplate.cs

using System;
using System.Collections;
using System.Collections.Generic;
using UnityEngine;
using Unity.Collections;
using Unity.Jobs;
using Unity.Burst;

namespace A_rosuko.JobSystemParallelForTemplate
{
    //インターフェイスを定義
    public interface IJobConnector
    {
        Vector3 GetPosition();
        Quaternion GetRotation();
        Vector3 GetMoveSpeed();
        Quaternion GetRotationSpeed();
        void SetResult(Vector3 position, Quaternion rotation);
        int MyArrayIndex { get; set; }
    }

    [DefaultExecutionOrder(-20)]//実行タイミングを調整する
    public class JobSystemParallelForTemplate : MonoBehaviour
    {
        //シングルトン
        private static JobSystemParallelForTemplate instance;
        private static bool isDestroy = false;

        //メモリ管理
        const int InitMemoryLength = 256;//初期化と追加でメモリを確保する時の配列サイズ
        const int JobBatchCount = 0;//Jobのバッチ実行時の分割数(0はコア数に自動設定)
        int memoryLength = 0;
        int ArrayUseLength = 0;//配列を使用中のlength

        //Jobの実行
        JobHandle jobHandle;
        bool isJobRunning = false;

        //WaitForFixedUpdateのキャッシュ
        WaitForFixedUpdate waitForFixedUpdate = new WaitForFixedUpdate();

        //追加と削除のBuffer
        List<IJobConnector> addBuffer = new List<IJobConnector>();
        List<IJobConnector> removeBuffer = new List<IJobConnector>();

        //Interfaceの参照の保持
        IJobConnector[] connectorArray = new IJobConnector[InitMemoryLength];

        //NativeArrayにコピーするときに使う普通の配列
        Vector3[] positionArray = new Vector3[InitMemoryLength];
        Quaternion[] rotationArray = new Quaternion[InitMemoryLength];
        Quaternion[] rotationSpeedArray = new Quaternion[InitMemoryLength];

        //Jobの計算に使うNativeArrayの配列
        NativeArray<Vector3> positions;
        NativeArray<Quaternion> rotations;
        NativeArray<Vector3> moveSpeeds;
        NativeArray<Quaternion> rotationSpeeds;

        //Jobを作成
        MyParallelForJob myParallelForJob = new MyParallelForJob();

        //Jobを定義
        [BurstCompile]
        struct MyParallelForJob : IJobParallelFor
        {
            public NativeArray<Vector3> positions;
            public NativeArray<Quaternion> rotations;

            [Unity.Collections.ReadOnly]
            public NativeArray<Vector3> moveSpeeds;
            [Unity.Collections.ReadOnly]
            public NativeArray<Quaternion> rotationSpeeds;

            void IJobParallelFor.Execute(int i)
            {
                positions[i] += moveSpeeds[i];
                rotations[i] *= rotationSpeeds[i];

                //高さが-50m以下であれば100m上に移動させる
                if (positions[i].y < -100.0f / 2.0f)
                {
                    positions[i] = new Vector3(positions[i].x, positions[i].y + 100.0f, positions[i].z);
                }
            }
        }

        private void Awake()
        {
            //メモリを確保
            InitMemory();
        }

        //Jobで使用するメモリを初期化
        void InitMemory()
        {
            this.memoryLength = InitMemoryLength;

            //初期設定でメモリを確保
            positions = new NativeArray<Vector3>(this.memoryLength, Allocator.Persistent);
            rotations = new NativeArray<Quaternion>(this.memoryLength, Allocator.Persistent);
            moveSpeeds = new NativeArray<Vector3>(this.memoryLength, Allocator.Persistent);
            rotationSpeeds = new NativeArray<Quaternion>(this.memoryLength, Allocator.Persistent);

            //Jobの参照を設定
            myParallelForJob.positions = positions;
            myParallelForJob.rotations = rotations;
            myParallelForJob.moveSpeeds = moveSpeeds;
            myParallelForJob.rotationSpeeds = rotationSpeeds;
        }

        void OnDestroy()
        {
            //Jobの完了を待つ
            jobHandle.Complete();
            isJobRunning = false;

            //メモリを解放
            DisposeMemory();
            instance = null;
        }

        //使用メモリの解放
        void DisposeMemory()
        {
            //NativeArrayの開放
            positions.Dispose();
            rotations.Dispose();
            moveSpeeds.Dispose();
            rotationSpeeds.Dispose();

            //参照を持つArrayをnullにする
            connectorArray = null;
        }

        //使用メモリのサイズ変更
        void Resize(int memoryLength)
        {
            this.memoryLength = memoryLength;

            //要素のコピーが必要ない場合はfalseを追加
            ResizeNativeArray(ref positions, memoryLength, false);
            ResizeNativeArray(ref rotations, memoryLength, false);
            ResizeNativeArray(ref moveSpeeds, memoryLength, true);
            ResizeNativeArray(ref rotationSpeeds, memoryLength, false);

            //Jobの参照を設定
            myParallelForJob.positions = positions;
            myParallelForJob.rotations = rotations;
            myParallelForJob.moveSpeeds = moveSpeeds;
            myParallelForJob.rotationSpeeds = rotationSpeeds;

            //配列のサイズを変更
            Array.Resize(ref connectorArray, memoryLength);
            Array.Resize(ref positionArray, memoryLength);
            Array.Resize(ref rotationArray, memoryLength);
            Array.Resize(ref rotationSpeedArray, memoryLength);
        }

        //実行開始
        void OnEnable()
        {
            //コルーチン開始
            StartCoroutine(LateFixedUpdate());
        }

        void OnDisable()
        {
            //コルーチン停止
            StopCoroutine(LateFixedUpdate());

            //jobが実行中だった場合を考慮
            jobHandle.Complete();
            isJobRunning = false;
        }

        //LateFixedUpdateが無いので代わり
        IEnumerator LateFixedUpdate()
        {
            while (true)
            {
                yield return waitForFixedUpdate;
                ArrayUpdate();//追加依頼と削除依頼のバッファを処理する

                if (ArrayUseLength == 0) continue;//要素がないから終了
                DataUpdate();//データを更新する。
                JobRun();//Jobの実行を開始する
            }
        }

        void FixedUpdate()
        {
            if (ArrayUseLength == 0) return;
            if (isJobRunning == false) return;
            jobHandle.Complete();
            isJobRunning = false;

            ResultReturn();//Jobの結果を返す
        }

        //Job利用の追加依頼をバッファに貯めておく
        public void AddRequest(IJobConnector item)
        {
            addBuffer.Add(item);
        }

        //Job利用の削除依頼をバッファに貯めておく
        public void RemoveRequest(IJobConnector item)
        {
            removeBuffer.Add(item);
        }

        //配列を更新
        void ArrayUpdate()
        {
            //connectorArrayがnullになっていた場合は削除(forで逆順処理)
            for (int i = ArrayUseLength - 1; 0 <= i; i--)
            {
                if (connectorArray[i].Equals(null))
                {
                    Remove(i);
                }
            }

            //追加バッファを処理
            foreach (IJobConnector item in addBuffer)
            {
                if (item.Equals(null)) continue;
                Add(item);
            }
            addBuffer.Clear();

            //削除バッファを処理
            foreach (IJobConnector item in removeBuffer)
            {
                if (item.Equals(null)) continue;
                //indexと中身が違っていたらエラー
                if (connectorArray[item.MyArrayIndex].Equals(item) == false) { Debug.LogError("No target in the index"); continue; }
                
                Remove(item.MyArrayIndex);
                item.MyArrayIndex = -1;
            }
            removeBuffer.Clear();
        }

        void Add(IJobConnector item)
        {
            //足りなければメモリサイズを拡張
            if (memoryLength < ArrayUseLength + 1)
            {
                //メモリサイズを+1では無くInitMemoryLengthずつ加算で増やす。
                memoryLength += InitMemoryLength;
                Resize(memoryLength);
            }

            //結果を返すために相手のInterfaceを保持する
            connectorArray[ArrayUseLength] = item;

            //毎フレーム更新しないデータはここで代入しておく。
            moveSpeeds[ArrayUseLength] = item.GetMoveSpeed();

            item.MyArrayIndex = ArrayUseLength;
            ++ArrayUseLength;
        }

        void Remove(int index)
        {
            if (index != -1)
            {
                --ArrayUseLength;

                //参照するデータは配列の最後と入れ替えてからnullにする。
                connectorArray[index] = connectorArray[ArrayUseLength];
                connectorArray[index].MyArrayIndex = index;//入れ替えた側のIndexを更新
                connectorArray[ArrayUseLength] = null;

                //更新しないデータはindexがずれるので配列の最後と入れ替える。
                moveSpeeds[index] = moveSpeeds[ArrayUseLength];
            }
        }

        //データを更新
        void DataUpdate()
        {
            for (int i = 0; i < ArrayUseLength; i++)
            {
                //毎フレーム更新するデータをArrayに代入
                positionArray[i] = connectorArray[i].GetPosition();
                rotationArray[i] = connectorArray[i].GetRotation();
                rotationSpeedArray[i] = connectorArray[i].GetRotationSpeed();
            }

            //NativeArrayの各要素へのアクセスは遅い
            //Arrayを経由してCopyFromでコピーした方が速い
            positions.CopyFrom(positionArray);
            rotations.CopyFrom(rotationArray);
            rotationSpeeds.CopyFrom(rotationSpeedArray);
        }

        void JobRun()
        {
            //Jobの実行順番を整理
            jobHandle = myParallelForJob.Schedule(ArrayUseLength, JobBatchCount);

            //Jobを実行
            JobHandle.ScheduleBatchedJobs();
            isJobRunning = true;
        }

        //結果を返す
        void ResultReturn()
        {
            //結果の反映
            positions.CopyTo(positionArray);
            rotations.CopyTo(rotationArray);

            for (int i = 0; i < ArrayUseLength; i++)
            {
                if (connectorArray[i].Equals(null)) continue;
                connectorArray[i].SetResult(positionArray[i], rotationArray[i]);
            }
        }

        public static JobSystemParallelForTemplate Instance
        {
            get
            {
                if (instance == null)
                {
                    if (isDestroy == false)
                    {
                        instance = FindObjectOfType<JobSystemParallelForTemplate>();
                        if (instance == null)
                        {
                            instance = new GameObject(typeof(JobSystemParallelForTemplate).Name).AddComponent<JobSystemParallelForTemplate>();
                        }
                    }
                }
                return instance;
            }
        }

        void OnApplicationQuit()
        {
            isDestroy = true;
            instance = null;
        }

        //NativeArrayのリサイズ用関数
        void ResizeNativeArray<T>(ref NativeArray<T> nativeArray, int length, bool dataCopy = true) where T : struct
        {
            if (dataCopy == true)
            {
                //データをコピーしてリサイズ
                var temp = new T[nativeArray.Length];
                nativeArray.CopyTo(temp);
                nativeArray.Dispose();
                nativeArray = new NativeArray<T>(length, Allocator.Persistent);
                Array.Resize(ref temp, length);//tempのサイズを変更
                nativeArray.CopyFrom(temp);
            }
            else
            {
                //データをコピーしないでリサイズ
                nativeArray.Dispose();
                nativeArray = new NativeArray<T>(length, Allocator.Persistent);
            }
        }
    }
}

Gistにもおいておきます。
リンクはこちら

補足

intreface == nullは正常に判定されないようなのでintreface.Equals(null)と書きました。
この記事を書いてる途中にRemoveの高速化案を思いついたので入れてみました。
interfaceの実装側に依存する書き方になるのであまり良くありませんが、頻繁に追加と削除を繰り返す場合に有効だと思います。
最初の実装ではArray内を検索して削除していたのですが、下記を変更しています。

メモリへAdd時にオブジェクト側にindexを渡してをキャッシュしておきます。
Remove時にindexも一緒に渡せば対象をArrayから探さずに削除できます。
別のオブジェクトのRemoveで順番の入れ替えが発生した場合はindexを更新します。
※Remove()内のconnectorArray[index].MyArrayIndex = index;をconnectorArray[ArrayUseLength] = null;の前に変更しました。(12/15修正)

まとめ

Jobの内容と実行タイミングを決めれば、あとはメモリ確保の違いだけでほぼ一本道で完成できると思います。
メモリの管理については愚直に１行づつ書いているので、数が多くなるとミスが発生しやすくなります。
構造体でまとめて扱えば楽そうですがNativeArrayのCopyFrom、CopyToのコピーが思ったより速かったので今回の実装にしました。
下記の参考のしゅみぷろ様のサイトではunsafeでポインタでNativeArrayにアクセスする方法も書かれていて参考になります。

さいごに

このスクリプトは実行タイミング、実行条件、実行内容、データ、それらの接続を分けることを意識して使いやすさを優先で書きました。
Transformを更新するだけの場合はIJobParallelForTransformを使った方が倍以上速いですし、データの渡し方も工夫の余地もまだありそうです。
もしかしたらECSを使うとすっきり書けるのかもしれませんが、難しかったのとpreviewだったのでよく分かってません。
良い書き方や改善案があれば教えてください。

ここに書かれている私が書いたスクリプトはCC0としますので好きに使って頂いてかまいません。
JobSystemは本当に高速に実行できるのでもっと気軽に簡単に扱えるようになるといいですね。
少し長くなってしまいましたが、最後まで読んでいただいてありがとうございました。
この記事が皆様のお役に立てればうれしいです。

参考

JobSystemの理解をするのにとても助かりました。
いつもわかりやすい記事をありがとうございます。

しゅみぷろ
SpringBoneのJobSystem化でわかったこと
 ECSとJobSystem 基礎
テラシュールブログ
【Unity】C# Job Systemを自分なりに解説してみる
かめくめ
UnityのC# Job SystemとBurst Compilerを使ってみる

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up