Fixed hang up in master/slave mode.

diff --git a/pool_process_query.c b/pool_process_query.c
index db6e50670e0ae449c3443dea0f232fea012e841d..d61cc3f190205300f3caabaabdb432f300b26b58 100644 (file)
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -322,7 +322,6 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
 
                                if (FD_ISSET(frontend->fd, &readmask))
                                {
-                                       pool_debug("XXXX");
                                        status = ProcessFrontendResponse(frontend, backend);
                                        if (status != POOL_CONTINUE)
                                                return status;
@@ -370,7 +369,7 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
                }
 
                /* this is the synchronous point */
-               if (REPLICATION || first_ready_for_query_received)
+               if (DUAL_MODE || first_ready_for_query_received)
                {
                        if (kind == 0)
                        {
@@ -618,58 +617,66 @@ static POOL_STATUS Query(POOL_CONNECTION *frontend,
                return POOL_CONTINUE;
        }
 
-       if (frontend &&
-               (strncasecmp("prepare", string, 7) == 0))
+       if (strncasecmp("prepare", string, 7) == 0)
        {
-               PreparedStatement *stmt;
-
-               stmt = get_prepared_command_portal_and_statement(string);
+               if (DUAL_MODE)
+                       force_replication = 1;          
 
-               if (stmt != NULL)
+               if (frontend)
                {
-                       pending_function = add_prepared_list;
-                       pending_prepared_stmt = stmt;
-                       if (REPLICATION)
-                               force_replication = 1;
-                       prepare_in_session = 1;
+                       PreparedStatement *stmt;
+
+                       stmt = get_prepared_command_portal_and_statement(string);
+
+                       if (stmt != NULL)
+                       {
+                               pending_function = add_prepared_list;
+                               pending_prepared_stmt = stmt;
+                               if (DUAL_MODE)
+                                       force_replication = 1;
+                               prepare_in_session = 1;
+                       }
                }
        }
-       else if (frontend &&
-                        strncasecmp("deallocate", string, 10) == 0)
+       else if (strncasecmp("deallocate", string, 10) == 0)
        {
-               char *query = string;
-               char *buf, *name;
-               query = skip_comment(query);
+               if (DUAL_MODE)
+                       force_replication = 1;          
 
-               /* skip "prepare" or "deallocate" */
-               while (*query && !isspace(*query))
-                       query++;
+               if (frontend)
+               {
+                       char *query = string;
+                       char *buf, *name;
+                       query = skip_comment(query);
 
-               /* skip spaces */
-               while (*query && isspace(*query))
-                       query++;
+                       /* skip "prepare" or "deallocate" */
+                       while (*query && !isspace(*query))
+                               query++;
 
-               buf = strdup(query);
-               name = strtok(buf, "\t\r\n (;");
+                       /* skip spaces */
+                       while (*query && isspace(*query))
+                               query++;
 
-               pending_function = del_prepared_list;
-               pending_prepared_stmt = malloc(sizeof(PreparedStatement));
-               if (pending_prepared_stmt == NULL)
-               {
-                       pool_error("SimpleForwardToBackend: malloc failed: %s", strerror(errno));
-                       return POOL_END;
-               }
+                       buf = strdup(query);
+                       name = strtok(buf, "\t\r\n (;");
 
-               pending_prepared_stmt->statement_name = normalize_prepared_stmt_name(name);
-               pending_prepared_stmt->portal_name = NULL;
-               if (pending_prepared_stmt->statement_name == NULL)
-               {
-                       pool_error("SimpleForwardToBackend: strdup failed: %s", strerror(errno));
-                       return POOL_END;
+                       pending_function = del_prepared_list;
+                       pending_prepared_stmt = malloc(sizeof(PreparedStatement));
+                       if (pending_prepared_stmt == NULL)
+                       {
+                               pool_error("SimpleForwardToBackend: malloc failed: %s", strerror(errno));
+                               return POOL_END;
+                       }
+
+                       pending_prepared_stmt->statement_name = normalize_prepared_stmt_name(name);
+                       pending_prepared_stmt->portal_name = NULL;
+                       if (pending_prepared_stmt->statement_name == NULL)
+                       {
+                               pool_error("SimpleForwardToBackend: strdup failed: %s", strerror(errno));
+                               return POOL_END;
+                       }
+                       free(buf);
                }
-               free(buf);
-               if (REPLICATION)
-                       force_replication = 1;
        }
 
        if (frontend &&
@@ -1107,7 +1114,7 @@ static POOL_STATUS Parse(POOL_CONNECTION *frontend,
                else if (send_extended_protocol_message(cp, "P", len, string))
                        return POOL_END;
 
-               if (!REPLICATION)
+               if (!DUAL_MODE)
                        break;
                else if (pool_config.replication_strict)
                {
@@ -1241,7 +1248,7 @@ static POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
                /* set transaction state */
                pool_debug("ReadyForQuery: transaction state: %c", state);
                MASTER(backend)->tstate = state;
-               if (REPLICATION)
+               if (DUAL_MODE)
                        SECONDARY(backend)->tstate = state;
 
                for (i = 0;i < backend->num;i++)
@@ -1261,15 +1268,34 @@ static POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
 
                pool_debug("ReadyForQuery: message length: %d", len);
 
-               state = pool_read_kind(backend);
-               if (state < 0)
-                       return POOL_END;
 
-               /* set transaction state */
-               pool_debug("ReadyForQuery: transaction state: %c", state);
-               MASTER(backend)->tstate = state;
-               if (REPLICATION)
+               /*
+                * Do not check transaction state in master/slave mode.
+                * Because SET, PREPARE, DEALLOCATE are replicated.
+                * If these queries are executed inside a transaction block,
+                * transation state is inconsistency. But it is no problem.
+                */
+               if (MASTER_SLAVE)
+               {
+                       char kind1;
+
+                       pool_read(MASTER(backend), &state, 1);
+                       MASTER(backend)->tstate = state;
+                       pool_read(SECONDARY(backend), &kind1, 1);
                        SECONDARY(backend)->tstate = state;
+               }
+               else
+               {
+                       state = pool_read_kind(backend);
+                       if (state < 0)
+                               return POOL_END;
+
+                       /* set transaction state */
+                       pool_debug("ReadyForQuery: transaction state: %c", state);
+                       MASTER(backend)->tstate = state;
+                       if (REPLICATION)
+                               SECONDARY(backend)->tstate = state;
+               }
        }
 
        if (send_ready)
@@ -2196,7 +2222,7 @@ static POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
                default:
                        if (MAJOR(backend) == PROTO_MAJOR_V3)
                        {
-                               if (MASTER_SLAVE)
+                               if (MASTER_SLAVE && TSTATE(backend) != 'I')
                                {
                                        master_slave_was_enabled = 1;
                                        MASTER_SLAVE = 0;
@@ -2205,7 +2231,7 @@ static POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
                                status = SimpleForwardToBackend(fkind, frontend, backend);
                                if (pool_flush(MASTER(backend)))
                                        status = POOL_ERROR;
-                               if (REPLICATION)
+                               if (DUAL_MODE)
                                        if (pool_flush(SECONDARY(backend)))
                                                status = POOL_ERROR;
                        }
@@ -2808,7 +2834,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
                return POOL_END;
        }
 
-       if (REPLICATION)
+       if (REPLICATION || MASTER_SLAVE)
        {
                status = pool_read(SECONDARY(backend), &len1, sizeof(len1));
                if (status < 0)
@@ -2835,7 +2861,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
        if (p == NULL)
                return POOL_END;
 
-       if (REPLICATION)
+       if (DUAL_MODE)
        {
                len1 = ntohl(len1) - 4;
                if (len1 <= 0)
@@ -2913,7 +2939,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
                                return POOL_END;
                        }
 
-                       if (!REPLICATION)
+                       if (!DUAL_MODE)
                                break;
                }
 
@@ -2944,7 +2970,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
                if (p1 == NULL)
                        return POOL_END;
        
-               if (REPLICATION)
+               if (DUAL_MODE)
                {
                        status = pool_read(SECONDARY(backend), &res2, sizeof(res2));
                        if (status < 0)
@@ -2981,7 +3007,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
 
        if (pool_write(MASTER(backend), &kind, 1))
                return POOL_END;
-       if (REPLICATION)
+       if (DUAL_MODE)
                if (pool_write(SECONDARY(backend), &kind, 1))
                        return POOL_END;
 
@@ -2992,7 +3018,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
 
        if (pool_write(MASTER(backend), &sendlen, sizeof(sendlen)))
                return POOL_END;
-       if (REPLICATION)
+       if (DUAL_MODE)
                if (pool_write(SECONDARY(backend), &sendlen, sizeof(sendlen)))
                        return POOL_END;
 
@@ -3012,7 +3038,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
 
        if (pool_write(MASTER(backend), p, len))
                return POOL_END;
-       if (REPLICATION)
+       if (DUAL_MODE)
                if (pool_write(SECONDARY(backend), p, len))
                        return POOL_END;
 
@@ -3096,7 +3122,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
                                return POOL_END;
                        }
 
-                       if (!REPLICATION)
+                       if (!DUAL_MODE)
                                break;
                }
 
@@ -3144,7 +3170,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
                if (pool_flush(MASTER(backend)))
                        return POOL_END;
 
-               if (REPLICATION)
+               if (DUAL_MODE)
                        if (pool_flush(SECONDARY(backend)))
                                return POOL_END;
        }