Rewrite handshake procedure

Author: knizhnik

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;

pglogical_output.c

@@ -383,8 +383,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)
@@ -408,8 +408,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);
 }
 
 /*
@@ -421,11 +421,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
@@ -541,11 +541,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);
 

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

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]