Support snapshot requests without valid GXID

This allows callers to request snapshots without first obtaining a GXID from
the GTM. This should help read-only transactions.

In passing, also make some adjustments to the proxy code and add a check for
messages that are proxied between proxy and GTM. Otherwise a mistmatch between
message proxying and later response processing can cause hard-to-find bugs
(like one we srtuggled while creating this patch)

We still want some more enhancements so that a snapshot once obtained can be
recorded at the GTM so that subsequent requests can get the same snapshot, for
example for serializable transactions

diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index c1e41ca6b97d74e265e31d990dba89ba7494f1b4..61e5eff5869b0e93205d30ed4eed8160e6e9afd8 100644 (file)
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -3061,7 +3061,7 @@ GetSnapshotDataFromGTM(Snapshot snapshot)
        GTM_Snapshot gtm_snapshot;
        bool canbe_grouped = (!FirstSnapshotSet) || (!IsolationUsesXactSnapshot());
 
-       gtm_snapshot = GetSnapshotGTM(GetCurrentTransactionId(), canbe_grouped);
+       gtm_snapshot = GetSnapshotGTM(GetCurrentTransactionIdIfAny(), canbe_grouped);
        
        if (!gtm_snapshot)
                ereport(ERROR,

diff --git a/src/gtm/client/fe-protocol.c b/src/gtm/client/fe-protocol.c
index 59467df238891cfd9f9f147dd0412cc91acce7bc..d18c531effc859016a3740a257681d01deee1ab6 100644 (file)
--- a/src/gtm/client/fe-protocol.c
+++ b/src/gtm/client/fe-protocol.c
@@ -763,7 +763,7 @@ gtmpqParseSuccess(GTM_Conn *conn, GTM_Result *result)
                                {
                                        result->gr_status = GTM_RESULT_ERROR;
                                        printfGTMPQExpBuffer(&conn->errorMessage, "buffer size not large enough for node list data");
-                       result->gr_status = GTM_RESULT_ERROR;
+                                       result->gr_status = GTM_RESULT_ERROR;
                                }
 
                                if (gtmpqGetnchar((char *)&buf, size, conn))

diff --git a/src/gtm/client/gtm_client.c b/src/gtm/client/gtm_client.c
index 382f6336c70f827d811e74f9cecc1410ca43bdf8..d4cc9d63247dad365a2e60348b2311068fd28a9b 100644 (file)
--- a/src/gtm/client/gtm_client.c
+++ b/src/gtm/client/gtm_client.c
@@ -971,8 +971,7 @@ get_snapshot(GTM_Conn *conn, GlobalTransactionId gxid, bool canbe_grouped)
 
         /* Start the message. */
        if (gtmpqPutMsgStart('C', true, conn) ||
-               gtmpqPutInt(MSG_SNAPSHOT_GET, sizeof (GTM_MessageType), conn) ||
-               gtmpqPutc(canbe_grouped, conn) ||
+               gtmpqPutInt(canbe_grouped ? MSG_SNAPSHOT_GET_MULTI : MSG_SNAPSHOT_GET, sizeof (GTM_MessageType), conn) ||
                gtmpqPutnchar((char *)&gxid, sizeof (GlobalTransactionId), conn))
                goto send_failed;
 

diff --git a/src/gtm/main/gtm_snap.c b/src/gtm/main/gtm_snap.c
index fa7e3ccc2df5d19dcfafa72c95577b8df7561b64..24651c8d562de66ee642d066919f97681eb3976f 100644 (file)
--- a/src/gtm/main/gtm_snap.c
+++ b/src/gtm/main/gtm_snap.c
@@ -23,6 +23,7 @@
 #include "gtm/libpq-int.h"
 #include "gtm/pqformat.h"
 
+static GTM_SnapshotData localSnapshot;
 /*
  * Get snapshot for the given transactions. If this is the first call in the
  * transaction, a fresh snapshot is taken and returned back. For a serializable
@@ -71,25 +72,29 @@ GTM_GetTransactionSnapshot(GTM_TransactionHandle handle[], int txn_count, int *s
 
        for (ii = 0; ii < txn_count; ii++)
        {
-               mygtm_txninfo = GTM_HandleToTransactionInfo(handle[ii]);
+               /*
+                * Even if the request does not contain a valid GXID, we still send
+                * down a snapshot, but mark the status field acoordingly
+                */
+               if (handle[ii] != InvalidTransactionHandle)
+                       mygtm_txninfo = GTM_HandleToTransactionInfo(handle[ii]);
+               else
+                       status[ii] = STATUS_NOT_FOUND;
 
                /*
                 * If the transaction does not exist, just mark the status field with
                 * a STATUS_ERROR code
                 */
-               if (mygtm_txninfo == NULL)
-                       status[ii] = STATUS_ERROR;
-               else if (snapshot == NULL)
+               if ((mygtm_txninfo != NULL) && (snapshot == NULL))
                        snapshot = &mygtm_txninfo->gti_current_snapshot;
        }
 
        /*
-        * If no valid transaction exists in the array, send an error message back.
-        * Otherwise, we should still get the snapshot and send it back. The
-        * invalid transaction ids are marked separately in the status array.
+        * If no valid transaction exists in the array, we record the snapshot in a
+        * local strucure and still send it out to the caller
         */
        if (snapshot == NULL)
-               return NULL;
+               snapshot = &localSnapshot;
 
        Assert(snapshot != NULL);
 
@@ -198,7 +203,7 @@ GTM_GetTransactionSnapshot(GTM_TransactionHandle handle[], int txn_count, int *s
                 * We have already gone through all the transaction handles above and
                 * marked the invalid handles with STATUS_ERROR
                 */
-               if (status[ii] == STATUS_ERROR)
+               if ((status[ii] == STATUS_ERROR) || (status[ii] == STATUS_NOT_FOUND))
                        continue;
 
                mygtm_txninfo = GTM_HandleToTransactionInfo(handle[ii]);
@@ -285,13 +290,6 @@ ProcessGetSnapshotCommand(Port *myport, StringInfo message, bool get_gxid)
        int txn_count = 1;
        const char *data = NULL;
 
-       /*
-        * Here we consume a byte which is a boolean to determine if snapshot can
-        * be grouped or not. This is used only by GTM-Proxy and it is useless for GTM
-        * so consume data.
-        */
-       pq_getmsgbyte(message);
-
        data = pq_getmsgbytes(message, sizeof (gxid));
        if (data == NULL)
                ereport(ERROR,

diff --git a/src/gtm/main/gtm_txn.c b/src/gtm/main/gtm_txn.c
index 06a099d95ced6e471253282db9c9330c8208acfb..3307e4d49be9d84c5bfa979df78f25065beca6a3 100644 (file)
--- a/src/gtm/main/gtm_txn.c
+++ b/src/gtm/main/gtm_txn.c
@@ -155,6 +155,9 @@ GTM_GXIDToHandle(GlobalTransactionId gxid)
        gtm_ListCell *elem = NULL;
        GTM_TransactionInfo *gtm_txninfo = NULL;
 
+       if (!GlobalTransactionIdIsValid(gxid))
+               return InvalidTransactionHandle;
+
        GTM_RWLockAcquire(&GTMTransactions.gt_TransArrayLock, GTM_LOCKMODE_READ);
 
        gtm_foreach(elem, GTMTransactions.gt_open_transactions)

diff --git a/src/gtm/proxy/proxy_main.c b/src/gtm/proxy/proxy_main.c
index 2bb55342fe0e5af338ee300472054d45289b5279..fbb6cfa8c47add422db4eacda7d0c037d448aa68 100644 (file)
--- a/src/gtm/proxy/proxy_main.c
+++ b/src/gtm/proxy/proxy_main.c
@@ -160,10 +160,6 @@ static void ProcessTransactionCommand(GTMProxy_ConnectionInfo *conninfo,
                GTM_Conn *gtm_conn, GTM_MessageType mtype, StringInfo message);
 static void ProcessSnapshotCommand(GTMProxy_ConnectionInfo *conninfo,
                GTM_Conn *gtm_conn, GTM_MessageType mtype, StringInfo message);
-static void ProcessSequenceCommand(GTMProxy_ConnectionInfo *conninfo,
-               GTM_Conn *gtm_conn, GTM_MessageType mtype, StringInfo message);
-static void ProcessBarrierCommand(GTMProxy_ConnectionInfo *conninfo,
-               GTM_Conn *gtm_conn, GTM_MessageType mtype, StringInfo message);
 
 static void GTMProxy_RegisterPGXCNode(GTMProxy_ConnectionInfo *conninfo,
                                                                          char *node_name,
@@ -189,6 +185,7 @@ static void UnregisterProxy(void);
 static GTM_Conn *ConnectGTM(void);
 static void ReleaseCmdBackup(GTMProxy_CommandInfo *cmdinfo);
 static void workerThreadReconnectToGTM(void);
+static bool IsProxiedMessage(GTM_MessageType mtype);
 
 /*
  * One-time initialization. It's called immediately after the main process
@@ -1581,47 +1578,47 @@ ProcessCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
 
        switch (mtype)
        {
-               case MSG_NODE_REGISTER:
-               case MSG_NODE_UNREGISTER:
+               case MSG_TXN_BEGIN_GETGXID_AUTOVACUUM:
+               case MSG_TXN_PREPARE:
+               case MSG_TXN_START_PREPARED:
+               case MSG_TXN_GET_GID_DATA:
+               case MSG_TXN_COMMIT_PREPARED:
+               case MSG_SNAPSHOT_GET:
+               case MSG_SEQUENCE_INIT:
+               case MSG_SEQUENCE_GET_CURRENT:
+               case MSG_SEQUENCE_GET_NEXT:
+               case MSG_SEQUENCE_GET_LAST:
+               case MSG_SEQUENCE_SET_VAL:
+               case MSG_SEQUENCE_RESET:
+               case MSG_SEQUENCE_CLOSE:
+               case MSG_SEQUENCE_RENAME:
+               case MSG_SEQUENCE_ALTER:
+               case MSG_BARRIER:
 #ifdef XCP
                case MSG_REGISTER_SESSION:
 #endif
+                       GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, input_message);
+                       break;
+
+
+               case MSG_NODE_REGISTER:
+               case MSG_NODE_UNREGISTER:
                        ProcessPGXCNodeCommand(conninfo, gtm_conn, mtype, input_message);
                        break;
 
                case MSG_TXN_BEGIN:
                case MSG_TXN_BEGIN_GETGXID:
-               case MSG_TXN_BEGIN_GETGXID_AUTOVACUUM:
-               case MSG_TXN_PREPARE:
-               case MSG_TXN_START_PREPARED:
                case MSG_TXN_COMMIT:
-               case MSG_TXN_COMMIT_PREPARED:
                case MSG_TXN_ROLLBACK:
                case MSG_TXN_GET_GXID:
-               case MSG_TXN_GET_GID_DATA:
                        ProcessTransactionCommand(conninfo, gtm_conn, mtype, input_message);
                        break;
 
-               case MSG_SNAPSHOT_GET:
+               case MSG_SNAPSHOT_GET_MULTI:
                case MSG_SNAPSHOT_GXID_GET:
                        ProcessSnapshotCommand(conninfo, gtm_conn, mtype, input_message);
                        break;
 
-               case MSG_SEQUENCE_INIT:
-               case MSG_SEQUENCE_GET_CURRENT:
-               case MSG_SEQUENCE_GET_NEXT:
-               case MSG_SEQUENCE_GET_LAST:
-               case MSG_SEQUENCE_SET_VAL:
-               case MSG_SEQUENCE_RESET:
-               case MSG_SEQUENCE_CLOSE:
-               case MSG_SEQUENCE_RENAME:
-               case MSG_SEQUENCE_ALTER:
-                       ProcessSequenceCommand(conninfo, gtm_conn, mtype, input_message);
-                       break;
-               case MSG_BARRIER:
-                       ProcessBarrierCommand(conninfo, gtm_conn, mtype, input_message);
-                       break;
-
                default:
                        ereport(FATAL,
                                        (EPROTO,
@@ -1730,6 +1727,42 @@ HandlePostCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn)
 
 }
 
+static bool
+IsProxiedMessage(GTM_MessageType mtype)
+{
+       switch (mtype)
+       {
+               case MSG_TXN_BEGIN:
+               case MSG_TXN_BEGIN_GETGXID_AUTOVACUUM:
+               case MSG_TXN_PREPARE:
+               case MSG_TXN_START_PREPARED:
+               /* There are not so many 2PC from application messages, so just proxy it. */
+               case MSG_TXN_COMMIT_PREPARED:
+               case MSG_TXN_GET_GXID:
+               case MSG_TXN_GET_GID_DATA:
+               case MSG_NODE_REGISTER:
+               case MSG_NODE_UNREGISTER:
+#ifdef XCP
+               case MSG_REGISTER_SESSION:
+#endif
+               case MSG_SNAPSHOT_GXID_GET:
+               case MSG_SEQUENCE_INIT:
+               case MSG_SEQUENCE_GET_CURRENT:
+               case MSG_SEQUENCE_GET_NEXT:
+               case MSG_SEQUENCE_GET_LAST:
+               case MSG_SEQUENCE_SET_VAL:
+               case MSG_SEQUENCE_RESET:
+               case MSG_SEQUENCE_CLOSE:
+               case MSG_SEQUENCE_RENAME:
+               case MSG_SEQUENCE_ALTER:
+               case MSG_SNAPSHOT_GET:
+                       return true;
+
+               default:
+                       return false;
+       }
+}
+
 static void
 ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
                GTM_Result *res)
@@ -1737,6 +1770,7 @@ ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
        StringInfoData buf;
        GlobalTransactionId gxid;
        GTM_Timestamp timestamp;
+       int     status;
 
        switch (cmdinfo->ci_mtype)
        {
@@ -1855,7 +1889,7 @@ ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
                        ReleaseCmdBackup(cmdinfo);
                        break;
 
-               case MSG_SNAPSHOT_GET:
+               case MSG_SNAPSHOT_GET_MULTI:
                        if ((res->gr_type != SNAPSHOT_GET_RESULT) &&
                                (res->gr_type != SNAPSHOT_GET_MULTI_RESULT))
                        {
@@ -1866,10 +1900,12 @@ ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
                        if (cmdinfo->ci_res_index >= res->gr_resdata.grd_txn_snap_multi.txn_count)
                        {
                                ReleaseCmdBackup(cmdinfo);
-                               elog(ERROR, "Too few GXIDs");
+                               elog(ERROR, "Too few GXIDs - %d:%d", cmdinfo->ci_res_index,
+                                               res->gr_resdata.grd_txn_snap_multi.txn_count);
                        }
 
-                       if (res->gr_resdata.grd_txn_snap_multi.status[cmdinfo->ci_res_index] == STATUS_OK)
+                       status = res->gr_resdata.grd_txn_snap_multi.status[cmdinfo->ci_res_index];
+                       if ((status == STATUS_OK) || (status == STATUS_NOT_FOUND))
                        {
                                int txn_count = 1;
                                int status = STATUS_OK;
@@ -1920,6 +1956,8 @@ ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
                case MSG_SEQUENCE_CLOSE:
                case MSG_SEQUENCE_RENAME:
                case MSG_SEQUENCE_ALTER:
+               case MSG_SNAPSHOT_GET:
+                       Assert(IsProxiedMessage(cmdinfo->ci_mtype));
                        if ((res->gr_proxyhdr.ph_conid == InvalidGTMProxyConnID) ||
                                (res->gr_proxyhdr.ph_conid >= GTM_PROXY_MAX_CONNECTIONS) ||
                                (thrinfo->thr_all_conns[res->gr_proxyhdr.ph_conid] != cmdinfo->ci_conn))
@@ -2237,11 +2275,6 @@ ProcessPGXCNodeCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
                        GTMProxy_ProxyPGXCNodeCommand(conninfo, gtm_conn, mtype, cmd_data);
                        break;
                }
-#ifdef XCP
-               case MSG_REGISTER_SESSION:
-                       GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, message);
-                       break;
-#endif
                default:
                        Assert(0);                      /* Shouldn't come here.. Keep compiler quiet */
        }
@@ -2282,14 +2315,6 @@ ProcessTransactionCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
                        elog(FATAL, "Support not yet added for these message types");
                        break;
 
-               case MSG_TXN_BEGIN_GETGXID_AUTOVACUUM:
-               case MSG_TXN_PREPARE:
-               case MSG_TXN_START_PREPARED:
-               case MSG_TXN_GET_GID_DATA:
-               case MSG_TXN_COMMIT_PREPARED:
-                       GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, message);
-                       break;
-
                default:
                        Assert(0);                      /* Shouldn't come here.. keep compiler quiet */
        }
@@ -2299,16 +2324,11 @@ static void
 ProcessSnapshotCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
                GTM_MessageType mtype, StringInfo message)
 {
-       bool canbe_grouped = false;
        GTMProxy_CommandData cmd_data;
 
        switch (mtype)
        {
-               case MSG_SNAPSHOT_GET:
-                       canbe_grouped = pq_getmsgbyte(message);
-                       if (!canbe_grouped)
-                               GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, message);
-                       else
+               case MSG_SNAPSHOT_GET_MULTI:
                        {
                                {
                                        const char *data = pq_getmsgbytes(message,
@@ -2334,35 +2354,6 @@ ProcessSnapshotCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
 
 }
 
-static void
-ProcessSequenceCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
-               GTM_MessageType mtype, StringInfo message)
-{
-       /*
-        * We proxy the Sequence messages as they are. Just add the connection
-        * identifier to it so that the response can be quickly sent back to the
-        * right backend.
-        *
-        * Write the message, but don't flush it just yet.
-        */
-       return GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, message);
-}
-
-static void
-ProcessBarrierCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
-               GTM_MessageType mtype, StringInfo message)
-{
-       /*
-        * We proxy the Barrier messages as they are. Just add the connection
-        * identifier to it so that the response can be quickly sent back to the
-        * right backend.
-        *
-        * Write the message, but don't flush it just yet.
-        */
-       return GTMProxy_ProxyCommand(conninfo, gtm_conn, mtype, message);
-}
-
-
 /*
  * Proxy the incoming message to the GTM server after adding our own identifier
  * to it. The rest of the message is forwarded as it is without even reading
@@ -2375,9 +2366,11 @@ GTMProxy_ProxyCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
        GTMProxy_CommandInfo *cmdinfo;
        GTMProxy_ThreadInfo *thrinfo = GetMyThreadInfo;
        GTM_ProxyMsgHeader proxyhdr;
-       char *unreadmsg;
+       const char *unreadmsg;
        int unreadmsglen;
 
+       Assert(IsProxiedMessage(mtype));
+
        proxyhdr.ph_conid = conninfo->con_id;
 
        unreadmsglen = pq_getmsgunreadlen(message);
@@ -2388,7 +2381,7 @@ GTMProxy_ProxyCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
                gtmpqPutnchar((char *)&proxyhdr, sizeof (GTM_ProxyMsgHeader), gtm_conn) ||
                gtmpqPutInt(mtype, sizeof (GTM_MessageType), gtm_conn) ||
                gtmpqPutnchar(unreadmsg, unreadmsglen, gtm_conn))
-               elog(ERROR, "Error proxing data");
+               elog(ERROR, "Error sending proxied message");
 
        /*
         * Add the message to the pending command list
@@ -2764,7 +2757,7 @@ GTMProxy_ProcessPendingCommands(GTMProxy_ThreadInfo *thrinfo)
                                thrinfo->thr_pending_commands[ii] = gtm_NIL;
                                break;
 
-                       case MSG_SNAPSHOT_GET:
+                       case MSG_SNAPSHOT_GET_MULTI:
                                if (gtmpqPutInt(MSG_SNAPSHOT_GET_MULTI, sizeof (GTM_MessageType), gtm_conn) ||
                                        gtmpqPutInt(gtm_list_length(thrinfo->thr_pending_commands[ii]), sizeof(int), gtm_conn))
                                        elog(ERROR, "Error sending data");

diff --git a/src/include/c.h b/src/include/c.h
index df22d50d4e40826520b3b50292199b71ca0aa670..15e2501cb87b6c6ddc971fd742af25f17e1a69d4 100644 (file)
--- a/src/include/c.h
+++ b/src/include/c.h
@@ -873,6 +873,7 @@ typedef NameData *Name;
 #define STATUS_EOF                             (-2)
 #define STATUS_FOUND                   (1)
 #define STATUS_WAITING                 (2)
+#define STATUS_NOT_FOUND               (3)
 
 
 /*