Skip to content

Commit

Permalink
Fix socket cleanup on allocation failure in kqueue and epoll datapaths (
Browse files Browse the repository at this point in the history 
microsoft#1739)

* Fix socket cleanup on allocation failure in kqueue and epoll datapaths

* Fix build

* Alloc fail linux

* Fixup codecheck

* Fixup failures

* Fix macos

* Somre more attemptS

* Build fix

* Better OOM Support

* Build fix

* Another fix

* Fix kqueue too

* forgot comma

* Fix linux weird error

* Fix windows build

* 1 more windows fix
  • Loading branch information
@ThadHouse
ThadHouse authored Jun 21, 2021
1 parent f391fbf commit 4c20de2
Show file tree
Hide file tree
Showing 4 changed files with 213 additions and 87 deletions.
11 changes: 7 additions & 4 deletions .azure/azure-pipelines.ci.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -483,28 +483,28 @@ stages:
image: ubuntu-latest
platform: linux
tls: openssl
extraArgs: -Filter -*CredValidation*:*NthAllocFail*
extraArgs: -Filter -*CredValidation*
- template: ./templates/run-bvt.yml
parameters:
image: ubuntu-latest
platform: linux
tls: openssl
extraArtifactDir: '_Sanitize'
extraArgs: -Filter -*CredValidation*:*NthAllocFail* -ExtraArtifactDir Sanitize
extraArgs: -Filter -*CredValidation* -ExtraArtifactDir Sanitize
- template: ./templates/run-bvt.yml
parameters:
image: macOS-10.15
platform: macos
tls: openssl
logProfile: None
extraArgs: -Filter -*CredValidation*:*NthAllocFail*
extraArgs: -Filter -*CredValidation*
- template: ./templates/run-bvt.yml
parameters:
image: ubuntu-latest
platform: linux
tls: openssl
extraArtifactDir: '_SystemCrypto'
extraArgs: -Filter -*CredValidation*:*NthAllocFail* -ExtraArtifactDir SystemCrypto
extraArgs: -Filter -*CredValidation* -ExtraArtifactDir SystemCrypto

#
# SpinQuic Tests
Expand All @@ -528,19 +528,22 @@ stages:
parameters:
image: windows-latest
platform: windows
allocFail: 100
tls: openssl
- template: ./templates/run-spinquic.yml
parameters:
image: ubuntu-latest
platform: linux
tls: openssl
allocFail: 100
extraArtifactDir: '_Sanitize'
extraArgs: -ExtraArtifactDir Sanitize
- template: ./templates/run-spinquic.yml
parameters:
image: ubuntu-latest
platform: linux
tls: openssl
allocFail: 100

#
# Code Coverage
Expand Down
148 changes: 103 additions & 45 deletions src/platform/datapath_epoll.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -449,6 +449,26 @@ CxPlatDataPathQuerySockoptSupport(
}
#endif

void
CxPlatProcessorContextUninitialize(
_In_ CXPLAT_DATAPATH_PROC_CONTEXT* ProcContext
)
{
const eventfd_t Value = 1;
eventfd_write(ProcContext->EventFd, Value);
CxPlatThreadWait(&ProcContext->EpollWaitThread);
CxPlatThreadDelete(&ProcContext->EpollWaitThread);

epoll_ctl(ProcContext->EpollFd, EPOLL_CTL_DEL, ProcContext->EventFd, NULL);
close(ProcContext->EventFd);
close(ProcContext->EpollFd);

CxPlatPoolUninitialize(&ProcContext->RecvBlockPool);
CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendDataPool);
}

QUIC_STATUS
CxPlatProcessorContextInitialize(
_In_ CXPLAT_DATAPATH* Datapath,
Expand All @@ -462,6 +482,7 @@ CxPlatProcessorContextInitialize(
int Ret = 0;
uint32_t RecvPacketLength = 0;
BOOLEAN EventFdAdded = FALSE;
BOOLEAN ThreadCreated = FALSE;

CXPLAT_DBG_ASSERT(Datapath != NULL);

Expand Down Expand Up @@ -560,47 +581,33 @@ CxPlatProcessorContextInitialize(
goto Exit;
}

ThreadCreated = true;

Exit:

if (QUIC_FAILED(Status)) {
if (EventFdAdded) {
epoll_ctl(EpollFd, EPOLL_CTL_DEL, EventFd, NULL);
}
if (EventFd != INVALID_SOCKET) {
close(EventFd);
}
if (EpollFd != INVALID_SOCKET) {
close(EpollFd);
if (ThreadCreated) {
CxPlatProcessorContextUninitialize(ProcContext);
} else {
if (EventFdAdded) {
epoll_ctl(EpollFd, EPOLL_CTL_DEL, EventFd, NULL);
}
if (EventFd != INVALID_SOCKET) {
close(EventFd);
}
if (EpollFd != INVALID_SOCKET) {
close(EpollFd);
}
CxPlatPoolUninitialize(&ProcContext->RecvBlockPool);
CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendDataPool);
}
CxPlatPoolUninitialize(&ProcContext->RecvBlockPool);
CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendDataPool);
}

return Status;
}

void
CxPlatProcessorContextUninitialize(
_In_ CXPLAT_DATAPATH_PROC_CONTEXT* ProcContext
)
{
const eventfd_t Value = 1;
eventfd_write(ProcContext->EventFd, Value);
CxPlatThreadWait(&ProcContext->EpollWaitThread);
CxPlatThreadDelete(&ProcContext->EpollWaitThread);

epoll_ctl(ProcContext->EpollFd, EPOLL_CTL_DEL, ProcContext->EventFd, NULL);
close(ProcContext->EventFd);
close(ProcContext->EpollFd);

CxPlatPoolUninitialize(&ProcContext->RecvBlockPool);
CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendBufferPool);
CxPlatPoolUninitialize(&ProcContext->SendDataPool);
}

QUIC_STATUS
CxPlatDataPathInitialize(
_In_ uint32_t ClientRecvContextLength,
Expand Down Expand Up @@ -1408,16 +1415,21 @@ CxPlatSocketContextStartReceive(
SocketContext->Binding,
Status,
"epoll_ctl failed");

//
// Return any allocations
//
for (ssize_t i = 0; i < CXPLAT_MAX_BATCH_RECEIVE; i++) {
if (SocketContext->CurrentRecvBlocks[i] != NULL) {
CxPlatRecvDataReturn(&SocketContext->CurrentRecvBlocks[i]->RecvPacket);
}
}

goto Error;
}

Error:

if (QUIC_FAILED(Status)) {
close(SocketContext->SocketFd);
SocketContext->SocketFd = INVALID_SOCKET;
}

return Status;
}

Expand Down Expand Up @@ -1536,14 +1548,21 @@ CxPlatSocketContextRecvComplete(
DatagramHead);
}

Drop:
Status = CxPlatSocketContextPrepareReceive(SocketContext);
Drop: ;

//
// Prepare can only fail under low memory condition. Treat it as a fatal
// error.
//
CXPLAT_FRE_ASSERT(QUIC_SUCCEEDED(Status));
int32_t RetryCount = 0;
do {
Status = CxPlatSocketContextPrepareReceive(SocketContext);
} while (!QUIC_SUCCEEDED(Status) && RetryCount < 10);

if (!QUIC_SUCCEEDED(Status)) {
QuicTraceEvent(
DatapathErrorStatus,
"[data][%p] ERROR, %u, %s.",
SocketContext->Binding,
Status,
"CxPlatSocketContextPrepareReceive failed multiple times. Receive will no longer work.");
}
}

//
Expand Down Expand Up @@ -1764,6 +1783,7 @@ CxPlatSocketCreateUdp(
{
QUIC_STATUS Status = QUIC_STATUS_SUCCESS;
BOOLEAN IsServerSocket = RemoteAddress == NULL;
int32_t SuccessfulStartReceives = -1;

CXPLAT_DBG_ASSERT(Datapath->UdpHandlers.Receive != NULL || InternalFlags & CXPLAT_SOCKET_FLAG_PCP);

Expand Down Expand Up @@ -1807,6 +1827,7 @@ CxPlatSocketCreateUdp(
for (uint32_t i = 0; i < SocketCount; i++) {
Binding->SocketContexts[i].Binding = Binding;
Binding->SocketContexts[i].SocketFd = INVALID_SOCKET;
Binding->SocketContexts[i].CleanupFd = INVALID_SOCKET;
for (ssize_t j = 0; j < CXPLAT_MAX_BATCH_RECEIVE; j++) {
Binding->SocketContexts[i].RecvIov[j].iov_len =
Binding->Mtu - CXPLAT_MIN_IPV4_HEADER_SIZE - CXPLAT_UDP_HEADER_SIZE;
Expand Down Expand Up @@ -1858,11 +1879,13 @@ CxPlatSocketCreateUdp(
//
*NewBinding = Binding;

SuccessfulStartReceives = 0;
for (uint32_t i = 0; i < SocketCount; i++) {
Status =
CxPlatSocketContextStartReceive(
&Binding->SocketContexts[i]);
if (QUIC_FAILED(Status)) {
SuccessfulStartReceives = (int32_t)i;
goto Exit;
}
}
Expand All @@ -1877,7 +1900,42 @@ CxPlatSocketCreateUdp(
DatapathDestroyed,
"[data][%p] Destroyed",
Binding);
// TODO - Clean up socket contexts

if (SuccessfulStartReceives >= 0) {
uint32_t CurrentSocket = 0;
Binding->Shutdown = TRUE;

//
// First shutdown any sockets that fully started
//
for (; CurrentSocket < (uint32_t)SuccessfulStartReceives; CurrentSocket++) {
CxPlatSocketContextUninitialize(&Binding->SocketContexts[CurrentSocket]);
}
//
// Then shutdown any sockets that failed to start
//
for (; CurrentSocket < SocketCount; CurrentSocket++) {
CxPlatSocketContextUninitializeComplete(&Binding->SocketContexts[CurrentSocket]);
}
} else {
//
// No sockets fully started. Only uninitialize static things
//
for (uint32_t i = 0; i < SocketCount; i++) {
CXPLAT_SOCKET_CONTEXT* SocketContext = &Binding->SocketContexts[i];
int EpollFd = SocketContext->ProcContext->EpollFd;
if (SocketContext->CleanupFd != INVALID_SOCKET) {
epoll_ctl(EpollFd, EPOLL_CTL_DEL, SocketContext->CleanupFd, NULL);
close(SocketContext->CleanupFd);
}
if (SocketContext->SocketFd != INVALID_SOCKET) {
epoll_ctl(EpollFd, EPOLL_CTL_DEL, SocketContext->SocketFd, NULL);
close(SocketContext->SocketFd);
}
CxPlatRundownRelease(&Binding->Rundown);
}
}
CxPlatRundownReleaseAndWait(&Binding->Rundown);
CxPlatRundownRelease(&Datapath->BindingsRundown);
CxPlatRundownUninitialize(&Binding->Rundown);
for (uint32_t i = 0; i < SocketCount; i++) {
Expand Down
Loading

0 comments on commit 4c20de2

Please sign in to comment.