Batch outbound logging packets to reduce pipe-write overhead

documentation/wiki/Nodes-Orchestration.md

@@ -33,6 +33,8 @@ Endpoints (nodes) communicate via named pipes (Windows or named pipes API implem
 
 The validation of transport is done via [proprietary handshake](https://github.com/dotnet/msbuild/blob/main/src/Build/BackEnd/Components/Communications/NodeProviderOutOfProcBase.cs#L501-L508).
 
+`NodeEndpointOutOfProcBase` combines queued logging packet frames into pipe writes up to a configurable byte budget. `MSBUILDLOGPACKETBATCHSIZE` sets that budget in bytes: the default is 64 KiB, zero disables batching, and values from zero through 1 MiB are accepted inclusively. Invalid values use the default. Each packet is serialized independently with its existing type, length, payload, and negotiated version; batching does not change the wire format or binary log contents. Pending frames are written when the queue drains, before control packets, and before shutdown, without waiting for more packets. Frames at least as large as a configured positive limit bypass the batch buffer. This setting controls user-space logging writes independently of `MSBUILDNODECONNECTIONBUFFERSIZE`, which controls the kernel pipe buffer size.
+
 
 ## Orchestration
 
@@ -69,4 +71,3 @@ Scheduler can (opt-in) dump a graph of dependencies from last build into a text
 
 Another mode of building is `graph build` - where project is build only once all its dependencies are resolved (so the build graph needs to be known and unchanged upfront).
 
-

src/Build.UnitTests/BackEnd/NodeEndpointReadLoop_Tests.cs

@@ -3,7 +3,9 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.IO;
+using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Build.BackEnd;
@@ -19,13 +21,15 @@ namespace Microsoft.Build.UnitTests.BackEnd;
 
 public sealed class NodeEndpointReadLoop_Tests(ITestOutputHelper output)
 {
+    private readonly ITestOutputHelper _output = output;
+
     [Theory]
     [InlineData(true, false)]
     [InlineData(false, false)]
     [InlineData(false, true)]
     public void ReadLoop_DistinguishesFailureFromExpectedDisconnect(bool throwOnRead, bool expectedDisconnect)
     {
-        using TestEnvironment env = TestEnvironment.Create(output);
+        using TestEnvironment env = TestEnvironment.Create(_output);
         using MemoryStream stream = throwOnRead ? new FailingReadStream() : new MemoryStream();
         using AutoResetEvent packetAvailable = new(false);
         using AutoResetEvent terminate = new(false);
@@ -70,6 +74,303 @@ public void ReadLoop_DistinguishesFailureFromExpectedDisconnect(bool throwOnRead
         }
     }
 
+    [Theory]
+    [InlineData(0, null, 3)]
+    [InlineData(0, "0", 5)]
+    [InlineData(0, "1048576", 3)]
+    [InlineData(1, null, 3)]
+    [InlineData(1, "0", 5)]
+    [InlineData(1, "1048576", 3)]
+    [InlineData(NodePacketTypeExtensions.PacketVersion, null, 3)]
+    [InlineData(NodePacketTypeExtensions.PacketVersion, "0", 5)]
+    [InlineData(NodePacketTypeExtensions.PacketVersion, "1048576", 3)]
+    public async Task LogBatch_PreservesSerializedFramesAndControlOrder(byte version, string? logPacketBatchSize, int expectedWrites)
+    {
+        INodePacket[] packets =
+        [
+            Log("first"), Log("second"),
+            new NodeBuildComplete(false, NodeBuildCompleteAction.Shutdown),
+            Log("third"), Log("fourth")
+        ];
+        byte[] expected = SerializeFrames(packets, version);
+        using CountingWriteStream stream = new();
+
+        await RunWritePump(stream, packets, version, logPacketBatchSize: logPacketBatchSize);
+
+        stream.ToArray().ShouldBe(expected);
+        stream.Writes.Count.ShouldBe(expectedWrites);
+    }
+
+    [Theory]
+    [InlineData(false, null)]
+    [InlineData(true, null)]
+    [InlineData(false, "0")]
+    [InlineData(true, "0")]
+    [InlineData(false, "64")]
+    [InlineData(true, "64")]
+    public async Task LogBatch_FlushesOnQueueEmptyOrTermination(bool terminating, string? logPacketBatchSize)
+    {
+        INodePacket[] packets = [new RawPacket(17), new RawPacket(23)];
+        using CountingWriteStream stream = new();
+
+        await RunWritePump(stream, packets, terminating: terminating, logPacketBatchSize: logPacketBatchSize);
+
+        stream.ToArray().ShouldBe(SerializeFrames(packets));
+        int[] expectedWrites = logPacketBatchSize == "0" ? [17, 23] : [40];
+        stream.Writes.ShouldBe(expectedWrites);
+    }
+
+    [Theory]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, -1, 1)]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, 0, 1)]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, 1, 2)]
+    [InlineData("64", 64, -1, 1)]
+    [InlineData("64", 64, 0, 1)]
+    [InlineData("64", 64, 1, 2)]
+    public async Task LogBatch_RespectsByteBudget(string? logPacketBatchSize, int byteBudget, int adjustment, int expectedWrites)
+    {
+        int firstSize = byteBudget / 2;
+        INodePacket[] packets = [new RawPacket(firstSize), new RawPacket(byteBudget - firstSize + adjustment)];
+        using CountingWriteStream stream = new();
+
+        await RunWritePump(stream, packets, logPacketBatchSize: logPacketBatchSize);
+
+        stream.ToArray().ShouldBe(SerializeFrames(packets));
+        stream.Writes.Count.ShouldBe(expectedWrites);
+        stream.Writes.ShouldAllBe(size => size <= byteBudget);
+    }
+
+    [Theory]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, -1)]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, 0)]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize, 1)]
+    [InlineData("64", 64, -1)]
+    [InlineData("64", 64, 0)]
+    [InlineData("64", 64, 1)]
+    public async Task LogBatch_WritesLargePacketsInOrder(string? logPacketBatchSize, int byteBudget, int adjustment)
+    {
+        int size = byteBudget + adjustment;
+        INodePacket[] packets = [new RawPacket(6), new RawPacket(size), new RawPacket(7)];
+        using CountingWriteStream stream = new();
+
+        await RunWritePump(stream, packets, logPacketBatchSize: logPacketBatchSize);
+
+        stream.ToArray().ShouldBe(SerializeFrames(packets));
+        stream.Writes.ShouldBe([6, size, 7]);
+    }
+
+    [Theory]
+    [InlineData(false, false, null)]
+    [InlineData(true, false, null)]
+    [InlineData(false, true, null)]
+    [InlineData(false, false, "0")]
+    [InlineData(false, false, "32")]
+    public async Task LogBatch_SerializationFailurePreservesPrefixAndBothErrors(bool controlPacket, bool failWrite, string? logPacketBatchSize)
+    {
+        using CountingWriteStream stream = new() { FailWrite = failWrite };
+        RawPacket prefix = new(17);
+        RawPacket failure = new(6, controlPacket ? NodePacketType.NodeBuildComplete : NodePacketType.LogMessage, () =>
+        {
+            if (controlPacket)
+            {
+                stream.ToArray().ShouldBe(SerializeFrames([prefix]));
+            }
+
+            throw new InvalidOperationException("Injected packet serialization failure.");
+        });
+
+        string diagnostic = await RunWritePump(stream, [prefix, failure, new RawPacket(9)], logPacketBatchSize: logPacketBatchSize);
+
+        diagnostic.ShouldContain("Injected packet serialization failure.");
+        stream.Writes.ShouldBe([17]);
+        stream.ToArray().ShouldBe(failWrite ? Array.Empty<byte>() : SerializeFrames([prefix]));
+        if (failWrite)
+        {
+            diagnostic.ShouldContain("Injected pipe write failure.");
+        }
+    }
+
+    [Theory]
+    [InlineData(null, 40, true)]
+    [InlineData("0", 17, false)]
+    [InlineData("32", 17, true)]
+    public async Task LogBatch_WriteFailureFailsLinkWithoutRetry(string? logPacketBatchSize, int failedWriteSize, bool secondPacketSerialized)
+    {
+        using CountingWriteStream stream = new() { FailWrite = true };
+        bool secondPacketWasSerialized = false;
+
+        string diagnostic = await RunWritePump(
+            stream,
+            [new RawPacket(17), new RawPacket(23, beforeTranslate: () => secondPacketWasSerialized = true)],
+            logPacketBatchSize: logPacketBatchSize);
+
+        diagnostic.ShouldContain("Injected pipe write failure.");
+        stream.Writes.ShouldBe([failedWriteSize]);
+        stream.Length.ShouldBe(0);
+        secondPacketWasSerialized.ShouldBe(secondPacketSerialized);
+    }
+
+    [Theory]
+    [InlineData(null, Traits.DefaultLogPacketBatchSize)]
+    [InlineData("", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("malformed", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("-1", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("2147483648", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("1048577", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("2147483647", Traits.DefaultLogPacketBatchSize)]
+    [InlineData("0", 0)]
+    [InlineData("1", 1)]
+    [InlineData("32768", 32768)]
+    [InlineData("1048576", 1048576)]
+    public void LogPacketBatchSize_ValidatesConfiguration(string? configured, int expected)
+    {
+        using TestEnvironment env = TestEnvironment.Create(_output);
+        env.SetEnvironmentVariable("MSBUILDLOGPACKETBATCHSIZE", configured);
+
+        Traits.Instance.LogPacketBatchSize.ShouldBe(expected);
+    }
+
+    [Fact]
+    public async Task LogBatch_CapturesConfiguredBudgetOncePerPump()
+    {
+        INodePacket[] packets =
+        [
+            new RawPacket(6, NodePacketType.NodeBuildComplete),
+            new RawPacket(17),
+            new RawPacket(15)
+        ];
+        using CountingWriteStream stream = new();
+
+        await RunWritePump(
+            stream,
+            packets,
+            logPacketBatchSize: "32",
+            afterFirstWrite: env =>
+            {
+                env.SetEnvironmentVariable("MSBUILDLOGPACKETBATCHSIZE", "0");
+                Traits.Instance.LogPacketBatchSize.ShouldBe(0);
+            });
+
+        stream.ToArray().ShouldBe(SerializeFrames(packets));
+        stream.Writes.ShouldBe([6, 32]);
+    }
+
+    private static LogMessagePacket Log(string message) =>
+        new(new KeyValuePair<int, BuildEventArgs>(0, new BuildMessageEventArgs(message, "", "", MessageImportance.Low)));
+
+    private static byte[] SerializeFrames(INodePacket[] packets, byte version = 0)
+    {
+        using MemoryStream outputStream = new();
+        using MemoryStream packetStream = new();
+        using ITranslator translator = BinaryTranslator.GetWriteTranslator(packetStream);
+        translator.NegotiatedPacketVersion = version;
+        foreach (INodePacket packet in packets)
+        {
+            packetStream.SetLength(0);
+            packetStream.WriteByte((byte)packet.Type);
+            translator.Writer.Write(0);
+            packet.Translate(translator);
+            int length = (int)packetStream.Position;
+            packetStream.Position = 1;
+            translator.Writer.Write(length - 5);
+            outputStream.Write(packetStream.GetBuffer(), 0, length);
+        }
+
+        return outputStream.ToArray();
+    }
+
+    private async Task<string> RunWritePump(
+        CountingWriteStream stream,
+        INodePacket[] packets,
+        byte version = 0,
+        bool terminating = false,
+        string? logPacketBatchSize = null,
+        Action<TestEnvironment>? afterFirstWrite = null)
+    {
+        using TestEnvironment env = TestEnvironment.Create(_output);
+        env.SetEnvironmentVariable("MSBUILDLOGPACKETBATCHSIZE", logPacketBatchSize);
+        using MemoryStream packetStream = new();
+        using BinaryWriter writer = new(packetStream);
+        using AutoResetEvent available = new(!terminating);
+        using AutoResetEvent terminate = new(terminating);
+        TestEndpoint endpoint = new();
+        // Initialize only serialization state, without opening a real named pipe.
+        SetField("_packetStream", packetStream);
+        SetField("_binaryWriter", writer);
+        SetField("_negotiatedWriteVersion", version);
+        stream.AfterWrite = () =>
+        {
+            Action<TestEnvironment>? callback = Interlocked.Exchange(ref afterFirstWrite, null);
+            callback?.Invoke(env);
+            terminate.Set();
+        };
+        DebugUtils.ResetDebugDumpPathInRunningTests = true;
+        _ = DebugUtils.DebugDumpPath;
+
+        Task pump = Task.Run(() => endpoint.RunReadLoop(stream, new ConcurrentQueue<INodePacket>(packets), available, terminate));
+        using CancellationTokenSource timeout = new();
+        Task completed = await Task.WhenAny(pump, Task.Delay(TimeSpan.FromSeconds(30), timeout.Token));
+        terminate.Set();
+        await pump;
+        timeout.Cancel();
+        completed.ShouldBe(pump, "the queue-empty flush should not wait for another packet or termination");
+        endpoint.LinkStatus.ShouldBe(LinkStatus.Failed);
+
+        string diagnostic = "";
+        if (DebugUtils.DumpFilePath is not null)
+        {
+            diagnostic = File.ReadAllText(DebugUtils.DumpFilePath);
+            env.WithTransientTestState(new TransientTestFile(DebugUtils.DebugDumpPath, Path.GetFileName(DebugUtils.DumpFilePath), diagnostic));
+            DebugUtils.ResetDebugDumpPathInRunningTests = true;
+        }
+
+        return diagnostic;
+
+        void SetField(string name, object value) =>
+            typeof(NodeEndpointOutOfProcBase).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.SetValue(endpoint, value);
+    }
+
+    private sealed class RawPacket(int frameSize, NodePacketType type = NodePacketType.LogMessage, Action? beforeTranslate = null) : INodePacket
+    {
+        private readonly byte[] _payload = new byte[frameSize - 5];
+        public NodePacketType Type => type;
+
+        public void Translate(ITranslator translator)
+        {
+            translator.Writer.BaseStream.Position.ShouldBe(5);
+            beforeTranslate?.Invoke();
+            translator.Writer.Write(_payload);
+        }
+    }
+
+    private sealed class CountingWriteStream : MemoryStream
+    {
+        private readonly TaskCompletionSource<int> _read = new();
+        public List<int> Writes { get; } = [];
+        public bool FailWrite { get; init; }
+        public Action? AfterWrite { get; set; }
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => _read.Task;
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            Writes.Add(count);
+            if (FailWrite)
+            {
+                throw new IOException("Injected pipe write failure.");
+            }
+
+            base.Write(buffer, offset, count);
+            AfterWrite?.Invoke();
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            _read.TrySetResult(0);
+            base.Dispose(disposing);
+        }
+    }
+
     private sealed class TestEndpoint : NodeEndpointOutOfProcBase
     {
         public TestEndpoint() => ChangeLinkStatus(LinkStatus.Active);

src/Framework/Traits.cs

@@ -148,6 +148,14 @@ public Traits()
         /// </summary>
         public readonly int NodeConnectionBufferSize = GetNodeConnectionBufferSize();
 
+        /// <summary>
+        /// Maximum combined size in bytes of outbound logging packets in one pipe write. Defaults to 64 KB;
+        /// zero disables batching. Values from zero through 1 MB are accepted, and invalid values use the default.
+        /// </summary>
+        internal const int DefaultLogPacketBatchSize = 64 * 1024;
+
+        public readonly int LogPacketBatchSize = GetLogPacketBatchSize();
+
         private static int GetNodeConnectionBufferSize()
         {
             int configured = EnvironmentUtilities.GetValueAsInt32OrDefault("MSBUILDNODECONNECTIONBUFFERSIZE", -1);
@@ -161,6 +169,12 @@ private static int GetNodeConnectionBufferSize()
             return ChangeWaves.AreFeaturesEnabled(ChangeWaves.Wave18_9) ? DefaultBufferSize : LegacyBufferSize;
         }
 
+        private static int GetLogPacketBatchSize()
+        {
+            int configured = EnvironmentUtilities.GetValueAsInt32OrDefault("MSBUILDLOGPACKETBATCHSIZE", DefaultLogPacketBatchSize);
+            return configured is >= 0 and <= 1024 * 1024 ? configured : DefaultLogPacketBatchSize;
+        }
+
         /// <summary>
         /// Launches a persistent RAR process.
         /// </summary>