From af7ca8ca286cbe562c179a8130e65b6961bb5bad Mon Sep 17 00:00:00 2001 From: Alexis Campailla Date: Fri, 19 Jun 2015 21:22:48 +0200 Subject: [PATCH] RejoinCOWPages and background threads should be synchronized RejoinCOWPages is copying dirty pages to a new view of the memory map. If another thread modifies the heap between when RejoincCOWPages copies the data and when the view is remapped, the modification will be lost, leading to a memory corruption. In short, when RejoincCOWPages is running, all other threads must be stopped. Fixes: https://github.com/MSOpenTech/redis/issues/244 --- src/Win32_Interop/Win32_Interop.vcxproj | 2 + src/Win32_Interop/Win32_QFork.cpp | 2 + src/Win32_Interop/Win32_ThreadControl.c | 111 ++++++++++++++++++++++++ src/Win32_Interop/Win32_ThreadControl.h | 43 +++++++++ src/Win32_Interop/win32fixes.c | 15 ++-- src/bio.c | 8 ++ src/redis.c | 31 ++++--- 7 files changed, 194 insertions(+), 18 deletions(-) create mode 100644 src/Win32_Interop/Win32_ThreadControl.c create mode 100644 src/Win32_Interop/Win32_ThreadControl.h diff --git a/src/Win32_Interop/Win32_Interop.vcxproj b/src/Win32_Interop/Win32_Interop.vcxproj index 7c86a2cd..fa3625ca 100644 --- a/src/Win32_Interop/Win32_Interop.vcxproj +++ b/src/Win32_Interop/Win32_Interop.vcxproj @@ -32,6 +32,7 @@ + @@ -51,6 +52,7 @@ + diff --git a/src/Win32_Interop/Win32_QFork.cpp b/src/Win32_Interop/Win32_QFork.cpp index 0d880ff7..10845077 100644 --- a/src/Win32_Interop/Win32_QFork.cpp +++ b/src/Win32_Interop/Win32_QFork.cpp @@ -43,6 +43,7 @@ #include "Win32_CommandLine.h" #include "Win32_RedisLog.h" #include "Win32_StackTrace.h" +#include "Win32_ThreadControl.h" #include #include @@ -1321,6 +1322,7 @@ extern "C" ParseCommandLineArguments(argc, argv); SetupLogging(); StackTraceInit(); + InitThreadControl(); } catch (system_error syserr) { exit(-1); } catch (runtime_error runerr) { diff --git a/src/Win32_Interop/Win32_ThreadControl.c b/src/Win32_Interop/Win32_ThreadControl.c new file mode 100644 index 00000000..4cbe871e --- /dev/null +++ b/src/Win32_Interop/Win32_ThreadControl.c @@ -0,0 +1,111 @@ +/* + * Copyright (c), Microsoft Open Technologies, Inc. + * All rights reserved. + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * - Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * - Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR + * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER + * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, + * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ + +#include +#include + + +volatile LONG g_NumWorkerThreads = 0; + +// Safe mode means the threads are not touching the heap, or they are suspended because of an explicit suspension request +// Threads in safe mode because they are not touching the heap will block if trying to transition to unsafe mode while a suspension is requested +volatile LONG g_NumWorkerThreadsInSafeMode = 0; + +volatile LONG g_SuspensionRequested = 0; +HANDLE g_hResumeFromSuspension; + +CRITICAL_SECTION g_ThreadControlMutex; + + +void InitThreadControl() { + InitializeCriticalSection(&g_ThreadControlMutex); + g_hResumeFromSuspension = CreateEvent(NULL, TRUE, TRUE, NULL); + if (!g_hResumeFromSuspension) { + exit(GetLastError()); + } +} + +void IncrementWorkerThreadCount() { + EnterCriticalSection(&g_ThreadControlMutex); + g_NumWorkerThreads++; + LeaveCriticalSection(&g_ThreadControlMutex); +} + +void DecrementWorkerThreadCount() { + EnterCriticalSection(&g_ThreadControlMutex); + g_NumWorkerThreads--; + LeaveCriticalSection(&g_ThreadControlMutex); +} + + +// Returns TRUE if threads are already in safe mode or suspended +BOOL SuspensionCompleted() { + BOOL result; + EnterCriticalSection(&g_ThreadControlMutex); + result = (g_NumWorkerThreadsInSafeMode == g_NumWorkerThreads); + LeaveCriticalSection(&g_ThreadControlMutex); + return result; +} + +// This is meant to be called from the main thread only. +void RequestSuspension() { + if (!g_SuspensionRequested) { + if (!ResetEvent(g_hResumeFromSuspension)) { + exit(GetLastError()); + } + InterlockedOr(&g_SuspensionRequested, 1); + } +} + +void ResumeFromSuspension() { + // This is meant to be called from the main thread only. + assert(g_SuspensionRequested && SuspensionCompleted()); + + InterlockedAnd(&g_SuspensionRequested, 0); + if (!SetEvent(g_hResumeFromSuspension)) { + exit(GetLastError()); + } +} + +void WorkerThread_EnterSafeMode() { + EnterCriticalSection(&g_ThreadControlMutex); + g_NumWorkerThreadsInSafeMode++; + LeaveCriticalSection(&g_ThreadControlMutex); +} + +void WorkerThread_ExitSafeMode() { + for(;;) { + EnterCriticalSection(&g_ThreadControlMutex); + if (g_SuspensionRequested) { + LeaveCriticalSection(&g_ThreadControlMutex); + if (WaitForSingleObject(g_hResumeFromSuspension, INFINITE) != WAIT_OBJECT_0) { + exit(GetLastError()); + } + continue; + } else { + g_NumWorkerThreadsInSafeMode--; + LeaveCriticalSection(&g_ThreadControlMutex); + break; + } + } +} + diff --git a/src/Win32_Interop/Win32_ThreadControl.h b/src/Win32_Interop/Win32_ThreadControl.h new file mode 100644 index 00000000..d79d5e12 --- /dev/null +++ b/src/Win32_Interop/Win32_ThreadControl.h @@ -0,0 +1,43 @@ +/* + * Copyright (c), Microsoft Open Technologies, Inc. + * All rights reserved. + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * - Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * - Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR + * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER + * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, + * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ + +#pragma once + +#ifdef __cplusplus +extern "C" { +#endif + + +void InitThreadControl(); +void IncrementWorkerThreadCount(); +void DecrementWorkerThreadCount(); +void RequestSuspension(); +BOOL SuspensionCompleted(); +void ResumeFromSuspension(); + +void WorkerThread_EnterSafeMode(); +void WorkerThread_ExitSafeMode(); + + +#ifdef __cplusplus +} +#endif diff --git a/src/Win32_Interop/win32fixes.c b/src/Win32_Interop/win32fixes.c index 0955b4a2..6aef31d0 100644 --- a/src/Win32_Interop/win32fixes.c +++ b/src/Win32_Interop/win32fixes.c @@ -20,6 +20,7 @@ #include #include //#include +#include "Win32_ThreadControl.h" /* Redefined here to avoid redis.h so it can be used in other projects */ #define REDIS_NOTUSED(V) ((void) V) @@ -149,12 +150,16 @@ typedef struct thread_params /* Proxy function by windows thread requirements */ static unsigned __stdcall win32_proxy_threadproc(void *arg) { + IncrementWorkerThreadCount(); + __try { + thread_params *p = (thread_params *) arg; + p->func(p->arg); - thread_params *p = (thread_params *) arg; - p->func(p->arg); - - /* Dealocate params */ - free(p); + /* Dealocate params */ + free(p); + } __finally { + DecrementWorkerThreadCount(); + } _endthreadex(0); return 0; diff --git a/src/bio.c b/src/bio.c index e187942d..c0b39308 100644 --- a/src/bio.c +++ b/src/bio.c @@ -64,6 +64,7 @@ #include "bio.h" #ifdef _WIN32 #include "win32_Interop/win32fixes.h" +#include "Win32_Interop/Win32_ThreadControl.h" #endif static pthread_t bio_threads[REDIS_BIO_NUM_OPS]; @@ -157,7 +158,10 @@ void *bioProcessBackgroundJobs(void *arg) { // needs much rework. Cancellability requires a shared event. #endif + WIN32_ONLY(WorkerThread_EnterSafeMode()); pthread_mutex_lock(&bio_mutex[type]); + WIN32_ONLY(WorkerThread_ExitSafeMode()); + /* Block SIGALRM so we are sure that only the main thread will * receive the watchdog signal. */ sigemptyset(&sigset); @@ -171,7 +175,9 @@ void *bioProcessBackgroundJobs(void *arg) { /* The loop always starts with the lock hold. */ if (listLength(bio_jobs[type]) == 0) { + WIN32_ONLY(WorkerThread_EnterSafeMode()); pthread_cond_wait(&bio_condvar[type],&bio_mutex[type]); + WIN32_ONLY(WorkerThread_ExitSafeMode()); continue; } /* Pop the job from the queue. */ @@ -193,7 +199,9 @@ void *bioProcessBackgroundJobs(void *arg) { /* Lock again before reiterating the loop, if there are no longer * jobs to process we'll block again in pthread_cond_wait(). */ + WIN32_ONLY(WorkerThread_EnterSafeMode()); pthread_mutex_lock(&bio_mutex[type]); + WIN32_ONLY(WorkerThread_ExitSafeMode()); listDelNode(bio_jobs[type],ln); bio_pending[type]--; } diff --git a/src/redis.c b/src/redis.c index 5a29f6c0..c04c1dbd 100644 --- a/src/redis.c +++ b/src/redis.c @@ -30,6 +30,7 @@ #ifdef _WIN32 #include "win32_Interop\win32_util.h" #include "Win32_Interop\Win32_FDAPI.h" +#include "Win32_Interop\Win32_ThreadControl.h" #include #define LOG_LOCAL0 0 #endif @@ -1108,19 +1109,23 @@ int serverCron(struct aeEventLoop *eventLoop, PORT_LONGLONG id, void *clientData /* Check if a background saving or AOF rewrite in progress terminated. */ if (server.rdb_child_pid != -1 || server.aof_child_pid != -1) { #ifdef _WIN32 - if (GetForkOperationStatus() == osCOMPLETE || GetForkOperationStatus() == osFAILED) { - int exitCode; - int bySignal; - bySignal = (int)(GetForkOperationStatus() == osFAILED); - redisLog(REDIS_WARNING, (bySignal ? "fork operation failed" : "fork operation complete")); - EndForkOperation(&exitCode); - if (server.rdb_child_pid != -1) { - backgroundSaveDoneHandler(exitCode, bySignal); - } else { - backgroundRewriteDoneHandler(exitCode, bySignal); - } - updateDictResizePolicy(); - } + if (GetForkOperationStatus() == osCOMPLETE || GetForkOperationStatus() == osFAILED) { + RequestSuspension(); + if (SuspensionCompleted()) { + int exitCode; + int bySignal; + bySignal = (int)(GetForkOperationStatus() == osFAILED); + redisLog(REDIS_WARNING, (bySignal ? "fork operation failed" : "fork operation complete")); + EndForkOperation(&exitCode); + ResumeFromSuspension(); + if (server.rdb_child_pid != -1) { + backgroundSaveDoneHandler(exitCode, bySignal); + } else { + backgroundRewriteDoneHandler(exitCode, bySignal); + } + updateDictResizePolicy(); + } + } #else int statloc; pid_t pid;