Refactor the dmesg collection logic to use overlapped IO and run in a single thread (#40766)
* Save state * Save state * Save state * Save state * Save state * Prepare for PR * Fix test hang * Apply PR suggestions * Apply PR feedback * Fix typo * Connect named pipe by default * Handle Connect correctly * Format * Apply PR feedback
Blue committed
Jun 11, 2026 at 17:30 UTC
3a2de0f0a694f3efce937e4fecfb9260fd451ac3
11 files changed
+764
-168
src/windows/common/Dmesg.cpp
+101
-117
@@ -15,56 +15,44 @@ Abstract:
15
#include "precomp.h"
16
#include "Dmesg.h"
17
18
+using wsl::windows::common::io::EventHandle;
19
+using wsl::windows::common::io::HandleWrapper;
20
+using wsl::windows::common::io::MultiHandleWait;
21
+using wsl::windows::common::io::ReadNamedPipe;
22
+using wsl::windows::common::io::WriteHandle;
23
+using wsl::windows::common::io::WriteNamedPipe;
24
+
25
DmesgCollector::DmesgCollector(
19
- GUID VmId, const wil::unique_event& ExitEvent, bool EnableTelemetry, bool EnableDebugConsole, const std::wstring& Com1PipeName, wil::unique_handle&& OutputHandle) :
26
+ GUID VmId, HANDLE ExitEvent, bool EnableTelemetry, bool EnableDebugConsole, const std::wstring& Com1PipeName, wil::unique_handle&& OutputHandle) :
27
m_com1PipeName(Com1PipeName),
28
+ m_vmExitEvent(ExitEvent),
29
+ m_outputHandle(std::move(OutputHandle)),
30
m_runtimeId(VmId),
31
m_debugConsole(EnableDebugConsole),
23
- m_telemetry(EnableTelemetry),
24
- m_outputHandle(std::move(OutputHandle))
32
+ m_telemetry(EnableTelemetry)
33
{
26
- m_exitEvent.reset(wsl::windows::common::wslutil::DuplicateHandle(ExitEvent.get()));
27
- m_overlappedEvent.create(wil::EventOptions::ManualReset);
28
- m_overlapped.hEvent = m_overlappedEvent.get();
29
- m_threadExit.create(wil::EventOptions::ManualReset);
30
- m_exitEvents = {m_threadExit.get(), m_exitEvent.get()};
34
}
35
36
DmesgCollector::~DmesgCollector()
37
{
35
- m_threadExit.SetEvent();
36
- if (m_earlyConsoleWorker.joinable())
38
+ m_threadExitEvent.SetEvent();
39
+ if (m_thread.joinable())
40
{
38
- m_earlyConsoleWorker.join();
39
- }
40
-
41
- if (m_virtioWorker.joinable())
42
- {
43
- m_virtioWorker.join();
41
+ m_thread.join();
42
}
43
}
44
45
std::shared_ptr<DmesgCollector> DmesgCollector::Create(
48
- GUID VmId,
49
- const wil::unique_event& ExitEvent,
50
- bool EnableTelemetry,
51
- bool EnableDebugConsole,
52
- const std::wstring& Com1PipeName,
53
- bool EnableEarlyBootConsole,
54
- wil::unique_handle&& OutputHandle)
46
+ GUID VmId, HANDLE ExitEvent, bool EnableTelemetry, bool EnableDebugConsole, const std::wstring& Com1PipeName, bool EnableEarlyBootConsole, wil::unique_handle&& OutputHandle)
47
{
48
auto dmesgCollector = std::shared_ptr<DmesgCollector>(
49
new DmesgCollector(VmId, ExitEvent, EnableTelemetry, EnableDebugConsole, Com1PipeName, std::move(OutputHandle)));
50
59
- if (FAILED(dmesgCollector->Start(EnableEarlyBootConsole)))
60
- {
61
- return {};
62
- }
63
-
51
+ dmesgCollector->Start(EnableEarlyBootConsole);
52
return dmesgCollector;
53
}
54
67
-std::pair<std::wstring, std::thread> DmesgCollector::StartDmesgThread(InputSource Source)
55
+std::pair<std::wstring, wil::unique_hfile> DmesgCollector::CreateConsolePipe()
56
{
57
std::wstring pipeName = wsl::windows::common::helpers::GetUniquePipeName();
58
wil::unique_hfile pipe(CreateNamedPipeW(
@@ -72,46 +60,89 @@ std::pair<std::wstring, std::thread> DmesgCollector::StartDmesgThread(InputSourc
60
61
THROW_LAST_ERROR_IF(!pipe);
62
75
- auto workerThread = std::thread([this, Source, Pipe = std::move(pipe)]() {
76
- try
77
- {
78
- wsl::windows::common::wslutil::SetThreadDescription(L"Dmesg");
63
+ return {std::move(pipeName), std::move(pipe)};
64
+}
65
80
- // When the pipe connects, start reading data.
81
- wsl::windows::common::helpers::ConnectPipe(Pipe.get(), INFINITE, m_exitEvents);
66
+void DmesgCollector::Run()
67
+try
68
+{
69
+ wsl::windows::common::wslutil::SetThreadDescription(L"Dmesg");
70
83
- std::vector<char> buffer(LX_RELAY_BUFFER_SIZE);
84
- const auto allBuffer = gsl::make_span(buffer);
85
- OVERLAPPED overlapped = {};
86
- const wil::unique_event overlappedEvent(wil::EventOptions::ManualReset);
87
- overlapped.hEvent = overlappedEvent.get();
88
- for (;;)
89
- {
90
- overlappedEvent.ResetEvent();
91
- const auto bytesRead = wsl::windows::common::relay::InterruptableRead(
92
- Pipe.get(), gslhelpers::convert_span<gsl::byte>(allBuffer), m_exitEvents, &overlapped);
71
+ MultiHandleWait io;
72
94
- if (bytesRead == 0)
95
- {
96
- break;
97
- }
73
+ if (m_earlyConsolePipe)
74
+ {
75
+ io.AddHandle(
76
+ std::make_unique<ReadNamedPipe>(
77
+ HandleWrapper{std::move(m_earlyConsolePipe)},
78
+ [this](const gsl::span<char>& Input) { ProcessInput(DmesgCollectorEarlyConsole, Input); }),
79
+ MultiHandleWait::IgnoreErrors);
80
+ }
81
99
- auto validBuffer = allBuffer.subspan(0, bytesRead);
100
- ProcessInput(Source, validBuffer);
101
- }
102
- }
103
- catch (...)
104
- {
105
- auto error = wil::ResultFromCaughtException();
106
- LOG_HR_IF(error, error != E_ABORT); // E_ABORT is expected during shutdown.
107
- }
108
- });
82
+ io.AddHandle(
83
+ std::make_unique<ReadNamedPipe>(
84
+ HandleWrapper{std::move(m_virtioConsolePipe)},
85
+ [this](const gsl::span<char>& Input) { ProcessInput(DmesgCollectorConsole, Input); }),
86
+ MultiHandleWait::IgnoreErrors);
87
+
88
+ if (m_outputHandle)
89
+ {
90
+ auto output = std::make_unique<WriteHandle>(
91
+ HandleWrapper{std::move(m_outputHandle), [this]() { m_outputWrite = nullptr; }}, std::vector<char>{}, false);
92
+ m_outputWrite = output.get();
93
+ io.AddHandle(std::move(output), MultiHandleWait::IgnoreErrors);
94
+ }
95
+
96
+ if (m_com1Pipe)
97
+ {
98
+ const bool reconnect = m_pipeServer && !m_debugConsole;
99
+
100
+ auto com1 = std::make_unique<WriteNamedPipe>(
101
+ HandleWrapper{std::move(m_com1Pipe), [this]() { m_com1Write = nullptr; }}, reconnect, !m_pipeServer);
102
+ m_com1Write = com1.get();
103
+ io.AddHandle(std::move(com1), MultiHandleWait::IgnoreErrors);
104
+ }
105
+
106
+ // The loop runs until either exit event is signaled.
107
+ io.AddHandle(std::make_unique<EventHandle>(m_threadExitEvent.get()), MultiHandleWait::CancelOnCompleted);
108
+ io.AddHandle(std::make_unique<EventHandle>(m_vmExitEvent), MultiHandleWait::CancelOnCompleted);
109
+
110
+ io.Run({});
111
+}
112
+CATCH_LOG()
113
+
114
+namespace {
115
+
116
+template <typename TWriter>
117
+void Push(TWriter& Writer, const gsl::span<char>& Input, const char* Target)
118
+{
119
+ constexpr size_t c_maxDmesgPendingBytes = 1024 * 1024;
120
+
121
+ const auto pending = Writer.PendingBytes();
122
+
123
+ // Don't fill the buffer past c_maxDmesgPendingBytes. If full, just drop the bytes with a warning.
124
+ if (pending + Input.size() > c_maxDmesgPendingBytes)
125
+ {
126
+ WSL_LOG(
127
+ "DmesgOutputDropped",
128
+ TraceLoggingValue(Target, "target"),
129
+ TraceLoggingValue(static_cast<uint64_t>(pending), "pendingBytes"));
130
110
- return std::pair{std::move(pipeName), std::move(workerThread)};
131
+ return;
132
+ }
133
+
134
+ Writer.Push(Input);
135
}
136
137
+} // namespace
138
+
139
void DmesgCollector::ProcessInput(InputSource Source, const gsl::span<char>& Input)
140
{
141
+ if (Input.empty())
142
+ {
143
+ return;
144
+ }
145
+
146
RingBuffer* ringBuffer = nullptr;
147
bool sendToComPipe = m_debugConsole;
148
if (Source == DmesgCollectorEarlyConsole)
@@ -122,7 +153,7 @@ void DmesgCollector::ProcessInput(InputSource Source, const gsl::span<char>& Inp
153
}
154
else
155
{
125
- sendToComPipe = !m_debugConsole && m_com1Pipe;
156
+ sendToComPipe = !m_debugConsole;
157
}
158
}
159
else
@@ -153,63 +184,18 @@ void DmesgCollector::ProcessInput(InputSource Source, const gsl::span<char>& Inp
184
}
185
}
186
156
- if (sendToComPipe)
187
+ if (sendToComPipe && m_com1Write)
188
{
158
- WriteToCom1(Input);
189
+ Push(*m_com1Write, Input, "com1");
190
}
191
161
- if (m_outputHandle != nullptr)
192
+ if (m_outputWrite != nullptr)
193
{
163
- m_overlappedEvent.ResetEvent();
164
- if (wsl::windows::common::relay::InterruptableWrite(
165
- m_outputHandle.get(), gslhelpers::convert_span<gsl::byte>(Input), m_exitEvents, &m_overlapped) == 0)
166
- {
167
- m_outputHandle = nullptr;
168
- }
194
+ Push(*m_outputWrite, Input, "output");
195
}
196
}
197
172
-void DmesgCollector::WriteToCom1(const gsl::span<char>& Input)
173
-{
174
- auto lock = m_lock.lock_exclusive();
175
- if (!m_com1Pipe)
176
- {
177
- return;
178
- }
179
-
180
- // If this is not writing to the debug console, emulate the normal
181
- // serial pipe behavior of waiting for a pipe connection.
182
- if (m_waitForConnection)
183
- {
184
- if (FAILED(wil::ResultFromException(
185
- [&]() { wsl::windows::common::helpers::ConnectPipe(m_com1Pipe.get(), INFINITE, m_exitEvents); })))
186
- {
187
- return;
188
- }
189
-
190
- m_waitForConnection = false;
191
- }
192
-
193
- m_overlappedEvent.ResetEvent();
194
- const auto buffer = gslhelpers::convert_span<gsl::byte>(Input);
195
- if (wsl::windows::common::relay::InterruptableWrite(m_com1Pipe.get(), buffer, m_exitEvents, &m_overlapped) == 0)
196
- {
197
- if (m_debugConsole || !m_pipeServer)
198
- {
199
- // A disconnect from the debug console, or from a pipe that was acting as the server, doesn't have any
200
- // reconnect mechanism, so don't try to write anymore bytes.
201
- m_com1Pipe.reset();
202
- }
203
- else
204
- {
205
- // Emulate the normal serial behavior of waiting for a pipe connection to write.
206
- m_waitForConnection = true;
207
- }
208
- }
209
-}
210
-
211
-HRESULT DmesgCollector::Start(bool EnableEarlyBootConsole)
212
-try
198
+void DmesgCollector::Start(bool EnableEarlyBootConsole)
199
{
200
if (!m_com1PipeName.empty())
201
{
@@ -232,8 +218,6 @@ try
218
if (m_com1Pipe)
219
{
220
m_pipeServer = true;
235
- // If the debug console is not active, may have to wait for a connection.
236
- m_waitForConnection = !m_debugConsole;
221
}
222
}
223
@@ -242,10 +226,10 @@ try
226
227
if (EnableEarlyBootConsole)
228
{
245
- std::tie(m_earlyConsoleName, m_earlyConsoleWorker) = StartDmesgThread(DmesgCollectorEarlyConsole);
229
+ std::tie(m_earlyConsoleName, m_earlyConsolePipe) = CreateConsolePipe();
230
}
231
248
- std::tie(m_virtioConsoleName, m_virtioWorker) = StartDmesgThread(DmesgCollectorConsole);
249
- return S_OK;
232
+ std::tie(m_virtioConsoleName, m_virtioConsolePipe) = CreateConsolePipe();
233
+
234
+ m_thread = std::thread([this]() { Run(); });
235
}
251
-CATCH_RETURN()
src/windows/common/Dmesg.h
+21
-23
@@ -15,6 +15,7 @@ Abstract:
15
#pragma once
16
17
#include "relay.hpp"
18
+#include "HandleIO.h"
19
#include "RingBuffer.h"
20
21
class DmesgCollector
@@ -35,7 +36,7 @@ public:
36
37
static std::shared_ptr<DmesgCollector> Create(
38
GUID VmId,
38
- const wil::unique_event& ExitEvent,
39
+ HANDLE ExitEvent,
40
bool EnableTelemetry,
41
bool EnableDebugConsole,
42
const std::wstring& Com1PipeName,
@@ -49,38 +50,35 @@ private:
50
DmesgCollectorConsole
51
};
52
52
- DmesgCollector(
53
- GUID VmId,
54
- const wil::unique_event& ExitEvent,
55
- bool EnableTelemetry,
56
- bool EnableDebugConsole,
57
- const std::wstring& Com1PipeName,
58
- wil::unique_handle&& OutputHandle = {});
53
+ DmesgCollector(GUID VmId, HANDLE ExitEvent, bool EnableTelemetry, bool EnableDebugConsole, const std::wstring& Com1PipeName, wil::unique_handle&& OutputHandle = {});
54
+
55
+ void Start(bool EnableEarlyBootConsole);
56
+
57
+ void Run();
58
+
59
+ static std::pair<std::wstring, wil::unique_hfile> CreateConsolePipe();
60
60
- HRESULT Start(bool EnableEarlyBootConsole);
61
- std::pair<std::wstring, std::thread> StartDmesgThread(InputSource Source);
61
void ProcessInput(InputSource Source, const gsl::span<char>& Input);
63
- void WriteToCom1(const gsl::span<char>& Input);
62
65
- wil::srwlock m_lock;
63
std::wstring m_com1PipeName;
64
std::wstring m_earlyConsoleName;
65
std::wstring m_virtioConsoleName;
69
- wil::unique_event m_exitEvent;
70
- wil::unique_event m_threadExit;
71
- std::vector<HANDLE> m_exitEvents;
66
+ HANDLE m_vmExitEvent;
67
+ wil::unique_event m_threadExitEvent{wil::EventOptions::ManualReset};
68
wil::unique_hfile m_com1Pipe;
69
+ wil::unique_handle m_outputHandle;
70
+ wil::unique_hfile m_earlyConsolePipe;
71
+ wil::unique_hfile m_virtioConsolePipe;
72
GUID m_runtimeId{};
74
- wil::unique_event m_overlappedEvent;
75
- _Guarded_by_(m_lock) OVERLAPPED m_overlapped {};
73
RingBuffer m_dmesgBuffer{LX_RELAY_BUFFER_SIZE};
74
RingBuffer m_dmesgEarlyBuffer{LX_RELAY_BUFFER_SIZE};
75
bool m_debugConsole;
76
bool m_telemetry;
80
- std::atomic<bool> m_earlyConsoleTransition = false;
81
- bool m_pipeServer;
82
- bool m_waitForConnection;
83
- std::thread m_earlyConsoleWorker;
84
- std::thread m_virtioWorker;
85
- wil::unique_handle m_outputHandle = nullptr;
77
+ bool m_earlyConsoleTransition = false;
78
+ bool m_pipeServer = false;
79
+
80
+ std::thread m_thread;
81
+
82
+ wsl::windows::common::io::WriteHandle* m_outputWrite = nullptr;
83
+ wsl::windows::common::io::WriteNamedPipe* m_com1Write = nullptr;
84
};
src/windows/common/HandleIO.cpp
+246
-13
@@ -14,9 +14,11 @@ using wsl::windows::common::io::LineBasedReadHandle;
14
using wsl::windows::common::io::MultiHandleWait;
15
using wsl::windows::common::io::OverlappedIOHandle;
16
using wsl::windows::common::io::ReadHandle;
17
+using wsl::windows::common::io::ReadNamedPipe;
18
using wsl::windows::common::io::ReadSocketMessageHandle;
19
using wsl::windows::common::io::SingleAcceptHandle;
20
using wsl::windows::common::io::WriteHandle;
21
+using wsl::windows::common::io::WriteNamedPipe;
22
23
namespace {
24
@@ -304,6 +306,60 @@ HANDLE ReadHandle::GetHandle() const
306
return Event.get();
307
}
308
309
+// ReadNamedPipe
310
+
311
+ReadNamedPipe::ReadNamedPipe(HandleWrapper&& Pipe, std::function<void(const gsl::span<char>& Buffer)>&& OnRead) :
312
+ ReadHandle(std::move(Pipe), std::move(OnRead))
313
+{
314
+}
315
+
316
+void ReadNamedPipe::Schedule()
317
+{
318
+ if (!m_connected)
319
+ {
320
+ WI_ASSERT(State == IOHandleStatus::Standby);
321
+
322
+ if (!ConnectNamedPipe(Handle.Get(), &Overlapped))
323
+ {
324
+ const auto error = GetLastError();
325
+ if (error == ERROR_IO_PENDING)
326
+ {
327
+ State = IOHandleStatus::Pending;
328
+ return;
329
+ }
330
+
331
+ THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_PIPE_CONNECTED, "Handle: 0x%p", (void*)Handle.Get());
332
+ }
333
+
334
+ m_connected = true;
335
+ }
336
+
337
+ ReadHandle::Schedule();
338
+}
339
+
340
+void ReadNamedPipe::Collect()
341
+{
342
+ if (!m_connected)
343
+ {
344
+ WI_ASSERT(State == IOHandleStatus::Pending);
345
+
346
+ DWORD bytes{};
347
+ if (!GetOverlappedResult(Handle.Get(), &Overlapped, &bytes, FALSE))
348
+ {
349
+ const auto error = GetLastError();
350
+ THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_PIPE_CONNECTED, "Handle: 0x%p", (void*)Handle.Get());
351
+ }
352
+
353
+ m_connected = true;
354
+
355
+ // Transition back to standby so the IO loop schedules the first read.
356
+ State = IOHandleStatus::Standby;
357
+ return;
358
+ }
359
+
360
+ ReadHandle::Collect();
361
+}
362
+
363
// SingleAcceptHandle
364
365
SingleAcceptHandle::SingleAcceptHandle(HandleWrapper&& ListenSocket, HandleWrapper&& AcceptedSocket, std::function<void()>&& OnAccepted) :
@@ -719,11 +775,20 @@ HANDLE ReadSocketMessageHandle::GetHandle() const
775
776
// WriteHandle
777
722
-WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, const std::vector<char>& Source) :
723
- Handle(std::move(MovedHandle)), Buffer(Source.size()), Offset(InitializeFileOffset(Handle.Get()))
778
+WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, const std::vector<char>& Source, bool CompleteOnDrained) :
779
+ Handle(std::move(MovedHandle)), Buffer(Source.size()), Offset(InitializeFileOffset(Handle.Get())), CompleteOnDrained(CompleteOnDrained)
780
{
725
- std::memcpy(Buffer.Span().data(), Source.data(), Source.size());
781
+ if (!Source.empty())
782
+ {
783
+ std::memcpy(Buffer.Span().data(), Source.data(), Source.size());
784
+ }
785
+
786
Overlapped.hEvent = Event.get();
787
+
788
+ if (!CompleteOnDrained && Buffer.Size() == 0)
789
+ {
790
+ State = IOHandleStatus::Idle;
791
+ }
792
}
793
794
WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, gsl::span<gsl::byte> Source) :
@@ -740,10 +805,37 @@ WriteHandle::~WriteHandle()
805
}
806
}
807
808
+void WriteHandle::SetCompleteOnDrained(bool Value)
809
+{
810
+ CompleteOnDrained = Value;
811
+}
812
+
813
+IOHandleStatus WriteHandle::DrainedState() const
814
+{
815
+ if (CompleteOnDrained)
816
+ {
817
+ return IOHandleStatus::Completed;
818
+ }
819
+
820
+ return Pending.empty() ? IOHandleStatus::Idle : IOHandleStatus::Standby;
821
+}
822
+
823
void WriteHandle::Schedule()
824
{
825
WI_ASSERT(State == IOHandleStatus::Standby);
826
827
+ if (!Pending.empty())
828
+ {
829
+ Buffer.Append(gsl::make_span(Pending));
830
+ Pending.clear();
831
+ }
832
+
833
+ if (Buffer.Size() == 0)
834
+ {
835
+ State = DrainedState();
836
+ return;
837
+ }
838
+
839
Event.ResetEvent();
840
841
Overlapped.Offset = Offset.LowPart;
@@ -759,7 +851,7 @@ void WriteHandle::Schedule()
851
Buffer.Consume(bytesWritten);
852
if (Buffer.Size() == 0)
853
{
762
- State = IOHandleStatus::Completed;
854
+ State = DrainedState();
855
}
856
}
857
else
@@ -787,20 +879,26 @@ void WriteHandle::Collect()
879
Buffer.Consume(bytesWritten);
880
if (Buffer.Size() == 0)
881
{
790
- State = IOHandleStatus::Completed;
882
+ State = DrainedState();
883
}
884
}
885
886
void WriteHandle::Push(const gsl::span<char>& Content)
887
{
796
- // Don't write if a WriteFile() is pending, since that could cause the buffer to reallocate.
797
- WI_ASSERT(State == IOHandleStatus::Standby || State == IOHandleStatus::Completed);
888
WI_ASSERT(!Content.empty());
889
800
- // Resize() throws E_UNEXPECTED if Buffer does not own its storage.
801
- Buffer.Append(Content);
890
+ // Put any pending output to a different buffer, since the active buffer could be in the middle of a write.
891
+ Pending.insert(Pending.end(), Content.begin(), Content.end());
892
803
- State = IOHandleStatus::Standby;
893
+ if (State == IOHandleStatus::Idle)
894
+ {
895
+ State = IOHandleStatus::Standby;
896
+ }
897
+}
898
+
899
+size_t WriteHandle::PendingBytes() const
900
+{
901
+ return Pending.size() + Buffer.Size();
902
}
903
904
HANDLE WriteHandle::GetHandle() const
@@ -808,10 +906,138 @@ HANDLE WriteHandle::GetHandle() const
906
return Event.get();
907
}
908
909
+WriteNamedPipe::WriteNamedPipe(HandleWrapper&& MovedPipe, bool Reconnect, bool Connected) :
910
+ Pipe(std::move(MovedPipe)), ReconnectOnFailure(Reconnect), NeedConnect(!Connected)
911
+{
912
+ ConnectOverlapped.hEvent = ConnectEvent.get();
913
+
914
+ Write.emplace(HandleWrapper{Pipe.Get()}, std::vector<char>{}, false);
915
+
916
+ State = IOHandleStatus::Idle;
917
+}
918
+
919
+WriteNamedPipe::~WriteNamedPipe()
920
+{
921
+ if (Connecting)
922
+ {
923
+ CancelPendingIo(Pipe.Get(), ConnectOverlapped);
924
+ }
925
+}
926
+
927
+void WriteNamedPipe::Reconnect()
928
+{
929
+ // Drop the disconnected client so a new one can connect, and retry the buffered data once reconnected.
930
+ LOG_IF_WIN32_BOOL_FALSE(DisconnectNamedPipe(Pipe.Get()));
931
+
932
+ NeedConnect = true;
933
+ State = IOHandleStatus::Standby;
934
+}
935
+
936
+void WriteNamedPipe::Schedule()
937
+{
938
+ WI_ASSERT(State == IOHandleStatus::Standby);
939
+
940
+ if (NeedConnect)
941
+ {
942
+ ConnectEvent.ResetEvent();
943
+ ConnectOverlapped.Offset = 0;
944
+ ConnectOverlapped.OffsetHigh = 0;
945
+
946
+ if (!ConnectNamedPipe(Pipe.Get(), &ConnectOverlapped))
947
+ {
948
+ const auto error = GetLastError();
949
+ if (error == ERROR_IO_PENDING)
950
+ {
951
+ Connecting = true;
952
+ State = IOHandleStatus::Pending;
953
+ return;
954
+ }
955
+
956
+ THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_PIPE_CONNECTED, "Handle: 0x%p", (void*)Pipe.Get());
957
+ }
958
+
959
+ NeedConnect = false;
960
+ }
961
+
962
+ try
963
+ {
964
+ Write->Schedule();
965
+ State = Write->GetState();
966
+ }
967
+ catch (...)
968
+ {
969
+ if (!ReconnectOnFailure)
970
+ {
971
+ throw;
972
+ }
973
+
974
+ LOG_CAUGHT_EXCEPTION();
975
+ Reconnect();
976
+ }
977
+}
978
+
979
+void WriteNamedPipe::Collect()
980
+{
981
+ WI_ASSERT(State == IOHandleStatus::Pending);
982
+
983
+ // Complete a pending connection, then let the loop schedule the first write.
984
+ if (Connecting)
985
+ {
986
+ Connecting = false;
987
+
988
+ DWORD bytes{};
989
+ if (!GetOverlappedResult(Pipe.Get(), &ConnectOverlapped, &bytes, FALSE))
990
+ {
991
+ const auto error = GetLastError();
992
+ THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_PIPE_CONNECTED, "Handle: 0x%p", (void*)Pipe.Get());
993
+ }
994
+
995
+ NeedConnect = false;
996
+ State = IOHandleStatus::Standby;
997
+ return;
998
+ }
999
+
1000
+ try
1001
+ {
1002
+ Write->Collect();
1003
+ State = Write->GetState();
1004
+ }
1005
+ catch (...)
1006
+ {
1007
+ if (!ReconnectOnFailure)
1008
+ {
1009
+ throw;
1010
+ }
1011
+
1012
+ LOG_CAUGHT_EXCEPTION();
1013
+ Reconnect();
1014
+ }
1015
+}
1016
+
1017
+HANDLE WriteNamedPipe::GetHandle() const
1018
+{
1019
+ return Connecting ? ConnectEvent.get() : Write->GetHandle();
1020
+}
1021
+
1022
+void WriteNamedPipe::Push(const gsl::span<char>& Content)
1023
+{
1024
+ Write->Push(Content);
1025
+
1026
+ if (State == IOHandleStatus::Idle)
1027
+ {
1028
+ State = IOHandleStatus::Standby;
1029
+ }
1030
+}
1031
+
1032
+size_t WriteNamedPipe::PendingBytes() const
1033
+{
1034
+ return Write ? Write->PendingBytes() : 0;
1035
+}
1036
+
1037
// DockerIORelayHandle
1038
1039
DockerIORelayHandle::DockerIORelayHandle(HandleWrapper&& ReadHandle, HandleWrapper&& Stdout, HandleWrapper&& Stderr, Format ReadFormat) :
814
- WriteStdout(std::move(Stdout)), WriteStderr(std::move(Stderr))
1040
+ WriteStdout(std::move(Stdout), {}, false), WriteStderr(std::move(Stderr), {}, false)
1041
{
1042
if (ReadFormat == Format::HttpChunked)
1043
{
@@ -850,7 +1076,7 @@ void DockerIORelayHandle::Schedule()
1076
{
1077
State = IOHandleStatus::Pending;
1078
}
853
- else if (ActiveHandle->GetState() == IOHandleStatus::Completed)
1079
+ else if (ActiveHandle->GetState() == IOHandleStatus::Completed || ActiveHandle->GetState() == IOHandleStatus::Idle)
1080
{
1081
if (RemainingBytes == 0)
1082
{
@@ -893,7 +1119,7 @@ void DockerIORelayHandle::Collect()
1119
// If the write is completed, switch back to reading.
1120
if (RemainingBytes == 0)
1121
{
896
- if (ActiveHandle->GetState() == IOHandleStatus::Completed)
1122
+ if (ActiveHandle->GetState() == IOHandleStatus::Completed || ActiveHandle->GetState() == IOHandleStatus::Idle)
1123
{
1124
ActiveHandle = nullptr;
1125
@@ -1098,6 +1324,13 @@ bool MultiHandleWait::Run(std::optional<std::chrono::milliseconds> Timeout)
1324
continue;
1325
}
1326
1327
+ // N.B. An Idle handle cannot be waited for since it's not doing any IO.
1328
+ if (entry.Handle->GetState() == IOHandleStatus::Idle)
1329
+ {
1330
+ ++it;
1331
+ continue;
1332
+ }
1333
+
1334
auto& callback = callbacks.emplace_back();
1335
1336
THROW_IF_WIN32_BOOL_FALSE(RegisterWaitForSingleObject(
src/windows/common/HandleIO.h
+90
-9
@@ -3,6 +3,7 @@
3
#pragma once
4
5
#include <concurrent_queue.h>
6
+#include <list>
7
8
#define LX_RELAY_BUFFER_SIZE 0x1000
9
@@ -12,7 +13,10 @@ enum class IOHandleStatus
13
{
14
Standby,
15
Pending,
15
- Completed
16
+ Completed,
17
+ // A persistent handle with no work to do. It stays registered with the MultiHandleWait loop but is not
18
+ // waited on; it returns to Standby once more data is pushed into it.
19
+ Idle
20
};
21
22
struct HandleWrapper
@@ -106,15 +110,35 @@ public:
110
void Collect() override;
111
HANDLE GetHandle() const override;
112
109
-private:
113
+protected:
114
HandleWrapper Handle;
115
+ OVERLAPPED Overlapped{};
116
+
117
+private:
118
std::function<void(const gsl::span<char>& Buffer)> OnRead;
119
wil::unique_event Event{wil::EventOptions::ManualReset};
113
- OVERLAPPED Overlapped{};
120
BufferWrapper Buffer{LX_RELAY_BUFFER_SIZE};
121
LARGE_INTEGER Offset{};
122
};
123
124
+// A ReadHandle for a server named pipe. It waits for a client to connect (ConnectNamedPipe) before
125
+// reading, so it can be scheduled directly with a freshly created server pipe. The connect reuses the
126
+// base read handle's overlapped and event, so the base destructor cancels a pending connect correctly.
127
+class ReadNamedPipe : public ReadHandle
128
+{
129
+public:
130
+ NON_COPYABLE(ReadNamedPipe);
131
+ NON_MOVABLE(ReadNamedPipe);
132
+
133
+ ReadNamedPipe(HandleWrapper&& Pipe, std::function<void(const gsl::span<char>& Buffer)>&& OnRead);
134
+
135
+ void Schedule() override;
136
+ void Collect() override;
137
+
138
+private:
139
+ bool m_connected = false;
140
+};
141
+
142
class SingleAcceptHandle : public OverlappedIOHandle
143
{
144
public:
@@ -211,20 +235,64 @@ public:
235
NON_COPYABLE(WriteHandle);
236
NON_MOVABLE(WriteHandle);
237
214
- WriteHandle(HandleWrapper&& Handle, const std::vector<char>& Buffer = {});
215
- WriteHandle(HandleWrapper&& Handle, gsl::span<gsl::byte> Span);
238
+ WriteHandle(HandleWrapper&& Handle, const std::vector<char>& Source = {}, bool CompleteOnDrained = true);
239
+ WriteHandle(HandleWrapper&& Handle, gsl::span<gsl::byte> Source);
240
~WriteHandle();
241
void Schedule() override;
242
void Collect() override;
243
HANDLE GetHandle() const override;
244
void Push(const gsl::span<char>& Buffer);
245
246
+ // Controls whether the writer completes (and is removed from the loop) or becomes Idle once its buffer drains.
247
+ void SetCompleteOnDrained(bool CompleteOnDrained);
248
+
249
+ // Returns the number of bytes that have been queued for writing but not yet written to the handle.
250
+ size_t PendingBytes() const;
251
+
252
private:
253
+ // Returns the state to adopt once the active buffer drains: Completed for one-shot writers, or Idle/Standby
254
+ // for reusable writers depending on whether more data is queued.
255
+ IOHandleStatus DrainedState() const;
256
+
257
HandleWrapper Handle;
258
wil::unique_event Event{wil::EventOptions::ManualReset};
259
OVERLAPPED Overlapped{};
260
BufferWrapper Buffer;
261
LARGE_INTEGER Offset{};
262
+ bool CompleteOnDrained = true;
263
+ std::vector<char> Pending;
264
+};
265
+
266
+// A persistent writer for a named pipe that transparently handles the server-side connection lifecycle. Data
267
+// pushed via Push() is written to the pipe by the IO loop. 'Connected' indicates the pipe already has a connected
268
+// peer.
269
+class WriteNamedPipe : public OverlappedIOHandle
270
+{
271
+public:
272
+ NON_COPYABLE(WriteNamedPipe);
273
+ NON_MOVABLE(WriteNamedPipe);
274
+
275
+ WriteNamedPipe(HandleWrapper&& Pipe, bool Reconnect, bool Connected);
276
+ ~WriteNamedPipe();
277
+ void Schedule() override;
278
+ void Collect() override;
279
+ HANDLE GetHandle() const override;
280
+ void Push(const gsl::span<char>& Buffer);
281
+
282
+ // Returns the number of bytes that have been queued for writing but not yet written to the pipe.
283
+ size_t PendingBytes() const;
284
+
285
+private:
286
+ // Drops the current client and arms a fresh connection so the next Schedule() reconnects before writing.
287
+ void Reconnect();
288
+
289
+ HandleWrapper Pipe;
290
+ std::optional<WriteHandle> Write;
291
+ wil::unique_event ConnectEvent{wil::EventOptions::ManualReset};
292
+ OVERLAPPED ConnectOverlapped{};
293
+ bool ReconnectOnFailure = false;
294
+ bool NeedConnect = false;
295
+ bool Connecting = false;
296
};
297
298
template <typename TRead = ReadHandle>
@@ -235,7 +303,7 @@ public:
303
NON_MOVABLE(RelayHandle);
304
305
RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output) :
238
- Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output))
306
+ Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output), {}, false)
307
{
308
}
309
@@ -246,10 +314,21 @@ public:
314
// If the Buffer is empty, then we're reading.
315
if (PendingBuffer.empty())
316
{
249
- // If the output buffer is empty and the reading end is completed, then we're done.
317
if (Read.GetState() == IOHandleStatus::Completed)
318
{
252
- State = IOHandleStatus::Completed;
319
+ // If all reading is complete, flush any pending writes before transitioning to Completed.
320
+ Write.SetCompleteOnDrained(true);
321
+
322
+ if (Write.PendingBytes() > 0)
323
+ {
324
+ Write.Schedule();
325
+ State = Write.GetState();
326
+ }
327
+ else
328
+ {
329
+ State = IOHandleStatus::Completed;
330
+ }
331
+
332
return;
333
}
334
@@ -391,7 +470,9 @@ private:
470
concurrency::concurrent_queue<Entry*> m_signaledHandles;
471
wil::unique_event m_handleSignaledEvent{wil::EventOptions::ManualReset};
472
394
- std::vector<std::unique_ptr<Entry>> m_handles;
473
+ // N.B. A std::list is used (rather than a vector) so handles can be added from a callback while Run() is
474
+ // iterating m_handles without invalidating the loop's iterator.
475
+ std::list<std::unique_ptr<Entry>> m_handles;
476
bool m_cancel = false;
477
};
478
src/windows/service/exe/HcsVirtualMachine.cpp
+1
-1
@@ -158,7 +158,7 @@ HcsVirtualMachine::HcsVirtualMachine(_In_ const WSLCSessionSettings* Settings)
158
}
159
160
m_dmesgCollector = DmesgCollector::Create(
161
- m_vmId, m_vmExitEvent, true, false, L"", FeatureEnabled(WslcFeatureFlagsEarlyBootDmesg), std::move(dmesgOutputHandle));
161
+ m_vmId, m_vmExitEvent.get(), true, false, L"", FeatureEnabled(WslcFeatureFlagsEarlyBootDmesg), std::move(dmesgOutputHandle));
162
163
if (FeatureEnabled(WslcFeatureFlagsEarlyBootDmesg))
164
{
src/windows/service/exe/HcsVirtualMachine.h
+1
-2
@@ -86,13 +86,12 @@ private:
86
std::wstring m_swiotlbOption;
87
88
wil::unique_socket m_listenSocket;
89
+ wil::unique_event m_vmExitEvent{wil::EventOptions::ManualReset};
90
std::shared_ptr<DmesgCollector> m_dmesgCollector;
91
std::shared_ptr<GuestDeviceManager> m_guestDeviceManager;
92
std::optional<wsl::core::Config> m_natConfig;
93
std::unique_ptr<wsl::core::INetworkingEngine> m_networkEngine;
94
94
- wil::unique_event m_vmExitEvent{wil::EventOptions::ManualReset};
95
-
95
std::map<ULONG, DiskInfo> m_attachedDisks;
96
std::bitset<MAX_VHD_COUNT> m_lunBitmap;
97
src/windows/service/exe/WslCoreVm.cpp
+1
-1
@@ -281,7 +281,7 @@ void WslCoreVm::Initialize(const GUID& VmId, const wil::shared_handle& UserToken
281
{
282
bool enableTelemetry = TraceLoggingProviderEnabled(g_hTraceLoggingProvider, WINEVENT_LEVEL_INFO, 0);
283
m_dmesgCollector = DmesgCollector::Create(
284
- VmId, m_vmExitEvent, enableTelemetry, m_vmConfig.EnableDebugConsole, m_comPipe0, m_vmConfig.EnableEarlyBootLogging, {});
284
+ VmId, m_vmExitEvent.get(), enableTelemetry, m_vmConfig.EnableDebugConsole, m_comPipe0, m_vmConfig.EnableEarlyBootLogging, {});
285
286
WSL_LOG("DMESG collector created");
287
test/windows/Common.cpp
+3
-2
@@ -1538,11 +1538,12 @@ std::wstring LxssGenerateTestConfig(TestConfigDefaults Default)
1538
L"mountDeviceTimeout=120000\n"
1539
L"kernelBootTimeout=120000\n"
1540
L"debugConsoleLogFile=" +
1541
- EscapePath(kernelLogs) +
1541
+ EscapePath(Default.debugConsoleLogFile.value_or(kernelLogs)) +
1542
L"\n"
1543
L"telemetry=false\n" +
1544
boolOptionToString(L"safeMode", Default.safeMode, false) + boolOptionToString(L"guiApplications", Default.guiApplications, true) +
1545
- L"earlyBootLogging=false\n" + networkingModeToString(Default.networkingMode) + drvFsModeToString(Default.drvFsMode);
1545
+ boolOptionToString(L"earlyBootLogging", Default.earlyBootLogging, false) +
1546
+ networkingModeToString(Default.networkingMode) + drvFsModeToString(Default.drvFsMode);
1547
1548
if (Default.kernel.has_value())
1549
{
test/windows/Common.h
+2
@@ -541,6 +541,8 @@ struct TestConfigDefaults
541
std::optional<size_t> vmIdleTimeout;
542
std::optional<bool> safeMode;
543
std::optional<bool> guiApplications;
544
+ std::optional<bool> earlyBootLogging;
545
+ std::optional<std::wstring> debugConsoleLogFile;
546
std::optional<DrvFsMode> drvFsMode;
547
std::optional<wsl::core::NetworkingMode> networkingMode;
548
const std::optional<std::wstring> vmSwitch;
test/windows/UnitTests.cpp
+61
@@ -2291,6 +2291,67 @@ Error code: Wsl/InstallDistro/WSL_E_DISTRO_NOT_FOUND
2291
VERIFY_ARE_EQUAL(L"", warnings);
2292
}
2293
2294
+ WSL2_TEST_METHOD(DmesgCollection)
2295
+ {
2296
+ const auto dmesgLogFile = std::filesystem::current_path() / L"test-dmesg.txt";
2297
+ auto cleanup = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&]() { DeleteFile(dmesgLogFile.c_str()); });
2298
+ WslConfigChange config(LxssGenerateTestConfig({}));
2299
+
2300
+ auto readDmesgLog = [&](uint64_t offset) -> std::string {
2301
+ wil::unique_hfile file(CreateFileW(
2302
+ dmesgLogFile.c_str(), GENERIC_READ, FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, nullptr, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, nullptr));
2303
+ if (!file)
2304
+ {
2305
+ return {};
2306
+ }
2307
+
2308
+ LARGE_INTEGER fileOffset{};
2309
+ fileOffset.QuadPart = static_cast<LONGLONG>(offset);
2310
+ THROW_LAST_ERROR_IF(!SetFilePointerEx(file.get(), fileOffset, nullptr, FILE_BEGIN));
2311
+
2312
+ return ReadToString(file.get());
2313
+ };
2314
+
2315
+ auto fileSize = [&]() -> uint64_t {
2316
+ WIN32_FILE_ATTRIBUTE_DATA attributes{};
2317
+ if (!GetFileAttributesExW(dmesgLogFile.c_str(), GetFileExInfoStandard, &attributes))
2318
+ {
2319
+ return 0;
2320
+ }
2321
+
2322
+ return (static_cast<uint64_t>(attributes.nFileSizeHigh) << 32) | attributes.nFileSizeLow;
2323
+ };
2324
+
2325
+ auto expectInDmesg = [&](bool earlyBootLogging, const std::string_view& expectedLine) -> std::string {
2326
+ config.Update(LxssGenerateTestConfig({.earlyBootLogging = earlyBootLogging, .debugConsoleLogFile = dmesgLogFile}));
2327
+
2328
+ const auto offset = fileSize();
2329
+
2330
+ VERIFY_ARE_EQUAL(LxsstuLaunchWsl(L"/bin/true"), 0L);
2331
+
2332
+ return wsl::shared::retry::RetryWithTimeout<std::string>(
2333
+ [&]() {
2334
+ auto content = readDmesgLog(offset);
2335
+ THROW_HR_IF(E_FAIL, content.find(expectedLine) == std::string::npos);
2336
+
2337
+ return content;
2338
+ },
2339
+ std::chrono::milliseconds(100),
2340
+ std::chrono::seconds(120));
2341
+ };
2342
+
2343
+ // 'Linux version' is printed during early boot. 'brd: module loaded' is printed after transitioning to the virtio console.
2344
+ {
2345
+ auto dmesg = expectInDmesg(true, "brd: module loaded");
2346
+ VERIFY_ARE_NOT_EQUAL(dmesg.find("Linux version"), std::string::npos);
2347
+ }
2348
+
2349
+ {
2350
+ auto dmesg = expectInDmesg(false, "brd: module loaded");
2351
+ VERIFY_ARE_EQUAL(dmesg.find("Linux version"), std::string::npos);
2352
+ }
2353
+ }
2354
+
2355
WSL2_TEST_METHOD(GuiApplications)
2356
{
2357
auto validateEnvironment = [&](bool systemdEnabled) {
test/windows/WSLCTests.cpp
+237
@@ -8303,6 +8303,216 @@ class WSLCTests
8303
VERIFY_ARE_EQUAL(static_cast<DWORD>(fileSize), bytesRead);
8304
VERIFY_IS_TRUE(readBuffer == writeBuffer);
8305
}
8306
+
8307
+ // Validate that WriteHandle behaves correctly when its buffer is fully written, and CompleteOnDrained is false.
8308
+ {
8309
+ auto [readPipe, writePipe] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, false);
8310
+ PartialHandleRead reader(readPipe.get());
8311
+
8312
+ wsl::windows::common::io::MultiHandleWait io;
8313
+ auto writerHandle =
8314
+ std::make_unique<WriteHandle>(wsl::windows::common::io::HandleWrapper{std::move(writePipe)}, std::vector<char>{}, false);
8315
+ auto* writer = writerHandle.get();
8316
+ io.AddHandle(std::move(writerHandle));
8317
+
8318
+ // A reusable writer with nothing queued is Idle, so Run() has no handle to wait on and returns.
8319
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8320
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8321
+
8322
+ // First write: a single Push() transitions the writer out of Idle and is delivered.
8323
+ std::string first = "first-chunk";
8324
+ writer->Push(gsl::make_span(first.data(), first.size()));
8325
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), first.size());
8326
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8327
+ reader.ExpectConsume(first);
8328
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8329
+
8330
+ // Reuse: the writer returned to Idle (not Completed) so it is still registered, and several
8331
+ // queued Push() calls accumulate and are written in order during the next Run().
8332
+ std::string a = "aaa";
8333
+ std::string b = "bbbb";
8334
+ std::string c = "cc";
8335
+ writer->Push(gsl::make_span(a.data(), a.size()));
8336
+ writer->Push(gsl::make_span(b.data(), b.size()));
8337
+ writer->Push(gsl::make_span(c.data(), c.size()));
8338
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), a.size() + b.size() + c.size());
8339
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8340
+ reader.ExpectConsume(a + b + c);
8341
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8342
+
8343
+ // Close the writer.
8344
+ writer->SetCompleteOnDrained(true);
8345
+ std::string exit = "exit";
8346
+ writer->Push(gsl::make_span(exit.data(), exit.size()));
8347
+
8348
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8349
+ reader.Expect(exit);
8350
+ reader.ExpectClosed();
8351
+ }
8352
+ }
8353
+
8354
+ TEST_METHOD(WriteNamedPipeContent)
8355
+ {
8356
+ using wsl::windows::common::io::HandleWrapper;
8357
+ using wsl::windows::common::io::MultiHandleWait;
8358
+ using wsl::windows::common::io::WriteNamedPipe;
8359
+
8360
+ auto createServerPipe = [](const std::wstring& name) {
8361
+ wil::unique_hfile pipe(CreateNamedPipeW(
8362
+ name.c_str(), PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT, 1, 4096, 4096, 0, nullptr));
8363
+ THROW_LAST_ERROR_IF(!pipe);
8364
+
8365
+ return pipe;
8366
+ };
8367
+
8368
+ auto connect = [](const std::wstring& name) {
8369
+ for (;;)
8370
+ {
8371
+ wil::unique_hfile client(CreateFileW(name.c_str(), GENERIC_READ, 0, nullptr, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, nullptr));
8372
+ if (client)
8373
+ {
8374
+ return client;
8375
+ }
8376
+
8377
+ const auto error = GetLastError();
8378
+ THROW_WIN32_IF(error, error != ERROR_PIPE_BUSY && error != ERROR_FILE_NOT_FOUND);
8379
+
8380
+ THROW_IF_WIN32_BOOL_FALSE(WaitNamedPipeW(name.c_str(), 30 * 1000));
8381
+ }
8382
+ };
8383
+
8384
+ auto push = [](WriteNamedPipe& writer, std::string& data) { writer.Push(gsl::make_span(data.data(), data.size())); };
8385
+
8386
+ // Scenario 1: a payload queued before any client exists is delivered once a client connects,
8387
+ // and PendingBytes() drops to zero after the write drains.
8388
+ {
8389
+ const auto name = wsl::windows::common::helpers::GetUniquePipeName();
8390
+
8391
+ MultiHandleWait io;
8392
+ auto writerHandle = std::make_unique<WriteNamedPipe>(HandleWrapper{createServerPipe(name)}, true, false);
8393
+ auto* writer = writerHandle.get();
8394
+ io.AddHandle(std::move(writerHandle));
8395
+
8396
+ // Connect a client up-front; the writer completes the handshake during Run().
8397
+ auto client = connect(name);
8398
+ PartialHandleRead reader(client.get());
8399
+
8400
+ std::string expected = "hello-named-pipe";
8401
+ push(*writer, expected);
8402
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), expected.size());
8403
+
8404
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8405
+
8406
+ reader.Expect(expected);
8407
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8408
+ }
8409
+
8410
+ // Scenario 2: multiple Push() calls accumulate and are delivered, in order, as a single stream.
8411
+ {
8412
+ const auto name = wsl::windows::common::helpers::GetUniquePipeName();
8413
+
8414
+ MultiHandleWait io;
8415
+ auto writerHandle = std::make_unique<WriteNamedPipe>(HandleWrapper{createServerPipe(name)}, true, false);
8416
+ auto* writer = writerHandle.get();
8417
+ io.AddHandle(std::move(writerHandle));
8418
+
8419
+ auto client = connect(name);
8420
+ PartialHandleRead reader(client.get());
8421
+
8422
+ std::string a = "aaaa";
8423
+ std::string b = "bbbbbb";
8424
+ std::string c = "cc";
8425
+ push(*writer, a);
8426
+ push(*writer, b);
8427
+ push(*writer, c);
8428
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), a.size() + b.size() + c.size());
8429
+
8430
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8431
+
8432
+ reader.Expect(a + b + c);
8433
+ }
8434
+
8435
+ // Scenario 3: when the connected client disconnects, the next write fails and the writer
8436
+ // reconnects, resuming delivery to a new client without losing the buffered payload.
8437
+ {
8438
+ const auto name = wsl::windows::common::helpers::GetUniquePipeName();
8439
+
8440
+ MultiHandleWait io;
8441
+ auto writerHandle = std::make_unique<WriteNamedPipe>(HandleWrapper{createServerPipe(name)}, true, false);
8442
+ auto* writer = writerHandle.get();
8443
+ io.AddHandle(std::move(writerHandle));
8444
+
8445
+ // Phase 1: the first client connects, reads the first payload, then disconnects (the reader
8446
+ // and client are scoped so the reader thread joins before the client handle closes).
8447
+ std::string first = "first-payload";
8448
+ {
8449
+ auto client1 = connect(name);
8450
+ PartialHandleRead reader1(client1.get());
8451
+
8452
+ push(*writer, first);
8453
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8454
+ reader1.Expect(first);
8455
+ }
8456
+
8457
+ // Phase 2: the next write fails against the now-closed client, triggering a reconnect. A
8458
+ // second client connects while Run() performs the reconnect and receives the buffered payload.
8459
+ std::string second = "second-payload";
8460
+ push(*writer, second);
8461
+
8462
+ wil::unique_hfile client2;
8463
+ std::thread connector([&]() { client2 = connect(name); });
8464
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8465
+ connector.join();
8466
+
8467
+ PartialHandleRead reader2(client2.get());
8468
+ reader2.Expect(second);
8469
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8470
+ }
8471
+
8472
+ // Scenario 4: a writer over an already-connected handle (Connected=true) skips the connection
8473
+ // handshake and behaves like a persistent WriteHandle, writing queued data straight to the handle.
8474
+ {
8475
+ auto [readPipe, writePipe] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, false);
8476
+ PartialHandleRead reader(readPipe.get());
8477
+
8478
+ MultiHandleWait io;
8479
+ auto writerHandle = std::make_unique<WriteNamedPipe>(HandleWrapper{std::move(writePipe)}, false, true);
8480
+ auto* writer = writerHandle.get();
8481
+ io.AddHandle(std::move(writerHandle));
8482
+
8483
+ std::string expected = "no-reconnect-path";
8484
+ push(*writer, expected);
8485
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), expected.size());
8486
+
8487
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8488
+
8489
+ reader.Expect(expected);
8490
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8491
+ }
8492
+
8493
+ // Scenario 5: Validate that the named pipe is connected if constructed with Connected = false.
8494
+ {
8495
+ const auto name = wsl::windows::common::helpers::GetUniquePipeName();
8496
+
8497
+ MultiHandleWait io;
8498
+ auto writerHandle = std::make_unique<WriteNamedPipe>(HandleWrapper{createServerPipe(name)}, false, false);
8499
+ auto* writer = writerHandle.get();
8500
+ io.AddHandle(std::move(writerHandle));
8501
+
8502
+ std::string expected = "handshake-without-reconnect";
8503
+ push(*writer, expected);
8504
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), expected.size());
8505
+
8506
+ // Connect the client after the payload is queued; the writer completes the handshake during Run().
8507
+ wil::unique_hfile client;
8508
+ std::thread connector([&]() { client = connect(name); });
8509
+ VERIFY_IS_TRUE(io.Run(std::chrono::seconds(30)));
8510
+ connector.join();
8511
+
8512
+ PartialHandleRead reader(client.get());
8513
+ reader.Expect(expected);
8514
+ VERIFY_ARE_EQUAL(writer->PendingBytes(), static_cast<size_t>(0));
8515
+ }
8516
}
8517
8518
TEST_METHOD(DockerIORelay)
@@ -8430,6 +8640,33 @@ class WSLCTests
8640
}
8641
}
8642
8643
+ TEST_METHOD(RelayHandleLargeBuffer)
8644
+ {
8645
+ using namespace wsl::windows::common::io;
8646
+
8647
+ auto [srcRead, srcWrite] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, true);
8648
+ auto [dstRead, dstWrite] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, true);
8649
+
8650
+ // A payload larger than the relay read buffer forces several read -> write cycles through the
8651
+ // RelayHandle's reused WriteHandle.
8652
+ const std::string payload(LX_RELAY_BUFFER_SIZE * 4 + 123, 'x');
8653
+
8654
+ MultiHandleWait io;
8655
+
8656
+ io.AddHandle(std::make_unique<WriteHandle>(std::move(srcWrite), std::vector<char>(payload.begin(), payload.end())));
8657
+ io.AddHandle(std::make_unique<RelayHandle<>>(std::move(srcRead), std::move(dstWrite)));
8658
+
8659
+ // Collect the relayed output.
8660
+ std::string output;
8661
+ io.AddHandle(std::make_unique<ReadHandle>(
8662
+ std::move(dstRead), [&](const gsl::span<char>& buffer) { output.append(buffer.data(), buffer.size()); }));
8663
+
8664
+ io.Run({});
8665
+
8666
+ VERIFY_ARE_EQUAL(payload.size(), output.size());
8667
+ VERIFY_IS_TRUE(payload == output);
8668
+ }
8669
+
8670
WSLC_TEST_METHOD(ContainerRecoveryFromStorage)
8671
{
8672
auto restore = ResetTestSession(); // Required to access the storage folder.