Make bio.c threads killable ASAP if needed.

We use this new bio.c feature in order to stop our I/O threads if there
is a memory test to do on crash. In this case we don't want anything
else than the main thread to run, otherwise the other threads may mess
with the heap and the memory test will report a false positive.

diff --git a/src/bio.c b/src/bio.c
index f3ddd67255337ed722f71a8c048bff248f155f38..0fec24d0bd091522b21e252782cd050ac4ac6165 100644 (file)
--- a/src/bio.c
+++ b/src/bio.c
@@ -61,6 +61,7 @@
 #include "redis.h"
 #include "bio.h"
 
 #include "redis.h"
 #include "bio.h"
 

+static pthread_t bio_threads[REDIS_BIO_NUM_OPS];

 static pthread_mutex_t bio_mutex[REDIS_BIO_NUM_OPS];
 static pthread_cond_t bio_condvar[REDIS_BIO_NUM_OPS];
 static list *bio_jobs[REDIS_BIO_NUM_OPS];
 static pthread_mutex_t bio_mutex[REDIS_BIO_NUM_OPS];
 static pthread_cond_t bio_condvar[REDIS_BIO_NUM_OPS];
 static list *bio_jobs[REDIS_BIO_NUM_OPS];


@@ -118,6 +119,7 @@ void bioInit(void) {
             redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");
             exit(1);
         }
             redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");
             exit(1);
         }

+        bio_threads[j] = thread;

     }
 }
 
     }
 }
 


@@ -140,7 +142,11 @@ void *bioProcessBackgroundJobs(void *arg) {
     unsigned long type = (unsigned long) arg;
     sigset_t sigset;
 
     unsigned long type = (unsigned long) arg;
     sigset_t sigset;
 

-    pthread_detach(pthread_self());
+    /* Make the thread killable at any time, so that bioKillThreads()
+     * can work reliably. */
+    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
+    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
+

     pthread_mutex_lock(&bio_mutex[type]);
     /* Block SIGALRM so we are sure that only the main thread will
      * receive the watchdog signal. */
     pthread_mutex_lock(&bio_mutex[type]);
     /* Block SIGALRM so we are sure that only the main thread will
      * receive the watchdog signal. */


@@ -192,56 +198,23 @@ unsigned long long bioPendingJobsOfType(int type) {
     return val;
 }
 
     return val;
 }
 

-#if 0 /* We don't use the following code for now, and bioWaitPendingJobsLE
-         probably needs a rewrite using conditional variables instead of the
-         current implementation. */
-         
-
-/* Wait until the number of pending jobs of the specified type are
- * less or equal to the specified number.
- *
- * This function may block for long time, it should only be used to perform
- * the following tasks:
- *
- * 1) To avoid that the main thread is pushing jobs of a given time so fast
- *    that the background thread can't process them at the same speed.
- *    So before creating a new job of a given type the main thread should
- *    call something like: bioWaitPendingJobsLE(job_type,10000);
- * 2) In order to perform special operations that make it necessary to be sure
- *    no one is touching shared resourced in the background.
- */
-void bioWaitPendingJobsLE(int type, unsigned long long num) {
-    unsigned long long iteration = 0;
+/* Kill the running bio threads in an unclean way. This function should be
+ * used only when it's critical to stop the threads for some reason.
+ * Currently Redis does this only on crash (for instance on SIGSEGV) in order
+ * to perform a fast memory check without other threads messing with memory. */
+void bioKillThreads(void) {
+    int err, j;

 
 

-    /* We poll the jobs queue aggressively to start, and gradually relax
-     * the polling speed if it is going to take too much time. */
-    while(1) {
-        iteration++;
-        if (iteration > 1000 && iteration <= 10000) {
-            usleep(100);
-        } else if (iteration > 10000) {
-            usleep(1000);
+    for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {
+        if (pthread_cancel(bio_threads[j]) == 0) {
+            if ((err = pthread_join(bio_threads[j],NULL)) != 0) {
+                redisLog(REDIS_WARNING,
+                    "Bio thread for job type #%d can be joined: %s",
+                        j, strerror(err));
+            } else {
+                redisLog(REDIS_WARNING,
+                    "Bio thread for job type #%d terminated",j);
+            }

         }
         }

-        if (bioPendingJobsOfType(type) <= num) break;
-    }
-}
-
-/* Return the older job of the specified type. */
-time_t bioOlderJobOfType(int type) {
-    time_t time;
-    listNode *ln;
-    struct bio_job *job;
-
-    pthread_mutex_lock(&bio_mutex[type]);
-    ln = listFirst(bio_jobs[type]);
-    if (ln == NULL) {
-        pthread_mutex_unlock(&bio_mutex[type]);
-        return 0;

     }
     }

-    job = ln->value;
-    time = job->time;
-    pthread_mutex_unlock(&bio_mutex[type]);
-    return time;

 }
 }

-
-#endif

diff --git a/src/bio.h b/src/bio.h
index 7c08a9b2959bce730d5318b15f272328b65490c9..85f03ad1af3af8290418322dbb2a36a56538f81f 100644 (file)
--- a/src/bio.h
+++ b/src/bio.h
@@ -33,6 +33,7 @@ void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3);
 unsigned long long bioPendingJobsOfType(int type);
 void bioWaitPendingJobsLE(int type, unsigned long long num);
 time_t bioOlderJobOfType(int type);
 unsigned long long bioPendingJobsOfType(int type);
 void bioWaitPendingJobsLE(int type, unsigned long long num);
 time_t bioOlderJobOfType(int type);

+void bioKillThreads(void);

 
 /* Background job opcodes */
 #define REDIS_BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */
 
 /* Background job opcodes */
 #define REDIS_BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */

diff --git a/src/debug.c b/src/debug.c
index 003d2bd408f41cc7ed2c2f749edaaea4fe21f444..d4ee1cb47103e586e260e949b92e005d1b0e184a 100644 (file)
--- a/src/debug.c
+++ b/src/debug.c
@@ -746,6 +746,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
 #if defined(HAVE_PROC_MAPS)
     /* Test memory */
     redisLog(REDIS_WARNING, "--- FAST MEMORY TEST");
 #if defined(HAVE_PROC_MAPS)
     /* Test memory */
     redisLog(REDIS_WARNING, "--- FAST MEMORY TEST");

+    bioKillThreads();

     if (memtest_test_linux_anonymous_maps()) {
         redisLog(REDIS_WARNING,
             "!!! MEMORY ERROR DETECTED! Check your memory ASAP !!!");
     if (memtest_test_linux_anonymous_maps()) {
         redisLog(REDIS_WARNING,
             "!!! MEMORY ERROR DETECTED! Check your memory ASAP !!!");