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;