Skip to content

Commit

Permalink
Update IpcStreamFactory state machine to handle being started on a th…
Browse files Browse the repository at this point in the history 
…read that ends (#43711)

Handles the case on Windows where Async IO is cancelled due to the thread coreclr started on ending.
  • Loading branch information
John Salem authored Nov 5, 2020
1 parent e7267cd commit d65f284
Show file tree
Hide file tree
Showing 2 changed files with 59 additions and 53 deletions.
108 changes: 55 additions & 53 deletions src/coreclr/src/debug/debug-pal/win/diagnosticsipc.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,7 @@ IpcStream *IpcStream::DiagnosticsIpc::Accept(ErrorCallback callback)
_ASSERTE(mode == ConnectionMode::LISTEN);

DWORD dwDummy = 0;
IpcStream *pStream = nullptr;
bool fSuccess = GetOverlappedResult(
_hPipe, // handle
&_oOverlap, // overlapped
Expand All @@ -144,11 +145,14 @@ IpcStream *IpcStream::DiagnosticsIpc::Accept(ErrorCallback callback)
{
if (callback != nullptr)
callback("Failed to GetOverlappedResults for NamedPipe server", ::GetLastError());
return nullptr;
// close the pipe (cleaned up and reset below)
::CloseHandle(_hPipe);
}
else
{
// create new IpcStream using handle (passes ownership to pStream)
pStream = new IpcStream(_hPipe, ConnectionMode::LISTEN);
}

// create new IpcStream using handle and reset the Server object so it can listen again
IpcStream *pStream = new IpcStream(_hPipe, ConnectionMode::LISTEN);

// reset the server
_hPipe = INVALID_HANDLE_VALUE;
Expand All @@ -159,7 +163,7 @@ IpcStream *IpcStream::DiagnosticsIpc::Accept(ErrorCallback callback)
if (!fSuccess)
{
delete pStream;
return nullptr;
pStream = nullptr;
}

return pStream;
Expand Down Expand Up @@ -434,43 +438,42 @@ bool IpcStream::Read(void *lpBuffer, const uint32_t nBytesToRead, uint32_t &nByt

if (!fSuccess)
{
DWORD dwError = GetLastError();

// if we're waiting infinitely, only make one syscall
if (timeoutMs == InfiniteTimeout)
if (timeoutMs == InfiniteTimeout && dwError == ERROR_IO_PENDING)
{
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesRead, // out actual number of bytes read
true) != 0; // block until async IO completes
}
else
else if (dwError == ERROR_IO_PENDING)
{
DWORD dwError = GetLastError();
if (dwError == ERROR_IO_PENDING)
// Wait on overlapped IO event (triggers when async IO is complete regardless of success)
// or timeout
DWORD dwWait = WaitForSingleObject(_oOverlap.hEvent, (DWORD)timeoutMs);
if (dwWait == WAIT_OBJECT_0)
{
// Wait on overlapped IO event (triggers when async IO is complete regardless of success)
// or timeout
DWORD dwWait = WaitForSingleObject(_oOverlap.hEvent, (DWORD)timeoutMs);
if (dwWait == WAIT_OBJECT_0)
{
// async IO compelted, get the result
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesRead, // out actual number of bytes read
true) != 0; // block until async IO completes
}
else
// async IO compelted, get the result
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesRead, // out actual number of bytes read
true) != 0; // block until async IO completes
}
else
{
// We either timed out or something else went wrong.
// For any error, attempt to cancel IO and ensure the cancel happened
if (CancelIoEx(_hPipe, overlap) != 0)
{
// We either timed out or something else went wrong.
// For any error, attempt to cancel IO and ensure the cancel happened
if (CancelIoEx(_hPipe, overlap) != 0)
{
// check if the async write beat the cancellation
fSuccess = GetOverlappedResult(_hPipe, overlap, &nNumberOfBytesRead, true) != 0;
// Failure here isn't recoverable, so return as such
}
// check if the async write beat the cancellation
fSuccess = GetOverlappedResult(_hPipe, overlap, &nNumberOfBytesRead, true) != 0;
// Failure here isn't recoverable, so return as such
}
}
}
// error is unrecoverable, so return as such
}

nBytesRead = static_cast<uint32_t>(nNumberOfBytesRead);
Expand All @@ -492,43 +495,42 @@ bool IpcStream::Write(const void *lpBuffer, const uint32_t nBytesToWrite, uint32

if (!fSuccess)
{
DWORD dwError = GetLastError();

// if we're waiting infinitely, only make one syscall
if (timeoutMs == InfiniteTimeout)
if (timeoutMs == InfiniteTimeout && dwError == ERROR_IO_PENDING)
{
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesWritten, // out actual number of bytes written
true) != 0; // block until async IO completes
}
else
else if (dwError == ERROR_IO_PENDING)
{
DWORD dwError = GetLastError();
if (dwError == ERROR_IO_PENDING)
// Wait on overlapped IO event (triggers when async IO is complete regardless of success)
// or timeout
DWORD dwWait = WaitForSingleObject(_oOverlap.hEvent, (DWORD)timeoutMs);
if (dwWait == WAIT_OBJECT_0)
{
// Wait on overlapped IO event (triggers when async IO is complete regardless of success)
// or timeout
DWORD dwWait = WaitForSingleObject(_oOverlap.hEvent, (DWORD)timeoutMs);
if (dwWait == WAIT_OBJECT_0)
{
// async IO compelted, get the result
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesWritten, // out actual number of bytes written
true) != 0; // block until async IO completes
}
else
// async IO compelted, get the result
fSuccess = GetOverlappedResult(_hPipe, // pipe
overlap, // overlapped
&nNumberOfBytesWritten, // out actual number of bytes written
true) != 0; // block until async IO completes
}
else
{
// We either timed out or something else went wrong.
// For any error, attempt to cancel IO and ensure the cancel happened
if (CancelIoEx(_hPipe, overlap) != 0)
{
// We either timed out or something else went wrong.
// For any error, attempt to cancel IO and ensure the cancel happened
if (CancelIoEx(_hPipe, overlap) != 0)
{
// check if the async write beat the cancellation
fSuccess = GetOverlappedResult(_hPipe, overlap, &nNumberOfBytesWritten, true) != 0;
// Failure here isn't recoverable, so return as such
}
// check if the async write beat the cancellation
fSuccess = GetOverlappedResult(_hPipe, overlap, &nNumberOfBytesWritten, true) != 0;
// Failure here isn't recoverable, so return as such
}
}
}
// error is unrecoverable, so return as such
}

nBytesWritten = static_cast<uint32_t>(nNumberOfBytesWritten);
Expand Down
4 changes: 4 additions & 0 deletions src/coreclr/src/vm/ipcstreamfactory.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,10 @@ IpcStream *IpcStreamFactory::GetNextAvailableStream(ErrorCallback callback)
if (pStream == nullptr) // only use first signaled stream; will get others on subsequent calls
{
pStream = ((DiagnosticPort*)(rgIpcPollHandles[i].pUserData))->GetConnectedStream(callback);
if (pStream == nullptr)
{
fSawError = true;
}
s_currentPort = (DiagnosticPort*)(rgIpcPollHandles[i].pUserData);
}
STRESS_LOG2(LF_DIAGNOSTICS_PORT, LL_INFO10, "IpcStreamFactory::GetNextAvailableStream - SIG :: Poll attempt: %d, connection %d signalled.\n", nPollAttempts, i);
Expand Down

0 comments on commit d65f284

Please sign in to comment.