Merge branch 'more_tests' of github.com:postgrespro/postgres_cluster into more_tests

Author: kelvich

contrib/mmts/Cluster.pm

@@ -233,7 +233,7 @@ sub stop
 			}
 		}
 	}
-
+	sleep(2);
 	return $ok;
 }
 

contrib/mmts/bgwpool.c

@@ -5,6 +5,7 @@
 #include "postmaster/bgworker.h"
 #include "storage/s_lock.h"
 #include "storage/spin.h"
+#include "storage/proc.h"
 #include "storage/pg_sema.h"
 #include "storage/shmem.h"
 #include "datatype/timestamp.h"
@@ -16,23 +17,41 @@
 bool MtmIsLogicalReceiver;
 int  MtmMaxWorkers;
 
+static BgwPool* pool;
+
+static void BgwShutdownWorker(int sig)
+{
+	BgwPoolStop(pool);
+}
+
 static void BgwPoolMainLoop(BgwPool* pool)
 {
     int size;
     void* work;
 	static PortalData fakePortal;
+	sigset_t sset;
 
 	MtmIsLogicalReceiver = true;
 
+	signal(SIGINT, BgwShutdownWorker);
+	signal(SIGQUIT, BgwShutdownWorker);
+	signal(SIGTERM, BgwShutdownWorker);
+
+	sigfillset(&sset);
+	sigprocmask(SIG_UNBLOCK, &sset, NULL);
+
     BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);
 	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 
-    while(true) { 
+    while (true) { 
         PGSemaphoreLock(&pool->available);
         SpinLockAcquire(&pool->lock);
+		if (pool->shutdown) { 
+			break;
+		}
         size = *(int*)&pool->queue[pool->head];
         Assert(size < pool->size);
         work = malloc(size);
@@ -64,6 +83,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 		pool->lastPeakTime = 0;
         SpinLockRelease(&pool->lock);
     }
+	SpinLockRelease(&pool->lock);
 }
 
 void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  char const* dbuser, size_t queueSize, size_t nWorkers)
@@ -75,6 +95,7 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  c
     PGSemaphoreReset(&pool->available);
     PGSemaphoreReset(&pool->overflow);
     SpinLockInit(&pool->lock);
+	pool->shutdown = false;
     pool->producerBlocked = false;
     pool->head = 0;
     pool->tail = 0;
@@ -167,7 +188,7 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 	}
 
     SpinLockAcquire(&pool->lock);
-    while (true) { 
+    while (!pool->shutdown) { 
         if ((pool->head <= pool->tail && pool->size - pool->tail < size + 4 && pool->head < size) 
             || (pool->head > pool->tail && pool->head - pool->tail < size + 4))
         {
@@ -204,3 +225,11 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
     SpinLockRelease(&pool->lock);            
 }
 
+void BgwPoolStop(BgwPool* pool)
+{
+    SpinLockAcquire(&pool->lock);
+	pool->shutdown = true;
+    SpinLockRelease(&pool->lock);            
+	PGSemaphoreUnlock(&pool->available);
+	PGSemaphoreUnlock(&pool->overflow);
+}

contrib/mmts/bgwpool.h

@@ -34,6 +34,7 @@ typedef struct
 	time_t lastPeakTime;
 	timestamp_t lastDynamicWorkerStartTime;
     bool   producerBlocked;
+	bool   shutdown;
     char   dbname[MAX_DBNAME_LEN];
 	char   dbuser[MAX_DBUSER_LEN];
     char*  queue;
@@ -51,4 +52,5 @@ extern size_t BgwPoolGetQueueSize(BgwPool* pool);
 
 extern timestamp_t BgwGetLastPeekTime(BgwPool* pool);
 
+extern void BgwPoolStop(BgwPool* pool);
 #endif

contrib/mmts/multimaster.c

@@ -583,7 +583,9 @@ MtmAdjustOldestXid(TransactionId xid)
 
 		for (ts = Mtm->transListHead; 
 			 ts != NULL 
+				 && (ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_COMMITTED) 
 				 && ts->csn < oldestSnapshot
+				 && !ts->isPinned
 				 && TransactionIdPrecedes(ts->xid, xid);
 			 prev = ts, ts = ts->next) 
 		{ 
@@ -653,6 +655,7 @@ static void MtmAddSubtransactions(MtmTransState* ts, TransactionId* subxids, int
         sts = (MtmTransState*)hash_search(MtmXid2State, &subxids[i], HASH_ENTER, &found);
         Assert(!found);
 		sts->isActive = false;
+		sts->isPinned = false;
         sts->status = ts->status;
         sts->csn = ts->csn;
 		sts->votingCompleted = true;
@@ -814,6 +817,7 @@ MtmCreateTransState(MtmCurrentTrans* x)
 	ts->isLocal = true;
 	ts->isPrepared = false;
 	ts->isTwoPhase = x->isTwoPhase;
+	ts->isPinned = false;
 	ts->votingCompleted = false;
 	if (!found) {
 		ts->isEnqueued = false;
@@ -963,8 +967,13 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 {
 	int result = 0;
 	int nConfigChanges = Mtm->nConfigChanges;
-	timestamp_t elapsed, start = MtmGetSystemTime();
-	timestamp_t deadline = 0;
+	timestamp_t prepareTime = ts->csn - ts->snapshot;
+	timestamp_t timeout = Max(prepareTime + MSEC_TO_USEC(MtmMin2PCTimeout), prepareTime*MtmMax2PCRatio/100);
+	timestamp_t deadline = MtmGetSystemTime() + timeout;
+	timestamp_t now;
+	
+	Assert(ts->csn > ts->snapshot);
+
 	/* Wait votes from all nodes until: */
 	while (!MtmVotingCompleted(ts)
 		   && (ts->isPrepared || nConfigChanges == Mtm->nConfigChanges))
@@ -980,19 +989,16 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 		if (result & WL_LATCH_SET) { 
 			ResetLatch(&MyProc->procLatch);			
 		}
-		elapsed = MtmGetSystemTime() - start;
+		now = MtmGetSystemTime();
 		MtmLock(LW_EXCLUSIVE);
-		if (deadline == 0 && ts->votedMask != 0) { 
-			deadline = Max(MSEC_TO_USEC(MtmMin2PCTimeout), elapsed*MtmMax2PCRatio/100);
-		} else { 
+		if (now > deadline) { 
 			if (ts->isPrepared) { 
 				/* resend precommit message */
 				MtmSend2PCMessage(ts, MSG_PRECOMMIT);
 			} else {
-				if (elapsed > deadline) { 
-					elog(WARNING, "Commit of distributed transaction is canceled because of %ld msec timeout expiration", USEC_TO_MSEC(elapsed));
-					MtmAbortTransaction(ts);
-				}
+				elog(WARNING, "Commit of distributed transaction is canceled because of %ld msec timeout expiration", USEC_TO_MSEC(timeout));
+				MtmAbortTransaction(ts);
+				break;
 			}
 		}
 	}
@@ -1005,7 +1011,7 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 		} else { 
 			if (Mtm->status != MTM_ONLINE) { 
 				elog(WARNING, "Commit of distributed transaction is canceled because node is switched to %s mode", MtmNodeStatusMnem[Mtm->status]);
-			} else if (nConfigChanges != Mtm->nConfigChanges) { 
+			} else { 
 				elog(WARNING, "Commit of distributed transaction is canceled because cluster configuration was changed");
 			} 
 			MtmAbortTransaction(ts);
@@ -1202,6 +1208,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				ts->status = TRANSACTION_STATUS_ABORTED;
 				ts->isLocal = true;
 				ts->isPrepared = false;
+				ts->isPinned = false;
 				ts->snapshot = x->snapshot;
 				ts->isTwoPhase = x->isTwoPhase;
 				ts->csn = MtmAssignCSN();	
@@ -1280,7 +1287,7 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 	}
 }
 
-void MtmBroadcastPollMessage(MtmTransState* ts)
+static void MtmBroadcastPollMessage(MtmTransState* ts)
 {
 	int i;
 	MtmArbiterMessage msg;
@@ -1293,7 +1300,7 @@ void MtmBroadcastPollMessage(MtmTransState* ts)
 
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
-		if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask, i))
+		if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask, i))
 		{
 			msg.node = i+1;
 			MTM_LOG3("Send request for transaction %s to node %d", msg.gid, msg.node);
@@ -1480,15 +1487,17 @@ static void MtmPollStatusOfPreparedTransactions(int disabledNodeId)
 			Assert(ts->gid[0]);
 			if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) { 
 				elog(LOG, "Abort transaction %s because its coordinator is disabled and it is not prepared at node %d", ts->gid, MtmNodeId);
-				//MtmUnlock();
+				ts->isPinned = true;
+				MtmUnlock();
 				MtmFinishPreparedTransaction(ts, false);
-				//MtmLock(LW_EXCLUSIVE);				
+				MtmLock(LW_EXCLUSIVE);				
+				ts->isPinned = false;
 			} else {
 				MTM_LOG1("Poll state of transaction %d (%s)", ts->xid, ts->gid);				
 				MtmBroadcastPollMessage(ts);
 			}
 		} else {
-			MTM_LOG2("Skip transaction %d (%s) with status %d gtid.node=%d gtid.xid=%d votedMask=%lx", 
+			MTM_LOG1("Skip transaction %d (%s) with status %d gtid.node=%d gtid.xid=%d votedMask=%lx", 
 					 ts->xid, ts->gid, ts->status, ts->gtid.node, ts->gtid.xid, ts->votedMask);
 		}
 	}
@@ -3216,8 +3225,13 @@ bool MtmFilterTransaction(char* record, int size)
 		duplicate = true;
 	}
 
-	MTM_LOG2("%s transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
-			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, flags, origin_node, origin_lsn, restart_lsn);
+	if (duplicate) {
+		MTM_LOG1("Ignore transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
+				 gid, replication_node, end_lsn, flags, origin_node, origin_lsn, restart_lsn);
+	} else {
+		MTM_LOG2("Apply transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
+				 gid, replication_node, end_lsn, flags, origin_node, origin_lsn, restart_lsn);
+	}
 	return duplicate;
 }
 
@@ -3831,7 +3845,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 		} else { 	
 			CommitTransactionCommand();
 			if (x->isSuspended) { 
-				elog(WARNING, "Transaction %s is left in prepared state because coordinator onde is not online", x->gid);
+				elog(WARNING, "Transaction %s is left in prepared state because coordinator node is not online", x->gid);
 			} else { 				
 				StartTransactionCommand();
 				if (x->status == TRANSACTION_STATUS_ABORTED) { 

contrib/mmts/multimaster.h

@@ -223,7 +223,8 @@ typedef struct MtmTransState
 	bool           isEnqueued;         /* Transaction is inserted in queue */
 	bool           isPrepared;         /* Transaction is prepared: now it is safe to commit transaction */
 	bool           isActive;           /* Transaction is active */
-	bool           isTwoPhase;         /* user level 2PC */
+	bool           isTwoPhase;         /* User level 2PC */
+	bool           isPinned;           /* Transaction oid potected from GC */
 	nodemask_t     participantsMask;   /* Mask of nodes involved in transaction */
 	nodemask_t     votedMask;          /* Mask of voted nodes */
 	TransactionId  xids[1];            /* [Mtm->nAllNodes]: transaction ID at replicas */
@@ -331,7 +332,6 @@ extern void  MtmExecutor(void* work, size_t size);
 extern void  MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd);
 extern void  MtmSendMessage(MtmArbiterMessage* msg);
 extern void  MtmAdjustSubtransactions(MtmTransState* ts);
-extern void  MtmBroadcastPollMessage(MtmTransState* ts);
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
 extern void  MtmLockNode(int nodeId, LWLockMode mode);

contrib/mmts/tests2/lib/bank_client.py

@@ -111,20 +111,24 @@ def status(self):
         while self.running:
             msg = yield from self.child_pipe.coro_recv()
             if msg == 'status':
-                # print('evloop: got status request')
-                serialized_aggs = {}
-                for name, aggregate in self.aggregates.items():
-                    serialized_aggs[name] = aggregate.as_dict() 
-                    aggregate.clear_values()
+                serialized_aggs = []
+
+                for conn_id, conn_aggs in self.aggregates.items():
+                    serialized_aggs.append({})
+                    for aggname, agg in conn_aggs.items():
+                        serialized_aggs[conn_id][aggname] = agg.as_dict()
+                        agg.clear_values()
+
                 self.child_pipe.send(serialized_aggs)
-                # print('evloop: sent status response')
             else:
                 print('evloop: unknown message')
 
     @asyncio.coroutine
     def exec_tx(self, tx_block, aggname_prefix, conn_i):
         aggname = "%s_%i" % (aggname_prefix, conn_i)
-        agg = self.aggregates[aggname] = MtmTxAggregate(aggname)
+        if conn_i not in self.aggregates:
+            self.aggregates[conn_i] = {}
+        agg = self.aggregates[conn_i][aggname_prefix] = MtmTxAggregate(aggname)
         pool = yield from aiopg.create_pool(self.dsns[conn_i])
         conn = yield from pool.acquire()
         cur = yield from conn.cursor()
@@ -167,8 +171,8 @@ def total_tx(self, conn, cur, agg):
         total = yield from cur.fetchone()
         if total[0] != 0:
             agg.isolation += 1
-            # print(self.oops)
-            # print('Isolation error, total = ', total[0])
+            print(self.oops)
+            print('Isolation error, total = ', total[0])
             # yield from cur.execute('select * from mtm.get_nodes_state()')
             # nodes_state = yield from cur.fetchall()
             # for i, col in enumerate(self.nodes_state_fields):
@@ -177,7 +181,6 @@ def total_tx(self, conn, cur, agg):
             #          print("%19s" % nodes_state[j][i], end="\t")
             #     print("\n")
 
-
     def run(self):
         # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
         self.loop = asyncio.get_event_loop()
@@ -196,13 +199,9 @@ def bgrun(self):
         self.evloop_process = multiprocessing.Process(target=self.run, args=())
         self.evloop_process.start()
 
-    # XXX: introduce periodic report from client?
     def get_aggregates(self, print=True):
-        # print('test: sending status request')
         self.parent_pipe.send('status')
-        # print('test: awaitng status response')
         resp = self.parent_pipe.recv()
-        # print('test: got status response')
         if print:
             MtmClient.print_aggregates(resp)
         return resp
@@ -216,7 +215,7 @@ def stop(self):
         self.evloop_process.terminate()
 
     @classmethod
-    def print_aggregates(cls, serialized_agg):
+    def print_aggregates(cls, aggs):
             columns = ['running_latency', 'max_latency', 'isolation', 'finish']
 
             # print table header
@@ -225,23 +224,17 @@ def print_aggregates(cls, serialized_agg):
                 print(col, end="\t")
             print("\n", end="")
 
-            serialized_agg
-
-            for aggname in sorted(serialized_agg.keys()):
-                agg = serialized_agg[aggname]
-                print("%s\t" % aggname, end="")
-                for col in columns:
-                    if col in agg:
+            for conn_id, agg_conn in enumerate(aggs):
+                for aggname, agg in agg_conn.items():
+                    print("Node %d: %s\t" % (conn_id + 1, aggname), end="")
+                    for col in columns:
                         if isinstance(agg[col], float):
                             print("%.2f\t" % (agg[col],), end="\t")
                         else:
                             print(agg[col], end="\t")
-                    else:
-                        print("-\t", end="")
-                print("")
+                    print("")
             print("")
 
-
 if __name__ == "__main__":
     c = MtmClient(['dbname=postgres user=postgres host=127.0.0.1',
         'dbname=postgres user=postgres host=127.0.0.1 port=5433',