8 /* ---------------------------------- MASTER -------------------------------- */
10 void replicationFeedSlaves(list
*slaves
, int dictid
, robj
**argv
, int argc
) {
15 listRewind(slaves
,&li
);
16 while((ln
= listNext(&li
))) {
17 redisClient
*slave
= ln
->value
;
19 /* Don't feed slaves that are still waiting for BGSAVE to start */
20 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_START
) continue;
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
) {
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;
40 selectcmd
= createObject(REDIS_STRING
,
41 sdscatprintf(sdsempty(),"select %d\r\n",dictid
));
42 selectcmd
->refcount
= 0;
45 addReply(slave
,selectcmd
);
46 slave
->slaveseldb
= dictid
;
48 addReplyMultiBulkLen(slave
,argc
);
49 for (j
= 0; j
< argc
; j
++) addReplyBulk(slave
,argv
[j
]);
53 void replicationFeedMonitors(redisClient
*c
, list
*monitors
, int dictid
, robj
**argv
, int argc
) {
57 sds cmdrepr
= sdsnew("+");
62 gettimeofday(&tv
,NULL
);
63 cmdrepr
= sdscatprintf(cmdrepr
,"%ld.%06ld ",(long)tv
.tv_sec
,(long)tv
.tv_usec
);
64 if (c
->flags
& REDIS_LUA_CLIENT
) {
65 cmdrepr
= sdscatprintf(cmdrepr
,"[%d lua] ", dictid
);
67 anetPeerToString(c
->fd
,ip
,&port
);
68 cmdrepr
= sdscatprintf(cmdrepr
,"[%d %s:%d] ", dictid
,ip
,port
);
71 for (j
= 0; j
< argc
; j
++) {
72 if (argv
[j
]->encoding
== REDIS_ENCODING_INT
) {
73 cmdrepr
= sdscatprintf(cmdrepr
, "\"%ld\"", (long)argv
[j
]->ptr
);
75 cmdrepr
= sdscatrepr(cmdrepr
,(char*)argv
[j
]->ptr
,
76 sdslen(argv
[j
]->ptr
));
79 cmdrepr
= sdscatlen(cmdrepr
," ",1);
81 cmdrepr
= sdscatlen(cmdrepr
,"\r\n",2);
82 cmdobj
= createObject(REDIS_STRING
,cmdrepr
);
84 listRewind(monitors
,&li
);
85 while((ln
= listNext(&li
))) {
86 redisClient
*monitor
= ln
->value
;
87 addReply(monitor
,cmdobj
);
92 void syncCommand(redisClient
*c
) {
93 /* ignore SYNC if aleady slave or in monitor mode */
94 if (c
->flags
& REDIS_SLAVE
) return;
96 /* Refuse SYNC requests if we are a slave but the link with our master
98 if (server
.masterhost
&& server
.repl_state
!= REDIS_REPL_CONNECTED
) {
99 addReplyError(c
,"Can't SYNC while not connected with my master");
103 /* SYNC can't be issued when the server has pending data to send to
104 * the client about already issued commands. We need a fresh reply
105 * buffer registering the differences between the BGSAVE and the current
106 * dataset, so that we can copy to other slaves if needed. */
107 if (listLength(c
->reply
) != 0) {
108 addReplyError(c
,"SYNC is invalid with pending input");
112 redisLog(REDIS_NOTICE
,"Slave ask for synchronization");
113 /* Here we need to check if there is a background saving operation
114 * in progress, or if it is required to start one */
115 if (server
.rdb_child_pid
!= -1) {
116 /* Ok a background save is in progress. Let's check if it is a good
117 * one for replication, i.e. if there is another slave that is
118 * registering differences since the server forked to save */
123 listRewind(server
.slaves
,&li
);
124 while((ln
= listNext(&li
))) {
126 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_END
) break;
129 /* Perfect, the server is already registering differences for
130 * another slave. Set the right state, and copy the buffer. */
131 copyClientOutputBuffer(c
,slave
);
132 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
133 redisLog(REDIS_NOTICE
,"Waiting for end of BGSAVE for SYNC");
135 /* No way, we need to wait for the next BGSAVE in order to
136 * register differences */
137 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_START
;
138 redisLog(REDIS_NOTICE
,"Waiting for next BGSAVE for SYNC");
141 /* Ok we don't have a BGSAVE in progress, let's start one */
142 redisLog(REDIS_NOTICE
,"Starting BGSAVE for SYNC");
143 if (rdbSaveBackground(server
.rdb_filename
) != REDIS_OK
) {
144 redisLog(REDIS_NOTICE
,"Replication failed, can't BGSAVE");
145 addReplyError(c
,"Unable to perform background save");
148 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
151 c
->flags
|= REDIS_SLAVE
;
153 listAddNodeTail(server
.slaves
,c
);
157 void sendBulkToSlave(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
158 redisClient
*slave
= privdata
;
161 char buf
[REDIS_IOBUF_LEN
];
162 ssize_t nwritten
, buflen
;
164 if (slave
->repldboff
== 0) {
165 /* Write the bulk write count before to transfer the DB. In theory here
166 * we don't know how much room there is in the output buffer of the
167 * socket, but in pratice SO_SNDLOWAT (the minimum count for output
168 * operations) will never be smaller than the few bytes we need. */
171 bulkcount
= sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
173 if (write(fd
,bulkcount
,sdslen(bulkcount
)) != (signed)sdslen(bulkcount
))
181 lseek(slave
->repldbfd
,slave
->repldboff
,SEEK_SET
);
182 buflen
= read(slave
->repldbfd
,buf
,REDIS_IOBUF_LEN
);
184 redisLog(REDIS_WARNING
,"Read error sending DB to slave: %s",
185 (buflen
== 0) ? "premature EOF" : strerror(errno
));
189 if ((nwritten
= write(fd
,buf
,buflen
)) == -1) {
190 redisLog(REDIS_VERBOSE
,"Write error sending DB to slave: %s",
195 slave
->repldboff
+= nwritten
;
196 if (slave
->repldboff
== slave
->repldbsize
) {
197 close(slave
->repldbfd
);
198 slave
->repldbfd
= -1;
199 aeDeleteFileEvent(server
.el
,slave
->fd
,AE_WRITABLE
);
200 slave
->replstate
= REDIS_REPL_ONLINE
;
201 if (aeCreateFileEvent(server
.el
, slave
->fd
, AE_WRITABLE
,
202 sendReplyToClient
, slave
) == AE_ERR
) {
206 addReplySds(slave
,sdsempty());
207 redisLog(REDIS_NOTICE
,"Synchronization with slave succeeded");
211 /* This function is called at the end of every backgrond saving.
212 * The argument bgsaveerr is REDIS_OK if the background saving succeeded
213 * otherwise REDIS_ERR is passed to the function.
215 * The goal of this function is to handle slaves waiting for a successful
216 * background saving in order to perform non-blocking synchronization. */
217 void updateSlavesWaitingBgsave(int bgsaveerr
) {
222 listRewind(server
.slaves
,&li
);
223 while((ln
= listNext(&li
))) {
224 redisClient
*slave
= ln
->value
;
226 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_START
) {
228 slave
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
229 } else if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_END
) {
230 struct redis_stat buf
;
232 if (bgsaveerr
!= REDIS_OK
) {
234 redisLog(REDIS_WARNING
,"SYNC failed. BGSAVE child returned an error");
237 if ((slave
->repldbfd
= open(server
.rdb_filename
,O_RDONLY
)) == -1 ||
238 redis_fstat(slave
->repldbfd
,&buf
) == -1) {
240 redisLog(REDIS_WARNING
,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno
));
243 slave
->repldboff
= 0;
244 slave
->repldbsize
= buf
.st_size
;
245 slave
->replstate
= REDIS_REPL_SEND_BULK
;
246 aeDeleteFileEvent(server
.el
,slave
->fd
,AE_WRITABLE
);
247 if (aeCreateFileEvent(server
.el
, slave
->fd
, AE_WRITABLE
, sendBulkToSlave
, slave
) == AE_ERR
) {
254 if (rdbSaveBackground(server
.rdb_filename
) != REDIS_OK
) {
257 listRewind(server
.slaves
,&li
);
258 redisLog(REDIS_WARNING
,"SYNC failed. BGSAVE failed");
259 while((ln
= listNext(&li
))) {
260 redisClient
*slave
= ln
->value
;
262 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_START
)
269 /* ----------------------------------- SLAVE -------------------------------- */
271 /* Abort the async download of the bulk dataset while SYNC-ing with master */
272 void replicationAbortSyncTransfer(void) {
273 redisAssert(server
.repl_state
== REDIS_REPL_TRANSFER
);
275 aeDeleteFileEvent(server
.el
,server
.repl_transfer_s
,AE_READABLE
);
276 close(server
.repl_transfer_s
);
277 close(server
.repl_transfer_fd
);
278 unlink(server
.repl_transfer_tmpfile
);
279 zfree(server
.repl_transfer_tmpfile
);
280 server
.repl_state
= REDIS_REPL_CONNECT
;
283 /* Asynchronously read the SYNC payload we receive from a master */
284 void readSyncBulkPayload(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
286 ssize_t nread
, readlen
;
288 REDIS_NOTUSED(privdata
);
291 /* If repl_transfer_left == -1 we still have to read the bulk length
292 * from the master reply. */
293 if (server
.repl_transfer_left
== -1) {
294 if (syncReadLine(fd
,buf
,1024,server
.repl_syncio_timeout
) == -1) {
295 redisLog(REDIS_WARNING
,
296 "I/O error reading bulk count from MASTER: %s",
302 redisLog(REDIS_WARNING
,
303 "MASTER aborted replication with an error: %s",
306 } else if (buf
[0] == '\0') {
307 /* At this stage just a newline works as a PING in order to take
308 * the connection live. So we refresh our last interaction
310 server
.repl_transfer_lastio
= time(NULL
);
312 } else if (buf
[0] != '$') {
313 redisLog(REDIS_WARNING
,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
316 server
.repl_transfer_left
= strtol(buf
+1,NULL
,10);
317 redisLog(REDIS_NOTICE
,
318 "MASTER <-> SLAVE sync: receiving %ld bytes from master",
319 server
.repl_transfer_left
);
324 readlen
= (server
.repl_transfer_left
< (signed)sizeof(buf
)) ?
325 server
.repl_transfer_left
: (signed)sizeof(buf
);
326 nread
= read(fd
,buf
,readlen
);
328 redisLog(REDIS_WARNING
,"I/O error trying to sync with MASTER: %s",
329 (nread
== -1) ? strerror(errno
) : "connection lost");
330 replicationAbortSyncTransfer();
333 server
.repl_transfer_lastio
= time(NULL
);
334 if (write(server
.repl_transfer_fd
,buf
,nread
) != nread
) {
335 redisLog(REDIS_WARNING
,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno
));
338 server
.repl_transfer_left
-= nread
;
339 /* Check if the transfer is now complete */
340 if (server
.repl_transfer_left
== 0) {
341 if (rename(server
.repl_transfer_tmpfile
,server
.rdb_filename
) == -1) {
342 redisLog(REDIS_WARNING
,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno
));
343 replicationAbortSyncTransfer();
346 redisLog(REDIS_NOTICE
, "MASTER <-> SLAVE sync: Loading DB in memory");
348 /* Before loading the DB into memory we need to delete the readable
349 * handler, otherwise it will get called recursively since
350 * rdbLoad() will call the event loop to process events from time to
351 * time for non blocking loading. */
352 aeDeleteFileEvent(server
.el
,server
.repl_transfer_s
,AE_READABLE
);
353 if (rdbLoad(server
.rdb_filename
) != REDIS_OK
) {
354 redisLog(REDIS_WARNING
,"Failed trying to load the MASTER synchronization DB from disk");
355 replicationAbortSyncTransfer();
358 /* Final setup of the connected slave <- master link */
359 zfree(server
.repl_transfer_tmpfile
);
360 close(server
.repl_transfer_fd
);
361 server
.master
= createClient(server
.repl_transfer_s
);
362 server
.master
->flags
|= REDIS_MASTER
;
363 server
.master
->authenticated
= 1;
364 server
.repl_state
= REDIS_REPL_CONNECTED
;
365 redisLog(REDIS_NOTICE
, "MASTER <-> SLAVE sync: Finished with success");
366 /* Restart the AOF subsystem now that we finished the sync. This
367 * will trigger an AOF rewrite, and when done will start appending
368 * to the new file. */
369 if (server
.aof_state
!= REDIS_AOF_OFF
) {
373 while (retry
-- && startAppendOnly() == REDIS_ERR
) {
374 redisLog(REDIS_WARNING
,"Failed enabling the AOF after successful master synchrnization! Trying it again in one second.");
378 redisLog(REDIS_WARNING
,"FATAL: this slave instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
387 replicationAbortSyncTransfer();
391 void syncWithMaster(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
392 char buf
[1024], tmpfile
[256];
393 int dfd
, maxtries
= 5;
395 REDIS_NOTUSED(privdata
);
398 /* If this event fired after the user turned the instance into a master
399 * with SLAVEOF NO ONE we must just return ASAP. */
400 if (server
.repl_state
== REDIS_REPL_NONE
) {
405 redisLog(REDIS_NOTICE
,"Non blocking connect for SYNC fired the event.");
406 /* This event should only be triggered once since it is used to have a
407 * non-blocking connect(2) to the master. It has been triggered when this
408 * function is called, so we can delete it. */
409 aeDeleteFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
);
411 /* AUTH with the master if required. */
412 if(server
.masterauth
) {
416 authlen
= snprintf(authcmd
,sizeof(authcmd
),"AUTH %s\r\n",server
.masterauth
);
417 if (syncWrite(fd
,authcmd
,authlen
,server
.repl_syncio_timeout
) == -1) {
418 redisLog(REDIS_WARNING
,"Unable to AUTH to MASTER: %s",
422 /* Read the AUTH result. */
423 if (syncReadLine(fd
,buf
,1024,server
.repl_syncio_timeout
) == -1) {
424 redisLog(REDIS_WARNING
,"I/O error reading auth result from MASTER: %s",
429 redisLog(REDIS_WARNING
,"Cannot AUTH to MASTER, is the masterauth password correct?");
434 /* Issue the SYNC command */
435 if (syncWrite(fd
,"SYNC \r\n",7,server
.repl_syncio_timeout
) == -1) {
436 redisLog(REDIS_WARNING
,"I/O error writing to MASTER: %s",
441 /* Prepare a suitable temp file for bulk transfer */
443 snprintf(tmpfile
,256,
444 "temp-%d.%ld.rdb",(int)time(NULL
),(long int)getpid());
445 dfd
= open(tmpfile
,O_CREAT
|O_WRONLY
|O_EXCL
,0644);
446 if (dfd
!= -1) break;
450 redisLog(REDIS_WARNING
,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno
));
454 /* Setup the non blocking download of the bulk file. */
455 if (aeCreateFileEvent(server
.el
,fd
, AE_READABLE
,readSyncBulkPayload
,NULL
)
458 redisLog(REDIS_WARNING
,"Can't create readable event for SYNC");
462 server
.repl_state
= REDIS_REPL_TRANSFER
;
463 server
.repl_transfer_left
= -1;
464 server
.repl_transfer_fd
= dfd
;
465 server
.repl_transfer_lastio
= time(NULL
);
466 server
.repl_transfer_tmpfile
= zstrdup(tmpfile
);
470 server
.repl_state
= REDIS_REPL_CONNECT
;
475 int connectWithMaster(void) {
478 fd
= anetTcpNonBlockConnect(NULL
,server
.masterhost
,server
.masterport
);
480 redisLog(REDIS_WARNING
,"Unable to connect to MASTER: %s",
485 if (aeCreateFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
,syncWithMaster
,NULL
) ==
489 redisLog(REDIS_WARNING
,"Can't create readable event for SYNC");
493 server
.repl_transfer_lastio
= time(NULL
);
494 server
.repl_transfer_s
= fd
;
495 server
.repl_state
= REDIS_REPL_CONNECTING
;
499 /* This function can be called when a non blocking connection is currently
500 * in progress to undo it. */
501 void undoConnectWithMaster(void) {
502 int fd
= server
.repl_transfer_s
;
504 redisAssert(server
.repl_state
== REDIS_REPL_CONNECTING
);
505 aeDeleteFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
);
507 server
.repl_transfer_s
= -1;
508 server
.repl_state
= REDIS_REPL_CONNECT
;
511 void slaveofCommand(redisClient
*c
) {
512 if (!strcasecmp(c
->argv
[1]->ptr
,"no") &&
513 !strcasecmp(c
->argv
[2]->ptr
,"one")) {
514 if (server
.masterhost
) {
515 sdsfree(server
.masterhost
);
516 server
.masterhost
= NULL
;
517 if (server
.master
) freeClient(server
.master
);
518 if (server
.repl_state
== REDIS_REPL_TRANSFER
)
519 replicationAbortSyncTransfer();
520 else if (server
.repl_state
== REDIS_REPL_CONNECTING
)
521 undoConnectWithMaster();
522 server
.repl_state
= REDIS_REPL_NONE
;
523 redisLog(REDIS_NOTICE
,"MASTER MODE enabled (user request)");
528 if ((getLongFromObjectOrReply(c
, c
->argv
[2], &port
, NULL
) != REDIS_OK
))
531 /* Check if we are already attached to the specified slave */
532 if (server
.masterhost
&& !strcasecmp(server
.masterhost
,c
->argv
[1]->ptr
)
533 && server
.masterport
== port
) {
534 redisLog(REDIS_NOTICE
,"SLAVE OF would result into synchronization with the master we are already connected with. No operation performed.");
535 addReplySds(c
,sdsnew("+OK Already connected to specified master\r\n"));
538 /* There was no previous master or the user specified a different one,
539 * we can continue. */
540 sdsfree(server
.masterhost
);
541 server
.masterhost
= sdsdup(c
->argv
[1]->ptr
);
542 server
.masterport
= port
;
543 if (server
.master
) freeClient(server
.master
);
544 if (server
.repl_state
== REDIS_REPL_TRANSFER
)
545 replicationAbortSyncTransfer();
546 server
.repl_state
= REDIS_REPL_CONNECT
;
547 redisLog(REDIS_NOTICE
,"SLAVE OF %s:%d enabled (user request)",
548 server
.masterhost
, server
.masterport
);
550 addReply(c
,shared
.ok
);
553 /* --------------------------- REPLICATION CRON ---------------------------- */
555 void replicationCron(void) {
556 /* Non blocking connection timeout? */
557 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_CONNECTING
&&
558 (time(NULL
)-server
.repl_transfer_lastio
) > server
.repl_timeout
)
560 redisLog(REDIS_WARNING
,"Timeout connecting to the MASTER...");
561 undoConnectWithMaster();
564 /* Bulk transfer I/O timeout? */
565 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_TRANSFER
&&
566 (time(NULL
)-server
.repl_transfer_lastio
) > server
.repl_timeout
)
568 redisLog(REDIS_WARNING
,"Timeout receiving bulk data from MASTER...");
569 replicationAbortSyncTransfer();
572 /* Timed out master when we are an already connected slave? */
573 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_CONNECTED
&&
574 (time(NULL
)-server
.master
->lastinteraction
) > server
.repl_timeout
)
576 redisLog(REDIS_WARNING
,"MASTER time out: no data nor PING received...");
577 freeClient(server
.master
);
580 /* Check if we should connect to a MASTER */
581 if (server
.repl_state
== REDIS_REPL_CONNECT
) {
582 redisLog(REDIS_NOTICE
,"Connecting to MASTER...");
583 if (connectWithMaster() == REDIS_OK
) {
584 redisLog(REDIS_NOTICE
,"MASTER <-> SLAVE sync started");
588 /* If we have attached slaves, PING them from time to time.
589 * So slaves can implement an explicit timeout to masters, and will
590 * be able to detect a link disconnection even if the TCP connection
591 * will not actually go down. */
592 if (!(server
.cronloops
% (server
.repl_ping_slave_period
*10))) {
596 listRewind(server
.slaves
,&li
);
597 while((ln
= listNext(&li
))) {
598 redisClient
*slave
= ln
->value
;
600 /* Don't ping slaves that are in the middle of a bulk transfer
601 * with the master for first synchronization. */
602 if (slave
->replstate
== REDIS_REPL_SEND_BULK
) continue;
603 if (slave
->replstate
== REDIS_REPL_ONLINE
) {
604 /* If the slave is online send a normal ping */
605 addReplySds(slave
,sdsnew("*1\r\n$4\r\nPING\r\n"));
607 /* Otherwise we are in the pre-synchronization stage.
608 * Just a newline will do the work of refreshing the
609 * connection last interaction time, and at the same time
610 * we'll be sure that being a single char there are no
611 * short-write problems. */
612 if (write(slave
->fd
, "\n", 1) == -1) {
613 /* Don't worry, it's just a ping. */