forked from redis/redis
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdscache.c
719 lines (642 loc) · 25.1 KB
/
dscache.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
#include "redis.h"
#include <fcntl.h>
#include <pthread.h>
#include <math.h>
#include <signal.h>
/* dscache.c - Disk store cache for disk store backend.
*
* When Redis is configured for using disk as backend instead of memory, the
* memory is used as a cache, so that recently accessed keys are taken in
* memory for fast read and write operations.
*
* Modified keys are marked to be flushed on disk, and will be flushed
* as long as the maxium configured flush time elapsed.
*
* This file implements the whole caching subsystem and contains further
* documentation. */
/* TODO:
*
* - The WATCH helper will be used to signal the cache system
* we need to flush a given key/dbid into disk, adding this key/dbid
* pair into a server.ds_cache_dirty linked list AND hash table (so that we
* don't add the same thing multiple times).
*
* - cron() checks if there are elements on this list. When there are things
* to flush, we create an IO Job for the I/O thread.
* NOTE: We disalbe object sharing when server.ds_enabled == 1 so objects
* that are referenced an IO job for flushing on disk are marked as
* o->storage == REDIS_DS_SAVING.
*
* - This is what we do on key lookup:
* 1) The key already exists in memory. object->storage == REDIS_DS_MEMORY
* or it is object->storage == REDIS_DS_DIRTY:
* We don't do nothing special, lookup, return value object pointer.
* 2) The key is in memory but object->storage == REDIS_DS_SAVING.
* When this happens we block waiting for the I/O thread to process
* this object. Then continue.
* 3) The key is not in memory. We block to load the key from disk.
* Of course the key may not be present at all on the disk store as well,
* in such case we just detect this condition and continue, returning
* NULL from lookup.
*
* - Preloading of needed keys:
* 1) As it was done with VM, also with this new system we try preloading
* keys a client is going to use. We block the client, load keys
* using the I/O thread, unblock the client. Same code as VM more or less.
*
* - Reclaiming memory.
* In cron() we detect our memory limit was reached. What we
* do is deleting keys that are REDIS_DS_MEMORY, using LRU.
*
* If this is not enough to return again under the memory limits we also
* start to flush keys that need to be synched on disk synchronously,
* removing it from the memory. We do this blocking as memory limit is a
* much "harder" barrirer in the new design.
*
* - IO thread operations are no longer stopped for sync loading/saving of
* things. When a key is found to be in the process of being saved
* we simply wait for the IO thread to end its work.
*
* Otherwise if there is to load a key without any IO thread operation
* just started it is blocking-loaded in the lookup function.
*
* - What happens when an object is destroyed?
*
* If o->storage == REDIS_DS_MEMORY then we simply destory the object.
* If o->storage == REDIS_DS_DIRTY we can still remove the object. It had
* changes not flushed on disk, but is being removed so
* who cares.
* if o->storage == REDIS_DS_SAVING then the object is being saved so
* it is impossible that its refcount == 1, must be at
* least two. When the object is saved the storage will
* be set back to DS_MEMORY.
*
* - What happens when keys are deleted?
*
* We simply schedule a key flush operation as usually, but when the
* IO thread will be created the object pointer will be set to NULL
* so the IO thread will know that the work to do is to delete the key
* from the disk store.
*
* - What happens with MULTI/EXEC?
*
* Good question.
*
* - If dsSet() fails on the write thread log the error and reschedule the
* key for flush.
*/
/* Virtual Memory is composed mainly of two subsystems:
* - Blocking Virutal Memory
* - Threaded Virtual Memory I/O
* The two parts are not fully decoupled, but functions are split among two
* different sections of the source code (delimited by comments) in order to
* make more clear what functionality is about the blocking VM and what about
* the threaded (not blocking) VM.
*
* Redis VM design:
*
* Redis VM is a blocking VM (one that blocks reading swapped values from
* disk into memory when a value swapped out is needed in memory) that is made
* unblocking by trying to examine the command argument vector in order to
* load in background values that will likely be needed in order to exec
* the command. The command is executed only once all the relevant keys
* are loaded into memory.
*
* This basically is almost as simple of a blocking VM, but almost as parallel
* as a fully non-blocking VM.
*/
void spawnIOThread(void);
/* =================== Virtual Memory - Blocking Side ====================== */
void dsInit(void) {
int pipefds[2];
size_t stacksize;
zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);
/* Open Disk Store */
if (dsOpen() != REDIS_OK) {
redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
exit(1);
};
/* Initialize threaded I/O for Object Cache */
server.io_newjobs = listCreate();
server.io_processing = listCreate();
server.io_processed = listCreate();
server.io_ready_clients = listCreate();
pthread_mutex_init(&server.io_mutex,NULL);
server.io_active_threads = 0;
if (pipe(pipefds) == -1) {
redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
,strerror(errno));
exit(1);
}
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);
/* Solaris may report a stacksize of 0, let's set it to 1 otherwise
* multiplying it by 2 in the while loop later will not really help ;) */
if (!stacksize) stacksize = 1;
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)
oom("creating file event");
/* Spawn our I/O thread */
spawnIOThread();
}
/* Compute how good candidate the specified object is for eviction.
* An higher number means a better candidate. */
double computeObjectSwappability(robj *o) {
/* actual age can be >= minage, but not < minage. As we use wrapping
* 21 bit clocks with minutes resolution for the LRU. */
return (double) estimateObjectIdleTime(o);
}
/* Try to free one entry from the diskstore object cache */
int cacheFreeOneEntry(void) {
int j, i;
struct dictEntry *best = NULL;
double best_swappability = 0;
redisDb *best_db = NULL;
robj *val;
sds key;
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
/* Why maxtries is set to 100?
* Because this way (usually) we'll find 1 object even if just 1% - 2%
* are swappable objects */
int maxtries = 100;
if (dictSize(db->dict) == 0) continue;
for (i = 0; i < 5; i++) {
dictEntry *de;
double swappability;
if (maxtries) maxtries--;
de = dictGetRandomKey(db->dict);
val = dictGetEntryVal(de);
/* Only swap objects that are currently in memory.
*
* Also don't swap shared objects: not a good idea in general and
* we need 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 (val->storage != REDIS_DS_MEMORY) {
if (maxtries) i--; /* don't count this try */
continue;
}
swappability = computeObjectSwappability(val);
if (!best || swappability > best_swappability) {
best = de;
best_swappability = swappability;
best_db = db;
}
}
}
if (best == NULL) {
/* FIXME: If there are objects marked as DS_DIRTY or DS_SAVING
* let's wait for this objects to be clear and retry...
*
* Object cache vm limit is considered an hard limit. */
return REDIS_ERR;
}
key = dictGetEntryKey(best);
val = dictGetEntryVal(best);
redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
key, best_swappability);
/* Delete this key from memory */
{
robj *kobj = createStringObject(key,sdslen(key));
dbDelete(best_db,kobj);
decrRefCount(kobj);
}
return REDIS_OK;
}
/* Return true if it's safe to swap out objects in a given moment.
* Basically we don't want to swap objects out while there is a BGSAVE
* or a BGAEOREWRITE running in backgroud. */
int dsCanTouchDiskStore(void) {
return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
}
/* =================== Virtual Memory - Threaded I/O ======================= */
void freeIOJob(iojob *j) {
decrRefCount(j->key);
/* j->val can be NULL if the job is about deleting the key from disk. */
if (j->val) decrRefCount(j->val);
zfree(j);
}
/* Every time a thread finished a Job, it writes a byte into the write side
* of an unix pipe in order to "awake" the main thread, and this function
* is called. */
void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
int mask)
{
char buf[1];
int retval, processed = 0, toprocess = -1;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata);
/* For every byte we read in the read side of the pipe, there is one
* I/O job completed to process. */
while((retval = read(fd,buf,1)) == 1) {
iojob *j;
listNode *ln;
redisLog(REDIS_DEBUG,"Processing I/O completed job");
/* Get the processed element (the oldest one) */
lockThreadedIO();
redisAssert(listLength(server.io_processed) != 0);
if (toprocess == -1) {
toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
if (toprocess <= 0) toprocess = 1;
}
ln = listFirst(server.io_processed);
j = ln->value;
listDelNode(server.io_processed,ln);
unlockThreadedIO();
/* Post process it in the main thread, as there are things we
* can do just here to avoid race conditions and/or invasive locks */
redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
(j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
(unsigned char*)j->key->ptr);
if (j->type == REDIS_IOJOB_LOAD) {
/* Create the key-value pair in the in-memory database */
if (j->val != NULL) {
dbAdd(j->db,j->key,j->val);
incrRefCount(j->val);
if (j->expire != -1) setExpire(j->db,j->key,j->expire);
} else {
/* The key does not exist. Create a negative cache entry
* for this key. */
/* FIXME: add this entry into the negative cache */
}
/* Handle clients waiting for this key to be loaded. */
handleClientsBlockedOnSwappedKey(j->db,j->key);
freeIOJob(j);
} else if (j->type == REDIS_IOJOB_SAVE) {
redisAssert(j->val->storage == REDIS_DS_SAVING);
j->val->storage = REDIS_DS_MEMORY;
freeIOJob(j);
}
processed++;
if (processed == toprocess) return;
}
if (retval < 0 && errno != EAGAIN) {
redisLog(REDIS_WARNING,
"WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
strerror(errno));
}
}
void lockThreadedIO(void) {
pthread_mutex_lock(&server.io_mutex);
}
void unlockThreadedIO(void) {
pthread_mutex_unlock(&server.io_mutex);
}
void *IOThreadEntryPoint(void *arg) {
iojob *j;
listNode *ln;
REDIS_NOTUSED(arg);
pthread_detach(pthread_self());
while(1) {
/* Get a new job to process */
lockThreadedIO();
if (listLength(server.io_newjobs) == 0) {
/* No new jobs in queue, exit. */
unlockThreadedIO();
sleep(1);
continue;
}
ln = listFirst(server.io_newjobs);
j = ln->value;
listDelNode(server.io_newjobs,ln);
/* Add the job in the processing queue */
listAddNodeTail(server.io_processing,j);
ln = listLast(server.io_processing); /* We use ln later to remove it */
unlockThreadedIO();
redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'",
(long) pthread_self(),
(j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
(void*)j, (char*)j->key->ptr);
/* Process the Job */
if (j->type == REDIS_IOJOB_LOAD) {
time_t expire;
j->val = dsGet(j->db,j->key,&expire);
if (j->val) j->expire = expire;
} else if (j->type == REDIS_IOJOB_SAVE) {
redisAssert(j->val->storage == REDIS_DS_SAVING);
if (j->val)
dsSet(j->db,j->key,j->val);
else
dsDel(j->db,j->key);
}
/* Done: insert the job into the processed queue */
redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
(long) pthread_self(), (void*)j, (char*)j->key->ptr);
lockThreadedIO();
listDelNode(server.io_processing,ln);
listAddNodeTail(server.io_processed,j);
unlockThreadedIO();
/* Signal the main thread there is new stuff to process */
redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
}
return NULL; /* never reached */
}
void spawnIOThread(void) {
pthread_t thread;
sigset_t mask, omask;
int err;
sigemptyset(&mask);
sigaddset(&mask,SIGCHLD);
sigaddset(&mask,SIGHUP);
sigaddset(&mask,SIGPIPE);
pthread_sigmask(SIG_SETMASK, &mask, &omask);
while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
strerror(err));
usleep(1000000);
}
pthread_sigmask(SIG_SETMASK, &omask, 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. */
void waitEmptyIOJobsQueue(void) {
while(1) {
int io_processed_len;
lockThreadedIO();
if (listLength(server.io_newjobs) == 0 &&
listLength(server.io_processing) == 0 &&
server.io_active_threads == 0)
{
unlockThreadedIO();
return;
}
/* While waiting for empty jobs queue condition we post-process some
* finshed job, as I/O threads may be hanging trying to write against
* the io_ready_pipe_write FD but there are so much pending jobs that
* it's blocking. */
io_processed_len = listLength(server.io_processed);
unlockThreadedIO();
if (io_processed_len) {
vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
(void*)0xdeadbeef,0);
usleep(1000); /* 1 millisecond */
} else {
usleep(10000); /* 10 milliseconds */
}
}
}
/* This function must be called while with threaded IO locked */
void queueIOJob(iojob *j) {
redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
(void*)j, j->type, (char*)j->key->ptr);
listAddNodeTail(server.io_newjobs,j);
if (server.io_active_threads < server.vm_max_threads)
spawnIOThread();
}
void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
iojob *j;
j = zmalloc(sizeof(*j));
j->type = type;
j->db = db;
j->key = key;
incrRefCount(key);
j->val = val;
if (val) incrRefCount(val);
lockThreadedIO();
queueIOJob(j);
unlockThreadedIO();
}
void cacheScheduleForFlush(redisDb *db, robj *key) {
dirtykey *dk;
dictEntry *de;
de = dictFind(db->dict,key->ptr);
if (de) {
robj *val = dictGetEntryVal(de);
if (val->storage == REDIS_DS_DIRTY)
return;
else
val->storage = REDIS_DS_DIRTY;
}
redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
dk = zmalloc(sizeof(*dk));
dk->db = db;
dk->key = key;
incrRefCount(key);
dk->ctime = time(NULL);
listAddNodeTail(server.cache_flush_queue, key);
}
void cacheCron(void) {
time_t now = time(NULL);
listNode *ln;
/* Sync stuff on disk */
while((ln = listFirst(server.cache_flush_queue)) != NULL) {
dirtykey *dk = ln->value;
if ((now - dk->ctime) >= server.cache_flush_delay) {
struct dictEntry *de;
robj *val;
redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);
/* Lookup the key. We need to check if it's still here and
* possibly access to the value. */
de = dictFind(dk->db->dict,dk->key->ptr);
if (de) {
val = dictGetEntryVal(de);
redisAssert(val->storage == REDIS_DS_DIRTY);
val->storage = REDIS_DS_SAVING;
} else {
/* Setting the value to NULL tells the IO thread to delete
* the key on disk. */
val = NULL;
}
dsCreateIOJob(REDIS_IOJOB_SAVE,dk->db,dk->key,val);
listDelNode(server.cache_flush_queue,ln);
} else {
break; /* too early */
}
}
/* Reclaim memory from the object cache */
while (server.ds_enabled && zmalloc_used_memory() >
server.cache_max_memory)
{
if (cacheFreeOneEntry() == REDIS_ERR) break;
}
}
/* ============ Virtual Memory - Blocking clients on missing keys =========== */
/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
* If the key is already in memory we don't need to block, regardless
* of the storage of the value object for this key:
*
* - If it's REDIS_DS_MEMORY we have the key in memory.
* - If it's REDIS_DS_DIRTY they key was modified, but still in memory.
* - if it's REDIS_DS_SAVING the key is being saved by an IO Job. When
* the client will lookup the key it will block if the key is still
* in this stage but it's more or less the best we can do.
*
* FIXME: we should try if it's actually better to suspend the client
* accessing an object that is being saved, and awake it only when
* the saving was completed.
*
* Otherwise if the key is not in memory, we block the client and start
* an IO Job to load it:
*
* the key is added to the io_keys list in the client structure, and also
* in the hash table mapping swapped keys to waiting clients, that is,
* server.io_waited_keys. */
int waitForSwappedKey(redisClient *c, robj *key) {
struct dictEntry *de;
list *l;
/* Return ASAP if the key is in memory */
de = dictFind(c->db->dict,key->ptr);
if (de != NULL) return 0;
/* Add the key to the list of keys this client is waiting for.
* This maps clients to keys they are waiting for. */
listAddNodeTail(c->io_keys,key);
incrRefCount(key);
/* Add the client to the swapped keys => clients waiting map. */
de = dictFind(c->db->io_keys,key);
if (de == NULL) {
int retval;
/* For every key we take a list of clients blocked for it */
l = listCreate();
retval = dictAdd(c->db->io_keys,key,l);
incrRefCount(key);
redisAssert(retval == DICT_OK);
} else {
l = dictGetEntryVal(de);
}
listAddNodeTail(l,c);
/* Are we already loading the key from disk? If not create a job */
if (de == NULL)
dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);
return 1;
}
/* Preload keys for any command with first, last and step values for
* the command keys prototype, as defined in the command table. */
void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int j, last;
if (cmd->vm_firstkey == 0) return;
last = cmd->vm_lastkey;
if (last < 0) last = argc+last;
for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) {
redisAssert(j < argc);
waitForSwappedKey(c,argv[j]);
}
}
/* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
* Note that the number of keys to preload is user-defined, so we need to
* apply a sanity check against argc. */
void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int i, num;
REDIS_NOTUSED(cmd);
num = atoi(argv[2]->ptr);
if (num > (argc-3)) return;
for (i = 0; i < num; i++) {
waitForSwappedKey(c,argv[3+i]);
}
}
/* Preload keys needed to execute the entire MULTI/EXEC block.
*
* This function is called by blockClientOnSwappedKeys when EXEC is issued,
* and will block the client when any command requires a swapped out value. */
void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int i, margc;
struct redisCommand *mcmd;
robj **margv;
REDIS_NOTUSED(cmd);
REDIS_NOTUSED(argc);
REDIS_NOTUSED(argv);
if (!(c->flags & REDIS_MULTI)) return;
for (i = 0; i < c->mstate.count; i++) {
mcmd = c->mstate.commands[i].cmd;
margc = c->mstate.commands[i].argc;
margv = c->mstate.commands[i].argv;
if (mcmd->vm_preload_proc != NULL) {
mcmd->vm_preload_proc(c,mcmd,margc,margv);
} else {
waitForMultipleSwappedKeys(c,mcmd,margc,margv);
}
}
}
/* Is this client attempting to run a command against swapped keys?
* If so, block it ASAP, load the keys in background, then resume it.
*
* The important idea about this function is that it can fail! If keys will
* still be swapped when the client is resumed, this key lookups will
* just block loading keys from disk. In practical terms this should only
* happen with SORT BY command or if there is a bug in this function.
*
* Return 1 if the client is marked as blocked, 0 if the client can
* continue as the keys it is going to access appear to be in memory. */
int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
if (cmd->vm_preload_proc != NULL) {
cmd->vm_preload_proc(c,cmd,c->argc,c->argv);
} else {
waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv);
}
/* If the client was blocked for at least one key, mark it as blocked. */
if (listLength(c->io_keys)) {
c->flags |= REDIS_IO_WAIT;
aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
server.cache_blocked_clients++;
return 1;
} else {
return 0;
}
}
/* Remove the 'key' from the list of blocked keys for a given client.
*
* The function returns 1 when there are no longer blocking keys after
* the current one was removed (and the client can be unblocked). */
int dontWaitForSwappedKey(redisClient *c, robj *key) {
list *l;
listNode *ln;
listIter li;
struct dictEntry *de;
/* The key object might be destroyed when deleted from the c->io_keys
* list (and the "key" argument is physically the same object as the
* object inside the list), so we need to protect it. */
incrRefCount(key);
/* Remove the key from the list of keys this client is waiting for. */
listRewind(c->io_keys,&li);
while ((ln = listNext(&li)) != NULL) {
if (equalStringObjects(ln->value,key)) {
listDelNode(c->io_keys,ln);
break;
}
}
redisAssert(ln != NULL);
/* Remove the client form the key => waiting clients map. */
de = dictFind(c->db->io_keys,key);
redisAssert(de != NULL);
l = dictGetEntryVal(de);
ln = listSearchKey(l,c);
redisAssert(ln != NULL);
listDelNode(l,ln);
if (listLength(l) == 0)
dictDelete(c->db->io_keys,key);
decrRefCount(key);
return listLength(c->io_keys) == 0;
}
/* Every time we now a key was loaded back in memory, we handle clients
* waiting for this key if any. */
void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
struct dictEntry *de;
list *l;
listNode *ln;
int len;
de = dictFind(db->io_keys,key);
if (!de) return;
l = dictGetEntryVal(de);
len = listLength(l);
/* Note: we can't use something like while(listLength(l)) as the list
* can be freed by the calling function when we remove the last element. */
while (len--) {
ln = listFirst(l);
redisClient *c = ln->value;
if (dontWaitForSwappedKey(c,key)) {
/* Put the client in the list of clients ready to go as we
* loaded all the keys about it. */
listAddNodeTail(server.io_ready_clients,c);
}
}
}