]> git.saurik.com Git - redis.git/blob - src/replication.c
Sorted Set type AOF rewrite using variadic ZADD.
[redis.git] / src / replication.c
1 #include "redis.h"
2
3 #include <sys/time.h>
4 #include <unistd.h>
5 #include <fcntl.h>
6 #include <sys/stat.h>
7
8 /* ---------------------------------- MASTER -------------------------------- */
9
10 void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
11 listNode *ln;
12 listIter li;
13 int j;
14
15 listRewind(slaves,&li);
16 while((ln = listNext(&li))) {
17 redisClient *slave = ln->value;
18
19 /* Don't feed slaves that are still waiting for BGSAVE to start */
20 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
21
22 /* Feed slaves that are waiting for the initial SYNC (so these commands
23 * are queued in the output buffer until the intial SYNC completes),
24 * or are already in sync with the master. */
25 if (slave->slaveseldb != dictid) {
26 robj *selectcmd;
27
28 switch(dictid) {
29 case 0: selectcmd = shared.select0; break;
30 case 1: selectcmd = shared.select1; break;
31 case 2: selectcmd = shared.select2; break;
32 case 3: selectcmd = shared.select3; break;
33 case 4: selectcmd = shared.select4; break;
34 case 5: selectcmd = shared.select5; break;
35 case 6: selectcmd = shared.select6; break;
36 case 7: selectcmd = shared.select7; break;
37 case 8: selectcmd = shared.select8; break;
38 case 9: selectcmd = shared.select9; break;
39 default:
40 selectcmd = createObject(REDIS_STRING,
41 sdscatprintf(sdsempty(),"select %d\r\n",dictid));
42 selectcmd->refcount = 0;
43 break;
44 }
45 addReply(slave,selectcmd);
46 slave->slaveseldb = dictid;
47 }
48 addReplyMultiBulkLen(slave,argc);
49 for (j = 0; j < argc; j++) addReplyBulk(slave,argv[j]);
50 }
51 }
52
53 void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) {
54 listNode *ln;
55 listIter li;
56 int j;
57 sds cmdrepr = sdsnew("+");
58 robj *cmdobj;
59 struct timeval tv;
60
61 gettimeofday(&tv,NULL);
62 cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
63 if (dictid != 0) cmdrepr = sdscatprintf(cmdrepr,"(db %d) ", dictid);
64
65 for (j = 0; j < argc; j++) {
66 if (argv[j]->encoding == REDIS_ENCODING_INT) {
67 cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
68 } else {
69 cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
70 sdslen(argv[j]->ptr));
71 }
72 if (j != argc-1)
73 cmdrepr = sdscatlen(cmdrepr," ",1);
74 }
75 cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
76 cmdobj = createObject(REDIS_STRING,cmdrepr);
77
78 listRewind(monitors,&li);
79 while((ln = listNext(&li))) {
80 redisClient *monitor = ln->value;
81 addReply(monitor,cmdobj);
82 }
83 decrRefCount(cmdobj);
84 }
85
86 void syncCommand(redisClient *c) {
87 /* ignore SYNC if aleady slave or in monitor mode */
88 if (c->flags & REDIS_SLAVE) return;
89
90 /* Refuse SYNC requests if we are a slave but the link with our master
91 * is not ok... */
92 if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED) {
93 addReplyError(c,"Can't SYNC while not connected with my master");
94 return;
95 }
96
97 /* SYNC can't be issued when the server has pending data to send to
98 * the client about already issued commands. We need a fresh reply
99 * buffer registering the differences between the BGSAVE and the current
100 * dataset, so that we can copy to other slaves if needed. */
101 if (listLength(c->reply) != 0) {
102 addReplyError(c,"SYNC is invalid with pending input");
103 return;
104 }
105
106 redisLog(REDIS_NOTICE,"Slave ask for synchronization");
107 /* Here we need to check if there is a background saving operation
108 * in progress, or if it is required to start one */
109 if (server.bgsavechildpid != -1) {
110 /* Ok a background save is in progress. Let's check if it is a good
111 * one for replication, i.e. if there is another slave that is
112 * registering differences since the server forked to save */
113 redisClient *slave;
114 listNode *ln;
115 listIter li;
116
117 listRewind(server.slaves,&li);
118 while((ln = listNext(&li))) {
119 slave = ln->value;
120 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
121 }
122 if (ln) {
123 /* Perfect, the server is already registering differences for
124 * another slave. Set the right state, and copy the buffer. */
125 listRelease(c->reply);
126 c->reply = listDup(slave->reply);
127 c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
128 redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
129 } else {
130 /* No way, we need to wait for the next BGSAVE in order to
131 * register differences */
132 c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
133 redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
134 }
135 } else {
136 /* Ok we don't have a BGSAVE in progress, let's start one */
137 redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
138 if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
139 redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
140 addReplyError(c,"Unable to perform background save");
141 return;
142 }
143 c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
144 }
145 c->repldbfd = -1;
146 c->flags |= REDIS_SLAVE;
147 c->slaveseldb = 0;
148 listAddNodeTail(server.slaves,c);
149 return;
150 }
151
152 void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
153 redisClient *slave = privdata;
154 REDIS_NOTUSED(el);
155 REDIS_NOTUSED(mask);
156 char buf[REDIS_IOBUF_LEN];
157 ssize_t nwritten, buflen;
158
159 if (slave->repldboff == 0) {
160 /* Write the bulk write count before to transfer the DB. In theory here
161 * we don't know how much room there is in the output buffer of the
162 * socket, but in pratice SO_SNDLOWAT (the minimum count for output
163 * operations) will never be smaller than the few bytes we need. */
164 sds bulkcount;
165
166 bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
167 slave->repldbsize);
168 if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
169 {
170 sdsfree(bulkcount);
171 freeClient(slave);
172 return;
173 }
174 sdsfree(bulkcount);
175 }
176 lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
177 buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
178 if (buflen <= 0) {
179 redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
180 (buflen == 0) ? "premature EOF" : strerror(errno));
181 freeClient(slave);
182 return;
183 }
184 if ((nwritten = write(fd,buf,buflen)) == -1) {
185 redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
186 strerror(errno));
187 freeClient(slave);
188 return;
189 }
190 slave->repldboff += nwritten;
191 if (slave->repldboff == slave->repldbsize) {
192 close(slave->repldbfd);
193 slave->repldbfd = -1;
194 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
195 slave->replstate = REDIS_REPL_ONLINE;
196 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
197 sendReplyToClient, slave) == AE_ERR) {
198 freeClient(slave);
199 return;
200 }
201 addReplySds(slave,sdsempty());
202 redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
203 }
204 }
205
206 /* This function is called at the end of every backgrond saving.
207 * The argument bgsaveerr is REDIS_OK if the background saving succeeded
208 * otherwise REDIS_ERR is passed to the function.
209 *
210 * The goal of this function is to handle slaves waiting for a successful
211 * background saving in order to perform non-blocking synchronization. */
212 void updateSlavesWaitingBgsave(int bgsaveerr) {
213 listNode *ln;
214 int startbgsave = 0;
215 listIter li;
216
217 listRewind(server.slaves,&li);
218 while((ln = listNext(&li))) {
219 redisClient *slave = ln->value;
220
221 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
222 startbgsave = 1;
223 slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
224 } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
225 struct redis_stat buf;
226
227 if (bgsaveerr != REDIS_OK) {
228 freeClient(slave);
229 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
230 continue;
231 }
232 if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
233 redis_fstat(slave->repldbfd,&buf) == -1) {
234 freeClient(slave);
235 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
236 continue;
237 }
238 slave->repldboff = 0;
239 slave->repldbsize = buf.st_size;
240 slave->replstate = REDIS_REPL_SEND_BULK;
241 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
242 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
243 freeClient(slave);
244 continue;
245 }
246 }
247 }
248 if (startbgsave) {
249 if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
250 listIter li;
251
252 listRewind(server.slaves,&li);
253 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
254 while((ln = listNext(&li))) {
255 redisClient *slave = ln->value;
256
257 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
258 freeClient(slave);
259 }
260 }
261 }
262 }
263
264 /* ----------------------------------- SLAVE -------------------------------- */
265
266 /* Abort the async download of the bulk dataset while SYNC-ing with master */
267 void replicationAbortSyncTransfer(void) {
268 redisAssert(server.replstate == REDIS_REPL_TRANSFER);
269
270 aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
271 close(server.repl_transfer_s);
272 close(server.repl_transfer_fd);
273 unlink(server.repl_transfer_tmpfile);
274 zfree(server.repl_transfer_tmpfile);
275 server.replstate = REDIS_REPL_CONNECT;
276 }
277
278 /* Asynchronously read the SYNC payload we receive from a master */
279 void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
280 char buf[4096];
281 ssize_t nread, readlen;
282 REDIS_NOTUSED(el);
283 REDIS_NOTUSED(privdata);
284 REDIS_NOTUSED(mask);
285
286 /* If repl_transfer_left == -1 we still have to read the bulk length
287 * from the master reply. */
288 if (server.repl_transfer_left == -1) {
289 if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
290 redisLog(REDIS_WARNING,
291 "I/O error reading bulk count from MASTER: %s",
292 strerror(errno));
293 goto error;
294 }
295
296 if (buf[0] == '-') {
297 redisLog(REDIS_WARNING,
298 "MASTER aborted replication with an error: %s",
299 buf+1);
300 goto error;
301 } else if (buf[0] == '\0') {
302 /* At this stage just a newline works as a PING in order to take
303 * the connection live. So we refresh our last interaction
304 * timestamp. */
305 server.repl_transfer_lastio = time(NULL);
306 return;
307 } else if (buf[0] != '$') {
308 redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
309 goto error;
310 }
311 server.repl_transfer_left = strtol(buf+1,NULL,10);
312 redisLog(REDIS_NOTICE,
313 "MASTER <-> SLAVE sync: receiving %ld bytes from master",
314 server.repl_transfer_left);
315 return;
316 }
317
318 /* Read bulk data */
319 readlen = (server.repl_transfer_left < (signed)sizeof(buf)) ?
320 server.repl_transfer_left : (signed)sizeof(buf);
321 nread = read(fd,buf,readlen);
322 if (nread <= 0) {
323 redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
324 (nread == -1) ? strerror(errno) : "connection lost");
325 replicationAbortSyncTransfer();
326 return;
327 }
328 server.repl_transfer_lastio = time(NULL);
329 if (write(server.repl_transfer_fd,buf,nread) != nread) {
330 redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
331 goto error;
332 }
333 server.repl_transfer_left -= nread;
334 /* Check if the transfer is now complete */
335 if (server.repl_transfer_left == 0) {
336 if (rename(server.repl_transfer_tmpfile,server.dbfilename) == -1) {
337 redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
338 replicationAbortSyncTransfer();
339 return;
340 }
341 redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
342 emptyDb();
343 /* Before loading the DB into memory we need to delete the readable
344 * handler, otherwise it will get called recursively since
345 * rdbLoad() will call the event loop to process events from time to
346 * time for non blocking loading. */
347 aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
348 if (rdbLoad(server.dbfilename) != REDIS_OK) {
349 redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
350 replicationAbortSyncTransfer();
351 return;
352 }
353 /* Final setup of the connected slave <- master link */
354 zfree(server.repl_transfer_tmpfile);
355 close(server.repl_transfer_fd);
356 server.master = createClient(server.repl_transfer_s);
357 server.master->flags |= REDIS_MASTER;
358 server.master->authenticated = 1;
359 server.replstate = REDIS_REPL_CONNECTED;
360 redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
361 /* Rewrite the AOF file now that the dataset changed. */
362 if (server.appendonly) rewriteAppendOnlyFileBackground();
363 }
364
365 return;
366
367 error:
368 replicationAbortSyncTransfer();
369 return;
370 }
371
372 void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
373 char buf[1024], tmpfile[256];
374 int dfd, maxtries = 5;
375 REDIS_NOTUSED(el);
376 REDIS_NOTUSED(privdata);
377 REDIS_NOTUSED(mask);
378
379 /* If this event fired after the user turned the instance into a master
380 * with SLAVEOF NO ONE we must just return ASAP. */
381 if (server.replstate == REDIS_REPL_NONE) {
382 close(fd);
383 return;
384 }
385
386 redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
387 /* This event should only be triggered once since it is used to have a
388 * non-blocking connect(2) to the master. It has been triggered when this
389 * function is called, so we can delete it. */
390 aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
391
392 /* AUTH with the master if required. */
393 if(server.masterauth) {
394 char authcmd[1024];
395 size_t authlen;
396
397 authlen = snprintf(authcmd,sizeof(authcmd),"AUTH %s\r\n",server.masterauth);
398 if (syncWrite(fd,authcmd,authlen,server.repl_syncio_timeout) == -1) {
399 redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
400 strerror(errno));
401 goto error;
402 }
403 /* Read the AUTH result. */
404 if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
405 redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
406 strerror(errno));
407 goto error;
408 }
409 if (buf[0] != '+') {
410 redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
411 goto error;
412 }
413 }
414
415 /* Issue the SYNC command */
416 if (syncWrite(fd,"SYNC \r\n",7,server.repl_syncio_timeout) == -1) {
417 redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
418 strerror(errno));
419 goto error;
420 }
421
422 /* Prepare a suitable temp file for bulk transfer */
423 while(maxtries--) {
424 snprintf(tmpfile,256,
425 "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
426 dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
427 if (dfd != -1) break;
428 sleep(1);
429 }
430 if (dfd == -1) {
431 redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
432 goto error;
433 }
434
435 /* Setup the non blocking download of the bulk file. */
436 if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
437 == AE_ERR)
438 {
439 redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
440 goto error;
441 }
442
443 server.replstate = REDIS_REPL_TRANSFER;
444 server.repl_transfer_left = -1;
445 server.repl_transfer_fd = dfd;
446 server.repl_transfer_lastio = time(NULL);
447 server.repl_transfer_tmpfile = zstrdup(tmpfile);
448 return;
449
450 error:
451 server.replstate = REDIS_REPL_CONNECT;
452 close(fd);
453 return;
454 }
455
456 int connectWithMaster(void) {
457 int fd;
458
459 fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport);
460 if (fd == -1) {
461 redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
462 strerror(errno));
463 return REDIS_ERR;
464 }
465
466 if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) ==
467 AE_ERR)
468 {
469 close(fd);
470 redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
471 return REDIS_ERR;
472 }
473
474 server.repl_transfer_lastio = time(NULL);
475 server.repl_transfer_s = fd;
476 server.replstate = REDIS_REPL_CONNECTING;
477 return REDIS_OK;
478 }
479
480 /* This function can be called when a non blocking connection is currently
481 * in progress to undo it. */
482 void undoConnectWithMaster(void) {
483 int fd = server.repl_transfer_s;
484
485 redisAssert(server.replstate == REDIS_REPL_CONNECTING);
486 aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
487 close(fd);
488 server.repl_transfer_s = -1;
489 server.replstate = REDIS_REPL_CONNECT;
490 }
491
492 void slaveofCommand(redisClient *c) {
493 if (!strcasecmp(c->argv[1]->ptr,"no") &&
494 !strcasecmp(c->argv[2]->ptr,"one")) {
495 if (server.masterhost) {
496 sdsfree(server.masterhost);
497 server.masterhost = NULL;
498 if (server.master) freeClient(server.master);
499 if (server.replstate == REDIS_REPL_TRANSFER)
500 replicationAbortSyncTransfer();
501 else if (server.replstate == REDIS_REPL_CONNECTING)
502 undoConnectWithMaster();
503 server.replstate = REDIS_REPL_NONE;
504 redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
505 }
506 } else {
507 sdsfree(server.masterhost);
508 server.masterhost = sdsdup(c->argv[1]->ptr);
509 server.masterport = atoi(c->argv[2]->ptr);
510 if (server.master) freeClient(server.master);
511 if (server.replstate == REDIS_REPL_TRANSFER)
512 replicationAbortSyncTransfer();
513 server.replstate = REDIS_REPL_CONNECT;
514 redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
515 server.masterhost, server.masterport);
516 }
517 addReply(c,shared.ok);
518 }
519
520 /* --------------------------- REPLICATION CRON ---------------------------- */
521
522 void replicationCron(void) {
523 /* Non blocking connection timeout? */
524 if (server.masterhost && server.replstate == REDIS_REPL_CONNECTING &&
525 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
526 {
527 redisLog(REDIS_WARNING,"Timeout connecting to the MASTER...");
528 undoConnectWithMaster();
529 }
530
531 /* Bulk transfer I/O timeout? */
532 if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
533 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
534 {
535 redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
536 replicationAbortSyncTransfer();
537 }
538
539 /* Timed out master when we are an already connected slave? */
540 if (server.masterhost && server.replstate == REDIS_REPL_CONNECTED &&
541 (time(NULL)-server.master->lastinteraction) > server.repl_timeout)
542 {
543 redisLog(REDIS_WARNING,"MASTER time out: no data nor PING received...");
544 freeClient(server.master);
545 }
546
547 /* Check if we should connect to a MASTER */
548 if (server.replstate == REDIS_REPL_CONNECT) {
549 redisLog(REDIS_NOTICE,"Connecting to MASTER...");
550 if (connectWithMaster() == REDIS_OK) {
551 redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started");
552 }
553 }
554
555 /* If we have attached slaves, PING them from time to time.
556 * So slaves can implement an explicit timeout to masters, and will
557 * be able to detect a link disconnection even if the TCP connection
558 * will not actually go down. */
559 if (!(server.cronloops % (server.repl_ping_slave_period*10))) {
560 listIter li;
561 listNode *ln;
562
563 listRewind(server.slaves,&li);
564 while((ln = listNext(&li))) {
565 redisClient *slave = ln->value;
566
567 /* Don't ping slaves that are in the middle of a bulk transfer
568 * with the master for first synchronization. */
569 if (slave->replstate == REDIS_REPL_SEND_BULK) continue;
570 if (slave->replstate == REDIS_REPL_ONLINE) {
571 /* If the slave is online send a normal ping */
572 addReplySds(slave,sdsnew("PING\r\n"));
573 } else {
574 /* Otherwise we are in the pre-synchronization stage.
575 * Just a newline will do the work of refreshing the
576 * connection last interaction time, and at the same time
577 * we'll be sure that being a single char there are no
578 * short-write problems. */
579 if (write(slave->fd, "\n", 1) == -1) {
580 /* Don't worry, it's just a ping. */
581 }
582 }
583 }
584 }
585 }