]> git.saurik.com Git - redis.git/blame - src/dscache.c
Better pretty printed makefile output
[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. */
365 if (dbAdd(j->db,j->key,j->val) == REDIS_OK) {
ad01a255 366 incrRefCount(j->val);
367 if (j->expire != -1) setExpire(j->db,j->key,j->expire);
368 }
5d46e370 369 } else {
370 /* Key not found on disk. If it is also not in memory
371 * as a cached object, nor there is a job writing it
372 * in background, we are sure the key does not exist
373 * currently.
374 *
375 * So we set a negative cache entry avoiding that the
376 * resumed client will block load what does not exist... */
bafa88c8 377 if (dictFind(j->db->dict,j->key->ptr) == NULL &&
5d46e370 378 (cacheScheduleIOGetFlags(j->db,j->key) &
379 (REDIS_IO_SAVE|REDIS_IO_SAVEINPROG)) == 0)
380 {
381 cacheSetKeyDoesNotExist(j->db,j->key);
382 }
4ab98823 383 }
3be00d7e 384 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_LOADINPROG);
5ef64098 385 handleClientsBlockedOnSwappedKey(j->db,j->key);
5f6e1183 386 } else if (j->type == REDIS_IOJOB_SAVE) {
3be00d7e 387 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG);
e2641e09 388 }
3a21cb99 389 freeIOJob(j);
e2641e09 390 processed++;
d9fac6c0 391 if (privdata == NULL) cacheScheduleIOPushJobs(0);
e2641e09 392 if (processed == toprocess) return;
393 }
394 if (retval < 0 && errno != EAGAIN) {
395 redisLog(REDIS_WARNING,
396 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
397 strerror(errno));
398 }
399}
400
401void lockThreadedIO(void) {
402 pthread_mutex_lock(&server.io_mutex);
403}
404
405void unlockThreadedIO(void) {
406 pthread_mutex_unlock(&server.io_mutex);
407}
408
e2641e09 409void *IOThreadEntryPoint(void *arg) {
410 iojob *j;
411 listNode *ln;
412 REDIS_NOTUSED(arg);
419e1cca 413 long long start;
e2641e09 414
415 pthread_detach(pthread_self());
98a9abb6 416 lockThreadedIO();
e2641e09 417 while(1) {
418 /* Get a new job to process */
e2641e09 419 if (listLength(server.io_newjobs) == 0) {
a440ecf0 420 /* Wait for more work to do */
05600eb8 421 redisLog(REDIS_DEBUG,"[T] wait for signal");
a440ecf0 422 pthread_cond_wait(&server.io_condvar,&server.io_mutex);
05600eb8 423 redisLog(REDIS_DEBUG,"[T] signal received");
1609a1c4 424 continue;
e2641e09 425 }
419e1cca 426 start = ustime();
05600eb8 427 redisLog(REDIS_DEBUG,"[T] %ld IO jobs to process",
c4b64a13 428 listLength(server.io_newjobs));
e2641e09 429 ln = listFirst(server.io_newjobs);
430 j = ln->value;
431 listDelNode(server.io_newjobs,ln);
432 /* Add the job in the processing queue */
e2641e09 433 listAddNodeTail(server.io_processing,j);
434 ln = listLast(server.io_processing); /* We use ln later to remove it */
435 unlockThreadedIO();
98a9abb6 436
05600eb8 437 redisLog(REDIS_DEBUG,"[T] %ld: new job type %s: %p about key '%s'",
5ef64098 438 (long) pthread_self(),
439 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
440 (void*)j, (char*)j->key->ptr);
e2641e09 441
442 /* Process the Job */
443 if (j->type == REDIS_IOJOB_LOAD) {
4ab98823 444 time_t expire;
445
446 j->val = dsGet(j->db,j->key,&expire);
447 if (j->val) j->expire = expire;
5ef64098 448 } else if (j->type == REDIS_IOJOB_SAVE) {
31222292 449 if (j->val) {
05600eb8 450 dsSet(j->db,j->key,j->val,j->expire);
31222292 451 } else {
5ef64098 452 dsDel(j->db,j->key);
31222292 453 }
e2641e09 454 }
455
456 /* Done: insert the job into the processed queue */
05600eb8 457 redisLog(REDIS_DEBUG,"[T] %ld completed the job: %p (key %s)",
e2641e09 458 (long) pthread_self(), (void*)j, (char*)j->key->ptr);
98a9abb6 459
05600eb8 460 redisLog(REDIS_DEBUG,"[T] lock IO");
e2641e09 461 lockThreadedIO();
05600eb8 462 redisLog(REDIS_DEBUG,"[T] IO locked");
e2641e09 463 listDelNode(server.io_processing,ln);
464 listAddNodeTail(server.io_processed,j);
e2641e09 465
466 /* Signal the main thread there is new stuff to process */
467 redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
8b108ed3 468 redisLog(REDIS_DEBUG,"TIME (%c): %lld\n", j->type == REDIS_IOJOB_LOAD ? 'L' : 'S', ustime()-start);
e2641e09 469 }
98a9abb6 470 /* never reached, but that's the full pattern... */
471 unlockThreadedIO();
472 return NULL;
e2641e09 473}
474
475void spawnIOThread(void) {
476 pthread_t thread;
477 sigset_t mask, omask;
478 int err;
479
480 sigemptyset(&mask);
481 sigaddset(&mask,SIGCHLD);
482 sigaddset(&mask,SIGHUP);
483 sigaddset(&mask,SIGPIPE);
484 pthread_sigmask(SIG_SETMASK, &mask, &omask);
485 while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
486 redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
487 strerror(err));
488 usleep(1000000);
489 }
490 pthread_sigmask(SIG_SETMASK, &omask, NULL);
491 server.io_active_threads++;
492}
493
f771dc23 494/* Wait that up to 'max' pending IO Jobs are processed by the I/O thread.
495 * From our point of view an IO job processed means that the count of
496 * server.io_processed must increase by one.
497 *
498 * If max is -1, all the pending IO jobs will be processed.
499 *
500 * Returns the number of IO jobs processed.
501 *
502 * NOTE: while this may appear like a busy loop, we are actually blocked
503 * by IO since we continuously acquire/release the IO lock. */
504int processActiveIOJobs(int max) {
505 int processed = 0;
506
507 while(max == -1 || max > 0) {
e2641e09 508 int io_processed_len;
509
05600eb8 510 redisLog(REDIS_DEBUG,"[P] lock IO");
e2641e09 511 lockThreadedIO();
05600eb8 512 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));
513
e2641e09 514 if (listLength(server.io_newjobs) == 0 &&
8d51fb6a 515 listLength(server.io_processing) == 0)
e2641e09 516 {
f771dc23 517 /* There is nothing more to process */
05600eb8 518 redisLog(REDIS_DEBUG,"[P] Nothing to process, unlock IO, return");
e2641e09 519 unlockThreadedIO();
f771dc23 520 break;
e2641e09 521 }
f771dc23 522
05600eb8 523#if 1
a440ecf0 524 /* If there are new jobs we need to signal the thread to
249ad25f 525 * process the next one. FIXME: drop this if useless. */
05600eb8 526 redisLog(REDIS_DEBUG,"[P] waitEmptyIOJobsQueue: new %d, processing %d, processed %d",
a440ecf0 527 listLength(server.io_newjobs),
05600eb8 528 listLength(server.io_processing),
529 listLength(server.io_processed));
5d46e370 530
a440ecf0 531 if (listLength(server.io_newjobs)) {
05600eb8 532 redisLog(REDIS_DEBUG,"[P] There are new jobs, signal");
a440ecf0 533 pthread_cond_signal(&server.io_condvar);
534 }
f771dc23 535#endif
536
537 /* Check if we can process some finished job */
e2641e09 538 io_processed_len = listLength(server.io_processed);
05600eb8 539 redisLog(REDIS_DEBUG,"[P] Unblock IO");
e2641e09 540 unlockThreadedIO();
05600eb8 541 redisLog(REDIS_DEBUG,"[P] Wait");
542 usleep(10000);
e2641e09 543 if (io_processed_len) {
c1ae36ae 544 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
545 (void*)0xdeadbeef,0);
f771dc23 546 processed++;
547 if (max != -1) max--;
e2641e09 548 }
549 }
f771dc23 550 return processed;
e2641e09 551}
552
f771dc23 553void waitEmptyIOJobsQueue(void) {
554 processActiveIOJobs(-1);
555}
556
557/* Process up to 'max' IO Jobs already completed by threads but still waiting
558 * processing from the main thread.
559 *
560 * If max == -1 all the pending jobs are processed.
561 *
562 * The number of processed jobs is returned. */
563int processPendingIOJobs(int max) {
564 int processed = 0;
565
566 while(max == -1 || max > 0) {
8d51fb6a 567 int io_processed_len;
568
569 lockThreadedIO();
570 io_processed_len = listLength(server.io_processed);
571 unlockThreadedIO();
f771dc23 572 if (io_processed_len == 0) break;
8d51fb6a 573 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
574 (void*)0xdeadbeef,0);
f771dc23 575 if (max != -1) max--;
576 processed++;
8d51fb6a 577 }
f771dc23 578 return processed;
579}
580
581void processAllPendingIOJobs(void) {
582 processPendingIOJobs(-1);
8d51fb6a 583}
584
e2641e09 585/* This function must be called while with threaded IO locked */
586void queueIOJob(iojob *j) {
587 redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
588 (void*)j, j->type, (char*)j->key->ptr);
589 listAddNodeTail(server.io_newjobs,j);
e2641e09 590}
591
249ad25f 592/* Consume all the IO scheduled operations, and all the thread IO jobs
593 * so that eventually the state of diskstore is a point-in-time snapshot.
594 *
595 * This is useful when we need to BGSAVE with diskstore enabled. */
596void cacheForcePointInTime(void) {
597 redisLog(REDIS_NOTICE,"Diskstore: synching on disk to reach point-in-time state.");
598 while (listLength(server.cache_io_queue) != 0) {
599 cacheScheduleIOPushJobs(REDIS_IO_ASAP);
600 processActiveIOJobs(1);
601 }
602 waitEmptyIOJobsQueue();
603 processAllPendingIOJobs();
604}
605
05600eb8 606void cacheCreateIOJob(int type, redisDb *db, robj *key, robj *val, time_t expire) {
e2641e09 607 iojob *j;
608
609 j = zmalloc(sizeof(*j));
5ef64098 610 j->type = type;
e2641e09 611 j->db = db;
612 j->key = key;
613 incrRefCount(key);
5ef64098 614 j->val = val;
1609a1c4 615 if (val) incrRefCount(val);
05600eb8 616 j->expire = expire;
e2641e09 617
618 lockThreadedIO();
619 queueIOJob(j);
98a9abb6 620 pthread_cond_signal(&server.io_condvar);
e2641e09 621 unlockThreadedIO();
e2641e09 622}
623
3be00d7e 624/* ============= Disk store cache - Scheduling of IO operations =============
625 *
626 * We use a queue and an hash table to hold the state of IO operations
627 * so that's fast to lookup if there is already an IO operation in queue
628 * for a given key.
629 *
630 * There are two types of IO operations for a given key:
631 * REDIS_IO_LOAD and REDIS_IO_SAVE.
632 *
633 * The function cacheScheduleIO() function pushes the specified IO operation
634 * in the queue, but avoid adding the same key for the same operation
635 * multiple times, thanks to the associated hash table.
636 *
637 * We take a set of flags per every key, so when the scheduled IO operation
638 * gets moved from the scheduled queue to the actual IO Jobs queue that
639 * is processed by the IO thread, we flag it as IO_LOADINPROG or
640 * IO_SAVEINPROG.
641 *
642 * So for every given key we always know if there is some IO operation
643 * scheduled, or in progress, for this key.
644 *
645 * NOTE: all this is very important in order to guarantee correctness of
646 * the Disk Store Cache. Jobs are always queued here. Load jobs are
647 * queued at the head for faster execution only in the case there is not
648 * already a write operation of some kind for this job.
649 *
650 * So we have ordering, but can do exceptions when there are no already
651 * operations for a given key. Also when we need to block load a given
652 * key, for an immediate lookup operation, we can check if the key can
653 * be accessed synchronously without race conditions (no IN PROGRESS
654 * operations for this key), otherwise we blocking wait for completion. */
655
656#define REDIS_IO_LOAD 1
657#define REDIS_IO_SAVE 2
658#define REDIS_IO_LOADINPROG 4
659#define REDIS_IO_SAVEINPROG 8
660
661void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) {
662 struct dictEntry *de = dictFind(db->io_queued,key);
663
664 if (!de) {
665 dictAdd(db->io_queued,key,(void*)flag);
666 incrRefCount(key);
667 return;
668 } else {
669 long flags = (long) dictGetEntryVal(de);
e37efb0d 670
671 if (flags & flag) {
672 redisLog(REDIS_WARNING,"Adding the same flag again: was: %ld, addede: %ld",flags,flag);
673 redisAssert(!(flags & flag));
674 }
3be00d7e 675 flags |= flag;
676 dictGetEntryVal(de) = (void*) flags;
f63f0928 677 }
3be00d7e 678}
679
680void cacheScheduleIODelFlag(redisDb *db, robj *key, long flag) {
681 struct dictEntry *de = dictFind(db->io_queued,key);
682 long flags;
683
684 redisAssert(de != NULL);
685 flags = (long) dictGetEntryVal(de);
686 redisAssert(flags & flag);
687 flags &= ~flag;
688 if (flags == 0) {
689 dictDelete(db->io_queued,key);
690 } else {
691 dictGetEntryVal(de) = (void*) flags;
692 }
693}
f63f0928 694
3be00d7e 695int cacheScheduleIOGetFlags(redisDb *db, robj *key) {
696 struct dictEntry *de = dictFind(db->io_queued,key);
697
698 return (de == NULL) ? 0 : ((long) dictGetEntryVal(de));
699}
700
701void cacheScheduleIO(redisDb *db, robj *key, int type) {
702 ioop *op;
703 long flags;
704
705 if ((flags = cacheScheduleIOGetFlags(db,key)) & type) return;
706
707 redisLog(REDIS_DEBUG,"Scheduling key %s for %s",
708 key->ptr, type == REDIS_IO_LOAD ? "loading" : "saving");
709 cacheScheduleIOAddFlag(db,key,type);
710 op = zmalloc(sizeof(*op));
711 op->type = type;
712 op->db = db;
713 op->key = key;
f63f0928 714 incrRefCount(key);
3be00d7e 715 op->ctime = time(NULL);
716
717 /* Give priority to load operations if there are no save already
718 * in queue for the same key. */
719 if (type == REDIS_IO_LOAD && !(flags & REDIS_IO_SAVE)) {
720 listAddNodeHead(server.cache_io_queue, op);
f771dc23 721 cacheScheduleIOPushJobs(REDIS_IO_ONLYLOADS);
3be00d7e 722 } else {
723 /* FIXME: probably when this happens we want to at least move
724 * the write job about this queue on top, and set the creation time
725 * to a value that will force processing ASAP. */
726 listAddNodeTail(server.cache_io_queue, op);
727 }
f63f0928 728}
729
9b24d8ad 730/* Push scheduled IO operations into IO Jobs that the IO thread can process.
f771dc23 731 *
732 * If flags include REDIS_IO_ONLYLOADS only load jobs are processed:this is
733 * useful since it's safe to push LOAD IO jobs from any place of the code, while
9b24d8ad 734 * SAVE io jobs should never be pushed while we are processing a command
735 * (not protected by lookupKey() that will block on keys in IO_SAVEINPROG
f771dc23 736 * state.
737 *
738 * The REDIS_IO_ASAP flag tells the function to don't wait for the IO job
739 * scheduled completion time, but just do the operation ASAP. This is useful
740 * when we need to reclaim memory from the IO queue.
741 */
419e1cca 742#define MAX_IO_JOBS_QUEUE 10
f771dc23 743int cacheScheduleIOPushJobs(int flags) {
f63f0928 744 time_t now = time(NULL);
745 listNode *ln;
418d5eaf 746 int jobs, topush = 0, pushed = 0;
c4b64a13 747
f03fe802 748 /* Don't push new jobs if there is a threaded BGSAVE in progress. */
749 if (server.bgsavethread != (pthread_t) -1) return 0;
750
9b24d8ad 751 /* Sync stuff on disk, but only if we have less
752 * than MAX_IO_JOBS_QUEUE IO jobs. */
c4b64a13 753 lockThreadedIO();
754 jobs = listLength(server.io_newjobs);
755 unlockThreadedIO();
756
9b24d8ad 757 topush = MAX_IO_JOBS_QUEUE-jobs;
c4b64a13 758 if (topush < 0) topush = 0;
e37efb0d 759 if (topush > (signed)listLength(server.cache_io_queue))
760 topush = listLength(server.cache_io_queue);
f63f0928 761
3be00d7e 762 while((ln = listFirst(server.cache_io_queue)) != NULL) {
763 ioop *op = ln->value;
f771dc23 764 struct dictEntry *de;
765 robj *val;
f63f0928 766
c4b64a13 767 if (!topush) break;
768 topush--;
769
f771dc23 770 if (op->type != REDIS_IO_LOAD && flags & REDIS_IO_ONLYLOADS) break;
771
f1df1739 772 /* Don't execute SAVE before the scheduled time for completion */
773 if (op->type == REDIS_IO_SAVE && !(flags & REDIS_IO_ASAP) &&
f771dc23 774 (now - op->ctime) < server.cache_flush_delay) break;
775
776 /* Don't add a SAVE job in the IO thread queue if there is already
777 * a save in progress for the same key. */
778 if (op->type == REDIS_IO_SAVE &&
779 cacheScheduleIOGetFlags(op->db,op->key) & REDIS_IO_SAVEINPROG)
3be00d7e 780 {
f771dc23 781 /* Move the operation at the end of the list if there
782 * are other operations, so we can try to process the next one.
783 * Otherwise break, nothing to do here. */
784 if (listLength(server.cache_io_queue) > 1) {
785 listDelNode(server.cache_io_queue,ln);
786 listAddNodeTail(server.cache_io_queue,op);
787 continue;
788 } else {
789 break;
e37efb0d 790 }
f771dc23 791 }
e37efb0d 792
f771dc23 793 redisLog(REDIS_DEBUG,"Creating IO %s Job for key %s",
794 op->type == REDIS_IO_LOAD ? "load" : "save", op->key->ptr);
3be00d7e 795
f771dc23 796 if (op->type == REDIS_IO_LOAD) {
05600eb8 797 cacheCreateIOJob(REDIS_IOJOB_LOAD,op->db,op->key,NULL,0);
f771dc23 798 } else {
05600eb8 799 time_t expire = -1;
800
f771dc23 801 /* Lookup the key, in order to put the current value in the IO
802 * Job. Otherwise if the key does not exists we schedule a disk
803 * store delete operation, setting the value to NULL. */
804 de = dictFind(op->db->dict,op->key->ptr);
805 if (de) {
806 val = dictGetEntryVal(de);
05600eb8 807 expire = getExpire(op->db,op->key);
f63f0928 808 } else {
f771dc23 809 /* Setting the value to NULL tells the IO thread to delete
810 * the key on disk. */
811 val = NULL;
f63f0928 812 }
05600eb8 813 cacheCreateIOJob(REDIS_IOJOB_SAVE,op->db,op->key,val,expire);
f63f0928 814 }
f771dc23 815 /* Mark the operation as in progress. */
816 cacheScheduleIODelFlag(op->db,op->key,op->type);
817 cacheScheduleIOAddFlag(op->db,op->key,
818 (op->type == REDIS_IO_LOAD) ? REDIS_IO_LOADINPROG :
819 REDIS_IO_SAVEINPROG);
820 /* Finally remove the operation from the queue.
821 * But we'll have trace of it in the hash table. */
822 listDelNode(server.cache_io_queue,ln);
823 decrRefCount(op->key);
824 zfree(op);
825 pushed++;
f63f0928 826 }
418d5eaf 827 return pushed;
9b24d8ad 828}
829
830void cacheCron(void) {
831 /* Push jobs */
832 cacheScheduleIOPushJobs(0);
f63f0928 833
834 /* Reclaim memory from the object cache */
835 while (server.ds_enabled && zmalloc_used_memory() >
836 server.cache_max_memory)
837 {
c15a3887 838 int done = 0;
839
840 if (cacheFreeOneEntry() == REDIS_OK) done++;
841 if (negativeCacheEvictOneEntry() == REDIS_OK) done++;
842 if (done == 0) break; /* nothing more to free */
f63f0928 843 }
844}
845
3be00d7e 846/* ========== Disk store cache - Blocking clients on missing keys =========== */
e2641e09 847
848/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
3be00d7e 849 * If the key is already in memory we don't need to block.
f63f0928 850 *
5ef64098 851 * FIXME: we should try if it's actually better to suspend the client
852 * accessing an object that is being saved, and awake it only when
853 * the saving was completed.
854 *
855 * Otherwise if the key is not in memory, we block the client and start
856 * an IO Job to load it:
857 *
858 * the key is added to the io_keys list in the client structure, and also
e2641e09 859 * in the hash table mapping swapped keys to waiting clients, that is,
860 * server.io_waited_keys. */
861int waitForSwappedKey(redisClient *c, robj *key) {
862 struct dictEntry *de;
e2641e09 863 list *l;
864
5ef64098 865 /* Return ASAP if the key is in memory */
e2641e09 866 de = dictFind(c->db->dict,key->ptr);
5ef64098 867 if (de != NULL) return 0;
e2641e09 868
d934e1e8 869 /* Don't wait for keys we are sure are not on disk either */
870 if (!cacheKeyMayExist(c->db,key)) return 0;
871
e2641e09 872 /* Add the key to the list of keys this client is waiting for.
873 * This maps clients to keys they are waiting for. */
874 listAddNodeTail(c->io_keys,key);
875 incrRefCount(key);
876
877 /* Add the client to the swapped keys => clients waiting map. */
878 de = dictFind(c->db->io_keys,key);
879 if (de == NULL) {
880 int retval;
881
882 /* For every key we take a list of clients blocked for it */
883 l = listCreate();
884 retval = dictAdd(c->db->io_keys,key,l);
885 incrRefCount(key);
886 redisAssert(retval == DICT_OK);
887 } else {
888 l = dictGetEntryVal(de);
889 }
890 listAddNodeTail(l,c);
891
892 /* Are we already loading the key from disk? If not create a job */
7493d2a0 893 if (de == NULL) {
894 int flags = cacheScheduleIOGetFlags(c->db,key);
895
896 /* It is possible that even if there are no clients waiting for
897 * a load operation, still we have a load operation in progress.
898 * For instance think to a client performing a GET and then
899 * closing the connection */
900 if ((flags & (REDIS_IO_LOAD|REDIS_IO_LOADINPROG)) == 0)
901 cacheScheduleIO(c->db,key,REDIS_IO_LOAD);
902 }
e2641e09 903 return 1;
904}
905
e2641e09 906/* Is this client attempting to run a command against swapped keys?
907 * If so, block it ASAP, load the keys in background, then resume it.
908 *
909 * The important idea about this function is that it can fail! If keys will
910 * still be swapped when the client is resumed, this key lookups will
911 * just block loading keys from disk. In practical terms this should only
912 * happen with SORT BY command or if there is a bug in this function.
913 *
914 * Return 1 if the client is marked as blocked, 0 if the client can
915 * continue as the keys it is going to access appear to be in memory. */
916int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
42b2621c 917 int *keyindex, numkeys, j, i;
918
919 /* EXEC is a special case, we need to preload all the commands
920 * queued into the transaction */
921 if (cmd->proc == execCommand) {
922 struct redisCommand *mcmd;
923 robj **margv;
924 int margc;
925
926 if (!(c->flags & REDIS_MULTI)) return 0;
927 for (i = 0; i < c->mstate.count; i++) {
928 mcmd = c->mstate.commands[i].cmd;
929 margc = c->mstate.commands[i].argc;
930 margv = c->mstate.commands[i].argv;
931
932 keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
933 REDIS_GETKEYS_PRELOAD);
b4b51446 934 for (j = 0; j < numkeys; j++) {
a07bc926 935 redisLog(REDIS_DEBUG,"Preloading %s",
b4b51446 936 (char*)margv[keyindex[j]]->ptr);
42b2621c 937 waitForSwappedKey(c,margv[keyindex[j]]);
b4b51446 938 }
42b2621c 939 getKeysFreeResult(keyindex);
940 }
941 } else {
942 keyindex = getKeysFromCommand(cmd,c->argv,c->argc,&numkeys,
943 REDIS_GETKEYS_PRELOAD);
b4b51446 944 for (j = 0; j < numkeys; j++) {
a07bc926 945 redisLog(REDIS_DEBUG,"Preloading %s",
b4b51446 946 (char*)c->argv[keyindex[j]]->ptr);
947 waitForSwappedKey(c,c->argv[keyindex[j]]);
948 }
42b2621c 949 getKeysFreeResult(keyindex);
950 }
e2641e09 951
952 /* If the client was blocked for at least one key, mark it as blocked. */
953 if (listLength(c->io_keys)) {
954 c->flags |= REDIS_IO_WAIT;
955 aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
5ef64098 956 server.cache_blocked_clients++;
e2641e09 957 return 1;
958 } else {
959 return 0;
960 }
961}
962
963/* Remove the 'key' from the list of blocked keys for a given client.
964 *
965 * The function returns 1 when there are no longer blocking keys after
966 * the current one was removed (and the client can be unblocked). */
967int dontWaitForSwappedKey(redisClient *c, robj *key) {
968 list *l;
969 listNode *ln;
970 listIter li;
971 struct dictEntry *de;
972
c8a10631
PN
973 /* The key object might be destroyed when deleted from the c->io_keys
974 * list (and the "key" argument is physically the same object as the
975 * object inside the list), so we need to protect it. */
976 incrRefCount(key);
977
e2641e09 978 /* Remove the key from the list of keys this client is waiting for. */
979 listRewind(c->io_keys,&li);
980 while ((ln = listNext(&li)) != NULL) {
981 if (equalStringObjects(ln->value,key)) {
982 listDelNode(c->io_keys,ln);
983 break;
984 }
985 }
986 redisAssert(ln != NULL);
987
988 /* Remove the client form the key => waiting clients map. */
989 de = dictFind(c->db->io_keys,key);
990 redisAssert(de != NULL);
991 l = dictGetEntryVal(de);
992 ln = listSearchKey(l,c);
993 redisAssert(ln != NULL);
994 listDelNode(l,ln);
995 if (listLength(l) == 0)
996 dictDelete(c->db->io_keys,key);
997
c8a10631 998 decrRefCount(key);
e2641e09 999 return listLength(c->io_keys) == 0;
1000}
1001
1002/* Every time we now a key was loaded back in memory, we handle clients
1003 * waiting for this key if any. */
1004void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
1005 struct dictEntry *de;
1006 list *l;
1007 listNode *ln;
1008 int len;
1009
1010 de = dictFind(db->io_keys,key);
1011 if (!de) return;
1012
1013 l = dictGetEntryVal(de);
1014 len = listLength(l);
1015 /* Note: we can't use something like while(listLength(l)) as the list
1016 * can be freed by the calling function when we remove the last element. */
1017 while (len--) {
1018 ln = listFirst(l);
1019 redisClient *c = ln->value;
1020
1021 if (dontWaitForSwappedKey(c,key)) {
1022 /* Put the client in the list of clients ready to go as we
1023 * loaded all the keys about it. */
1024 listAddNodeTail(server.io_ready_clients,c);
1025 }
1026 }
1027}