Using threading

I've haven't been keeping up with the multithreading support with Unity. I'm so use to the approach I use below because previously Unity didn't have a strong thread codebase. But now I've heard that coroutines are more reliable and they even included this Job system I only know little about that seems to mimic .net's thread and task system. I would really like some direction on where to start my learning of this new system. There seems to be a lot of topics but I'm not sure which to focus on as the standard for the engine.

Code (CSharp):

1. public class ThreadedJob
2. {
3.     private bool m_IsDone = false;
4.     private object m_Handle = new object();
5.     protected System.Threading.Thread m_Thread = null;
6.     protected bool stopThread;
7.     public bool IsDone
8.     {
9.         get
10.         {
11.             bool tmp;
12.             lock (m_Handle)
13.             {
14.                 tmp = m_IsDone;
15.             }
16.             return tmp;
17.         }
18.         set
19.         {
20.             lock (m_Handle)
21.             {
22.                 m_IsDone = value;
23.             }
24.         }
25.     }
26.  
27.  
28.     public virtual void Start()
29.     {
30.         stopThread = false;
31.         m_Thread = new System.Threading.Thread(Run);
32.         m_Thread.IsBackground = true;
33.         m_Thread.Start();
34.     }
35.  
36.     public virtual void Abort()
37.     {
38.         if(m_Thread != null)
39.         m_Thread.Abort();
40.     }
41.  
42.     public virtual void StopThread()
43.     {
44.         stopThread = true;
45.     }
46.  
47.     protected virtual void ThreadFunction() { }
48.  
49.     protected virtual void OnFinished() { }
50.  
51.     public virtual bool Update()
52.     {
53.         if (IsDone)
54.         {
55.             OnFinished();
56.             return true;
57.         }
58.         return false;
59.     }
60.     protected void Run()
61.     {
62.         ThreadFunction();
63.         IsDone = true;
64.     }
65. }

 

Steps to success to learn a system:

- decide first if you actually NEED it (this is highly dubious for threading under Unity, but there are a few uses where it might provide a benefit; mostly threading under Unity simply gives no benefit unless you actually know what you're doing)

- contrive an example context that lets you learn how it works, OR

- be lucky to find someone who wants to pay you to learn how it works.

Either way, most of the same "you must do 99.999% of things on the main thread" still applies, and likely always will apply.

Alternate technologies available under Unity include the Jobs/ECS/DOTS stuff, but all of the learning steps above apply just as much as anything you want to learn.

EDIT: see what @R1PFake wrote below... I totally forgot about that option!

 

Like the other comment says it depends on the context, but I would like to add one thing and suggest to use tasks instead of "raw" threads if you want to go the ".NET route" combine it with async / await.

Unity implements a custom SynchronizationContext to better support await / tasks, becuse the "callback" code after await will be called on the main thread (instead of a pool thread like the default SynchronizationContext would do) so it's "easier" to use.

And some of the newer Unity system (like Addressable) also support await / tasks "by default"

 

I've been reading since I made this thread and tasks seem to be the recommended approach so far.

@Kurt-Dekker I think I remember seeing you in that one Minecraft thread, but essentially I'm rewriting my threading system to utilize the updated commands for generating mesh data for procedural content and dealing with math operations. Having tasks take care of that then using a callback function to deal with the resulting data seems like the best pattern. Now it's a matter of whether I could just repurpose my ThreadedJob custom class or make something new to make use of the pattern better. I can figure that out on my own but all advice is welcomed.

 

what you want sounds possible in jobs system or threads, have done similar things with a dedicated thread for background processing of procedural stuff, then made a TaskFactory and TaskScheduler to allow me to create Tasks that can be awaited for work happening on this thread. Took a similar approach to pathfinding as well.

 

That sounds like what I want to do.

The design I'm looking at is a queue based task system. So for procedural for instance, to deal with stuff like lighting and mesh generation, a background thread will pick it up from the queue, work on it, then pass it back to be processed by the main Unity thread for mesh applications with vertex lighting and the new mesh data that was created (vertices, color, etc).

 

yeah if you need help later i can show what that implementation looks like, in my case it was a single thread, with a queue of jobs but you could do a similar idea with a thread pool as well if you need more then 1 job processing at a time

 

Oh forgot to add, you do open yourself up to a new class of weird looking errors by doing this that you will need to check for.

Say you await a task on your thread from a method on a MonoBehaviour, then that object gets destroyed while that call is being awaited with work happening in a other thread. The synchronization context will still do its job and bring you back to the method you started in, but accessing nearly everything on that object will result in a missing reference exception since it got destroyed on the C++ side of the engine. So i found in cases where this is possible i have often needed to do a `if (this == null) return;` to ensure i just exit early if that situation happens. A other thing to handle this is with CancellationTokens and making sure your task properly handles Cancellation. Then in the MB that called and awaited the task you can have OnDestroy invoke cancellation.

 

Yeah that makes total sense. Seems like an error I would have been stomped on if I didn't know that single instance.

 

Update:

I managed to implement a Task Manager that does exactly what I want. I did some debugging to take care of the bugs on the Unity side of things, but quite frankly, this baby really works out quite nicely. I managed to just offload the mesh building and calculations to the separate thread, and then queue the ready objects. The background thread has a small delay. The main queue thread has a timeout to keep the main thread from being blocked for forever if there's lots of work to do. Basically a throttle system to keep things rolling smoothly.

I also did not realize how hard it was to debug things in a task. So I had to use the Unity attach debugger to find out that I was accidentally putting a Unity api call into the threaded action. That was my bad, but it was a easy fix.

Things I still need to do is implement the queue sorter task that will basically keep any chunks waiting in the background queue sorted by distance to keep the closest to the front. Though with my computer, the queue is always emptying too fast so maybe I don't have to worry about it.

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using System.Threading;
4. using System.Threading.Tasks;
5. using System;
6. using System.Diagnostics;
7. using UnityEngine;
8.  
9. public enum TaskPriority
10. {
11.     IMMEDIATE,
12.     LOW,
13.     HIGH
14. }
15.  
16. public static class TaskManager
17. {
18.     public static readonly LinkedList<Action> backgroundTaskQueue = new LinkedList<Action>();
19.     public static readonly LinkedList<Action> mainThreadTaskQueue = new LinkedList<Action>();
20.     static long backgroundDelay_ms = 10;
21.     static long queueSorterDelay_ms = 2000;
22.     static long mainTimeout_ms = 10;
23.  
24.     public static Stopwatch mainStartTime = new Stopwatch();
25.     static Stopwatch bgThreadSleep = new Stopwatch();
26.     public static bool isActive;
27.  
28.     public static Task mainTask;
29.     public static Task bgTask;
30.  
31.     static CancellationTokenSource mainCancelTokenSource = new CancellationTokenSource();
32.     static CancellationToken mainCancelToken = mainCancelTokenSource.Token;
33.  
34.  
35.     public static void EnqueueTask_BG(Action newAction, TaskPriority priority)
36.     {
37.         lock (backgroundTaskQueue)
38.         {
39.             switch (priority)
40.             {
41.                 case TaskPriority.LOW:
42.                     backgroundTaskQueue.AddLast(newAction);
43.                     break;
44.                 case TaskPriority.HIGH:
45.                     backgroundTaskQueue.AddFirst(newAction);
46.                     break;
47.                 case TaskPriority.IMMEDIATE:
48.                     Task.Run(newAction);
49.                     break;
50.             }
51.         }
52.     }
53.  
54.     public static void EnqueueTask_Main(Action mainAction, TaskPriority priority)
55.     {
56.         switch (priority)
57.         {
58.             case TaskPriority.LOW:
59.                 mainThreadTaskQueue.AddLast(mainAction);
60.                 break;
61.             case TaskPriority.HIGH:
62.                 mainThreadTaskQueue.AddFirst(mainAction);
63.                 break;
64.             case TaskPriority.IMMEDIATE:
65.                 Task.Run(mainAction);
66.                 break;
67.         }
68.  
69.     }
70.  
71.     public static void DoUpdate()
72.     {
73.  
74.     }
75.  
76.     static async void GetNextBackgroundTask()
77.     {
78.         while (isActive && !mainCancelToken.IsCancellationRequested)
79.         {
80.             if ((bgThreadSleep.IsRunning && bgThreadSleep.ElapsedMilliseconds > backgroundDelay_ms) ||
81.                 !bgThreadSleep.IsRunning)
82.             {
83.                 bgThreadSleep.Stop();
84.                 bgThreadSleep.Reset();
85.  
86.                 if (backgroundTaskQueue.Count > 0)
87.                 {
88.                     lock (backgroundTaskQueue)
89.                     {
90.                         Task.Run(backgroundTaskQueue.First.Value).ContinueWith((t) =>
91.                         {
92.                             if (t.IsFaulted) UnityEngine.Debug.LogError($"Message: {mainTask.Exception?.Message}\n Stacktrace {mainTask.Exception?.StackTrace.ToString()}");
93.                         });
94.                         backgroundTaskQueue.RemoveFirst();
95.                     }
96.                 }
97.                 else
98.                 {
99.                     bgThreadSleep.Start();
100.                 }
101.             }
102.  
103.          
104.         }
105.     }
106.  
107.     public static void SetMainTaskFinished()
108.     {
109.         lock (mainStartTime)
110.         {
111.             mainStartTime.Stop();
112.         }
113.     }
114.  
115.     public static void DisableTaskManager()
116.     {
117.         isActive = false;
118.         mainCancelTokenSource.Cancel();
119.     }
120.  
121.     public static void EnableTaskManager()
122.     {
123.         isActive = true;
124.      
125.         bgTask = Task.Run(GetNextBackgroundTask);
126.     }
127.  
128.     public static IEnumerator GetNextMainTask()
129.     {
130.         while (isActive)
131.         {        
132.             if (mainStartTime.Elapsed.TotalMilliseconds < mainTimeout_ms)
133.             {
134.                 if (mainThreadTaskQueue.Count > 0)
135.                 {
136.  
137.                     mainThreadTaskQueue.First.Value.Invoke();
138.                     mainThreadTaskQueue.RemoveFirst();
139.                     mainStartTime.Start();
140.                  
141.                 }
142.                 else
143.                 {
144.                     mainStartTime.Reset();
145.                     yield return new WaitForSeconds(mainTimeout_ms /1000);
146.                 }
147.             }
148.             else
149.             {
150.                 mainStartTime.Reset();
151.                 yield return new WaitForSeconds(mainTimeout_ms / 1000);
152.             }
153.         }
154.      
155.     }
156.  
157. }
158.  

 

So i stripped out the business logic, and just show how i setup my TaskFactories and Schedulers and what not, and in this case its from a custom pathfinding solution i needed to implement.

Code (CSharp):

1. using System.Collections.Generic;
2. using UnityEngine;
3. using System;
4. using System.Collections.Concurrent;
5. using System.Threading;
6. using System.Threading.Tasks;
7.  
8. public class Pathfinding : Singleton<Pathfinding> {
9.     private TaskFactory<Node[]> _factory;
10.  
11.  
12.     private void Awake() {
13.         _factory = new TaskFactory<Node[]>(new PathfindingTaskScheduler());
14.     }
15.  
16.     private void OnDestroy() {
17.         if (_factory.Scheduler is IDisposable scheduler) {
18.             scheduler.Dispose();
19.         }
20.     }
21.  
22.     public Task<Node[]> FindPath(Vector3 seekerPosition, Vector3 targetPosition, CancellationToken cancellationToken = new CancellationToken()) {
23.         return _factory.StartNew(() => {
24.             if (cancellationToken.IsCancellationRequested) {
25.                 throw new TaskCanceledException();
26.             }
27.  
28.             // DO the threaed logic here
29.  
30.         }, cancellationToken);
31.     }
32. }
33.  
34. internal sealed class PathfindingTaskScheduler : TaskScheduler, IDisposable {
35.     private readonly BlockingCollection<Task> _tasksCollection = new BlockingCollection<Task>(new ConcurrentQueue<Task>());
36.     private readonly Thread _thread;
37.     private bool _disposed;
38.  
39.     public PathfindingTaskScheduler() {
40.         _thread = new Thread(Run) {Name = "Pathfinding", IsBackground = true};
41.         if (!_thread.IsAlive) {
42.             _thread.Start();
43.         }
44.     }
45.  
46.     private void Run() {
47.         foreach (var task in _tasksCollection.GetConsumingEnumerable()) {
48.             TryExecuteTask(task);
49.         }
50.         _tasksCollection.Dispose();
51.     }
52.  
53.     protected override IEnumerable<Task> GetScheduledTasks() {
54.         return _tasksCollection;
55.     }
56.  
57.     protected override void QueueTask(Task task) {
58.         if (_disposed) return;
59.         _tasksCollection.Add(task);
60.     }
61.  
62.     protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) {
63.         return Thread.CurrentThread == _thread && TryExecuteTask(task);
64.     }
65.  
66.     public void Dispose() {
67.         if (_disposed) return;
68.         _disposed = true;
69.         _tasksCollection.CompleteAdding();
70.     }
71. }
72.  
73.  

Then usage wise i would just do something like

Code (CSharp):

1. using (PathfindingTcs = new CancellationTokenSource()) {
2.     try {
3.         var navPath = await Pathfinding.Instance.FindPath(startingNode.WorldPosition, DestinationNode.WorldPosition, PathfindingTcs.Token);
4.         // do stuff with the found path
5.     }
6.     catch (TaskCanceledException) {
7.         // cleanup for a cancelled path
8.         return
9.     }
10. }
11.  

PathfindingTcs is a field, and i can call cancel on the token source if forget path is called on my object or if it gets destroyed. the using statement just makes sure its disposed of when no longer needed

 

Wow that's way more complicated than what I did. But I'm still fairly new to the Task system and cancellation process. I can see the dispose process is heavily involved, which I guess I should have realized due to the way threading works. I just use a standard bool that is switched on and off, with OnApplicationQuit taking care of disabling it to stop all processes. However, I have a feeling this isnt reliable due to how fast things may happen. What if I disable and turn the bool condition to false, but the task never gets to realize that? Then again, I took the advice of doing the (this == null) check to my Action business logic, so I think I'm fine there.

 

yeah once its setup, it works pretty good, since even if i had more then 1 type of thing i wanted to process on that thread, its as simple as calling _factory.StartNew giving it different logic to work with for your use case the cancelation stuff may not matter, but it was important for my.

What i am doing is not all that dissimilar to yours, i am just using some of the types .net gives me for creating tasks to use as handles to work being done on the thread, and to support await

 

Do you see any pros or cons from doing it that way or is my way still fine and dandy?

 

well think your is a little strange that you have a background and main task, generally the Task handle for work happening on the thread, and anything you do that happens on the other thread returns a Task so you can either await it, or set a callback to execute when done. Also nice thing with how i am approaching it is that its returning data via the task once awaited.

 

Mostly cause in my case, I have thousands of chunk objects to apply meshes to. This way I can manage a throttle to make sure it doesn't hang up the main thread too much. Only reason I'm doing it really.