Well, my intention here is to allow users to execute Unity methods/properties in other threads by using this Dispatcher implementation:
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using System.Threading.Tasks;
using UnityEngine;
namespace DepotToolkit.Utils
{
/// <summary>
/// A thread-safe class which holds a queue with actions to execute on the next Update() method. It can be used to make calls to the main thread for
/// things such as UI Manipulation in Unity. It was developed for use in combination with the Firebase Unity plugin, which uses separate threads for event handling
/// </summary>
public class UnityDispatcher : MonoBehaviour
// : MonoSingleton<UnityDispatcher>
{
// TODO: Add timeout from https://gist.github.com/phizaz/f67f1ba7d25d30d8972c13b7b6a29319
private static readonly ConcurrentQueue<Action> _executionQueue = new ConcurrentQueue<Action>();
public static ObservableCollection<Exception> Exceptions { get; } = new ObservableCollection<Exception>();
// TODO
//private static readonly Queue<Action<object>> _executionQueue1 = new Queue<Action<object>>();
//private static readonly Queue<Action<object, object>> _executionQueue2 = new Queue<Action<object, object>>();
private void Exceptions_OnCollectionChanged(object sender, NotifyCollectionChangedEventArgs e)
{
foreach (var newItem in e.NewItems)
{
var item = newItem as Exception;
Debug.LogError(item);
}
}
public void Update()
{
while (_executionQueue.Count > 0)
{
if (_executionQueue.TryDequeue(out var act))
act.Invoke();
}
}
/// <summary>
/// Locks the queue and adds the IEnumerator to the queue
/// </summary>
/// <param name="action">IEnumerator function that will be executed from the main thread.</param>
public void Enqueue(Action action)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(ActionWrapper(action));
});
}
public void Enqueue(EnumeratorJobWrapper job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
public void Enqueue(JobWrapper job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
public void Enqueue(JobWrapper<object> job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
public void Enqueue<T>(JobWrapper<T> job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
public void Enqueue<T1, TTResult>(JobWrapper<T1, TTResult> job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
public void Enqueue<T>(JobEnumeratorWrapper<T> job)
{
_executionQueue.Enqueue(() =>
{
StartCoroutine(JobWrapper(job));
});
}
/// <summary>
/// Locks the queue and adds the Action to the queue
/// </summary>
/// <param name="action">function that will be executed from the main thread.</param>
public void Invoke(Action action)
{
Enqueue(action);
}
// Alias for Enqueue
public void Invoke(EnumeratorJobWrapper job)
{
Enqueue(job);
}
public void Invoke(JobWrapper job)
{
Enqueue(job);
}
public void Invoke(JobWrapper<object> job)
{
Enqueue(job);
}
public void Invoke<T>(JobWrapper<T> job)
{
Enqueue(job);
}
public void Invoke<T1, TResult>(JobWrapper<T1, TResult> job)
{
Enqueue(job);
}
/// <summary>
/// Locks the queue and adds the Action to the queue, returning a Task which is completed when the action completes
/// </summary>
/// <param name="action">function that will be executed from the main thread.</param>
/// <returns>A Task that can be awaited until the action completes</returns>
public Task EnqueueAsync(Action action)
{
var tcs = new TaskCompletionSource<bool>();
void WrappedAction()
{
try
{
action();
tcs.TrySetResult(true);
}
catch (Exception ex)
{
tcs.TrySetException(ex);
}
}
Enqueue(WrappedAction);
return tcs.Task;
}
private IEnumerator ActionWrapper(Action a)
{
a();
yield return null;
}
private IEnumerator JobWrapper(EnumeratorJobWrapper wrap)
{
var enumerator = wrap.Callback;
while (enumerator.MoveNext())
{
object obj = null;
bool isException = false;
try
{
obj = enumerator.Current;
}
catch (Exception ex)
{
Debug.LogException(ex);
isException = true;
}
if (!isException)
yield return obj;
else
yield break;
}
wrap.Finished.Set();
}
private IEnumerator JobWrapper(JobWrapper wrap)
{
try
{
wrap.Callback();
}
catch (Exception ex)
{
Debug.LogException(ex);
//Exceptions.Add(ex);
}
yield return null;
wrap.Finished.Set();
}
private IEnumerator JobWrapper(JobWrapper<object> wrap)
{
object result = null;
try
{
wrap.Job.Invoke(out result);
}
catch (Exception ex)
{
Debug.LogException(ex);
//Exceptions.Add(ex);
}
wrap.Result = result;
yield return null;
wrap.Finished.Set();
}
private IEnumerator JobWrapper<T>(JobWrapper<T> wrap)
{
T result = default;
try
{
wrap.Job.Invoke(out result);
}
catch (Exception ex)
{
Debug.LogException(ex);
//Exceptions.Add(ex);
}
wrap.Result = result;
yield return null;
wrap.Finished.Set();
}
private IEnumerator JobWrapper<T1, TResult>(JobWrapper<T1, TResult> wrap)
{
TResult result = default;
try
{
wrap.Job.Invoke(wrap.Value, out result);
}
catch (Exception ex)
{
Debug.LogException(ex);
//Exceptions.Add(ex);
}
wrap.Result = result;
yield return null;
wrap.Finished.Set();
}
private IEnumerator JobWrapper<T>(JobEnumeratorWrapper<T> wrap)
{
var enumerator = wrap.Job.Invoke(out var result);
while (enumerator.MoveNext()) yield return null;
wrap.Result = result;
wrap.Finished.Set();
}
private static UnityDispatcher _instance;
public static UnityDispatcher Instance
{
get
{
if (_instance == null)
{
try
{
// Search for existing instance.
_instance = (UnityDispatcher)FindObjectOfType(typeof(UnityDispatcher));
// Create new instance if one doesn't already exist.
if (_instance == null)
{
// Need to create a new GameObject to attach the singleton to.
var singletonObject = new GameObject();
_instance = singletonObject.AddComponent<UnityDispatcher>();
singletonObject.name = $"{nameof(UnityDispatcher)} (Singleton)";
//((MonoSingleton<T>)_instance).IsStarted = false;
// Make instance persistent.
if (Application.isPlaying) DontDestroyOnLoad(singletonObject);
}
}
catch
{
// This is not thread safe
throw new Exception(
"UnityMainThreadDispatcher could not find the UnityMainThreadDispatcher object. Please ensure you have added the MainThreadExecutor Prefab to your scene.");
}
}
return _instance;
}
}
private void Awake()
{
Exceptions.CollectionChanged += Exceptions_OnCollectionChanged;
if (_instance == null)
{
_instance = this;
DontDestroyOnLoad(this.gameObject);
}
}
private void OnDestroy()
{
_instance = null;
}
}
}
Where the JobWrapper class is this:
using System;
using System.Collections;
using System.Threading;
namespace DepotToolkit.Utils
{
public class EnumeratorJobWrapper
{
public IEnumerator Callback { get; }
public ManualResetEvent Finished { get; }
private EnumeratorJobWrapper()
{
}
public EnumeratorJobWrapper(IEnumerator callback, ManualResetEvent finished)
{
Callback = callback;
Finished = finished;
}
}
public class JobWrapper
{
public Action Callback { get; }
public ManualResetEvent Finished { get; }
private JobWrapper()
{
}
public JobWrapper(Action callback, ManualResetEvent finished)
{
Callback = callback;
Finished = finished;
}
}
public class JobWrapper<T>
{
public delegate void DelegateJob(out T result);
public DelegateJob Job { get; }
public ManualResetEvent Finished { get; }
public T Result { get; set; }
private JobWrapper()
{
}
public JobWrapper(DelegateJob job, ManualResetEvent finished)
{
Job = job;
Finished = finished;
}
}
public class JobWrapper<T1, TResult>
{
public delegate void DelegateJob(T1 t1, out TResult result);
public DelegateJob Job { get; }
public ManualResetEvent Finished { get; }
public T1 Value { get; }
public TResult Result { get; set; }
private JobWrapper()
{
}
public JobWrapper(T1 value, DelegateJob job, ManualResetEvent finished)
{
Value = value;
Job = job;
Finished = finished;
}
}
}
An example of what I’m doing:
using System;
using System.Threading;
using DepotToolkit.Utils;
using UnityEngine;
using UnityEngine.Core;
using Object = UnityEngine.Object;
namespace DepotToolkit.CommonCode
{
public static class ThreadSafeUtils
{
internal class ThreadChecker : MonoSingleton<ThreadChecker>
{
public Thread UnityThread { get; private set; }
private void Awake()
{
UnityThread = Thread.CurrentThread;
}
}
public static bool IsMainThread => Thread.CurrentThread.Equals(ThreadChecker.Instance.UnityThread);
public static T() ThreadSafe_GetComponentsInChildren<T>(this GameObject obj)
{
if (IsMainThread)
return obj.GetComponentsInChildren<T>();
var mre = new ManualResetEvent(false);
var job = new JobWrapper<GameObject, T()>(obj, Impl_GetComponentsInChildren, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
}
private static void Impl_GetComponentsInChildren<T>(GameObject obj, out T() result)
{
result = obj.GetComponentsInChildren<T>();
}
// Object.Instantiate
public static GameObject Instantiate(GameObject obj)
{
if (IsMainThread)
return Object.Instantiate(obj);
var mre = new ManualResetEvent(false);
var job = new JobWrapper<GameObject, GameObject>(obj, Impl_Instantiate, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
}
private static void Impl_Instantiate(GameObject obj, out GameObject result)
{
result = Object.Instantiate(obj);
}
}
}
As you can see I’m checking If I’m in the current thread by creating a MonoSingleton instance of a MonoBehaviour:
using System;
using System.Threading;
using DepotToolkit.Utils;
using uzLib.Lite.ExternalCode.Extensions;
using uzSourceToolkit.ThirdParty.uSrcTools.Extensions;
namespace UnityEngine.Core
{
#if UNITY_2020 || UNITY_2019 || UNITY_2018 || UNITY_2017 || UNITY_5
/// <summary>
/// Inherit from this base class to create a singleton.
/// e.g. public class MyClassName : Singleton<MyClassName> {}
/// </summary>
public class MonoSingleton<T> : MonoBehaviour // , IStarted
where T : MonoBehaviour
{
// Check to see if we're about to be destroyed.
private static bool m_ShuttingDown;
private static readonly object m_Lock = new object();
protected static T m_Instance;
/// <summary>
/// Access singleton instance through this propriety.
/// </summary>
public static T Instance
{
get
{
if (m_ShuttingDown)
{
// FIX: Shutting down takes affect after playing a scene, this makes (ie) SteamWorkshopWrapper unavailable on Editor
// (check for ExecuteInEditrMode attribute, if available, shutthing down shuld not be performed)
// ReSharper disable once ConditionIsAlwaysTrueOrFalse
// if (MonoSingletonSettings.ShowWarning)
//#pragma warning disable 162
// // ReSharper disable once HeuristicUnreachableCode
// Debug.LogWarning("(Singleton) Instance '" + typeof(T) +
// "' already destroyed. Returning null.");
//#pragma warning restore 162
return null;
}
lock (m_Lock)
{
if (m_Instance == null)
{
// Search for existing instance.
try
{
m_Instance = (T)FindObjectOfType(typeof(T));
}
catch (UnityException)
{
// Call with the Dispatcher...
var mre = new ManualResetEvent(false);
var job = new JobWrapper<T>(Impl_GetInstance, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
m_Instance = job.Result;
}
// Create new instance if one doesn't already exist.
if (m_Instance == null)
{
// Need to create a new GameObject to attach the singleton to.
try
{
var singletonObject = new GameObject();
m_Instance = singletonObject.AddComponent<T>();
singletonObject.name = typeof(T).Name + " (Singleton)";
//((MonoSingleton<T>)m_Instance).IsStarted = false;
// Make instance persistent.
if (IsPlaying()) DoAction(() => DontDestroyOnLoad(singletonObject)); // TODO
}
catch (UnityException)
{
var singletonObject = CreateGameObject();
m_Instance = AddComponent<T>(singletonObject);
DoAction(() => singletonObject.name = typeof(T).Name + " (Singleton)");
}
}
}
return m_Instance;
}
}
protected set => m_Instance = value;
}
private static void Impl_GetInstance(out T result)
{
result = (T)FindObjectOfType(typeof(T));
}
private static void Impl_CreateInstance(out GameObject obj)
{
obj = new GameObject();
}
private static void Impl_IsPlaying(out bool result)
{
result = Application.isPlaying;
}
private static void Impl_AddComponent<TMethod>(GameObject obj, out TMethod result)
where TMethod : Component
{
result = obj.AddComponent<TMethod>();
}
private static GameObject CreateGameObject()
{
var mre = new ManualResetEvent(false);
var job = new JobWrapper<GameObject>(Impl_CreateInstance, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
}
private static TMethod AddComponent<TMethod>(GameObject gameObject)
where TMethod : Component
{
var mre = new ManualResetEvent(false);
var job = new JobWrapper<GameObject, TMethod>(gameObject, Impl_AddComponent, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
}
private static void DoAction(Action action)
{
var mre = new ManualResetEvent(false);
var job = new JobWrapper(action, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
}
private static bool IsPlaying()
{
var mre = new ManualResetEvent(false);
var job = new JobWrapper<bool>(Impl_IsPlaying, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
}
public bool ExecuteInEditMode => GetType().IsExecutingInEditMode();
public bool IsStarted { get; set; }
public static T Create()
{
var go = new GameObject(typeof(T).Name);
return go.GetOrAddComponent<T>();
}
private void OnApplicationQuit()
{
if (!ExecuteInEditMode) m_ShuttingDown = true;
}
private void OnDestroy()
{
if (!ExecuteInEditMode) m_ShuttingDown = true;
}
}
#endif
}
As you can see, the MonoSingleton class is already protected from external thread calls by using try/catch blocks with UnityExceptions.
My main three concerns about my implementation are the following:
- I’m not able to figure out how to make easier the implementation of the
JobWrapper, as you can see every time I need to create a thread-safe method I use the following four lines that are repeated lots and lots of times:
var mre = new ManualResetEvent(false);
var job = new JobWrapper<GameObject, T()>(obj, Impl_GetComponentsInChildren, mre);
UnityDispatcher.Instance.Invoke(job);
mre.WaitOne();
return job.Result;
- On the attached code from above you can see
Impl_GetComponentsInChildren which is the alias for doing this simple thing (THAT IS EXECUTED ON THE UNITY THREAD):
private static void Impl_GetComponentsInChildren<T>(GameObject obj, out T() result)
{
result = obj.GetComponentsInChildren<T>();
}
My question here is there a nicer way to do this?
- Has my Unity Dispatcher a good design on the memory / CPU usage part? I mean, this will be called lots and lots of times and I didn’t benchmark this yet. So, can you suggest to me something lighter or a tip to keep in mind on this topic?
