From 4c20de207c1a700123a14cb5d6c9f77c27aade27 Mon Sep 17 00:00:00 2001 From: Thad House Date: Mon, 21 Jun 2021 15:56:09 -0700 Subject: [PATCH] Fix socket cleanup on allocation failure in kqueue and epoll datapaths (#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 --- .azure/azure-pipelines.ci.yml | 11 ++- src/platform/datapath_epoll.c | 148 ++++++++++++++++++++++---------- src/platform/datapath_kqueue.c | 120 ++++++++++++++++++-------- src/platform/datapath_winuser.c | 21 ++++- 4 files changed, 213 insertions(+), 87 deletions(-) diff --git a/.azure/azure-pipelines.ci.yml b/.azure/azure-pipelines.ci.yml index 1538b77951..152e203caf 100644 --- a/.azure/azure-pipelines.ci.yml +++ b/.azure/azure-pipelines.ci.yml @@ -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 @@ -528,12 +528,14 @@ 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 @@ -541,6 +543,7 @@ stages: image: ubuntu-latest platform: linux tls: openssl + allocFail: 100 # # Code Coverage diff --git a/src/platform/datapath_epoll.c b/src/platform/datapath_epoll.c index 22fae9844c..7bdf8b4e75 100644 --- a/src/platform/datapath_epoll.c +++ b/src/platform/datapath_epoll.c @@ -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, @@ -462,6 +482,7 @@ CxPlatProcessorContextInitialize( int Ret = 0; uint32_t RecvPacketLength = 0; BOOLEAN EventFdAdded = FALSE; + BOOLEAN ThreadCreated = FALSE; CXPLAT_DBG_ASSERT(Datapath != NULL); @@ -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, @@ -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; } @@ -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."); + } } // @@ -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); @@ -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; @@ -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; } } @@ -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++) { diff --git a/src/platform/datapath_kqueue.c b/src/platform/datapath_kqueue.c index 2385c0cf06..5593a47373 100644 --- a/src/platform/datapath_kqueue.c +++ b/src/platform/datapath_kqueue.c @@ -372,6 +372,25 @@ CxPlatSocketSendInternal( _In_ BOOLEAN IsPendedSend ); +void +CxPlatProcessorContextUninitialize( + _In_ CXPLAT_DATAPATH_PROC_CONTEXT* ProcContext + ) +{ + struct kevent Event = {0}; + EV_SET(&Event, ProcContext->KqueueFd, EVFILT_USER, EV_ADD | EV_CLEAR, NOTE_TRIGGER, 0, NULL); + kevent(ProcContext->KqueueFd, &Event, 1, NULL, 0, NULL); + CxPlatThreadWait(&ProcContext->KqueueWaitThread); + CxPlatThreadDelete(&ProcContext->KqueueWaitThread); + + close(ProcContext->KqueueFd); + + CxPlatPoolUninitialize(&ProcContext->RecvBlockPool); + CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool); + CxPlatPoolUninitialize(&ProcContext->SendBufferPool); + CxPlatPoolUninitialize(&ProcContext->SendDataPool); +} + QUIC_STATUS CxPlatProcessorContextInitialize( _In_ CXPLAT_DATAPATH* Datapath, @@ -382,6 +401,7 @@ CxPlatProcessorContextInitialize( QUIC_STATUS Status = QUIC_STATUS_SUCCESS; int KqueueFd = INVALID_SOCKET; uint32_t RecvPacketLength = 0; + BOOLEAN ThreadCreated = FALSE; CXPLAT_DBG_ASSERT(Datapath != NULL); @@ -451,37 +471,22 @@ CxPlatProcessorContextInitialize( Exit: if (QUIC_FAILED(Status)) { - if (KqueueFd != INVALID_SOCKET) { - close(KqueueFd); + if (ThreadCreated) { + CxPlatProcessorContextUninitialize(ProcContext); + } else { + if (KqueueFd != INVALID_SOCKET) { + close(KqueueFd); + } + 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 - ) -{ - struct kevent Event = {0}; - EV_SET(&Event, ProcContext->KqueueFd, EVFILT_USER, EV_ADD | EV_CLEAR, NOTE_TRIGGER, 0, NULL); - kevent(ProcContext->KqueueFd, &Event, 1, NULL, 0, NULL); - CxPlatThreadWait(&ProcContext->KqueueWaitThread); - CxPlatThreadDelete(&ProcContext->KqueueWaitThread); - - close(ProcContext->KqueueFd); - - CxPlatPoolUninitialize(&ProcContext->RecvBlockPool); - CxPlatPoolUninitialize(&ProcContext->LargeSendBufferPool); - CxPlatPoolUninitialize(&ProcContext->SendBufferPool); - CxPlatPoolUninitialize(&ProcContext->SendDataPool); -} - QUIC_STATUS CxPlatDataPathInitialize( _In_ uint32_t ClientRecvContextLength, @@ -1140,16 +1145,19 @@ CxPlatSocketContextStartReceive( SocketContext->Binding, Status, "kevent failed"); + + // + // Return any allocations + // + if (SocketContext->CurrentRecvBlock != NULL) { + CxPlatRecvDataReturn(&SocketContext->CurrentRecvBlock->RecvPacket); + } + goto Error; } Error: - if (QUIC_FAILED(Status)) { - close(SocketContext->SocketFd); - SocketContext->SocketFd = INVALID_SOCKET; - } - return Status; } @@ -1243,13 +1251,19 @@ CxPlatSocketContextRecvComplete( RecvPacket); } - Status = CxPlatSocketContextPrepareReceive(SocketContext); + int32_t RetryCount = 0; + do { + Status = CxPlatSocketContextPrepareReceive(SocketContext); + } while (!QUIC_SUCCEEDED(Status) && RetryCount < 10); - // - // Prepare can only fail under low memory condition. Treat it as a fatal - // error. - // - CXPLAT_FRE_ASSERT(QUIC_SUCCEEDED(Status)); + if (!QUIC_SUCCEEDED(Status)) { + QuicTraceEvent( + DatapathErrorStatus, + "[data][%p] ERROR, %u, %s.", + SocketContext->Binding, + Status, + "CxPlatSocketContextPrepareReceive failed multiple times. Receive will no longer work."); + } } // @@ -1412,6 +1426,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); @@ -1494,11 +1509,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; } } @@ -1513,7 +1530,36 @@ 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]; + if (SocketContext->SocketFd != INVALID_SOCKET) { + close(SocketContext->SocketFd); + } + CxPlatRundownRelease(&Binding->Rundown); + } + } + CxPlatRundownReleaseAndWait(&Binding->Rundown); CxPlatRundownRelease(&Datapath->BindingsRundown); CxPlatRundownUninitialize(&Binding->Rundown); for (uint32_t i = 0; i < SocketCount; i++) { diff --git a/src/platform/datapath_winuser.c b/src/platform/datapath_winuser.c index b9adf8746a..a6aa984e50 100644 --- a/src/platform/datapath_winuser.c +++ b/src/platform/datapath_winuser.c @@ -2833,6 +2833,11 @@ CxPlatSocketStartReceive( DatapathProc->Index); if (SocketProc->CurrentRecvContext == NULL) { Status = QUIC_STATUS_OUT_OF_MEMORY; + QuicTraceEvent( + AllocFailure, + "Allocation of '%s' failed. (%llu bytes)", + "Socket Receive Buffer", + SocketProc->Parent->Datapath->RecvPayloadOffset + MAX_URO_PAYLOAD_LENGTH); goto Error; } } @@ -3157,7 +3162,21 @@ CxPlatDataPathUdpRecvComplete( // // Try to start a new receive. // - (void)CxPlatSocketStartReceive(SocketProc, DatapathProc); + int32_t RetryCount = 0; + QUIC_STATUS Status; + do { + Status = CxPlatSocketStartReceive(SocketProc, DatapathProc); + } while (!QUIC_SUCCEEDED(Status) && RetryCount < 10); + + if (!QUIC_SUCCEEDED(Status)) { + QuicTraceEvent( + DatapathErrorStatus, + "[data][%p] ERROR, %u, %s.", + SocketProc->Parent, + Status, + "CxPlatSocketStartReceive failed multiple times. Receive will no longer work."); + } + } void