@samitouri / QOSAMI-WSL-2 / commits / d4b49348

Move IO related classes to their own file & namespace (#40534)

* Move IO related classes to their own file & namespace * Add file * Format

Blue committed May 14, 2026 at 12:46 UTC d4b49348fbc5f424fadb1609777c9dc1f22145a7

30 files changed +1548 -1446

src/shared/inc/SocketChannel.h

+6 -6

@@ -184,7 +184,7 @@ public:

184 #ifdef WIN32

185

186 auto io = CreateIO();

187 - io.AddHandle(std::make_unique<windows::common::relay::WriteHandle>(m_socket.get(), span));

187 + io.AddHandle(std::make_unique<windows::common::io::WriteHandle>(m_socket.get(), span));

188

189 io.Run(TimeoutToMilliseconds(timeout));

190

@@ -605,17 +605,17 @@ public:

605

606 private:

607 #ifdef WIN32

608 - windows::common::relay::MultiHandleWait CreateIO() const

608 + windows::common::io::MultiHandleWait CreateIO() const

609 {

610 - wsl::windows::common::relay::MultiHandleWait io;

610 + wsl::windows::common::io::MultiHandleWait io;

611

612 for (const auto event : m_exitEvents)

613 {

614 io.AddHandle(

615 - std::make_unique<windows::common::relay::EventHandle>(

615 + std::make_unique<windows::common::io::EventHandle>(

616 event,

617 [this, event]() { THROW_HR_MSG(E_ABORT, "Exit event 0x%p signaled on channel: %hs", event, m_name.c_str()); }),

618 - windows::common::relay::MultiHandleWait::CancelOnCompleted | windows::common::relay::MultiHandleWait::NeedNotComplete);

618 + windows::common::io::MultiHandleWait::CancelOnCompleted | windows::common::io::MultiHandleWait::NeedNotComplete);

619 }

620

621 return io;

@@ -637,7 +637,7 @@ private:

637

638 gsl::span<gsl::byte> message;

639

640 - io.AddHandle(std::make_unique<windows::common::relay::ReadSocketMessageHandle>(

640 + io.AddHandle(std::make_unique<windows::common::io::ReadSocketMessageHandle>(

641 m_socket.get(), m_buffer, [&message](auto& received) { message = received; }));

642

643 io.Run(TimeoutToMilliseconds(timeout));

src/windows/WslcSDK/IOCallback.cpp

+1 -1

@@ -17,7 +17,7 @@ Abstract:

17 IOCallback::IOCallback(IWSLCProcess* process, const WslcContainerProcessIOCallbackOptions& options) :

18 m_process(process), m_callbackOptions(std::make_unique<WslcContainerProcessIOCallbackOptions>(options))

19 {

20 - using namespace wsl::windows::common::relay;

20 + using namespace wsl::windows::common::io;

22 auto addIOCallback = [&](WslcProcessIOHandle ioHandle, WslcStdIOCallback callback, PVOID context) {

23 std::function<void(const gsl::span<char>& Buffer)> function;

src/windows/WslcSDK/IOCallback.h

+1 -1

@@ -35,6 +35,6 @@ private:

35 wil::com_ptr<IWSLCProcess> m_process;

36 std::unique_ptr<WslcContainerProcessIOCallbackOptions> m_callbackOptions;

37 std::thread m_thread;

38 - wsl::windows::common::relay::MultiHandleWait m_io;

38 + wsl::windows::common::io::MultiHandleWait m_io;

39 wil::unique_event m_cancelEvent{wil::EventOptions::ManualReset};

40 };

src/windows/WslcSDK/wslcsdk.cpp

+1 -1

@@ -268,7 +268,7 @@ struct ImageFileResolver

268 }

269

270 private:

271 - wsl::windows::common::relay::HandleWrapper m_fileHandle;

271 + wsl::windows::common::io::HandleWrapper m_fileHandle;

272 ULONGLONG m_length;

273 };

274

src/windows/common/CMakeLists.txt

@@ -18,6 +18,7 @@ set(SOURCES

18 helpers.cpp

19 hvsocket.cpp

20 interop.cpp

21 + HandleIO.cpp

22 Localization.cpp

23 lxssbusclient.cpp

24 lxssclient.cpp

@@ -101,6 +102,7 @@ set(HEADERS

102 hvsocket.hpp

103 INetworkingEngine.h

104 interop.hpp

105 + HandleIO.h

106 LxssMessagePort.h

107 LxssPort.h

108 LxssSecurity.h

src/windows/common/HandleIO.cpp new

+1047

@@ -0,0 +1,1047 @@

2 +

3 +#include "precomp.h"

4 +#include "HandleIO.h"

5 +#pragma hdrstop

6 +

7 +using wsl::windows::common::io::BufferWrapper;

8 +using wsl::windows::common::io::DockerIORelayHandle;

9 +using wsl::windows::common::io::EventHandle;

10 +using wsl::windows::common::io::HandleWrapper;

11 +using wsl::windows::common::io::HTTPChunkBasedReadHandle;

12 +using wsl::windows::common::io::IOHandleStatus;

13 +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::ReadSocketMessageHandle;

18 +using wsl::windows::common::io::SingleAcceptHandle;

19 +using wsl::windows::common::io::WriteHandle;

20 +

21 +namespace {

22 +

23 +LARGE_INTEGER InitializeFileOffset(HANDLE File)

24 +{

25 + LARGE_INTEGER Offset{};

26 + if (GetFileType(File) == FILE_TYPE_DISK)

27 + {

28 + LOG_IF_WIN32_BOOL_FALSE(SetFilePointerEx(File, {}, &Offset, FILE_CURRENT));

29 + }

30 +

31 + return Offset;

32 +}

33 +

34 +void CancelPendingIo(auto Handle, OVERLAPPED& Overlapped)

35 +{

36 + DWORD bytesTransferred{};

37 + if (CancelIoEx((HANDLE)Handle, &Overlapped))

38 + {

39 + if constexpr (std::is_same_v<decltype(Handle), SOCKET>)

40 + {

41 + if (!WSAGetOverlappedResult(Handle, &Overlapped, &bytesTransferred, true, nullptr))

42 + {

43 + auto error = WSAGetLastError();

44 + LOG_LAST_ERROR_IF(error != WSAECONNABORTED && error != WSA_OPERATION_ABORTED && error != WSAECONNRESET);

45 + }

46 + }

47 + else

48 + {

49 + static_assert(std::is_same_v<decltype(Handle), HANDLE>);

50 + if (!GetOverlappedResult(Handle, &Overlapped, &bytesTransferred, true))

51 + {

52 + auto error = GetLastError();

53 + LOG_LAST_ERROR_IF(error != ERROR_CONNECTION_ABORTED && error != ERROR_OPERATION_ABORTED);

54 + }

55 + }

56 + }

57 + else

58 + {

59 + // ERROR_NOT_FOUND is returned if there was no IO to cancel.

60 + LOG_LAST_ERROR_IF(GetLastError() != ERROR_NOT_FOUND);

61 + }

62 +}

63 +

64 +} // namespace

65 +

66 +// HandleWrapper

67 +

68 +HandleWrapper::HandleWrapper(wil::unique_handle&& handle, std::function<void()>&& OnClose) :

69 + Handle(handle.get()), OwnedHandle(std::move(handle)), OnClose(std::move(OnClose))

70 +{

71 +}

72 +

73 +HandleWrapper::HandleWrapper(wil::unique_socket&& handle, std::function<void()>&& OnClose) :

74 + Handle((HANDLE)handle.get()), OwnedHandle(wil::unique_socket{handle.release()}), OnClose(std::move(OnClose))

75 +{

76 +}

77 +

78 +HandleWrapper::HandleWrapper(wil::unique_event&& handle, std::function<void()>&& OnClose) :

79 + Handle(handle.get()), OwnedHandle(wil::unique_handle{handle.release()}), OnClose(std::move(OnClose))

80 +{

81 +}

82 +

83 +HandleWrapper::HandleWrapper(SOCKET handle, std::function<void()>&& OnClose) :

84 + Handle(reinterpret_cast<HANDLE>(handle)), OnClose(std::move(OnClose))

85 +{

86 +}

87 +

88 +HandleWrapper::HandleWrapper(HANDLE handle, std::function<void()>&& OnClose) : Handle(handle), OnClose(std::move(OnClose))

89 +{

90 +}

91 +

92 +HandleWrapper::HandleWrapper(wil::unique_hfile&& handle, std::function<void()>&& OnClose) :

93 + Handle(handle.get()), OwnedHandle(wil::unique_handle{handle.release()}), OnClose(std::move(OnClose))

94 +{

95 +}

96 +

97 +HandleWrapper::~HandleWrapper()

98 +{

99 + Reset();

100 +}

101 +

102 +HANDLE HandleWrapper::Get() const

103 +{

104 + return Handle;

105 +}

106 +

107 +void HandleWrapper::Reset()

108 +{

109 + if (OnClose != nullptr)

110 + {

111 + OnClose();

112 + OnClose = nullptr;

113 + }

114 +

115 + OwnedHandle = {};

116 + Handle = nullptr;

117 +}

118 +

119 +// BufferWrapper

120 +

121 +BufferWrapper::BufferWrapper(size_t size) : m_owned(std::in_place, size)

122 +{

123 +}

124 +

125 +BufferWrapper::BufferWrapper(gsl::span<gsl::byte> span) : m_unowned(span)

126 +{

127 +}

128 +

129 +bool BufferWrapper::Owned() const noexcept

130 +{

131 + return m_owned.has_value();

132 +}

133 +

134 +void BufferWrapper::Resize(size_t size)

135 +{

136 + THROW_HR_IF_MSG(E_UNEXPECTED, !Owned(), "BufferWrapper::Resize called on a non-owned buffer");

137 + m_owned->resize(size);

138 +}

139 +

140 +void BufferWrapper::Append(gsl::span<char> Span)

141 +{

142 + THROW_HR_IF_MSG(E_UNEXPECTED, !Owned(), "BufferWrapper::Append called on a non-owned buffer");

143 +

144 + m_owned->insert(m_owned->end(), Span.begin(), Span.end());

145 +}

146 +

147 +void BufferWrapper::Consume(size_t bytes) noexcept

148 +{

149 + WI_ASSERT(bytes <= Size());

150 + if (Owned())

151 + {

152 + m_owned->erase(m_owned->begin(), m_owned->begin() + bytes);

153 + }

154 + else

155 + {

156 + m_unowned = m_unowned.subspan(bytes);

157 + }

158 +}

159 +

160 +gsl::span<gsl::byte> BufferWrapper::Span() noexcept

161 +{

162 + return Owned() ? gsl::make_span(reinterpret_cast<gsl::byte*>(m_owned->data()), m_owned->size()) : m_unowned;

163 +}

164 +

165 +size_t BufferWrapper::Size() const noexcept

166 +{

167 + return Owned() ? m_owned->size() : m_unowned.size();

168 +}

169 +

170 +// OverlappedIOHandle

171 +

172 +IOHandleStatus OverlappedIOHandle::GetState() const

173 +{

174 + return State;

175 +}

176 +

177 +// EventHandle

178 +

179 +EventHandle::EventHandle(HandleWrapper&& Handle, std::function<void()>&& OnSignalled) :

180 + Handle(std::move(Handle)), OnSignalled(std::move(OnSignalled))

181 +{

182 +}

183 +

184 +void EventHandle::Schedule()

185 +{

186 + State = IOHandleStatus::Pending;

187 +}

188 +

189 +void EventHandle::Collect()

190 +{

191 + State = IOHandleStatus::Completed;

192 + OnSignalled();

193 +}

194 +

195 +HANDLE EventHandle::GetHandle() const

196 +{

197 + return Handle.Get();

198 +}

199 +

200 +// ReadHandle

201 +

202 +ReadHandle::ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead) :

203 + Handle(std::move(MovedHandle)), OnRead(OnRead), Offset(InitializeFileOffset(Handle.Get()))

204 +{

205 + Overlapped.hEvent = Event.get();

206 +}

207 +

208 +ReadHandle::~ReadHandle()

209 +{

210 + if (State == IOHandleStatus::Pending)

211 + {

212 + CancelPendingIo(Handle.Get(), Overlapped);

213 + }

214 +}

215 +

216 +void ReadHandle::Schedule()

217 +{

218 + WI_ASSERT(State == IOHandleStatus::Standby);

219 +

220 + Event.ResetEvent();

221 +

222 + // Schedule the read.

223 + DWORD bytesRead{};

224 + Overlapped.Offset = Offset.LowPart;

225 + Overlapped.OffsetHigh = Offset.HighPart;

226 + auto* bufferData = reinterpret_cast<char*>(Buffer.Span().data());

227 + if (ReadFile(Handle.Get(), bufferData, static_cast<DWORD>(Buffer.Size()), &bytesRead, &Overlapped))

228 + {

229 + Offset.QuadPart += bytesRead;

230 +

231 + // Signal the read.

232 + OnRead(gsl::make_span<char>(bufferData, static_cast<size_t>(bytesRead)));

233 +

234 + // ReadFile completed immediately, process the result right away.

235 + if (bytesRead == 0)

236 + {

237 + State = IOHandleStatus::Completed;

238 + return; // Handle is completely read, don't try again.

239 + }

240 +

241 + // Read was done synchronously, remain in 'standby' state.

242 + }

243 + else

244 + {

245 + auto error = GetLastError();

246 + if (error == ERROR_HANDLE_EOF || error == ERROR_BROKEN_PIPE)

247 + {

248 + // Signal an empty read for EOF.

249 + OnRead({});

250 +

251 + State = IOHandleStatus::Completed;

252 + return;

253 + }

254 +

255 + THROW_LAST_ERROR_IF_MSG(error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)Handle.Get());

256 +

257 + // The read is pending, update to 'Pending'

258 + State = IOHandleStatus::Pending;

259 + }

260 +}

261 +

262 +void ReadHandle::Collect()

263 +{

264 + WI_ASSERT(State == IOHandleStatus::Pending);

265 +

266 + // Transition back to standby

267 + State = IOHandleStatus::Standby;

268 +

269 + // Complete the read.

270 + DWORD bytesRead{};

271 + if (!GetOverlappedResult(Handle.Get(), &Overlapped, &bytesRead, false))

272 + {

273 + auto error = GetLastError();

274 + THROW_WIN32_IF(error, error != ERROR_HANDLE_EOF && error != ERROR_BROKEN_PIPE);

275 +

276 + // We received ERROR_HANDLE_EOF or ERROR_BROKEN_PIPE. Validate that this was indeed a zero byte read.

277 + WI_ASSERT(bytesRead == 0);

278 + }

279 +

280 + Offset.QuadPart += bytesRead;

281 +

282 + // Signal the read.

283 + OnRead(gsl::make_span<char>(reinterpret_cast<char*>(Buffer.Span().data()), static_cast<size_t>(bytesRead)));

284 +

285 + // Transition to Complete if this was a zero byte read.

286 + if (bytesRead == 0)

287 + {

288 + State = IOHandleStatus::Completed;

289 + }

290 +}

291 +

292 +HANDLE ReadHandle::GetHandle() const

293 +{

294 + return Event.get();

295 +}

296 +

297 +// SingleAcceptHandle

298 +

299 +SingleAcceptHandle::SingleAcceptHandle(HandleWrapper&& ListenSocket, HandleWrapper&& AcceptedSocket, std::function<void()>&& OnAccepted) :

300 + ListenSocket(std::move(ListenSocket)), AcceptedSocket(std::move(AcceptedSocket)), OnAccepted(std::move(OnAccepted))

301 +{

302 + Overlapped.hEvent = Event.get();

303 +}

304 +

305 +SingleAcceptHandle::~SingleAcceptHandle()

306 +{

307 + if (State == IOHandleStatus::Pending)

308 + {

309 + LOG_IF_WIN32_BOOL_FALSE(CancelIoEx(ListenSocket.Get(), &Overlapped));

310 +

311 + DWORD bytesProcessed{};

312 + DWORD flagsReturned{};

313 + if (!WSAGetOverlappedResult((SOCKET)ListenSocket.Get(), &Overlapped, &bytesProcessed, TRUE, &flagsReturned))

314 + {

315 + auto error = GetLastError();

316 + LOG_LAST_ERROR_IF(error != ERROR_CONNECTION_ABORTED && error != ERROR_OPERATION_ABORTED);

317 + }

318 + }

319 +}

320 +

321 +void SingleAcceptHandle::Schedule()

322 +{

323 + WI_ASSERT(State == IOHandleStatus::Standby);

324 +

325 + // Schedule the accept.

326 + DWORD bytesReturned{};

327 + if (AcceptEx((SOCKET)ListenSocket.Get(), (SOCKET)AcceptedSocket.Get(), &AcceptBuffer, 0, sizeof(SOCKADDR_STORAGE), sizeof(SOCKADDR_STORAGE), &bytesReturned, &Overlapped))

328 + {

329 + // Accept completed immediately.

330 + State = IOHandleStatus::Completed;

331 + OnAccepted();

332 + }

333 + else

334 + {

335 + auto error = WSAGetLastError();

336 + THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)ListenSocket.Get());

337 +

338 + State = IOHandleStatus::Pending;

339 + }

340 +}

341 +

342 +void SingleAcceptHandle::Collect()

343 +{

344 + WI_ASSERT(State == IOHandleStatus::Pending);

345 +

346 + DWORD bytesReceived{};

347 + DWORD flagsReturned{};

348 +

349 + THROW_IF_WIN32_BOOL_FALSE(WSAGetOverlappedResult((SOCKET)ListenSocket.Get(), &Overlapped, &bytesReceived, false, &flagsReturned));

350 +

351 + State = IOHandleStatus::Completed;

352 + OnAccepted();

353 +}

354 +

355 +HANDLE SingleAcceptHandle::GetHandle() const

356 +{

357 + return Event.get();

358 +}

359 +

360 +// LineBasedReadHandle

361 +

362 +LineBasedReadHandle::LineBasedReadHandle(HandleWrapper&& Handle, std::function<void(const gsl::span<char>& Line)>&& OnLine, bool Crlf) :

363 + ReadHandle(std::move(Handle), [this](const gsl::span<char>& Buffer) { OnRead(Buffer); }), OnLine(OnLine), Crlf(Crlf)

364 +{

365 +}

366 +

367 +LineBasedReadHandle::~LineBasedReadHandle()

368 +{

369 + // N.B. PendingBuffer can contain remaining data is an exception was thrown during parsing.

370 +}

371 +

372 +void LineBasedReadHandle::OnRead(const gsl::span<char>& Buffer)

373 +{

374 + // If we reach of the end, signal a line with the remaining buffer.

375 + if (Buffer.empty() && !PendingBuffer.empty())

376 + {

377 + OnLine(PendingBuffer);

378 + PendingBuffer.clear();

379 + return;

380 + }

381 +

382 + auto begin = Buffer.begin();

383 + auto end = std::ranges::find(Buffer, Crlf ? '\r' : '\n');

384 + while (end != Buffer.end())

385 + {

386 + if (Crlf)

387 + {

388 + end++; // Move to the following '\n'

389 +

390 + if (end == Buffer.end() || *end != '\n') // Incomplete CRLF sequence. Append to buffer and continue.

391 + {

392 + PendingBuffer.insert(PendingBuffer.end(), begin, end);

393 + begin = end;

394 + end = std::ranges::find(end, Buffer.end(), '\r');

395 + continue;

396 + }

397 + }

398 +

399 + // Discard the '\r' in CRLF mode.

400 + PendingBuffer.insert(PendingBuffer.end(), begin, Crlf ? end - 1 : end);

401 +

402 + if (!PendingBuffer.empty())

403 + {

404 + OnLine(PendingBuffer);

405 + PendingBuffer.clear();

406 + }

407 +

408 + begin = end + 1;

409 + end = std::ranges::find(begin, Buffer.end(), Crlf ? '\r' : '\n');

410 + }

411 +

412 + PendingBuffer.insert(PendingBuffer.end(), begin, end);

413 +}

414 +

415 +// HTTPChunkBasedReadHandle

416 +

417 +HTTPChunkBasedReadHandle::HTTPChunkBasedReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Line)>&& OnChunk) :

418 + ReadHandle(std::move(MovedHandle), [this](const gsl::span<char>& Buffer) { OnRead(Buffer); }), OnChunk(std::move(OnChunk))

419 +{

420 +}

421 +

422 +HTTPChunkBasedReadHandle::~HTTPChunkBasedReadHandle()

423 +{

424 + // N.B. PendingBuffer can contain remaining data is an exception was thrown during parsing.

425 + LOG_HR_IF(E_UNEXPECTED, !PendingBuffer.empty() || PendingChunkSize != 0 || ExpectHeader);

426 +}

427 +

428 +void HTTPChunkBasedReadHandle::OnRead(const gsl::span<char>& Input)

429 +{

430 + // See: https://httpwg.org/specs/rfc9112.html#field.transfer-encoding

431 +

432 + if (Input.empty())

433 + {

434 + // N.B. The body can be terminated by a zero-length chunk.

435 + THROW_HR_IF(E_INVALIDARG, PendingChunkSize != 0 || ExpectHeader);

436 + }

437 +

438 + auto buffer = Input;

439 +

440 + auto advance = [&](size_t count) {

441 + WI_ASSERT(buffer.size() >= count);

442 + buffer = buffer.subspan(count);

443 + };

444 +

445 + while (!buffer.empty())

446 + {

447 + if (PendingChunkSize == 0)

448 + {

449 + // Consume CRLF's between chunks.

450 + if (PendingBuffer.empty() && (buffer.front() == '\r' || buffer.front() == '\n'))

451 + {

452 + advance(1);

453 + continue;

454 + }

455 +

456 + ExpectHeader = true;

457 +

458 + auto end = std::ranges::find(buffer, '\n');

459 + PendingBuffer.insert(PendingBuffer.end(), buffer.begin(), end);

460 + if (end == buffer.end())

461 + {

462 + // Incomplete size header, buffer until next read.

463 + break;

464 + }

465 + // Advance beyond the LF

466 + advance(end - buffer.begin() + 1);

467 +

468 + THROW_HR_IF_MSG(

469 + E_INVALIDARG,

470 + PendingBuffer.size() < 2 || PendingBuffer.back() != '\r',

471 + "Malformed chunk header: %hs",

472 + PendingBuffer.c_str());

473 + PendingBuffer.erase(PendingBuffer.end() - 1, PendingBuffer.end()); // Remove CR.

474 +

475 +#ifdef WSLC_HTTP_DEBUG

476 +

477 + WSL_LOG("HTTPChunkHeader", TraceLoggingValue(PendingBuffer.c_str(), "Size"));

478 +

479 +#endif

480 +

481 + try

482 + {

483 + size_t parsed{};

484 + PendingChunkSize = std::stoul(PendingBuffer.c_str(), &parsed, 16);

485 + THROW_HR_IF(E_INVALIDARG, parsed != PendingBuffer.size());

486 + }

487 + catch (...)

488 + {

489 + THROW_HR_MSG(E_INVALIDARG, "Failed to parse chunk size: %hs", PendingBuffer.c_str());

490 + }

491 +

492 + ExpectHeader = false;

493 + PendingBuffer.clear();

494 + }

495 + else

496 + {

497 + // Consume the chunk.

498 + auto consumedBytes = std::min(PendingChunkSize, buffer.size());

499 + PendingBuffer.append(buffer.data(), consumedBytes);

500 + advance(consumedBytes);

501 +

502 + WI_ASSERT(PendingChunkSize >= consumedBytes);

503 + PendingChunkSize -= consumedBytes;

504 +

505 + if (PendingChunkSize == 0)

506 + {

507 +

508 +#ifdef WSLC_HTTP_DEBUG

509 +

510 + WSL_LOG("HTTPChunk", TraceLoggingValue(PendingBuffer.c_str(), "Content"));

511 +

512 +#endif

513 + OnChunk(PendingBuffer);

514 + PendingBuffer.clear();

515 + }

516 + }

517 + }

518 +}

519 +

520 +// ReadSocketMessageHandle

521 +

522 +ReadSocketMessageHandle::ReadSocketMessageHandle(

523 + HandleWrapper&& MovedSocket, std::vector<gsl::byte>& Buffer, std::function<void(const gsl::span<gsl::byte>& Message)>&& OnMessage) :

524 + Socket(std::move(MovedSocket)), Buffer(Buffer), OnMessage(std::move(OnMessage))

525 +{

526 + Overlapped.hEvent = Event.get();

527 +

528 + if (Buffer.size() < sizeof(MESSAGE_HEADER))

529 + {

530 + Buffer.resize(sizeof(MESSAGE_HEADER));

531 + }

532 +}

533 +

534 +ReadSocketMessageHandle::~ReadSocketMessageHandle()

535 +{

536 + if (State == IOHandleStatus::Pending)

537 + {

538 + CancelPendingIo((SOCKET)Socket.Get(), Overlapped);

539 + }

540 +}

541 +

542 +void ReadSocketMessageHandle::ScheduleRecv()

543 +{

544 + Event.ResetEvent();

545 +

546 + auto target = gsl::make_span(Buffer).subspan(CurrentOffset, BytesRemaining);

547 + WSABUF wsaBuf = {gsl::narrow_cast<ULONG>(target.size()), reinterpret_cast<CHAR*>(target.data())};

548 + DWORD bytesRead{};

549 + DWORD flags = 0;

550 + if (WSARecv(reinterpret_cast<SOCKET>(Socket.Get()), &wsaBuf, 1, &bytesRead, &flags, &Overlapped, nullptr) == 0)

551 + {

552 + ProcessRecvResult(bytesRead);

553 + }

554 + else

555 + {

556 + auto error = WSAGetLastError();

557 + if (error == WSAECONNABORTED || error == WSAECONNRESET)

558 + {

559 + ProcessRecvResult(0);

560 + return;

561 + }

562 +

563 + THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != WSA_IO_PENDING, "Socket: 0x%p", (void*)Socket.Get());

564 +

565 + State = IOHandleStatus::Pending;

566 + }

567 +}

568 +

569 +void ReadSocketMessageHandle::ProcessRecvResult(DWORD BytesRead)

570 +{

571 + if (BytesRead == 0)

572 + {

573 + // If the socket was closed before any bytes of the next message were read, signal a clean end-of-stream.

574 + // If some bytes were already buffered, the peer closed mid-message which is a protocol error.

575 + THROW_HR_IF_MSG(

576 + E_UNEXPECTED,

577 + CurrentOffset > 0,

578 + "Socket closed before a complete message could be read. ReadingHeader: %d, CurrentOffset: %zu, BytesRemaining: %zu",

579 + ReadingHeader,

580 + CurrentOffset,

581 + BytesRemaining);

582 +

583 + OnMessage({});

584 + State = IOHandleStatus::Completed;

585 + return;

586 + }

587 +

588 + CurrentOffset += BytesRead;

589 + BytesRemaining -= BytesRead;

590 +

591 + if (BytesRemaining > 0)

592 + {

593 + return;

594 + }

595 +

596 + if (ReadingHeader)

597 + {

598 + auto messageSize = gslhelpers::get_struct<MESSAGE_HEADER>(gsl::make_span(Buffer.data(), sizeof(MESSAGE_HEADER)))->MessageSize;

599 +

600 + THROW_HR_IF_MSG(E_UNEXPECTED, messageSize < sizeof(MESSAGE_HEADER), "Unexpected message size: %u", messageSize);

601 + THROW_HR_IF_MSG(E_UNEXPECTED, messageSize > 4 * 1024 * 1024, "Message size too large: %u", messageSize);

602 +

603 + if (messageSize == sizeof(MESSAGE_HEADER))

604 + {

605 + OnMessage(gsl::make_span(Buffer.data(), messageSize));

606 + State = IOHandleStatus::Completed;

607 + return;

608 + }

609 +

610 + if (Buffer.size() < messageSize)

611 + {

612 + Buffer.resize(messageSize);

613 + }

614 +

615 + ReadingHeader = false;

616 + CurrentOffset = sizeof(MESSAGE_HEADER);

617 + BytesRemaining = messageSize - sizeof(MESSAGE_HEADER);

618 + }

619 + else

620 + {

621 + auto messageSize = gslhelpers::get_struct<MESSAGE_HEADER>(gsl::make_span(Buffer.data(), sizeof(MESSAGE_HEADER)))->MessageSize;

622 + OnMessage(gsl::make_span(Buffer.data(), messageSize));

623 + State = IOHandleStatus::Completed;

624 + }

625 +}

626 +

627 +void ReadSocketMessageHandle::Schedule()

628 +{

629 + WI_ASSERT(State == IOHandleStatus::Standby);

630 + ScheduleRecv();

631 +}

632 +

633 +void ReadSocketMessageHandle::Collect()

634 +{

635 + WI_ASSERT(State == IOHandleStatus::Pending);

636 +

637 + State = IOHandleStatus::Standby;

638 +

639 + DWORD bytesRead{};

640 + DWORD flags{};

641 + if (!WSAGetOverlappedResult(reinterpret_cast<SOCKET>(Socket.Get()), &Overlapped, &bytesRead, FALSE, &flags))

642 + {

643 + long error = WSAGetLastError();

644 + THROW_WIN32_IF(error, error != WSAECONNABORTED && error != WSAECONNRESET);

645 +

646 + WI_ASSERT(bytesRead == 0);

647 + }

648 +

649 + ProcessRecvResult(bytesRead);

650 +}

651 +

652 +HANDLE ReadSocketMessageHandle::GetHandle() const

653 +{

654 + return Event.get();

655 +}

656 +

657 +// WriteHandle

658 +

659 +WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, const std::vector<char>& Source) :

660 + Handle(std::move(MovedHandle)), Buffer(Source.size()), Offset(InitializeFileOffset(Handle.Get()))

661 +{

662 + std::memcpy(Buffer.Span().data(), Source.data(), Source.size());

663 + Overlapped.hEvent = Event.get();

664 +}

665 +

666 +WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, gsl::span<gsl::byte> Source) :

667 + Handle(std::move(MovedHandle)), Buffer(Source), Offset(InitializeFileOffset(Handle.Get()))

668 +{

669 + Overlapped.hEvent = Event.get();

670 +}

671 +

672 +WriteHandle::~WriteHandle()

673 +{

674 + if (State == IOHandleStatus::Pending)

675 + {

676 + CancelPendingIo(Handle.Get(), Overlapped);

677 + }

678 +}

679 +

680 +void WriteHandle::Schedule()

681 +{

682 + WI_ASSERT(State == IOHandleStatus::Standby);

683 +

684 + Event.ResetEvent();

685 +

686 + Overlapped.Offset = Offset.LowPart;

687 + Overlapped.OffsetHigh = Offset.HighPart;

688 +

689 + // Schedule the write.

690 + const auto buffer = Buffer.Span();

691 + DWORD bytesWritten{};

692 + if (WriteFile(Handle.Get(), buffer.data(), static_cast<DWORD>(buffer.size()), &bytesWritten, &Overlapped))

693 + {

694 + Offset.QuadPart += bytesWritten;

695 +

696 + Buffer.Consume(bytesWritten);

697 + if (Buffer.Size() == 0)

698 + {

699 + State = IOHandleStatus::Completed;

700 + }

701 + }

702 + else

703 + {

704 + auto error = GetLastError();

705 + THROW_LAST_ERROR_IF_MSG(error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)Handle.Get());

706 +

707 + // The write is pending, update to 'Pending'

708 + State = IOHandleStatus::Pending;

709 + }

710 +}

711 +

712 +void WriteHandle::Collect()

713 +{

714 + WI_ASSERT(State == IOHandleStatus::Pending);

715 +

716 + // Transition back to standby

717 + State = IOHandleStatus::Standby;

718 +

719 + // Complete the write.

720 + DWORD bytesWritten{};

721 + THROW_IF_WIN32_BOOL_FALSE(GetOverlappedResult(Handle.Get(), &Overlapped, &bytesWritten, false));

722 + Offset.QuadPart += bytesWritten;

723 +

724 + Buffer.Consume(bytesWritten);

725 + if (Buffer.Size() == 0)

726 + {

727 + State = IOHandleStatus::Completed;

728 + }

729 +}

730 +

731 +void WriteHandle::Push(const gsl::span<char>& Content)

732 +{

733 + // Don't write if a WriteFile() is pending, since that could cause the buffer to reallocate.

734 + WI_ASSERT(State == IOHandleStatus::Standby || State == IOHandleStatus::Completed);

735 + WI_ASSERT(!Content.empty());

736 +

737 + // Resize() throws E_UNEXPECTED if Buffer does not own its storage.

738 + Buffer.Append(Content);

739 +

740 + State = IOHandleStatus::Standby;

741 +}

742 +

743 +HANDLE WriteHandle::GetHandle() const

744 +{

745 + return Event.get();

746 +}

747 +

748 +// DockerIORelayHandle

749 +

750 +DockerIORelayHandle::DockerIORelayHandle(HandleWrapper&& ReadHandle, HandleWrapper&& Stdout, HandleWrapper&& Stderr, Format ReadFormat) :

751 + WriteStdout(std::move(Stdout)), WriteStderr(std::move(Stderr))

752 +{

753 + if (ReadFormat == Format::HttpChunked)

754 + {

755 + Read = std::make_unique<HTTPChunkBasedReadHandle>(

756 + std::move(ReadHandle), [this](const gsl::span<char>& Line) { this->OnRead(Line); });

757 + }

758 + else

759 + {

760 + Read =

761 + std::make_unique<io::ReadHandle>(std::move(ReadHandle), [this](const gsl::span<char>& Buffer) { this->OnRead(Buffer); });

762 + }

763 +}

764 +

765 +void DockerIORelayHandle::Schedule()

766 +{

767 + WI_ASSERT(State == IOHandleStatus::Standby);

768 + WI_ASSERT(Read->GetState() != IOHandleStatus::Pending);

769 +

770 + // If we have an active handle and a buffer, try to flush that first.

771 + if (ActiveHandle != nullptr && !PendingBuffer.empty())

772 + {

773 + // Push the data to the selected handle.

774 + DWORD bytesToWrite = std::min(static_cast<DWORD>(RemainingBytes), static_cast<DWORD>(PendingBuffer.size()));

775 +

776 + ActiveHandle->Push(gsl::make_span(PendingBuffer.data(), bytesToWrite));

777 +

778 + // Consume the written bytes.

779 + RemainingBytes -= bytesToWrite;

780 + PendingBuffer.erase(PendingBuffer.begin(), PendingBuffer.begin() + bytesToWrite);

781 +

782 + // Schedule the write.

783 + ActiveHandle->Schedule();

784 +

785 + // If the write is pending, update to 'Pending'

786 + if (ActiveHandle->GetState() == IOHandleStatus::Pending)

787 + {

788 + State = IOHandleStatus::Pending;

789 + }

790 + else if (ActiveHandle->GetState() == IOHandleStatus::Completed)

791 + {

792 + if (RemainingBytes == 0)

793 + {

794 + // Switch back to reading if we've written all bytes for this chunk.

795 + ActiveHandle = nullptr;

796 +

797 + ProcessNextHeader();

798 + }

799 + }

800 + }

801 + else

802 + {

803 + if (Read->GetState() == IOHandleStatus::Completed)

804 + {

805 + LOG_HR_IF(E_UNEXPECTED, ActiveHandle != nullptr);

806 +

807 + // No more data to read, we're done.

808 + State = IOHandleStatus::Completed;

809 + return;

810 + }

811 +

812 + // Schedule a read from the input.

813 + Read->Schedule();

814 + if (Read->GetState() == IOHandleStatus::Pending)

815 + {

816 + State = IOHandleStatus::Pending;

817 + }

818 + }

819 +}

820 +

821 +void DockerIORelayHandle::Collect()

822 +{

823 + WI_ASSERT(State == IOHandleStatus::Pending);

824 +

825 + if (ActiveHandle != nullptr && ActiveHandle->GetState() == IOHandleStatus::Pending)

826 + {

827 + // Complete the write.

828 + ActiveHandle->Collect();

829 +

830 + // If the write is completed, switch back to reading.

831 + if (RemainingBytes == 0)

832 + {

833 + if (ActiveHandle->GetState() == IOHandleStatus::Completed)

834 + {

835 + ActiveHandle = nullptr;

836 + }

837 + }

838 +

839 + // Transition back to standby if there's still data to read.

840 + // Otherwise switch to Completed since everything is done.

841 + if (Read->GetState() == IOHandleStatus::Completed)

842 + {

843 + LOG_HR_IF(E_UNEXPECTED, RemainingBytes != 0);

844 +

845 + State = IOHandleStatus::Completed;

846 + }

847 + else

848 + {

849 + State = IOHandleStatus::Standby;

850 + }

851 + }

852 + else

853 + {

854 + WI_ASSERT(Read->GetState() == IOHandleStatus::Pending);

855 +

856 + // Complete the read.

857 + Read->Collect();

858 +

859 + // Transition back to standby.

860 + State = IOHandleStatus::Standby;

861 + }

862 +}

863 +

864 +HANDLE DockerIORelayHandle::GetHandle() const

865 +{

866 + if (ActiveHandle != nullptr && ActiveHandle->GetState() == IOHandleStatus::Pending)

867 + {

868 + return ActiveHandle->GetHandle();

869 + }

870 + else

871 + {

872 + return Read->GetHandle();

873 + }

874 +}

875 +

876 +void DockerIORelayHandle::ProcessNextHeader()

877 +{

878 + if (PendingBuffer.size() < sizeof(MultiplexedHeader))

879 + {

880 + // Not enough data for a header yet.

881 + return;

882 + }

883 +

884 + const auto* header = reinterpret_cast<const MultiplexedHeader*>(PendingBuffer.data());

885 + RemainingBytes = ntohl(header->Length);

886 +

887 + if (header->Fd == 1)

888 + {

889 + ActiveHandle = &WriteStdout;

890 + }

891 + else if (header->Fd == 2)

892 + {

893 + ActiveHandle = &WriteStderr;

894 + }

895 + else

896 + {

897 + THROW_HR_MSG(E_INVALIDARG, "Invalid Docker IO multiplexed header fd: %u", header->Fd);

898 + }

899 +

900 + // Consume the header.

901 + PendingBuffer.erase(PendingBuffer.begin(), PendingBuffer.begin() + sizeof(MultiplexedHeader));

902 +}

903 +

904 +void DockerIORelayHandle::OnRead(const gsl::span<char>& Buffer)

905 +{

906 + PendingBuffer.insert(PendingBuffer.end(), Buffer.begin(), Buffer.end());

907 +

908 + if (ActiveHandle == nullptr)

909 + {

910 + // If no handle is active, expect a header.

911 + ProcessNextHeader();

912 + }

913 +}

914 +

915 +// MultiHandleWait

916 +

917 +void MultiHandleWait::AddHandle(std::unique_ptr<OverlappedIOHandle>&& handle, Flags flags)

918 +{

919 + m_handles.emplace_back(flags, std::move(handle));

920 +}

921 +

922 +void MultiHandleWait::Cancel()

923 +{

924 + m_cancel = true;

925 +}

926 +

927 +bool MultiHandleWait::Run(std::optional<std::chrono::milliseconds> Timeout)

928 +{

929 + m_cancel = false; // Run may be called multiple times.

930 +

931 + std::optional<std::chrono::steady_clock::time_point> deadline;

932 +

933 + if (Timeout.has_value())

934 + {

935 + deadline = std::chrono::steady_clock::now() + Timeout.value();

936 + }

937 +

938 + // Run until all handles are completed.

939 +

940 + while (!m_handles.empty() && !m_cancel)

941 + {

942 + // Schedule IO on each handle until all are either pending, or completed.

943 + for (size_t i = 0; i < m_handles.size() && !m_cancel; i++)

944 + {

945 + while (m_handles[i].second->GetState() == IOHandleStatus::Standby && !m_cancel)

946 + {

947 + try

948 + {

949 + m_handles[i].second->Schedule();

950 + }

951 + catch (...)

952 + {

953 + if (WI_IsFlagSet(m_handles[i].first, Flags::IgnoreErrors))

954 + {

955 + m_handles[i].second.reset(); // Reset the handle so it can be deleted.

956 + break;

957 + }

958 + else

959 + {

960 + throw;

961 + }

962 + }

963 + }

964 + }

965 +

966 + // Remove completed handles from m_handles.

967 + bool hasHandleToWaitFor = false;

968 + for (auto it = m_handles.begin(); it != m_handles.end();)

969 + {

970 + if (!it->second)

971 + {

972 + it = m_handles.erase(it);

973 + }

974 + else if (it->second->GetState() == IOHandleStatus::Completed)

975 + {

976 + if (WI_IsFlagSet(it->first, Flags::CancelOnCompleted))

977 + {

978 + m_cancel = true; // Cancel the IO if a handle with CancelOnCompleted is in the completed state.

979 + }

980 +

981 + it = m_handles.erase(it);

982 + }

983 + else

984 + {

985 + // If only NeedNotComplete handles are left, we want to exit Run.

986 + if (WI_IsFlagClear(it->first, Flags::NeedNotComplete))

987 + {

988 + hasHandleToWaitFor = true;

989 + }

990 + ++it;

991 + }

992 + }

993 +

994 + if (!hasHandleToWaitFor || m_cancel)

995 + {

996 + break;

997 + }

998 +

999 + // Wait for the next operation to complete.

1000 + std::vector<HANDLE> waitHandles;

1001 + for (const auto& e : m_handles)

1002 + {

1003 + waitHandles.emplace_back(e.second->GetHandle());

1004 + }

1005 +

1006 + DWORD waitTimeout = INFINITE;

1007 + if (deadline.has_value())

1008 + {

1009 + auto miliseconds =

1010 + std::chrono::duration_cast<std::chrono::milliseconds>(deadline.value() - std::chrono::steady_clock::now()).count();

1011 +

1012 + waitTimeout = static_cast<DWORD>(std::max(0LL, miliseconds));

1013 + }

1014 +

1015 + auto result = WaitForMultipleObjects(static_cast<DWORD>(waitHandles.size()), waitHandles.data(), false, waitTimeout);

1016 + if (result == WAIT_TIMEOUT)

1017 + {

1018 + THROW_WIN32(ERROR_TIMEOUT);

1019 + }

1020 + else if (result >= WAIT_OBJECT_0 && result < WAIT_OBJECT_0 + m_handles.size())

1021 + {

1022 + auto index = result - WAIT_OBJECT_0;

1023 +

1024 + try

1025 + {

1026 + m_handles[index].second->Collect();

1027 + }

1028 + catch (...)

1029 + {

1030 + if (WI_IsFlagSet(m_handles[index].first, Flags::IgnoreErrors))

1031 + {

1032 + m_handles.erase(m_handles.begin() + index);

1033 + }

1034 + else

1035 + {

1036 + throw;

1037 + }

1038 + }

1039 + }

1040 + else

1041 + {

1042 + THROW_LAST_ERROR_MSG("Timeout: %lu, Count: %llu", waitTimeout, waitHandles.size());

1043 + }

1044 + }

1045 +

1046 + return !m_cancel;

1047 +}

src/windows/common/HandleIO.h new

+380

@@ -0,0 +1,380 @@

2 +

3 +#pragma once

4 +

5 +#define LX_RELAY_BUFFER_SIZE 0x1000

6 +

7 +namespace wsl::windows::common::io {

8 +

9 +enum class IOHandleStatus

10 +{

11 + Standby,

12 + Pending,

13 + Completed

14 +};

15 +

16 +struct HandleWrapper

17 +{

18 + DEFAULT_MOVABLE(HandleWrapper);

19 + NON_COPYABLE(HandleWrapper)

20 +

21 + HandleWrapper(wil::unique_handle&& handle, std::function<void()>&& OnClose = []() {});

22 + HandleWrapper(wil::unique_socket&& handle, std::function<void()>&& OnClose = []() {});

23 + HandleWrapper(wil::unique_event&& handle, std::function<void()>&& OnClose = []() {});

24 + HandleWrapper(SOCKET handle, std::function<void()>&& OnClose = []() {});

25 + HandleWrapper(HANDLE handle, std::function<void()>&& OnClose = []() {});

26 + HandleWrapper(wil::unique_hfile&& handle, std::function<void()>&& OnClose = []() {});

27 + ~HandleWrapper();

28 +

29 + HANDLE Get() const;

30 + void Reset();

31 +

32 +private:

33 + HANDLE Handle{};

34 + std::variant<wil::unique_handle, wil::unique_socket> OwnedHandle;

35 + std::function<void()> OnClose;

36 +};

37 +

38 +// A buffer that may either own its underlying storage (constructed from a size, allocating an

39 +// internal std::vector<char>) or borrow it from a caller-provided gsl::span<gsl::byte>.

40 +class BufferWrapper

41 +{

42 +public:

43 + DEFAULT_MOVABLE(BufferWrapper);

44 + NON_COPYABLE(BufferWrapper);

45 +

46 + explicit BufferWrapper(size_t size);

47 + explicit BufferWrapper(gsl::span<gsl::byte> span);

48 +

49 + bool Owned() const noexcept;

50 + void Resize(size_t size);

51 + void Append(gsl::span<char> Span);

52 + void Consume(size_t bytes) noexcept;

53 + gsl::span<gsl::byte> Span() noexcept;

54 + size_t Size() const noexcept;

55 +

56 +private:

57 + std::optional<std::vector<char>> m_owned;

58 + gsl::span<gsl::byte> m_unowned;

59 +};

60 +

61 +class OverlappedIOHandle

62 +{

63 +public:

64 + NON_COPYABLE(OverlappedIOHandle)

65 + NON_MOVABLE(OverlappedIOHandle)

66 +

67 + OverlappedIOHandle() = default;

68 + virtual ~OverlappedIOHandle() = default;

69 + virtual void Schedule() = 0;

70 + virtual void Collect() = 0;

71 + virtual HANDLE GetHandle() const = 0;

72 + IOHandleStatus GetState() const;

73 +

74 +protected:

75 + IOHandleStatus State = IOHandleStatus::Standby;

76 +};

77 +

78 +class EventHandle : public OverlappedIOHandle

79 +{

80 +public:

81 + NON_COPYABLE(EventHandle)

82 + NON_MOVABLE(EventHandle)

83 +

84 + EventHandle(HandleWrapper&& Handle, std::function<void()>&& OnSignalled = []() {});

85 + void Schedule() override;

86 + void Collect() override;

87 + HANDLE GetHandle() const override;

88 +

89 +private:

90 + HandleWrapper Handle;

91 + std::function<void()> OnSignalled;

92 +};

93 +

94 +class ReadHandle : public OverlappedIOHandle

95 +{

96 +public:

97 + NON_COPYABLE(ReadHandle);

98 + NON_MOVABLE(ReadHandle);

99 +

100 + ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead);

101 + virtual ~ReadHandle();

102 +

103 + void Schedule() override;

104 + void Collect() override;

105 + HANDLE GetHandle() const override;

106 +

107 +private:

108 + HandleWrapper Handle;

109 + std::function<void(const gsl::span<char>& Buffer)> OnRead;

110 + wil::unique_event Event{wil::EventOptions::ManualReset};

111 + OVERLAPPED Overlapped{};

112 + BufferWrapper Buffer{LX_RELAY_BUFFER_SIZE};

113 + LARGE_INTEGER Offset{};

114 +};

115 +

116 +class SingleAcceptHandle : public OverlappedIOHandle

117 +{

118 +public:

119 + NON_COPYABLE(SingleAcceptHandle)

120 + NON_MOVABLE(SingleAcceptHandle)

121 +

122 + SingleAcceptHandle(HandleWrapper&& ListenSocket, HandleWrapper&& AcceptedSocket, std::function<void()>&& OnAccepted);

123 + ~SingleAcceptHandle();

124 +

125 + void Schedule() override;

126 + void Collect() override;

127 + HANDLE GetHandle() const override;

128 +

129 +private:

130 + HandleWrapper ListenSocket;

131 + HandleWrapper AcceptedSocket;

132 + wil::unique_event Event{wil::EventOptions::ManualReset};

133 + OVERLAPPED Overlapped{};

134 + std::function<void()> OnAccepted;

135 + char AcceptBuffer[2 * sizeof(SOCKADDR_STORAGE)];

136 +};

137 +

138 +class LineBasedReadHandle : public ReadHandle

139 +{

140 +public:

141 + NON_COPYABLE(LineBasedReadHandle);

142 + NON_MOVABLE(LineBasedReadHandle);

143 +

144 + LineBasedReadHandle(HandleWrapper&& Handle, std::function<void(const gsl::span<char>& Buffer)>&& OnLine, bool Crlf);

145 + ~LineBasedReadHandle();

146 +

147 +private:

148 + void OnRead(const gsl::span<char>& Buffer);

149 +

150 + std::function<void(const gsl::span<char>& Buffer)> OnLine;

151 + std::string PendingBuffer;

152 + bool Crlf{};

153 +};

154 +

155 +class HTTPChunkBasedReadHandle : public ReadHandle

156 +{

157 +public:

158 + NON_COPYABLE(HTTPChunkBasedReadHandle);

159 + NON_MOVABLE(HTTPChunkBasedReadHandle);

160 +

161 + HTTPChunkBasedReadHandle(HandleWrapper&& Handler, std::function<void(const gsl::span<char>& Buffer)>&& OnChunk);

162 + ~HTTPChunkBasedReadHandle();

163 +

164 + void OnRead(const gsl::span<char>& Line);

165 +

166 +private:

167 + std::function<void(const gsl::span<char>& Buffer)> OnChunk;

168 + std::string PendingBuffer;

169 + uint64_t PendingChunkSize = 0;

170 + bool ExpectHeader = true;

171 +};

172 +

173 +class ReadSocketMessageHandle : public OverlappedIOHandle

174 +{

175 +public:

176 + NON_COPYABLE(ReadSocketMessageHandle);

177 + NON_MOVABLE(ReadSocketMessageHandle);

178 +

179 + ReadSocketMessageHandle(HandleWrapper&& Socket, std::vector<gsl::byte>& Buffer, std::function<void(const gsl::span<gsl::byte>& Message)>&& OnMessage);

180 + ~ReadSocketMessageHandle();

181 +

182 + void Schedule() override;

183 + void Collect() override;

184 + HANDLE GetHandle() const override;

185 +

186 +private:

187 + void ScheduleRecv();

188 + void ProcessRecvResult(DWORD BytesRead);

189 +

190 + HandleWrapper Socket;

191 + std::vector<gsl::byte>& Buffer;

192 + std::function<void(const gsl::span<gsl::byte>& Message)> OnMessage;

193 + wil::unique_event Event{wil::EventOptions::ManualReset};

194 + OVERLAPPED Overlapped{};

195 + bool ReadingHeader = true;

196 + size_t BytesRemaining = sizeof(MESSAGE_HEADER);

197 + size_t CurrentOffset = 0;

198 +};

199 +

200 +class WriteHandle : public OverlappedIOHandle

201 +{

202 +public:

203 + NON_COPYABLE(WriteHandle);

204 + NON_MOVABLE(WriteHandle);

205 +

206 + WriteHandle(HandleWrapper&& Handle, const std::vector<char>& Buffer = {});

207 + WriteHandle(HandleWrapper&& Handle, gsl::span<gsl::byte> Span);

208 + ~WriteHandle();

209 + void Schedule() override;

210 + void Collect() override;

211 + HANDLE GetHandle() const override;

212 + void Push(const gsl::span<char>& Buffer);

213 +

214 +private:

215 + HandleWrapper Handle;

216 + wil::unique_event Event{wil::EventOptions::ManualReset};

217 + OVERLAPPED Overlapped{};

218 + BufferWrapper Buffer;

219 + LARGE_INTEGER Offset{};

220 +};

221 +

222 +template <typename TRead = ReadHandle>

223 +class RelayHandle : public OverlappedIOHandle

224 +{

225 +public:

226 + NON_COPYABLE(RelayHandle);

227 + NON_MOVABLE(RelayHandle);

228 +

229 + RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output) :

230 + Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output))

231 + {

232 + }

233 +

234 + void Schedule() override

235 + {

236 + WI_ASSERT(State == IOHandleStatus::Standby);

237 +

238 + // If the Buffer is empty, then we're reading.

239 + if (PendingBuffer.empty())

240 + {

241 + // If the output buffer is empty and the reading end is completed, then we're done.

242 + if (Read.GetState() == IOHandleStatus::Completed)

243 + {

244 + State = IOHandleStatus::Completed;

245 + return;

246 + }

247 +

248 + Read.Schedule();

249 +

250 + // If the read is pending, update to 'Pending'

251 + if (Read.GetState() == IOHandleStatus::Pending)

252 + {

253 + State = IOHandleStatus::Pending;

254 + }

255 + }

256 + else

257 + {

258 + Write.Push(PendingBuffer);

259 + PendingBuffer.clear();

260 +

261 + Write.Schedule();

262 +

263 + if (Write.GetState() == IOHandleStatus::Pending)

264 + {

265 + // The write is pending, update to 'Pending'

266 + State = IOHandleStatus::Pending;

267 + }

268 + }

269 + }

270 +

271 + void Collect() override

272 + {

273 + WI_ASSERT(State == IOHandleStatus::Pending);

274 +

275 + // Transition back to standby

276 + State = IOHandleStatus::Standby;

277 +

278 + if (Read.GetState() == IOHandleStatus::Pending)

279 + {

280 + Read.Collect();

281 + }

282 + else

283 + {

284 + WI_ASSERT(Write.GetState() == IOHandleStatus::Pending);

285 + Write.Collect();

286 + }

287 + }

288 +

289 + HANDLE GetHandle() const override

290 + {

291 + if (Read.GetState() == IOHandleStatus::Pending)

292 + {

293 + return Read.GetHandle();

294 + }

295 + else

296 + {

297 + WI_ASSERT(Write.GetState() == IOHandleStatus::Pending);

298 + return Write.GetHandle();

299 + }

300 + }

301 +

302 +private:

303 + void OnRead(const gsl::span<char>& Content)

304 + {

305 + PendingBuffer.insert(PendingBuffer.end(), Content.begin(), Content.end());

306 + }

307 +

308 + TRead Read;

309 + WriteHandle Write;

310 + std::vector<char> PendingBuffer;

311 +};

312 +

313 +class DockerIORelayHandle : public OverlappedIOHandle

314 +{

315 +public:

316 + NON_COPYABLE(DockerIORelayHandle);

317 + NON_MOVABLE(DockerIORelayHandle);

318 +

319 + enum class Format

320 + {

321 + Raw,

322 + HttpChunked

323 + };

324 +

325 + DockerIORelayHandle(HandleWrapper&& Input, HandleWrapper&& Stdout, HandleWrapper&& Stderr, Format ReadFormat);

326 + void Schedule() override;

327 + void Collect() override;

328 + HANDLE GetHandle() const override;

329 +

330 +#pragma pack(push, 1)

331 + struct MultiplexedHeader

332 + {

333 + uint8_t Fd;

334 + char Zeroes[3];

335 + uint32_t Length;

336 + };

337 +#pragma pack(pop)

338 +

339 + static_assert(sizeof(MultiplexedHeader) == 8);

340 +

341 +private:

342 + void OnRead(const gsl::span<char>& Buffer);

343 + void ProcessNextHeader();

344 +

345 + std::unique_ptr<OverlappedIOHandle> Read;

346 + WriteHandle WriteStdout;

347 + WriteHandle WriteStderr;

348 + std::vector<char> PendingBuffer;

349 + WriteHandle* ActiveHandle = nullptr;

350 + size_t RemainingBytes = 0;

351 +};

352 +

353 +class MultiHandleWait

354 +{

355 +public:

356 + NON_COPYABLE(MultiHandleWait);

357 + DEFAULT_MOVABLE(MultiHandleWait);

358 +

359 + enum Flags

360 + {

361 + None = 0,

362 + CancelOnCompleted = 1,

363 + IgnoreErrors = 2,

364 + NeedNotComplete = 4,

365 + };

366 +

367 + MultiHandleWait() = default;

368 +

369 + void AddHandle(std::unique_ptr<OverlappedIOHandle>&& handle, Flags flags = Flags::None);

370 + bool Run(std::optional<std::chrono::milliseconds> Timeout);

371 + void Cancel();

372 +

373 +private:

374 + std::vector<std::pair<Flags, std::unique_ptr<OverlappedIOHandle>>> m_handles;

375 + bool m_cancel = false;

376 +};

377 +

378 +DEFINE_ENUM_FLAG_OPERATORS(MultiHandleWait::Flags);

379 +

380 +} // namespace wsl::windows::common::io

src/windows/common/SubProcess.cpp

+4 -4

@@ -16,7 +16,7 @@ Abstract:

17 #include "SubProcess.h"

19 -using namespace wsl::windows::common::relay;

19 +using namespace wsl::windows::common::io;

20 using wsl::windows::common::SubProcess;

22 SubProcess::SubProcess(LPCWSTR ApplicationName, LPCWSTR CommandLine, DWORD Flags, DWORD StartupFlags) :

@@ -224,11 +224,11 @@ SubProcess::ProcessOutput SubProcess::RunAndCaptureOutput(DWORD Timeout, HANDLE

224

225 m_stdOut = stdoutWrite.get();

226

227 - relay::MultiHandleWait io;

227 + io::MultiHandleWait io;

228 std::string stdoutNative;

229 std::string stderrNative;

230

231 - io.AddHandle(std::make_unique<relay::ReadHandle>(

231 + io.AddHandle(std::make_unique<io::ReadHandle>(

232 std::move(stdoutRead), [&](const gsl::span<char>& buffer) { stdoutNative.append(buffer.data(), buffer.size()); }));

233

234 wil::unique_hfile stderrWrite;

@@ -240,7 +240,7 @@ SubProcess::ProcessOutput SubProcess::RunAndCaptureOutput(DWORD Timeout, HANDLE

240

241 m_stdErr = stderrWrite.get();

242

243 - io.AddHandle(std::make_unique<relay::ReadHandle>(

243 + io.AddHandle(std::make_unique<io::ReadHandle>(

244 std::move(stderrRead), [&](const gsl::span<char>& buffer) { stderrNative.append(buffer.data(), buffer.size()); }));

245 }

246 else

src/windows/common/WSLCProcessLauncher.cpp

+4 -4

@@ -132,11 +132,11 @@ int RunningWSLCProcess::Wait(DWORD TimeoutMs)

132 return GetExitCode();

133 }

134

135 -RunningWSLCProcess::ProcessResult RunningWSLCProcess::WaitAndCaptureOutput(DWORD TimeoutMs, std::vector<std::unique_ptr<relay::OverlappedIOHandle>>&& ExtraHandles)

135 +RunningWSLCProcess::ProcessResult RunningWSLCProcess::WaitAndCaptureOutput(DWORD TimeoutMs, std::vector<std::unique_ptr<io::OverlappedIOHandle>>&& ExtraHandles)

136 {

137 RunningWSLCProcess::ProcessResult result;

138

139 - relay::MultiHandleWait io;

139 + io::MultiHandleWait io;

140

141 // Add a callback on IO for each std handle.

142

@@ -148,7 +148,7 @@ RunningWSLCProcess::ProcessResult RunningWSLCProcess::WaitAndCaptureOutput(DWORD

148 result.Output[Index].insert(result.Output[Index].end(), Content.begin(), Content.end());

149 };

150

151 - io.AddHandle(std::make_unique<relay::ReadHandle>(std::move(stdHandle), std::move(ioCallback)));

151 + io.AddHandle(std::make_unique<io::ReadHandle>(std::move(stdHandle), std::move(ioCallback)));

152 };

153

154 if (WI_IsFlagSet(m_flags, WSLCProcessFlagsTty))

@@ -169,7 +169,7 @@ RunningWSLCProcess::ProcessResult RunningWSLCProcess::WaitAndCaptureOutput(DWORD

169 // Add a callback for when the process exits.

170 auto exitCallback = [&]() { result.Code = GetExitCode(); };

171

172 - io.AddHandle(std::make_unique<relay::EventHandle>(GetExitEvent(), std::move(exitCallback)));

172 + io.AddHandle(std::make_unique<io::EventHandle>(GetExitEvent(), std::move(exitCallback)));

173

174 io.Run(std::chrono::milliseconds(TimeoutMs));

175

src/windows/common/WSLCProcessLauncher.h

+1 -1

@@ -35,7 +35,7 @@ public:

35 NON_COPYABLE(RunningWSLCProcess);

36 DEFAULT_MOVABLE(RunningWSLCProcess);

38 - ProcessResult WaitAndCaptureOutput(DWORD TimeoutMs = INFINITE, std::vector<std::unique_ptr<relay::OverlappedIOHandle>>&& ExtraHandles = {});

38 + ProcessResult WaitAndCaptureOutput(DWORD TimeoutMs = INFINITE, std::vector<std::unique_ptr<io::OverlappedIOHandle>>&& ExtraHandles = {});

39 int Wait(DWORD TimeoutMs = INFINITE);

40 virtual wil::unique_handle GetStdHandle(int Index) = 0;

41 virtual wil::unique_event GetExitEvent() = 0;

src/windows/common/relay.cpp

-871

@@ -16,21 +16,8 @@ Abstract:

16 #include "relay.hpp"

17 #pragma hdrstop

19 -using wsl::windows::common::relay::DockerIORelayHandle;

20 -using wsl::windows::common::relay::EventHandle;

21 -using wsl::windows::common::relay::HandleWrapper;

22 -using wsl::windows::common::relay::HTTPChunkBasedReadHandle;

23 -using wsl::windows::common::relay::IOHandleStatus;

24 -using wsl::windows::common::relay::LineBasedReadHandle;

25 -using wsl::windows::common::relay::MultiHandleWait;

26 -using wsl::windows::common::relay::OverlappedIOHandle;

27 -using wsl::windows::common::relay::ReadHandle;

28 -using wsl::windows::common::relay::ReadSocketMessageHandle;

29 -using wsl::windows::common::relay::RelayHandle;

19 using wsl::windows::common::relay::ScopedMultiRelay;

20 using wsl::windows::common::relay::ScopedRelay;

32 -using wsl::windows::common::relay::SingleAcceptHandle;

33 -using wsl::windows::common::relay::WriteHandle;

22 namespace {

@@ -991,861 +978,3 @@ try

978 }

979 }

980 CATCH_LOG()

994 -

995 -void MultiHandleWait::AddHandle(std::unique_ptr<OverlappedIOHandle>&& handle, Flags flags)

996 -{

997 - m_handles.emplace_back(flags, std::move(handle));

998 -}

999 -

1000 -void MultiHandleWait::Cancel()

1001 -{

1002 - m_cancel = true;

1003 -}

1004 -bool MultiHandleWait::Run(std::optional<std::chrono::milliseconds> Timeout)

1005 -{

1006 - m_cancel = false; // Run may be called multiple times.

1007 -

1008 - std::optional<std::chrono::steady_clock::time_point> deadline;

1009 -

1010 - if (Timeout.has_value())

1011 - {

1012 - deadline = std::chrono::steady_clock::now() + Timeout.value();

1013 - }

1014 -

1015 - // Run until all handles are completed.

1016 -

1017 - while (!m_handles.empty() && !m_cancel)

1018 - {

1019 - // Schedule IO on each handle until all are either pending, or completed.

1020 - for (size_t i = 0; i < m_handles.size() && !m_cancel; i++)

1021 - {

1022 - while (m_handles[i].second->GetState() == IOHandleStatus::Standby && !m_cancel)

1023 - {

1024 - try

1025 - {

1026 - m_handles[i].second->Schedule();

1027 - }

1028 - catch (...)

1029 - {

1030 - if (WI_IsFlagSet(m_handles[i].first, Flags::IgnoreErrors))

1031 - {

1032 - m_handles[i].second.reset(); // Reset the handle so it can be deleted.

1033 - break;

1034 - }

1035 - else

1036 - {

1037 - throw;

1038 - }

1039 - }

1040 - }

1041 - }

1042 -

1043 - // Remove completed handles from m_handles.

1044 - bool hasHandleToWaitFor = false;

1045 - for (auto it = m_handles.begin(); it != m_handles.end();)

1046 - {

1047 - if (!it->second)

1048 - {

1049 - it = m_handles.erase(it);

1050 - }

1051 - else if (it->second->GetState() == IOHandleStatus::Completed)

1052 - {

1053 - if (WI_IsFlagSet(it->first, Flags::CancelOnCompleted))

1054 - {

1055 - m_cancel = true; // Cancel the IO if a handle with CancelOnCompleted is in the completed state.

1056 - }

1057 -

1058 - it = m_handles.erase(it);

1059 - }

1060 - else

1061 - {

1062 - // If only NeedNotComplete handles are left, we want to exit Run.

1063 - if (WI_IsFlagClear(it->first, Flags::NeedNotComplete))

1064 - {

1065 - hasHandleToWaitFor = true;

1066 - }

1067 - ++it;

1068 - }

1069 - }

1070 -

1071 - if (!hasHandleToWaitFor || m_cancel)

1072 - {

1073 - break;

1074 - }

1075 -

1076 - // Wait for the next operation to complete.

1077 - std::vector<HANDLE> waitHandles;

1078 - for (const auto& e : m_handles)

1079 - {

1080 - waitHandles.emplace_back(e.second->GetHandle());

1081 - }

1082 -

1083 - DWORD waitTimeout = INFINITE;

1084 - if (deadline.has_value())

1085 - {

1086 - auto miliseconds =

1087 - std::chrono::duration_cast<std::chrono::milliseconds>(deadline.value() - std::chrono::steady_clock::now()).count();

1088 -

1089 - waitTimeout = static_cast<DWORD>(std::max(0LL, miliseconds));

1090 - }

1091 -

1092 - auto result = WaitForMultipleObjects(static_cast<DWORD>(waitHandles.size()), waitHandles.data(), false, waitTimeout);

1093 - if (result == WAIT_TIMEOUT)

1094 - {

1095 - THROW_WIN32(ERROR_TIMEOUT);

1096 - }

1097 - else if (result >= WAIT_OBJECT_0 && result < WAIT_OBJECT_0 + m_handles.size())

1098 - {

1099 - auto index = result - WAIT_OBJECT_0;

1100 -

1101 - try

1102 - {

1103 - m_handles[index].second->Collect();

1104 - }

1105 - catch (...)

1106 - {

1107 - if (WI_IsFlagSet(m_handles[index].first, Flags::IgnoreErrors))

1108 - {

1109 - m_handles.erase(m_handles.begin() + index);

1110 - }

1111 - else

1112 - {

1113 - throw;

1114 - }

1115 - }

1116 - }

1117 - else

1118 - {

1119 - THROW_LAST_ERROR_MSG("Timeout: %lu, Count: %llu", waitTimeout, waitHandles.size());

1120 - }

1121 - }

1122 -

1123 - return !m_cancel;

1124 -}

1125 -

1126 -IOHandleStatus OverlappedIOHandle::GetState() const

1127 -{

1128 - return State;

1129 -}

1130 -

1131 -EventHandle::EventHandle(HandleWrapper&& Handle, std::function<void()>&& OnSignalled) :

1132 - Handle(std::move(Handle)), OnSignalled(std::move(OnSignalled))

1133 -{

1134 -}

1135 -

1136 -void EventHandle::Schedule()

1137 -{

1138 - State = IOHandleStatus::Pending;

1139 -}

1140 -

1141 -void EventHandle::Collect()

1142 -{

1143 - State = IOHandleStatus::Completed;

1144 - OnSignalled();

1145 -}

1146 -

1147 -HANDLE EventHandle::GetHandle() const

1148 -{

1149 - return Handle.Get();

1150 -}

1151 -

1152 -ReadHandle::ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead) :

1153 - Handle(std::move(MovedHandle)), OnRead(OnRead), Offset(InitializeFileOffset(Handle.Get()))

1154 -{

1155 - Overlapped.hEvent = Event.get();

1156 -}

1157 -

1158 -ReadHandle::~ReadHandle()

1159 -{

1160 - if (State == IOHandleStatus::Pending)

1161 - {

1162 - CancelPendingIo(Handle.Get(), Overlapped);

1163 - }

1164 -}

1165 -

1166 -void ReadHandle::Schedule()

1167 -{

1168 - WI_ASSERT(State == IOHandleStatus::Standby);

1169 -

1170 - Event.ResetEvent();

1171 -

1172 - // Schedule the read.

1173 - DWORD bytesRead{};

1174 - Overlapped.Offset = Offset.LowPart;

1175 - Overlapped.OffsetHigh = Offset.HighPart;

1176 - auto* bufferData = reinterpret_cast<char*>(Buffer.Span().data());

1177 - if (ReadFile(Handle.Get(), bufferData, static_cast<DWORD>(Buffer.Size()), &bytesRead, &Overlapped))

1178 - {

1179 - Offset.QuadPart += bytesRead;

1180 -

1181 - // Signal the read.

1182 - OnRead(gsl::make_span<char>(bufferData, static_cast<size_t>(bytesRead)));

1183 -

1184 - // ReadFile completed immediately, process the result right away.

1185 - if (bytesRead == 0)

1186 - {

1187 - State = IOHandleStatus::Completed;

1188 - return; // Handle is completely read, don't try again.

1189 - }

1190 -

1191 - // Read was done synchronously, remain in 'standby' state.

1192 - }

1193 - else

1194 - {

1195 - auto error = GetLastError();

1196 - if (error == ERROR_HANDLE_EOF || error == ERROR_BROKEN_PIPE)

1197 - {

1198 - // Signal an empty read for EOF.

1199 - OnRead({});

1200 -

1201 - State = IOHandleStatus::Completed;

1202 - return;

1203 - }

1204 -

1205 - THROW_LAST_ERROR_IF_MSG(error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)Handle.Get());

1206 -

1207 - // The read is pending, update to 'Pending'

1208 - State = IOHandleStatus::Pending;

1209 - }

1210 -}

1211 -

1212 -void ReadHandle::Collect()

1213 -{

1214 - WI_ASSERT(State == IOHandleStatus::Pending);

1215 -

1216 - // Transition back to standby

1217 - State = IOHandleStatus::Standby;

1218 -

1219 - // Complete the read.

1220 - DWORD bytesRead{};

1221 - if (!GetOverlappedResult(Handle.Get(), &Overlapped, &bytesRead, false))

1222 - {

1223 - auto error = GetLastError();

1224 - THROW_WIN32_IF(error, error != ERROR_HANDLE_EOF && error != ERROR_BROKEN_PIPE);

1225 -

1226 - // We received ERROR_HANDLE_EOF or ERROR_BROKEN_PIPE. Validate that this was indeed a zero byte read.

1227 - WI_ASSERT(bytesRead == 0);

1228 - }

1229 -

1230 - Offset.QuadPart += bytesRead;

1231 -

1232 - // Signal the read.

1233 - OnRead(gsl::make_span<char>(reinterpret_cast<char*>(Buffer.Span().data()), static_cast<size_t>(bytesRead)));

1234 -

1235 - // Transition to Complete if this was a zero byte read.

1236 - if (bytesRead == 0)

1237 - {

1238 - State = IOHandleStatus::Completed;

1239 - }

1240 -}

1241 -

1242 -HANDLE ReadHandle::GetHandle() const

1243 -{

1244 - return Event.get();

1245 -}

1246 -

1247 -SingleAcceptHandle::SingleAcceptHandle(HandleWrapper&& ListenSocket, HandleWrapper&& AcceptedSocket, std::function<void()>&& OnAccepted) :

1248 - ListenSocket(std::move(ListenSocket)), AcceptedSocket(std::move(AcceptedSocket)), OnAccepted(std::move(OnAccepted))

1249 -{

1250 - Overlapped.hEvent = Event.get();

1251 -}

1252 -

1253 -SingleAcceptHandle::~SingleAcceptHandle()

1254 -{

1255 - if (State == IOHandleStatus::Pending)

1256 - {

1257 - LOG_IF_WIN32_BOOL_FALSE(CancelIoEx(ListenSocket.Get(), &Overlapped));

1258 -

1259 - DWORD bytesProcessed{};

1260 - DWORD flagsReturned{};

1261 - if (!WSAGetOverlappedResult((SOCKET)ListenSocket.Get(), &Overlapped, &bytesProcessed, TRUE, &flagsReturned))

1262 - {

1263 - auto error = GetLastError();

1264 - LOG_LAST_ERROR_IF(error != ERROR_CONNECTION_ABORTED && error != ERROR_OPERATION_ABORTED);

1265 - }

1266 - }

1267 -}

1268 -

1269 -void SingleAcceptHandle::Schedule()

1270 -{

1271 - WI_ASSERT(State == IOHandleStatus::Standby);

1272 -

1273 - // Schedule the accept.

1274 - DWORD bytesReturned{};

1275 - if (AcceptEx((SOCKET)ListenSocket.Get(), (SOCKET)AcceptedSocket.Get(), &AcceptBuffer, 0, sizeof(SOCKADDR_STORAGE), sizeof(SOCKADDR_STORAGE), &bytesReturned, &Overlapped))

1276 - {

1277 - // Accept completed immediately.

1278 - State = IOHandleStatus::Completed;

1279 - OnAccepted();

1280 - }

1281 - else

1282 - {

1283 - auto error = WSAGetLastError();

1284 - THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)ListenSocket.Get());

1285 -

1286 - State = IOHandleStatus::Pending;

1287 - }

1288 -}

1289 -

1290 -void SingleAcceptHandle::Collect()

1291 -{

1292 - WI_ASSERT(State == IOHandleStatus::Pending);

1293 -

1294 - DWORD bytesReceived{};

1295 - DWORD flagsReturned{};

1296 -

1297 - THROW_IF_WIN32_BOOL_FALSE(WSAGetOverlappedResult((SOCKET)ListenSocket.Get(), &Overlapped, &bytesReceived, false, &flagsReturned));

1298 -

1299 - State = IOHandleStatus::Completed;

1300 - OnAccepted();

1301 -}

1302 -

1303 -HANDLE SingleAcceptHandle::GetHandle() const

1304 -{

1305 - return Event.get();

1306 -}

1307 -

1308 -LineBasedReadHandle::LineBasedReadHandle(HandleWrapper&& Handle, std::function<void(const gsl::span<char>& Line)>&& OnLine, bool Crlf) :

1309 - ReadHandle(std::move(Handle), [this](const gsl::span<char>& Buffer) { OnRead(Buffer); }), OnLine(OnLine), Crlf(Crlf)

1310 -{

1311 -}

1312 -

1313 -LineBasedReadHandle::~LineBasedReadHandle()

1314 -{

1315 - // N.B. PendingBuffer can contain remaining data is an exception was thrown during parsing.

1316 -}

1317 -

1318 -void LineBasedReadHandle::OnRead(const gsl::span<char>& Buffer)

1319 -{

1320 - // If we reach of the end, signal a line with the remaining buffer.

1321 - if (Buffer.empty() && !PendingBuffer.empty())

1322 - {

1323 - OnLine(PendingBuffer);

1324 - PendingBuffer.clear();

1325 - return;

1326 - }

1327 -

1328 - auto begin = Buffer.begin();

1329 - auto end = std::ranges::find(Buffer, Crlf ? '\r' : '\n');

1330 - while (end != Buffer.end())

1331 - {

1332 - if (Crlf)

1333 - {

1334 - end++; // Move to the following '\n'

1335 -

1336 - if (end == Buffer.end() || *end != '\n') // Incomplete CRLF sequence. Append to buffer and continue.

1337 - {

1338 - PendingBuffer.insert(PendingBuffer.end(), begin, end);

1339 - begin = end;

1340 - end = std::ranges::find(end, Buffer.end(), '\r');

1341 - continue;

1342 - }

1343 - }

1344 -

1345 - // Discard the '\r' in CRLF mode.

1346 - PendingBuffer.insert(PendingBuffer.end(), begin, Crlf ? end - 1 : end);

1347 -

1348 - if (!PendingBuffer.empty())

1349 - {

1350 - OnLine(PendingBuffer);

1351 - PendingBuffer.clear();

1352 - }

1353 -

1354 - begin = end + 1;

1355 - end = std::ranges::find(begin, Buffer.end(), Crlf ? '\r' : '\n');

1356 - }

1357 -

1358 - PendingBuffer.insert(PendingBuffer.end(), begin, end);

1359 -}

1360 -

1361 -HTTPChunkBasedReadHandle::HTTPChunkBasedReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Line)>&& OnChunk) :

1362 - ReadHandle(std::move(MovedHandle), [this](const gsl::span<char>& Buffer) { OnRead(Buffer); }), OnChunk(std::move(OnChunk))

1363 -{

1364 -}

1365 -

1366 -HTTPChunkBasedReadHandle::~HTTPChunkBasedReadHandle()

1367 -{

1368 - // N.B. PendingBuffer can contain remaining data is an exception was thrown during parsing.

1369 - LOG_HR_IF(E_UNEXPECTED, !PendingBuffer.empty() || PendingChunkSize != 0 || ExpectHeader);

1370 -}

1371 -

1372 -void HTTPChunkBasedReadHandle::OnRead(const gsl::span<char>& Input)

1373 -{

1374 - // See: https://httpwg.org/specs/rfc9112.html#field.transfer-encoding

1375 -

1376 - if (Input.empty())

1377 - {

1378 - // N.B. The body can be terminated by a zero-length chunk.

1379 - THROW_HR_IF(E_INVALIDARG, PendingChunkSize != 0 || ExpectHeader);

1380 - }

1381 -

1382 - auto buffer = Input;

1383 -

1384 - auto advance = [&](size_t count) {

1385 - WI_ASSERT(buffer.size() >= count);

1386 - buffer = buffer.subspan(count);

1387 - };

1388 -

1389 - while (!buffer.empty())

1390 - {

1391 - if (PendingChunkSize == 0)

1392 - {

1393 - // Consume CRLF's between chunks.

1394 - if (PendingBuffer.empty() && (buffer.front() == '\r' || buffer.front() == '\n'))

1395 - {

1396 - advance(1);

1397 - continue;

1398 - }

1399 -

1400 - ExpectHeader = true;

1401 -

1402 - auto end = std::ranges::find(buffer, '\n');

1403 - PendingBuffer.insert(PendingBuffer.end(), buffer.begin(), end);

1404 - if (end == buffer.end())

1405 - {

1406 - // Incomplete size header, buffer until next read.

1407 - break;

1408 - }

1409 - // Advance beyond the LF

1410 - advance(end - buffer.begin() + 1);

1411 -

1412 - THROW_HR_IF_MSG(

1413 - E_INVALIDARG,

1414 - PendingBuffer.size() < 2 || PendingBuffer.back() != '\r',

1415 - "Malformed chunk header: %hs",

1416 - PendingBuffer.c_str());

1417 - PendingBuffer.erase(PendingBuffer.end() - 1, PendingBuffer.end()); // Remove CR.

1418 -

1419 -#ifdef WSLC_HTTP_DEBUG

1420 -

1421 - WSL_LOG("HTTPChunkHeader", TraceLoggingValue(PendingBuffer.c_str(), "Size"));

1422 -

1423 -#endif

1424 -

1425 - try

1426 - {

1427 - size_t parsed{};

1428 - PendingChunkSize = std::stoul(PendingBuffer.c_str(), &parsed, 16);

1429 - THROW_HR_IF(E_INVALIDARG, parsed != PendingBuffer.size());

1430 - }

1431 - catch (...)

1432 - {

1433 - THROW_HR_MSG(E_INVALIDARG, "Failed to parse chunk size: %hs", PendingBuffer.c_str());

1434 - }

1435 -

1436 - ExpectHeader = false;

1437 - PendingBuffer.clear();

1438 - }

1439 - else

1440 - {

1441 - // Consume the chunk.

1442 - auto consumedBytes = std::min(PendingChunkSize, buffer.size());

1443 - PendingBuffer.append(buffer.data(), consumedBytes);

1444 - advance(consumedBytes);

1445 -

1446 - WI_ASSERT(PendingChunkSize >= consumedBytes);

1447 - PendingChunkSize -= consumedBytes;

1448 -

1449 - if (PendingChunkSize == 0)

1450 - {

1451 -

1452 -#ifdef WSLC_HTTP_DEBUG

1453 -

1454 - WSL_LOG("HTTPChunk", TraceLoggingValue(PendingBuffer.c_str(), "Content"));

1455 -

1456 -#endif

1457 - OnChunk(PendingBuffer);

1458 - PendingBuffer.clear();

1459 - }

1460 - }

1461 - }

1462 -}

1463 -

1464 -ReadSocketMessageHandle::ReadSocketMessageHandle(

1465 - HandleWrapper&& MovedSocket, std::vector<gsl::byte>& Buffer, std::function<void(const gsl::span<gsl::byte>& Message)>&& OnMessage) :

1466 - Socket(std::move(MovedSocket)), Buffer(Buffer), OnMessage(std::move(OnMessage))

1467 -{

1468 - Overlapped.hEvent = Event.get();

1469 -

1470 - if (Buffer.size() < sizeof(MESSAGE_HEADER))

1471 - {

1472 - Buffer.resize(sizeof(MESSAGE_HEADER));

1473 - }

1474 -}

1475 -

1476 -ReadSocketMessageHandle::~ReadSocketMessageHandle()

1477 -{

1478 - if (State == IOHandleStatus::Pending)

1479 - {

1480 - CancelPendingIo((SOCKET)Socket.Get(), Overlapped);

1481 - }

1482 -}

1483 -

1484 -void ReadSocketMessageHandle::ScheduleRecv()

1485 -{

1486 - Event.ResetEvent();

1487 -

1488 - auto target = gsl::make_span(Buffer).subspan(CurrentOffset, BytesRemaining);

1489 - WSABUF wsaBuf = {gsl::narrow_cast<ULONG>(target.size()), reinterpret_cast<CHAR*>(target.data())};

1490 - DWORD bytesRead{};

1491 - DWORD flags = 0;

1492 - if (WSARecv(reinterpret_cast<SOCKET>(Socket.Get()), &wsaBuf, 1, &bytesRead, &flags, &Overlapped, nullptr) == 0)

1493 - {

1494 - ProcessRecvResult(bytesRead);

1495 - }

1496 - else

1497 - {

1498 - auto error = WSAGetLastError();

1499 - if (error == WSAECONNABORTED || error == WSAECONNRESET)

1500 - {

1501 - ProcessRecvResult(0);

1502 - return;

1503 - }

1504 -

1505 - THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != WSA_IO_PENDING, "Socket: 0x%p", (void*)Socket.Get());

1506 -

1507 - State = IOHandleStatus::Pending;

1508 - }

1509 -}

1510 -

1511 -void ReadSocketMessageHandle::ProcessRecvResult(DWORD BytesRead)

1512 -{

1513 - if (BytesRead == 0)

1514 - {

1515 - // If the socket was closed before any bytes of the next message were read, signal a clean end-of-stream.

1516 - // If some bytes were already buffered, the peer closed mid-message which is a protocol error.

1517 - THROW_HR_IF_MSG(

1518 - E_UNEXPECTED,

1519 - CurrentOffset > 0,

1520 - "Socket closed before a complete message could be read. ReadingHeader: %d, CurrentOffset: %zu, BytesRemaining: %zu",

1521 - ReadingHeader,

1522 - CurrentOffset,

1523 - BytesRemaining);

1524 -

1525 - OnMessage({});

1526 - State = IOHandleStatus::Completed;

1527 - return;

1528 - }

1529 -

1530 - CurrentOffset += BytesRead;

1531 - BytesRemaining -= BytesRead;

1532 -

1533 - if (BytesRemaining > 0)

1534 - {

1535 - return;

1536 - }

1537 -

1538 - if (ReadingHeader)

1539 - {

1540 - auto messageSize = gslhelpers::get_struct<MESSAGE_HEADER>(gsl::make_span(Buffer.data(), sizeof(MESSAGE_HEADER)))->MessageSize;

1541 -

1542 - THROW_HR_IF_MSG(E_UNEXPECTED, messageSize < sizeof(MESSAGE_HEADER), "Unexpected message size: %u", messageSize);

1543 - THROW_HR_IF_MSG(E_UNEXPECTED, messageSize > 4 * 1024 * 1024, "Message size too large: %u", messageSize);

1544 -

1545 - if (messageSize == sizeof(MESSAGE_HEADER))

1546 - {

1547 - OnMessage(gsl::make_span(Buffer.data(), messageSize));

1548 - State = IOHandleStatus::Completed;

1549 - return;

1550 - }

1551 -

1552 - if (Buffer.size() < messageSize)

1553 - {

1554 - Buffer.resize(messageSize);

1555 - }

1556 -

1557 - ReadingHeader = false;

1558 - CurrentOffset = sizeof(MESSAGE_HEADER);

1559 - BytesRemaining = messageSize - sizeof(MESSAGE_HEADER);

1560 - }

1561 - else

1562 - {

1563 - auto messageSize = gslhelpers::get_struct<MESSAGE_HEADER>(gsl::make_span(Buffer.data(), sizeof(MESSAGE_HEADER)))->MessageSize;

1564 - OnMessage(gsl::make_span(Buffer.data(), messageSize));

1565 - State = IOHandleStatus::Completed;

1566 - }

1567 -}

1568 -

1569 -void ReadSocketMessageHandle::Schedule()

1570 -{

1571 - WI_ASSERT(State == IOHandleStatus::Standby);

1572 - ScheduleRecv();

1573 -}

1574 -

1575 -void ReadSocketMessageHandle::Collect()

1576 -{

1577 - WI_ASSERT(State == IOHandleStatus::Pending);

1578 -

1579 - State = IOHandleStatus::Standby;

1580 -

1581 - DWORD bytesRead{};

1582 - DWORD flags{};

1583 - if (!WSAGetOverlappedResult(reinterpret_cast<SOCKET>(Socket.Get()), &Overlapped, &bytesRead, FALSE, &flags))

1584 - {

1585 - long error = WSAGetLastError();

1586 - THROW_WIN32_IF(error, error != WSAECONNABORTED && error != WSAECONNRESET);

1587 -

1588 - WI_ASSERT(bytesRead == 0);

1589 - }

1590 -

1591 - ProcessRecvResult(bytesRead);

1592 -}

1593 -

1594 -HANDLE ReadSocketMessageHandle::GetHandle() const

1595 -{

1596 - return Event.get();

1597 -}

1598 -

1599 -WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, const std::vector<char>& Source) :

1600 - Handle(std::move(MovedHandle)), Buffer(Source.size()), Offset(InitializeFileOffset(Handle.Get()))

1601 -{

1602 - std::memcpy(Buffer.Span().data(), Source.data(), Source.size());

1603 - Overlapped.hEvent = Event.get();

1604 -}

1605 -

1606 -WriteHandle::WriteHandle(HandleWrapper&& MovedHandle, gsl::span<gsl::byte> Source) :

1607 - Handle(std::move(MovedHandle)), Buffer(Source), Offset(InitializeFileOffset(Handle.Get()))

1608 -{

1609 - Overlapped.hEvent = Event.get();

1610 -}

1611 -

1612 -WriteHandle::~WriteHandle()

1613 -{

1614 - if (State == IOHandleStatus::Pending)

1615 - {

1616 - CancelPendingIo(Handle.Get(), Overlapped);

1617 - }

1618 -}

1619 -

1620 -void WriteHandle::Schedule()

1621 -{

1622 - WI_ASSERT(State == IOHandleStatus::Standby);

1623 -

1624 - Event.ResetEvent();

1625 -

1626 - Overlapped.Offset = Offset.LowPart;

1627 - Overlapped.OffsetHigh = Offset.HighPart;

1628 -

1629 - // Schedule the write.

1630 - const auto buffer = Buffer.Span();

1631 - DWORD bytesWritten{};

1632 - if (WriteFile(Handle.Get(), buffer.data(), static_cast<DWORD>(buffer.size()), &bytesWritten, &Overlapped))

1633 - {

1634 - Offset.QuadPart += bytesWritten;

1635 -

1636 - Buffer.Consume(bytesWritten);

1637 - if (Buffer.Size() == 0)

1638 - {

1639 - State = IOHandleStatus::Completed;

1640 - }

1641 - }

1642 - else

1643 - {

1644 - auto error = GetLastError();

1645 - THROW_LAST_ERROR_IF_MSG(error != ERROR_IO_PENDING, "Handle: 0x%p", (void*)Handle.Get());

1646 -

1647 - // The write is pending, update to 'Pending'

1648 - State = IOHandleStatus::Pending;

1649 - }

1650 -}

1651 -

1652 -void WriteHandle::Collect()

1653 -{

1654 - WI_ASSERT(State == IOHandleStatus::Pending);

1655 -

1656 - // Transition back to standby

1657 - State = IOHandleStatus::Standby;

1658 -

1659 - // Complete the write.

1660 - DWORD bytesWritten{};

1661 - THROW_IF_WIN32_BOOL_FALSE(GetOverlappedResult(Handle.Get(), &Overlapped, &bytesWritten, false));

1662 - Offset.QuadPart += bytesWritten;

1663 -

1664 - Buffer.Consume(bytesWritten);

1665 - if (Buffer.Size() == 0)

1666 - {

1667 - State = IOHandleStatus::Completed;

1668 - }

1669 -}

1670 -

1671 -void WriteHandle::Push(const gsl::span<char>& Content)

1672 -{

1673 - // Don't write if a WriteFile() is pending, since that could cause the buffer to reallocate.

1674 - WI_ASSERT(State == IOHandleStatus::Standby || State == IOHandleStatus::Completed);

1675 - WI_ASSERT(!Content.empty());

1676 -

1677 - // Resize() throws E_UNEXPECTED if Buffer does not own its storage.

1678 - Buffer.Append(Content);

1679 -

1680 - State = IOHandleStatus::Standby;

1681 -}

1682 -

1683 -HANDLE WriteHandle::GetHandle() const

1684 -{

1685 - return Event.get();

1686 -}

1687 -

1688 -DockerIORelayHandle::DockerIORelayHandle(HandleWrapper&& ReadHandle, HandleWrapper&& Stdout, HandleWrapper&& Stderr, Format ReadFormat) :

1689 - WriteStdout(std::move(Stdout)), WriteStderr(std::move(Stderr))

1690 -{

1691 - if (ReadFormat == Format::HttpChunked)

1692 - {

1693 - Read = std::make_unique<HTTPChunkBasedReadHandle>(

1694 - std::move(ReadHandle), [this](const gsl::span<char>& Line) { this->OnRead(Line); });

1695 - }

1696 - else

1697 - {

1698 - Read = std::make_unique<relay::ReadHandle>(

1699 - std::move(ReadHandle), [this](const gsl::span<char>& Buffer) { this->OnRead(Buffer); });

1700 - }

1701 -}

1702 -

1703 -void DockerIORelayHandle::Schedule()

1704 -{

1705 - WI_ASSERT(State == IOHandleStatus::Standby);

1706 - WI_ASSERT(Read->GetState() != IOHandleStatus::Pending);

1707 -

1708 - // If we have an active handle and a buffer, try to flush that first.

1709 - if (ActiveHandle != nullptr && !PendingBuffer.empty())

1710 - {

1711 - // Push the data to the selected handle.

1712 - DWORD bytesToWrite = std::min(static_cast<DWORD>(RemainingBytes), static_cast<DWORD>(PendingBuffer.size()));

1713 -

1714 - ActiveHandle->Push(gsl::make_span(PendingBuffer.data(), bytesToWrite));

1715 -

1716 - // Consume the written bytes.

1717 - RemainingBytes -= bytesToWrite;

1718 - PendingBuffer.erase(PendingBuffer.begin(), PendingBuffer.begin() + bytesToWrite);

1719 -

1720 - // Schedule the write.

1721 - ActiveHandle->Schedule();

1722 -

1723 - // If the write is pending, update to 'Pending'

1724 - if (ActiveHandle->GetState() == IOHandleStatus::Pending)

1725 - {

1726 - State = IOHandleStatus::Pending;

1727 - }

1728 - else if (ActiveHandle->GetState() == IOHandleStatus::Completed)

1729 - {

1730 - if (RemainingBytes == 0)

1731 - {

1732 - // Switch back to reading if we've written all bytes for this chunk.

1733 - ActiveHandle = nullptr;

1734 -

1735 - ProcessNextHeader();

1736 - }

1737 - }

1738 - }

1739 - else

1740 - {

1741 - if (Read->GetState() == IOHandleStatus::Completed)

1742 - {

1743 - LOG_HR_IF(E_UNEXPECTED, ActiveHandle != nullptr);

1744 -

1745 - // No more data to read, we're done.

1746 - State = IOHandleStatus::Completed;

1747 - return;

1748 - }

1749 -

1750 - // Schedule a read from the input.

1751 - Read->Schedule();

1752 - if (Read->GetState() == IOHandleStatus::Pending)

1753 - {

1754 - State = IOHandleStatus::Pending;

1755 - }

1756 - }

1757 -}

1758 -

1759 -void DockerIORelayHandle::Collect()

1760 -{

1761 - WI_ASSERT(State == IOHandleStatus::Pending);

1762 -

1763 - if (ActiveHandle != nullptr && ActiveHandle->GetState() == IOHandleStatus::Pending)

1764 - {

1765 - // Complete the write.

1766 - ActiveHandle->Collect();

1767 -

1768 - // If the write is completed, switch back to reading.

1769 - if (RemainingBytes == 0)

1770 - {

1771 - if (ActiveHandle->GetState() == IOHandleStatus::Completed)

1772 - {

1773 - ActiveHandle = nullptr;

1774 - }

1775 - }

1776 -

1777 - // Transition back to standby if there's still data to read.

1778 - // Otherwise switch to Completed since everything is done.

1779 - if (Read->GetState() == IOHandleStatus::Completed)

1780 - {

1781 - LOG_HR_IF(E_UNEXPECTED, RemainingBytes != 0);

1782 -

1783 - State = IOHandleStatus::Completed;

1784 - }

1785 - else

1786 - {

1787 - State = IOHandleStatus::Standby;

1788 - }

1789 - }

1790 - else

1791 - {

1792 - WI_ASSERT(Read->GetState() == IOHandleStatus::Pending);

1793 -

1794 - // Complete the read.

1795 - Read->Collect();

1796 -

1797 - // Transition back to standby.

1798 - State = IOHandleStatus::Standby;

1799 - }

1800 -}

1801 -

1802 -HANDLE DockerIORelayHandle::GetHandle() const

1803 -{

1804 - if (ActiveHandle != nullptr && ActiveHandle->GetState() == IOHandleStatus::Pending)

1805 - {

1806 - return ActiveHandle->GetHandle();

1807 - }

1808 - else

1809 - {

1810 - return Read->GetHandle();

1811 - }

1812 -}

1813 -

1814 -void DockerIORelayHandle::ProcessNextHeader()

1815 -{

1816 - if (PendingBuffer.size() < sizeof(MultiplexedHeader))

1817 - {

1818 - // Not enough data for a header yet.

1819 - return;

1820 - }

1821 -

1822 - const auto* header = reinterpret_cast<const MultiplexedHeader*>(PendingBuffer.data());

1823 - RemainingBytes = ntohl(header->Length);

1824 -

1825 - if (header->Fd == 1)

1826 - {

1827 - ActiveHandle = &WriteStdout;

1828 - }

1829 - else if (header->Fd == 2)

1830 - {

1831 - ActiveHandle = &WriteStderr;

1832 - }

1833 - else

1834 - {

1835 - THROW_HR_MSG(E_INVALIDARG, "Invalid Docker IO multiplexed header fd: %u", header->Fd);

1836 - }

1837 -

1838 - // Consume the header.

1839 - PendingBuffer.erase(PendingBuffer.begin(), PendingBuffer.begin() + sizeof(MultiplexedHeader));

1840 -}

1841 -

1842 -void DockerIORelayHandle::OnRead(const gsl::span<char>& Buffer)

1843 -{

1844 - PendingBuffer.insert(PendingBuffer.end(), Buffer.begin(), Buffer.end());

1845 -

1846 - if (ActiveHandle == nullptr)

1847 - {

1848 - // If no handle is active, expect a header.

1849 - ProcessNextHeader();

1850 - }

1851 -}

src/windows/common/relay.hpp

+3 -457

@@ -16,11 +16,12 @@ Abstract:

17 #include <winsock2.h>

18 #include "ConsoleState.h"

19 -

20 -#define LX_RELAY_BUFFER_SIZE 0x1000

19 +#include "HandleIO.h"

21 namespace wsl::windows::common::relay {

23 +using namespace wsl::windows::common::io;

24 +

25 std::thread CreateThread(_In_ HANDLE InputHandle, _In_ HANDLE OutputHandle, _In_opt_ HANDLE ExitHandle = nullptr, _In_ size_t BufferSize = LX_RELAY_BUFFER_SIZE);

27 std::thread CreateThread(_In_ wil::unique_handle&& InputHandle, _In_ HANDLE OutputHandle, _In_opt_ HANDLE ExitHandle = nullptr, _In_ size_t BufferSize = LX_RELAY_BUFFER_SIZE);

@@ -157,459 +158,4 @@ private:

158 std::function<void()> m_onDestroy;

159 };

160

160 -enum class IOHandleStatus

161 -{

162 - Standby,

163 - Pending,

164 - Completed

165 -};

166 -

167 -struct HandleWrapper

168 -{

169 - DEFAULT_MOVABLE(HandleWrapper);

170 - NON_COPYABLE(HandleWrapper)

171 -

172 - HandleWrapper(

173 - wil::unique_handle&& handle, std::function<void()>&& OnClose = []() {}) :

174 - Handle(handle.get()), OwnedHandle(std::move(handle)), OnClose(std::move(OnClose))

175 - {

176 - }

177 -

178 - HandleWrapper(

179 - wil::unique_socket&& handle, std::function<void()>&& OnClose = []() {}) :

180 - Handle((HANDLE)handle.get()), OwnedHandle(wil::unique_socket{handle.release()}), OnClose(std::move(OnClose))

181 - {

182 - }

183 -

184 - HandleWrapper(

185 - wil::unique_event&& handle, std::function<void()>&& OnClose = []() {}) :

186 - Handle(handle.get()), OwnedHandle(wil::unique_handle{handle.release()}), OnClose(std::move(OnClose))

187 - {

188 - }

189 -

190 - HandleWrapper(

191 - SOCKET handle, std::function<void()>&& OnClose = []() {}) :

192 - Handle(reinterpret_cast<HANDLE>(handle)), OnClose(std::move(OnClose))

193 - {

194 - }

195 -

196 - HandleWrapper(HANDLE handle, std::function<void()>&& OnClose = []() {}) : Handle(handle), OnClose(std::move(OnClose))

197 - {

198 - }

199 -

200 - HandleWrapper(

201 - wil::unique_hfile&& handle, std::function<void()>&& OnClose = []() {}) :

202 - Handle(handle.get()), OwnedHandle(wil::unique_handle{handle.release()}), OnClose(std::move(OnClose))

203 - {

204 - }

205 -

206 - ~HandleWrapper()

207 - {

208 - Reset();

209 - }

210 -

211 - HANDLE Get() const

212 - {

213 - return Handle;

214 - }

215 -

216 - void Reset()

217 - {

218 - if (OnClose != nullptr)

219 - {

220 - OnClose();

221 - OnClose = nullptr;

222 - }

223 -

224 - OwnedHandle = {};

225 - Handle = nullptr;

226 - }

227 -

228 -private:

229 - HANDLE Handle{};

230 - std::variant<wil::unique_handle, wil::unique_socket> OwnedHandle;

231 - std::function<void()> OnClose;

232 -};

233 -

234 -// A buffer that may either own its underlying storage (constructed from a size, allocating an

235 -// internal std::vector<char>) or borrow it from a caller-provided gsl::span<gsl::byte>.

236 -class BufferWrapper

237 -{

238 -public:

239 - DEFAULT_MOVABLE(BufferWrapper);

240 - NON_COPYABLE(BufferWrapper);

241 -

242 - explicit BufferWrapper(size_t size) : m_owned(std::in_place, size)

243 - {

244 - }

245 -

246 - explicit BufferWrapper(gsl::span<gsl::byte> span) : m_unowned(span)

247 - {

248 - }

249 -

250 - bool Owned() const noexcept

251 - {

252 - return m_owned.has_value();

253 - }

254 -

255 - void Resize(size_t size)

256 - {

257 - THROW_HR_IF_MSG(E_UNEXPECTED, !Owned(), "BufferWrapper::Resize called on a non-owned buffer");

258 - m_owned->resize(size);

259 - }

260 -

261 - void Append(gsl::span<char> Span)

262 - {

263 - THROW_HR_IF_MSG(E_UNEXPECTED, !Owned(), "BufferWrapper::Append called on a non-owned buffer");

264 -

265 - m_owned->insert(m_owned->end(), Span.begin(), Span.end());

266 - }

267 -

268 - void Consume(size_t bytes) noexcept

269 - {

270 - WI_ASSERT(bytes <= Size());

271 - if (Owned())

272 - {

273 - m_owned->erase(m_owned->begin(), m_owned->begin() + bytes);

274 - }

275 - else

276 - {

277 - m_unowned = m_unowned.subspan(bytes);

278 - }

279 - }

280 -

281 - gsl::span<gsl::byte> Span() noexcept

282 - {

283 - return Owned() ? gsl::make_span(reinterpret_cast<gsl::byte*>(m_owned->data()), m_owned->size()) : m_unowned;

284 - }

285 -

286 - size_t Size() const noexcept

287 - {

288 - return Owned() ? m_owned->size() : m_unowned.size();

289 - }

290 -

291 -private:

292 - std::optional<std::vector<char>> m_owned;

293 - gsl::span<gsl::byte> m_unowned;

294 -};

295 -

296 -class OverlappedIOHandle

297 -{

298 -public:

299 - NON_COPYABLE(OverlappedIOHandle)

300 - NON_MOVABLE(OverlappedIOHandle)

301 -

302 - OverlappedIOHandle() = default;

303 - virtual ~OverlappedIOHandle() = default;

304 - virtual void Schedule() = 0;

305 - virtual void Collect() = 0;

306 - virtual HANDLE GetHandle() const = 0;

307 - IOHandleStatus GetState() const;

308 -

309 -protected:

310 - IOHandleStatus State = IOHandleStatus::Standby;

311 -};

312 -

313 -class EventHandle : public OverlappedIOHandle

314 -{

315 -public:

316 - NON_COPYABLE(EventHandle)

317 - NON_MOVABLE(EventHandle)

318 -

319 - EventHandle(HandleWrapper&& Handle, std::function<void()>&& OnSignalled = []() {});

320 - void Schedule() override;

321 - void Collect() override;

322 - HANDLE GetHandle() const override;

323 -

324 -private:

325 - HandleWrapper Handle;

326 - std::function<void()> OnSignalled;

327 -};

328 -

329 -class ReadHandle : public OverlappedIOHandle

330 -{

331 -public:

332 - NON_COPYABLE(ReadHandle);

333 - NON_MOVABLE(ReadHandle);

334 -

335 - ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead);

336 - virtual ~ReadHandle();

337 -

338 - void Schedule() override;

339 - void Collect() override;

340 - HANDLE GetHandle() const override;

341 -

342 -private:

343 - HandleWrapper Handle;

344 - std::function<void(const gsl::span<char>& Buffer)> OnRead;

345 - wil::unique_event Event{wil::EventOptions::ManualReset};

346 - OVERLAPPED Overlapped{};

347 - BufferWrapper Buffer{LX_RELAY_BUFFER_SIZE};

348 - LARGE_INTEGER Offset{};

349 -};

350 -

351 -class SingleAcceptHandle : public OverlappedIOHandle

352 -{

353 -public:

354 - NON_COPYABLE(SingleAcceptHandle)

355 - NON_MOVABLE(SingleAcceptHandle)

356 -

357 - SingleAcceptHandle(HandleWrapper&& ListenSocket, HandleWrapper&& AcceptedSocket, std::function<void()>&& OnAccepted);

358 - ~SingleAcceptHandle();

359 -

360 - void Schedule() override;

361 - void Collect() override;

362 - HANDLE GetHandle() const override;

363 -

364 -private:

365 - HandleWrapper ListenSocket;

366 - HandleWrapper AcceptedSocket;

367 - wil::unique_event Event{wil::EventOptions::ManualReset};

368 - OVERLAPPED Overlapped{};

369 - std::function<void()> OnAccepted;

370 - char AcceptBuffer[2 * sizeof(SOCKADDR_STORAGE)];

371 -};

372 -

373 -class LineBasedReadHandle : public ReadHandle

374 -{

375 -public:

376 - NON_COPYABLE(LineBasedReadHandle);

377 - NON_MOVABLE(LineBasedReadHandle);

378 -

379 - LineBasedReadHandle(HandleWrapper&& Handle, std::function<void(const gsl::span<char>& Buffer)>&& OnLine, bool Crlf);

380 - ~LineBasedReadHandle();

381 -

382 -private:

383 - void OnRead(const gsl::span<char>& Buffer);

384 -

385 - std::function<void(const gsl::span<char>& Buffer)> OnLine;

386 - std::string PendingBuffer;

387 - bool Crlf{};

388 -};

389 -

390 -class HTTPChunkBasedReadHandle : public ReadHandle

391 -{

392 -public:

393 - NON_COPYABLE(HTTPChunkBasedReadHandle);

394 - NON_MOVABLE(HTTPChunkBasedReadHandle);

395 -

396 - HTTPChunkBasedReadHandle(HandleWrapper&& Handler, std::function<void(const gsl::span<char>& Buffer)>&& OnChunk);

397 - ~HTTPChunkBasedReadHandle();

398 -

399 - void OnRead(const gsl::span<char>& Line);

400 -

401 -private:

402 - std::function<void(const gsl::span<char>& Buffer)> OnChunk;

403 - std::string PendingBuffer;

404 - uint64_t PendingChunkSize = 0;

405 - bool ExpectHeader = true;

406 -};

407 -

408 -class ReadSocketMessageHandle : public OverlappedIOHandle

409 -{

410 -public:

411 - NON_COPYABLE(ReadSocketMessageHandle);

412 - NON_MOVABLE(ReadSocketMessageHandle);

413 -

414 - ReadSocketMessageHandle(HandleWrapper&& Socket, std::vector<gsl::byte>& Buffer, std::function<void(const gsl::span<gsl::byte>& Message)>&& OnMessage);

415 - ~ReadSocketMessageHandle();

416 -

417 - void Schedule() override;

418 - void Collect() override;

419 - HANDLE GetHandle() const override;

420 -

421 -private:

422 - void ScheduleRecv();

423 - void ProcessRecvResult(DWORD BytesRead);

424 -

425 - HandleWrapper Socket;

426 - std::vector<gsl::byte>& Buffer;

427 - std::function<void(const gsl::span<gsl::byte>& Message)> OnMessage;

428 - wil::unique_event Event{wil::EventOptions::ManualReset};

429 - OVERLAPPED Overlapped{};

430 - bool ReadingHeader = true;

431 - size_t BytesRemaining = sizeof(MESSAGE_HEADER);

432 - size_t CurrentOffset = 0;

433 -};

434 -

435 -class WriteHandle : public OverlappedIOHandle

436 -{

437 -public:

438 - NON_COPYABLE(WriteHandle);

439 - NON_MOVABLE(WriteHandle);

440 -

441 - WriteHandle(HandleWrapper&& Handle, const std::vector<char>& Buffer = {});

442 - WriteHandle(HandleWrapper&& Handle, gsl::span<gsl::byte> Span);

443 - ~WriteHandle();

444 - void Schedule() override;

445 - void Collect() override;

446 - HANDLE GetHandle() const override;

447 - void Push(const gsl::span<char>& Buffer);

448 -

449 -private:

450 - HandleWrapper Handle;

451 - wil::unique_event Event{wil::EventOptions::ManualReset};

452 - OVERLAPPED Overlapped{};

453 - BufferWrapper Buffer;

454 - LARGE_INTEGER Offset{};

455 -};

456 -

457 -template <typename TRead = ReadHandle>

458 -class RelayHandle : public OverlappedIOHandle

459 -{

460 -public:

461 - NON_COPYABLE(RelayHandle);

462 - NON_MOVABLE(RelayHandle);

463 -

464 - RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output) :

465 - Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output))

466 - {

467 - }

468 -

469 - void Schedule() override

470 - {

471 - WI_ASSERT(State == IOHandleStatus::Standby);

472 -

473 - // If the Buffer is empty, then we're reading.

474 - if (PendingBuffer.empty())

475 - {

476 - // If the output buffer is empty and the reading end is completed, then we're done.

477 - if (Read.GetState() == IOHandleStatus::Completed)

478 - {

479 - State = IOHandleStatus::Completed;

480 - return;

481 - }

482 -

483 - Read.Schedule();

484 -

485 - // If the read is pending, update to 'Pending'

486 - if (Read.GetState() == IOHandleStatus::Pending)

487 - {

488 - State = IOHandleStatus::Pending;

489 - }

490 - }

491 - else

492 - {

493 - Write.Push(PendingBuffer);

494 - PendingBuffer.clear();

495 -

496 - Write.Schedule();

497 -

498 - if (Write.GetState() == IOHandleStatus::Pending)

499 - {

500 - // The write is pending, update to 'Pending'

501 - State = IOHandleStatus::Pending;

502 - }

503 - }

504 - }

505 -

506 - void Collect() override

507 - {

508 - WI_ASSERT(State == IOHandleStatus::Pending);

509 -

510 - // Transition back to standby

511 - State = IOHandleStatus::Standby;

512 -

513 - if (Read.GetState() == IOHandleStatus::Pending)

514 - {

515 - Read.Collect();

516 - }

517 - else

518 - {

519 - WI_ASSERT(Write.GetState() == IOHandleStatus::Pending);

520 - Write.Collect();

521 - }

522 - }

523 -

524 - HANDLE GetHandle() const override

525 - {

526 - if (Read.GetState() == IOHandleStatus::Pending)

527 - {

528 - return Read.GetHandle();

529 - }

530 - else

531 - {

532 - WI_ASSERT(Write.GetState() == IOHandleStatus::Pending);

533 - return Write.GetHandle();

534 - }

535 - }

536 -

537 -private:

538 - void OnRead(const gsl::span<char>& Content)

539 - {

540 - PendingBuffer.insert(PendingBuffer.end(), Content.begin(), Content.end());

541 - }

542 -

543 - TRead Read;

544 - WriteHandle Write;

545 - std::vector<char> PendingBuffer;

546 -};

547 -

548 -class DockerIORelayHandle : public OverlappedIOHandle

549 -{

550 -public:

551 - NON_COPYABLE(DockerIORelayHandle);

552 - NON_MOVABLE(DockerIORelayHandle);

553 -

554 - enum class Format

555 - {

556 - Raw,

557 - HttpChunked

558 - };

559 -

560 - DockerIORelayHandle(HandleWrapper&& Input, HandleWrapper&& Stdout, HandleWrapper&& Stderr, Format ReadFormat);

561 - void Schedule() override;

562 - void Collect() override;

563 - HANDLE GetHandle() const override;

564 -

565 -#pragma pack(push, 1)

566 - struct MultiplexedHeader

567 - {

568 - uint8_t Fd;

569 - char Zeroes[3];

570 - uint32_t Length;

571 - };

572 -#pragma pack(pop)

573 -

574 - static_assert(sizeof(MultiplexedHeader) == 8);

575 -

576 -private:

577 - void OnRead(const gsl::span<char>& Buffer);

578 - void ProcessNextHeader();

579 -

580 - std::unique_ptr<OverlappedIOHandle> Read;

581 - WriteHandle WriteStdout;

582 - WriteHandle WriteStderr;

583 - std::vector<char> PendingBuffer;

584 - WriteHandle* ActiveHandle = nullptr;

585 - size_t RemainingBytes = 0;

586 -};

587 -

588 -class MultiHandleWait

589 -{

590 -public:

591 - NON_COPYABLE(MultiHandleWait);

592 - DEFAULT_MOVABLE(MultiHandleWait);

593 -

594 - enum Flags

595 - {

596 - None = 0,

597 - CancelOnCompleted = 1,

598 - IgnoreErrors = 2,

599 - NeedNotComplete = 4,

600 - };

601 -

602 - MultiHandleWait() = default;

603 -

604 - void AddHandle(std::unique_ptr<OverlappedIOHandle>&& handle, Flags flags = Flags::None);

605 - bool Run(std::optional<std::chrono::milliseconds> Timeout);

606 - void Cancel();

607 -

608 -private:

609 - std::vector<std::pair<Flags, std::unique_ptr<OverlappedIOHandle>>> m_handles;

610 - bool m_cancel = false;

611 -};

612 -

613 -DEFINE_ENUM_FLAG_OPERATORS(MultiHandleWait::Flags);

614 -

161 } // namespace wsl::windows::common::relay

src/windows/common/socket.cpp

+3 -3

@@ -20,15 +20,15 @@ Abstract:

20 bool wsl::windows::common::socket::CancellableAccept(

21 _In_ SOCKET ListenSocket, _In_ SOCKET Socket, _In_ DWORD Timeout, _In_opt_ HANDLE ExitHandle, _In_ const std::source_location& Location)

22 {

23 - relay::MultiHandleWait io;

23 + io::MultiHandleWait io;

25 bool accepted = false;

27 - io.AddHandle(std::make_unique<relay::SingleAcceptHandle>(ListenSocket, Socket, [&]() { accepted = true; }), relay::MultiHandleWait::CancelOnCompleted);

27 + io.AddHandle(std::make_unique<io::SingleAcceptHandle>(ListenSocket, Socket, [&]() { accepted = true; }), io::MultiHandleWait::CancelOnCompleted);

29 if (ExitHandle != nullptr)

30 {

31 - io.AddHandle(std::make_unique<relay::EventHandle>(ExitHandle), relay::MultiHandleWait::CancelOnCompleted);

31 + io.AddHandle(std::make_unique<io::EventHandle>(ExitHandle), io::MultiHandleWait::CancelOnCompleted);

32 }

34 io.Run(std::chrono::milliseconds(Timeout));

src/windows/wslc/services/ConsoleService.cpp

+3 -3

@@ -18,8 +18,8 @@ Abstract:

18 namespace wsl::windows::wslc::services {

20 using wsl::windows::common::ClientRunningWSLCProcess;

21 -using wsl::windows::common::relay::ReadHandle;

22 -using wsl::windows::common::relay::RelayHandle;

21 +using wsl::windows::common::io::ReadHandle;

22 +using wsl::windows::common::io::RelayHandle;

24 bool ConsoleService::RelayInteractiveTty(ClientRunningWSLCProcess& Process, HANDLE Tty, bool triggerRefresh)

25 {

@@ -70,7 +70,7 @@ bool ConsoleService::RelayInteractiveTty(ClientRunningWSLCProcess& Process, HAND

71 void ConsoleService::RelayNonTtyProcess(wil::unique_handle&& Stdin, wil::unique_handle&& Stdout, wil::unique_handle&& Stderr)

72 {

73 - wsl::windows::common::relay::MultiHandleWait io;

73 + wsl::windows::common::io::MultiHandleWait io;

75 // Create a thread to relay stdin to the pipe.

76 wil::unique_event exitEvent(wil::EventOptions::ManualReset);

src/windows/wslc/services/ContainerService.cpp

+3 -3

@@ -521,13 +521,13 @@ void ContainerService::Logs(Session& session, const std::string& id, bool follow

521

522 THROW_IF_FAILED(container->Logs(flags, &stdoutHandle, &stderrHandle, 0, 0, tail));

523

524 - wsl::windows::common::relay::MultiHandleWait io;

525 - io.AddHandle(std::make_unique<wsl::windows::common::relay::RelayHandle<wsl::windows::common::relay::ReadHandle>>(

524 + wsl::windows::common::io::MultiHandleWait io;

525 + io.AddHandle(std::make_unique<wsl::windows::common::io::RelayHandle<wsl::windows::common::io::ReadHandle>>(

526 stdoutHandle.Release(), GetStdHandle(STD_OUTPUT_HANDLE)));

527

528 if (!stderrHandle.Empty()) // This handle is only used for non-tty processes.

529 {

530 - io.AddHandle(std::make_unique<wsl::windows::common::relay::RelayHandle<wsl::windows::common::relay::ReadHandle>>(

530 + io.AddHandle(std::make_unique<wsl::windows::common::io::RelayHandle<wsl::windows::common::io::ReadHandle>>(

531 stderrHandle.Release(), GetStdHandle(STD_ERROR_HANDLE)));

532 }

533

src/windows/wslcsession/DockerEventTracker.cpp

+1 -1

@@ -83,7 +83,7 @@ DockerEventTracker::DockerEventTracker(DockerHTTPClient& dockerClient, WSLCSessi

84 auto socket = dockerClient.MonitorEvents();

86 - relay.AddHandle(std::make_unique<common::relay::HTTPChunkBasedReadHandle>(std::move(socket), std::move(onChunk)));

86 + relay.AddHandle(std::make_unique<common::io::HTTPChunkBasedReadHandle>(std::move(socket), std::move(onChunk)));

87 }

89 DockerEventTracker::~DockerEventTracker()

src/windows/wslcsession/DockerHTTPClient.cpp

+6 -6

@@ -31,8 +31,8 @@ Abstract:

31 namespace http = boost::beast::http;

32 using boost::beast::http::verb;

33 using wsl::windows::common::docker_schema::EmptyRequest;

34 -using wsl::windows::common::relay::HandleWrapper;

35 -using wsl::windows::common::relay::MultiHandleWait;

34 +using wsl::windows::common::io::HandleWrapper;

35 +using wsl::windows::common::io::MultiHandleWait;

36 using wsl::windows::service::wslc::DockerHTTPClient;

37 using namespace wsl::windows::common;

@@ -618,7 +618,7 @@ std::pair<DockerHTTPClient::HTTPResponse, std::string> DockerHTTPClient::SendReq

618 auto onHttpResponse = [&](const auto& response) { responseHeader = response; };

619 MultiHandleWait io;

620

621 - io.AddHandle(std::make_unique<relay::EventHandle>(m_exitingEvent, [&]() { THROW_HR(E_ABORT); }));

621 + io.AddHandle(std::make_unique<io::EventHandle>(m_exitingEvent, [&]() { THROW_HR(E_ABORT); }));

622 io.AddHandle(std::make_unique<DockerHttpResponseHandle>(*context, std::move(onHttpResponse), std::move(OnResponse)), MultiHandleWait::CancelOnCompleted);

623

624 io.Run({});

@@ -633,7 +633,7 @@ DockerHTTPClient::DockerHttpResponseHandle::DockerHttpResponseHandle(

633 std::function<void(const HTTPResponse&)>&& onResponseHeader,

634 std::function<void(const gsl::span<char>&)>&& onResponseBytes,

635 std::function<void()>&& onCompleted) :

636 - common::relay::ReadHandle(

636 + common::io::ReadHandle(

637 HandleWrapper{context.stream.native_handle()}, std::bind(&DockerHttpResponseHandle::OnRead, this, std::placeholders::_1)),

638 Context(context),

639 OnResponseHeader(std::move(onResponseHeader)),

@@ -644,7 +644,7 @@ DockerHTTPClient::DockerHttpResponseHandle::DockerHttpResponseHandle(

644

645 DockerHTTPClient::DockerHttpResponseHandle::~DockerHttpResponseHandle()

646 {

647 - if (State == common::relay::IOHandleStatus::Completed)

647 + if (State == common::io::IOHandleStatus::Completed)

648 {

649 OnCompleted();

650 }

@@ -731,7 +731,7 @@ void DockerHTTPClient::DockerHttpResponseHandle::OnResponseBytes(const gsl::span

731 *RemainingContentLength -= consume;

732 if (*RemainingContentLength == 0)

733 {

734 - State = common::relay::IOHandleStatus::Completed;

734 + State = common::io::IOHandleStatus::Completed;

735 }

736

737 span = span.subspan(0, consume);

src/windows/wslcsession/DockerHTTPClient.h

+2 -2

@@ -188,7 +188,7 @@ public:

188

189 wil::unique_socket MonitorEvents();

190

191 - struct DockerHttpResponseHandle : public common::relay::ReadHandle

191 + struct DockerHttpResponseHandle : public common::io::ReadHandle

192 {

193 NON_COPYABLE(DockerHttpResponseHandle);

194 NON_MOVABLE(DockerHttpResponseHandle);

@@ -212,7 +212,7 @@ public:

212 boost::beast::http::response_parser<boost::beast::http::buffer_body> Parser;

213 size_t LineFeeds = 0;

214 std::optional<size_t> RemainingContentLength;

215 - std::optional<common::relay::HTTPChunkBasedReadHandle> ResponseParser;

215 + std::optional<common::io::HTTPChunkBasedReadHandle> ResponseParser;

216 };

217

218 private:

src/windows/wslcsession/IORelay.cpp

+8 -8

@@ -14,9 +14,9 @@ Abstract:

15 #include "IORelay.h"

17 -using wsl::windows::common::relay::DockerIORelayHandle;

18 -using wsl::windows::common::relay::MultiHandleWait;

19 -using wsl::windows::common::relay::OverlappedIOHandle;

17 +using wsl::windows::common::io::DockerIORelayHandle;

18 +using wsl::windows::common::io::MultiHandleWait;

19 +using wsl::windows::common::io::OverlappedIOHandle;

20 using wsl::windows::service::wslc::IORelay;

22 IORelay::IORelay()

@@ -29,15 +29,15 @@ IORelay::~IORelay()

29 Stop();

30 }

32 -void IORelay::AddHandle(std::unique_ptr<common::relay::OverlappedIOHandle>&& Handle)

32 +void IORelay::AddHandle(std::unique_ptr<common::io::OverlappedIOHandle>&& Handle)

33 {

34 - std::vector<std::unique_ptr<common::relay::OverlappedIOHandle>> handles;

34 + std::vector<std::unique_ptr<common::io::OverlappedIOHandle>> handles;

35 handles.emplace_back(std::move(Handle));

37 AddHandles(std::move(handles));

38 }

40 -void IORelay::AddHandles(std::vector<std::unique_ptr<common::relay::OverlappedIOHandle>>&& Handles)

40 +void IORelay::AddHandles(std::vector<std::unique_ptr<common::io::OverlappedIOHandle>>&& Handles)

41 {

42 WI_ASSERT(!m_exit);

@@ -73,7 +73,7 @@ try

73 {

74 common::wslutil::SetThreadDescription(L"IORelay");

76 - windows::common::relay::MultiHandleWait io;

76 + windows::common::io::MultiHandleWait io;

78 // N.B. All the IO must happen on the thread.

79 // If the thread that scheduled the IO exits, the IO is cancelled.

@@ -90,7 +90,7 @@ try

90 m_pendingHandles.clear();

91 }

93 - io.AddHandle(std::make_unique<common::relay::EventHandle>(m_refreshEvent.get()), MultiHandleWait::CancelOnCompleted);

93 + io.AddHandle(std::make_unique<common::io::EventHandle>(m_refreshEvent.get()), MultiHandleWait::CancelOnCompleted);

94 io.Run({});

95 }

96 }

src/windows/wslcsession/IORelay.h

+3 -3

@@ -25,8 +25,8 @@ public:

25 IORelay();

26 ~IORelay();

28 - void AddHandles(std::vector<std::unique_ptr<common::relay::OverlappedIOHandle>>&& Handles);

29 - void AddHandle(std::unique_ptr<common::relay::OverlappedIOHandle>&& Handle);

28 + void AddHandles(std::vector<std::unique_ptr<common::io::OverlappedIOHandle>>&& Handles);

29 + void AddHandle(std::unique_ptr<common::io::OverlappedIOHandle>&& Handle);

31 void Stop();

@@ -36,7 +36,7 @@ private:

37 std::mutex m_pendingHandlesLock;

38 wil::unique_event m_refreshEvent{wil::EventOptions::None};

39 - std::vector<std::unique_ptr<common::relay::OverlappedIOHandle>> m_pendingHandles;

39 + std::vector<std::unique_ptr<common::io::OverlappedIOHandle>> m_pendingHandles;

41 std::thread m_thread;

42 std::atomic<bool> m_exit = false;

src/windows/wslcsession/PortRelayHandle.cpp

+7 -7

@@ -34,7 +34,7 @@ PortRelayAcceptHandle::PortRelayAcceptHandle(

35 PortRelayAcceptHandle::~PortRelayAcceptHandle()

36 {

37 - if (State == relay::IOHandleStatus::Pending)

37 + if (State == io::IOHandleStatus::Pending)

38 {

39 LOG_IF_WIN32_BOOL_FALSE(CancelIoEx(reinterpret_cast<HANDLE>(ListenSocket.get()), &Overlapped));

@@ -50,7 +50,7 @@ PortRelayAcceptHandle::~PortRelayAcceptHandle()

51 void PortRelayAcceptHandle::Schedule()

52 {

53 - WI_ASSERT(State == relay::IOHandleStatus::Standby);

53 + WI_ASSERT(State == io::IOHandleStatus::Standby);

55 // Create a new socket for accepting

56 AcceptedSocket.reset(WSASocket(Family, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED));

@@ -61,22 +61,22 @@ void PortRelayAcceptHandle::Schedule()

61 if (AcceptEx(ListenSocket.get(), AcceptedSocket.get(), AcceptBuffer, 0, sizeof(SOCKADDR_STORAGE), sizeof(SOCKADDR_STORAGE), &bytesReturned, &Overlapped))

62 {

63 // Accept completed immediately

64 - State = relay::IOHandleStatus::Completed;

64 + State = io::IOHandleStatus::Completed;

65 }

66 else

67 {

68 auto error = WSAGetLastError();

69 THROW_HR_IF_MSG(HRESULT_FROM_WIN32(error), error != ERROR_IO_PENDING, "Handle: 0x%p", reinterpret_cast<void*>(ListenSocket.get()));

71 - State = relay::IOHandleStatus::Pending;

71 + State = io::IOHandleStatus::Pending;

72 }

73 }

75 void PortRelayAcceptHandle::Collect()

76 {

77 - WI_ASSERT(State == relay::IOHandleStatus::Pending || State == relay::IOHandleStatus::Completed);

77 + WI_ASSERT(State == io::IOHandleStatus::Pending || State == io::IOHandleStatus::Completed);

79 - if (State == relay::IOHandleStatus::Pending)

79 + if (State == io::IOHandleStatus::Pending)

80 {

81 DWORD bytesReceived{};

82 DWORD flagsReturned{};

@@ -87,7 +87,7 @@ void PortRelayAcceptHandle::Collect()

87 LaunchRelay(std::move(AcceptedSocket));

89 // Go back to standby to accept the next connection

90 - State = relay::IOHandleStatus::Standby;

90 + State = io::IOHandleStatus::Standby;

91 }

93 HANDLE PortRelayAcceptHandle::GetHandle() const

src/windows/wslcsession/PortRelayHandle.h

+1 -1

@@ -20,7 +20,7 @@ namespace wsl::windows::service::wslc {

21 class IORelay;

23 -class PortRelayAcceptHandle : public common::relay::OverlappedIOHandle

23 +class PortRelayAcceptHandle : public common::io::OverlappedIOHandle

24 {

25 public:

26 NON_COPYABLE(PortRelayAcceptHandle)

src/windows/wslcsession/WSLCContainer.cpp

+10 -10

@@ -25,12 +25,12 @@ Abstract:

26 using wsl::windows::common::COMServiceExecutionContext;

27 using wsl::windows::common::docker_schema::ErrorResponse;

28 -using wsl::windows::common::relay::DockerIORelayHandle;

29 -using wsl::windows::common::relay::HandleWrapper;

30 -using wsl::windows::common::relay::HTTPChunkBasedReadHandle;

31 -using wsl::windows::common::relay::OverlappedIOHandle;

32 -using wsl::windows::common::relay::ReadHandle;

33 -using wsl::windows::common::relay::RelayHandle;

28 +using wsl::windows::common::io::DockerIORelayHandle;

29 +using wsl::windows::common::io::HandleWrapper;

30 +using wsl::windows::common::io::HTTPChunkBasedReadHandle;

31 +using wsl::windows::common::io::OverlappedIOHandle;

32 +using wsl::windows::common::io::ReadHandle;

33 +using wsl::windows::common::io::RelayHandle;

34 using wsl::windows::service::wslc::ContainerPortMapping;

35 using wsl::windows::service::wslc::IWSLCVolume;

36 using wsl::windows::service::wslc::NetworkEntry;

@@ -47,7 +47,7 @@ using wsl::windows::service::wslc::WSLCSession;

47 using wsl::windows::service::wslc::WSLCVirtualMachine;

48 using wsl::windows::service::wslc::WSLCVolumeMount;

50 -using namespace wsl::windows::common::relay;

50 +using namespace wsl::windows::common::io;

51 using namespace wsl::windows::common::docker_schema;

52 using namespace wsl::windows::common::wslutil;

53 using namespace std::chrono_literals;

@@ -1023,7 +1023,7 @@ void WSLCContainerImpl::Export(WSLCHandle OutHandle) const

1023

1024 auto userHandle = m_wslcSession.OpenUserHandle(OutHandle);

1025

1026 - wsl::windows::common::relay::MultiHandleWait io = m_wslcSession.CreateIOContext();

1026 + wsl::windows::common::io::MultiHandleWait io = m_wslcSession.CreateIOContext();

1027

1028 std::string errorJson;

1029 auto accumulateError = [&](const gsl::span<char>& buffer) {

@@ -1039,7 +1039,7 @@ void WSLCContainerImpl::Export(WSLCHandle OutHandle) const

1039 {

1040 io.AddHandle(

1041 std::make_unique<RelayHandle<HTTPChunkBasedReadHandle>>(HandleWrapper{std::move(SocketCodePair.second)}, userHandle.Get()),

1042 - wsl::windows::common::relay::MultiHandleWait::CancelOnCompleted);

1042 + wsl::windows::common::io::MultiHandleWait::CancelOnCompleted);

1043 }

1044

1045 // Release the lock so the container can still be interacted with while the export is in progress.

@@ -1884,7 +1884,7 @@ std::unique_ptr<RelayedProcessIO> WSLCContainerImpl::CreateRelayedProcessIO(wil:

1884 fds.emplace(WSLCFDStderr, TypedHandle{wil::unique_handle{stderrRead.release()}, WSLCHandleTypePipe});

1885

1886 ioHandles.emplace_back(std::make_unique<DockerIORelayHandle>(

1887 - std::move(stream), std::move(stdoutWrite), std::move(stderrWrite), common::relay::DockerIORelayHandle::Format::Raw));

1887 + std::move(stream), std::move(stdoutWrite), std::move(stderrWrite), common::io::DockerIORelayHandle::Format::Raw));

1888

1889 m_ioRelay.AddHandles(std::move(ioHandles));

1890

src/windows/wslcsession/WSLCProcessIO.cpp

+1 -1

@@ -20,7 +20,7 @@ using wsl::windows::service::wslc::RelayedProcessIO;

20 using wsl::windows::service::wslc::TTYProcessIO;

21 using wsl::windows::service::wslc::TypedHandle;

22 using wsl::windows::service::wslc::VMProcessIO;

23 -using namespace wsl::windows::common::relay;

23 +using namespace wsl::windows::common::io;

25 RelayedProcessIO::RelayedProcessIO(std::map<ULONG, TypedHandle>&& fds) : m_relayedHandles(std::move(fds))

26 {

src/windows/wslcsession/WSLCSession.cpp

+23 -25

@@ -22,7 +22,7 @@ Abstract:

22 #include "wslpolicies.h"

24 using namespace wsl::windows::common;

25 -using relay::MultiHandleWait;

25 +using io::MultiHandleWait;

26 using wsl::shared::Localization;

27 using wsl::windows::service::wslc::UserCOMCallback;

28 using wsl::windows::service::wslc::UserHandle;

@@ -311,8 +311,7 @@ try

311 m_volumes.emplace(m_dockerClient.value(), m_virtualMachine.value(), m_eventTracker.value(), m_storageVhdPath.parent_path());

312

313 // Monitor for unexpected VM exit.

314 - m_ioRelay.AddHandle(

315 - std::make_unique<windows::common::relay::EventHandle>(m_vmExitedEvent.get(), std::bind(&WSLCSession::OnVmExited, this)));

314 + m_ioRelay.AddHandle(std::make_unique<windows::common::io::EventHandle>(m_vmExitedEvent.get(), std::bind(&WSLCSession::OnVmExited, this)));

315

316 // Recover any existing resources from storage.

317 RecoverExistingNetworks();

@@ -492,13 +491,13 @@ ServiceRunningProcess WSLCSession::StartProcess(

491

492 auto process = launcher.Launch(*m_virtualMachine);

493

495 - m_ioRelay.AddHandle(std::make_unique<windows::common::relay::LineBasedReadHandle>(

494 + m_ioRelay.AddHandle(std::make_unique<windows::common::io::LineBasedReadHandle>(

495 process.GetStdHandle(1), [this, LogSource](const auto& data) { OnProcessLog(data, LogSource); }, false));

496

498 - m_ioRelay.AddHandle(std::make_unique<windows::common::relay::LineBasedReadHandle>(

497 + m_ioRelay.AddHandle(std::make_unique<windows::common::io::LineBasedReadHandle>(

498 process.GetStdHandle(2), [this, LogSource](const auto& data) { OnProcessLog(data, LogSource); }, false));

499

501 - m_ioRelay.AddHandle(std::make_unique<windows::common::relay::EventHandle>(process.GetExitEvent(), std::move(ExitCallback)));

500 + m_ioRelay.AddHandle(std::make_unique<windows::common::io::EventHandle>(process.GetExitEvent(), std::move(ExitCallback)));

501

502 return process;

503 }

@@ -762,8 +761,8 @@ try

761

762 auto io = CreateIOContext();

763

765 - io.AddHandle(std::make_unique<relay::RelayHandle<relay::ReadHandle>>(

766 - buildFileHandle.Get(), common::relay::HandleWrapper{buildProcess.GetStdHandle(WSLCFDStdin)}));

764 + io.AddHandle(std::make_unique<io::RelayHandle<io::ReadHandle>>(

765 + buildFileHandle.Get(), common::io::HandleWrapper{buildProcess.GetStdHandle(WSLCFDStdin)}));

766

767 bool verbose = WI_IsFlagSet(Options->Flags, WSLCBuildImageFlagsVerbose);

768 std::string allOutput;

@@ -920,11 +919,11 @@ try

919 // With --progress=rawjson, docker writes progress to stderr and the final image ID to stdout on success (empty on

920 // failure). Stdout is drained into allOutput (shown only on error) and its EOF signals build completion.

921 io.AddHandle(

923 - std::make_unique<relay::ReadHandle>(

922 + std::make_unique<io::ReadHandle>(

923 buildProcess.GetStdHandle(1), [&](const auto& content) { allOutput.append(content.begin(), content.end()); }),

925 - relay::MultiHandleWait::CancelOnCompleted);

924 + io::MultiHandleWait::CancelOnCompleted);

925

927 - io.AddHandle(std::make_unique<relay::LineBasedReadHandle>(buildProcess.GetStdHandle(2), captureOutput, false));

926 + io.AddHandle(std::make_unique<io::LineBasedReadHandle>(buildProcess.GetStdHandle(2), captureOutput, false));

927

928 // Handle cancellation within the IO loop (NeedNotComplete) so pipes keep draining.

929 bool cancelled = false;

@@ -935,7 +934,7 @@ try

934 THROW_LAST_ERROR_IF_NULL(killTimer);

935

936 io.AddHandle(

938 - std::make_unique<relay::EventHandle>(

937 + std::make_unique<io::EventHandle>(

938 CancelEvent,

939 [&]() {

940 cancelled = true;

@@ -943,12 +942,12 @@ try

942 LARGE_INTEGER dueTime{.QuadPart = -10LL * 10 * 1000 * 1000}; // 10 seconds

943 THROW_IF_WIN32_BOOL_FALSE(SetWaitableTimer(killTimer.get(), &dueTime, 0, nullptr, nullptr, FALSE));

944 }),

946 - relay::MultiHandleWait::NeedNotComplete);

945 + io::MultiHandleWait::NeedNotComplete);

946

947 io.AddHandle(

949 - std::make_unique<relay::EventHandle>(

948 + std::make_unique<io::EventHandle>(

949 killTimer.get(), [&]() { LOG_IF_FAILED(buildProcess.Get().Signal(WSLCSignalSIGKILL)); }),

951 - relay::MultiHandleWait::NeedNotComplete);

950 + io::MultiHandleWait::NeedNotComplete);

951 }

952

953 try

@@ -1111,9 +1110,8 @@ void WSLCSession::ImportImageImpl(DockerHTTPClient::HTTPRequestContext& Request,

1110 LOG_LAST_ERROR_IF(shutdown(socket, SD_SEND) == SOCKET_ERROR);

1111 };

1112

1114 - io.AddHandle(std::make_unique<relay::RelayHandle<relay::ReadHandle>>(

1115 - common::relay::HandleWrapper{userHandle.Get(), std::move(onInputComplete)},

1116 - common::relay::HandleWrapper{Request.stream.native_handle()}));

1113 + io.AddHandle(std::make_unique<io::RelayHandle<io::ReadHandle>>(

1114 + common::io::HandleWrapper{userHandle.Get(), std::move(onInputComplete)}, common::io::HandleWrapper{Request.stream.native_handle()}));

1115

1116 io.AddHandle(

1117 std::make_unique<DockerHTTPClient::DockerHttpResponseHandle>(Request, std::move(onHttpResponse), std::move(onProgress)),

@@ -1170,14 +1168,14 @@ void WSLCSession::SaveImageImpl(std::pair<uint32_t, wil::unique_socket>& SocketC

1168 };

1169

1170 io.AddHandle(

1173 - std::make_unique<relay::ReadHandle>(common::relay::HandleWrapper{std::move(SocketCodePair.second)}, std::move(accumulateError)),

1171 + std::make_unique<io::ReadHandle>(common::io::HandleWrapper{std::move(SocketCodePair.second)}, std::move(accumulateError)),

1172 MultiHandleWait::CancelOnCompleted);

1173 }

1174 else

1175 {

1176 io.AddHandle(

1179 - std::make_unique<relay::RelayHandle<relay::HTTPChunkBasedReadHandle>>(

1180 - common::relay::HandleWrapper{std::move(SocketCodePair.second)}, userHandle.Get()),

1177 + std::make_unique<io::RelayHandle<io::HTTPChunkBasedReadHandle>>(

1178 + common::io::HandleWrapper{std::move(SocketCodePair.second)}, userHandle.Get()),

1179 MultiHandleWait::CancelOnCompleted);

1180 }

1181

@@ -2644,20 +2642,20 @@ HRESULT WSLCSession::InterfaceSupportsErrorInfo(REFIID riid)

2642

2643 MultiHandleWait WSLCSession::CreateIOContext(HANDLE CancelHandle)

2644 {

2647 - relay::MultiHandleWait io;

2645 + io::MultiHandleWait io;

2646

2647 // Cancel with E_ABORT if the session is terminating.

2650 - io.AddHandle(std::make_unique<relay::EventHandle>(

2648 + io.AddHandle(std::make_unique<io::EventHandle>(

2649 m_sessionTerminatingEvent.get(), [this]() { THROW_HR_MSG(E_ABORT, "Session %lu is terminating", m_id); }));

2650

2651 // Cancel with E_ABORT if the client process exits.

2654 - io.AddHandle(std::make_unique<relay::EventHandle>(

2652 + io.AddHandle(std::make_unique<io::EventHandle>(

2653 wslutil::OpenCallingProcess(SYNCHRONIZE), [this]() { THROW_HR_MSG(E_ABORT, "Client process has exited"); }));

2654

2655 if (CancelHandle != nullptr)

2656 {

2657 io.AddHandle(

2660 - std::make_unique<relay::EventHandle>(CancelHandle, []() { THROW_HR_MSG(E_ABORT, "Cancellation handle was signaled"); }));

2658 + std::make_unique<io::EventHandle>(CancelHandle, []() { THROW_HR_MSG(E_ABORT, "Cancellation handle was signaled"); }));

2659 }

2660

2661 return io;

src/windows/wslcsession/WSLCSession.h

+1 -1

@@ -150,7 +150,7 @@ public:

150 IFACEMETHOD(MapVmPort)(_In_ int Family, _In_ unsigned short WindowsPort, _In_ unsigned short LinuxPort) override;

151 IFACEMETHOD(UnmapVmPort)(_In_ int Family, _In_ unsigned short WindowsPort, _In_ unsigned short LinuxPort) override;

152

153 - common::relay::MultiHandleWait CreateIOContext(HANDLE CancelHandle = nullptr);

153 + common::io::MultiHandleWait CreateIOContext(HANDLE CancelHandle = nullptr);

154

155 UserHandle OpenUserHandle(WSLCHandle Handle);

156 void ReleaseUserHandle(HANDLE Handle);

test/windows/Common.cpp

+4 -4

@@ -2814,13 +2814,13 @@ try

2814 }

2815 CATCH_LOG();

2816

2817 -class ReadHandleWithTargetValue : public wsl::windows::common::relay::ReadHandle

2817 +class ReadHandleWithTargetValue : public wsl::windows::common::io::ReadHandle

2818 {

2819 public:

2820 NON_COPYABLE(ReadHandleWithTargetValue);

2821 NON_MOVABLE(ReadHandleWithTargetValue);

2822

2823 - ReadHandleWithTargetValue(wsl::windows::common::relay::HandleWrapper&& MovedHandle, std::string_view targetValue) :

2823 + ReadHandleWithTargetValue(wsl::windows::common::io::HandleWrapper&& MovedHandle, std::string_view targetValue) :

2824 ReadHandle(std::move(MovedHandle), [this](const auto& buffer) { m_readBuffer.append(buffer.data(), buffer.size()); }),

2825 m_targetValue(targetValue)

2826 {

@@ -2841,7 +2841,7 @@ public:

2841 private:

2842 void CheckIfTargetFound()

2843 {

2844 - using namespace wsl::windows::common::relay;

2844 + using namespace wsl::windows::common::io;

2845

2846 if (State == IOHandleStatus::Standby || State == IOHandleStatus::Completed)

2847 {

@@ -2867,7 +2867,7 @@ private:

2867

2868 void WaitForOutput(wil::unique_handle handle, std::string_view targetValue, std::chrono::milliseconds timeout)

2869 {

2870 - wsl::windows::common::relay::MultiHandleWait io;

2870 + wsl::windows::common::io::MultiHandleWait io;

2871 io.AddHandle(std::make_unique<ReadHandleWithTargetValue>(std::move(handle), targetValue));

2872 io.Run(timeout);

2873 }

test/windows/UnitTests.cpp

+3 -3

@@ -6729,9 +6729,9 @@ Error code: Wsl/InstallDistro/WSL_E_INVALID_JSON\r\n",

6729 bool callbackInvoked = false;

6730 std::vector<gsl::byte> message;

6731

6732 - wsl::windows::common::relay::MultiHandleWait io;

6733 - io.AddHandle(std::make_unique<wsl::windows::common::relay::ReadSocketMessageHandle>(

6734 - wsl::windows::common::relay::HandleWrapper{std::move(server)}, buffer, [&callbackInvoked, &message](const gsl::span<gsl::byte>& received) {

6732 + wsl::windows::common::io::MultiHandleWait io;

6733 + io.AddHandle(std::make_unique<wsl::windows::common::io::ReadSocketMessageHandle>(

6734 + wsl::windows::common::io::HandleWrapper{std::move(server)}, buffer, [&callbackInvoked, &message](const gsl::span<gsl::byte>& received) {

6735 callbackInvoked = true;

6736 message.assign(received.begin(), received.end());

6737 }));

test/windows/WSLCTests.cpp

+16 -16

@@ -28,8 +28,8 @@ using wsl::windows::common::RunningWSLCContainer;

28 using wsl::windows::common::RunningWSLCProcess;

29 using wsl::windows::common::WSLCContainerLauncher;

30 using wsl::windows::common::WSLCProcessLauncher;

31 -using wsl::windows::common::relay::OverlappedIOHandle;

32 -using wsl::windows::common::relay::WriteHandle;

31 +using wsl::windows::common::io::OverlappedIOHandle;

32 +using wsl::windows::common::io::WriteHandle;

33 using namespace wsl::windows::common::wslutil;

35 extern std::wstring g_testDataPath;

@@ -7204,12 +7204,12 @@ class WSLCTests

7204 std::vector<std::string> lines;

7205 auto onData = [&](const gsl::span<char>& data) { lines.emplace_back(data.data(), data.size()); };

7206

7207 - wsl::windows::common::relay::MultiHandleWait io;

7207 + wsl::windows::common::io::MultiHandleWait io;

7208

7209 - io.AddHandle(std::make_unique<wsl::windows::common::relay::LineBasedReadHandle>(std::move(readPipe), std::move(onData), Crlf));

7209 + io.AddHandle(std::make_unique<wsl::windows::common::io::LineBasedReadHandle>(std::move(readPipe), std::move(onData), Crlf));

7210

7211 std::vector<char> buffer{Data.begin(), Data.end()};

7212 - io.AddHandle(std::make_unique<wsl::windows::common::relay::WriteHandle>(std::move(writePipe), buffer));

7212 + io.AddHandle(std::make_unique<wsl::windows::common::io::WriteHandle>(std::move(writePipe), buffer));

7213

7214 io.Run({});

7215

@@ -7268,12 +7268,12 @@ class WSLCTests

7268 std::vector<std::string> chunks;

7269 auto onData = [&](const gsl::span<char>& data) { chunks.emplace_back(data.data(), data.size()); };

7270

7271 - wsl::windows::common::relay::MultiHandleWait io;

7271 + wsl::windows::common::io::MultiHandleWait io;

7272

7273 - io.AddHandle(std::make_unique<wsl::windows::common::relay::HTTPChunkBasedReadHandle>(std::move(readPipe), std::move(onData)));

7273 + io.AddHandle(std::make_unique<wsl::windows::common::io::HTTPChunkBasedReadHandle>(std::move(readPipe), std::move(onData)));

7274

7275 std::vector<char> buffer{Data.begin(), Data.end()};

7276 - io.AddHandle(std::make_unique<wsl::windows::common::relay::WriteHandle>(std::move(writePipe), buffer));

7276 + io.AddHandle(std::make_unique<wsl::windows::common::io::WriteHandle>(std::move(writePipe), buffer));

7277

7278 io.Run({});

7279

@@ -7336,8 +7336,8 @@ class WSLCTests

7336 std::vector<std::string> chunks;

7337 auto onData = [&](const gsl::span<char>& data) { chunks.emplace_back(data.data(), data.size()); };

7338

7339 - auto reader = std::make_unique<wsl::windows::common::relay::HTTPChunkBasedReadHandle>(

7340 - wsl::windows::common::relay::HandleWrapper{nullptr}, std::move(onData));

7339 + auto reader = std::make_unique<wsl::windows::common::io::HTTPChunkBasedReadHandle>(

7340 + wsl::windows::common::io::HandleWrapper{nullptr}, std::move(onData));

7341

7342 std::string allData;

7343 for (const auto& datum : Data)

@@ -7407,9 +7407,9 @@ class WSLCTests

7407 auto [readPipe, writePipe] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, false);

7408

7409 std::string readData;

7410 - wsl::windows::common::relay::MultiHandleWait io;

7410 + wsl::windows::common::io::MultiHandleWait io;

7411

7412 - io.AddHandle(std::make_unique<wsl::windows::common::relay::ReadHandle>(std::move(readPipe), [&](const gsl::span<char>& buffer) {

7412 + io.AddHandle(std::make_unique<wsl::windows::common::io::ReadHandle>(std::move(readPipe), [&](const gsl::span<char>& buffer) {

7413 if (!buffer.empty())

7414 {

7415 readData.append(buffer.data(), buffer.size());

@@ -7441,7 +7441,7 @@ class WSLCTests

7441 std::filesystem::remove("write-handle-test");

7442 });

7443

7444 - wsl::windows::common::relay::MultiHandleWait io;

7444 + wsl::windows::common::io::MultiHandleWait io;

7445 io.AddHandle(std::make_unique<WriteHandle>(outputFile.get(), writeBuffer));

7446 io.Run({});

7447

@@ -7461,7 +7461,7 @@ class WSLCTests

7461

7462 TEST_METHOD(DockerIORelay)

7463 {

7464 - using namespace wsl::windows::common::relay;

7464 + using namespace wsl::windows::common::io;

7465

7466 auto runTest = [](const std::vector<char>& Input, const std::string& ExpectedStdout, const std::string& ExpectedStderr) {

7467 auto [readPipe, writePipe] = wsl::windows::common::wslutil::OpenAnonymousPipe(16 * 1024, true, false);

@@ -8881,8 +8881,8 @@ class WSLCTests

8881 std::string output;

8882 auto onRead = [&](const gsl::span<char>& data) { output.append(data.data(), data.size()); };

8883

8884 - wsl::windows::common::relay::MultiHandleWait io;

8885 - io.AddHandle(std::make_unique<wsl::windows::common::relay::ReadHandle>(TtyOut, std::move(onRead)));

8884 + wsl::windows::common::io::MultiHandleWait io;

8885 + io.AddHandle(std::make_unique<wsl::windows::common::io::ReadHandle>(TtyOut, std::move(onRead)));

8886

8887 io.Run(60s);

8888

test/windows/WslcSdkTests.cpp

+3 -3

@@ -85,12 +85,12 @@ ProcessOutput WaitForProcessOutput(WslcProcess process, std::chrono::millisecond

86 // Read stdout / stderr concurrently so that full pipe buffers do not stall the process.

87 ProcessOutput output;

88 - wsl::windows::common::relay::MultiHandleWait io;

88 + wsl::windows::common::io::MultiHandleWait io;

90 - io.AddHandle(std::make_unique<wsl::windows::common::relay::ReadHandle>(

90 + io.AddHandle(std::make_unique<wsl::windows::common::io::ReadHandle>(

91 std::move(ownedStdout), [&](const auto& buffer) { output.stdoutOutput.append(buffer.data(), buffer.size()); }));

93 - io.AddHandle(std::make_unique<wsl::windows::common::relay::ReadHandle>(

93 + io.AddHandle(std::make_unique<wsl::windows::common::io::ReadHandle>(

94 std::move(ownedStderr), [&](const auto& buffer) { output.stderrOutput.append(buffer.data(), buffer.size()); }));

96 auto timeoutTime = std::chrono::steady_clock::now() + timeout;