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

@@ -32,6 +32,7 @@
     <ClCompile Include="win32_rfdmap.cpp" />
     <ClCompile Include="Win32_service.cpp" />
     <ClCompile Include="Win32_StackTrace.cpp" />
+    <ClCompile Include="Win32_ThreadControl.c" />
     <ClCompile Include="win32_util.c" />
     <ClCompile Include="Win32_variadicFunctor.cpp" />
     <ClCompile Include="win32_wsiocp.c" />
@@ -51,6 +52,7 @@
     <ClInclude Include="Win32_Service.h" />
     <ClInclude Include="Win32_SmartHandle.h" />
     <ClInclude Include="Win32_StackTrace.h" />
+    <ClInclude Include="Win32_ThreadControl.h" />
     <ClInclude Include="win32_types.h" />
     <ClInclude Include="win32_util.h" />
     <ClInclude Include="Win32_variadicFunctor.h" />

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 <vector>
 #include <map>
@@ -1321,6 +1322,7 @@ extern "C"
             ParseCommandLineArguments(argc, argv);
             SetupLogging();
             StackTraceInit();
+            InitThreadControl();
         } catch (system_error syserr) {
             exit(-1);
         } catch (runtime_error runerr) {

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 <Windows.h>
+#include <assert.h>
+
+
+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;
+        }
+    }
+}
+

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

src/Win32_Interop/win32fixes.c

@@ -20,6 +20,7 @@
 #include <string.h>
 #include <assert.h>
 //#include <io.h>
+#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;

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]--;
     }

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 <locale.h>
 #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;