]>
Commit | Line | Data |
---|---|---|
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 | void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { | |
9 | listNode *ln; | |
10 | listIter li; | |
11 | int outc = 0, j; | |
12 | robj **outv; | |
13 | /* We need 1+(ARGS*3) objects since commands are using the new protocol | |
14 | * and we one 1 object for the first "*<count>\r\n" multibulk count, then | |
15 | * for every additional object we have "$<count>\r\n" + object + "\r\n". */ | |
16 | robj *static_outv[REDIS_STATIC_ARGS*3+1]; | |
17 | robj *lenobj; | |
18 | ||
19 | if (argc <= REDIS_STATIC_ARGS) { | |
20 | outv = static_outv; | |
21 | } else { | |
22 | outv = zmalloc(sizeof(robj*)*(argc*3+1)); | |
23 | } | |
24 | ||
25 | lenobj = createObject(REDIS_STRING, | |
26 | sdscatprintf(sdsempty(), "*%d\r\n", argc)); | |
27 | lenobj->refcount = 0; | |
28 | outv[outc++] = lenobj; | |
29 | for (j = 0; j < argc; j++) { | |
30 | lenobj = createObject(REDIS_STRING, | |
31 | sdscatprintf(sdsempty(),"$%lu\r\n", | |
32 | (unsigned long) stringObjectLen(argv[j]))); | |
33 | lenobj->refcount = 0; | |
34 | outv[outc++] = lenobj; | |
35 | outv[outc++] = argv[j]; | |
36 | outv[outc++] = shared.crlf; | |
37 | } | |
38 | ||
39 | /* Increment all the refcounts at start and decrement at end in order to | |
40 | * be sure to free objects if there is no slave in a replication state | |
41 | * able to be feed with commands */ | |
42 | for (j = 0; j < outc; j++) incrRefCount(outv[j]); | |
43 | listRewind(slaves,&li); | |
44 | while((ln = listNext(&li))) { | |
45 | redisClient *slave = ln->value; | |
46 | ||
47 | /* Don't feed slaves that are still waiting for BGSAVE to start */ | |
48 | if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue; | |
49 | ||
50 | /* Feed all the other slaves, MONITORs and so on */ | |
51 | if (slave->slaveseldb != dictid) { | |
52 | robj *selectcmd; | |
53 | ||
54 | switch(dictid) { | |
55 | case 0: selectcmd = shared.select0; break; | |
56 | case 1: selectcmd = shared.select1; break; | |
57 | case 2: selectcmd = shared.select2; break; | |
58 | case 3: selectcmd = shared.select3; break; | |
59 | case 4: selectcmd = shared.select4; break; | |
60 | case 5: selectcmd = shared.select5; break; | |
61 | case 6: selectcmd = shared.select6; break; | |
62 | case 7: selectcmd = shared.select7; break; | |
63 | case 8: selectcmd = shared.select8; break; | |
64 | case 9: selectcmd = shared.select9; break; | |
65 | default: | |
66 | selectcmd = createObject(REDIS_STRING, | |
67 | sdscatprintf(sdsempty(),"select %d\r\n",dictid)); | |
68 | selectcmd->refcount = 0; | |
69 | break; | |
70 | } | |
71 | addReply(slave,selectcmd); | |
72 | slave->slaveseldb = dictid; | |
73 | } | |
74 | for (j = 0; j < outc; j++) addReply(slave,outv[j]); | |
75 | } | |
76 | for (j = 0; j < outc; j++) decrRefCount(outv[j]); | |
77 | if (outv != static_outv) zfree(outv); | |
78 | } | |
79 | ||
80 | void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) { | |
81 | listNode *ln; | |
82 | listIter li; | |
83 | int j; | |
84 | sds cmdrepr = sdsnew("+"); | |
85 | robj *cmdobj; | |
86 | struct timeval tv; | |
87 | ||
88 | gettimeofday(&tv,NULL); | |
89 | cmdrepr = sdscatprintf(cmdrepr,"%ld.%ld ",(long)tv.tv_sec,(long)tv.tv_usec); | |
90 | if (dictid != 0) cmdrepr = sdscatprintf(cmdrepr,"(db %d) ", dictid); | |
91 | ||
92 | for (j = 0; j < argc; j++) { | |
93 | if (argv[j]->encoding == REDIS_ENCODING_INT) { | |
94 | cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr); | |
95 | } else { | |
96 | cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr, | |
97 | sdslen(argv[j]->ptr)); | |
98 | } | |
99 | if (j != argc-1) | |
100 | cmdrepr = sdscatlen(cmdrepr," ",1); | |
101 | } | |
102 | cmdrepr = sdscatlen(cmdrepr,"\r\n",2); | |
103 | cmdobj = createObject(REDIS_STRING,cmdrepr); | |
104 | ||
105 | listRewind(monitors,&li); | |
106 | while((ln = listNext(&li))) { | |
107 | redisClient *monitor = ln->value; | |
108 | addReply(monitor,cmdobj); | |
109 | } | |
110 | decrRefCount(cmdobj); | |
111 | } | |
112 | ||
113 | int syncWrite(int fd, char *ptr, ssize_t size, int timeout) { | |
114 | ssize_t nwritten, ret = size; | |
115 | time_t start = time(NULL); | |
116 | ||
117 | timeout++; | |
118 | while(size) { | |
119 | if (aeWait(fd,AE_WRITABLE,1000) & AE_WRITABLE) { | |
120 | nwritten = write(fd,ptr,size); | |
121 | if (nwritten == -1) return -1; | |
122 | ptr += nwritten; | |
123 | size -= nwritten; | |
124 | } | |
125 | if ((time(NULL)-start) > timeout) { | |
126 | errno = ETIMEDOUT; | |
127 | return -1; | |
128 | } | |
129 | } | |
130 | return ret; | |
131 | } | |
132 | ||
133 | int syncRead(int fd, char *ptr, ssize_t size, int timeout) { | |
134 | ssize_t nread, totread = 0; | |
135 | time_t start = time(NULL); | |
136 | ||
137 | timeout++; | |
138 | while(size) { | |
139 | if (aeWait(fd,AE_READABLE,1000) & AE_READABLE) { | |
140 | nread = read(fd,ptr,size); | |
141 | if (nread <= 0) return -1; | |
142 | ptr += nread; | |
143 | size -= nread; | |
144 | totread += nread; | |
145 | } | |
146 | if ((time(NULL)-start) > timeout) { | |
147 | errno = ETIMEDOUT; | |
148 | return -1; | |
149 | } | |
150 | } | |
151 | return totread; | |
152 | } | |
153 | ||
154 | int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) { | |
155 | ssize_t nread = 0; | |
156 | ||
157 | size--; | |
158 | while(size) { | |
159 | char c; | |
160 | ||
161 | if (syncRead(fd,&c,1,timeout) == -1) return -1; | |
162 | if (c == '\n') { | |
163 | *ptr = '\0'; | |
164 | if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0'; | |
165 | return nread; | |
166 | } else { | |
167 | *ptr++ = c; | |
168 | *ptr = '\0'; | |
169 | nread++; | |
170 | } | |
171 | } | |
172 | return nread; | |
173 | } | |
174 | ||
175 | void syncCommand(redisClient *c) { | |
176 | /* ignore SYNC if aleady slave or in monitor mode */ | |
177 | if (c->flags & REDIS_SLAVE) return; | |
178 | ||
179 | /* Refuse SYNC requests if we are a slave but the link with our master | |
180 | * is not ok... */ | |
181 | if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED) { | |
182 | addReplyError(c,"Can't SYNC while not connected with my master"); | |
183 | return; | |
184 | } | |
185 | ||
186 | /* SYNC can't be issued when the server has pending data to send to | |
187 | * the client about already issued commands. We need a fresh reply | |
188 | * buffer registering the differences between the BGSAVE and the current | |
189 | * dataset, so that we can copy to other slaves if needed. */ | |
190 | if (listLength(c->reply) != 0) { | |
191 | addReplyError(c,"SYNC is invalid with pending input"); | |
192 | return; | |
193 | } | |
194 | ||
195 | redisLog(REDIS_NOTICE,"Slave ask for synchronization"); | |
196 | /* Here we need to check if there is a background saving operation | |
197 | * in progress, or if it is required to start one */ | |
198 | if (server.bgsavechildpid != -1) { | |
199 | /* Ok a background save is in progress. Let's check if it is a good | |
200 | * one for replication, i.e. if there is another slave that is | |
201 | * registering differences since the server forked to save */ | |
202 | redisClient *slave; | |
203 | listNode *ln; | |
204 | listIter li; | |
205 | ||
206 | listRewind(server.slaves,&li); | |
207 | while((ln = listNext(&li))) { | |
208 | slave = ln->value; | |
209 | if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break; | |
210 | } | |
211 | if (ln) { | |
212 | /* Perfect, the server is already registering differences for | |
213 | * another slave. Set the right state, and copy the buffer. */ | |
214 | listRelease(c->reply); | |
215 | c->reply = listDup(slave->reply); | |
216 | c->replstate = REDIS_REPL_WAIT_BGSAVE_END; | |
217 | redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC"); | |
218 | } else { | |
219 | /* No way, we need to wait for the next BGSAVE in order to | |
220 | * register differences */ | |
221 | c->replstate = REDIS_REPL_WAIT_BGSAVE_START; | |
222 | redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC"); | |
223 | } | |
224 | } else { | |
225 | /* Ok we don't have a BGSAVE in progress, let's start one */ | |
226 | redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC"); | |
227 | if (rdbSaveBackground(server.dbfilename) != REDIS_OK) { | |
228 | redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE"); | |
229 | addReplyError(c,"Unable to perform background save"); | |
230 | return; | |
231 | } | |
232 | c->replstate = REDIS_REPL_WAIT_BGSAVE_END; | |
233 | } | |
234 | c->repldbfd = -1; | |
235 | c->flags |= REDIS_SLAVE; | |
236 | c->slaveseldb = 0; | |
237 | listAddNodeTail(server.slaves,c); | |
238 | return; | |
239 | } | |
240 | ||
241 | void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) { | |
242 | redisClient *slave = privdata; | |
243 | REDIS_NOTUSED(el); | |
244 | REDIS_NOTUSED(mask); | |
245 | char buf[REDIS_IOBUF_LEN]; | |
246 | ssize_t nwritten, buflen; | |
247 | ||
248 | if (slave->repldboff == 0) { | |
249 | /* Write the bulk write count before to transfer the DB. In theory here | |
250 | * we don't know how much room there is in the output buffer of the | |
251 | * socket, but in pratice SO_SNDLOWAT (the minimum count for output | |
252 | * operations) will never be smaller than the few bytes we need. */ | |
253 | sds bulkcount; | |
254 | ||
255 | bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long) | |
256 | slave->repldbsize); | |
257 | if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount)) | |
258 | { | |
259 | sdsfree(bulkcount); | |
260 | freeClient(slave); | |
261 | return; | |
262 | } | |
263 | sdsfree(bulkcount); | |
264 | } | |
265 | lseek(slave->repldbfd,slave->repldboff,SEEK_SET); | |
266 | buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN); | |
267 | if (buflen <= 0) { | |
268 | redisLog(REDIS_WARNING,"Read error sending DB to slave: %s", | |
269 | (buflen == 0) ? "premature EOF" : strerror(errno)); | |
270 | freeClient(slave); | |
271 | return; | |
272 | } | |
273 | if ((nwritten = write(fd,buf,buflen)) == -1) { | |
274 | redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s", | |
275 | strerror(errno)); | |
276 | freeClient(slave); | |
277 | return; | |
278 | } | |
279 | slave->repldboff += nwritten; | |
280 | if (slave->repldboff == slave->repldbsize) { | |
281 | close(slave->repldbfd); | |
282 | slave->repldbfd = -1; | |
283 | aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); | |
284 | slave->replstate = REDIS_REPL_ONLINE; | |
285 | if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, | |
286 | sendReplyToClient, slave) == AE_ERR) { | |
287 | freeClient(slave); | |
288 | return; | |
289 | } | |
290 | addReplySds(slave,sdsempty()); | |
291 | redisLog(REDIS_NOTICE,"Synchronization with slave succeeded"); | |
292 | } | |
293 | } | |
294 | ||
295 | /* This function is called at the end of every backgrond saving. | |
296 | * The argument bgsaveerr is REDIS_OK if the background saving succeeded | |
297 | * otherwise REDIS_ERR is passed to the function. | |
298 | * | |
299 | * The goal of this function is to handle slaves waiting for a successful | |
300 | * background saving in order to perform non-blocking synchronization. */ | |
301 | void updateSlavesWaitingBgsave(int bgsaveerr) { | |
302 | listNode *ln; | |
303 | int startbgsave = 0; | |
304 | listIter li; | |
305 | ||
306 | listRewind(server.slaves,&li); | |
307 | while((ln = listNext(&li))) { | |
308 | redisClient *slave = ln->value; | |
309 | ||
310 | if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) { | |
311 | startbgsave = 1; | |
312 | slave->replstate = REDIS_REPL_WAIT_BGSAVE_END; | |
313 | } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) { | |
314 | struct redis_stat buf; | |
315 | ||
316 | if (bgsaveerr != REDIS_OK) { | |
317 | freeClient(slave); | |
318 | redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error"); | |
319 | continue; | |
320 | } | |
321 | if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 || | |
322 | redis_fstat(slave->repldbfd,&buf) == -1) { | |
323 | freeClient(slave); | |
324 | redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno)); | |
325 | continue; | |
326 | } | |
327 | slave->repldboff = 0; | |
328 | slave->repldbsize = buf.st_size; | |
329 | slave->replstate = REDIS_REPL_SEND_BULK; | |
330 | aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); | |
331 | if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) { | |
332 | freeClient(slave); | |
333 | continue; | |
334 | } | |
335 | } | |
336 | } | |
337 | if (startbgsave) { | |
338 | if (rdbSaveBackground(server.dbfilename) != REDIS_OK) { | |
339 | listIter li; | |
340 | ||
341 | listRewind(server.slaves,&li); | |
342 | redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed"); | |
343 | while((ln = listNext(&li))) { | |
344 | redisClient *slave = ln->value; | |
345 | ||
346 | if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) | |
347 | freeClient(slave); | |
348 | } | |
349 | } | |
350 | } | |
351 | } | |
352 | ||
353 | int syncWithMaster(void) { | |
354 | char buf[1024], tmpfile[256], authcmd[1024]; | |
355 | long dumpsize; | |
356 | int fd = anetTcpConnect(NULL,server.masterhost,server.masterport); | |
357 | int dfd, maxtries = 5; | |
358 | ||
359 | if (fd == -1) { | |
360 | redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", | |
361 | strerror(errno)); | |
362 | return REDIS_ERR; | |
363 | } | |
364 | ||
365 | /* AUTH with the master if required. */ | |
366 | if(server.masterauth) { | |
367 | snprintf(authcmd, 1024, "AUTH %s\r\n", server.masterauth); | |
368 | if (syncWrite(fd, authcmd, strlen(server.masterauth)+7, 5) == -1) { | |
369 | close(fd); | |
370 | redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s", | |
371 | strerror(errno)); | |
372 | return REDIS_ERR; | |
373 | } | |
374 | /* Read the AUTH result. */ | |
375 | if (syncReadLine(fd,buf,1024,3600) == -1) { | |
376 | close(fd); | |
377 | redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s", | |
378 | strerror(errno)); | |
379 | return REDIS_ERR; | |
380 | } | |
381 | if (buf[0] != '+') { | |
382 | close(fd); | |
383 | redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?"); | |
384 | return REDIS_ERR; | |
385 | } | |
386 | } | |
387 | ||
388 | /* Issue the SYNC command */ | |
389 | if (syncWrite(fd,"SYNC \r\n",7,5) == -1) { | |
390 | close(fd); | |
391 | redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s", | |
392 | strerror(errno)); | |
393 | return REDIS_ERR; | |
394 | } | |
395 | /* Read the bulk write count */ | |
396 | if (syncReadLine(fd,buf,1024,3600) == -1) { | |
397 | close(fd); | |
398 | redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s", | |
399 | strerror(errno)); | |
400 | return REDIS_ERR; | |
401 | } | |
402 | if (buf[0] == '-') { | |
403 | close(fd); | |
404 | redisLog(REDIS_WARNING,"MASTER aborted replication with an error: %s", | |
405 | buf+1); | |
406 | return REDIS_ERR; | |
407 | } else if (buf[0] != '$') { | |
408 | close(fd); | |
409 | redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?"); | |
410 | return REDIS_ERR; | |
411 | } | |
412 | dumpsize = strtol(buf+1,NULL,10); | |
413 | redisLog(REDIS_NOTICE,"Receiving %ld bytes data dump from MASTER",dumpsize); | |
414 | /* Read the bulk write data on a temp file */ | |
415 | while(maxtries--) { | |
416 | snprintf(tmpfile,256, | |
417 | "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid()); | |
418 | dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644); | |
419 | if (dfd != -1) break; | |
420 | sleep(1); | |
421 | } | |
422 | if (dfd == -1) { | |
423 | close(fd); | |
424 | redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno)); | |
425 | return REDIS_ERR; | |
426 | } | |
427 | while(dumpsize) { | |
428 | int nread, nwritten; | |
429 | ||
430 | nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024); | |
431 | if (nread <= 0) { | |
432 | redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s", | |
433 | (nread == -1) ? strerror(errno) : "connection lost"); | |
434 | close(fd); | |
435 | close(dfd); | |
436 | return REDIS_ERR; | |
437 | } | |
438 | nwritten = write(dfd,buf,nread); | |
439 | if (nwritten == -1) { | |
440 | redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno)); | |
441 | close(fd); | |
442 | close(dfd); | |
443 | return REDIS_ERR; | |
444 | } | |
445 | dumpsize -= nread; | |
446 | } | |
447 | close(dfd); | |
448 | if (rename(tmpfile,server.dbfilename) == -1) { | |
449 | redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno)); | |
450 | unlink(tmpfile); | |
451 | close(fd); | |
452 | return REDIS_ERR; | |
453 | } | |
454 | emptyDb(); | |
455 | if (rdbLoad(server.dbfilename) != REDIS_OK) { | |
456 | redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk"); | |
457 | close(fd); | |
458 | return REDIS_ERR; | |
459 | } | |
460 | server.master = createClient(fd); | |
461 | server.master->flags |= REDIS_MASTER; | |
462 | server.master->authenticated = 1; | |
463 | server.replstate = REDIS_REPL_CONNECTED; | |
464 | return REDIS_OK; | |
465 | } | |
466 | ||
467 | void slaveofCommand(redisClient *c) { | |
468 | if (!strcasecmp(c->argv[1]->ptr,"no") && | |
469 | !strcasecmp(c->argv[2]->ptr,"one")) { | |
470 | if (server.masterhost) { | |
471 | sdsfree(server.masterhost); | |
472 | server.masterhost = NULL; | |
473 | if (server.master) freeClient(server.master); | |
474 | server.replstate = REDIS_REPL_NONE; | |
475 | redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)"); | |
476 | } | |
477 | } else { | |
478 | sdsfree(server.masterhost); | |
479 | server.masterhost = sdsdup(c->argv[1]->ptr); | |
480 | server.masterport = atoi(c->argv[2]->ptr); | |
481 | if (server.master) freeClient(server.master); | |
482 | server.replstate = REDIS_REPL_CONNECT; | |
483 | redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)", | |
484 | server.masterhost, server.masterport); | |
485 | } | |
486 | addReply(c,shared.ok); | |
487 | } |