Rewrite handshake procedure

knizhnik · knizhnik · commit 13a25d7925a4 · 2016-08-05T16:19:38.000+03:00
diff --git a/contrib/mmts/arbiter.c b/contrib/mmts/arbiter.c
@@ -396,7 +396,7 @@ static int MtmConnectSocket(int node, int port, int timeout)
     unsigned i, n_addrs = sizeof(addrs) / sizeof(addrs[0]);
 	MtmHandshakeMessage req;
 	MtmArbiterMessage   resp;
-	int sd;
+	int rc, sd;
 	timestamp_t start = MtmGetSystemTime();
 	char const* host = Mtm->nodes[node].con.hostName;
 
@@ -476,7 +476,10 @@ static int MtmConnectSocket(int node, int port, int timeout)
 		close(sd);
 		goto Retry;
 	}
-	if (MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout) != 1 || MtmReadSocket(sd, &resp, sizeof resp) != sizeof(resp)) { 
+	while ((rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout)) == 0) { 
+		elog(WARNING, "Arbiter waiting response for handshake message from %s:%d: rc=%d, error=%d", host, port, rc, errno);	
+	}
+	if (rc != 1 || MtmReadSocket(sd, &resp, sizeof resp) != sizeof(resp)) { 
 		elog(WARNING, "Arbiter failed to receive response for handshake message from %s:%d: errno=%d", host, port, errno);
 		close(sd);
 		goto Retry;
diff --git a/contrib/mmts/pglogical_output.c b/contrib/mmts/pglogical_output.c
@@ -376,8 +376,8 @@ pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
 	/* If the record didn't originate locally, send origin info */
 	send_replication_origin &= txn->origin_id != InvalidRepOriginId;
 
-	OutputPluginPrepareWrite(ctx, !send_replication_origin);
 	if (data->api) { 
+		OutputPluginPrepareWrite(ctx, !send_replication_origin);
 		data->api->write_begin(ctx->out, data, txn);
 
 		if (send_replication_origin)
@@ -401,8 +401,8 @@ pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
 				replorigin_by_oid(txn->origin_id, true, &origin))
 			data->api->write_origin(ctx->out, origin, txn->origin_lsn);
 		}
+		OutputPluginWrite(ctx, true);
 	}
-	OutputPluginWrite(ctx, true);
 }
 
 /*
@@ -414,11 +414,11 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 {
 	PGLogicalOutputData* data = (PGLogicalOutputData*)ctx->output_plugin_private;
 
-	OutputPluginPrepareWrite(ctx, true);
 	if (data->api) { 
+		OutputPluginPrepareWrite(ctx, true);
 		data->api->write_commit(ctx->out, data, txn, commit_lsn);
+		OutputPluginWrite(ctx, true);
 	}
-	OutputPluginWrite(ctx, true);
 }
 
 void
@@ -522,11 +522,11 @@ send_startup_message(LogicalDecodingContext *ctx,
 	 * not.
 	 */
 
-	OutputPluginPrepareWrite(ctx, last_message);
 	if (data->api) {
+		OutputPluginPrepareWrite(ctx, last_message);
 		data->api->write_startup_message(ctx->out, msg);
+		OutputPluginWrite(ctx, last_message);
 	}
-	OutputPluginWrite(ctx, last_message);
 
 	pfree(msg);
 
diff --git a/contrib/mmts/tests2/docker-entrypoint.sh b/contrib/mmts/tests2/docker-entrypoint.sh
@@ -80,9 +80,6 @@ if [ "$1" = 'postgres' ]; then
 			checkpoint_timeout = 30
 			log_autovacuum_min_duration = 0
 
-			raftable.id = $NODE_ID
-			raftable.peers = '$RAFT_PEERS'
-
 			multimaster.workers = 4
 			multimaster.use_raftable = true
 			multimaster.queue_size=52857600
diff --git a/contrib/mmts/tests2/lib/bank_client.py b/contrib/mmts/tests2/lib/bank_client.py
@@ -151,6 +151,7 @@ def exec_tx(self, name, tx_block):
     def check_total(self):
 
         def tx(conn, cur):
+            conn.commit()
             cur.execute('select sum(amount) from bank_test')
             res = cur.fetchone()
             total = res[0]

Original file line number	Diff line number	Diff line change
`@@ -376,8 +376,8 @@ pg_decode_begin_txn(LogicalDecodingContext ctx, ReorderBufferTXN txn)`
`376`	`376`	`/* If the record didn't originate locally, send origin info */`
`377`	`377`	`send_replication_origin &= txn->origin_id != InvalidRepOriginId;`
`378`	`378`
`379`		`- OutputPluginPrepareWrite(ctx, !send_replication_origin);`
`380`	`379`	`if (data->api) {`
	`380`	`+ OutputPluginPrepareWrite(ctx, !send_replication_origin);`
`381`	`381`	`data->api->write_begin(ctx->out, data, txn);`
`382`	`382`
`383`	`383`	`if (send_replication_origin)`
`@@ -401,8 +401,8 @@ pg_decode_begin_txn(LogicalDecodingContext ctx, ReorderBufferTXN txn)`
`401`	`401`	`replorigin_by_oid(txn->origin_id, true, &origin))`
`402`	`402`	`data->api->write_origin(ctx->out, origin, txn->origin_lsn);`
`403`	`403`	`}`
	`404`	`+ OutputPluginWrite(ctx, true);`
`404`	`405`	`}`
`405`		`- OutputPluginWrite(ctx, true);`
`406`	`406`	`}`
`407`	`407`
`408`	`408`	`/*`
`@@ -414,11 +414,11 @@ pg_decode_commit_txn(LogicalDecodingContext ctx, ReorderBufferTXN txn,`
`414`	`414`	`{`
`415`	`415`	`PGLogicalOutputData* data = (PGLogicalOutputData*)ctx->output_plugin_private;`
`416`	`416`
`417`		`- OutputPluginPrepareWrite(ctx, true);`
`418`	`417`	`if (data->api) {`
	`418`	`+ OutputPluginPrepareWrite(ctx, true);`
`419`	`419`	`data->api->write_commit(ctx->out, data, txn, commit_lsn);`
	`420`	`+ OutputPluginWrite(ctx, true);`
`420`	`421`	`}`
`421`		`- OutputPluginWrite(ctx, true);`
`422`	`422`	`}`
`423`	`423`
`424`	`424`	`void`
`@@ -522,11 +522,11 @@ send_startup_message(LogicalDecodingContext *ctx,`
`522`	`522`	`* not.`
`523`	`523`	`*/`
`524`	`524`
`525`		`- OutputPluginPrepareWrite(ctx, last_message);`
`526`	`525`	`if (data->api) {`
	`526`	`+ OutputPluginPrepareWrite(ctx, last_message);`
`527`	`527`	`data->api->write_startup_message(ctx->out, msg);`
	`528`	`+ OutputPluginWrite(ctx, last_message);`
`528`	`529`	`}`
`529`		`- OutputPluginWrite(ctx, last_message);`
`530`	`530`
`531`	`531`	`pfree(msg);`
`532`	`532`