Take steps to avoid threadpool starvation

src/Build/BackEnd/Components/Caching/ResultsCache.cs

@@ -318,7 +318,7 @@ private static bool CheckResults(BuildResult result, List<string> targets, bool
             bool missingTargetFound = false;
             foreach (string target in targets)
             {
-                if (!result.HasResultsForTarget(target) || (result[target].ResultCode == TargetResultCode.Skipped && !skippedResultsAreOK))
+                if (!result.TryGetResultsForTarget(target, out TargetResult targetResult) || (targetResult.ResultCode == TargetResultCode.Skipped && !skippedResultsAreOK))
                 {
                     if (checkTargetsMissingResults)
                     {
@@ -334,7 +334,7 @@ private static bool CheckResults(BuildResult result, List<string> targets, bool
                 {
                     // If the result was a failure and we have not seen any skipped targets up to this point, then we conclude we do
                     // have results for this request, and they indicate failure.
-                    if (result[target].ResultCode == TargetResultCode.Failure && (!checkTargetsMissingResults || !missingTargetFound))
+                    if (targetResult.ResultCode == TargetResultCode.Failure && (!checkTargetsMissingResults || !missingTargetFound))
                     {
                         return true;
                     }

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

@@ -10,15 +10,14 @@
 using System.IO.Pipes;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 #if FEATURE_PIPE_SECURITY
 using System.Security.Principal;
 #endif
 
 #if FEATURE_APM
 using Microsoft.Build.Eventing;
-#else
-using System.Threading;
 #endif
 using Microsoft.Build.Internal;
 using Microsoft.Build.Shared;
@@ -580,7 +579,7 @@ private enum ExitPacketState
             /// <summary>
             /// A queue used for enqueuing packets to write to the stream asynchronously.
             /// </summary>
-            private BlockingCollection<INodePacket> _packetWriteQueue = new BlockingCollection<INodePacket>();
+            private ConcurrentQueue<INodePacket> _packetWriteQueue = new ConcurrentQueue<INodePacket>();
 
             /// <summary>
             /// A task representing the last packet write, so we can chain packet writes one after another.
@@ -715,7 +714,7 @@ public void SendData(INodePacket packet)
                 {
                     _exitPacketState = ExitPacketState.ExitPacketQueued;
                 }
-                _packetWriteQueue.Add(packet);
+                _packetWriteQueue.Enqueue(packet);
                 DrainPacketQueue();
             }
 
@@ -737,65 +736,63 @@ private void DrainPacketQueue()
                 {
                     // average latency between the moment this runs and when the delegate starts
                     // running is about 100-200 microseconds (unless there's thread pool saturation)
-                    _packetWriteDrainTask = _packetWriteDrainTask.ContinueWith(_ =>
+                    _packetWriteDrainTask = _packetWriteDrainTask.ContinueWith(
+                        SendDataCoreAsync,
+                        this,
+                        TaskScheduler.Default).Unwrap();
+
+                    static async Task SendDataCoreAsync(Task _, object state)
                     {
-                        while (_packetWriteQueue.TryTake(out var packet))
+                        NodeContext context = (NodeContext)state;
+                        while (context._packetWriteQueue.TryDequeue(out var packet))
                         {
-                            SendDataCore(packet);
-                        }
-                    }, TaskScheduler.Default);
-                }
-            }
+                            MemoryStream writeStream = context._writeBufferMemoryStream;
 
-            /// <summary>
-            /// Actually writes and sends the packet. This can't be called in parallel
-            /// because it reuses the _writeBufferMemoryStream, and this is why we use
-            /// the _packetWriteDrainTask to serially chain invocations one after another.
-            /// </summary>
-            /// <param name="packet">The packet to send.</param>
-            private void SendDataCore(INodePacket packet)
-            {
-                MemoryStream writeStream = _writeBufferMemoryStream;
+                            // clear the buffer but keep the underlying capacity to avoid reallocations
+                            writeStream.SetLength(0);
 
-                // clear the buffer but keep the underlying capacity to avoid reallocations
-                writeStream.SetLength(0);
+                            ITranslator writeTranslator = BinaryTranslator.GetWriteTranslator(writeStream);
+                            try
+                            {
+                                writeStream.WriteByte((byte)packet.Type);
 
-                ITranslator writeTranslator = BinaryTranslator.GetWriteTranslator(writeStream);
-                try
-                {
-                    writeStream.WriteByte((byte)packet.Type);
+                                // Pad for the packet length
+                                WriteInt32(writeStream, 0);
+                                packet.Translate(writeTranslator);
 
-                    // Pad for the packet length
-                    WriteInt32(writeStream, 0);
-                    packet.Translate(writeTranslator);
+                                int writeStreamLength = (int)writeStream.Position;
 
-                    int writeStreamLength = (int)writeStream.Position;
+                                // Now plug in the real packet length
+                                writeStream.Position = 1;
+                                WriteInt32(writeStream, writeStreamLength - 5);
 
-                    // Now plug in the real packet length
-                    writeStream.Position = 1;
-                    WriteInt32(writeStream, writeStreamLength - 5);
+                                byte[] writeStreamBuffer = writeStream.GetBuffer();
 
-                    byte[] writeStreamBuffer = writeStream.GetBuffer();
+                                for (int i = 0; i < writeStreamLength; i += MaxPacketWriteSize)
+                                {
+                                    int lengthToWrite = Math.Min(writeStreamLength - i, MaxPacketWriteSize);
+#pragma warning disable CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                                    await context._serverToClientStream.WriteAsync(writeStreamBuffer, i, lengthToWrite, CancellationToken.None);
+#pragma warning restore CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                                }
 
-                    for (int i = 0; i < writeStreamLength; i += MaxPacketWriteSize)
-                    {
-                        int lengthToWrite = Math.Min(writeStreamLength - i, MaxPacketWriteSize);
-                        _serverToClientStream.Write(writeStreamBuffer, i, lengthToWrite);
-                    }
-                    if (IsExitPacket(packet))
-                    {
-                        _exitPacketState = ExitPacketState.ExitPacketSent;
+                                if (IsExitPacket(packet))
+                                {
+                                    context._exitPacketState = ExitPacketState.ExitPacketSent;
+                                }
+                            }
+                            catch (IOException e)
+                            {
+                                // Do nothing here because any exception will be caught by the async read handler
+                                CommunicationsUtilities.Trace(context._nodeId, "EXCEPTION in SendData: {0}", e);
+                            }
+                            catch (ObjectDisposedException) // This happens if a child dies unexpectedly
+                            {
+                                // Do nothing here because any exception will be caught by the async read handler
+                            }
+                        }
                     }
                 }
-                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);
-                }
-                catch (ObjectDisposedException) // This happens if a child dies unexpectedly
-                {
-                    // Do nothing here because any exception will be caught by the async read handler
-                }
             }
 
             private static bool IsExitPacket(INodePacket packet)
@@ -806,7 +803,7 @@ private static bool IsExitPacket(INodePacket packet)
             /// <summary>
             /// Avoid having a BinaryWriter just to write a 4-byte int
             /// </summary>
-            private void WriteInt32(MemoryStream stream, int value)
+            private static void WriteInt32(MemoryStream stream, int value)
             {
                 stream.WriteByte((byte)value);
                 stream.WriteByte((byte)(value >> 8));

src/Build/BackEnd/Components/RequestBuilder/TargetBuilder.cs

@@ -557,9 +557,8 @@ await PushTargets(errorTargets, currentTargetEntry, currentTargetEntry.Lookup, t
         /// <returns>True to skip the target, false otherwise.</returns>
         private bool CheckSkipTarget(ref bool stopProcessingStack, TargetEntry currentTargetEntry)
         {
-            if (_buildResult.HasResultsForTarget(currentTargetEntry.Name))
+            if (_buildResult.TryGetResultsForTarget(currentTargetEntry.Name, out TargetResult targetResult))
             {
-                TargetResult targetResult = _buildResult[currentTargetEntry.Name] as TargetResult;
                 ErrorUtilities.VerifyThrowInternalNull(targetResult, "targetResult");
 
                 if (targetResult.ResultCode != TargetResultCode.Skipped)
@@ -676,12 +675,9 @@ private async Task<bool> PushTargets(IList<TargetSpecification> targets, TargetE
                 {
                     // Don't build any Before or After targets for which we already have results.  Unlike other targets,
                     // we don't explicitly log a skipped-with-results message because it is not interesting.
-                    if (_buildResult.HasResultsForTarget(targetSpecification.TargetName))
+                    if (_buildResult.TryGetResultsForTarget(targetSpecification.TargetName, out TargetResult targetResult) && targetResult.ResultCode != TargetResultCode.Skipped)
                     {
-                        if (_buildResult[targetSpecification.TargetName].ResultCode != TargetResultCode.Skipped)
-                        {
-                            continue;
-                        }
+                        continue;
                     }
                 }
 

src/Build/BackEnd/Components/Scheduler/Scheduler.cs

@@ -1019,11 +1019,8 @@ private bool IsTraversalRequest(BuildRequest request)
         private void AssignUnscheduledRequestsWithConfigurationCountLevelling(List<ScheduleResponse> responses, HashSet<int> idleNodes)
         {
             // Assign requests but try to keep the same number of configurations on each node
-            List<int> nodesByConfigurationCountAscending = new List<int>(_availableNodes.Keys);
-            nodesByConfigurationCountAscending.Sort(delegate (int left, int right)
-            {
-                return Comparer<int>.Default.Compare(_schedulingData.GetConfigurationsCountByNode(left, true /* excludeTraversals */, _configCache), _schedulingData.GetConfigurationsCountByNode(right, true /* excludeTraversals */, _configCache));
-            });
+            // Use OrderBy to sort since it will cache the lookup in configCache which. This reduces the number of times we have to acquire the lock.
+            IEnumerable<int> nodesByConfigurationCountAscending = _availableNodes.Keys.OrderBy(x => _schedulingData.GetConfigurationsCountByNode(x, excludeTraversals: true, _configCache));
 
             // Assign projects to nodes, preferring to assign work to nodes with the fewest configurations first.
             foreach (int nodeId in nodesByConfigurationCountAscending)

src/Build/BackEnd/Shared/BuildResult.cs

@@ -10,6 +10,7 @@
 using Microsoft.Build.Shared;
 using Microsoft.Build.Shared.FileSystem;
 using Microsoft.Build.Framework;
+using System.Diagnostics.CodeAnalysis;
 
 namespace Microsoft.Build.Execution
 {
@@ -601,6 +602,16 @@ public bool HasResultsForTarget(string target)
             return _resultsByTarget?.ContainsKey(target) ?? false;
         }
 
+        public bool TryGetResultsForTarget(string target, [NotNullWhen(true)] out TargetResult? value)
+        {
+            if (_resultsByTarget is null)
+            {
+                value = default;
+                return false;
+            }
+
+            return _resultsByTarget.TryGetValue(target, out value);
+        }
         #region INodePacket Members
 
         /// <summary>

src/Shared/CommunicationsUtilities.cs

@@ -540,26 +540,20 @@ internal static int ReadIntForHandshake(this PipeStream stream, byte? byteToAcce
             else
 #endif
             {
-                // Legacy approach with an early-abort for connection attempts from ancient MSBuild.exes
-                for (int i = 0; i < bytes.Length; i++)
+                int bytesRead = stream.Read(bytes, 0, bytes.Length);
+                if (bytesRead != bytes.Length)
                 {
-                    int read = stream.ReadByte();
-
-                    if (read == -1)
-                    {
-                        // We've unexpectly reached end of stream.
-                        // We are now in a bad state, disconnect on our end
-                        throw new IOException(String.Format(CultureInfo.InvariantCulture, "Unexpected end of stream while reading for handshake"));
-                    }
-
-                    bytes[i] = Convert.ToByte(read);
+                    // We've unexpectly reached end of stream.
+                    // We are now in a bad state, disconnect on our end
+                    throw new IOException(String.Format(CultureInfo.InvariantCulture, "Unexpected end of stream while reading for handshake"));
+                }
 
-                    if (i == 0 && byteToAccept != null && byteToAccept != bytes[0])
-                    {
-                        stream.WriteIntForHandshake(0x0F0F0F0F);
-                        stream.WriteIntForHandshake(0x0F0F0F0F);
-                        throw new InvalidOperationException(String.Format(CultureInfo.InvariantCulture, "Client: rejected old host. Received byte {0} instead of {1}.", bytes[0], byteToAccept));
-                    }
+                // Legacy approach with an early-abort for connection attempts from ancient MSBuild.exes
+                if (byteToAccept != null && byteToAccept != bytes[0])
+                {
+                    stream.WriteIntForHandshake(0x0F0F0F0F);
+                    stream.WriteIntForHandshake(0x0F0F0F0F);
+                    throw new InvalidOperationException(String.Format(CultureInfo.InvariantCulture, "Client: rejected old host. Received byte {0} instead of {1}.", bytes[0], byteToAccept));
                 }
             }
 

src/Shared/NodeEndpointOutOfProcBase.cs

@@ -152,7 +152,7 @@ public LinkStatus LinkStatus
         public void Listen(INodePacketFactory factory)
         {
             ErrorUtilities.VerifyThrow(_status == LinkStatus.Inactive, "Link not inactive.  Status is {0}", _status);
-            ErrorUtilities.VerifyThrowArgumentNull(factory);
+            ErrorUtilities.VerifyThrowArgumentNull(factory, nameof(factory));
             _packetFactory = factory;
 
             InitializeAsyncPacketThread();
@@ -314,7 +314,7 @@ private void InternalDisconnect()
         /// <param name="packet">The packet to be transmitted.</param>
         private void EnqueuePacket(INodePacket packet)
         {
-            ErrorUtilities.VerifyThrowArgumentNull(packet);
+            ErrorUtilities.VerifyThrowArgumentNull(packet, nameof(packet));
             ErrorUtilities.VerifyThrow(_packetQueue != null, "packetQueue is null");
             ErrorUtilities.VerifyThrow(_packetAvailable != null, "packetAvailable is null");
             _packetQueue.Enqueue(packet);
@@ -531,15 +531,13 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
             {
                 // Ordering is important.  We want packetAvailable to supercede terminate otherwise we will not properly wait for all
                 // packets to be sent by other threads which are shutting down, such as the logging thread.
-                WaitHandle[] handles =
-                [
+                WaitHandle[] handles = new WaitHandle[] {
 #if FEATURE_APM
                     result.AsyncWaitHandle,
 #else
                     ((IAsyncResult)readTask).AsyncWaitHandle,
 #endif
-                    localPacketAvailable, localTerminatePacketPump
-                ];
+                    localPacketAvailable, localTerminatePacketPump };
 
                 int waitId = WaitHandle.WaitAny(handles);
                 switch (waitId)