using System; using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; using UnityEngine; namespace Unity.Pipeline.Threading { /// /// Dispatches work to Unity's main thread from background threads (required for accessing Unity /// APIs from HTTP request handlers). /// /// Each pipeline server owns its own instance (no global singleton): it is initialized on Start /// and pumped from the main thread — auto-pumped via EditorApplication.update in the editor, and /// by RuntimePipelineManager.Update in a player. /// public class Dispatcher { private readonly ConcurrentQueue m_WorkQueue = new ConcurrentQueue(); private volatile bool m_IsInitialized; private int m_MainThreadId = -1; public bool IsInitialized => m_IsInitialized; /// /// Initialize the dispatcher. Must be called from Unity's main thread. /// public void Initialize() { if (m_IsInitialized) return; m_MainThreadId = Thread.CurrentThread.ManagedThreadId; m_IsInitialized = true; #if UNITY_EDITOR UnityEditor.EditorApplication.update += ProcessWorkQueue; #endif } /// /// Shutdown the dispatcher and cancel any pending work. /// public void Shutdown() { if (!m_IsInitialized) return; #if UNITY_EDITOR UnityEditor.EditorApplication.update -= ProcessWorkQueue; #endif while (m_WorkQueue.TryDequeue(out var item)) { try { item.SetException(new OperationCanceledException("Dispatcher is shutting down")); } catch { } } m_IsInitialized = false; } /// /// Execute a function on the main thread and return the result (synchronous wait). /// public T Invoke(Func function, int timeoutMs = 60000) { if (!m_IsInitialized) throw new InvalidOperationException("Dispatcher must be initialized first"); if (IsMainThread()) return function(); var workItem = new WorkItem(function); m_WorkQueue.Enqueue(workItem); var startTime = DateTime.UtcNow; var task = workItem.TaskCompletionSource.Task; while (!task.IsCompleted) { if ((DateTime.UtcNow - startTime).TotalMilliseconds > timeoutMs) throw new TimeoutException($"Main thread operation timed out after {timeoutMs}ms"); Thread.Sleep(1); } if (task.IsFaulted) throw task.Exception?.GetBaseException() ?? new Exception("Unknown error"); if (task.IsCanceled) throw new OperationCanceledException("Main thread operation was cancelled"); return task.Result; } /// /// Execute an action on the main thread. /// public void Invoke(Action action, int timeoutMs = 60000) { Invoke(() => { action(); return null; }, timeoutMs); } /// /// Execute a function on the main thread and return the result (async version). /// public async Task InvokeAsync(Func function, int timeoutMs = 60000) { return await Task.Run(() => Invoke(function, timeoutMs)); } /// /// Execute an action on the main thread (async version). /// public async Task InvokeAsync(Action action, int timeoutMs = 60000) { await Task.Run(() => Invoke(action, timeoutMs)); } /// /// Check if we're currently on Unity's main thread. /// public bool IsMainThread() { return m_MainThreadId != -1 && Thread.CurrentThread.ManagedThreadId == m_MainThreadId; } /// /// Process queued work items. Called from EditorApplication.update or MonoBehaviour.Update. /// /// Max items to process per call, to limit frame-rate impact. public void ProcessWorkQueue(int maxItemsPerFrame) { int processedCount = 0; while (processedCount < maxItemsPerFrame && m_WorkQueue.TryDequeue(out var workItem)) { try { workItem.Execute(); } catch (Exception ex) { Debug.LogError($"Dispatcher work item failed: {ex.Message}"); workItem.SetException(ex); } processedCount++; } } public void ProcessWorkQueue() { ProcessWorkQueue(10); } private abstract class WorkItem { public abstract void Execute(); public abstract void SetException(Exception exception); } private class WorkItem : WorkItem { private readonly Func m_Function; public TaskCompletionSource TaskCompletionSource { get; } public WorkItem(Func function) { m_Function = function ?? throw new ArgumentNullException(nameof(function)); TaskCompletionSource = new TaskCompletionSource(); } public override void Execute() { try { var result = m_Function(); TaskCompletionSource.SetResult(result); } catch (Exception ex) { TaskCompletionSource.SetException(ex); } } public override void SetException(Exception exception) { TaskCompletionSource.SetException(exception); } } } }