pglogical_receiver.c

/*-------------------------------------------------------------------------
 *
 * receiver_raw.c
 *		Receive and apply logical changes generated by decoder_raw. This
 *		creates some basics for a multi-master cluster using vanilla
 *		PostgreSQL without modifying its code.
 *
 * Copyright (c) 1996-2015, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		receiver_raw/receiver_raw.c
 *
 *-------------------------------------------------------------------------
 */

/* Some general headers for custom bgworker facility */

#include <unistd.h>
#include "postgres.h"
#include "fmgr.h"
#include "miscadmin.h"
#include "pg_socket.h"
#include "pqexpbuffer.h"
#include "access/xact.h"
#include "access/clog.h"
#include "access/transam.h"
#include "lib/stringinfo.h"
#include "libpq/pqformat.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/proc.h"
#include "utils/guc.h"
#include "utils/snapmgr.h"
#include "utils/memutils.h"
#include "executor/spi.h"
#include "replication/origin.h"
#include "utils/portal.h"
#include "tcop/pquery.h"
#include "libpq-int.h"

#include "multimaster.h"
#include "spill.h"
#include "state.h"
#include "bgwpool.h"

#define ERRCODE_DUPLICATE_OBJECT_STR  "42710"
#define RECEIVER_SUSPEND_TIMEOUT (1*USECS_PER_SEC)

/* Signal handling */
static volatile sig_atomic_t got_sigterm = false;
static volatile sig_atomic_t got_sighup = false;

/* GUC variables */
static int receiver_idle_time = 0;
static bool receiver_sync_mode = true; /* We need sync mode to have up-to-date values of catalog_xmin in replication slots */

/* Worker name */
static char worker_proc[BGW_MAXLEN];

/* Lastly written positions */
static lsn_t output_written_lsn = INVALID_LSN;
lsn_t MtmSenderWalEnd;

/* Stream functions */
static void fe_sendint64(int64 i, char *buf);
static int64 fe_recvint64(char *buf);

static void
receiver_raw_sigterm(SIGNAL_ARGS)
{
	int save_errno = errno;
	got_sigterm = true;
	if (MyProc)
		SetLatch(&MyProc->procLatch);
	errno = save_errno;
}

static void
receiver_raw_sighup(SIGNAL_ARGS)
{
	int save_errno = errno;
	got_sighup = true;
	if (MyProc)
		SetLatch(&MyProc->procLatch);
	errno = save_errno;
}

/*
 * Send a Standby Status Update message to server.
 */
static bool
sendFeedback(PGconn *conn, int64 now, int node_id)
{
	char		replybuf[1 + 8 + 8 + 8 + 8 + 1];
	int		 len = 0;
	lsn_t output_applied_lsn = output_written_lsn;
	lsn_t output_flushed_lsn = MtmGetFlushPosition(node_id);

	replybuf[len] = 'r';
	len += 1;
	fe_sendint64(output_written_lsn, &replybuf[len]);	/* write */
	len += 8;
	fe_sendint64(output_flushed_lsn, &replybuf[len]);	/* flush */
	len += 8;
	fe_sendint64(output_applied_lsn, &replybuf[len]);	/* apply */
	len += 8;
	fe_sendint64(now, &replybuf[len]);	/* sendTime */
	len += 8;

	/* No reply requested from server */
	replybuf[len] = 0;
	len += 1;

	if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
	{
		ereport(LOG, (MTM_ERRMSG("%s: could not send feedback packet: %s",
							 worker_proc, PQerrorMessage(conn))));
		return false;
	}

	return true;
}

/*
 * Converts an int64 to network byte order.
 */
static void
fe_sendint64(int64 i, char *buf)
{
	uint32	  n32;

	/* High order half first, since we're doing MSB-first */
	n32 = (uint32) (i >> 32);
	n32 = htonl(n32);
	memcpy(&buf[0], &n32, 4);

	/* Now the low order half */
	n32 = (uint32) i;
	n32 = htonl(n32);
	memcpy(&buf[4], &n32, 4);
}

/*
 * Converts an int64 from network byte order to native format.
 */
static int64
fe_recvint64(char *buf)
{
	int64	   result;
	uint32	  h32;
	uint32	  l32;

	memcpy(&h32, buf, 4);
	memcpy(&l32, buf + 4, 4);
	h32 = ntohl(h32);
	l32 = ntohl(l32);

	result = h32;
	result <<= 32;
	result |= l32;

	return result;
}

static int64
feGetCurrentTimestamp(void)
{
	int64	   result;
	struct timeval tp;

	gettimeofday(&tp, NULL);

	result = (int64) tp.tv_sec -
		((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);

	result = (result * USECS_PER_SEC) + tp.tv_usec;

	return result;
}

static void
feTimestampDifference(int64 start_time, int64 stop_time,
					  long *secs, int *microsecs)
{
	int64	   diff = stop_time - start_time;

	if (diff <= 0)
	{
		*secs = 0;
		*microsecs = 0;
	}
	else
	{
		*secs = (long) (diff / USECS_PER_SEC);
		*microsecs = (int) (diff % USECS_PER_SEC);
	}
}

static char const* const MtmReplicationModeName[] =
{
	"exit",
	"recovered",   /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
	"recovery",	   /* perform recorvery of the node by applying all data from theslot from specified point */
	"create_new",  /* destination node is recovered: drop old slot and restart from roveredLsn position */
	"open_existed" /* normal mode: use existed slot or create new one and start receiving data from it from the rememered position */
};

static void
pglogical_receiver_main(Datum main_arg)
{
	int nodeId = DatumGetInt32(main_arg);
	/* Variables for replication connection */
	PQExpBuffer query;
	PGconn *conn;
	PGresult *res;
	MtmReplicationMode mode;

	ByteBuffer buf;
	/* Buffer for COPY data */
	char	*copybuf = NULL;
	int spill_file = -1;
	StringInfoData spill_info;
	char *slotName;
	char* connString = psprintf("replication=database %s", Mtm->nodes[nodeId-1].con.connStr);
	static PortalData fakePortal;
	int i;

	MtmBackgroundWorker = true;

	ByteBufferAlloc(&buf);

	slotName = psprintf(MULTIMASTER_SLOT_PATTERN, MtmNodeId);

	MtmIsLogicalReceiver = true;

	initStringInfo(&spill_info);

	/* Register functions for SIGTERM/SIGHUP management */
	pqsignal(SIGHUP, receiver_raw_sighup);
	pqsignal(SIGTERM, receiver_raw_sigterm);

	MtmCreateSpillDirectory(nodeId);

	Mtm->nodes[nodeId-1].receiverPid = MyProcPid;
	Mtm->nodes[nodeId-1].receiverStartTime = MtmGetSystemTime();
	MtmReplicationNodeId = nodeId;

	sprintf(worker_proc, "mtm_pglogical_receiver_%d_%d", MtmNodeId, nodeId);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();

	/* Connect to a database */
	BackgroundWorkerInitializeConnection(MtmDatabaseName, MtmDatabaseUser);
	ActivePortal = &fakePortal;
	ActivePortal->status = PORTAL_ACTIVE;
	ActivePortal->sourceText = "";

	/*
	 * Set proper restartLsn for all origins
	 */
	MtmLock(LW_EXCLUSIVE);
	for (i = 0; i < Mtm->nAllNodes; i++)
	{
		char	   *originName;
		RepOriginId originId;

		StartTransactionCommand();
		originName = psprintf(MULTIMASTER_SLOT_PATTERN, i + 1);
		originId = replorigin_by_name(originName, true);
		if (originId == InvalidRepOriginId) {
			originId = replorigin_create(originName);
		}
		CommitTransactionCommand();
		if (Mtm->nodes[i].restartLSN == INVALID_LSN)
			Mtm->nodes[i].restartLSN = replorigin_get_progress(originId, true);
		Mtm->nodes[i].originId = originId;
	}
	MtmUnlock();

	/* This is main loop of logical replication.
	 * In case of errors we will try to reestablish connection.
	 * Also reconnet is forced when node is switch to recovery mode
	 */
	while (!got_sigterm)
	{
		int	 count;
		ConnStatusType status;
		lsn_t originStartPos;
		int timeline;

		/*
		 * Determine when and how we should open replication slot.
		 * Druing recovery we need to open only one replication slot from which node should receive all transactions.
		 * Slots at other nodes should be removed
		 */
		mode = MtmGetReplicationMode(nodeId, &got_sigterm);
		MTM_LOG1("[STATE] Node %i: wal_receiver starts in %s mode", nodeId, MtmReplicationModeName[mode]);

		if (mode == REPLMODE_RECOVERY)
			enableFsync = false;
		else
			enableFsync = true;

		if (mode == REPLMODE_EXIT)
		{
			break;
		}
		timeline = Mtm->nodes[nodeId-1].timeline;
		count = Mtm->recoveryCount;

		/* Establish connection to remote server */
		conn = PQconnectdb_safe(connString, 0);
		status = PQstatus(conn);
		if (status != CONNECTION_OK)
		{
			ereport(WARNING, (MTM_ERRMSG("%s: Could not establish connection to remote server (%s), status = %d, error = %s",
									 worker_proc, connString, status, PQerrorMessage(conn))));
			goto OnError;
		}

		query = createPQExpBuffer();

		/* Start logical replication at specified position */
		originStartPos = replorigin_get_progress(Mtm->nodes[nodeId-1].originId, false);
		if (originStartPos == INVALID_LSN || Mtm->nodes[nodeId-1].manualRecovery) {
			/*
			 * We are just creating new replication slot.
			 * It is assumed that state of local and remote nodes is the same at this moment.
			 * They are either empty, either new node is synchronized using base_backup.
			 * So we assume that LSNs are the same for local and remote node
			 */
			appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", slotName, MULTIMASTER_NAME);
			res = PQexec(conn, query->data);
			if (PQresultStatus(res) != PGRES_TUPLES_OK)
			{
				const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
				if (!sqlstate || strcmp(sqlstate, ERRCODE_DUPLICATE_OBJECT_STR) != 0)
				{
					PQclear(res);
					ereport(ERROR, (MTM_ERRMSG("%s: Could not create logical slot", worker_proc)));

					goto OnError;
				}
			}
			PQclear(res);
			resetPQExpBuffer(query);
			Mtm->nodes[nodeId-1].manualRecovery = false;
		} else {
			if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
				MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
				Mtm->nodes[nodeId-1].restartLSN = originStartPos;
			}
			MTM_LOG1("Restart logical receiver at position %llx from node %d", originStartPos, nodeId);
		}

		MTM_LOG1("Start replication on slot %s from node %d at position %llx, mode %s, recovered lsn %llx",
				 slotName, nodeId, originStartPos, MtmReplicationModeName[mode], Mtm->recoveredLSN);

		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s', \"mtm_restart_pos\" '%llx', \"mtm_recovered_pos\" '%llx')",
						  slotName,
						  (uint32) (originStartPos >> 32),
						  (uint32) originStartPos,
						  MULTIMASTER_MAX_PROTO_VERSION,
						  MULTIMASTER_MIN_PROTO_VERSION,
						  MtmReplicationModeName[mode],
						  originStartPos,
						  Mtm->recoveredLSN
			);
		res = PQexec(conn, query->data);
		if (PQresultStatus(res) != PGRES_COPY_BOTH)
		{
			// int i, n_deleted_slots = 0;

			elog(WARNING, "Can't find slot on node%d. Shutting down receiver. %s", nodeId, PQresultErrorMessage(res));
			goto OnError;
			// Mtm->nodes[nodeId-1].slotDeleted = true;
			// for (i = 0; i < Mtm->nAllNodes; i++)
			// {
			// 	if (Mtm->nodes[i].slotDeleted)
			// 		n_deleted_slots++;
			// }
			// if (n_deleted_slots == Mtm->nAllNodes - 1)
			// {
			// 	elog(FATAL, "All neighbour nopes have no replication slot for us. Exiting.");
			// }
			// proc_exit(1);
		}
		PQclear(res);
		resetPQExpBuffer(query);

		MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_WAL_RECEIVER_START);

		while (!got_sigterm)
		{
			int rc, hdr_len;
			/* Wait necessary amount of time */
			rc = WaitLatch(&MyProc->procLatch,
						   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
						   receiver_idle_time * 1L);
			ResetLatch(&MyProc->procLatch);
			/* Process signals */
			if (got_sighup)
			{
				/* Process config file */
				ProcessConfigFile(PGC_SIGHUP);
				got_sighup = false;
				ereport(LOG, (MTM_ERRMSG("%s: processed SIGHUP", worker_proc)));
			}

			if (got_sigterm)
			{
				/* Simply exit */
				ereport(LOG, (MTM_ERRMSG("%s: processed SIGTERM", worker_proc)));
				proc_exit(0);
			}

			/* Emergency bailout if postmaster has died */
			if (rc & WL_POSTMASTER_DEATH)
			{
				BgwPoolStop(&Mtm->pool);
				proc_exit(1);
			}

			if (Mtm->status == MTM_DISABLED || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId))
			{
				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
				break;
			}
			if (count != Mtm->recoveryCount) {
				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because node was recovered", worker_proc)));
				break;
			}

			if (timeline != Mtm->nodes[nodeId-1].timeline) {
				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because node %d timeline is changed", worker_proc, nodeId)));
				break;
			}

			/*
			 * Receive data.
			 */
			while (true)
			{
				lsn_t walEnd;
				char* stmt;

				/* Some cleanup */
				if (copybuf != NULL)
				{
					PQfreemem(copybuf);
					copybuf = NULL;
				}

				rc = PQgetCopyData(conn, &copybuf, 1);
				if (rc <= 0) {
					break;
				}

				/*
				 * Check message received from server:
				 * - 'k', keepalive message
				 * - 'w', check for streaming header
				 */
				if (copybuf[0] == 'k')
				{
					int			pos;
					bool		replyRequested;

					/*
					 * Parse the keepalive message, enclosed in the CopyData message.
					 * We just check if the server requested a reply, and ignore the
					 * rest.
					 */
					pos = 1;	/* skip msgtype 'k' */

					/*
					 * In this message is the latest WAL position that server has
					 * considered as sent to this receiver.
					 */
					walEnd = fe_recvint64(&copybuf[pos]);
					pos += 8;	/* read walEnd */
					pos += 8;	/* skip sendTime */
					if (rc < pos + 1)
					{
						ereport(LOG, (MTM_ERRMSG("%s: streaming header too small: %d",
											 worker_proc, rc)));
						goto OnError;
					}
					replyRequested = copybuf[pos];

					/* Update written position */
					output_written_lsn = Max(walEnd, output_written_lsn);

					/*
					 * If the server requested an immediate reply, send one.
					 * If sync mode is sent reply in all cases to ensure that
					 * server knows how far replay has been done.
					 * In recovery mode also always send reply to provide master with more precise information
					 * about recovery progress
					 */
					if (replyRequested || receiver_sync_mode || Mtm->status == MTM_RECOVERY)
					{
						int64 now = feGetCurrentTimestamp();

						MtmUpdateLsnMapping(nodeId, walEnd);
						/* Leave if feedback is not sent properly */
						if (!sendFeedback(conn, now, nodeId)) {
							goto OnError;
						}
					}
					continue;
				}
				else if (copybuf[0] != 'w')
				{
					ereport(LOG, (MTM_ERRMSG("%s: Incorrect streaming header",
										 worker_proc)));
					goto OnError;
				}

				/* Now fetch the data */
				hdr_len = 1;		/* msgtype 'w' */
				fe_recvint64(&copybuf[hdr_len]);
				hdr_len += 8;		/* dataStart */
				walEnd = fe_recvint64(&copybuf[hdr_len]);
				hdr_len += 8;		/* WALEnd */
				hdr_len += 8;		/* sendTime */

				/* WAL position of the end of this message at WAL sender */
				MtmSenderWalEnd = walEnd;

				/*ereport(LOG, (MTM_ERRMSG("%s: receive message %c length %d", worker_proc, copybuf[hdr_len], rc - hdr_len)));*/

				Assert(rc >= hdr_len);

				if (rc > hdr_len)
				{
					int msg_len = rc - hdr_len;
					stmt = copybuf + hdr_len;
					MTM_LOG3("Receive message %c from node %d", stmt[0], nodeId);
					if (buf.used + msg_len + 1 >= MtmTransSpillThreshold*1024L) {
						if (spill_file < 0) {
							int file_id;
							spill_file = MtmCreateSpillFile(nodeId, &file_id);
							pq_sendbyte(&spill_info, 'F');
							pq_sendint(&spill_info, nodeId, 4);
							pq_sendint(&spill_info, file_id, 4);
						}
						ByteBufferAppend(&buf, ")", 1);
						pq_sendbyte(&spill_info, '(');
						pq_sendint(&spill_info, buf.used, 4);
						MtmSpillToFile(spill_file, buf.data, buf.used);
						ByteBufferReset(&buf);
					}
					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'C'))) {
						MTM_LOG3("Process '%c' message from %d", stmt[1], nodeId);
						if (stmt[0] == 'M' && stmt[1] == 'C') { /* concurrent DDL should be executed by parallel workers */
							MtmExecute(stmt, msg_len);
						} else {
							MtmExecutor(stmt, msg_len); /* all other messages can be processed by receiver itself */
						}
					} else {
						ByteBufferAppend(&buf, stmt, msg_len);
						if (stmt[0] == 'C') /* commit */
						{
							if (!MtmFilterTransaction(stmt, msg_len))
							{
								if (spill_file >= 0) {
									ByteBufferAppend(&buf, ")", 1);
									pq_sendbyte(&spill_info, '(');
									pq_sendint(&spill_info, buf.used, 4);
									MtmSpillToFile(spill_file, buf.data, buf.used);
									MtmCloseSpillFile(spill_file);
									MtmExecute(spill_info.data, spill_info.len);
									spill_file = -1;
									resetStringInfo(&spill_info);
								} else {
									if (MtmPreserveCommitOrder && buf.used == msg_len) {
										/* Perform commit-prepared and rollback-prepared requested directly in receiver */
										timestamp_t stop, start = MtmGetSystemTime();
										MtmExecutor(buf.data, buf.used);
										stop = MtmGetSystemTime();
										if (stop - start > USECS_PER_SEC) {
											elog(WARNING, "Commit of prepared transaction takes %lld usec, flags=%x", stop - start, stmt[1]);
										}
									} else {
										/* all other commits should be applied in place */
										// Assert(stmt[1] == PGLOGICAL_PREPARE || stmt[1] == PGLOGICAL_COMMIT || stmt[1] == PGLOGICAL_PRECOMMIT_PREPARED);
										MtmExecute(buf.data, buf.used);
									}
								}
							} else if (spill_file >= 0) {
								MtmCloseSpillFile(spill_file);
								resetStringInfo(&spill_info);
								spill_file = -1;
							}
							ByteBufferReset(&buf);
						}
					}
				}
				/* Update written position */
				output_written_lsn = Max(walEnd, output_written_lsn);
			}

			/* No data, move to next loop */
			if (rc == 0)
			{
				/*
				 * In async mode, and no data available. We block on reading but
				 * not more than the specified timeout, so that we can send a
				 * response back to the client.
				 */
				int			r;
				fd_set	  input_mask;
				int64	   message_target = 0;
				int64	   fsync_target = 0;
				struct timeval timeout;
				struct timeval *timeoutptr = NULL;
				int64	   targettime;
				long		secs;
				int		 usecs;
				int64		now;

				FD_ZERO(&input_mask);
				FD_SET(PQsocket(conn), &input_mask);

				/* Now compute when to wakeup. */
				targettime = message_target;

				if (fsync_target > 0 && fsync_target < targettime)
					targettime = fsync_target;
				now = feGetCurrentTimestamp();
				feTimestampDifference(now,
									  targettime,
									  &secs,
									  &usecs);
				if (secs <= 0)
					timeout.tv_sec = 1; /* Always sleep at least 1 sec */
				else
					timeout.tv_sec = secs;
				timeout.tv_usec = usecs;
				timeoutptr = &timeout;

				r = pg_select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr, conn->isRsocket);
				if (r == 0)
				{
					int64 now = feGetCurrentTimestamp();

					MtmUpdateLsnMapping(nodeId, INVALID_LSN);
					sendFeedback(conn, now, nodeId);
				}
				else if (r < 0 && errno == EINTR)
				{
					/*
					 * Got a timeout or signal. Continue the loop and either
					 * deliver a status packet to the server or just go back into
					 * blocking.
					 */
					continue;
				}
				else if (r < 0)
				{
					ereport(LOG, (MTM_ERRMSG("%s: Incorrect status received.",
										 worker_proc)));

					goto OnError;
				}

				/* Else there is actually data on the socket */
				if (PQconsumeInput(conn) == 0)
				{
					ereport(LOG, (MTM_ERRMSG("%s: Data remaining on the socket.",
										 worker_proc)));
					goto OnError;
				}
				continue;
			}

			/* End of copy stream */
			if (rc == -1)
			{
				ereport(LOG, (MTM_ERRMSG("%s: COPY Stream has abruptly ended...",
									 worker_proc)));
				goto OnError;
			}

			/* Failure when reading copy stream, leave */
			if (rc == -2)
			{
				ereport(LOG, (MTM_ERRMSG("%s: Failure while receiving changes...",
									 worker_proc)));
				goto OnError;
			}
		}
		PQfinish(conn);
		continue;

	  OnError:
		PQfinish(conn);
		MtmReleaseRecoverySlot(nodeId);
		MtmSleep(RECEIVER_SUSPEND_TIMEOUT);
	}
	ByteBufferFree(&buf);
	/* Restart this bgworker */
	proc_exit(1);
}

void MtmStartReceiver(int nodeId, bool dynamic)
{
	BackgroundWorker worker;
	MemoryContext oldContext = MemoryContextSwitchTo(TopMemoryContext);

	MemSet(&worker, 0, sizeof(BackgroundWorker));
	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |	BGWORKER_BACKEND_DATABASE_CONNECTION;
	worker.bgw_start_time = BgWorkerStart_ConsistentState;
	worker.bgw_main = pglogical_receiver_main;
	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;

	/* Worker parameter and registration */
	snprintf(worker.bgw_name, BGW_MAXLEN, "mtm_pglogical_receiver_%d_%d", MtmNodeId, nodeId);

	worker.bgw_main_arg = Int32GetDatum(nodeId);
	if (dynamic) {
		BackgroundWorkerHandle *handle;
		if (!RegisterDynamicBackgroundWorker(&worker, &handle)) {
			elog(WARNING, "Failed to start background worker, please increase max_worker_processes configuration parameter (current value is %d)", max_worker_processes);
		}
	} else {
		RegisterBackgroundWorker(&worker);
	}

	MemoryContextSwitchTo(oldContext);
}

void MtmStartReceivers(void)
{
	int i;
	for (i = 0; i < MtmNodes; i++) {
		if (i+1 != MtmNodeId) {
			MtmStartReceiver(i+1, false);
		}
	}
}