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(list
*monitors
, int dictid
, robj
**argv
, int argc
) {
57 sds cmdrepr
= sdsnew("+");
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
);
65 for (j
= 0; j
< argc
; j
++) {
66 if (argv
[j
]->encoding
== REDIS_ENCODING_INT
) {
67 cmdrepr
= sdscatprintf(cmdrepr
, "\"%ld\"", (long)argv
[j
]->ptr
);
69 cmdrepr
= sdscatrepr(cmdrepr
,(char*)argv
[j
]->ptr
,
70 sdslen(argv
[j
]->ptr
));
73 cmdrepr
= sdscatlen(cmdrepr
," ",1);
75 cmdrepr
= sdscatlen(cmdrepr
,"\r\n",2);
76 cmdobj
= createObject(REDIS_STRING
,cmdrepr
);
78 listRewind(monitors
,&li
);
79 while((ln
= listNext(&li
))) {
80 redisClient
*monitor
= ln
->value
;
81 addReply(monitor
,cmdobj
);
86 void syncCommand(redisClient
*c
) {
87 /* ignore SYNC if aleady slave or in monitor mode */
88 if (c
->flags
& REDIS_SLAVE
) return;
90 /* Refuse SYNC requests if we are a slave but the link with our master
92 if (server
.masterhost
&& server
.repl_state
!= REDIS_REPL_CONNECTED
) {
93 addReplyError(c
,"Can't SYNC while not connected with my master");
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");
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
.rdb_child_pid
!= -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 */
117 listRewind(server
.slaves
,&li
);
118 while((ln
= listNext(&li
))) {
120 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_END
) break;
123 /* Perfect, the server is already registering differences for
124 * another slave. Set the right state, and copy the buffer. */
125 copyClientOutputBuffer(c
,slave
);
126 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
127 redisLog(REDIS_NOTICE
,"Waiting for end of BGSAVE for SYNC");
129 /* No way, we need to wait for the next BGSAVE in order to
130 * register differences */
131 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_START
;
132 redisLog(REDIS_NOTICE
,"Waiting for next BGSAVE for SYNC");
135 /* Ok we don't have a BGSAVE in progress, let's start one */
136 redisLog(REDIS_NOTICE
,"Starting BGSAVE for SYNC");
137 if (rdbSaveBackground(server
.rdb_filename
) != REDIS_OK
) {
138 redisLog(REDIS_NOTICE
,"Replication failed, can't BGSAVE");
139 addReplyError(c
,"Unable to perform background save");
142 c
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
145 c
->flags
|= REDIS_SLAVE
;
147 listAddNodeTail(server
.slaves
,c
);
151 void sendBulkToSlave(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
152 redisClient
*slave
= privdata
;
155 char buf
[REDIS_IOBUF_LEN
];
156 ssize_t nwritten
, buflen
;
158 if (slave
->repldboff
== 0) {
159 /* Write the bulk write count before to transfer the DB. In theory here
160 * we don't know how much room there is in the output buffer of the
161 * socket, but in pratice SO_SNDLOWAT (the minimum count for output
162 * operations) will never be smaller than the few bytes we need. */
165 bulkcount
= sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
167 if (write(fd
,bulkcount
,sdslen(bulkcount
)) != (signed)sdslen(bulkcount
))
175 lseek(slave
->repldbfd
,slave
->repldboff
,SEEK_SET
);
176 buflen
= read(slave
->repldbfd
,buf
,REDIS_IOBUF_LEN
);
178 redisLog(REDIS_WARNING
,"Read error sending DB to slave: %s",
179 (buflen
== 0) ? "premature EOF" : strerror(errno
));
183 if ((nwritten
= write(fd
,buf
,buflen
)) == -1) {
184 redisLog(REDIS_VERBOSE
,"Write error sending DB to slave: %s",
189 slave
->repldboff
+= nwritten
;
190 if (slave
->repldboff
== slave
->repldbsize
) {
191 close(slave
->repldbfd
);
192 slave
->repldbfd
= -1;
193 aeDeleteFileEvent(server
.el
,slave
->fd
,AE_WRITABLE
);
194 slave
->replstate
= REDIS_REPL_ONLINE
;
195 if (aeCreateFileEvent(server
.el
, slave
->fd
, AE_WRITABLE
,
196 sendReplyToClient
, slave
) == AE_ERR
) {
200 addReplySds(slave
,sdsempty());
201 redisLog(REDIS_NOTICE
,"Synchronization with slave succeeded");
205 /* This function is called at the end of every backgrond saving.
206 * The argument bgsaveerr is REDIS_OK if the background saving succeeded
207 * otherwise REDIS_ERR is passed to the function.
209 * The goal of this function is to handle slaves waiting for a successful
210 * background saving in order to perform non-blocking synchronization. */
211 void updateSlavesWaitingBgsave(int bgsaveerr
) {
216 listRewind(server
.slaves
,&li
);
217 while((ln
= listNext(&li
))) {
218 redisClient
*slave
= ln
->value
;
220 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_START
) {
222 slave
->replstate
= REDIS_REPL_WAIT_BGSAVE_END
;
223 } else if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_END
) {
224 struct redis_stat buf
;
226 if (bgsaveerr
!= REDIS_OK
) {
228 redisLog(REDIS_WARNING
,"SYNC failed. BGSAVE child returned an error");
231 if ((slave
->repldbfd
= open(server
.rdb_filename
,O_RDONLY
)) == -1 ||
232 redis_fstat(slave
->repldbfd
,&buf
) == -1) {
234 redisLog(REDIS_WARNING
,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno
));
237 slave
->repldboff
= 0;
238 slave
->repldbsize
= buf
.st_size
;
239 slave
->replstate
= REDIS_REPL_SEND_BULK
;
240 aeDeleteFileEvent(server
.el
,slave
->fd
,AE_WRITABLE
);
241 if (aeCreateFileEvent(server
.el
, slave
->fd
, AE_WRITABLE
, sendBulkToSlave
, slave
) == AE_ERR
) {
248 if (rdbSaveBackground(server
.rdb_filename
) != REDIS_OK
) {
251 listRewind(server
.slaves
,&li
);
252 redisLog(REDIS_WARNING
,"SYNC failed. BGSAVE failed");
253 while((ln
= listNext(&li
))) {
254 redisClient
*slave
= ln
->value
;
256 if (slave
->replstate
== REDIS_REPL_WAIT_BGSAVE_START
)
263 /* ----------------------------------- SLAVE -------------------------------- */
265 /* Abort the async download of the bulk dataset while SYNC-ing with master */
266 void replicationAbortSyncTransfer(void) {
267 redisAssert(server
.repl_state
== REDIS_REPL_TRANSFER
);
269 aeDeleteFileEvent(server
.el
,server
.repl_transfer_s
,AE_READABLE
);
270 close(server
.repl_transfer_s
);
271 close(server
.repl_transfer_fd
);
272 unlink(server
.repl_transfer_tmpfile
);
273 zfree(server
.repl_transfer_tmpfile
);
274 server
.repl_state
= REDIS_REPL_CONNECT
;
277 /* Asynchronously read the SYNC payload we receive from a master */
278 void readSyncBulkPayload(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
280 ssize_t nread
, readlen
;
282 REDIS_NOTUSED(privdata
);
285 /* If repl_transfer_left == -1 we still have to read the bulk length
286 * from the master reply. */
287 if (server
.repl_transfer_left
== -1) {
288 if (syncReadLine(fd
,buf
,1024,server
.repl_syncio_timeout
) == -1) {
289 redisLog(REDIS_WARNING
,
290 "I/O error reading bulk count from MASTER: %s",
296 redisLog(REDIS_WARNING
,
297 "MASTER aborted replication with an error: %s",
300 } else if (buf
[0] == '\0') {
301 /* At this stage just a newline works as a PING in order to take
302 * the connection live. So we refresh our last interaction
304 server
.repl_transfer_lastio
= time(NULL
);
306 } else if (buf
[0] != '$') {
307 redisLog(REDIS_WARNING
,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
310 server
.repl_transfer_left
= strtol(buf
+1,NULL
,10);
311 redisLog(REDIS_NOTICE
,
312 "MASTER <-> SLAVE sync: receiving %ld bytes from master",
313 server
.repl_transfer_left
);
318 readlen
= (server
.repl_transfer_left
< (signed)sizeof(buf
)) ?
319 server
.repl_transfer_left
: (signed)sizeof(buf
);
320 nread
= read(fd
,buf
,readlen
);
322 redisLog(REDIS_WARNING
,"I/O error trying to sync with MASTER: %s",
323 (nread
== -1) ? strerror(errno
) : "connection lost");
324 replicationAbortSyncTransfer();
327 server
.repl_transfer_lastio
= time(NULL
);
328 if (write(server
.repl_transfer_fd
,buf
,nread
) != nread
) {
329 redisLog(REDIS_WARNING
,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno
));
332 server
.repl_transfer_left
-= nread
;
333 /* Check if the transfer is now complete */
334 if (server
.repl_transfer_left
== 0) {
335 if (rename(server
.repl_transfer_tmpfile
,server
.rdb_filename
) == -1) {
336 redisLog(REDIS_WARNING
,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno
));
337 replicationAbortSyncTransfer();
340 redisLog(REDIS_NOTICE
, "MASTER <-> SLAVE sync: Loading DB in memory");
342 /* Before loading the DB into memory we need to delete the readable
343 * handler, otherwise it will get called recursively since
344 * rdbLoad() will call the event loop to process events from time to
345 * time for non blocking loading. */
346 aeDeleteFileEvent(server
.el
,server
.repl_transfer_s
,AE_READABLE
);
347 if (rdbLoad(server
.rdb_filename
) != REDIS_OK
) {
348 redisLog(REDIS_WARNING
,"Failed trying to load the MASTER synchronization DB from disk");
349 replicationAbortSyncTransfer();
352 /* Final setup of the connected slave <- master link */
353 zfree(server
.repl_transfer_tmpfile
);
354 close(server
.repl_transfer_fd
);
355 server
.master
= createClient(server
.repl_transfer_s
);
356 server
.master
->flags
|= REDIS_MASTER
;
357 server
.master
->authenticated
= 1;
358 server
.repl_state
= REDIS_REPL_CONNECTED
;
359 redisLog(REDIS_NOTICE
, "MASTER <-> SLAVE sync: Finished with success");
360 /* Restart the AOF subsystem now that we finished the sync. This
361 * will trigger an AOF rewrite, and when done will start appending
362 * to the new file. */
363 if (server
.aof_state
!= REDIS_AOF_OFF
) {
367 while (retry
-- && startAppendOnly() == REDIS_ERR
) {
368 redisLog(REDIS_WARNING
,"Failed enabling the AOF after successful master synchrnization! Trying it again in one second.");
372 redisLog(REDIS_WARNING
,"FATAL: this slave instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
381 replicationAbortSyncTransfer();
385 void syncWithMaster(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
386 char buf
[1024], tmpfile
[256];
387 int dfd
, maxtries
= 5;
389 REDIS_NOTUSED(privdata
);
392 /* If this event fired after the user turned the instance into a master
393 * with SLAVEOF NO ONE we must just return ASAP. */
394 if (server
.repl_state
== REDIS_REPL_NONE
) {
399 redisLog(REDIS_NOTICE
,"Non blocking connect for SYNC fired the event.");
400 /* This event should only be triggered once since it is used to have a
401 * non-blocking connect(2) to the master. It has been triggered when this
402 * function is called, so we can delete it. */
403 aeDeleteFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
);
405 /* AUTH with the master if required. */
406 if(server
.masterauth
) {
410 authlen
= snprintf(authcmd
,sizeof(authcmd
),"AUTH %s\r\n",server
.masterauth
);
411 if (syncWrite(fd
,authcmd
,authlen
,server
.repl_syncio_timeout
) == -1) {
412 redisLog(REDIS_WARNING
,"Unable to AUTH to MASTER: %s",
416 /* Read the AUTH result. */
417 if (syncReadLine(fd
,buf
,1024,server
.repl_syncio_timeout
) == -1) {
418 redisLog(REDIS_WARNING
,"I/O error reading auth result from MASTER: %s",
423 redisLog(REDIS_WARNING
,"Cannot AUTH to MASTER, is the masterauth password correct?");
428 /* Issue the SYNC command */
429 if (syncWrite(fd
,"SYNC \r\n",7,server
.repl_syncio_timeout
) == -1) {
430 redisLog(REDIS_WARNING
,"I/O error writing to MASTER: %s",
435 /* Prepare a suitable temp file for bulk transfer */
437 snprintf(tmpfile
,256,
438 "temp-%d.%ld.rdb",(int)time(NULL
),(long int)getpid());
439 dfd
= open(tmpfile
,O_CREAT
|O_WRONLY
|O_EXCL
,0644);
440 if (dfd
!= -1) break;
444 redisLog(REDIS_WARNING
,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno
));
448 /* Setup the non blocking download of the bulk file. */
449 if (aeCreateFileEvent(server
.el
,fd
, AE_READABLE
,readSyncBulkPayload
,NULL
)
452 redisLog(REDIS_WARNING
,"Can't create readable event for SYNC");
456 server
.repl_state
= REDIS_REPL_TRANSFER
;
457 server
.repl_transfer_left
= -1;
458 server
.repl_transfer_fd
= dfd
;
459 server
.repl_transfer_lastio
= time(NULL
);
460 server
.repl_transfer_tmpfile
= zstrdup(tmpfile
);
464 server
.repl_state
= REDIS_REPL_CONNECT
;
469 int connectWithMaster(void) {
472 fd
= anetTcpNonBlockConnect(NULL
,server
.masterhost
,server
.masterport
);
474 redisLog(REDIS_WARNING
,"Unable to connect to MASTER: %s",
479 if (aeCreateFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
,syncWithMaster
,NULL
) ==
483 redisLog(REDIS_WARNING
,"Can't create readable event for SYNC");
487 server
.repl_transfer_lastio
= time(NULL
);
488 server
.repl_transfer_s
= fd
;
489 server
.repl_state
= REDIS_REPL_CONNECTING
;
493 /* This function can be called when a non blocking connection is currently
494 * in progress to undo it. */
495 void undoConnectWithMaster(void) {
496 int fd
= server
.repl_transfer_s
;
498 redisAssert(server
.repl_state
== REDIS_REPL_CONNECTING
);
499 aeDeleteFileEvent(server
.el
,fd
,AE_READABLE
|AE_WRITABLE
);
501 server
.repl_transfer_s
= -1;
502 server
.repl_state
= REDIS_REPL_CONNECT
;
505 void slaveofCommand(redisClient
*c
) {
506 if (!strcasecmp(c
->argv
[1]->ptr
,"no") &&
507 !strcasecmp(c
->argv
[2]->ptr
,"one")) {
508 if (server
.masterhost
) {
509 sdsfree(server
.masterhost
);
510 server
.masterhost
= NULL
;
511 if (server
.master
) freeClient(server
.master
);
512 if (server
.repl_state
== REDIS_REPL_TRANSFER
)
513 replicationAbortSyncTransfer();
514 else if (server
.repl_state
== REDIS_REPL_CONNECTING
)
515 undoConnectWithMaster();
516 server
.repl_state
= REDIS_REPL_NONE
;
517 redisLog(REDIS_NOTICE
,"MASTER MODE enabled (user request)");
520 sdsfree(server
.masterhost
);
521 server
.masterhost
= sdsdup(c
->argv
[1]->ptr
);
522 server
.masterport
= atoi(c
->argv
[2]->ptr
);
523 if (server
.master
) freeClient(server
.master
);
524 if (server
.repl_state
== REDIS_REPL_TRANSFER
)
525 replicationAbortSyncTransfer();
526 server
.repl_state
= REDIS_REPL_CONNECT
;
527 redisLog(REDIS_NOTICE
,"SLAVE OF %s:%d enabled (user request)",
528 server
.masterhost
, server
.masterport
);
530 addReply(c
,shared
.ok
);
533 /* --------------------------- REPLICATION CRON ---------------------------- */
535 void replicationCron(void) {
536 /* Non blocking connection timeout? */
537 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_CONNECTING
&&
538 (time(NULL
)-server
.repl_transfer_lastio
) > server
.repl_timeout
)
540 redisLog(REDIS_WARNING
,"Timeout connecting to the MASTER...");
541 undoConnectWithMaster();
544 /* Bulk transfer I/O timeout? */
545 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_TRANSFER
&&
546 (time(NULL
)-server
.repl_transfer_lastio
) > server
.repl_timeout
)
548 redisLog(REDIS_WARNING
,"Timeout receiving bulk data from MASTER...");
549 replicationAbortSyncTransfer();
552 /* Timed out master when we are an already connected slave? */
553 if (server
.masterhost
&& server
.repl_state
== REDIS_REPL_CONNECTED
&&
554 (time(NULL
)-server
.master
->lastinteraction
) > server
.repl_timeout
)
556 redisLog(REDIS_WARNING
,"MASTER time out: no data nor PING received...");
557 freeClient(server
.master
);
560 /* Check if we should connect to a MASTER */
561 if (server
.repl_state
== REDIS_REPL_CONNECT
) {
562 redisLog(REDIS_NOTICE
,"Connecting to MASTER...");
563 if (connectWithMaster() == REDIS_OK
) {
564 redisLog(REDIS_NOTICE
,"MASTER <-> SLAVE sync started");
568 /* If we have attached slaves, PING them from time to time.
569 * So slaves can implement an explicit timeout to masters, and will
570 * be able to detect a link disconnection even if the TCP connection
571 * will not actually go down. */
572 if (!(server
.cronloops
% (server
.repl_ping_slave_period
*10))) {
576 listRewind(server
.slaves
,&li
);
577 while((ln
= listNext(&li
))) {
578 redisClient
*slave
= ln
->value
;
580 /* Don't ping slaves that are in the middle of a bulk transfer
581 * with the master for first synchronization. */
582 if (slave
->replstate
== REDIS_REPL_SEND_BULK
) continue;
583 if (slave
->replstate
== REDIS_REPL_ONLINE
) {
584 /* If the slave is online send a normal ping */
585 addReplySds(slave
,sdsnew("PING\r\n"));
587 /* Otherwise we are in the pre-synchronization stage.
588 * Just a newline will do the work of refreshing the
589 * connection last interaction time, and at the same time
590 * we'll be sure that being a single char there are no
591 * short-write problems. */
592 if (write(slave
->fd
, "\n", 1) == -1) {
593 /* Don't worry, it's just a ping. */