]> git.saurik.com Git - redis.git/blame - src/dscache.c
DB API refactoring. The changes were designed together with Pieter Noordhuis.
[redis.git] / src / dscache.c
CommitLineData
e2641e09 1#include "redis.h"
2
3#include <fcntl.h>
4#include <pthread.h>
5#include <math.h>
6#include <signal.h>
7
33388d43 8/* dscache.c - Disk store cache for disk store backend.
9 *
10 * When Redis is configured for using disk as backend instead of memory, the
11 * memory is used as a cache, so that recently accessed keys are taken in
12 * memory for fast read and write operations.
13 *
14 * Modified keys are marked to be flushed on disk, and will be flushed
15 * as long as the maxium configured flush time elapsed.
16 *
17 * This file implements the whole caching subsystem and contains further
18 * documentation. */
19
20/* TODO:
133cf28e 21 *
22 * WARNING: most of the following todo items and design issues are no
23 * longer relevant with the new design. Here as a checklist to see if
24 * some old ideas still apply.
33388d43 25 *
16d77878 26 * - What happens when an object is destroyed?
27 *
d158dc28 28 * If the object is destroyed since semantically it was deleted or
29 * replaced with something new, we don't care if there was a SAVE
30 * job pending for it. Anyway when the IO JOb will be created we'll get
31 * the pointer of the current value.
16d77878 32 *
d158dc28 33 * If the object is already a REDIS_IO_SAVEINPROG object, then it is
34 * impossible that we get a decrRefCount() that will reach refcount of zero
35 * since the object is both in the dataset and in the io job entry.
16d77878 36 *
37 * - What happens with MULTI/EXEC?
38 *
d158dc28 39 * Good question. Without some kind of versioning with a global counter
40 * it is not possible to have trasactions on disk, but they are still
41 * useful since from the point of view of memory and client bugs it is
42 * a protection anyway. Also it's useful for WATCH.
4ab98823 43 *
d158dc28 44 * Btw there is to check what happens when WATCH gets combined to keys
45 * that gets removed from the object cache. Should be save but better
46 * to check.
98a9abb6 47 *
d158dc28 48 * - Check if/why INCR will not update the LRU info for the object.
8e6bb671 49 *
50 * - Fix/Check the following race condition: a key gets a DEL so there is
51 * a write operation scheduled against this key. Later the same key will
52 * be the argument of a GET, but the write operation was still not
53 * completed (to delete the file). If the GET will be for some reason
54 * a blocking loading (via lookup) we can load the old value on memory.
55 *
56 * This problems can be fixed with negative caching. We can use it
57 * to optimize the system, but also when a key is deleted we mark
58 * it as non existing on disk as well (in a way that this cache
59 * entry can't be evicted, setting time to 0), then we avoid looking at
60 * the disk at all if the key can't be there. When an IO Job complete
61 * a deletion, we set the time of the negative caching to a non zero
62 * value so it will be evicted later.
63 *
64 * Are there other patterns like this where we load stale data?
d934e1e8 65 *
66 * Also, make sure that key preloading is ONLY done for keys that are
67 * not marked as cacheKeyDoesNotExist(), otherwise, again, we can load
68 * data from disk that should instead be deleted.
aa81e4d5 69 *
d158dc28 70 * - dsSet() should use rename(2) in order to avoid corruptions.
4942145d 71 *
72 * - Don't add a LOAD if there is already a LOADINPROGRESS, or is this
73 * impossible since anyway the io_keys stuff will work as lock?
d158dc28 74 *
75 * - Serialize special encoded things in a raw form.
fad97fbe 76 *
77 * - When putting IO read operations on top of the queue, do this only if
78 * the already-on-top operation is not a save or if it is a save that
79 * is scheduled for later execution. If there is a save that is ready to
80 * fire, let's insert the load operation just before the first save that
81 * is scheduled for later exection for instance.
82 *
83 * - Support MULTI/EXEC transactions via a journal file, that is played on
84 * startup to check if there is cleanup to do. This way we can implement
85 * transactions with our simple file based KV store.
33388d43 86 */
87
e2641e09 88/* Virtual Memory is composed mainly of two subsystems:
89 * - Blocking Virutal Memory
90 * - Threaded Virtual Memory I/O
91 * The two parts are not fully decoupled, but functions are split among two
92 * different sections of the source code (delimited by comments) in order to
93 * make more clear what functionality is about the blocking VM and what about
94 * the threaded (not blocking) VM.
95 *
96 * Redis VM design:
97 *
98 * Redis VM is a blocking VM (one that blocks reading swapped values from
99 * disk into memory when a value swapped out is needed in memory) that is made
100 * unblocking by trying to examine the command argument vector in order to
101 * load in background values that will likely be needed in order to exec
102 * the command. The command is executed only once all the relevant keys
103 * are loaded into memory.
104 *
105 * This basically is almost as simple of a blocking VM, but almost as parallel
106 * as a fully non-blocking VM.
107 */
108
f34a6cd8 109void spawnIOThread(void);
f771dc23 110int cacheScheduleIOPushJobs(int flags);
111int processActiveIOJobs(int max);
f34a6cd8 112
e2641e09 113/* =================== Virtual Memory - Blocking Side ====================== */
114
f2da3a62 115void dsInit(void) {
e2641e09 116 int pipefds[2];
117 size_t stacksize;
e2641e09 118
f2da3a62 119 zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
e2641e09 120
67b0b41c 121 redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);
f2da3a62 122 /* Open Disk Store */
123 if (dsOpen() != REDIS_OK) {
124 redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
e2641e09 125 exit(1);
f2da3a62 126 };
e2641e09 127
f2da3a62 128 /* Initialize threaded I/O for Object Cache */
e2641e09 129 server.io_newjobs = listCreate();
130 server.io_processing = listCreate();
131 server.io_processed = listCreate();
132 server.io_ready_clients = listCreate();
133 pthread_mutex_init(&server.io_mutex,NULL);
98a9abb6 134 pthread_cond_init(&server.io_condvar,NULL);
36c17a53 135 pthread_mutex_init(&server.bgsavethread_mutex,NULL);
e2641e09 136 server.io_active_threads = 0;
137 if (pipe(pipefds) == -1) {
f2da3a62 138 redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
e2641e09 139 ,strerror(errno));
140 exit(1);
141 }
142 server.io_ready_pipe_read = pipefds[0];
143 server.io_ready_pipe_write = pipefds[1];
144 redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
145 /* LZF requires a lot of stack */
146 pthread_attr_init(&server.io_threads_attr);
147 pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
556bdfba 148
149 /* Solaris may report a stacksize of 0, let's set it to 1 otherwise
150 * multiplying it by 2 in the while loop later will not really help ;) */
151 if (!stacksize) stacksize = 1;
152
e2641e09 153 while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
154 pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
155 /* Listen for events in the threaded I/O pipe */
156 if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
157 vmThreadedIOCompletedJob, NULL) == AE_ERR)
158 oom("creating file event");
e2641e09 159
f2da3a62 160 /* Spawn our I/O thread */
161 spawnIOThread();
e2641e09 162}
163
f2da3a62 164/* Compute how good candidate the specified object is for eviction.
165 * An higher number means a better candidate. */
e2641e09 166double computeObjectSwappability(robj *o) {
167 /* actual age can be >= minage, but not < minage. As we use wrapping
168 * 21 bit clocks with minutes resolution for the LRU. */
f081eaf1 169 return (double) estimateObjectIdleTime(o);
e2641e09 170}
171
f2da3a62 172/* Try to free one entry from the diskstore object cache */
173int cacheFreeOneEntry(void) {
e2641e09 174 int j, i;
175 struct dictEntry *best = NULL;
176 double best_swappability = 0;
177 redisDb *best_db = NULL;
178 robj *val;
179 sds key;
180
181 for (j = 0; j < server.dbnum; j++) {
182 redisDb *db = server.db+j;
183 /* Why maxtries is set to 100?
184 * Because this way (usually) we'll find 1 object even if just 1% - 2%
185 * are swappable objects */
186 int maxtries = 100;
187
69bfffb4 188 for (i = 0; i < 5 && dictSize(db->dict); i++) {
e2641e09 189 dictEntry *de;
190 double swappability;
3be00d7e 191 robj keyobj;
192 sds keystr;
e2641e09 193
194 if (maxtries) maxtries--;
195 de = dictGetRandomKey(db->dict);
3be00d7e 196 keystr = dictGetEntryKey(de);
e2641e09 197 val = dictGetEntryVal(de);
3be00d7e 198 initStaticStringObject(keyobj,keystr);
199
200 /* Don't remove objects that are currently target of a
201 * read or write operation. */
202 if (cacheScheduleIOGetFlags(db,&keyobj) != 0) {
e2641e09 203 if (maxtries) i--; /* don't count this try */
204 continue;
205 }
206 swappability = computeObjectSwappability(val);
207 if (!best || swappability > best_swappability) {
208 best = de;
209 best_swappability = swappability;
210 best_db = db;
211 }
212 }
213 }
f2da3a62 214 if (best == NULL) {
c5b6f461 215 /* Not able to free a single object? we should check if our
418d5eaf 216 * IO queues have stuff in queue, and try to consume the queue
217 * otherwise we'll use an infinite amount of memory if changes to
218 * the dataset are faster than I/O */
219 if (listLength(server.cache_io_queue) > 0) {
f771dc23 220 redisLog(REDIS_DEBUG,"--- Busy waiting IO to reclaim memory");
221 cacheScheduleIOPushJobs(REDIS_IO_ASAP);
222 processActiveIOJobs(1);
418d5eaf 223 return REDIS_OK;
224 }
225 /* Nothing to free at all... */
f2da3a62 226 return REDIS_ERR;
227 }
e2641e09 228 key = dictGetEntryKey(best);
229 val = dictGetEntryVal(best);
230
f2da3a62 231 redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
e2641e09 232 key, best_swappability);
233
f2da3a62 234 /* Delete this key from memory */
235 {
236 robj *kobj = createStringObject(key,sdslen(key));
237 dbDelete(best_db,kobj);
238 decrRefCount(kobj);
e2641e09 239 }
5ef64098 240 return REDIS_OK;
e2641e09 241}
242
d934e1e8 243/* ==================== Disk store negative caching ========================
244 *
245 * When disk store is enabled, we need negative caching, that is, to remember
246 * keys that are for sure *not* on the disk key-value store.
247 *
3be00d7e 248 * This is usefuls because without negative caching cache misses will cost us
249 * a disk lookup, even if the same non existing key is accessed again and again.
d934e1e8 250 *
3be00d7e 251 * With negative caching we remember that the key is not on disk, so if it's
252 * not in memory and we have a negative cache entry, we don't try a disk
253 * access at all.
254 */
d934e1e8 255
3be00d7e 256/* Returns true if the specified key may exists on disk, that is, we don't
257 * have an entry in our negative cache for this key */
d934e1e8 258int cacheKeyMayExist(redisDb *db, robj *key) {
259 return dictFind(db->io_negcache,key) == NULL;
260}
261
3be00d7e 262/* Set the specified key as an entry that may possibily exist on disk, that is,
263 * remove the negative cache entry for this key if any. */
d934e1e8 264void cacheSetKeyMayExist(redisDb *db, robj *key) {
265 dictDelete(db->io_negcache,key);
266}
267
3be00d7e 268/* Set the specified key as non existing on disk, that is, create a negative
269 * cache entry for this key. */
d934e1e8 270void cacheSetKeyDoesNotExist(redisDb *db, robj *key) {
d934e1e8 271 if (dictReplace(db->io_negcache,key,(void*)time(NULL))) {
272 incrRefCount(key);
273 }
274}
275
c15a3887 276/* Remove one entry from negative cache using approximated LRU. */
277int negativeCacheEvictOneEntry(void) {
278 struct dictEntry *de;
279 robj *best = NULL;
280 redisDb *best_db = NULL;
281 time_t time, best_time = 0;
282 int j;
283
284 for (j = 0; j < server.dbnum; j++) {
285 redisDb *db = server.db+j;
286 int i;
287
288 if (dictSize(db->io_negcache) == 0) continue;
289 for (i = 0; i < 3; i++) {
290 de = dictGetRandomKey(db->io_negcache);
291 time = (time_t) dictGetEntryVal(de);
292
293 if (best == NULL || time < best_time) {
294 best = dictGetEntryKey(de);
295 best_db = db;
296 best_time = time;
297 }
298 }
299 }
300 if (best) {
301 dictDelete(best_db->io_negcache,best);
302 return REDIS_OK;
303 } else {
304 return REDIS_ERR;
305 }
306}
307
d934e1e8 308/* ================== Disk store cache - Threaded I/O ====================== */
e2641e09 309
310void freeIOJob(iojob *j) {
e2641e09 311 decrRefCount(j->key);
5ef64098 312 /* j->val can be NULL if the job is about deleting the key from disk. */
313 if (j->val) decrRefCount(j->val);
e2641e09 314 zfree(j);
315}
316
317/* Every time a thread finished a Job, it writes a byte into the write side
318 * of an unix pipe in order to "awake" the main thread, and this function
419e1cca 319 * is called.
320 *
d9fac6c0 321 * If privdata == NULL the function will try to put more jobs in the queue
322 * of IO jobs to process as more room is made. privdata is equal to NULL
323 * when the function is called from the event loop, so we want to push
324 * more IO jobs in the queue. Instead when the function is called by
325 * other functions that want to create a write-barrier to avoid race
326 * conditions we don't push new jobs in the queue. */
e2641e09 327void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
328 int mask)
329{
330 char buf[1];
f34a6cd8 331 int retval, processed = 0, toprocess = -1;
e2641e09 332 REDIS_NOTUSED(el);
333 REDIS_NOTUSED(mask);
e2641e09 334
335 /* For every byte we read in the read side of the pipe, there is one
336 * I/O job completed to process. */
337 while((retval = read(fd,buf,1)) == 1) {
338 iojob *j;
339 listNode *ln;
e2641e09 340
341 redisLog(REDIS_DEBUG,"Processing I/O completed job");
342
343 /* Get the processed element (the oldest one) */
344 lockThreadedIO();
345 redisAssert(listLength(server.io_processed) != 0);
346 if (toprocess == -1) {
347 toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
348 if (toprocess <= 0) toprocess = 1;
349 }
350 ln = listFirst(server.io_processed);
351 j = ln->value;
352 listDelNode(server.io_processed,ln);
353 unlockThreadedIO();
f34a6cd8 354
e2641e09 355 /* Post process it in the main thread, as there are things we
356 * can do just here to avoid race conditions and/or invasive locks */
5ef64098 357 redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
358 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
359 (unsigned char*)j->key->ptr);
e2641e09 360 if (j->type == REDIS_IOJOB_LOAD) {
5ef64098 361 /* Create the key-value pair in the in-memory database */
4ab98823 362 if (j->val != NULL) {
3be00d7e 363 /* Note: it's possible that the key is already in memory
364 * due to a blocking load operation. */
f85cd526 365 if (dictFind(j->db->dict,j->key->ptr) == NULL) {
366 dbAdd(j->db,j->key,j->val);
ad01a255 367 incrRefCount(j->val);
368 if (j->expire != -1) setExpire(j->db,j->key,j->expire);
369 }
5d46e370 370 } else {
371 /* Key not found on disk. If it is also not in memory
372 * as a cached object, nor there is a job writing it
373 * in background, we are sure the key does not exist
374 * currently.
375 *
376 * So we set a negative cache entry avoiding that the
377 * resumed client will block load what does not exist... */
bafa88c8 378 if (dictFind(j->db->dict,j->key->ptr) == NULL &&
5d46e370 379 (cacheScheduleIOGetFlags(j->db,j->key) &
380 (REDIS_IO_SAVE|REDIS_IO_SAVEINPROG)) == 0)
381 {
382 cacheSetKeyDoesNotExist(j->db,j->key);
383 }
4ab98823 384 }
3be00d7e 385 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_LOADINPROG);
5ef64098 386 handleClientsBlockedOnSwappedKey(j->db,j->key);
5f6e1183 387 } else if (j->type == REDIS_IOJOB_SAVE) {
3be00d7e 388 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG);
e2641e09 389 }
3a21cb99 390 freeIOJob(j);
e2641e09 391 processed++;
d9fac6c0 392 if (privdata == NULL) cacheScheduleIOPushJobs(0);
e2641e09 393 if (processed == toprocess) return;
394 }
395 if (retval < 0 && errno != EAGAIN) {
396 redisLog(REDIS_WARNING,
397 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
398 strerror(errno));
399 }
400}
401
402void lockThreadedIO(void) {
403 pthread_mutex_lock(&server.io_mutex);
404}
405
406void unlockThreadedIO(void) {
407 pthread_mutex_unlock(&server.io_mutex);
408}
409
e2641e09 410void *IOThreadEntryPoint(void *arg) {
411 iojob *j;
412 listNode *ln;
413 REDIS_NOTUSED(arg);
419e1cca 414 long long start;
e2641e09 415
416 pthread_detach(pthread_self());
98a9abb6 417 lockThreadedIO();
e2641e09 418 while(1) {
419 /* Get a new job to process */
e2641e09 420 if (listLength(server.io_newjobs) == 0) {
a440ecf0 421 /* Wait for more work to do */
05600eb8 422 redisLog(REDIS_DEBUG,"[T] wait for signal");
a440ecf0 423 pthread_cond_wait(&server.io_condvar,&server.io_mutex);
05600eb8 424 redisLog(REDIS_DEBUG,"[T] signal received");
1609a1c4 425 continue;
e2641e09 426 }
419e1cca 427 start = ustime();
05600eb8 428 redisLog(REDIS_DEBUG,"[T] %ld IO jobs to process",
c4b64a13 429 listLength(server.io_newjobs));
e2641e09 430 ln = listFirst(server.io_newjobs);
431 j = ln->value;
432 listDelNode(server.io_newjobs,ln);
433 /* Add the job in the processing queue */
e2641e09 434 listAddNodeTail(server.io_processing,j);
435 ln = listLast(server.io_processing); /* We use ln later to remove it */
436 unlockThreadedIO();
98a9abb6 437
05600eb8 438 redisLog(REDIS_DEBUG,"[T] %ld: new job type %s: %p about key '%s'",
5ef64098 439 (long) pthread_self(),
440 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
441 (void*)j, (char*)j->key->ptr);
e2641e09 442
443 /* Process the Job */
444 if (j->type == REDIS_IOJOB_LOAD) {
4ab98823 445 time_t expire;
446
447 j->val = dsGet(j->db,j->key,&expire);
448 if (j->val) j->expire = expire;
5ef64098 449 } else if (j->type == REDIS_IOJOB_SAVE) {
31222292 450 if (j->val) {
05600eb8 451 dsSet(j->db,j->key,j->val,j->expire);
31222292 452 } else {
5ef64098 453 dsDel(j->db,j->key);
31222292 454 }
e2641e09 455 }
456
457 /* Done: insert the job into the processed queue */
05600eb8 458 redisLog(REDIS_DEBUG,"[T] %ld completed the job: %p (key %s)",
e2641e09 459 (long) pthread_self(), (void*)j, (char*)j->key->ptr);
98a9abb6 460
05600eb8 461 redisLog(REDIS_DEBUG,"[T] lock IO");
e2641e09 462 lockThreadedIO();
05600eb8 463 redisLog(REDIS_DEBUG,"[T] IO locked");
e2641e09 464 listDelNode(server.io_processing,ln);
465 listAddNodeTail(server.io_processed,j);
e2641e09 466
467 /* Signal the main thread there is new stuff to process */
468 redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
8b108ed3 469 redisLog(REDIS_DEBUG,"TIME (%c): %lld\n", j->type == REDIS_IOJOB_LOAD ? 'L' : 'S', ustime()-start);
e2641e09 470 }
98a9abb6 471 /* never reached, but that's the full pattern... */
472 unlockThreadedIO();
473 return NULL;
e2641e09 474}
475
476void spawnIOThread(void) {
477 pthread_t thread;
478 sigset_t mask, omask;
479 int err;
480
481 sigemptyset(&mask);
482 sigaddset(&mask,SIGCHLD);
483 sigaddset(&mask,SIGHUP);
484 sigaddset(&mask,SIGPIPE);
485 pthread_sigmask(SIG_SETMASK, &mask, &omask);
486 while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
487 redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
488 strerror(err));
489 usleep(1000000);
490 }
491 pthread_sigmask(SIG_SETMASK, &omask, NULL);
492 server.io_active_threads++;
493}
494
f771dc23 495/* Wait that up to 'max' pending IO Jobs are processed by the I/O thread.
496 * From our point of view an IO job processed means that the count of
497 * server.io_processed must increase by one.
498 *
499 * If max is -1, all the pending IO jobs will be processed.
500 *
501 * Returns the number of IO jobs processed.
502 *
503 * NOTE: while this may appear like a busy loop, we are actually blocked
504 * by IO since we continuously acquire/release the IO lock. */
505int processActiveIOJobs(int max) {
506 int processed = 0;
507
508 while(max == -1 || max > 0) {
e2641e09 509 int io_processed_len;
510
05600eb8 511 redisLog(REDIS_DEBUG,"[P] lock IO");
e2641e09 512 lockThreadedIO();
05600eb8 513 redisLog(REDIS_DEBUG,"Waiting IO jobs processing: new:%d proessing:%d processed:%d",listLength(server.io_newjobs),listLength(server.io_processing),listLength(server.io_processed));
514
e2641e09 515 if (listLength(server.io_newjobs) == 0 &&
8d51fb6a 516 listLength(server.io_processing) == 0)
e2641e09 517 {
f771dc23 518 /* There is nothing more to process */
05600eb8 519 redisLog(REDIS_DEBUG,"[P] Nothing to process, unlock IO, return");
e2641e09 520 unlockThreadedIO();
f771dc23 521 break;
e2641e09 522 }
f771dc23 523
05600eb8 524#if 1
a440ecf0 525 /* If there are new jobs we need to signal the thread to
249ad25f 526 * process the next one. FIXME: drop this if useless. */
05600eb8 527 redisLog(REDIS_DEBUG,"[P] waitEmptyIOJobsQueue: new %d, processing %d, processed %d",
a440ecf0 528 listLength(server.io_newjobs),
05600eb8 529 listLength(server.io_processing),
530 listLength(server.io_processed));
5d46e370 531
a440ecf0 532 if (listLength(server.io_newjobs)) {
05600eb8 533 redisLog(REDIS_DEBUG,"[P] There are new jobs, signal");
a440ecf0 534 pthread_cond_signal(&server.io_condvar);
535 }
f771dc23 536#endif
537
538 /* Check if we can process some finished job */
e2641e09 539 io_processed_len = listLength(server.io_processed);
05600eb8 540 redisLog(REDIS_DEBUG,"[P] Unblock IO");
e2641e09 541 unlockThreadedIO();
05600eb8 542 redisLog(REDIS_DEBUG,"[P] Wait");
543 usleep(10000);
e2641e09 544 if (io_processed_len) {
c1ae36ae 545 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
546 (void*)0xdeadbeef,0);
f771dc23 547 processed++;
548 if (max != -1) max--;
e2641e09 549 }
550 }
f771dc23 551 return processed;
e2641e09 552}
553
f771dc23 554void waitEmptyIOJobsQueue(void) {
555 processActiveIOJobs(-1);
556}
557
558/* Process up to 'max' IO Jobs already completed by threads but still waiting
559 * processing from the main thread.
560 *
561 * If max == -1 all the pending jobs are processed.
562 *
563 * The number of processed jobs is returned. */
564int processPendingIOJobs(int max) {
565 int processed = 0;
566
567 while(max == -1 || max > 0) {
8d51fb6a 568 int io_processed_len;
569
570 lockThreadedIO();
571 io_processed_len = listLength(server.io_processed);
572 unlockThreadedIO();
f771dc23 573 if (io_processed_len == 0) break;
8d51fb6a 574 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
575 (void*)0xdeadbeef,0);
f771dc23 576 if (max != -1) max--;
577 processed++;
8d51fb6a 578 }
f771dc23 579 return processed;
580}
581
582void processAllPendingIOJobs(void) {
583 processPendingIOJobs(-1);
8d51fb6a 584}
585
e2641e09 586/* This function must be called while with threaded IO locked */
587void queueIOJob(iojob *j) {
588 redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
589 (void*)j, j->type, (char*)j->key->ptr);
590 listAddNodeTail(server.io_newjobs,j);
e2641e09 591}
592
249ad25f 593/* Consume all the IO scheduled operations, and all the thread IO jobs
594 * so that eventually the state of diskstore is a point-in-time snapshot.
595 *
596 * This is useful when we need to BGSAVE with diskstore enabled. */
597void cacheForcePointInTime(void) {
598 redisLog(REDIS_NOTICE,"Diskstore: synching on disk to reach point-in-time state.");
599 while (listLength(server.cache_io_queue) != 0) {
600 cacheScheduleIOPushJobs(REDIS_IO_ASAP);
601 processActiveIOJobs(1);
602 }
603 waitEmptyIOJobsQueue();
604 processAllPendingIOJobs();
605}
606
05600eb8 607void cacheCreateIOJob(int type, redisDb *db, robj *key, robj *val, time_t expire) {
e2641e09 608 iojob *j;
609
610 j = zmalloc(sizeof(*j));
5ef64098 611 j->type = type;
e2641e09 612 j->db = db;
613 j->key = key;
614 incrRefCount(key);
5ef64098 615 j->val = val;
1609a1c4 616 if (val) incrRefCount(val);
05600eb8 617 j->expire = expire;
e2641e09 618
619 lockThreadedIO();
620 queueIOJob(j);
98a9abb6 621 pthread_cond_signal(&server.io_condvar);
e2641e09 622 unlockThreadedIO();
e2641e09 623}
624
3be00d7e 625/* ============= Disk store cache - Scheduling of IO operations =============
626 *
627 * We use a queue and an hash table to hold the state of IO operations
628 * so that's fast to lookup if there is already an IO operation in queue
629 * for a given key.
630 *
631 * There are two types of IO operations for a given key:
632 * REDIS_IO_LOAD and REDIS_IO_SAVE.
633 *
634 * The function cacheScheduleIO() function pushes the specified IO operation
635 * in the queue, but avoid adding the same key for the same operation
636 * multiple times, thanks to the associated hash table.
637 *
638 * We take a set of flags per every key, so when the scheduled IO operation
639 * gets moved from the scheduled queue to the actual IO Jobs queue that
640 * is processed by the IO thread, we flag it as IO_LOADINPROG or
641 * IO_SAVEINPROG.
642 *
643 * So for every given key we always know if there is some IO operation
644 * scheduled, or in progress, for this key.
645 *
646 * NOTE: all this is very important in order to guarantee correctness of
647 * the Disk Store Cache. Jobs are always queued here. Load jobs are
648 * queued at the head for faster execution only in the case there is not
649 * already a write operation of some kind for this job.
650 *
651 * So we have ordering, but can do exceptions when there are no already
652 * operations for a given key. Also when we need to block load a given
653 * key, for an immediate lookup operation, we can check if the key can
654 * be accessed synchronously without race conditions (no IN PROGRESS
655 * operations for this key), otherwise we blocking wait for completion. */
656
657#define REDIS_IO_LOAD 1
658#define REDIS_IO_SAVE 2
659#define REDIS_IO_LOADINPROG 4
660#define REDIS_IO_SAVEINPROG 8
661
662void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) {
663 struct dictEntry *de = dictFind(db->io_queued,key);
664
665 if (!de) {
666 dictAdd(db->io_queued,key,(void*)flag);
667 incrRefCount(key);
668 return;
669 } else {
670 long flags = (long) dictGetEntryVal(de);
e37efb0d 671
672 if (flags & flag) {
673 redisLog(REDIS_WARNING,"Adding the same flag again: was: %ld, addede: %ld",flags,flag);
674 redisAssert(!(flags & flag));
675 }
3be00d7e 676 flags |= flag;
677 dictGetEntryVal(de) = (void*) flags;
f63f0928 678 }
3be00d7e 679}
680
681void cacheScheduleIODelFlag(redisDb *db, robj *key, long flag) {
682 struct dictEntry *de = dictFind(db->io_queued,key);
683 long flags;
684
685 redisAssert(de != NULL);
686 flags = (long) dictGetEntryVal(de);
687 redisAssert(flags & flag);
688 flags &= ~flag;
689 if (flags == 0) {
690 dictDelete(db->io_queued,key);
691 } else {
692 dictGetEntryVal(de) = (void*) flags;
693 }
694}
f63f0928 695
3be00d7e 696int cacheScheduleIOGetFlags(redisDb *db, robj *key) {
697 struct dictEntry *de = dictFind(db->io_queued,key);
698
699 return (de == NULL) ? 0 : ((long) dictGetEntryVal(de));
700}
701
702void cacheScheduleIO(redisDb *db, robj *key, int type) {
703 ioop *op;
704 long flags;
705
706 if ((flags = cacheScheduleIOGetFlags(db,key)) & type) return;
707
708 redisLog(REDIS_DEBUG,"Scheduling key %s for %s",
709 key->ptr, type == REDIS_IO_LOAD ? "loading" : "saving");
710 cacheScheduleIOAddFlag(db,key,type);
711 op = zmalloc(sizeof(*op));
712 op->type = type;
713 op->db = db;
714 op->key = key;
f63f0928 715 incrRefCount(key);
3be00d7e 716 op->ctime = time(NULL);
717
718 /* Give priority to load operations if there are no save already
719 * in queue for the same key. */
720 if (type == REDIS_IO_LOAD && !(flags & REDIS_IO_SAVE)) {
721 listAddNodeHead(server.cache_io_queue, op);
f771dc23 722 cacheScheduleIOPushJobs(REDIS_IO_ONLYLOADS);
3be00d7e 723 } else {
724 /* FIXME: probably when this happens we want to at least move
725 * the write job about this queue on top, and set the creation time
726 * to a value that will force processing ASAP. */
727 listAddNodeTail(server.cache_io_queue, op);
728 }
f63f0928 729}
730
9b24d8ad 731/* Push scheduled IO operations into IO Jobs that the IO thread can process.
f771dc23 732 *
733 * If flags include REDIS_IO_ONLYLOADS only load jobs are processed:this is
734 * useful since it's safe to push LOAD IO jobs from any place of the code, while
9b24d8ad 735 * SAVE io jobs should never be pushed while we are processing a command
736 * (not protected by lookupKey() that will block on keys in IO_SAVEINPROG
f771dc23 737 * state.
738 *
739 * The REDIS_IO_ASAP flag tells the function to don't wait for the IO job
740 * scheduled completion time, but just do the operation ASAP. This is useful
741 * when we need to reclaim memory from the IO queue.
742 */
419e1cca 743#define MAX_IO_JOBS_QUEUE 10
f771dc23 744int cacheScheduleIOPushJobs(int flags) {
f63f0928 745 time_t now = time(NULL);
746 listNode *ln;
418d5eaf 747 int jobs, topush = 0, pushed = 0;
c4b64a13 748
f03fe802 749 /* Don't push new jobs if there is a threaded BGSAVE in progress. */
750 if (server.bgsavethread != (pthread_t) -1) return 0;
751
9b24d8ad 752 /* Sync stuff on disk, but only if we have less
753 * than MAX_IO_JOBS_QUEUE IO jobs. */
c4b64a13 754 lockThreadedIO();
755 jobs = listLength(server.io_newjobs);
756 unlockThreadedIO();
757
9b24d8ad 758 topush = MAX_IO_JOBS_QUEUE-jobs;
c4b64a13 759 if (topush < 0) topush = 0;
e37efb0d 760 if (topush > (signed)listLength(server.cache_io_queue))
761 topush = listLength(server.cache_io_queue);
f63f0928 762
3be00d7e 763 while((ln = listFirst(server.cache_io_queue)) != NULL) {
764 ioop *op = ln->value;
f771dc23 765 struct dictEntry *de;
766 robj *val;
f63f0928 767
c4b64a13 768 if (!topush) break;
769 topush--;
770
f771dc23 771 if (op->type != REDIS_IO_LOAD && flags & REDIS_IO_ONLYLOADS) break;
772
f1df1739 773 /* Don't execute SAVE before the scheduled time for completion */
774 if (op->type == REDIS_IO_SAVE && !(flags & REDIS_IO_ASAP) &&
f771dc23 775 (now - op->ctime) < server.cache_flush_delay) break;
776
777 /* Don't add a SAVE job in the IO thread queue if there is already
778 * a save in progress for the same key. */
779 if (op->type == REDIS_IO_SAVE &&
780 cacheScheduleIOGetFlags(op->db,op->key) & REDIS_IO_SAVEINPROG)
3be00d7e 781 {
f771dc23 782 /* Move the operation at the end of the list if there
783 * are other operations, so we can try to process the next one.
784 * Otherwise break, nothing to do here. */
785 if (listLength(server.cache_io_queue) > 1) {
786 listDelNode(server.cache_io_queue,ln);
787 listAddNodeTail(server.cache_io_queue,op);
788 continue;
789 } else {
790 break;
e37efb0d 791 }
f771dc23 792 }
e37efb0d 793
f771dc23 794 redisLog(REDIS_DEBUG,"Creating IO %s Job for key %s",
795 op->type == REDIS_IO_LOAD ? "load" : "save", op->key->ptr);
3be00d7e 796
f771dc23 797 if (op->type == REDIS_IO_LOAD) {
05600eb8 798 cacheCreateIOJob(REDIS_IOJOB_LOAD,op->db,op->key,NULL,0);
f771dc23 799 } else {
05600eb8 800 time_t expire = -1;
801
f771dc23 802 /* Lookup the key, in order to put the current value in the IO
803 * Job. Otherwise if the key does not exists we schedule a disk
804 * store delete operation, setting the value to NULL. */
805 de = dictFind(op->db->dict,op->key->ptr);
806 if (de) {
807 val = dictGetEntryVal(de);
05600eb8 808 expire = getExpire(op->db,op->key);
f63f0928 809 } else {
f771dc23 810 /* Setting the value to NULL tells the IO thread to delete
811 * the key on disk. */
812 val = NULL;
f63f0928 813 }
05600eb8 814 cacheCreateIOJob(REDIS_IOJOB_SAVE,op->db,op->key,val,expire);
f63f0928 815 }
f771dc23 816 /* Mark the operation as in progress. */
817 cacheScheduleIODelFlag(op->db,op->key,op->type);
818 cacheScheduleIOAddFlag(op->db,op->key,
819 (op->type == REDIS_IO_LOAD) ? REDIS_IO_LOADINPROG :
820 REDIS_IO_SAVEINPROG);
821 /* Finally remove the operation from the queue.
822 * But we'll have trace of it in the hash table. */
823 listDelNode(server.cache_io_queue,ln);
824 decrRefCount(op->key);
825 zfree(op);
826 pushed++;
f63f0928 827 }
418d5eaf 828 return pushed;
9b24d8ad 829}
830
831void cacheCron(void) {
832 /* Push jobs */
833 cacheScheduleIOPushJobs(0);
f63f0928 834
835 /* Reclaim memory from the object cache */
836 while (server.ds_enabled && zmalloc_used_memory() >
837 server.cache_max_memory)
838 {
c15a3887 839 int done = 0;
840
841 if (cacheFreeOneEntry() == REDIS_OK) done++;
842 if (negativeCacheEvictOneEntry() == REDIS_OK) done++;
843 if (done == 0) break; /* nothing more to free */
f63f0928 844 }
845}
846
3be00d7e 847/* ========== Disk store cache - Blocking clients on missing keys =========== */
e2641e09 848
849/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
3be00d7e 850 * If the key is already in memory we don't need to block.
f63f0928 851 *
5ef64098 852 * FIXME: we should try if it's actually better to suspend the client
853 * accessing an object that is being saved, and awake it only when
854 * the saving was completed.
855 *
856 * Otherwise if the key is not in memory, we block the client and start
857 * an IO Job to load it:
858 *
859 * the key is added to the io_keys list in the client structure, and also
e2641e09 860 * in the hash table mapping swapped keys to waiting clients, that is,
861 * server.io_waited_keys. */
862int waitForSwappedKey(redisClient *c, robj *key) {
863 struct dictEntry *de;
e2641e09 864 list *l;
865
5ef64098 866 /* Return ASAP if the key is in memory */
e2641e09 867 de = dictFind(c->db->dict,key->ptr);
5ef64098 868 if (de != NULL) return 0;
e2641e09 869
d934e1e8 870 /* Don't wait for keys we are sure are not on disk either */
871 if (!cacheKeyMayExist(c->db,key)) return 0;
872
e2641e09 873 /* Add the key to the list of keys this client is waiting for.
874 * This maps clients to keys they are waiting for. */
875 listAddNodeTail(c->io_keys,key);
876 incrRefCount(key);
877
878 /* Add the client to the swapped keys => clients waiting map. */
879 de = dictFind(c->db->io_keys,key);
880 if (de == NULL) {
881 int retval;
882
883 /* For every key we take a list of clients blocked for it */
884 l = listCreate();
885 retval = dictAdd(c->db->io_keys,key,l);
886 incrRefCount(key);
887 redisAssert(retval == DICT_OK);
888 } else {
889 l = dictGetEntryVal(de);
890 }
891 listAddNodeTail(l,c);
892
893 /* Are we already loading the key from disk? If not create a job */
7493d2a0 894 if (de == NULL) {
895 int flags = cacheScheduleIOGetFlags(c->db,key);
896
897 /* It is possible that even if there are no clients waiting for
898 * a load operation, still we have a load operation in progress.
899 * For instance think to a client performing a GET and then
900 * closing the connection */
901 if ((flags & (REDIS_IO_LOAD|REDIS_IO_LOADINPROG)) == 0)
902 cacheScheduleIO(c->db,key,REDIS_IO_LOAD);
903 }
e2641e09 904 return 1;
905}
906
e2641e09 907/* Is this client attempting to run a command against swapped keys?
908 * If so, block it ASAP, load the keys in background, then resume it.
909 *
910 * The important idea about this function is that it can fail! If keys will
911 * still be swapped when the client is resumed, this key lookups will
912 * just block loading keys from disk. In practical terms this should only
913 * happen with SORT BY command or if there is a bug in this function.
914 *
915 * Return 1 if the client is marked as blocked, 0 if the client can
916 * continue as the keys it is going to access appear to be in memory. */
917int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
42b2621c 918 int *keyindex, numkeys, j, i;
919
920 /* EXEC is a special case, we need to preload all the commands
921 * queued into the transaction */
922 if (cmd->proc == execCommand) {
923 struct redisCommand *mcmd;
924 robj **margv;
925 int margc;
926
927 if (!(c->flags & REDIS_MULTI)) return 0;
928 for (i = 0; i < c->mstate.count; i++) {
929 mcmd = c->mstate.commands[i].cmd;
930 margc = c->mstate.commands[i].argc;
931 margv = c->mstate.commands[i].argv;
932
933 keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
934 REDIS_GETKEYS_PRELOAD);
b4b51446 935 for (j = 0; j < numkeys; j++) {
a07bc926 936 redisLog(REDIS_DEBUG,"Preloading %s",
b4b51446 937 (char*)margv[keyindex[j]]->ptr);
42b2621c 938 waitForSwappedKey(c,margv[keyindex[j]]);
b4b51446 939 }
42b2621c 940 getKeysFreeResult(keyindex);
941 }
942 } else {
943 keyindex = getKeysFromCommand(cmd,c->argv,c->argc,&numkeys,
944 REDIS_GETKEYS_PRELOAD);
b4b51446 945 for (j = 0; j < numkeys; j++) {
a07bc926 946 redisLog(REDIS_DEBUG,"Preloading %s",
b4b51446 947 (char*)c->argv[keyindex[j]]->ptr);
948 waitForSwappedKey(c,c->argv[keyindex[j]]);
949 }
42b2621c 950 getKeysFreeResult(keyindex);
951 }
e2641e09 952
953 /* If the client was blocked for at least one key, mark it as blocked. */
954 if (listLength(c->io_keys)) {
955 c->flags |= REDIS_IO_WAIT;
956 aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
5ef64098 957 server.cache_blocked_clients++;
e2641e09 958 return 1;
959 } else {
960 return 0;
961 }
962}
963
964/* Remove the 'key' from the list of blocked keys for a given client.
965 *
966 * The function returns 1 when there are no longer blocking keys after
967 * the current one was removed (and the client can be unblocked). */
968int dontWaitForSwappedKey(redisClient *c, robj *key) {
969 list *l;
970 listNode *ln;
971 listIter li;
972 struct dictEntry *de;
973
c8a10631
PN
974 /* The key object might be destroyed when deleted from the c->io_keys
975 * list (and the "key" argument is physically the same object as the
976 * object inside the list), so we need to protect it. */
977 incrRefCount(key);
978
e2641e09 979 /* Remove the key from the list of keys this client is waiting for. */
980 listRewind(c->io_keys,&li);
981 while ((ln = listNext(&li)) != NULL) {
982 if (equalStringObjects(ln->value,key)) {
983 listDelNode(c->io_keys,ln);
984 break;
985 }
986 }
987 redisAssert(ln != NULL);
988
989 /* Remove the client form the key => waiting clients map. */
990 de = dictFind(c->db->io_keys,key);
991 redisAssert(de != NULL);
992 l = dictGetEntryVal(de);
993 ln = listSearchKey(l,c);
994 redisAssert(ln != NULL);
995 listDelNode(l,ln);
996 if (listLength(l) == 0)
997 dictDelete(c->db->io_keys,key);
998
c8a10631 999 decrRefCount(key);
e2641e09 1000 return listLength(c->io_keys) == 0;
1001}
1002
1003/* Every time we now a key was loaded back in memory, we handle clients
1004 * waiting for this key if any. */
1005void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
1006 struct dictEntry *de;
1007 list *l;
1008 listNode *ln;
1009 int len;
1010
1011 de = dictFind(db->io_keys,key);
1012 if (!de) return;
1013
1014 l = dictGetEntryVal(de);
1015 len = listLength(l);
1016 /* Note: we can't use something like while(listLength(l)) as the list
1017 * can be freed by the calling function when we remove the last element. */
1018 while (len--) {
1019 ln = listFirst(l);
1020 redisClient *c = ln->value;
1021
1022 if (dontWaitForSwappedKey(c,key)) {
1023 /* Put the client in the list of clients ready to go as we
1024 * loaded all the keys about it. */
1025 listAddNodeTail(server.io_ready_clients,c);
1026 }
1027 }
1028}