Fixes(#16) - Disable LazyExecute(SingleThreadEventLoop & LoopExecutor)

cuteant · Aug 28, 2020 · ada8a3a · ada8a3a
1 parent 78eeb31
commit ada8a3a
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 44 deletions.
diff --git a/src/DotNetty.Common/Concurrency/SingleThreadEventExecutor.cs b/src/DotNetty.Common/Concurrency/SingleThreadEventExecutor.cs
@@ -1048,7 +1048,11 @@ public override void Execute(IRunnable task)
             }
             else
             {
-                InternalLazyExecute(task);
+                // netty 第一个任务进来，不管是否延迟任务，都会启动线程
+                // 防止线程启动后，第一个进来的就是 lazy task
+                var firstTask = _firstTask;
+                if (firstTask) { _firstTask = false; }
+                InternalExecute(task, firstTask);
             }
         }
 
@@ -1057,15 +1061,6 @@ public override void LazyExecute(IRunnable task)
             InternalExecute(task, false);
         }
 
-        protected virtual void InternalLazyExecute(IRunnable task)
-        {
-            // netty 第一个任务进来，不管是否延迟任务，都会启动线程
-            // 防止线程启动后，第一个进来的就是 lazy task
-            var firstTask = _firstTask;
-            if (firstTask) { _firstTask = false; }
-            InternalExecute(task, firstTask);
-        }
-
         [MethodImpl(InlineMethod.AggressiveOptimization)]
         private void InternalExecute(IRunnable task, bool immediate)
         {

diff --git a/src/DotNetty.Transport.Libuv/LoopExecutor.cs b/src/DotNetty.Transport.Libuv/LoopExecutor.cs
@@ -207,6 +207,24 @@ protected sealed override IQueue<IRunnable> NewTaskQueue(int maxPendingTasks)
             return new CompatibleConcurrentQueue<IRunnable>();
         }
 
+        public override void Execute(IRunnable task)
+        {
+            InternalExecute(task);
+        }
+
+        public override void LazyExecute(IRunnable task)
+        {
+            InternalExecute(task);
+        }
+
+        [MethodImpl(InlineMethod.AggressiveOptimization)]
+        private void InternalExecute(IRunnable task)
+        {
+            AddTask(task);
+
+            WakeUp(InEventLoop);
+        }
+
         protected override void WakeUp(bool inEventLoop)
         {
             // If the executor is not in the event loop, wake up the loop by async handle immediately.

diff --git a/src/DotNetty.Transport/Channels/SingleThreadEventLoop.cs b/src/DotNetty.Transport/Channels/SingleThreadEventLoop.cs
@@ -45,7 +45,6 @@ public class SingleThreadEventLoop : SingleThreadEventLoopBase
 
         private readonly long _breakoutNanosInterval;
         private readonly ManualResetEventSlim _emptyEvent;
-        private bool _firstTask; // 不需要设置 volatile
 
         public SingleThreadEventLoop(IEventLoopGroup parent)
             : this(parent, DefaultBreakoutInterval)
@@ -85,7 +84,6 @@ public SingleThreadEventLoop(IEventLoopGroup parent, IThreadFactory threadFactor
         public SingleThreadEventLoop(IEventLoopGroup parent, IThreadFactory threadFactory, IRejectedExecutionHandler rejectedHandler, TimeSpan breakoutInterval)
             : base(parent, threadFactory, false, int.MaxValue, rejectedHandler)
         {
-            _firstTask = true;
             _emptyEvent = new ManualResetEventSlim(false, 1);
             _breakoutNanosInterval = PreciseTime.ToDelayNanos(breakoutInterval);
             Start();
@@ -115,40 +113,20 @@ protected override void Run()
         /// <inheritdoc />
         public sealed override void Execute(IRunnable task)
         {
-            if (!(task is ILazyRunnable)
-#if DEBUG
-                && WakesUpForTask(task)
-#endif
-                )
-            {
-                InternalExecute(task, true);
-            }
-            else
-            {
-                InternalLazyExecute(task);
-            }
+            InternalExecute(task);
         }
 
         public sealed override void LazyExecute(IRunnable task)
         {
-            InternalExecute(task, false);
-        }
-
-        protected override void InternalLazyExecute(IRunnable task)
-        {
-            // netty 第一个任务进来，不管是否延迟任务，都会启动线程
-            // 防止线程启动后，第一个进来的就是 lazy task
-            var firstTask = _firstTask;
-            if (firstTask) { _firstTask = false; }
-            InternalExecute(task, firstTask);
+            InternalExecute(task);
         }
 
         [MethodImpl(InlineMethod.AggressiveOptimization)]
-        private void InternalExecute(IRunnable task, bool immediate)
+        private void InternalExecute(IRunnable task)
         {
             AddTask(task);
 
-            if (immediate) { _emptyEvent.Set(); }
+            if (!InEventLoop) { _emptyEvent.Set(); }
         }
 
         protected sealed override void WakeUp(bool inEventLoop)
@@ -171,7 +149,10 @@ protected sealed override IRunnable PollTask()
 #else
             _emptyEvent.Reset();
 #endif
-            if (_taskQueue.TryDequeue(out task) || IsShuttingDown) { return task; }
+            if (_taskQueue.TryDequeue(out task) || IsShuttingDown) // revisit queue as producer might have put a task in meanwhile
+            {
+                return task;
+            }
 
             // revisit queue as producer might have put a task in meanwhile
             if (ScheduledTaskQueue.TryPeek(out IScheduledRunnable nextScheduledTask))
@@ -180,14 +161,8 @@ protected sealed override IRunnable PollTask()
                 if ((ulong)delayNanos > 0UL) // delayNanos >= 0
                 {
                     var timeout = PreciseTime.ToMilliseconds(delayNanos);
-                    if (_emptyEvent.Wait((int)Math.Min(timeout, MaxDelayMilliseconds)))
-                    {
-                        if (_taskQueue.TryDequeue(out task)) { return task; }
-                    }
+                    _emptyEvent.Wait((int)Math.Min(timeout, MaxDelayMilliseconds));
                 }
-
-                FetchFromScheduledTaskQueue();
-                _taskQueue.TryDequeue(out task);
             }
             else
             {

diff --git a/test/DotNetty.Transport.Tests/Channel/Local/LocalChannelTest.cs b/test/DotNetty.Transport.Tests/Channel/Local/LocalChannelTest.cs
@@ -995,6 +995,7 @@ public TestHandler0(IPromise promise, IPromise assertPromise)
 
             public override void ChannelActive(IChannelHandlerContext ctx)
             {
+                Thread.Sleep(10);
                 // Ensure the promise was done before the handler method is triggered.
                 if (_promise.IsCompleted)
                 {