Reset memory context after each iteration of pglogical apply

Author: knizhnik

arbiter.c

@@ -724,9 +724,9 @@ static void MtmSender(Datum arg)
 	MTM_ELOG(LOG, "Start arbiter sender %d", MyProcPid);
 	InitializeTimeouts();
 
-	signal(SIGINT, SetStop);
-	signal(SIGQUIT, SetStop);
-	signal(SIGTERM, SetStop);
+	pqsignal(SIGINT, SetStop);
+	pqsignal(SIGQUIT, SetStop);
+	pqsignal(SIGTERM, SetStop);
 
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
@@ -803,9 +803,9 @@ static bool MtmRecovery()
 
 static void MtmMonitor(Datum arg)
 {
-	signal(SIGINT, SetStop);
-	signal(SIGQUIT, SetStop);
-	signal(SIGTERM, SetStop);
+	pqsignal(SIGINT, SetStop);
+	pqsignal(SIGQUIT, SetStop);
+	pqsignal(SIGTERM, SetStop);
 
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
@@ -840,9 +840,9 @@ static void MtmReceiver(Datum arg)
     max_fd = 0;
 #endif
 
-	signal(SIGINT, SetStop);
-	signal(SIGQUIT, SetStop);
-	signal(SIGTERM, SetStop);
+	pqsignal(SIGINT, SetStop);
+	pqsignal(SIGQUIT, SetStop);
+	pqsignal(SIGTERM, SetStop);
 
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();

bgwpool.c

@@ -35,9 +35,9 @@ static void BgwPoolMainLoop(BgwPool* pool)
 	MtmIsLogicalReceiver = true;
 	MtmPool = pool;
 
-	signal(SIGINT, BgwShutdownWorker);
-	signal(SIGQUIT, BgwShutdownWorker);
-	signal(SIGTERM, BgwShutdownWorker);
+	pqsignal(SIGINT, BgwShutdownWorker);
+	pqsignal(SIGQUIT, BgwShutdownWorker);
+	pqsignal(SIGTERM, BgwShutdownWorker);
 
     BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);

multimaster.c

@@ -528,8 +528,8 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
         if (ts != NULL /*&& ts->status != TRANSACTION_STATUS_IN_PROGRESS*/)
         {
             if (ts->csn > MtmTx.snapshot) { 
-                MTM_LOG4("%d: tuple with xid=%d(csn=%lld) is invisible in snapshot %lld",
-						 MyProcPid, xid, ts->csn, MtmTx.snapshot);
+                MTM_LOG4("%d: tuple with xid=%lld(csn=%lld) is invisible in snapshot %lld",
+						 MyProcPid, (long64)xid, ts->csn, MtmTx.snapshot);
 				if (MtmGetSystemTime() - start > USECS_PER_SEC) { 
 					MTM_ELOG(WARNING, "Backend %d waits for transaction %s (%llu) status %lld usecs", MyProcPid, ts->gid, (long64)xid, MtmGetSystemTime() - start);
 				}
@@ -569,8 +569,8 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
             else
             {
                 bool invisible = ts->status != TRANSACTION_STATUS_COMMITTED;
-                MTM_LOG4("%d: tuple with xid=%d(csn= %lld) is %s in snapshot %lld",
-						 MyProcPid, xid, ts->csn, invisible ? "rollbacked" : "committed", MtmTx.snapshot);
+                MTM_LOG4("%d: tuple with xid=%lld(csn= %lld) is %s in snapshot %lld",
+						 MyProcPid, (long64)xid, ts->csn, invisible ? "rollbacked" : "committed", MtmTx.snapshot);
                 MtmUnlock();
 				if (MtmGetSystemTime() - start > USECS_PER_SEC) { 
 					MTM_ELOG(WARNING, "Backend %d waits for %s transaction %s (%llu) %lld usecs", MyProcPid, invisible ? "rollbacked" : "committed", 
@@ -581,7 +581,7 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
         }
         else
         {
-            MTM_LOG4("%d: visibility check is skipped for transaction %u in snapshot %llu", MyProcPid, xid, MtmTx.snapshot);
+            MTM_LOG4("%d: visibility check is skipped for transaction %llu in snapshot %llu", MyProcPid, (long64)xid, MtmTx.snapshot);
 			MtmUnlock();
 			return PgXidInMVCCSnapshot(xid, snapshot);
         }
@@ -4945,6 +4945,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			break;
 
 		case T_DropStmt:
+	    case T_TruncateStmt:
 			{
 				DropStmt *stmt = (DropStmt *) parsetree;
 				if (stmt->removeType == OBJECT_INDEX && stmt->concurrent)

pglogical_apply.c

@@ -408,16 +408,13 @@ process_remote_message(StringInfo s)
 				if (MtmVacuumStmt != NULL) { 
 					ExecVacuum(MtmVacuumStmt, 1);
 				} else if (MtmIndexStmt != NULL) {
-					MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
 					Oid relid =	RangeVarGetRelidExtended(MtmIndexStmt->relation, ShareUpdateExclusiveLock,
 														 false, false,
 														 NULL,
 														 NULL);
 					/* Run parse analysis ... */
 					MtmIndexStmt = transformIndexStmt(relid, MtmIndexStmt, messageBody);
 
-					MemoryContextSwitchTo(oldContext);
-
 					DefineIndex(relid,		/* OID of heap relation */
 								MtmIndexStmt,
 								InvalidOid, /* no predefined OID */
@@ -618,6 +615,7 @@ read_rel(StringInfo s, LOCKMODE mode)
 	RangeVar*	rv;
 	Oid			remote_relid = pq_getmsgint(s, 4);
 	Oid         local_relid;
+	MemoryContext old_context;
 
 	local_relid = pglogical_relid_map_get(remote_relid);
 	if (local_relid == InvalidOid) { 
@@ -630,7 +628,9 @@ read_rel(StringInfo s, LOCKMODE mode)
 		rv->relname = (char *) pq_getmsgbytes(s, relnamelen);
 
 		local_relid = RangeVarGetRelidExtended(rv, mode, false, false, NULL, NULL);
+		old_context = MemoryContextSwitchTo(TopMemoryContext);
 		pglogical_relid_map_put(remote_relid, local_relid);
+		MemoryContextSwitchTo(old_context);
 	} else { 
 		nspnamelen = pq_getmsgbyte(s);
 		s->cursor += nspnamelen;
@@ -1060,7 +1060,8 @@ void MtmExecutor(void* work, size_t size)
 	int spill_file = -1;
 	int save_cursor = 0;
 	int save_len = 0;
-	MemoryContext topContext;
+	MemoryContext old_context;
+	MemoryContext top_context;
 
     s.data = work;
     s.len = size;
@@ -1074,13 +1075,15 @@ void MtmExecutor(void* work, size_t size)
 												ALLOCSET_DEFAULT_INITSIZE,
 												ALLOCSET_DEFAULT_MAXSIZE);
     }
-	topContext = MemoryContextSwitchTo(MtmApplyContext);
-
+	top_context = MemoryContextSwitchTo(MtmApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;
     PG_TRY();
     {    
-        while (true) { 
+		bool inside_transaction = true;
+        do { 
             char action = pq_getmsgbyte(&s);
+			old_context = MemoryContextSwitchTo(MtmApplyContext);
+	
             MTM_LOG2("%d: REMOTE process action %c", MyProcPid, action);
 #if 0
 			if (Mtm->status == MTM_RECOVERY) { 
@@ -1091,84 +1094,81 @@ void MtmExecutor(void* work, size_t size)
             switch (action) {
                 /* BEGIN */
             case 'B':
-			    if (process_remote_begin(&s)) { 				   
-					continue;
-				} else { 
-					break;
-				}
+			    inside_transaction = process_remote_begin(&s);
+				break;
                 /* COMMIT */
             case 'C':
   			    close_rel(rel);
                 process_remote_commit(&s);
+				inside_transaction = false;
                 break;
                 /* INSERT */
             case 'I':
-                process_remote_insert(&s, rel);
-                continue;
+			    process_remote_insert(&s, rel);
+                break;
                 /* UPDATE */
             case 'U':
                 process_remote_update(&s, rel);
-                continue;
+                break;
                 /* DELETE */
             case 'D':
                 process_remote_delete(&s, rel);
-                continue;
+                break;
             case 'R':
   			    close_rel(rel);
                 rel = read_rel(&s, RowExclusiveLock);
-                continue;
+                break;
 			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;
+				break;
 			}
  		    case '(':
 			{
 			    size_t size = pq_getmsgint(&s, 4);    
-				s.data = palloc(size);
+				s.data = MemoryContextAlloc(TopMemoryContext, size);
 				save_cursor = s.cursor;
 				save_len = s.len;
 				s.cursor = 0;
 				s.len = size;
 				MtmReadSpillFile(spill_file, s.data, size);
-				continue;
+				break;
 			}
   		    case ')':
 			{
   			    pfree(s.data);
 				s.data = work;
   			    s.cursor = save_cursor;
 				s.len = save_len;
-				continue;
+				break;
 			}
 			case 'M':
 			{
-				if (process_remote_message(&s)) { 
-					break;
-				}
-				continue;
+				inside_transaction = !process_remote_message(&s);
+				break;
 			}
 			case 'Z':
 			{
 				MtmRecoveryCompleted();
+				inside_transaction = false;
 				break;
 			}
             default:
                 MTM_ELOG(ERROR, "unknown action of type %c", action);
             }        
-            break;
-        }
+			MemoryContextSwitchTo(old_context);
+			MemoryContextResetAndDeleteChildren(MtmApplyContext);
+        } while (inside_transaction);
     }
     PG_CATCH();
     {
-		MemoryContext oldcontext;
 		MtmReleaseLock();
-		oldcontext = MemoryContextSwitchTo(MtmApplyContext);
+		old_context = MemoryContextSwitchTo(MtmApplyContext);
 		MtmHandleApplyError();
-		MemoryContextSwitchTo(oldcontext);
+		MemoryContextSwitchTo(old_context);
 		EmitErrorReport();
         FlushErrorState();
 		MTM_LOG1("%d: REMOTE begin abort transaction %llu", MyProcPid, (long64)MtmGetCurrentTransactionId());
@@ -1178,12 +1178,15 @@ void MtmExecutor(void* work, size_t size)
 		MTM_LOG2("%d: REMOTE end abort transaction %llu", MyProcPid, (long64)MtmGetCurrentTransactionId());
     }
     PG_END_TRY();
+	if (s.data != work) { 
+		pfree(s.data);
+	}
 #if 0 /* spill file is expecrted to be closed by tranaction commit or rollback */
 	if (spill_file >= 0) { 
 		MtmCloseSpillFile(spill_file);
 	}
 #endif
+	MemoryContextSwitchTo(top_context);
     MemoryContextResetAndDeleteChildren(MtmApplyContext);
-	MemoryContextSwitchTo(topContext);
 }