Fix bug: the pipeline mode socket return an unexpected result after reconnecting (#2358)

* fix bug: the pipeline mode socket return an unexpected result after reconnecting

* fix typos: pipeline is right

---------

Co-authored-by: marcofu <marcofu@tencent.com>

cluster_library.c

@@ -1097,7 +1097,7 @@ PHP_REDIS_API int cluster_map_keyspace(redisCluster *c) {
                 memset(c->master, 0, sizeof(redisClusterNode*)*REDIS_CLUSTER_SLOTS);
             }
         }
-        redis_sock_disconnect(seed, 0);
+        redis_sock_disconnect(seed, 0, 1);
         if (mapped) break;
     } ZEND_HASH_FOREACH_END();
 
@@ -1218,13 +1218,13 @@ PHP_REDIS_API void cluster_disconnect(redisCluster *c, int force) {
         if (node == NULL) continue;
 
         /* Disconnect from the master */
-        redis_sock_disconnect(node->sock, force);
+        redis_sock_disconnect(node->sock, force, 1);
 
         /* We also want to disconnect any slave connections so they will be pooled
          * in the event we are using persistent connections and connection pooling. */
         if (node->slaves) {
             ZEND_HASH_FOREACH_PTR(node->slaves, slave) {
-                redis_sock_disconnect(slave->sock, force);
+                redis_sock_disconnect(slave->sock, force, 1);
             } ZEND_HASH_FOREACH_END();
         }
     } ZEND_HASH_FOREACH_END();
@@ -1603,7 +1603,7 @@ PHP_REDIS_API short cluster_send_command(redisCluster *c, short slot, const char
         return -1;
     } else if (timedout || resp == -1) {
         // Make sure the socket is reconnected, it such that it is in a clean state
-        redis_sock_disconnect(c->cmd_sock, 1);
+        redis_sock_disconnect(c->cmd_sock, 1, 1);
 
         if (timedout) {
             CLUSTER_THROW_EXCEPTION("Timed out attempting to find data in the correct node!", 0);

library.c

@@ -359,7 +359,8 @@ redis_check_eof(RedisSock *redis_sock, zend_bool no_retry, zend_bool no_throw)
         for (retry_index = 0; !no_retry && retry_index < redis_sock->max_retries; ++retry_index) {
             /* close existing stream before reconnecting */
             if (redis_sock->stream) {
-                redis_sock_disconnect(redis_sock, 1);
+                /* reconnect no need to reset mode, it will cause pipeline mode socket exception */
+                redis_sock_disconnect(redis_sock, 1, 0);
             }
             /* Sleep based on our backoff algorithm */
             zend_ulong delay = redis_backoff_compute(&redis_sock->backoff, retry_index);
@@ -390,7 +391,7 @@ redis_check_eof(RedisSock *redis_sock, zend_bool no_retry, zend_bool no_throw)
         }
     }
     /* close stream and mark socket as failed */
-    redis_sock_disconnect(redis_sock, 1);
+    redis_sock_disconnect(redis_sock, 1, 1);
     redis_sock->status = REDIS_SOCK_STATUS_FAILED;
     if (!no_throw) {
         REDIS_THROW_EXCEPTION( errmsg, 0);
@@ -3058,7 +3059,7 @@ PHP_REDIS_API int redis_sock_connect(RedisSock *redis_sock)
     ConnectionPool *p = NULL;
 
     if (redis_sock->stream != NULL) {
-        redis_sock_disconnect(redis_sock, 0);
+        redis_sock_disconnect(redis_sock, 0, 1);
     }
 
     address = ZSTR_VAL(redis_sock->host);
@@ -3206,7 +3207,7 @@ redis_sock_server_open(RedisSock *redis_sock)
  * redis_sock_disconnect
  */
 PHP_REDIS_API int
-redis_sock_disconnect(RedisSock *redis_sock, int force)
+redis_sock_disconnect(RedisSock *redis_sock, int force, int is_reset_mode)
 {
     if (redis_sock == NULL) {
         return FAILURE;
@@ -3228,7 +3229,9 @@ redis_sock_disconnect(RedisSock *redis_sock, int force)
         }
         redis_sock->stream = NULL;
     }
-    redis_sock->mode = ATOMIC;
+    if (is_reset_mode) {
+        redis_sock->mode = ATOMIC;
+    }
     redis_sock->status = REDIS_SOCK_STATUS_DISCONNECTED;
     redis_sock->watching = 0;
 
@@ -4107,7 +4110,7 @@ redis_sock_gets(RedisSock *redis_sock, char *buf, int buf_size, size_t *line_siz
             snprintf(buf, buf_size, "read error on connection to %s:%d", ZSTR_VAL(redis_sock->host), redis_sock->port);
         }
         // Close our socket
-        redis_sock_disconnect(redis_sock, 1);
+        redis_sock_disconnect(redis_sock, 1, 1);
 
         // Throw a read error exception
         REDIS_THROW_EXCEPTION(buf, 0);

library.h

@@ -83,7 +83,7 @@ PHP_REDIS_API char *redis_sock_auth_cmd(RedisSock *redis_sock, int *cmdlen);
 PHP_REDIS_API void redis_sock_set_auth(RedisSock *redis_sock, zend_string *user, zend_string *pass);
 PHP_REDIS_API void redis_sock_set_auth_zval(RedisSock *redis_sock, zval *zv);
 PHP_REDIS_API void redis_sock_free_auth(RedisSock *redis_sock);
-PHP_REDIS_API int redis_sock_disconnect(RedisSock *redis_sock, int force);
+PHP_REDIS_API int redis_sock_disconnect(RedisSock *redis_sock, int force, int is_reset_mode);
 PHP_REDIS_API zval *redis_sock_read_multibulk_reply_zval(RedisSock *redis_sock, zval *z_tab);
 PHP_REDIS_API int redis_sock_read_single_line(RedisSock *redis_sock, char *buffer,
     size_t buflen, size_t *linelen, int set_err);

redis.c

@@ -192,7 +192,7 @@ free_redis_object(zend_object *object)
 
     zend_object_std_dtor(&redis->std);
     if (redis->sock) {
-        redis_sock_disconnect(redis->sock, 0);
+        redis_sock_disconnect(redis->sock, 0, 1);
         redis_free_socket(redis->sock);
     }
 }
@@ -573,7 +573,7 @@ redis_connect(INTERNAL_FUNCTION_PARAMETERS, int persistent)
 
     /* if there is a redis sock already we have to remove it */
     if (redis->sock) {
-        redis_sock_disconnect(redis->sock, 0);
+        redis_sock_disconnect(redis->sock, 0, 1);
         redis_free_socket(redis->sock);
     }
 
@@ -632,7 +632,7 @@ PHP_METHOD(Redis, close)
 {
     RedisSock *redis_sock = redis_sock_get_connected(INTERNAL_FUNCTION_PARAM_PASSTHRU);
 
-    if (redis_sock_disconnect(redis_sock, 1) == SUCCESS) {
+    if (redis_sock_disconnect(redis_sock, 1, 1) == SUCCESS) {
         RETURN_TRUE;
     }
     RETURN_FALSE;

redis_session.c

@@ -111,7 +111,7 @@ redis_pool_free(redis_pool *pool) {
     rpm = pool->head;
     while (rpm) {
         next = rpm->next;
-        redis_sock_disconnect(rpm->redis_sock, 0);
+        redis_sock_disconnect(rpm->redis_sock, 0, 1);
         redis_free_socket(rpm->redis_sock);
         efree(rpm);
         rpm = next;
@@ -1103,7 +1103,7 @@ PS_UPDATE_TIMESTAMP_FUNC(rediscluster) {
     /* Attempt to send EXPIRE command */
     c->readonly = 0;
     if (cluster_send_command(c,slot,cmd,cmdlen) < 0 || c->err) {
-        php_error_docref(NULL, E_NOTICE, "Redis unable to update session expiry"); 
+        php_error_docref(NULL, E_NOTICE, "Redis unable to update session expiry");
         efree(cmd);
         return FAILURE;
     }
@@ -1146,7 +1146,7 @@ PS_READ_FUNC(rediscluster) {
         cmdlen = redis_spprintf(NULL, NULL, &cmd, "GET", "s", skey, skeylen);
         c->readonly = 1;
     }
-    
+
     efree(skey);
 
     /* Attempt to kick off our command */

sentinel_library.c

@@ -8,7 +8,7 @@ free_redis_sentinel_object(zend_object *object)
     redis_sentinel_object *obj = PHPREDIS_GET_OBJECT(redis_sentinel_object, object);
 
     if (obj->sock) {
-        redis_sock_disconnect(obj->sock, 0);
+        redis_sock_disconnect(obj->sock, 0, 1);
         redis_free_socket(obj->sock);
     }
     zend_object_std_dtor(&obj->std);