#include "lzf.h" /* LZF compression library */
#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
+/* #define REDIS_HELGRIND_FRIENDLY */
+#if defined(__GNUC__) && defined(REDIS_HELGRIND_FRIENDLY)
+#warning "Remember to undef REDIS_HELGRIND_FRIENDLY before to commit"
+#endif
+
/* Error codes */
#define REDIS_OK 0
#define REDIS_ERR -1
#define REDIS_VM_MAX_NEAR_PAGES 65536
#define REDIS_VM_MAX_RANDOM_JUMP 4096
#define REDIS_VM_MAX_THREADS 32
+#define REDIS_THREAD_STACK_SIZE (1024*1024*4)
/* The following is the number of completed I/O jobs to process when the
* handelr is called. 1 is the minimum, and also the default, as it allows
* to block as little as possible other accessing clients. While Virtual
int sort_bypattern;
/* Virtual memory configuration */
int vm_enabled;
+ char *vm_swap_file;
off_t vm_page_size;
off_t vm_pages;
unsigned long long vm_max_memory;
pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
+ pthread_attr_t io_threads_attr; /* attributes for threads creation */
int io_active_threads; /* Number of running I/O threads */
int vm_max_threads; /* Max number of I/O threads running at the same time */
/* Our main thread is blocked on the event loop, locking for sockets ready
static void queueIOJob(iojob *j);
static int vmWriteObjectOnSwap(robj *o, off_t page);
static robj *vmReadObjectFromSwap(off_t page, int type);
-static void waitZeroActiveThreads(void);
+static void waitEmptyIOJobsQueue(void);
+static void vmReopenSwapFile(void);
static void authCommand(redisClient *c);
static void pingCommand(redisClient *c);
now = time(NULL);
strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
- fprintf(fp,"%s %c ",buf,c[level]);
+ fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
vfprintf(fp, fmt, ap);
fprintf(fp,"\n");
fflush(fp);
server.blockedclients = 0;
server.maxmemory = 0;
server.vm_enabled = 0;
+ server.vm_swap_file = zstrdup("/tmp/redis-%p.vm");
server.vm_page_size = 256; /* 256 bytes per page */
server.vm_pages = 1024*1024*100; /* 104 millions of pages */
server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */
goto loaderr;
}
} else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
- server.requirepass = zstrdup(argv[1]);
+ server.requirepass = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
- server.pidfile = zstrdup(argv[1]);
+ server.pidfile = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
- server.dbfilename = zstrdup(argv[1]);
+ server.dbfilename = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"vm-enabled") && argc == 2) {
if ((server.vm_enabled = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
}
+ } else if (!strcasecmp(argv[0],"vm-swap-file") && argc == 2) {
+ zfree(server.vm_swap_file);
+ server.vm_swap_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"vm-max-memory") && argc == 2) {
server.vm_max_memory = strtoll(argv[1], NULL, 10);
} else if (!strcasecmp(argv[0],"vm-page-size") && argc == 2) {
o->ptr = ptr;
o->refcount = 1;
if (server.vm_enabled) {
+ /* Note that this code may run in the context of an I/O thread
+ * and accessing to server.unixtime in theory is an error
+ * (no locks). But in practice this is safe, and even if we read
+ * garbage Redis will not fail, as it's just a statistical info */
o->vm.atime = server.unixtime;
o->storage = REDIS_VM_MEMORY;
}
outlen = sdslen(obj->ptr)-4;
if (outlen <= 0) return 0;
if ((out = zmalloc(outlen+1)) == NULL) return 0;
- printf("Calling LZF with ptr: %p\n", (void*)obj->ptr);
- fflush(stdout);
comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
if (comprlen == 0) {
zfree(out);
int j;
time_t now = time(NULL);
+ /* Wait for I/O therads to terminate, just in case this is a
+ * foreground-saving, to avoid seeking the swap file descriptor at the
+ * same time. */
+ if (server.vm_enabled)
+ waitEmptyIOJobsQueue();
+
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
pid_t childpid;
if (server.bgsavechildpid != -1) return REDIS_ERR;
- if (server.vm_enabled) waitZeroActiveThreads();
+ if (server.vm_enabled) waitEmptyIOJobsQueue();
if ((childpid = fork()) == 0) {
/* Child */
+ if (server.vm_enabled) vmReopenSwapFile();
close(server.fd);
if (rdbSave(filename) == REDIS_OK) {
exit(0);
if (server.appendonly) {
/* Append only file: fsync() the AOF and exit */
fsync(server.appendfd);
+ if (server.vm_enabled) unlink(server.vm_swap_file);
exit(0);
} else {
/* Snapshotting. Perform a SYNC SAVE and exit */
unlink(server.pidfile);
redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
redisLog(REDIS_WARNING,"Server exit now, bye bye...");
+ if (server.vm_enabled) unlink(server.vm_swap_file);
exit(0);
} else {
/* Ooops.. error saving! The best we can do is to continue operating.
);
}
if (server.vm_enabled) {
+ lockThreadedIO();
info = sdscatprintf(info,
"vm_conf_max_memory:%llu\r\n"
"vm_conf_page_size:%llu\r\n"
(unsigned long) listLength(server.io_clients),
(unsigned long) server.io_active_threads
);
+ unlockThreadedIO();
}
for (j = 0; j < server.dbnum; j++) {
long long keys, vkeys;
pid_t childpid;
if (server.bgrewritechildpid != -1) return REDIS_ERR;
- if (server.vm_enabled) waitZeroActiveThreads();
+ if (server.vm_enabled) waitEmptyIOJobsQueue();
if ((childpid = fork()) == 0) {
/* Child */
char tmpfile[256];
- close(server.fd);
+ if (server.vm_enabled) vmReopenSwapFile();
+ close(server.fd);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
exit(0);
*/
/* =================== Virtual Memory - Blocking Side ====================== */
+
+/* substitute the first occurrence of '%p' with the process pid in the
+ * swap file name. */
+static void expandVmSwapFilename(void) {
+ char *p = strstr(server.vm_swap_file,"%p");
+ sds new;
+
+ if (!p) return;
+ new = sdsempty();
+ *p = '\0';
+ new = sdscat(new,server.vm_swap_file);
+ new = sdscatprintf(new,"%ld",(long) getpid());
+ new = sdscat(new,p+2);
+ zfree(server.vm_swap_file);
+ server.vm_swap_file = new;
+}
+
static void vmInit(void) {
off_t totsize;
int pipefds[2];
+ size_t stacksize;
+
+ if (server.vm_max_threads != 0)
+ zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
- server.vm_fp = fopen("/tmp/redisvm","w+b");
+ expandVmSwapFilename();
+ redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file);
+ if ((server.vm_fp = fopen(server.vm_swap_file,"r+b")) == NULL) {
+ server.vm_fp = fopen(server.vm_swap_file,"w+b");
+ }
if (server.vm_fp == NULL) {
- redisLog(REDIS_WARNING,"Impossible to open the swap file. Exiting.");
+ redisLog(REDIS_WARNING,
+ "Impossible to open the swap file: %s. Exiting.",
+ strerror(errno));
exit(1);
}
server.vm_fd = fileno(server.vm_fp);
redisLog(REDIS_VERBOSE,"Allocated %lld bytes page table for %lld pages",
(long long) (server.vm_pages+7)/8, server.vm_pages);
memset(server.vm_bitmap,0,(server.vm_pages+7)/8);
- /* Try to remove the swap file, so the OS will really delete it from the
- * file system when Redis exists. */
- unlink("/tmp/redisvm");
/* Initialize threaded I/O (used by Virtual Memory) */
server.io_newjobs = listCreate();
server.io_ready_pipe_read = pipefds[0];
server.io_ready_pipe_write = pipefds[1];
redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
+ /* LZF requires a lot of stack */
+ pthread_attr_init(&server.io_threads_attr);
+ pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
+ while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
+ pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
/* Listen for events in the threaded I/O pipe */
if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
vmThreadedIOCompletedJob, NULL) == AE_ERR)
de = dictGetRandomKey(db->dict);
key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
- if (key->storage != REDIS_VM_MEMORY) {
+ /* Only swap objects that are currently in memory.
+ *
+ * Also don't swap shared objects if threaded VM is on, as we
+ * try to ensure that the main thread does not touch the
+ * object while the I/O thread is using it, but we can't
+ * control other keys without adding additional mutex. */
+ if (key->storage != REDIS_VM_MEMORY ||
+ (server.vm_max_threads != 0 && val->refcount != 1)) {
if (maxtries) i--; /* don't count this try */
continue;
}
struct dictEntry *de;
redisLog(REDIS_DEBUG,"Processing I/O completed job");
- assert(listLength(server.io_processed) != 0);
/* Get the processed element (the oldest one) */
lockThreadedIO();
+ assert(listLength(server.io_processed) != 0);
ln = listFirst(server.io_processed);
j = ln->value;
listDelNode(server.io_processed,ln);
/* Now we know the amount of pages required to swap this object.
* Let's find some space for it, and queue this task again
* rebranded as REDIS_IOJOB_DO_SWAP. */
- if (vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR) {
- /* Ooops... no space! */
+ if (!vmCanSwapOut() ||
+ vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR)
+ {
+ /* Ooops... no space or we can't swap as there is
+ * a fork()ed Redis trying to save stuff on disk. */
freeIOJob(j);
+ key->storage = REDIS_VM_MEMORY; /* undo operation */
} else {
/* Note that we need to mark this pages as used now,
* if the job will be canceled, we'll mark them as freed
freeIOJob(j);
/* Put a few more swap requests in queue if we are still
* out of memory */
- if (zmalloc_used_memory() > server.vm_max_memory) {
+ if (vmCanSwapOut() && zmalloc_used_memory() > server.vm_max_memory){
int more = 1;
while(more) {
lockThreadedIO();
int i;
assert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING);
+again:
lockThreadedIO();
/* Search for a matching key in one of the queues */
for (i = 0; i < 3; i++) {
if (job->canceled) continue; /* Skip this, already canceled. */
if (compareStringObjects(job->key,o) == 0) {
- redisLog(REDIS_DEBUG,"*** CANCELED %p (%s)\n",
- (void*)job, (char*)o->ptr);
+ redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (LIST ID %d)\n",
+ (void*)job, (char*)o->ptr, i);
/* Mark the pages as free since the swap didn't happened
* or happened but is now discarded. */
if (job->type == REDIS_IOJOB_DO_SWAP)
listDelNode(lists[i],ln);
break;
case 1: /* io_processing */
+ /* Oh Shi- the thread is messing with the Job, and
+ * probably with the object if this is a
+ * PREPARE_SWAP or DO_SWAP job. Better to wait for the
+ * job to move into the next queue... */
+ if (job->type != REDIS_IOJOB_LOAD) {
+ /* Yes, we try again and again until the job
+ * is completed. */
+ unlockThreadedIO();
+ /* But let's wait some time for the I/O thread
+ * to finish with this job. After all this condition
+ * should be very rare. */
+ usleep(1);
+ goto again;
+ } else {
+ job->canceled = 1;
+ break;
+ }
case 2: /* io_processed */
+ /* The job was already processed, that's easy...
+ * just mark it as canceled so that we'll ignore it
+ * when processing completed jobs. */
job->canceled = 1;
break;
}
/* Get a new job to process */
lockThreadedIO();
if (listLength(server.io_newjobs) == 0) {
+#ifdef REDIS_HELGRIND_FRIENDLY
+ /* No new jobs? Wait and retry, because to be Helgrind
+ * (valgrind --tool=helgrind) what's needed is to take
+ * the same threads running instead to create/destroy threads
+ * as needed (otherwise valgrind will fail) */
+ unlockThreadedIO();
+ usleep(1); /* Give some time for the I/O thread to work. */
+ continue;
+#endif
/* No new jobs in queue, exit. */
redisLog(REDIS_DEBUG,"Thread %lld exiting, nothing to do",
(long long) pthread_self());
static void spawnIOThread(void) {
pthread_t thread;
- pthread_create(&thread,NULL,IOThreadEntryPoint,NULL);
+ pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL);
server.io_active_threads++;
}
/* We need to wait for the last thread to exit before we are able to
* fork() in order to BGSAVE or BGREWRITEAOF. */
-static void waitZeroActiveThreads(void) {
+static void waitEmptyIOJobsQueue(void) {
while(1) {
lockThreadedIO();
- if (server.io_active_threads == 0) {
+ if (listLength(server.io_newjobs) == 0 &&
+ listLength(server.io_processing) == 0 &&
+ server.io_active_threads == 0)
+ {
unlockThreadedIO();
return;
}
}
}
+static void vmReopenSwapFile(void) {
+ fclose(server.vm_fp);
+ server.vm_fp = fopen(server.vm_swap_file,"r+b");
+ if (server.vm_fp == NULL) {
+ redisLog(REDIS_WARNING,"Can't re-open the VM swap file: %s. Exiting.",
+ server.vm_swap_file);
+ exit(1);
+ }
+ server.vm_fd = fileno(server.vm_fp);
+}
+
/* This function must be called while with threaded IO locked */
static void queueIOJob(iojob *j) {
redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
"+Key at:%p refcount:%d, value at:%p refcount:%d "
"encoding:%d serializedlength:%lld\r\n",
(void*)key, key->refcount, (void*)val, val->refcount,
- val->encoding, rdbSavedObjectLen(val,NULL)));
+ val->encoding, (long long) rdbSavedObjectLen(val,NULL)));
} else {
addReplySds(c,sdscatprintf(sdsempty(),
"+Key at:%p refcount:%d, value swapped at: page %llu "
projects / redis.git / blobdiff