Skip to content

Migrate relay IO to MultiHandleWait infrastructure #40730

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
benhillis wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Migrate relay IO to MultiHandleWait infrastructure #40730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 25 additions & 22 deletions src/windows/common/Dmesg.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ Module Name:
#include "precomp.h"
#include "Dmesg.h"

namespace io = wsl::windows::common::io;

DmesgCollector::DmesgCollector(
GUID VmId, const wil::unique_event& ExitEvent, bool EnableTelemetry, bool EnableDebugConsole, const std::wstring& Com1PipeName, wil::unique_handle&& OutputHandle) :
m_com1PipeName(Com1PipeName),
Expand Down Expand Up @@ -80,25 +82,27 @@ std::pair<std::wstring, std::thread> DmesgCollector::StartDmesgThread(InputSourc
// When the pipe connects, start reading data.
wsl::windows::common::helpers::ConnectPipe(Pipe.get(), INFINITE, m_exitEvents);

std::vector<char> buffer(LX_RELAY_BUFFER_SIZE);
const auto allBuffer = gsl::make_span(buffer);
OVERLAPPED overlapped = {};
const wil::unique_event overlappedEvent(wil::EventOptions::ManualReset);
overlapped.hEvent = overlappedEvent.get();
for (;;)
{
overlappedEvent.ResetEvent();
const auto bytesRead = wsl::windows::common::relay::InterruptableRead(
Pipe.get(), gslhelpers::convert_span<gsl::byte>(allBuffer), m_exitEvents, &overlapped);

if (bytesRead == 0)
{
break;
}

auto validBuffer = allBuffer.subspan(0, bytesRead);
ProcessInput(Source, validBuffer);
}
io::MultiHandleWait ioWait;
ioWait.AddHandle(
std::make_unique<io::ReadHandle>(
io::HandleWrapper{Pipe.get()},
[this, Source](const gsl::span<char>& buffer) {
if (!buffer.empty())
{
ProcessInput(Source, buffer);
}
}),
io::MultiHandleWait::IgnoreErrors);

ioWait.AddHandle(
std::make_unique<io::EventHandle>(io::HandleWrapper{m_exitEvents[0]}),
io::MultiHandleWait::CancelOnCompleted | io::MultiHandleWait::NeedNotComplete);

ioWait.AddHandle(
std::make_unique<io::EventHandle>(io::HandleWrapper{m_exitEvents[1]}),
io::MultiHandleWait::CancelOnCompleted | io::MultiHandleWait::NeedNotComplete);

ioWait.Run(std::nullopt);
}
catch (...)
{
Expand Down Expand Up @@ -161,8 +165,7 @@ void DmesgCollector::ProcessInput(InputSource Source, const gsl::span<char>& Inp
if (m_outputHandle != nullptr)
{
m_overlappedEvent.ResetEvent();
if (wsl::windows::common::relay::InterruptableWrite(
m_outputHandle.get(), gslhelpers::convert_span<gsl::byte>(Input), m_exitEvents, &m_overlapped) == 0)
if (io::InterruptableWrite(m_outputHandle.get(), gslhelpers::convert_span<gsl::byte>(Input), m_exitEvents, &m_overlapped) == 0)
{
m_outputHandle = nullptr;
}
Expand Down Expand Up @@ -192,7 +195,7 @@ void DmesgCollector::WriteToCom1(const gsl::span<char>& Input)

m_overlappedEvent.ResetEvent();
const auto buffer = gslhelpers::convert_span<gsl::byte>(Input);
if (wsl::windows::common::relay::InterruptableWrite(m_com1Pipe.get(), buffer, m_exitEvents, &m_overlapped) == 0)
if (io::InterruptableWrite(m_com1Pipe.get(), buffer, m_exitEvents, &m_overlapped) == 0)
{
if (m_debugConsole || !m_pipeServer)
{
Expand Down
115 changes: 113 additions & 2 deletions src/windows/common/HandleIO.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,8 +209,8 @@ HANDLE EventHandle::GetHandle() const

// ReadHandle

ReadHandle::ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead) :
Handle(std::move(MovedHandle)), OnRead(OnRead), Offset(InitializeFileOffset(Handle.Get()))
ReadHandle::ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead, size_t BufferSize) :
Handle(std::move(MovedHandle)), OnRead(std::move(OnRead)), Buffer(BufferSize), Offset(InitializeFileOffset(Handle.Get()))
{
Overlapped.hEvent = Event.get();
}
Expand Down Expand Up @@ -1133,3 +1133,114 @@ bool MultiHandleWait::Run(std::optional<std::chrono::milliseconds> Timeout)

return !m_cancel;
}

DWORD wsl::windows::common::io::InterruptableRead(
_In_ HANDLE InputHandle, _In_ gsl::span<gsl::byte> Buffer, _In_ const std::vector<HANDLE>& ExitHandles, _In_opt_ LPOVERLAPPED Overlapped)
{
// Initialize an overlapped structure if one was not provided by the caller.
OVERLAPPED overlapped = {};
wil::unique_event overlappedEvent = {};
if (!ARGUMENT_PRESENT(Overlapped))
{
overlappedEvent.create(wil::EventOptions::ManualReset);
overlapped.hEvent = overlappedEvent.get();
Overlapped = &overlapped;
}

DWORD bytesRead = 0;
if (!ReadFile(InputHandle, Buffer.data(), gsl::narrow_cast<DWORD>(Buffer.size()), &bytesRead, Overlapped))
{
auto lastError = GetLastError();
if ((lastError == ERROR_HANDLE_EOF) || (lastError == ERROR_BROKEN_PIPE))
{
return 0;
}

THROW_LAST_ERROR_IF_MSG(lastError != ERROR_IO_PENDING, "Handle: 0x%p", (void*)InputHandle);

auto cancelRead = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&] {
CancelIoEx(InputHandle, Overlapped);
GetOverlappedResult(InputHandle, Overlapped, &bytesRead, TRUE);
});

// Wait for the read to complete, or the client to exit.
if (!InterruptableWait(Overlapped->hEvent, ExitHandles))
{
return 0;
}

if (!GetOverlappedResult(InputHandle, Overlapped, &bytesRead, FALSE))
{
lastError = GetLastError();
if ((lastError == ERROR_HANDLE_EOF) || (lastError == ERROR_BROKEN_PIPE))
{
return 0;
}

THROW_LAST_ERROR();
}

cancelRead.release();
}

return bytesRead;
}

bool wsl::windows::common::io::InterruptableWait(_In_ HANDLE WaitObject, _In_ const std::vector<HANDLE>& ExitHandles)
{
// Wait for the object to become signaled or one of the exit handles to be signaled.
std::vector<HANDLE> waitObjects{WaitObject};
for (const auto& exitHandle : ExitHandles)
{
waitObjects.push_back(exitHandle);
}

const DWORD waitResult = WaitForMultipleObjects(gsl::narrow_cast<DWORD>(waitObjects.size()), waitObjects.data(), FALSE, INFINITE);
if (waitResult != WAIT_OBJECT_0)
{
if (waitResult > WAIT_OBJECT_0 && waitResult < WAIT_OBJECT_0 + waitObjects.size())
{
return false;
}

THROW_HR_MSG(E_FAIL, "WaitForMultipleObjects %d", waitResult);
}

return true;
}

DWORD wsl::windows::common::io::InterruptableWrite(
_In_ HANDLE OutputHandle, _In_ gsl::span<const gsl::byte> Buffer, _In_ const std::vector<HANDLE>& ExitHandles, _In_ LPOVERLAPPED Overlapped)
{
const DWORD bytesToWrite = gsl::narrow_cast<DWORD>(Buffer.size());
DWORD bytesWritten = 0;
BOOL success = WriteFile(OutputHandle, Buffer.data(), bytesToWrite, &bytesWritten, Overlapped);
if (!success)
{
const auto lastError = GetLastError();
if (lastError == ERROR_NO_DATA)
{
return 0;
}

THROW_LAST_ERROR_IF(lastError != ERROR_IO_PENDING);

auto cancelWrite = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&] {
CancelIoEx(OutputHandle, Overlapped);
GetOverlappedResult(OutputHandle, Overlapped, &bytesWritten, TRUE);
});

if (InterruptableWait(Overlapped->hEvent, ExitHandles))
{
success = GetOverlappedResult(OutputHandle, Overlapped, &bytesWritten, FALSE);
if (success)
{
cancelWrite.release();
}
}
}

WI_ASSERT(!success || (bytesWritten == bytesToWrite));

return bytesWritten;
}
24 changes: 20 additions & 4 deletions src/windows/common/HandleIO.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ class ReadHandle : public OverlappedIOHandle
NON_COPYABLE(ReadHandle);
NON_MOVABLE(ReadHandle);

ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead);
ReadHandle(HandleWrapper&& MovedHandle, std::function<void(const gsl::span<char>& Buffer)>&& OnRead, size_t BufferSize = LX_RELAY_BUFFER_SIZE);
virtual ~ReadHandle();

void Schedule() override;
Expand All @@ -111,7 +111,7 @@ class ReadHandle : public OverlappedIOHandle
std::function<void(const gsl::span<char>& Buffer)> OnRead;
wil::unique_event Event{wil::EventOptions::ManualReset};
OVERLAPPED Overlapped{};
BufferWrapper Buffer{LX_RELAY_BUFFER_SIZE};
BufferWrapper Buffer;
LARGE_INTEGER Offset{};
};

Expand Down Expand Up @@ -234,8 +234,16 @@ class RelayHandle : public OverlappedIOHandle
NON_COPYABLE(RelayHandle);
NON_MOVABLE(RelayHandle);

RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output) :
Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output))
RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output, size_t BufferSize = LX_RELAY_BUFFER_SIZE)
requires std::is_same_v<TRead, ReadHandle>
:
Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }, BufferSize), Write(std::move(Output))
{
}

RelayHandle(HandleWrapper&& Input, HandleWrapper&& Output)
requires(!std::is_same_v<TRead, ReadHandle>)
: Read(std::move(Input), [this](const gsl::span<char>& Buffer) { return OnRead(Buffer); }), Write(std::move(Output))
{
}

Expand Down Expand Up @@ -397,4 +405,12 @@ class MultiHandleWait

DEFINE_ENUM_FLAG_OPERATORS(MultiHandleWait::Flags);

// Standalone cancellable IO operations. These perform a single overlapped read/write
// and can be cancelled via exit event handles.
DWORD InterruptableRead(_In_ HANDLE InputHandle, _In_ gsl::span<gsl::byte> Buffer, _In_ const std::vector<HANDLE>& ExitHandles, _In_opt_ LPOVERLAPPED Overlapped = nullptr);

DWORD InterruptableWrite(_In_ HANDLE OutputHandle, _In_ gsl::span<const gsl::byte> Buffer, _In_ const std::vector<HANDLE>& ExitHandles, _In_ LPOVERLAPPED Overlapped);

bool InterruptableWait(_In_ HANDLE WaitObject, _In_ const std::vector<HANDLE>& ExitHandles = {});

} // namespace wsl::windows::common::io
Loading