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