Revert "Switch back to using thread pool to process packet queue (#12113)"

This reverts commit 1bb8932. It
appeared to cause DevDiv#2524487 and related bugs.

Author: rainersigwald

src/Build/BackEnd/Components/Communications/NodeProviderOutOfProcBase.cs

@@ -161,29 +161,22 @@ protected void ShutdownAllNodes(bool nodeReuse, NodeContextTerminateDelegate ter
                 int timeout = 30;
 
                 // Attempt to connect to the process with the handshake without low priority.
-                using Stream nodeStream = TryConnectToProcess(nodeProcess.Id, timeout, NodeProviderOutOfProc.GetHandshake(nodeReuse, false));
-                if (nodeStream != null)
-                {
-                    ShutdownNode(terminateNode, factory, nodeProcess, nodeStream);
-                }
-                else
+                Stream nodeStream = TryConnectToProcess(nodeProcess.Id, timeout, NodeProviderOutOfProc.GetHandshake(nodeReuse, false));
+
+                if (nodeStream == null)
                 {
                     // If we couldn't connect attempt to connect to the process with the handshake including low priority.
-                    using Stream lowPriorityConnection = TryConnectToProcess(nodeProcess.Id, timeout, NodeProviderOutOfProc.GetHandshake(nodeReuse, true));
-
-                    if (lowPriorityConnection != null)
-                    {
-                        ShutdownNode(terminateNode, factory, nodeProcess, lowPriorityConnection);
-                    }
+                    nodeStream = TryConnectToProcess(nodeProcess.Id, timeout, NodeProviderOutOfProc.GetHandshake(nodeReuse, true));
                 }
-            }
 
-            static void ShutdownNode(NodeContextTerminateDelegate terminateNode, INodePacketFactory factory, Process nodeProcess, Stream nodeStream)
-            {
-                // If we're able to connect to such a process, send a packet requesting its termination
-                CommunicationsUtilities.Trace("Shutting down node with pid = {0}", nodeProcess.Id);
-                NodeContext nodeContext = new NodeContext(0, nodeProcess, nodeStream, factory, terminateNode);
-                nodeContext.SendData(new NodeBuildComplete(false /* no node reuse */));
+                if (nodeStream != null)
+                {
+                    // If we're able to connect to such a process, send a packet requesting its termination
+                    CommunicationsUtilities.Trace("Shutting down node with pid = {0}", nodeProcess.Id);
+                    NodeContext nodeContext = new NodeContext(0, nodeProcess, nodeStream, factory, terminateNode);
+                    nodeContext.SendData(new NodeBuildComplete(false /* no node reuse */));
+                    nodeStream.Dispose();
+                }
             }
         }
 
@@ -607,14 +600,19 @@ private enum ExitPacketState
             private readonly NodeContextTerminateDelegate _terminateDelegate;
 
             /// <summary>
-            /// A task representing the work to consume enqueued packets.
+            /// A dedicated thread to consume enqueued packets.
+            /// </summary>
+            private readonly Thread _drainPacketQueueThread;
+
+            /// <summary>
+            /// Used to signal the consuming thread that a packet has been enqueued;
             /// </summary>
-            private readonly Task _packetWriteDrainTask;
+            private readonly AutoResetEvent _packetEnqueued;
 
             /// <summary>
-            /// Used to signal the consuming task that a packet has been enqueued.
+            /// Used to signal that the exit packet has been sent and we no longer need to wait for the queue to drain.
             /// </summary>
-            private readonly SemaphoreSlim _packetEnqueued;
+            private readonly CancellationTokenSource _packetQueueDrainDelayCancellation;
 
             /// <summary>
             /// Tracks the state of the packet sent to terminate the node.
@@ -650,9 +648,13 @@ public NodeContext(int nodeId, Process process,
 #endif
 
                 _packetWriteQueue = new ConcurrentQueue<INodePacket>();
-                _packetEnqueued = new SemaphoreSlim(0, 1);
+                _packetEnqueued = new AutoResetEvent(false);
+                _packetQueueDrainDelayCancellation = new CancellationTokenSource();
 
-                _packetWriteDrainTask = DrainPacketQueue();
+                // specify the smallest stack size - 64kb
+                _drainPacketQueueThread = new Thread(DrainPacketQueue, 64 * 1024);
+                _drainPacketQueueThread.IsBackground = true;
+                _drainPacketQueueThread.Start(this);
             }
 
             /// <summary>
@@ -758,35 +760,30 @@ public void SendData(INodePacket packet)
                 {
                     _exitPacketState = ExitPacketState.ExitPacketQueued;
                 }
-
                 _packetWriteQueue.Enqueue(packet);
-
-                if (_packetEnqueued.CurrentCount == 0)
-                {
-                    // If the semaphore is not already signaled, signal it to wake up the draining task.
-                    _packetEnqueued.Release();
-                }
+                _packetEnqueued.Set();
             }
 
             /// <summary>
-            /// Use a threadpool thread to drain the queue and be careful not to block it where possible.
+            /// We use a dedicated thread to avoid blocking a threadpool thread.
             /// </summary>
             /// <remarks>Usually there'll be a single packet in the queue, but sometimes
             /// a burst of SendData comes in, with 10-20 packets scheduled.</remarks>
-            private async Task DrainPacketQueue()
+            private void DrainPacketQueue(object state)
             {
-                MemoryStream writeStream = _writeBufferMemoryStream;
-                Stream serverToClientStream = _pipeStream;
-                ITranslator writeTranslator = _writeTranslator;
+                NodeContext context = (NodeContext)state;
+                MemoryStream writeStream = context._writeBufferMemoryStream;
+                Stream serverToClientStream = context._pipeStream;
 
                 while (true)
                 {
-                    await _packetEnqueued.WaitAsync();
-                    while (_packetWriteQueue.TryDequeue(out INodePacket packet))
+                    context._packetEnqueued.WaitOne();
+                    while (context._packetWriteQueue.TryDequeue(out INodePacket packet))
                     {
                         // clear the buffer but keep the underlying capacity to avoid reallocations
                         writeStream.SetLength(0);
 
+                        ITranslator writeTranslator = context._writeTranslator;
                         try
                         {
                             writeStream.WriteByte((byte)packet.Type);
@@ -806,29 +803,22 @@ private async Task DrainPacketQueue()
                             for (int i = 0; i < writeStreamLength; i += MaxPacketWriteSize)
                             {
                                 int lengthToWrite = Math.Min(writeStreamLength - i, MaxPacketWriteSize);
-#if NET
-                                await serverToClientStream.WriteAsync(writeStreamBuffer.AsMemory(i, lengthToWrite));
-#else
-                                await serverToClientStream.WriteAsync(writeStreamBuffer, i, lengthToWrite);
-#endif
+
+                                serverToClientStream.Write(writeStreamBuffer, i, lengthToWrite);
                             }
 
                             if (IsExitPacket(packet))
                             {
-                                _exitPacketState = ExitPacketState.ExitPacketSent;
+                                context._exitPacketState = ExitPacketState.ExitPacketSent;
+                                context._packetQueueDrainDelayCancellation.Cancel();
 
                                 return;
                             }
-
-                            if (packet is NodeBuildComplete)
-                            {
-                                return;
-                            }
                         }
                         catch (IOException e)
                         {
                             // Do nothing here because any exception will be caught by the async read handler
-                            CommunicationsUtilities.Trace(_nodeId, "EXCEPTION in SendData: {0}", e);
+                            CommunicationsUtilities.Trace(context._nodeId, "EXCEPTION in SendData: {0}", e);
                         }
                         catch (ObjectDisposedException) // This happens if a child dies unexpectedly
                         {
@@ -860,7 +850,6 @@ private static void WriteInt32(MemoryStream stream, int value)
             private void Close()
             {
                 _pipeStream.Dispose();
-                _packetEnqueued.Dispose();
                 _terminateDelegate(_nodeId);
             }
 
@@ -874,16 +863,14 @@ public async Task WaitForExitAsync(ILoggingService loggingService)
                     // Wait up to 100ms until all remaining packets are sent.
                     // We don't need to wait long, just long enough for the Task to start running on the ThreadPool.
 #if NET
-                    await _packetWriteDrainTask.WaitAsync(TimeSpan.FromMilliseconds(100)).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+                    await Task.Delay(100, _packetQueueDrainDelayCancellation.Token).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
 #else
-                    using (var cts = new CancellationTokenSource(100))
-                    {
-                        await Task.WhenAny(_packetWriteDrainTask, Task.Delay(100, cts.Token));
-                        cts.Cancel();
-                    }
+                    await Task.WhenAny(Task.Delay(100, _packetQueueDrainDelayCancellation.Token));
 #endif
                 }
 
+                _packetQueueDrainDelayCancellation?.Dispose();
+
                 if (_exitPacketState == ExitPacketState.ExitPacketSent)
                 {
                     CommunicationsUtilities.Trace("Waiting for node with pid = {0} to exit", _process.Id);