Merge branch 'master' of github.com:postgrespro/postgres_cluster

Author: kelvich

README

@@ -1,4 +1,4 @@
-PostgreSQL Database Management System
+pPostgreSQL Database Management System
 =====================================
 
 This directory contains the source code distribution of the PostgreSQL

contrib/mmts/bgwpool.c

@@ -16,9 +16,6 @@ typedef struct
     int id;
 } BgwPoolExecutorCtx;
 
-size_t n_snapshots;
-size_t n_active;
-
 static void BgwPoolMainLoop(Datum arg)
 {
     BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)arg;
@@ -36,7 +33,8 @@ static void BgwPoolMainLoop(Datum arg)
         size = *(int*)&pool->queue[pool->head];
         Assert(size < pool->size);
         work = malloc(size);
-        pool->active -= 1;
+        pool->pending -= 1;
+        pool->active += 1;
         if (pool->head + size + 4 > pool->size) { 
             memcpy(work, pool->queue, size);
             pool->head = INTALIGN(size);
@@ -54,6 +52,9 @@ static void BgwPoolMainLoop(Datum arg)
         SpinLockRelease(&pool->lock);
         pool->executor(id, work, size);
         free(work);
+        SpinLockAcquire(&pool->lock);
+        pool->active -= 1;
+        SpinLockRelease(&pool->lock);
     }
 }
 
@@ -71,6 +72,7 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, si
     pool->tail = 0;
     pool->size = queueSize;
     pool->active = 0;
+    pool->pending = 0;
     strcpy(pool->dbname, dbname);
 }
 
@@ -126,9 +128,7 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
             PGSemaphoreLock(&pool->overflow);
             SpinLockAcquire(&pool->lock);
         } else {
-            pool->active += 1;
-            n_snapshots += 1;
-            n_active += pool->active;
+            pool->pending += 1;
             *(int*)&pool->queue[pool->tail] = size;
             if (pool->size - pool->tail >= size + 4) { 
                 memcpy(&pool->queue[pool->tail+4], work, size);

contrib/mmts/bgwpool.h

@@ -20,6 +20,7 @@ typedef struct
     size_t tail;
     size_t size;
     size_t active;
+    size_t pending;
     bool   producerBlocked;
     char   dbname[MAX_DBNAME_LEN];
     char*  queue;

contrib/mmts/multimaster--1.0.sql

@@ -24,18 +24,22 @@ AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
 
-CREATE TYPE mtm.node_state AS (id integer, disabled bool, disconnected bool, catchUp bool, slotLag bigint, avgTransDelay bigint, lastStatusChange timestamp, connStr text);
+CREATE TYPE mtm.node_state AS ("id" integer, "disabled" bool, "disconnected" bool, "catchUp" bool, "slotLag" bigint, "avgTransDelay" bigint, "lastStatusChange" timestamp, "connStr" text);
 
 CREATE FUNCTION mtm.get_nodes_state() RETURNS SETOF mtm.node_state
 AS 'MODULE_PATHNAME','mtm_get_nodes_state'
 LANGUAGE C;
 
-CREATE TYPE mtm.cluster_state AS (status text, disabledNodeMask bigint, disconnectedNodeMask bigint, catchUpNodeMask bigint, nNodes integer, nActiveQueries integer, queueSize bigint, transCount bigint, timeShift bigint, recoverySlot integer);
+CREATE TYPE mtm.cluster_state AS ("status" text, "disabledNodeMask" bigint, "disconnectedNodeMask" bigint, "catchUpNodeMask" bigint, "nNodes" integer, "nActiveQueries" integer, "nPendingQueries" integer, "queueSize" bigint, "transCount" bigint, "timeShift" bigint, "recoverySlot" integer);
 
 CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state 
 AS 'MODULE_PATHNAME','mtm_get_cluster_state'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.get_cluster_info() RETURNS SETOF mtm.cluster_state 
+AS 'MODULE_PATHNAME','mtm_get_cluster_info'
+LANGUAGE C;
+
 CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;

contrib/mmts/multimaster.c

@@ -108,6 +108,7 @@ PG_FUNCTION_INFO_V1(mtm_recover_node);
 PG_FUNCTION_INFO_V1(mtm_get_snapshot);
 PG_FUNCTION_INFO_V1(mtm_get_nodes_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_state);
+PG_FUNCTION_INFO_V1(mtm_get_cluster_info);
 PG_FUNCTION_INFO_V1(mtm_make_table_local);
 PG_FUNCTION_INFO_V1(mtm_dump_lock_graph);
 
@@ -166,7 +167,8 @@ char const* const MtmNodeStatusMnem[] =
 	"Connected",
 	"Online",
 	"Recovery",
-	"InMinor"
+	"InMinor",
+	"OutOfService"
 };
 
 bool  MtmDoReplication;
@@ -1014,6 +1016,26 @@ void MtmAbortTransaction(MtmTransState* ts)
  * -------------------------------------------
  */
 
+void MtmHandleApplyError(void)
+{
+	ErrorData *edata = CopyErrorData();
+	switch (edata->sqlerrcode) { 
+	  case ERRCODE_DISK_FULL:
+	  case ERRCODE_INSUFFICIENT_RESOURCES:
+	  case ERRCODE_IO_ERROR:
+	  case ERRCODE_DATA_CORRUPTED:
+	  case ERRCODE_INDEX_CORRUPTED:
+	  case ERRCODE_SYSTEM_ERROR:
+	  case ERRCODE_INTERNAL_ERROR:
+	  case ERRCODE_OUT_OF_MEMORY:		  
+		  elog(WARNING, "Node is excluded from cluster because of non-recoverable error %d", edata->sqlerrcode);
+		  MtmSwitchClusterMode(MTM_OUT_OF_SERVICE);
+		  kill(PostmasterPid, SIGQUIT);
+		  break;
+	}
+}
+
+
 void MtmRecoveryCompleted(void)
 {
 	MTM_LOG1("Recovery of node %d is completed", MtmNodeId);
@@ -1609,7 +1631,7 @@ _PG_init(void)
 		"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",
 		"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
 		&Mtm2PCMinTimeout,
-		10000,
+		100000, /* 100 seconds */
 		0,
 		INT_MAX,
 		PGC_BACKEND,
@@ -1624,7 +1646,7 @@ _PG_init(void)
 		"Percent of prepare time for maximal time of second phase of two-pahse commit",
 		"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
 		&Mtm2PCPrepareRatio,
-		100,
+		1000, /* 10 times */
 		0,
 		INT_MAX,
 		PGC_BACKEND,
@@ -2178,10 +2200,9 @@ mtm_get_snapshot(PG_FUNCTION_ARGS)
 typedef struct
 {
 	int       nodeId;
-	char*     connStrPtr;
 	TupleDesc desc;
-    Datum     values[8];
-    bool      nulls[8];
+    Datum     values[Natts_mtm_nodes_state];
+    bool      nulls[Natts_mtm_nodes_state];
 } MtmGetNodeStateCtx;
 
 Datum
@@ -2190,7 +2211,6 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
     FuncCallContext* funcctx;
 	MtmGetNodeStateCtx* usrfctx;
 	MemoryContext oldcontext;
-	char* p;
 	int64 lag;
     bool is_first_call = SRF_IS_FIRSTCALL();
 
@@ -2200,7 +2220,6 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 		usrfctx = (MtmGetNodeStateCtx*)palloc(sizeof(MtmGetNodeStateCtx));
 		get_call_result_type(fcinfo, NULL, &usrfctx->desc);
 		usrfctx->nodeId = 1;
-		usrfctx->connStrPtr = pstrdup(MtmConnStrs);
 		memset(usrfctx->nulls, false, sizeof(usrfctx->nulls));
 		funcctx->user_fctx = usrfctx;
 		MemoryContextSwitchTo(oldcontext);      
@@ -2219,23 +2238,19 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	usrfctx->nulls[4] = lag < 0;
 	usrfctx->values[5] = Int64GetDatum(Mtm->transCount ? Mtm->nodes[usrfctx->nodeId-1].transDelay/Mtm->transCount : 0);
 	usrfctx->values[6] = TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[usrfctx->nodeId-1].lastStatusChangeTime));
-	p = strchr(usrfctx->connStrPtr, ',');
-	if (p != NULL) { 
-		*p++ = '\0';
-	}
-	usrfctx->values[7] = CStringGetTextDatum(usrfctx->connStrPtr);
-	usrfctx->connStrPtr = p;
+	usrfctx->values[7] = CStringGetTextDatum(Mtm->nodes[usrfctx->nodeId-1].con.connStr);
 	usrfctx->nodeId += 1;
 
 	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(heap_form_tuple(usrfctx->desc, usrfctx->values, usrfctx->nulls)));
 }
 
+
 Datum
 mtm_get_cluster_state(PG_FUNCTION_ARGS)
 {
 	TupleDesc desc;
-    Datum     values[10];
-    bool      nulls[10] = {false};
+    Datum     values[Natts_mtm_cluster_state];
+    bool      nulls[Natts_mtm_cluster_state] = {false};
 	get_call_result_type(fcinfo, NULL, &desc);
 
 	values[0] = CStringGetTextDatum(MtmNodeStatusMnem[Mtm->status]);
@@ -2244,16 +2259,73 @@ mtm_get_cluster_state(PG_FUNCTION_ARGS)
 	values[3] = Int64GetDatum(Mtm->nodeLockerMask);
 	values[4] = Int32GetDatum(Mtm->nNodes);
 	values[5] = Int32GetDatum((int)Mtm->pool.active);
-	values[6] = Int64GetDatum(BgwPoolGetQueueSize(&Mtm->pool));
-	values[7] = Int64GetDatum(Mtm->transCount);
-	values[8] = Int64GetDatum(Mtm->timeShift);
-	values[9] = Int32GetDatum(Mtm->recoverySlot);
-	nulls[9] = Mtm->recoverySlot == 0;
+	values[6] = Int32GetDatum((int)Mtm->pool.pending);
+	values[7] = Int64GetDatum(BgwPoolGetQueueSize(&Mtm->pool));
+	values[8] = Int64GetDatum(Mtm->transCount);
+	values[9] = Int64GetDatum(Mtm->timeShift);
+	values[10] = Int32GetDatum(Mtm->recoverySlot);
 
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc, values, nulls)));
 }
 
 
+typedef struct
+{
+	int       nodeId;
+} MtmGetClusterInfoCtx;
+
+
+Datum
+mtm_get_cluster_info(PG_FUNCTION_ARGS)
+{
+
+    FuncCallContext* funcctx;
+	MtmGetClusterInfoCtx* usrfctx;
+	MemoryContext oldcontext;
+	TupleDesc desc;
+    bool is_first_call = SRF_IS_FIRSTCALL();
+	int i;
+	PGconn* conn;
+	PGresult *result;
+	char* values[Natts_mtm_cluster_state];
+	HeapTuple tuple;
+
+    if (is_first_call) { 
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);       
+		usrfctx = (MtmGetClusterInfoCtx*)palloc(sizeof(MtmGetNodeStateCtx));
+		get_call_result_type(fcinfo, NULL, &desc);
+		funcctx->attinmeta = TupleDescGetAttInMetadata(desc);
+		usrfctx->nodeId = 1;
+		funcctx->user_fctx = usrfctx;
+		MemoryContextSwitchTo(oldcontext);      
+    }
+    funcctx = SRF_PERCALL_SETUP();	
+	usrfctx = (MtmGetClusterInfoCtx*)funcctx->user_fctx;
+	if (usrfctx->nodeId > MtmNodes) {
+		SRF_RETURN_DONE(funcctx);      
+	}	
+	conn = PQconnectdb(Mtm->nodes[usrfctx->nodeId-1].con.connStr);
+	if (PQstatus(conn) != CONNECTION_OK) {
+		elog(ERROR, "Failed to establish connection '%s' to node %d", Mtm->nodes[usrfctx->nodeId-1].con.connStr, usrfctx->nodeId);
+	}
+	result = PQexec(conn, "select * from mtm.get_cluster_state()");
+
+	if (PQresultStatus(result) != PGRES_TUPLES_OK || PQntuples(result) != 1) { 
+		elog(ERROR, "Failed to receive data from %d", usrfctx->nodeId);
+	}
+
+	for (i = 0; i < Natts_mtm_cluster_state; i++) { 
+		values[i] = PQgetvalue(result, 0, i);
+	}
+	tuple = BuildTupleFromCStrings(funcctx->attinmeta, values);
+	PQclear(result);
+	PQfinish(conn);
+	usrfctx->nodeId += 1;
+	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+}
+
+
 Datum mtm_make_table_local(PG_FUNCTION_ARGS)
 {
 	Oid	reloid = PG_GETARG_OID(1);

contrib/mmts/multimaster.h

@@ -56,6 +56,9 @@
 #define Anum_mtm_local_tables_rel_schema 1
 #define Anum_mtm_local_tables_rel_name	 2
 
+#define Natts_mtm_cluster_state 11
+#define Natts_mtm_nodes_state   8
+
 typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)
 
@@ -94,11 +97,12 @@ typedef enum
 typedef enum
 {
 	MTM_INITIALIZATION, /* Initial status */
-	MTM_OFFLINE,        /* Node is out of quorum */
+	MTM_OFFLINE,        /* Node is excluded from cluster */
 	MTM_CONNECTED,      /* Arbiter is established connections with other nodes */
 	MTM_ONLINE,         /* Ready to receive client's queries */
 	MTM_RECOVERY,       /* Node is in recovery process */
-	MTM_IN_MINORITY     /* Node is out of quorum */
+	MTM_IN_MINORITY,    /* Node is out of quorum */
+	MTM_OUT_OF_SERVICE  /* Node is not avaiable to to critical, non-recoverable error */
 } MtmNodeStatus;
 
 typedef enum
@@ -235,5 +239,6 @@ extern void  MtmCheckQuorum(void);
 extern bool  MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN);
 extern void  MtmRecoveryCompleted(void);
 extern void  MtmMakeTableLocal(char* schema, char* name);
+extern void  MtmHandleApplyError(void);
 
 #endif

contrib/mmts/pglogical_apply.c

@@ -951,6 +951,7 @@ void MtmExecutor(int id, void* work, size_t size)
     }
     PG_CATCH();
     {
+		MtmHandleApplyError();
 		EmitErrorReport();
         FlushErrorState();
 		MTM_LOG2("%d: REMOTE begin abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());