Spill applied transactions to the disk

Author: knizhnik

Makefile

@@ -1,5 +1,5 @@
 MODULE_big = multimaster
-OBJS = multimaster.o raftable.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o
+OBJS = multimaster.o raftable.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o
 
 override CPPFLAGS += -I../raftable
 
@@ -15,7 +15,7 @@ all: multimaster.so
 tests/dtmbench:
 	make -C tests
 
-PG_CPPFLAGS = -I$(libpq_srcdir) -DUSE_PGLOGICAL_OUTPUT
+PG_CPPFLAGS = -I$(libpq_srcdir)
 SHLIB_LINK = $(libpq)
 
 ifdef USE_PGXS

multimaster.c

@@ -184,6 +184,7 @@ int   MtmConnectTimeout;
 int   MtmKeepaliveTimeout;
 int   MtmReconnectAttempts;
 int   MtmNodeDisableDelay;
+int   MtmTransSpillThreshold;
 bool  MtmUseRaftable;
 bool  MtmUseDtm;
 MtmConnectionInfo* MtmConnections;
@@ -1247,6 +1248,7 @@ MtmBuildConnectivityMatrix(nodemask_t* matrix, bool nowait)
 			matrix[i] |= ((matrix[j] >> i) & 1) << j;
 			matrix[j] |= ((matrix[i] >> j) & 1) << i;
 		}
+		matrix[i] &= ~((nodemask_t)1 << i);
 	}
 	return true;
 }	
@@ -1630,6 +1632,21 @@ _PG_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		return;
 
+	DefineCustomIntVariable(
+		"multimaster.trans_spill_threshold",
+		"Maximal size (Mb) of transaction after which transaction is written to the disk",
+		NULL,
+		&MtmTransSpillThreshold,
+		1000, /* 1Gb */
+		0,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.twopc_min_timeout",
 		"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",

multimaster.h

@@ -205,6 +205,7 @@ extern int   MtmConnectTimeout;
 extern int   MtmReconnectAttempts;
 extern int   MtmKeepaliveTimeout;
 extern int   MtmNodeDisableDelay;
+extern int   MtmTransSpillThreshold;
 extern bool  MtmUseDtm;
 extern HTAB* MtmXid2State;
 

pglogical_apply.c

@@ -50,6 +50,7 @@
 
 #include "multimaster.h"
 #include "pglogical_relid_map.h"
+#include "spill.h"
 
 typedef struct TupleData
 {
@@ -901,6 +902,9 @@ void MtmExecutor(int id, void* work, size_t size)
 {
     StringInfoData s;
     Relation rel = NULL;
+	int spill_file = -1;
+	int save_cursor;
+	int save_len;
     s.data = work;
     s.len = size;
     s.maxlen = -1;
@@ -944,6 +948,33 @@ void MtmExecutor(int id, void* work, size_t size)
             case 'R':
                 rel = read_rel(&s, RowExclusiveLock);
                 continue;
+			case 'F':
+			{
+				int node_id = pq_getmsgint(&s, 4);
+				int file_id = pq_getmsgint(&s, 4);
+				Assert(spill_file < 0);
+				spill_file = MtmOpenSpillFile(node_id, file_id);
+				continue;
+			}
+ 		    case '(':
+			{
+			    int64 size = pq_getmsgint(&s, 4);    
+				s.data = palloc(size);
+				save_cursor = s.cursor;
+				save_len = s.len;
+				s.cursor = 0;
+				s.len = size;
+				MtmReadSpillFile(spill_file, s.data, size);
+				continue;
+			}
+  		    case ')':
+			{
+  			    pfree(s.data);
+				s.data = work;
+  			    s.cursor = save_cursor;
+				s.len = save_len;
+				continue;
+			}
             default:
                 elog(ERROR, "unknown action of type %c", action);
             }        
@@ -963,7 +994,9 @@ void MtmExecutor(int id, void* work, size_t size)
 		MTM_LOG2("%d: REMOTE end abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());
     }
     PG_END_TRY();
-
+	if (spill_file >= 0) { 
+		MtmCloseSpillFile(spill_file);
+	}
     MemoryContextResetAndDeleteChildren(ApplyContext);
 }
 

pglogical_receiver.c

@@ -24,6 +24,7 @@
 #include "access/clog.h"
 #include "access/transam.h"
 #include "lib/stringinfo.h"
+#include "libpq/pqformat.h"
 #include "pgstat.h"
 #include "postmaster/bgworker.h"
 #include "storage/ipc.h"
@@ -35,6 +36,7 @@
 #include "replication/origin.h"
 
 #include "multimaster.h"
+#include "spill.h"
 
 /* Allow load of this module in shared libs */
 
@@ -213,20 +215,23 @@ pglogical_receiver_main(Datum main_arg)
 	PGresult *res;
 	MtmSlotMode mode;
 
-#ifndef USE_PGLOGICAL_OUTPUT
-    bool insideTrans = false;
-#endif
     ByteBuffer buf;
 	XLogRecPtr originStartPos = 0;
 	RepOriginId originId;
 	char* originName;
 	/* Buffer for COPY data */
 	char	*copybuf = NULL;
+	int spill_file = -1;
+	StringInfoData spill_info;
+
+	initStringInfo(&spill_info);
 
 	/* Register functions for SIGTERM/SIGHUP management */
 	pqsignal(SIGHUP, receiver_raw_sighup);
 	pqsignal(SIGTERM, receiver_raw_sigterm);
 
+	MtmCreateSpillDirectory(args->remote_node);
+
     sprintf(worker_proc, "mtm_pglogical_receiver_%d_%d", args->local_node, args->remote_node);
 
 	/* We're now ready to receive signals */
@@ -449,34 +454,38 @@ pglogical_receiver_main(Datum main_arg)
 			if (rc > hdr_len)
 			{
                 stmt = copybuf + hdr_len;
-           
-#ifdef USE_PGLOGICAL_OUTPUT
+				
+				if (buf.used >= MtmTransSpillThreshold) { 
+					if (spill_file < 0) {
+						int file_id;
+						spill_file = MtmCreateSpillFile(args->remote_node, &file_id);
+						pq_sendbyte(&spill_info, 'F');
+						pq_sendint(&spill_info, args->remote_node, 4);
+						pq_sendint(&spill_info, file_id, 4);
+					}
+					ByteBufferAppend(&buf, ")", 1);
+					pq_sendbyte(&spill_info, '(');
+					pq_sendint(&spill_info, buf.used, 4);
+					MtmSpillToFile(spill_file, buf.data, buf.used);
+					ByteBufferReset(&buf);
+				}
                 ByteBufferAppend(&buf, stmt, rc - hdr_len);
                 if (stmt[0] == 'C') /* commit */
-                { 
-					MtmExecute(buf.data, buf.used);
+                {
+					if (spill_file >= 0) { 
+						ByteBufferAppend(&buf, ")", 1);
+						pq_sendbyte(&spill_info, '(');
+						pq_sendint(&spill_info, buf.used, 4);
+						MtmSpillToFile(spill_file, buf.data, buf.used);
+						MtmCloseSpillFile(spill_file);
+						MtmExecute(spill_info.data, spill_info.len);
+						spill_file = -1;
+						resetStringInfo(&spill_info);
+					} else { 
+						MtmExecute(buf.data, buf.used);
+					}
                     ByteBufferReset(&buf);
                 }
-#else
-                if (strncmp(stmt, "BEGIN ", 6) == 0) { 
-                    TransactionId xid;
-                    int rc = sscanf(stmt + 6, "%u", &xid);
-                    Assert(rc == 1);
-                    ByteBufferAppendInt32(&buf, xid);
-                    Assert(!insideTrans);
-                    insideTrans = true;
-                } else if (strncmp(stmt, "COMMIT;", 7) == 0) { 
-                    Assert(insideTrans);
-                    Assert(buf.used > 4);
-                    buf.data[buf.used-1] = '\0'; /* replace last ';' with '\0' to make string zero terminated */
-                    MMExecute(buf.data, buf.used);
-                    ByteBufferReset(&buf);
-                    insideTrans = false;
-                } else {
-                    Assert(insideTrans);
-                    ByteBufferAppend(&buf, stmt, rc - hdr_len/*strlen(stmt)*/);
-                }
-#endif
             }
 			/* Update written position */
 			output_written_lsn = Max(walEnd, output_written_lsn);
@@ -575,6 +584,7 @@ void MtmStartReceivers(void)
 {
     int i;
 	BackgroundWorker worker;
+
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
     worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
@@ -586,6 +596,7 @@ void MtmStartReceivers(void)
             ReceiverArgs* ctx = (ReceiverArgs*)palloc(sizeof(ReceiverArgs));
             ctx->receiver_conn_string = psprintf("replication=database %s", MtmConnections[i].connStr);
             sprintf(ctx->receiver_slot, MULTIMASTER_SLOT_PATTERN, MtmNodeId);
+		
             ctx->local_node = MtmNodeId;
             ctx->remote_node = i+1;
 
@@ -598,45 +609,3 @@ void MtmStartReceivers(void)
     }
 }
 
-#ifndef USE_PGLOGICAL_OUTPUT
-void MMExecutor(int id, void* work, size_t size)
-{
-    TransactionId xid = *(TransactionId*)work;
-    char* stmts = (char*)work + 4;
-    bool finished = false;
-
-    MMJoinTransaction(xid);
-
-    SetCurrentStatementStartTimestamp();               
-    StartTransactionCommand();
-    SPI_connect();
-    PushActiveSnapshot(GetTransactionSnapshot());
-
-    PG_TRY();
-    {
-        int rc = SPI_execute(stmts, false, 0);
-        SPI_finish();
-        PopActiveSnapshot();
-        finished = true;
-        if (rc != SPI_OK_INSERT && rc != SPI_OK_UPDATE && rc != SPI_OK_DELETE) {
-            ereport(LOG, (errmsg("Executor %d: failed to apply transaction %u",
-                                 id, xid)));
-            AbortCurrentTransaction();
-        } else { 
-            CommitTransactionCommand();
-        }
-    }
-    PG_CATCH();
-    {
-        FlushErrorState();
-        if (!finished) {
-            SPI_finish();
-            if (ActiveSnapshotSet()) { 
-                PopActiveSnapshot();
-            }
-        }
-        AbortCurrentTransaction();
-    }
-    PG_END_TRY();
-}
-#endif