*
* Currently there is no way for the creator of the job to be notified about
* the completion of the operation, this will only be added when/if needed.
+ *
+ * ----------------------------------------------------------------------------
+ *
+ * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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.
+ * * Neither the name of Redis nor the names of its contributors may be used
+ * to endorse or promote products derived from this software without
+ * specific prior written permission.
+ *
+ * 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 OWNER 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 "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];
redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");
exit(1);
}
+ bio_threads[j] = thread;
}
}
void *bioProcessBackgroundJobs(void *arg) {
struct bio_job *job;
unsigned long type = (unsigned long) arg;
+ sigset_t sigset;
+
+ /* 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_detach(pthread_self());
pthread_mutex_lock(&bio_mutex[type]);
+ /* Block SIGALRM so we are sure that only the main thread will
+ * receive the watchdog signal. */
+ sigemptyset(&sigset);
+ sigaddset(&sigset, SIGALRM);
+ if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))
+ redisLog(REDIS_WARNING,
+ "Warning: can't mask SIGALRM in bio.c thread: %s", strerror(errno));
+
while(1) {
listNode *ln;
/* Process the job accordingly to its type. */
if (type == REDIS_BIO_CLOSE_FILE) {
close((long)job->arg1);
+ } else if (type == REDIS_BIO_AOF_FSYNC) {
+ aof_fsync((long)job->arg1);
} else {
redisPanic("Wrong job type in bioProcessBackgroundJobs().");
}
return val;
}
-/* 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;
-}
-
projects / redis.git / blobdiff