How to convert that piece of code into unity job system

I'm trying to convert the following code into parallel job using unity.

The original code is:

Code (CSharp):

1.     {
2.         if (history_scan_depth > history_capacity)
3.         {
4.             history_scan_depth = history_capacity;
5.         }
6.  
7.         for (int i = 0; i < layer_size; i++)
8.         {
9.             byte b = src[i];
10.             uint avrg = 0;
11.             for (int k = 0; k < history_scan_depth; k++)
12.             {
13.                 avrg += history[k * layer_size + i];
14.  
15.             }
16.             average[i] = (byte)(avrg / history_scan_depth);
17.  
18.             if (Math.Abs(b - average[i]) < data_delta)
19.             {
20.                 dst[i] = b;
21.             }
22.  
23.  
24.             history[layer * layer_size + i] = b;
25.         }
26.  
27.         layer++;
28.         if (layer == history_scan_depth)
29.         {
30.             layer = 0;
31.         }
32.     }

the current solution that I have tried has a lot of exceptions like :System.IndexOutOfRangeException: Index {0} is out of restricted IJobParallelFor range [{1}...{2}] in ReadWriteBuffer IndexOutOfRangeException: Index 3686400 is out of restricted IJobParallelFor range [0...3686399] in ReadWriteBuffer

Code (CSharp):

1. Code (CSharp):
2.  
3.   private const int width = 1280;
4.    private const int height = 720;
5.    private int history_capacity = 3;
6.    private int layer_size = width * height * 4;
7.    private int layer = 0;
8.    public int history_scan_depth = 50;
9.    public int data_delta = 50;
10.    private Texture2D _outputTexture;
11.  
12.    private NativeArray<byte> inputArray;
13.    private NativeArray<byte> outputArray;
14.  
15.    private NativeArray<byte> historyArray;
16.    private NativeArray<byte> averageArray;
17.  
18.    private int _history_scan_depth;
19.    private int _history_capacity;
20.    private int _data_delta;
21.    private int _layer;
22.    private void Awake()
23.    {
24.        historyArray = new NativeArray<byte>(history_capacity * layer_size, Allocator.Persistent);
25.        averageArray = new NativeArray<byte>(layer_size, Allocator.Persistent);
26.        outputArray = new NativeArray<byte>(layer_size, Allocator.Persistent);
27.  
28.    }
29.  
30.    private void OnDestroy()
31.    {
32.        historyArray.Dispose(); // We need to manually dispose our NativeArrays. The arrays retrieved from the Textures are disposed by the textures so we don't care here
33.        averageArray.Dispose();
34.        outputArray.Dispose();
35.  
36.    }
37.  
38.    private void Update()
39.    {
40.  
41.    }
42.  
43.    // [BurstCompile] // <-- For this you need to add the Burst package using the Unity package manager, this attribute is completely optional but it improves performance by a lot (like 10 times faster)
44.    public struct ConvertJobParallel : IJobParallelFor
45.    {
46.        public NativeArray<byte> input;
47.        public NativeArray<byte> output;
48.        public NativeArray<byte> history;
49.        public NativeArray<byte> average;
50.        public int _history_scan_depth;
51.        public int _history_capacity;
52.        public int _layer_size;
53.        public int _data_delta;
54.        public int _layer;
55.  
56.        // This gets called by the job system. index ranges between 0 and the 'arrayLength' specified in the first parameter of Schedule()
57.        public void Execute(int index)
58.        {
59.      
60.        
61.            if (_history_scan_depth > _history_capacity)
62.            {
63.                _history_scan_depth = _history_capacity;
64.            }
65.  
66.            byte b = input[index];
67.            uint avrg = 0;
68.            for (int k = 0; k < _history_scan_depth; k++)
69.            {
70.                avrg += history[k * _layer_size + index];
71.  
72.            }
73.            average[index] = (byte)(avrg / _history_scan_depth);
74.  
75.            if (Math.Abs(b - average[index]) < _data_delta)
76.            {
77.                output[index] = b;
78.            }
79.  
80.  
81.            history[_layer * _layer_size + index] = b;
82.  
83.  
84.            _layer++;
85.            if (_layer == _history_scan_depth)
86.            {
87.                _layer = 0;
88.            }
89.        
90.        }
91.  
92.    }

1. 
   

 

Because you are treating history as a 2D array, you need to add a [NativeDisableParallelForRestriction] attribute. You may also want to add [ReadOnly] attributes to input.

 

@DreamingImLatios thanks for your answer. I added the attribute [NativeDisableParallelForRestriction] still I get exception on that line
IndexOutOfRangeException: Index 11059200 is out of range of '11059200' Length.
avrg += history[k * _layer_size + index];

 

Here is modified code
historyArray = new NativeArray<byte>(history_capacity * layer_size, Allocator.Persistent);
averageArray = new NativeArray<byte>(layer_size, Allocator.Persistent);
outputArray = new NativeArray<byte>(layer_size, Allocator.Persistent);

Code (CSharp):

1.   public struct ConvertJobParallel : IJobParallelFor
2.     {
3.  
4.         [ReadOnly("input_texture")] public NativeArray<byte> input_texture;
5.  
6.         public NativeArray<byte> output;
7.         [NativeDisableParallelForRestriction] public NativeArray<byte> history;
8.         [NativeDisableParallelForRestriction] public NativeArray<byte> average;
9.         public int _history_scan_depth;
10.         public int _history_capacity;
11.         public int _layer_size;
12.         public int _data_delta;
13.         public int _layer;
14.  
15.         // This gets called by the job system. index ranges between 0 and the 'arrayLength' specified in the first parameter of Schedule()
16.         public void Execute(int index)
17.         {
18.  
19.  
20.             if (_history_scan_depth > _history_capacity)
21.             {
22.                 _history_scan_depth = _history_capacity;
23.             }
24.  
25.             byte b = input_texture[index];
26.  
27.  
28.             uint avrg = 0;
29.             for (int k = 0; k < _history_scan_depth; k++)
30.             {
31.                 avrg += history[k * _layer_size + index];
32.  
33.             }
34.             average[index] = (byte)(avrg / _history_scan_depth);
35.  
36.             if (Math.Abs(b - average[index]) < _data_delta)
37.             {
38.                 output[index] = b;
39.             }
40.  
41.             output[index] = b;
42.  
43.             history[_layer * _layer_size + index] = b;
44.  
45.  
46.             _layer++;
47.             if (_layer == _history_scan_depth)
48.             {
49.                 _layer = 0;
50.             }
51.  
52.         }
53.  
54.     }

 

You are indexing one element larger than the last element in your history array. This is a bug in your code. I would put an if statement that checks if the calculated index is greater than or equal to the array length, and if so, Debug.Log the different parameters that lead to that index calculation. Use string interpolation when using Debug.Log in Bursted code.

 

@DreamingImLatios I fixed that problem, the problem right now is that code is very slow and drops the performance so there are no benefits of it

 

Are you missing [BurstCompile]?

 

@DreamingImLatios
Wrote it still very slow, I'm doing a compute shader approach, can you help me with that please ?

 

Shader initialization:

Code (CSharp):

1.     handleDepthFilter = shader.FindKernel("CSMain");
2.  
3.  
4.         if (handleDepthFilter < 0)
5.         {
6.             Debug.Log("Initialization failed.");
7.             enabled = false;
8.             return;
9.         }
10.  
11.         history = new int[history_capacity * layer_size];
12.  
13.         average = new int[layer_size];
14.  
15.  
16.         shader.SetInt("history_scan_depth", history_scan_depth);
17.         shader.SetInt("data_delta", data_delta);
18.         shader.SetInt("history_capacity", history_capacity);
19.         shader.SetInt("layer_size", layer_size);
20.         shader.SetInt("layer", layer);
21.  
22.         ComputeBuffer history_buffer = new ComputeBuffer(history_capacity * layer_size, sizeof(int));
23.         ComputeBuffer average_buffer = new ComputeBuffer(layer_size, sizeof(int));
24.  
25.         history_buffer.SetData(history);
26.         average_buffer.SetData(average);
27.         shader.SetBuffer(handleDepthFilter, "history_buffer", history_buffer);
28.         shader.SetBuffer(handleDepthFilter, "average_buffer", average_buffer);
29.  
30.         _outputTexture = new RenderTexture(width, height, 0, RenderTextureFormat.ARGB32);
31.         _outputTexture.enableRandomWrite = true;
32.         _outputTexture.Create();
33.        

Shader dispatching:

Code (CSharp):

1.  private void ImageUpdated(ref Texture2D zedTextureDepth)
2.     {
3.  
4.        Texture2D copyDepthRW = duplicateTexture(zedTextureDepth);
5.  
6.  
7.         shader.SetTexture(handleDepthFilter, "Texture", copyDepthRW);
8.         shader.SetTexture(handleDepthFilter, "Result", _outputTexture);
9.  
10.         shader.Dispatch(handleDepthFilter, (width) / 8,
11.            (height) / 8, 1);

}



Shader Code:

Code (CSharp):

1. #pragma kernel CSMain
2.  
3. // Create a RenderTexture with enableRandomWrite flag and set it
4. // with cs.SetTexture
5. RWTexture2D<float4> Result;
6. Texture2D<float4> Texture;
7. RWStructuredBuffer<int> history_buffer;
8. RWStructuredBuffer<int> average_buffer;
9. int history_scan_depth;
10. int history_capacity;
11. int data_delta;
12. int layer;
13. int layer_size;
14.  
15. [numthreads(8,8,1)]
16. void CSMain (uint3 id : SV_DispatchThreadID)
17. {
18.     // TODO: insert actual code here!
19.     if (history_scan_depth > history_capacity)
20.     {
21.         history_scan_depth = history_capacity;
22.     }
23.  
24.     float4 b = Texture[id.xy];
25.    
26.  
27.     uint avrg = 0;
28.     for (int k = 0; k < history_scan_depth; k++)
29.     {
30.         avrg += history_buffer[k * layer_size + id.xy];
31.  
32.     }
33.     average_buffer[id] = (byte)(avrg / history_scan_depth);
34.  
35.     if (abs(b - average[id.xy) < data_delta)
36.     {
37.         Result[id] = b;
38.     }
39.  
40.  
41.     history_buffer[layer * layer_size + .xy] = b;
42.  
43.  
44.     _layer++;
45.     if (_layer == history_scan_depth)
46.     {
47.         _layer = 0;
48.     }
49. }
50.  

 

This isn't a compute shader forum. All you have told me is that things are slow. I have no idea if the original non-job algorithm was equally slow. And I have no idea if you are actually running Burst (sometimes Burst fails to compile a job so it runs without Burst). Please include some timeline profiler screenshots.

 

@DreamingImLatios

The main problem right now with parallel code is getting the texture bytes that I want to process. using GetRawTextureData WITHOUT doing anything on the Execute function Drops the frame. now it's 10FPS.

The other problem is how would I convert that serial code into parallel code, basically the serial code loops over layer_size which is texture resoultion 1280*720, how would I let the theads divides into that texture pixels do the average and keep it in history.

Code (CSharp):

1. private void ImageUpdated(ref Texture2D zedTextureDepth)
2.     {
3.    
4.         inputArray = zedTextureDepth.GetRawTextureData<byte>();
5.    }
6.  
7.  
8.   private void Update()
9.     {
10.         var job = new ConvertJobParallel()
11.         {
12.             input_texture = inputArray,
13.             output = outputArray,
14.             history = historyArray,
15.             average = averageArray,
16.             _history_scan_depth = history_scan_depth,
17.             _history_capacity = history_capacity,
18.             _layer_size = layer_size,
19.             _data_delta = data_delta,
20.             _layer = layer
21.         };
22.            JobHandle handle1 = job.Schedule(inputArray.Length, 100); // The 100 here is a magic number, it is basically a number that says how many items (Color32 in our example) are handled in one 'batch' where each batch can run async on any thread
23.            JobHandle.ScheduleBatchedJobs(); // Start running (all) our previously scheduled jobs
24.           handle1.Complete(); // Wait on the main thread fot this job to complete
25.     }
26.  
27.  
28. [BurstCompile] // <-- For this you need to add the Burst package using the Unity package manager, this attribute is completely optional but it improves performance by a lot (like 10 times faster)
29.     public struct ConvertJobParallel : IJobParallelFor
30.     {
31.         [ReadOnly("input_texture")] public NativeArray<byte> input_texture;
32.         public NativeArray<byte> output;
33.         [NativeDisableParallelForRestriction] public NativeArray<byte> history;
34.         [NativeDisableParallelForRestriction] public NativeArray<byte> average;
35.         public int _history_scan_depth;
36.         public int _history_capacity;
37.         public int _layer_size;
38.         public int _data_delta;
39.         public int _layer;
40.         // This gets called by the job system. index ranges between 0 and the 'arrayLength' specified in the first parameter of Schedule()
41.         public void Execute(int index)
42.         {
43.             if (_history_scan_depth > _history_capacity)
44.             {
45.                 _history_scan_depth = _history_capacity;
46.             }
47.             byte b = input_texture[index];
48.             output[index] = b;
49.             /*
50.          
51.             uint avrg = 0;
52.             for (int k = 0; k < _history_scan_depth; k++)
53.             {
54.                 avrg += history[k * _layer_size + index];
55.             }
56.             average[index] = (byte)(avrg / _history_scan_depth);
57.             if (Abs(b - average[index]) < _data_delta)
58.             {
59.                 output[index] = b;
60.             }
61.             history[_layer * _layer_size + index] = b;
62.             _layer++;
63.             if (_layer == _history_scan_depth)
64.             {
65.                 _layer = 0;
66.             }
67.            */
68.         }
69.     }

 

Oh. This is a RenderTexture? In that case your slowness is due to reading back the texture from the GPU synchronously. Either you need to keep that texture on the GPU and use compute, or you need to use asynchronous readback APIs.

You are going to have to be more specific about what your requirements are, what you are trying to do, and how each part of your existing code accomplishes that if you wish for me to help you, as my mind is just not clicking with your code as is today.