MIGRATE: retry one time on I/O error.

Now that we cache connections, a retry attempt makes sure that the
operation don't fail just because there is an existing connection error
on the socket, like the other end closing the connection.

Unfortunately this condition is not detectable using
getsockopt(SO_ERROR), so the only option left is to retry.

We don't retry on timeouts.

diff --git a/src/cluster.c b/src/cluster.c
index a1e576bd03e85c50fd10e07caf934bc6676d8dec..cbcdf373df8d8b3ed69d13979537db9a2098d1b6 100644 (file)
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -1741,12 +1741,19 @@ void migrateCloseTimedoutSockets(void) {
 
 /* MIGRATE host port key dbid timeout [COPY | REPLACE] */
 void migrateCommand(redisClient *c) {
 
 /* MIGRATE host port key dbid timeout [COPY | REPLACE] */
 void migrateCommand(redisClient *c) {

-    int fd, copy = 0, replace = 0, j;
+    int fd, copy, replace, j;

     long timeout;
     long dbid;
     long timeout;
     long dbid;

-    long long ttl = 0, expireat;
+    long long ttl, expireat;

     robj *o;
     rio cmd, payload;
     robj *o;
     rio cmd, payload;

+    int retry_num = 0;
+
+try_again:
+    /* Initialization */
+    copy = 0;
+    replace = 0;
+    ttl = 0;

 
     /* Parse additional options */
     for (j = 6; j < c->argc; j++) {
 
     /* Parse additional options */
     for (j = 6; j < c->argc; j++) {


@@ -1809,6 +1816,7 @@ void migrateCommand(redisClient *c) {
         redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"REPLACE",7));
 
     /* Tranfer the query to the other node in 64K chunks. */
         redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"REPLACE",7));
 
     /* Tranfer the query to the other node in 64K chunks. */

+    errno = 0;

     {
         sds buf = cmd.io.buffer.ptr;
         size_t pos = 0, towrite;
     {
         sds buf = cmd.io.buffer.ptr;
         size_t pos = 0, towrite;


@@ -1857,15 +1865,19 @@ void migrateCommand(redisClient *c) {
     return;
 
 socket_wr_err:
     return;
 
 socket_wr_err:

-    addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n"));

     sdsfree(cmd.io.buffer.ptr);
     migrateCloseSocket(c->argv[1],c->argv[2]);
     sdsfree(cmd.io.buffer.ptr);
     migrateCloseSocket(c->argv[1],c->argv[2]);

+    if (errno != ETIMEDOUT && retry_num++ == 0) goto try_again;
+    addReplySds(c,
+        sdsnew("-IOERR error or timeout writing to target instance\r\n"));

     return;
 
 socket_rd_err:
     return;
 
 socket_rd_err:

-    addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n"));

     sdsfree(cmd.io.buffer.ptr);
     migrateCloseSocket(c->argv[1],c->argv[2]);
     sdsfree(cmd.io.buffer.ptr);
     migrateCloseSocket(c->argv[1],c->argv[2]);

+    if (errno != ETIMEDOUT && retry_num++ == 0) goto try_again;
+    addReplySds(c,
+        sdsnew("-IOERR error or timeout reading from target node\r\n"));

     return;
 }
 
     return;
 }