continuations and recycle are inlined

src/DotNetty.Common/Concurrency/AbstractPromise.cs

@@ -17,8 +17,7 @@ public abstract class AbstractPromise : IPromise, IValueTaskSource
 
         static readonly Exception CanceledException = new OperationCanceledException();
         static readonly Exception CompletedNoException = new Exception();
-
-        protected IEventExecutor executor;
+
         protected Exception exception;
 
         Action<object> callback;
@@ -131,8 +130,7 @@ bool TryExecuteCallback()
 
             try
             {
-                Contract.Requires(this.executor != null);
-                this.executor.Execute(this.callback, this.callbackState);
+                this.callback(this.callbackState);
                 return true;
             }
             finally

src/DotNetty.Common/Concurrency/AbstractRecyclablePromise.cs

@@ -40,31 +40,28 @@ protected override bool TryComplete0(Exception exception)
             }
             catch
             {
-                this.executor.Execute(this.Recycle);
+                this.Recycle();
                 throw;
             }
 
             if (completed)
             {
-                this.executor.Execute(this.Recycle);
+                this.Recycle();
             }
 
             return completed;
         }
 
-        protected void Init(IEventExecutor executor)
+        protected void Init()
         {
-            this.executor = executor;
             this.recycled = false;
         }
 
         protected virtual void Recycle()
         {
-            this.executor = null;
             this.exception = null;
             this.ClearCallback();
             this.recycled = true;
-
             this.handle.Release(this);
         }
 
@@ -81,11 +78,6 @@ void ThrowIfRecycled()
             {
                 throw new InvalidOperationException("Attempt to use recycled channel promise");
             }
-
-            if (this.executor == null)
-            {
-                throw new InvalidOperationException("Attempt to use recyclable channel promise without executor");
-            }
         }
     }
 }

src/DotNetty.Transport/Channels/AbstractChannel.cs

@@ -192,6 +192,8 @@ public IChannel Read()
         public ValueTask WriteAsync(object msg) => this.pipeline.WriteAsync(msg);
 
         public ValueTask WriteAndFlushAsync(object message) => this.pipeline.WriteAndFlushAsync(message);
+
+        public ValueTask WriteAndFlushAsync(object message, bool notifyComplete) => this.pipeline.WriteAndFlushAsync(message, notifyComplete);
 
         public Task CloseCompletion => this.closeFuture.Task;
 

src/DotNetty.Transport/Channels/AbstractChannelHandlerContext.cs

@@ -797,7 +797,7 @@ public ValueTask WriteAsync(object msg)
         {
             Contract.Requires(msg != null);
             // todo: check for cancellation
-            return this.WriteAsync(msg, false);
+            return this.WriteAsync(msg, FlushMode.NoFlush);
         }
 
         ValueTask InvokeWriteAsync(object msg) => this.Added ? this.InvokeWriteAsync0(msg) : this.WriteAsync(msg);
@@ -853,41 +853,45 @@ void InvokeFlush0()
             }
         }
 
-        public ValueTask WriteAndFlushAsync(object message)
+        public ValueTask WriteAndFlushAsync(object message) => this.WriteAndFlushAsync(message, true);
+
+        public ValueTask WriteAndFlushAsync(object message, bool notifyComplete)
         {
             Contract.Requires(message != null);
             // todo: check for cancellation
 
-            return this.WriteAsync(message, true);
+            return this.WriteAsync(message, notifyComplete ? FlushMode.Flush : FlushMode.VoidFlush);
         }
 
-        ValueTask InvokeWriteAndFlushAsync(object msg)
+        ValueTask InvokeWriteAndFlushAsync(object msg, bool notifyComplete)
         {
             if (this.Added)
             {
                 ValueTask task = this.InvokeWriteAsync0(msg);
+                //flush can synchronously complete write, hence Task allocation required to capture result
+                task = notifyComplete ? task.Preserve() : default(ValueTask);
                 this.InvokeFlush0();
                 return task;
             }
             return this.WriteAndFlushAsync(msg);
         }
 
-        ValueTask WriteAsync(object msg, bool flush)
+        ValueTask WriteAsync(object msg, FlushMode mode)
         {
             AbstractChannelHandlerContext next = this.FindContextOutbound();
             object m = this.pipeline.Touch(msg, next);
             IEventExecutor nextExecutor = next.Executor;
             if (nextExecutor.InEventLoop)
             {
-                return flush
-                    ? next.InvokeWriteAndFlushAsync(m)
-                    : next.InvokeWriteAsync(m);
+                return mode == FlushMode.NoFlush
+                    ? next.InvokeWriteAsync(m)
+                    : next.InvokeWriteAndFlushAsync(m, mode == FlushMode.Flush);
             }
             else
             {
-                AbstractWriteTask task = flush 
-                    ? WriteAndFlushTask.NewInstance(next, m)
-                    : (AbstractWriteTask)WriteTask.NewInstance(next, m);
+                AbstractWriteTask task = mode == FlushMode.NoFlush 
+                    ? WriteTask.NewInstance(next, m)
+                    : (AbstractWriteTask)WriteAndFlushTask.NewInstance(next, m, mode == FlushMode.Flush);
                 SafeExecuteOutbound(nextExecutor, task, msg);
                 return task;
             }
@@ -974,6 +978,13 @@ static void SafeExecuteOutbound(IEventExecutor executor, AbstractWriteTask task,
 
         public override string ToString() => $"{typeof(IChannelHandlerContext).Name} ({this.Name}, {this.Channel})";
 
+        enum FlushMode : byte
+        {
+            NoFlush = 0,
+            VoidFlush = 1,
+            Flush = 2
+        }
+
         abstract class AbstractWriteTask : AbstractRecyclablePromise, IRunnable
         {
             static readonly bool EstimateTaskSizeOnSubmit =
@@ -989,7 +1000,7 @@ abstract class AbstractWriteTask : AbstractRecyclablePromise, IRunnable
 
             protected static void Init(AbstractWriteTask task, AbstractChannelHandlerContext ctx, object msg)
             {
-                task.Init(ctx.Executor);
+                task.Init();
                 task.ctx = ctx;
                 task.msg = msg;
 
@@ -1047,7 +1058,7 @@ public void Run()
                 }
             }
 
-            protected virtual ValueTask WriteAsync(AbstractChannelHandlerContext ctx, object msg) => ctx.InvokeWriteAsync(msg);
+            protected abstract ValueTask WriteAsync(AbstractChannelHandlerContext ctx, object msg);
 
             /*public override void Recycle()
             {
@@ -1072,30 +1083,31 @@ public static WriteTask NewInstance(AbstractChannelHandlerContext ctx, object ms
                 : base(handle)
             {
             }
+
+            protected override ValueTask WriteAsync(AbstractChannelHandlerContext ctx, object msg) => ctx.InvokeWriteAsync(msg);
         }
 
         sealed class WriteAndFlushTask : AbstractWriteTask
         {
+            bool notifyComplete;
+
             static readonly ThreadLocalPool<WriteAndFlushTask> Recycler = new ThreadLocalPool<WriteAndFlushTask>(handle => new WriteAndFlushTask(handle));
 
-            public static WriteAndFlushTask NewInstance(AbstractChannelHandlerContext ctx, object msg) 
+            public static WriteAndFlushTask NewInstance(AbstractChannelHandlerContext ctx, object msg, bool notifyComplete) 
             {
                 WriteAndFlushTask task = Recycler.Take();
                 Init(task, ctx, msg);
+                task.notifyComplete = notifyComplete;
                 return task;
             }
 
             WriteAndFlushTask(ThreadLocalPool.Handle handle)
                 : base(handle)
             {
             }
+
+            protected override ValueTask WriteAsync(AbstractChannelHandlerContext ctx, object msg) => ctx.InvokeWriteAndFlushAsync(msg, this.notifyComplete);
 
-            protected override ValueTask WriteAsync(AbstractChannelHandlerContext ctx, object msg)
-            {
-                ValueTask result = base.WriteAsync(ctx, msg);
-                ctx.InvokeFlush();
-                return result;
-            }
         }
     }
 }

src/DotNetty.Transport/Channels/BatchingPendingWriteQueue.cs

@@ -334,7 +334,7 @@ sealed class PendingWrite : AbstractRecyclablePromise
             public static PendingWrite NewInstance(IEventExecutor executor, object msg, int size)
             {
                 PendingWrite write = Pool.Take();
-                write.Init(executor);
+                write.Init();
                 write.Add(msg, size);
                 return write;
             }

src/DotNetty.Transport/Channels/ChannelOutboundBuffer.cs

@@ -808,7 +808,7 @@ sealed class Entry : AbstractRecyclablePromise
             public static Entry NewInstance(IEventExecutor executor, object msg, int size)
             {
                 Entry entry = Pool.Take();
-                entry.Init(executor);
+                entry.Init();
                 entry.Message = msg;
                 entry.PendingSize = size;
                 return entry;

src/DotNetty.Transport/Channels/DefaultChannelPipeline.cs

@@ -831,6 +831,8 @@ public IChannelPipeline Flush()
         }
 
         public ValueTask WriteAndFlushAsync(object msg) => this.tail.WriteAndFlushAsync(msg);
+
+        public ValueTask WriteAndFlushAsync(object msg, bool notifyComplete) => this.tail.WriteAndFlushAsync(msg, notifyComplete);
 
         string FilterName(string name, IChannelHandler handler)
         {

src/DotNetty.Transport/Channels/IChannel.cs

@@ -67,5 +67,7 @@ public interface IChannel : IAttributeMap, IComparable<IChannel>
         IChannel Flush();
 
         ValueTask WriteAndFlushAsync(object message);
+
+        ValueTask WriteAndFlushAsync(object message, bool notifyComplete);
     }
 }

src/DotNetty.Transport/Channels/IChannelHandlerContext.cs

@@ -72,6 +72,8 @@ public interface IChannelHandlerContext : IAttributeMap
         IChannelHandlerContext Flush();
 
         ValueTask WriteAndFlushAsync(object message);
+
+        ValueTask WriteAndFlushAsync(object message, bool notifyComplete);
 
         /// <summary>
         /// Request to bind to the given <see cref="EndPoint"/>.

src/DotNetty.Transport/Channels/IChannelPipeline.cs

@@ -695,5 +695,7 @@ public interface IChannelPipeline : IEnumerable<IChannelHandler>
         /// Shortcut for calling both <see cref="WriteAsync"/> and <see cref="Flush"/>.
         /// </summary>
         ValueTask WriteAndFlushAsync(object msg);
+
+        ValueTask WriteAndFlushAsync(object msg, bool notifyComplete);
     }
 }

src/DotNetty.Transport/Channels/PendingWriteQueue.cs

@@ -294,7 +294,7 @@ sealed class PendingWrite : AbstractRecyclablePromise
             public static PendingWrite NewInstance(IEventExecutor executor, object msg, int size)
             {
                 PendingWrite write = Pool.Take();
-                write.Init(executor);
+                write.Init();
                 write.Size = size;
                 write.Msg = msg;
                 return write;