Changed fiber pause to use FiberExecutionEventArgs.

Author: tosh-coding

Readme.md

@@ -163,7 +163,7 @@ while (...)
 |:-|:-|
 | Threads | `Thread.Sleep()` |
 | Fiber on dedicated thread | `Thread.Sleep()` |
-| Fiber on shared threads | `fiber.Enqueue(Func<Task<Action>>) & await Task.Delay()` |
+| Fiber on shared threads | `fiber.Enqueue(Action<FiberExecutionEventArgs>) & FiberExecutionEventArgs.Pause()/Resume(Action)` |
 | Asynchronous control flow | `await Task.Deley()` |
 
 ## Fibers ##

src/AsyncFiberWorks/Channels/Channel.cs

@@ -33,7 +33,10 @@ public IDisposable Subscribe(IAsyncExecutionContext executionContext, Func<T, Ta
         {
             return this._channel.AddHandler((msg) =>
             {
-                executionContext.Enqueue(() => receive.Invoke(msg));
+                executionContext.Enqueue((e) =>
+                {
+                    e.PauseWhileRunning(async () => await receive.Invoke(msg));
+                });
             });
         }
 

src/AsyncFiberWorks/Core/FiberExecutionEventArgs.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Core
+{
+    /// <summary>
+    /// Fiber execution notification handler arguments.
+    /// </summary>
+    public class FiberExecutionEventArgs : EventArgs
+    {
+        private Action _pause;
+        private Action<Action> _resume;
+
+        /// <summary>
+        /// Constructor.
+        /// </summary>
+        /// <param name="pause"></param>
+        /// <param name="resume"></param>
+        public FiberExecutionEventArgs(Action pause, Action<Action> resume)
+        {
+            _pause = pause;
+            _resume = resume;
+        }
+
+        /// <summary>
+        /// Pauses the consumption of the task queue.
+        /// This is only called during an Execute in the fiber.
+        /// </summary>
+        public void Pause()
+        {
+            _pause();
+        }
+
+        /// <summary>
+        /// Resumes consumption of a paused task queue.
+        /// </summary>
+        /// <param name="action">The action to be taken immediately after the resume.</param>
+        public void Resume(Action action)
+        {
+            _resume(action);
+        }
+
+        /// <summary>
+        /// Pause the fiber while the task is running.
+        /// </summary>
+        /// <param name="runningTask"></param>
+        public void PauseWhileRunning(Func<Task<Action>> runningTask)
+        {
+            this.Pause();
+            Task.Run(async () =>
+            {
+                Action resumingAction = default;
+                try
+                {
+                    resumingAction = await runningTask.Invoke();
+                }
+                finally
+                {
+                    this.Resume(resumingAction);
+                }
+            });
+        }
+    }
+}

src/AsyncFiberWorks/Core/IAsyncExecutionContext.cs

@@ -11,7 +11,7 @@ public interface IAsyncExecutionContext
         /// <summary>
         /// Enqueue a single action. It is executed sequentially.
         /// </summary>
-        /// <param name="func">Task generator. This is done after a pause in the fiber. The generated task is monitored and takes action to resume after completion.</param>
-        void Enqueue(Func<Task<Action>> func);
+        /// <param name="action">Action to be executed.</param>
+        void Enqueue(Action<FiberExecutionEventArgs> action);
     }
 }

src/AsyncFiberWorks/Fibers/PoolFiber.cs

@@ -14,6 +14,7 @@ public sealed class PoolFiber : IFiber
         private readonly object _lock = new object();
         private readonly IThreadPool _pool;
         private readonly IExecutor _executor;
+        private readonly FiberExecutionEventArgs _eventArgs;
 
         private Queue<Action> _queue = new Queue<Action>();
         private Queue<Action> _toPass = new Queue<Action>();
@@ -32,6 +33,7 @@ public PoolFiber(IThreadPool pool, IExecutor executor)
         {
             _pool = pool;
             _executor = executor;
+            _eventArgs = new FiberExecutionEventArgs(this.Pause, this.Resume);
         }
 
         /// <summary>
@@ -203,27 +205,12 @@ private void Resume(Action action)
         }
 
         /// <summary>
-        /// Enqueue a single task.
+        /// Enqueue a single action. It is executed sequentially.
         /// </summary>
-        /// <param name="func">Task generator. This is done after a pause in the fiber. The generated task is monitored and takes action to resume after completion.</param>
-        public void Enqueue(Func<Task<Action>> func)
+        /// <param name="action">Action to be executed.</param>
+        public void Enqueue(Action<FiberExecutionEventArgs> action)
         {
-            this.Enqueue(() =>
-            {
-                this.Pause();
-                Task.Run(async () =>
-                {
-                    Action resumingAction = default;
-                    try
-                    {
-                        resumingAction = await func.Invoke();
-                    }
-                    finally
-                    {
-                        this.Resume(resumingAction);
-                    }
-                });
-            });
+            this.Enqueue(() => action(_eventArgs));
         }
     }
 }

src/AsyncFiberWorks/Fibers/StubFiber.cs

@@ -16,6 +16,7 @@ public sealed class StubFiber : IFiber
         private readonly object _lock = new object();
         private readonly ConcurrentQueue<Action> _pending = new ConcurrentQueue<Action>();
         private readonly IExecutor _executor;
+        private readonly FiberExecutionEventArgs _eventArgs;
 
         private int _paused = 0;
         private Action _resumeAction = null;
@@ -35,6 +36,7 @@ public StubFiber()
         public StubFiber(IExecutor executor)
         {
             _executor = executor;
+            _eventArgs = new FiberExecutionEventArgs(this.Pause, this.Resume);
         }
 
         /// <summary>
@@ -172,27 +174,12 @@ private void ExecuteResumeProcess()
         }
 
         /// <summary>
-        /// Enqueue a single task.
+        /// Enqueue a single action. It is executed sequentially.
         /// </summary>
-        /// <param name="func">Task generator. This is done after a pause in the fiber. The generated task is monitored and takes action to resume after completion.</param>
-        public void Enqueue(Func<Task<Action>> func)
+        /// <param name="action">Action to be executed.</param>
+        public void Enqueue(Action<FiberExecutionEventArgs> action)
         {
-            this.Enqueue(() =>
-            {
-                this.Pause();
-                Task.Run(async () =>
-                {
-                    Action resumingAction = default;
-                    try
-                    {
-                        resumingAction = await func.Invoke();
-                    }
-                    finally
-                    {
-                        this.Resume(resumingAction);
-                    }
-                });
-            });
+            this.Enqueue(() => action(_eventArgs));
         }
     }
 }

src/AsyncFiberWorks/Fibers/ThreadFiber.cs

@@ -11,9 +11,15 @@ namespace AsyncFiberWorks.Fibers
     /// </summary>
     public class ThreadFiber : IFiber, IDisposable
     {
+        private readonly object _lock = new object();
         private readonly IDedicatedConsumerThreadWork _queue;
         private readonly UserWorkerThread _workerThread;
+        private readonly FiberExecutionEventArgs _eventArgs;
         private bool _stopped = false;
+        private bool _paused;
+        private bool _resuming;
+        private AutoResetEvent _autoReset = null;
+        private Action _resumeAction = null;
 
         /// <summary>
         /// Create a thread fiber with the default queue.
@@ -51,6 +57,7 @@ public ThreadFiber(string threadName)
         public ThreadFiber(IDedicatedConsumerThreadWork queue, string threadName, bool isBackground = true, ThreadPriority priority = ThreadPriority.Normal)
         {
             _queue = queue;
+            _eventArgs = new FiberExecutionEventArgs(this.Pause, this.Resume);
             _workerThread = new UserWorkerThread(queue, threadName, isBackground, priority);
             _workerThread.Start();
         }
@@ -61,6 +68,7 @@ public ThreadFiber(IDedicatedConsumerThreadWork queue, string threadName, bool i
         /// <param name="consumerThread">A consumer thread.</param>
         public ThreadFiber(UserWorkerThread consumerThread)
         {
+            _eventArgs = new FiberExecutionEventArgs(this.Pause, this.Resume);
             _workerThread = consumerThread;
             _workerThread.Start();
         }
@@ -85,6 +93,50 @@ public void Dispose()
             Stop();
         }
 
+        /// <summary>
+        /// Pauses the consumption of the task queue.
+        /// This is only called during an Execute in the fiber.
+        /// </summary>
+        /// <exception cref="InvalidOperationException">Pause was called twice.</exception>
+        private void Pause()
+        {
+            lock (_lock)
+            {
+                if (_paused)
+                {
+                    throw new InvalidOperationException("Pause was called twice.");
+                }
+                _paused = true;
+                if (_autoReset == null)
+                {
+                    _autoReset = new AutoResetEvent(false);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Resumes consumption of a paused task queue.
+        /// </summary>
+        /// <param name="action">The action to be taken immediately after the resume.</param>
+        /// <exception cref="InvalidOperationException">Resume was called in the unpaused state.</exception>
+        private void Resume(Action action)
+        {
+            lock (_lock)
+            {
+                if (!_paused)
+                {
+                    throw new InvalidOperationException("Resume was called in the unpaused state.");
+                }
+                if (_resuming)
+                {
+                    throw new InvalidOperationException("Resume was called twice.");
+                }
+                _resuming = true;
+                _resumeAction = action;
+                _autoReset.Set();
+            }
+        }
+
         /// <summary>
         /// Enqueue a single action.
         /// </summary>
@@ -95,31 +147,36 @@ public void Enqueue(Action action)
         }
 
         /// <summary>
-        /// Enqueue a single task.
+        /// Enqueue a single action. It is executed sequentially.
         /// </summary>
-        /// <param name="func">Task generator. This is done after a pause in the fiber. The generated task is monitored and takes action to resume after completion.</param>
-        public void Enqueue(Func<Task<Action>> func)
+        /// <param name="action">Action to be executed.</param>
+        public void Enqueue(Action<FiberExecutionEventArgs> action)
         {
             this.Enqueue(() =>
             {
-                var tcs = new TaskCompletionSource<Action>(TaskCreationOptions.RunContinuationsAsynchronously);
-                Task.Run(async () =>
+                action(_eventArgs);
+                bool tmpPaused;
+                lock (_lock)
                 {
-                    Action resumingAction = default;
+                    tmpPaused = _paused;
+                }
+                if (tmpPaused)
+                {
+                    _autoReset.WaitOne();
                     try
                     {
-                        resumingAction = await func.Invoke();
+                        _resumeAction();
                     }
                     finally
                     {
-                        tcs.SetResult(resumingAction);
+                        lock (_lock)
+                        {
+                            _paused = false;
+                            _resuming = false;
+                            _resumeAction = null;
+                        }
                     }
-                });
-
-                // This is in a dedicated thread. Blocking OK.
-                tcs.Task.Wait();
-                var act = tcs.Task.Result;
-                act?.Invoke();
+                }
             });
         }
     }