Commit fe63046e authored by Oran Agra's avatar Oran Agra
Browse files

diskless replication on slave side

parent c77081a4
...@@ -193,6 +193,20 @@ int anetSendTimeout(char *err, int fd, long long ms) { ...@@ -193,6 +193,20 @@ int anetSendTimeout(char *err, int fd, long long ms) {
return ANET_OK; return ANET_OK;
} }
/* Set the socket receive timeout (SO_RCVTIMEO socket option) to the specified
* number of milliseconds, or disable it if the 'ms' argument is zero. */
int anetRecvTimeout(char *err, int fd, long long ms) {
struct timeval tv;
tv.tv_sec = ms/1000;
tv.tv_usec = (ms%1000)*1000;
if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) == -1) {
anetSetError(err, "setsockopt SO_RCVTIMEO: %s", strerror(errno));
return ANET_ERR;
}
return ANET_OK;
}
/* anetGenericResolve() is called by anetResolve() and anetResolveIP() to /* anetGenericResolve() is called by anetResolve() and anetResolveIP() to
* do the actual work. It resolves the hostname "host" and set the string * do the actual work. It resolves the hostname "host" and set the string
* representation of the IP address into the buffer pointed by "ipbuf". * representation of the IP address into the buffer pointed by "ipbuf".
......
...@@ -67,6 +67,7 @@ int anetEnableTcpNoDelay(char *err, int fd); ...@@ -67,6 +67,7 @@ int anetEnableTcpNoDelay(char *err, int fd);
int anetDisableTcpNoDelay(char *err, int fd); int anetDisableTcpNoDelay(char *err, int fd);
int anetTcpKeepAlive(char *err, int fd); int anetTcpKeepAlive(char *err, int fd);
int anetSendTimeout(char *err, int fd, long long ms); int anetSendTimeout(char *err, int fd, long long ms);
int anetRecvTimeout(char *err, int fd, long long ms);
int anetPeerToString(int fd, char *ip, size_t ip_len, int *port); int anetPeerToString(int fd, char *ip, size_t ip_len, int *port);
int anetKeepAlive(char *err, int fd, int interval); int anetKeepAlive(char *err, int fd, int interval);
int anetSockName(int fd, char *ip, size_t ip_len, int *port); int anetSockName(int fd, char *ip, size_t ip_len, int *port);
......
...@@ -620,7 +620,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -620,7 +620,7 @@ int loadAppendOnlyFile(char *filename) {
server.aof_state = REDIS_AOF_OFF; server.aof_state = REDIS_AOF_OFF;
fakeClient = createFakeClient(); fakeClient = createFakeClient();
startLoading(fp); startLoadingFile(fp);
while(1) { while(1) {
int argc, j; int argc, j;
......
...@@ -1187,18 +1187,19 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) { ...@@ -1187,18 +1187,19 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
/* Mark that we are loading in the global state and setup the fields /* Mark that we are loading in the global state and setup the fields
* needed to provide loading stats. */ * needed to provide loading stats. */
void startLoading(FILE *fp) { void startLoading(size_t size) {
struct stat sb;
/* Load the DB */ /* Load the DB */
server.loading = 1; server.loading = 1;
server.loading_start_time = time(NULL); server.loading_start_time = time(NULL);
server.loading_loaded_bytes = 0; server.loading_loaded_bytes = 0;
if (fstat(fileno(fp), &sb) == -1) { server.loading_total_bytes = size;
server.loading_total_bytes = 0; }
} else {
server.loading_total_bytes = sb.st_size; void startLoadingFile(FILE *fp) {
} struct stat sb;
if (fstat(fileno(fp), &sb) == -1)
sb.st_size = 0;
startLoading(sb.st_size);
} }
/* Refresh the loading progress info */ /* Refresh the loading progress info */
...@@ -1233,66 +1234,74 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -1233,66 +1234,74 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
} }
int rdbLoad(char *filename) { int rdbLoad(char *filename) {
FILE *fp;
int retval;
rio rdb;
struct stat sb;
if ((fp = fopen(filename,"r")) == NULL) return REDIS_ERR;
if (fstat(fileno(fp), &sb) == -1)
sb.st_size = 0;
rioInitWithFile(&rdb,fp);
retval = rdbLoadRio(&rdb, sb.st_size);
fclose(fp);
return retval;
}
int rdbLoadRio(rio *rdb, size_t size) {
uint32_t dbid; uint32_t dbid;
int type, rdbver; int type, rdbver;
redisDb *db = server.db+0; redisDb *db = server.db+0;
char buf[1024]; char buf[1024];
long long expiretime, now = mstime(); long long expiretime, now = mstime();
FILE *fp;
rio rdb;
if ((fp = fopen(filename,"r")) == NULL) return REDIS_ERR;
rioInitWithFile(&rdb,fp); rdb->update_cksum = rdbLoadProgressCallback;
rdb.update_cksum = rdbLoadProgressCallback; rdb->max_processing_chunk = server.loading_process_events_interval_bytes;
rdb.max_processing_chunk = server.loading_process_events_interval_bytes; if (rioRead(rdb,buf,9) == 0) goto eoferr;
if (rioRead(&rdb,buf,9) == 0) goto eoferr;
buf[9] = '\0'; buf[9] = '\0';
if (memcmp(buf,"REDIS",5) != 0) { if (memcmp(buf,"REDIS",5) != 0) {
fclose(fp);
redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file"); redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
errno = EINVAL; errno = EINVAL;
return REDIS_ERR; return REDIS_ERR;
} }
rdbver = atoi(buf+5); rdbver = atoi(buf+5);
if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) { if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) {
fclose(fp);
redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver); redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
errno = EINVAL; errno = EINVAL;
return REDIS_ERR; return REDIS_ERR;
} }
startLoading(fp); startLoading(size);
while(1) { while(1) {
robj *key, *val; robj *key, *val;
expiretime = -1; expiretime = -1;
/* Read type. */ /* Read type. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
/* Handle special types. */ /* Handle special types. */
if (type == REDIS_RDB_OPCODE_EXPIRETIME) { if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
/* EXPIRETIME: load an expire associated with the next key /* EXPIRETIME: load an expire associated with the next key
* to load. Note that after loading an expire we need to * to load. Note that after loading an expire we need to
* load the actual type, and continue. */ * load the actual type, and continue. */
if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr; if ((expiretime = rdbLoadTime(rdb)) == -1) goto eoferr;
/* We read the time so we need to read the object type again. */ /* We read the time so we need to read the object type again. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
/* the EXPIRETIME opcode specifies time in seconds, so convert /* the EXPIRETIME opcode specifies time in seconds, so convert
* into milliseconds. */ * into milliseconds. */
expiretime *= 1000; expiretime *= 1000;
} else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) { } else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) {
/* EXPIRETIME_MS: milliseconds precision expire times introduced /* EXPIRETIME_MS: milliseconds precision expire times introduced
* with RDB v3. Like EXPIRETIME but no with more precision. */ * with RDB v3. Like EXPIRETIME but no with more precision. */
if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr; if ((expiretime = rdbLoadMillisecondTime(rdb)) == -1) goto eoferr;
/* We read the time so we need to read the object type again. */ /* We read the time so we need to read the object type again. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
} else if (type == REDIS_RDB_OPCODE_EOF) { } else if (type == REDIS_RDB_OPCODE_EOF) {
/* EOF: End of file, exit the main loop. */ /* EOF: End of file, exit the main loop. */
break; break;
} else if (type == REDIS_RDB_OPCODE_SELECTDB) { } else if (type == REDIS_RDB_OPCODE_SELECTDB) {
/* SELECTDB: Select the specified database. */ /* SELECTDB: Select the specified database. */
if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) if ((dbid = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR)
goto eoferr; goto eoferr;
if (dbid >= (unsigned)server.dbnum) { if (dbid >= (unsigned)server.dbnum) {
redisLog(REDIS_WARNING, redisLog(REDIS_WARNING,
...@@ -1307,9 +1316,9 @@ int rdbLoad(char *filename) { ...@@ -1307,9 +1316,9 @@ int rdbLoad(char *filename) {
/* RESIZEDB: Hint about the size of the keys in the currently /* RESIZEDB: Hint about the size of the keys in the currently
* selected data base, in order to avoid useless rehashing. */ * selected data base, in order to avoid useless rehashing. */
uint32_t db_size, expires_size; uint32_t db_size, expires_size;
if ((db_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) if ((db_size = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR)
goto eoferr; goto eoferr;
if ((expires_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) if ((expires_size = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR)
goto eoferr; goto eoferr;
dictExpand(db->dict,db_size); dictExpand(db->dict,db_size);
dictExpand(db->expires,expires_size); dictExpand(db->expires,expires_size);
...@@ -1321,8 +1330,8 @@ int rdbLoad(char *filename) { ...@@ -1321,8 +1330,8 @@ int rdbLoad(char *filename) {
* *
* An AUX field is composed of two strings: key and value. */ * An AUX field is composed of two strings: key and value. */
robj *auxkey, *auxval; robj *auxkey, *auxval;
if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((auxkey = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((auxval = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
if (((char*)auxkey->ptr)[0] == '%') { if (((char*)auxkey->ptr)[0] == '%') {
/* All the fields with a name staring with '%' are considered /* All the fields with a name staring with '%' are considered
...@@ -1344,9 +1353,9 @@ int rdbLoad(char *filename) { ...@@ -1344,9 +1353,9 @@ int rdbLoad(char *filename) {
} }
/* Read key */ /* Read key */
if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((key = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
/* Read value */ /* Read value */
if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr; if ((val = rdbLoadObject(type,rdb)) == NULL) goto eoferr;
/* Check if the key already expired. This function is used when loading /* Check if the key already expired. This function is used when loading
* an RDB file from disk, either at startup, or when an RDB was * an RDB file from disk, either at startup, or when an RDB was
* received from the master. In the latter case, the master is * received from the master. In the latter case, the master is
...@@ -1367,9 +1376,9 @@ int rdbLoad(char *filename) { ...@@ -1367,9 +1376,9 @@ int rdbLoad(char *filename) {
} }
/* Verify the checksum if RDB version is >= 5 */ /* Verify the checksum if RDB version is >= 5 */
if (rdbver >= 5 && server.rdb_checksum) { if (rdbver >= 5 && server.rdb_checksum) {
uint64_t cksum, expected = rdb.cksum; uint64_t cksum, expected = rdb->cksum;
if (rioRead(&rdb,&cksum,8) == 0) goto eoferr; if (rioRead(rdb,&cksum,8) == 0) goto eoferr;
memrev64ifbe(&cksum); memrev64ifbe(&cksum);
if (cksum == 0) { if (cksum == 0) {
redisLog(REDIS_WARNING,"RDB file was saved with checksum disabled: no check performed."); redisLog(REDIS_WARNING,"RDB file was saved with checksum disabled: no check performed.");
...@@ -1379,7 +1388,6 @@ int rdbLoad(char *filename) { ...@@ -1379,7 +1388,6 @@ int rdbLoad(char *filename) {
} }
} }
fclose(fp);
stopLoading(); stopLoading();
return REDIS_OK; return REDIS_OK;
...@@ -1572,7 +1580,7 @@ int rdbSaveToSlavesSockets(void) { ...@@ -1572,7 +1580,7 @@ int rdbSaveToSlavesSockets(void) {
clientids[numfds] = slave->id; clientids[numfds] = slave->id;
fds[numfds++] = slave->fd; fds[numfds++] = slave->fd;
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END; slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
/* Put the socket in non-blocking mode to simplify RDB transfer. /* Put the socket in blocking mode to simplify RDB transfer.
* We'll restore it when the children returns (since duped socket * We'll restore it when the children returns (since duped socket
* will share the O_NONBLOCK attribute with the parent). */ * will share the O_NONBLOCK attribute with the parent). */
anetBlock(NULL,slave->fd); anetBlock(NULL,slave->fd);
...@@ -1646,6 +1654,7 @@ int rdbSaveToSlavesSockets(void) { ...@@ -1646,6 +1654,7 @@ int rdbSaveToSlavesSockets(void) {
zfree(msg); zfree(msg);
} }
zfree(clientids); zfree(clientids);
rioFreeFdset(&slave_sockets);
exitFromChild((retval == REDIS_OK) ? 0 : 1); exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else { } else {
/* Parent */ /* Parent */
......
...@@ -105,6 +105,7 @@ uint32_t rdbLoadLen(rio *rdb, int *isencoded); ...@@ -105,6 +105,7 @@ uint32_t rdbLoadLen(rio *rdb, int *isencoded);
int rdbSaveObjectType(rio *rdb, robj *o); int rdbSaveObjectType(rio *rdb, robj *o);
int rdbLoadObjectType(rio *rdb); int rdbLoadObjectType(rio *rdb);
int rdbLoad(char *filename); int rdbLoad(char *filename);
int rdbLoadRio(rio *rdb, size_t size);
int rdbSaveBackground(char *filename); int rdbSaveBackground(char *filename);
int rdbSaveToSlavesSockets(void); int rdbSaveToSlavesSockets(void);
void rdbRemoveTempFile(pid_t childpid); void rdbRemoveTempFile(pid_t childpid);
......
...@@ -1490,6 +1490,9 @@ void initServerConfig(void) { ...@@ -1490,6 +1490,9 @@ void initServerConfig(void) {
server.cached_master = NULL; server.cached_master = NULL;
server.repl_master_initial_offset = -1; server.repl_master_initial_offset = -1;
server.repl_state = REDIS_REPL_NONE; server.repl_state = REDIS_REPL_NONE;
server.repl_transfer_tmpfile = NULL;
server.repl_transfer_fd = -1;
server.repl_transfer_s = -1;
server.repl_syncio_timeout = REDIS_REPL_SYNCIO_TIMEOUT; server.repl_syncio_timeout = REDIS_REPL_SYNCIO_TIMEOUT;
server.repl_serve_stale_data = REDIS_DEFAULT_SLAVE_SERVE_STALE_DATA; server.repl_serve_stale_data = REDIS_DEFAULT_SLAVE_SERVE_STALE_DATA;
server.repl_slave_ro = REDIS_DEFAULT_SLAVE_READ_ONLY; server.repl_slave_ro = REDIS_DEFAULT_SLAVE_READ_ONLY;
......
...@@ -1195,7 +1195,8 @@ long long replicationGetSlaveOffset(void); ...@@ -1195,7 +1195,8 @@ long long replicationGetSlaveOffset(void);
char *replicationGetSlaveName(redisClient *c); char *replicationGetSlaveName(redisClient *c);
/* Generic persistence functions */ /* Generic persistence functions */
void startLoading(FILE *fp); void startLoading(size_t size);
void startLoadingFile(FILE *fp);
void loadingProgress(off_t pos); void loadingProgress(off_t pos);
void stopLoading(void); void stopLoading(void);
......
...@@ -823,9 +823,13 @@ void replicationAbortSyncTransfer(void) { ...@@ -823,9 +823,13 @@ void replicationAbortSyncTransfer(void) {
aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
close(server.repl_transfer_s); close(server.repl_transfer_s);
if (!server.repl_diskless_sync) {
close(server.repl_transfer_fd); close(server.repl_transfer_fd);
unlink(server.repl_transfer_tmpfile); unlink(server.repl_transfer_tmpfile);
zfree(server.repl_transfer_tmpfile); zfree(server.repl_transfer_tmpfile);
server.repl_transfer_tmpfile = NULL;
server.repl_transfer_fd = -1;
}
server.repl_state = REDIS_REPL_CONNECT; server.repl_state = REDIS_REPL_CONNECT;
} }
...@@ -877,6 +881,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -877,6 +881,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
char buf[4096]; char buf[4096];
ssize_t nread, readlen; ssize_t nread, readlen;
off_t left; off_t left;
sds initialCommandStream = NULL;
REDIS_NOTUSED(el); REDIS_NOTUSED(el);
REDIS_NOTUSED(privdata); REDIS_NOTUSED(privdata);
REDIS_NOTUSED(mask); REDIS_NOTUSED(mask);
...@@ -942,7 +947,8 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -942,7 +947,8 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
return; return;
} }
/* Read bulk data */ if (!server.repl_diskless_sync) {
/* read the data from the socket, store it to a file and search for the EOF */
if (usemark) { if (usemark) {
readlen = sizeof(buf); readlen = sizeof(buf);
} else { } else {
...@@ -1010,13 +1016,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1010,13 +1016,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
if (server.repl_transfer_read == server.repl_transfer_size) if (server.repl_transfer_read == server.repl_transfer_size)
eof_reached = 1; eof_reached = 1;
} }
if (!eof_reached)
if (eof_reached) {
if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
replicationAbortSyncTransfer();
return; return;
} }
/* We reach here when the slave is using diskless replication,
* or when we are done reading from the socket to the rdb file. */
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Flushing old data"); redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Flushing old data");
signalFlushedDb(-1); signalFlushedDb(-1);
emptyDb(replicationEmptyDbCallback); emptyDb(replicationEmptyDbCallback);
...@@ -1026,15 +1031,64 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1026,15 +1031,64 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
* time for non blocking loading. */ * time for non blocking loading. */
aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory"); redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
if (server.repl_diskless_sync) {
rio rdb;
rioInitWithFd(&rdb,fd);
/* Put the socket in blocking mode to simplify RDB transfer.
* We'll restore it when the RDB is received. */
anetBlock(NULL,fd);
anetRecvTimeout(NULL,fd,server.repl_timeout*1000);
if (rdbLoadRio(&rdb, server.repl_transfer_size) != REDIS_OK) {
redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
replicationAbortSyncTransfer();
rioFreeFd(&rdb, NULL);
return;
}
if (usemark) {
if (!rioRead(&rdb,buf,REDIS_RUN_ID_SIZE) || memcmp(buf,eofmark,REDIS_RUN_ID_SIZE) != 0) {
redisLog(REDIS_WARNING,"replication stream EOF marker is broken");
replicationAbortSyncTransfer();
rioFreeFd(&rdb, NULL);
return;
}
}
/* get the unread command stream from the rio buffer */
rioFreeFd(&rdb, &initialCommandStream);
/* Restore the socket as non-blocking. */
anetNonBlock(NULL,fd);
anetRecvTimeout(NULL,fd,0);
} else {
if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
replicationAbortSyncTransfer();
return;
}
if (rdbLoad(server.rdb_filename) != REDIS_OK) { if (rdbLoad(server.rdb_filename) != REDIS_OK) {
redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk"); redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
replicationAbortSyncTransfer(); replicationAbortSyncTransfer();
return; return;
} }
/* Final setup of the connected slave <- master link */
zfree(server.repl_transfer_tmpfile); zfree(server.repl_transfer_tmpfile);
close(server.repl_transfer_fd); close(server.repl_transfer_fd);
server.repl_transfer_fd = -1;
server.repl_transfer_tmpfile = NULL;
}
/* Final setup of the connected slave <- master link */
replicationCreateMasterClient(server.repl_transfer_s); replicationCreateMasterClient(server.repl_transfer_s);
/* feed the initial command stream into the client input buffer */
if (initialCommandStream) {
int nread = sdslen(initialCommandStream);
server.current_client = server.master;
server.master->querybuf = sdscatsds(server.master->querybuf, initialCommandStream);
server.master->reploff += nread;
server.master->lastinteraction = server.unixtime;
server.stat_net_input_bytes += nread;
processInputBuffer(server.master);
server.current_client = NULL;
sdsfree(initialCommandStream);
}
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success"); redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
/* Restart the AOF subsystem now that we finished the sync. This /* Restart the AOF subsystem now that we finished the sync. This
* will trigger an AOF rewrite, and when done will start appending * will trigger an AOF rewrite, and when done will start appending
...@@ -1052,8 +1106,6 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1052,8 +1106,6 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
exit(1); exit(1);
} }
} }
}
return; return;
error: error:
...@@ -1341,6 +1393,7 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1341,6 +1393,7 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
} }
/* Prepare a suitable temp file for bulk transfer */ /* Prepare a suitable temp file for bulk transfer */
if (!server.repl_diskless_sync) {
while(maxtries--) { while(maxtries--) {
snprintf(tmpfile,256, snprintf(tmpfile,256,
"temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid()); "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
...@@ -1352,6 +1405,9 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1352,6 +1405,9 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno)); redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
goto error; goto error;
} }
server.repl_transfer_tmpfile = zstrdup(tmpfile);
server.repl_transfer_fd = dfd;
}
/* Setup the non blocking download of the bulk file. */ /* Setup the non blocking download of the bulk file. */
if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL) if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
...@@ -1367,13 +1423,17 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1367,13 +1423,17 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
server.repl_transfer_size = -1; server.repl_transfer_size = -1;
server.repl_transfer_read = 0; server.repl_transfer_read = 0;
server.repl_transfer_last_fsync_off = 0; server.repl_transfer_last_fsync_off = 0;
server.repl_transfer_fd = dfd;
server.repl_transfer_lastio = server.unixtime; server.repl_transfer_lastio = server.unixtime;
server.repl_transfer_tmpfile = zstrdup(tmpfile);
return; return;
error: error:
close(fd); close(fd);
if (server.repl_transfer_fd != -1)
close(server.repl_transfer_fd);
if (server.repl_transfer_tmpfile)
zfree(server.repl_transfer_tmpfile);
server.repl_transfer_tmpfile = NULL;
server.repl_transfer_fd = -1;
server.repl_transfer_s = -1; server.repl_transfer_s = -1;
server.repl_state = REDIS_REPL_CONNECT; server.repl_state = REDIS_REPL_CONNECT;
return; return;
......
...@@ -157,13 +157,106 @@ void rioInitWithFile(rio *r, FILE *fp) { ...@@ -157,13 +157,106 @@ void rioInitWithFile(rio *r, FILE *fp) {
r->io.file.autosync = 0; r->io.file.autosync = 0;
} }
/* ------------------- File descriptor implementation ------------------- */
static size_t rioFdWrite(rio *r, const void *buf, size_t len) {
REDIS_NOTUSED(r);
REDIS_NOTUSED(buf);
REDIS_NOTUSED(len);
return 0; /* Error, this target does not yet support writing. */
}
/* Returns 1 or 0 for success/failure. */
static size_t rioFdRead(rio *r, void *buf, size_t len) {
size_t avail = sdslen(r->io.fd.buf)-r->io.fd.pos;
/* if the buffer is too small for the entire request: realloc */
if (sdslen(r->io.fd.buf) + sdsavail(r->io.fd.buf) < len)
r->io.fd.buf = sdsMakeRoomFor(r->io.fd.buf, len - sdslen(r->io.fd.buf));
/* if the remaining unused buffer is not large enough: memmove so that we can read the rest */
if (len > avail && sdsavail(r->io.fd.buf) < len - avail) {
sdsrange(r->io.fd.buf, r->io.fd.pos, -1);
r->io.fd.pos = 0;
}
/* if we don't already have all the data in the sds, read more */
while (len > sdslen(r->io.fd.buf) - r->io.fd.pos) {
size_t toread = len - (sdslen(r->io.fd.buf) - r->io.fd.pos);
/* read either what's missing, or REDIS_IOBUF_LEN, the bigger of the two */
if (toread < REDIS_IOBUF_LEN)
toread = REDIS_IOBUF_LEN;
if (toread > sdsavail(r->io.fd.buf))
toread = sdsavail(r->io.fd.buf);
int retval = read(r->io.fd.fd, (char*)r->io.fd.buf + sdslen(r->io.fd.buf), toread);
if (retval == -1) {
if (errno == EWOULDBLOCK) errno = ETIMEDOUT;
return 0;
}
sdsIncrLen(r->io.fd.buf, retval);
}
memcpy(buf, (char*)r->io.fd.buf + r->io.fd.pos, len);
r->io.fd.pos += len;
return len;
}
/* Returns read/write position in file. */
static off_t rioFdTell(rio *r) {
off_t pos = lseek(r->io.fd.fd, 0, SEEK_CUR);
return pos - sdslen(r->io.fd.buf) + r->io.fd.pos;
}
/* Flushes any buffer to target device if applicable. Returns 1 on success
* and 0 on failures. */
static int rioFdFlush(rio *r) {
/* Our flush is implemented by the write method, that recognizes a
* buffer set to NULL with a count of zero as a flush request. */
return rioFdWrite(r,NULL,0);
}
static const rio rioFdIO = {
rioFdRead,
rioFdWrite,
rioFdTell,
rioFdFlush,
NULL, /* update_checksum */
0, /* current checksum */
0, /* bytes read or written */
0, /* read/write chunk size */
{ { NULL, 0 } } /* union for io-specific vars */
};
void rioInitWithFd(rio *r, int fd) {
*r = rioFdIO;
r->io.fd.fd = fd;
r->io.fd.pos = 0;
r->io.fd.buf = sdsnewlen(NULL, REDIS_IOBUF_LEN);
sdsclear(r->io.fd.buf);
}
/* release the rio stream.
* optionally returns the unread buffered data. */
void rioFreeFd(rio *r, sds* out_remainingBufferedData) {
if(out_remainingBufferedData && (size_t)r->io.fd.pos < sdslen(r->io.fd.buf)) {
if (r->io.fd.pos > 0)
sdsrange(r->io.fd.buf, r->io.fd.pos, -1);
*out_remainingBufferedData = r->io.fd.buf;
} else {
sdsfree(r->io.fd.buf);
if (out_remainingBufferedData)
*out_remainingBufferedData = NULL;
}
r->io.fd.buf = NULL;
}
/* ------------------- File descriptors set implementation ------------------- */ /* ------------------- File descriptors set implementation ------------------- */
/* Returns 1 or 0 for success/failure. /* Returns 1 or 0 for success/failure.
* The function returns success as long as we are able to correctly write * The function returns success as long as we are able to correctly write
* to at least one file descriptor. * to at least one file descriptor.
* *
* When buf is NULL adn len is 0, the function performs a flush operation * When buf is NULL and len is 0, the function performs a flush operation
* if there is some pending buffer, so this function is also used in order * if there is some pending buffer, so this function is also used in order
* to implement rioFdsetFlush(). */ * to implement rioFdsetFlush(). */
static size_t rioFdsetWrite(rio *r, const void *buf, size_t len) { static size_t rioFdsetWrite(rio *r, const void *buf, size_t len) {
...@@ -176,7 +269,7 @@ static size_t rioFdsetWrite(rio *r, const void *buf, size_t len) { ...@@ -176,7 +269,7 @@ static size_t rioFdsetWrite(rio *r, const void *buf, size_t len) {
* a given size, we actually write to the sockets. */ * a given size, we actually write to the sockets. */
if (len) { if (len) {
r->io.fdset.buf = sdscatlen(r->io.fdset.buf,buf,len); r->io.fdset.buf = sdscatlen(r->io.fdset.buf,buf,len);
len = 0; /* Prevent entering the while belove if we don't flush. */ len = 0; /* Prevent entering the while below if we don't flush. */
if (sdslen(r->io.fdset.buf) > REDIS_IOBUF_LEN) doflush = 1; if (sdslen(r->io.fdset.buf) > REDIS_IOBUF_LEN) doflush = 1;
} }
...@@ -276,6 +369,7 @@ void rioInitWithFdset(rio *r, int *fds, int numfds) { ...@@ -276,6 +369,7 @@ void rioInitWithFdset(rio *r, int *fds, int numfds) {
r->io.fdset.buf = sdsempty(); r->io.fdset.buf = sdsempty();
} }
/* release the rio stream. */
void rioFreeFdset(rio *r) { void rioFreeFdset(rio *r) {
zfree(r->io.fdset.fds); zfree(r->io.fdset.fds);
zfree(r->io.fdset.state); zfree(r->io.fdset.state);
......
...@@ -73,6 +73,12 @@ struct _rio { ...@@ -73,6 +73,12 @@ struct _rio {
off_t buffered; /* Bytes written since last fsync. */ off_t buffered; /* Bytes written since last fsync. */
off_t autosync; /* fsync after 'autosync' bytes written. */ off_t autosync; /* fsync after 'autosync' bytes written. */
} file; } file;
/* file descriptor */
struct {
int fd; /* File descriptor. */
off_t pos;
sds buf;
} fd;
/* Multiple FDs target (used to write to N sockets). */ /* Multiple FDs target (used to write to N sockets). */
struct { struct {
int *fds; /* File descriptors. */ int *fds; /* File descriptors. */
...@@ -126,8 +132,12 @@ static inline int rioFlush(rio *r) { ...@@ -126,8 +132,12 @@ static inline int rioFlush(rio *r) {
void rioInitWithFile(rio *r, FILE *fp); void rioInitWithFile(rio *r, FILE *fp);
void rioInitWithBuffer(rio *r, sds s); void rioInitWithBuffer(rio *r, sds s);
void rioInitWithFd(rio *r, int fd);
void rioInitWithFdset(rio *r, int *fds, int numfds); void rioInitWithFdset(rio *r, int *fds, int numfds);
void rioFreeFdset(rio *r);
void rioFreeFd(rio *r, sds* out_remainingBufferedData);
size_t rioWriteBulkCount(rio *r, char prefix, int count); size_t rioWriteBulkCount(rio *r, char prefix, int count);
size_t rioWriteBulkString(rio *r, const char *buf, size_t len); size_t rioWriteBulkString(rio *r, const char *buf, size_t len);
size_t rioWriteBulkLongLong(rio *r, long long l); size_t rioWriteBulkLongLong(rio *r, long long l);
......
proc start_bg_complex_data {host port db ops} {
set tclsh [info nameofexecutable]
exec $tclsh tests/helpers/bg_complex_data.tcl $host $port $db $ops &
}
proc stop_bg_complex_data {handle} {
catch {exec /bin/kill -9 $handle}
}
start_server {tags {"repl"}} { start_server {tags {"repl"}} {
start_server {} { start_server {} {
......
proc start_bg_complex_data {host port db ops} {
set tclsh [info nameofexecutable]
exec $tclsh tests/helpers/bg_complex_data.tcl $host $port $db $ops &
}
proc stop_bg_complex_data {handle} {
catch {exec /bin/kill -9 $handle}
}
# Creates a master-slave pair and breaks the link continuously to force # Creates a master-slave pair and breaks the link continuously to force
# partial resyncs attempts, all this while flooding the master with # partial resyncs attempts, all this while flooding the master with
# write queries. # write queries.
......
...@@ -94,16 +94,18 @@ start_server {tags {"repl"}} { ...@@ -94,16 +94,18 @@ start_server {tags {"repl"}} {
} }
} }
foreach dl {no yes} { foreach mdl {no yes} {
foreach sdl {no yes} {
start_server {tags {"repl"}} { start_server {tags {"repl"}} {
set master [srv 0 client] set master [srv 0 client]
$master config set repl-diskless-sync $dl $master config set repl-diskless-sync $mdl
$master config set repl-diskless-sync-delay 1
set master_host [srv 0 host] set master_host [srv 0 host]
set master_port [srv 0 port] set master_port [srv 0 port]
set slaves {} set slaves {}
set load_handle0 [start_write_load $master_host $master_port 3] set load_handle0 [start_bg_complex_data $master_host $master_port 9 100000000]
set load_handle1 [start_write_load $master_host $master_port 5] set load_handle1 [start_bg_complex_data $master_host $master_port 11 100000000]
set load_handle2 [start_write_load $master_host $master_port 20] set load_handle2 [start_bg_complex_data $master_host $master_port 12 100000000]
set load_handle3 [start_write_load $master_host $master_port 8] set load_handle3 [start_write_load $master_host $master_port 8]
set load_handle4 [start_write_load $master_host $master_port 4] set load_handle4 [start_write_load $master_host $master_port 4]
start_server {} { start_server {} {
...@@ -112,8 +114,11 @@ foreach dl {no yes} { ...@@ -112,8 +114,11 @@ foreach dl {no yes} {
lappend slaves [srv 0 client] lappend slaves [srv 0 client]
start_server {} { start_server {} {
lappend slaves [srv 0 client] lappend slaves [srv 0 client]
test "Connect multiple slaves at the same time (issue #141), diskless=$dl" { test "Connect multiple slaves at the same time (issue #141), master diskless=$mdl, slave diskless=$sdl" {
# Send SALVEOF commands to slaves # Send SALVEOF commands to slaves
[lindex $slaves 0] config set repl-diskless-sync $sdl
[lindex $slaves 1] config set repl-diskless-sync $sdl
[lindex $slaves 2] config set repl-diskless-sync $sdl
[lindex $slaves 0] slaveof $master_host $master_port [lindex $slaves 0] slaveof $master_host $master_port
[lindex $slaves 1] slaveof $master_host $master_port [lindex $slaves 1] slaveof $master_host $master_port
[lindex $slaves 2] slaveof $master_host $master_port [lindex $slaves 2] slaveof $master_host $master_port
...@@ -146,9 +151,9 @@ foreach dl {no yes} { ...@@ -146,9 +151,9 @@ foreach dl {no yes} {
} }
# Stop the write load # Stop the write load
stop_write_load $load_handle0 stop_bg_complex_data $load_handle0
stop_write_load $load_handle1 stop_bg_complex_data $load_handle1
stop_write_load $load_handle2 stop_bg_complex_data $load_handle2
stop_write_load $load_handle3 stop_write_load $load_handle3
stop_write_load $load_handle4 stop_write_load $load_handle4
...@@ -176,4 +181,5 @@ foreach dl {no yes} { ...@@ -176,4 +181,5 @@ foreach dl {no yes} {
} }
} }
} }
}
} }
...@@ -371,3 +371,15 @@ proc start_write_load {host port seconds} { ...@@ -371,3 +371,15 @@ proc start_write_load {host port seconds} {
proc stop_write_load {handle} { proc stop_write_load {handle} {
catch {exec /bin/kill -9 $handle} catch {exec /bin/kill -9 $handle}
} }
# Execute a background process writing complex data for the specified number
# of ops to the specified Redis instance.
proc start_bg_complex_data {host port db ops} {
set tclsh [info nameofexecutable]
exec $tclsh tests/helpers/bg_complex_data.tcl $host $port $db $ops &
}
# Stop a process generating write load executed with start_bg_complex_data.
proc stop_bg_complex_data {handle} {
catch {exec /bin/kill -9 $handle}
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment