Fix bugs

pgxn/neon/libpagestore.c

@@ -1,4 +1,4 @@
-]/*-------------------------------------------------------------------------
+/*-------------------------------------------------------------------------
  *
  * libpagestore.c
  *	  Handles network communications with the remote pagestore.
@@ -25,11 +25,14 @@
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "portability/instr_time.h"
+#include "postmaster/bgworker.h"
 #include "postmaster/interrupt.h"
 #include "storage/buf_internals.h"
 #include "storage/ipc.h"
 #include "storage/lwlock.h"
 #include "storage/pg_shmem.h"
+#include "storage/proc.h"
+#include "storage/procsignal.h"
 #include "utils/guc.h"
 
 #include "neon.h"
@@ -67,7 +70,7 @@ static int	stripe_size;
 
 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 
-void CommunicatorMain(Datum main_arg)
+void CommunicatorMain(Datum main_arg);
 
 typedef struct
 {
@@ -76,7 +79,6 @@ typedef struct
 } ShardMap;
 
 static ShardMap shard_map;
-static PageServer* page_servers;
 
 #if PG_VERSION_NUM >= 150000
 static shmem_request_hook_type prev_shmem_request_hook = NULL;
@@ -90,6 +92,10 @@ static NeonCommunicatorChannel* channels;
 int MyProcNumber;
 #endif
 
+#ifndef PG_IOV_MAX
+#define PG_IOV_MAX 32
+#endif
+
 static bool am_communicator = false;
 
 typedef enum PSConnectionState {
@@ -133,24 +139,29 @@ typedef struct
 
 static PageServer* page_servers;
 
-static bool pageserver_flush(int chan_no);
 static void pageserver_disconnect(int chan_no);
 
 
 static void* communicator_read_loop(void* arg);
-static void* communicator_erite_loop(void* arg);
+static void* communicator_write_loop(void* arg);
 
 static void
 log_error_message(NeonErrorResponse* err)
 {
 	int save_pid = MyProcPid;
 	pthread_mutex_lock(&mutex);
-	MyProcPid = ProcGlobal->allProcs[resp->procno].pid;
+	MyProcPid = ProcGlobal->allProcs[err->req.u.recepient.procno].pid;
 	neon_log(LOG, "Server returns error for request %d: %s", err->req.tag, err->message);
 	MyProcPid = save_pid;
 	pthread_mutex_unlock(&mutex);
 }
 
+static Size
+NumberOfChannels(void)
+{
+	return MAX_SHARDS * parallel_connections;
+}
+
 /*
  * Parse a comma-separated list of connection strings into a ShardMap.
  *
@@ -238,13 +249,13 @@ AssignPageserverConnstring(const char *newval, void *extra)
 		elog(ERROR, "could not parse shard map");
 	}
 
-	for (page_servers == NULL)
+	if (page_servers == NULL)
 	{
 		page_servers = (PageServer*)calloc(NumberOfChannels(), sizeof(PageServer));
 	}
 	for (size_t i = 0; i < old_num_shards; i++)
 	{
-		if (page_servers[i].state = PS_Connected)
+		if (page_servers[i].state == PS_Connected)
 		{
 			/* TODO: race condition */
 			page_servers[i].state = PS_Expired;
@@ -259,69 +270,6 @@ AssignPageserverConnstring(const char *newval, void *extra)
 	}
 }
 
-/*
- * Get the current number of shards, and/or the connection string for a
- * particular shard from the shard map in shared memory.
- *
- * If num_shards_p is not NULL, it is set to the current number of shards.
- *
- * If connstr_p is not NULL, the connection string for 'chan_no' is copied to
- * it. It must point to a buffer at least MAX_PAGESERVER_CONNSTRING_SIZE bytes
- * long.
- *
- * As a side-effect, if the shard map in shared memory had changed since the
- * last call, terminates all existing connections to all pageservers.
- */
-static void
-load_shard_map(int chan_no, char *connstr_p, int *num_shards_p)
-{
-	uint64		begin_update_counter;
-	uint64		end_update_counter;
-	ShardMap   *shard_map = &pagestore_shared->shard_map;
-	int	num_shards;
-
-	/*
-	 * Postmaster can update the shared memory values concurrently, in which
-	 * case we would copy a garbled mix of the old and new values. We will
-	 * detect it because the counter's won't match, and retry. But it's
-	 * important that we don't do anything within the retry-loop that would
-	 * depend on the string having valid contents.
-	 */
-	do
-	{
-		begin_update_counter = pg_atomic_read_u64(&pagestore_shared->begin_update_counter);
-		end_update_counter = pg_atomic_read_u64(&pagestore_shared->end_update_counter);
-
-		num_shards = shard_map->num_shards;
-		if (connstr_p && chan_no < MAX_SHARDS)
-			strlcpy(connstr_p, shard_map->connstring[chan_no], MAX_PAGESERVER_CONNSTRING_SIZE);
-		pg_memory_barrier();
-	}
-	while (begin_update_counter != end_update_counter
-		   || begin_update_counter != pg_atomic_read_u64(&pagestore_shared->begin_update_counter)
-		   || end_update_counter != pg_atomic_read_u64(&pagestore_shared->end_update_counter));
-
-	if (connstr_p && chan_no >= num_shards)
-		neon_log(ERROR, "Shard %d is greater or equal than number of shards %d",
-				 chan_no, num_shards);
-
-	/*
-	 * If any of the connection strings changed, reset all connections.
-	 */
-	if (pagestore_local_counter != end_update_counter)
-	{
-		for (int i = 0; i < MAX_SHARDS; i++)
-		{
-			if (page_servers[i].conn)
-				pageserver_disconnect(i);
-		}
-		pagestore_local_counter = end_update_counter;
-	}
-
-	if (num_shards_p)
-		*num_shards_p = num_shards;
-}
-
 #define MB (1024*1024)
 
 int
@@ -331,13 +279,13 @@ get_shard_number(NRelFileInfo rinfo, BlockNumber blocknum)
 
 #if PG_MAJORVERSION_NUM < 16
 	hash = murmurhash32(rinfo.relNode);
-	hash = hash_combine(hash, murmurhash32(>blocknum / stripe_size));
+	hash = hash_combine(hash, murmurhash32(blocknum / stripe_size));
 #else
 	hash = murmurhash32(rinfo.relNumber);
-	hash = hash_combine(hash, murmurhash32(tag->blocknum / stripe_size));
+	hash = hash_combine(hash, murmurhash32(blocknum / stripe_size));
 #endif
 
-	return hash % n_shards;
+	return hash % shard_map.num_shards;
 }
 
 static inline void
@@ -361,16 +309,15 @@ cleanup_and_disconnect(PageServer *ps)
 /*
  * Like pchmop but uses malloc instead palloc for thread safety
  */
-
-char*
-static chomp(char const* in)
+static char*
+chomp(char const* in)
 {
 	size_t		n;
 
 	n = strlen(in);
 	while (n > 0 && in[n - 1] == '\n')
 		n--;
-	return strdup(in, n);
+	return strndup(in, n);
 }
 
 /*
@@ -388,15 +335,7 @@ pageserver_connect(int chan_no, int elevel)
 	PageServer *ps = &page_servers[chan_no];
 	char		connstr[MAX_PAGESERVER_CONNSTRING_SIZE];
 	char        pagestream_query[MAX_PS_QUERY_LENGTH];
-
-	/*
-	 * Get the connection string for this shard. If the shard map has been
-	 * updated since we last looked, this will also disconnect any existing
-	 * pageserver connections as a side effect.
-	 * Note that connstr is used both during connection start, and when we
-	 * log the successful connection.
-	 */
-	load_shard_map(chan_no, connstr, NULL);
+	int			shard_no = chan_no / parallel_connections;
 
 	switch (ps->state)
 	{
@@ -409,9 +348,9 @@ pageserver_connect(int chan_no, int elevel)
 		int64		us_since_last_attempt;
 
 		/* Make sure we start with a clean slate */
-		cleanup_and_disconnect(shard);
+		cleanup_and_disconnect(ps);
 
-		neon_shard_log(chan_no, DEBUG5, "Connection state: Disconnected");
+		neon_shard_log(shard_no, DEBUG5, "Connection state: Disconnected");
 
 		now = GetCurrentTimestamp();
 		us_since_last_attempt = (int64) (now - ps->last_reconnect_time);
@@ -449,7 +388,7 @@ pageserver_connect(int chan_no, int elevel)
 		 * we don't currently use that capability anywhere.
 		 */
 		keywords[0] = "dbname";
-		values[0] = shard_map.connstrings[chan_no / parallel_connections];
+		values[0] = shard_map.connstring[chan_no / parallel_connections];
 		n_pgsql_params = 1;
 
 		if (neon_auth_token)
@@ -466,7 +405,7 @@ pageserver_connect(int chan_no, int elevel)
 		if (PQstatus(ps->conn) == CONNECTION_BAD)
 		{
 			char	   *msg = chomp(PQerrorMessage(ps->conn));
-			cleanup_and_disconnect(shard);
+			cleanup_and_disconnect(ps);
 			ereport(elevel,
 					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
 						errmsg(NEON_TAG "[shard %d] could not establish connection to pageserver", chan_no),
@@ -482,7 +421,7 @@ pageserver_connect(int chan_no, int elevel)
 		int			ps_send_query_ret;
 		bool		connected = false;
 		int poll_result = PGRES_POLLING_WRITING;
-		neon_shard_log(chan_no, DEBUG5, "Connection state: Connecting_Startup");
+		neon_shard_log(shard_no, DEBUG5, "Connection state: Connecting_Startup");
 
 		do
 		{
@@ -493,22 +432,22 @@ pageserver_connect(int chan_no, int elevel)
 				{
 					char	   *pqerr = PQerrorMessage(ps->conn);
 					char	   *msg = NULL;
-					neon_shard_log(chan_no, DEBUG5, "POLLING_FAILED");
+					neon_shard_log(shard_no, DEBUG5, "POLLING_FAILED");
 
 					if (pqerr)
 						msg = chomp(pqerr);
 
-					cleanup_and_disconnect(shard);
+					cleanup_and_disconnect(ps);
 
 					if (msg)
 					{
-						neon_shard_log(chan_no, elevel,
+						neon_shard_log(shard_no, elevel,
 									   "could not connect to pageserver: %s",
 									   msg);
 						free(msg);
 					}
 					else
-						neon_shard_log(chan_no, elevel,
+						neon_shard_log(shard_no, elevel,
 									   "could not connect to pageserver");
 
 					return false;
@@ -558,7 +497,7 @@ pageserver_connect(int chan_no, int elevel)
 
 				break;
 			case PGRES_POLLING_OK:
-				neon_shard_log(chan_no, DEBUG5, "POLLING_OK");
+				neon_shard_log(shard_no, DEBUG5, "POLLING_OK");
 				connected = true;
 				break;
 			}
@@ -571,7 +510,7 @@ pageserver_connect(int chan_no, int elevel)
 		ps->last_connect_time = GetCurrentTimestamp();
 
 		/* Allocate wait event set in critical section */
-		pthread_lock(&mutex);
+		pthread_mutex_lock(&mutex);
 #if PG_MAJORVERSION_NUM >= 17
 		ps->wes_read = CreateWaitEventSet(NULL, 3);
 #else
@@ -582,16 +521,16 @@ pageserver_connect(int chan_no, int elevel)
 		AddWaitEventToSet(ps->wes_read, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
 						  NULL, NULL);
 		AddWaitEventToSet(ps->wes_read, WL_SOCKET_READABLE, PQsocket(ps->conn), NULL, NULL);
-		pthread_unlock(&mutex);
+		pthread_mutex_unlock(&mutex);
 
 
 		switch (neon_protocol_version)
 		{
 		case 3:
-			snprintf(pagesstream_query, sizeof pagestream_query, "pagestream_v3 %s %s", neon_tenant, neon_timeline);
+			snprintf(pagestream_query, sizeof pagestream_query, "pagestream_v3 %s %s", neon_tenant, neon_timeline);
 			break;
 		case 2:
-			snprintf(pagesstream_query, sizeof pagestream_query, "pagestream_v2 %s %s", neon_tenant, neon_timeline);
+			snprintf(pagestream_query, sizeof pagestream_query, "pagestream_v2 %s %s", neon_tenant, neon_timeline);
 			break;
 		default:
 			elog(ERROR, "unexpected neon_protocol_version %d", neon_protocol_version);
@@ -601,7 +540,7 @@ pageserver_connect(int chan_no, int elevel)
 		{
 			char	   *msg = chomp(PQerrorMessage(ps->conn));
 
-			cleanup_and_disconnect(shard);
+			cleanup_and_disconnect(ps);
 
 			ereport(elevel,
 					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
@@ -614,9 +553,9 @@ pageserver_connect(int chan_no, int elevel)
 		ps_send_query_ret = PQsendQuery(ps->conn, pagestream_query);
 		if (ps_send_query_ret != 1)
 		{
-			cleanup_and_disconnect(shard);
+			cleanup_and_disconnect(ps);
 
-			neon_shard_log(chan_no, elevel, "could not send pagestream command to pageserver");
+			neon_shard_log(shard_no, elevel, "could not send pagestream command to pageserver");
 			return false;
 		}
 
@@ -625,12 +564,12 @@ pageserver_connect(int chan_no, int elevel)
 	/* FALLTHROUGH */
 	case PS_Connecting_PageStream:
 	{
-		neon_shard_log(chan_no, DEBUG5, "Connection state: Connecting_PageStream");
+		neon_shard_log(shard_no, DEBUG5, "Connection state: Connecting_PageStream");
 
 		if (PQstatus(ps->conn) == CONNECTION_BAD)
 		{
 			char	   *msg = chomp(PQerrorMessage(ps->conn));
-			cleanup_and_disconnect(shard);
+			cleanup_and_disconnect(ps);
 			ereport(elevel,
 					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
 						errmsg(NEON_TAG "[shard %d] could not establish connection to pageserver", chan_no),
@@ -657,8 +596,8 @@ pageserver_connect(int chan_no, int elevel)
 				{
 					char	   *msg = chomp(PQerrorMessage(ps->conn));
 
-					cleanup_and_disconnect(shard);
-					neon_shard_log(chan_no, elevel, "could not complete handshake with pageserver: %s",
+					cleanup_and_disconnect(ps);
+					neon_shard_log(shard_no, elevel, "could not complete handshake with pageserver: %s",
 								   msg);
 					free(msg);
 					return false;
@@ -678,11 +617,11 @@ pageserver_connect(int chan_no, int elevel)
 		 */
 		ps->delay_us = MIN_RECONNECT_INTERVAL_USEC;
 
-		neon_shard_log(chan_no, DEBUG5, "Connection state: Connected");
-		neon_shard_log(chan_no, LOG, "libpagestore: connected to '%s' with protocol version %d", connstr, neon_protocol_version);
+		neon_shard_log(shard_no, DEBUG5, "Connection state: Connected");
+		neon_shard_log(shard_no, LOG, "libpagestore: connected to '%s' with protocol version %d", connstr, neon_protocol_version);
 		return true;
 	default:
-		neon_shard_log(chan_no, ERROR, "libpagestore: invalid connection state %d", ps->state);
+		neon_shard_log(shard_no, ERROR, "libpagestore: invalid connection state %d", ps->state);
 	}
 	/* This shouldn't be hit */
 	Assert(false);
@@ -703,6 +642,7 @@ call_PQgetCopyData(int chan_no, char **buffer)
 				last_log_ts,
 				since_last_log;
 	bool		logged = false;
+	int			shard_no = chan_no / parallel_connections;
 
 	/*
 	 * As a debugging aid, if we don't get a response for a long time, print a
@@ -743,7 +683,7 @@ retry:
 			{
 				char	   *msg = chomp(PQerrorMessage(pageserver_conn));
 
-				neon_shard_log(chan_no, LOG, "could not get response from pageserver: %s", msg);
+				neon_shard_log(shard_no, LOG, "could not get response from pageserver: %s", msg);
 				free(msg);
 				return -1;
 			}
@@ -785,7 +725,7 @@ retry:
 				}
 			}
 #endif
-			neon_shard_log(chan_no, LOG, "no response received from pageserver for %0.3f s, still waiting (sent " UINT64_FORMAT " requests, received " UINT64_FORMAT " responses) (socket sndbuf=%d recvbuf=%d)",
+			neon_shard_log(shard_no, LOG, "no response received from pageserver for %0.3f s, still waiting (sent " UINT64_FORMAT " requests, received " UINT64_FORMAT " responses) (socket sndbuf=%d recvbuf=%d)",
 						   INSTR_TIME_GET_DOUBLE(since_start),
 						   ps->nrequests_sent, ps->nresponses_received, sndbuf, recvbuf);
 			last_log_ts = now;
@@ -804,7 +744,7 @@ retry:
 		INSTR_TIME_SET_CURRENT(now);
 		since_start = now;
 		INSTR_TIME_SUBTRACT(since_start, start_ts);
-		neon_shard_log(chan_no, LOG, "received response from pageserver after %0.3f s",
+		neon_shard_log(shard_no, LOG, "received response from pageserver after %0.3f s",
 					   INSTR_TIME_GET_DOUBLE(since_start));
 	}
 
@@ -835,17 +775,18 @@ pageserver_disconnect(int chan_no)
 }
 
 static bool
-pageserver_send(int chan_no, StringInfo* msg)
+pageserver_send(int chan_no, StringInfo msg)
 {
 	PageServer *ps = &page_servers[chan_no];
 	PGconn	   *pageserver_conn;
+	int			shard_no = chan_no / parallel_connections;
 
 	MyNeonCounters->pageserver_requests_sent_total++;
 
 	/* If the connection was lost for some reason, reconnect */
 	if (ps->state == PS_Connected && PQstatus(ps->conn) == CONNECTION_BAD)
 	{
-		neon_shard_log(chan_no, LOG, "pageserver_send disconnect bad connection");
+		neon_shard_log(shard_no, LOG, "pageserver_send disconnect bad connection");
 		pageserver_disconnect(chan_no);
 		pageserver_conn = NULL;
 	}
@@ -863,7 +804,7 @@ pageserver_send(int chan_no, StringInfo* msg)
 	{
 		if (ps->state == PS_Expired)
 		{
-			neon_shard_log(chan_no, LOG, "pageserver_send disconnect expired connection");
+			neon_shard_log(shard_no, LOG, "pageserver_send disconnect expired connection");
 			pageserver_disconnect(chan_no);
 			pageserver_conn = NULL;
 		}
@@ -893,39 +834,40 @@ pageserver_send(int chan_no, StringInfo* msg)
 	ps->nrequests_sent++;
 	if (PQputCopyData(pageserver_conn, msg->data, msg->len) <= 0)
 	{
-		char	   *errmsg = chomp(PQerrorMessage(pageserver_conn));
+		char	   *msg = chomp(PQerrorMessage(pageserver_conn));
 
 		pageserver_disconnect(chan_no);
-		neon_shard_log(chan_no, LOG, "pageserver_send disconnected: failed to send page request (try to reconnect): %s", errmsg);
-		free(errmsg);
+		neon_shard_log(shard_no, LOG, "pageserver_send disconnected: failed to send page request (try to reconnect): %s", msg);
+		free(msg);
 		return false;
 	}
 	if (PQflush(pageserver_conn))
 	{
-		char	   *errmsg = chomp(PQerrorMessage(pageserver_conn));
+		char	   *msg = chomp(PQerrorMessage(pageserver_conn));
 
 		pageserver_disconnect(chan_no);
-		neon_shard_log(chan_no, LOG, "pageserver_flush disconnect because failed to flush page requests: %s", errmsg);
-		free(errmsg);
+		neon_shard_log(shard_no, LOG, "pageserver_flush disconnect because failed to flush page requests: %s", msg);
+		free(msg);
 		return false;
 	}
 	return true;
 }
 
 static NeonResponse *
-pageserver_receive(int chan_no, StringInfo* buf)
+pageserver_receive(int chan_no)
 {
 	StringInfoData resp_buff;
 	NeonResponse *resp;
-	PageServer *shard = &page_servers[chan_no];
+	PageServer *ps = &page_servers[chan_no];
 	PGconn	   *pageserver_conn = ps->conn;
+	int			shard_no = chan_no / parallel_connections;
 	/* read response */
 	int			rc;
 
 	/* TODO: fix race condition between sender and receivcer */
 	if (ps->state != PS_Connected)
 	{
-		neon_shard_log(chan_no, LOG,
+		neon_shard_log(shard_no, LOG,
 					   "pageserver_receive: returning NULL for non-connected pageserver connection: 0x%02x",
 					   ps->state);
 		return NULL;
@@ -948,7 +890,7 @@ pageserver_receive(int chan_no, StringInfo* buf)
 		}
 		PG_CATCH();
 		{
-			neon_shard_log(chan_no, LOG, "pageserver_receive: disconnect due to failure while parsing response");
+			neon_shard_log(shard_no, LOG, "pageserver_receive: disconnect due to failure while parsing response");
 			pageserver_disconnect(chan_no);
 			PG_RE_THROW();
 		}
@@ -956,7 +898,7 @@ pageserver_receive(int chan_no, StringInfo* buf)
 	}
 	else if (rc == -1)
 	{
-		neon_shard_log(chan_no, LOG, "pageserver_receive disconnect: psql end of copy data: %s", chomp(PQerrorMessage(pageserver_conn)));
+		neon_shard_log(shard_no, LOG, "pageserver_receive disconnect: psql end of copy data: %s", chomp(PQerrorMessage(pageserver_conn)));
 		pageserver_disconnect(chan_no);
 		resp = NULL;
 	}
@@ -965,12 +907,12 @@ pageserver_receive(int chan_no, StringInfo* buf)
 		char	   *msg = chomp(PQerrorMessage(pageserver_conn));
 
 		pageserver_disconnect(chan_no);
-		neon_shard_log(chan_no, ERROR, "pageserver_receive disconnect: could not read COPY data: %s", msg);
+		neon_shard_log(shard_no, ERROR, "pageserver_receive disconnect: could not read COPY data: %s", msg);
 	}
 	else
 	{
 		pageserver_disconnect(chan_no);
-		neon_shard_log(chan_no, ERROR, "pageserver_receive disconnect: unexpected PQgetCopyData return value: %d", rc);
+		neon_shard_log(shard_no, ERROR, "pageserver_receive disconnect: unexpected PQgetCopyData return value: %d", rc);
 	}
 
 	ps->nresponses_received++;
@@ -991,17 +933,11 @@ RequestBufferSize(void)
 	return (readahead_buffer_size + PG_IOV_MAX) * MaxBackends + (parallel_connections - 1) / parallel_connections;
 }
 
-static Size
-NumberOfChannels(void)
-{
-	return MAX_SHARDS * parallel_connections;
-}
-
 static Size
 CommunicatorShmemSize(void)
 {
-	return RequestBufferSize() * NumberOfChannels() * sizeof(CommunicatorRequest)
-		+ NumberOfChannels() * sizeof(CommunicatorChannel)
+	return RequestBufferSize() * NumberOfChannels() * sizeof(NeonCommunicatorRequest)
+		+ NumberOfChannels() * sizeof(NeonCommunicatorChannel)
 		+ sizeof(NeonCommunicatorResponse) * MaxBackends;
 }
 
@@ -1017,7 +953,7 @@ PagestoreShmemInit(void)
 	bool		found;
 
 	#if PG_VERSION_NUM < 170000
-	MyProcNumber = MyProc->procno;
+	MyProcNumber = MyProc->pgprocno;
 	#endif
 
 	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
@@ -1029,9 +965,9 @@ PagestoreShmemInit(void)
 		size_t n_channels = NumberOfChannels();
 		NeonCommunicatorRequest* requests = (NeonCommunicatorRequest*)(channels + n_channels);
 
-		for (size_t i = 0; i < channels; i++)
+		for (size_t i = 0; i < n_channels; i++)
 		{
-			NeonCommunicatorChannel* chan = channels[i]'
+			NeonCommunicatorChannel* chan = &channels[i];
 			pg_atomic_init_u64(&chan->write_pos, 0);
 			pg_atomic_init_u64(&chan->read_pos, 0);
 			InitLatch(&chan->latch);
@@ -1081,16 +1017,16 @@ pagestore_prepare_shmem(void)
 }
 
 void
-communicator_send_request(int shard, NeonCommunicationRequest* req)
+communicator_send_request(int shard, NeonCommunicatorRequest* req)
 {
 	/* bind backend to the particular channel */
-	NeonCommunicationChannel* chan = channels[shard * parallel_connections + (MyBackendId % parallel_connections)];
+	NeonCommunicatorChannel* chan = &channels[shard * parallel_connections + (MyProcNumber % parallel_connections)];
 	size_t ring_size = RequestBufferSize();
 	uint64 write_pos = pg_atomic_add_fetch_u64(&chan->write_pos, 1); /* reserve write position */
 	uint64 read_pos;
 
-	Assert(req->hdr.reqid == 0); /* ring overflow should not happen */
-	req->hdr.procno = MyProcNumber;
+	Assert(req->hdr.u.reqid == 0); /* ring overflow should not happen */
+	req->hdr.u.recepient.procno = MyProcNumber;
 
 	/* copy request */
 	chan->requests[(size_t)(write_pos % ring_size)] = *req;
@@ -1100,7 +1036,7 @@ communicator_send_request(int shard, NeonCommunicationRequest* req)
 		read_pos = write_pos;
 	} while (!pg_atomic_compare_exchange_u64(&chan->read_pos, &read_pos, write_pos+1) && read_pos <= write_pos);
 
-	responses[MyProcNumber] = req->hdr.tag;
+	responses[MyProcNumber].tag = req->hdr.tag;
 	SetLatch(&chan->latch);
 }
 
@@ -1122,7 +1058,7 @@ communicator_receive_response(void)
 }
 
 int64
-communicator_request(int shard, NeonCommunicationRequest* req)
+communicator_request(int shard, NeonCommunicatorRequest* req)
 {
 	communicator_send_request(shard, req);
 	return communicator_receive_response();
@@ -1219,14 +1155,6 @@ pg_init_libpagestore(void)
 							PGC_SIGHUP,
 							GUC_UNIT_MB,
 							NULL, NULL, NULL);
-	DefineCustomIntVariable("neon.max_reconnect_attempts",
-							"Maximal attempts to reconnect to pages server (with 1 second timeout)",
-							NULL,
-							&max_reconnect_attempts,
-							60, 0, INT_MAX,
-							PGC_USERSET,
-							0,
-							NULL, NULL, NULL);
 	DefineCustomIntVariable("neon.readahead_buffer_size",
 							"number of prefetches to buffer",
 							"This buffer is used to hold and manage prefetched "
@@ -1263,12 +1191,6 @@ pg_init_libpagestore(void)
 
 	register_communicator_worker();
 
-	if (page_server != NULL)
-		neon_log(ERROR, "libpagestore already loaded");
-
-	neon_log(PageStoreTrace, "libpagestore already loaded");
-	page_server = &api;
-
 	/*
 	 * Retrieve the auth token to use when connecting to pageserver and
 	 * safekeepers
@@ -1285,13 +1207,11 @@ pg_init_libpagestore(void)
 		dbsize_hook = neon_dbsize;
 	}
 
-	memset(page_servers, 0, sizeof(page_servers));
-
 	lfc_init();
 }
 
 static void
-allocStringInfo(StringInfo* s, size_t size)
+allocStringInfo(StringInfo s, size_t size)
 {
 	s->data = (char *)malloc(size);
 	s->maxlen = size;
@@ -1303,33 +1223,34 @@ communicator_write_loop(void* arg)
 {
 	uint64 read_start_pos = 0;
 	size_t chan_no = (size_t)arg;
-	NeonCommunicatorChannel* chan = channels[chan_no]
+	NeonCommunicatorChannel* chan = &channels[chan_no];
 	size_t n_channels = NumberOfChannels();
-	StrintgInfoData s;
+	StringInfoData s;
 
-	allocStringInfo(&s, NAX_REQUEST_SIZE);
+	allocStringInfo(&s, MAX_REQUEST_SIZE);
 
 	while (true)
 	{
-		NeonCommunicatorrequest* req;
+		NeonCommunicatorRequest* req;
 		uint64 read_end_pos;
 
 		/* Number of shards is decreased */
-		if (chan_no >= shard_map.num_shards * parallel_connection)
+		if (chan_no >= shard_map.num_shards * parallel_connections)
 			return NULL;
 
 		read_end_pos = pg_atomic_read_u64(&chan->read_pos);
 		Assert(read_start_pos <= read_end_pos);
 		while (read_start_pos == read_end_pos)
 		{
-			int events = WaitLatch(&chan->latch, WL_LATCH_SET|WL_POSTMASTER_DEATH, WAIT_EVENT_NEON_PS_SEND);
+			int events = WaitLatch(&chan->latch, WL_LATCH_SET|WL_POSTMASTER_DEATH, -1, WAIT_EVENT_NEON_PS_SEND);
 			if (events & WL_POSTMASTER_DEATH)
 				break;
 		}
-		req = chan->requests[read_start_pos % n_channels++];
+		req = &chan->requests[read_start_pos % n_channels++];
 		nm_pack_request(&s, &req->hdr);
-		pageserver_send(id, &s);
-		req->hdr.reqid = 0; /* mark requests as processed */
+		Assert(s.maxlen == MAX_REQUEST_SIZE); /* was not reallocated */
+		pageserver_send(chan_no, &s);
+		req->hdr.u.reqid = 0; /* mark requests as processed */
 	}
 }
 
@@ -1343,13 +1264,13 @@ communicator_read_loop(void* arg)
 	while (true)
 	{
 		/* Number of shards is decreased */
-		if (chan_no >= shard_map.num_shards * parallel_connection)
+		if (chan_no >= shard_map.num_shards * parallel_connections)
 			return NULL;
 
 		resp = pageserver_receive(chan_no);
 		if (resp == NULL)
 		{
-			pg_usleep(RECEVIER_RETRY_DELAY);
+			pg_usleep(RECEIVER_RETRY_DELAY);
 			continue;
 		}
 		switch (resp->tag)
@@ -1364,25 +1285,30 @@ communicator_read_loop(void* arg)
 				value = ((NeonDbSizeResponse*)resp)->db_size;
 				break;
 			case T_NeonGetPageResponse:
-				if (resp->recepient.bufid == InvalidBuffer)
+			{
+				NeonGetPageResponse* page_resp = (NeonGetPageResponse*)resp;
+				if (resp->u.recepient.bufid == InvalidBuffer)
 				{
-					NeonGetPageResponse* page_resp = (NeonGetPageResponse*)resp;
 					/* result of prefetch */
 					lfc_prefetch(page_resp->req.rinfo, page_resp->req.forknum, page_resp->req.blkno, page_resp->page, resp->not_modified_since);
+					free(resp);
 					continue; /* no need to notify backend */
-				} else {
-					memcpy(BufferGetBlock(resp->recepient.bufid), page_resp->page, BLCKSZ);
+				}
+				else
+				{
+					memcpy(BufferGetBlock(resp->u.recepient.bufid), page_resp->page, BLCKSZ);
 				}
 				break;
+			}
 			case T_NeonErrorResponse:
 				log_error_message((NeonErrorResponse *) resp);
 				break;
 			default:
 				break;
 		}
-		responses[resp->recepient.procno].value = value;
-		responses[resp->recepient.procno].tag = resp->tag;
-		SetLatch(&ProcGlobal->allProcs[resp->procno].procLatch);
+		responses[resp->u.recepient.procno].value = value;
+		responses[resp->u.recepient.procno].tag = resp->tag;
+		SetLatch(&ProcGlobal->allProcs[resp->u.recepient.procno].procLatch);
 		free(resp);
 	}
 }

pgxn/neon/pagestore_smgr.c

@@ -66,6 +66,7 @@
 #include "storage/md.h"
 #include "storage/smgr.h"
 
+#include "neon.h"
 #include "neon_perf_counters.h"
 #include "pagestore_client.h"
 #include "bitmap.h"
@@ -119,7 +120,7 @@ static bool (*old_redo_read_buffer_filter) (XLogReaderState *record, uint8 block
 static BlockNumber neon_nblocks(SMgrRelation reln, ForkNumber forknum);
 
 void
-nm_pack_request(StringInfo* s, NeonRequest *msg)
+nm_pack_request(StringInfo s, NeonRequest *msg)
 {
 	resetStringInfo(s);
 	pq_sendbyte(s, msg->tag);
@@ -194,14 +195,13 @@ nm_pack_request(StringInfo* s, NeonRequest *msg)
 			neon_log(ERROR, "unexpected neon message tag 0x%02x", msg->tag);
 			break;
 	}
-	return s;
 }
 
 /*
  * We can not used palloc because it is not thread safe
  &*/
-void*
-static memalloc(size_t size)
+static void*
+memalloc(size_t size)
 {
 	return calloc(size, 1);
 }
@@ -690,11 +690,6 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, co
 static void
 neon_init(void)
 {
-	Size		prfs_size;
-
-	if (MyPState != NULL)
-		return;
-
 	/*
 	 * Sanity check that theperf counters array is sized correctly. We got
 	 * this wrong once, and the formula for max number of backends and aux
@@ -959,6 +954,7 @@ static bool
 neon_exists(SMgrRelation reln, ForkNumber forkNum)
 {
 	neon_request_lsns request_lsns;
+	BlockNumber  n_blocks;
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -1017,12 +1013,12 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
 						  REL_METADATA_PSEUDO_BLOCKNO, &request_lsns, 1, NULL);
 	{
 		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonNblocksRequest,
-			.hdr.bufid = InvalidBuffer,
-			.hdr.lsn = request_lsns.request_lsn,
-			.hdr.not_modified_since = request_lsns.not_modified_since,
+			.exists.hdr.tag = T_NeonNblocksRequest,
+			.exists.hdr.u.recepient.bufid = InvalidBuffer,
+			.exists.hdr.lsn = request_lsns.request_lsn,
+			.exists.hdr.not_modified_since = request_lsns.not_modified_since,
 			.exists.rinfo = InfoFromSMgrRel(reln),
-			.exists.forknum = forknum,
+			.exists.forknum = forkNum,
 		};
 		return communicator_request(0, &request) != 0;
 	}
@@ -1358,7 +1354,7 @@ static bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			  int nblocks)
 {
-	BufferTag	tag;
+	neon_request_lsns request_lsns[PG_IOV_MAX];
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -1374,20 +1370,24 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			neon_log(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
+
+	neon_get_request_lsns(InfoFromSMgrRel(reln),
+						  forknum, blocknum,
+						  request_lsns, nblocks, NULL);
 	for (int i = 0; i < nblocks; i++)
 	{
 		if (!lfc_cache_contains(InfoFromSMgrRel(reln), forknum, blocknum + i))
 		{
 			NeonCommunicatorRequest request = {
-				.hdr.tag = T_NeonGetPageRequest,
-				.hdr.bufid = InvalidBuffer,
-				.hdr.lsn = request_lsns.request_lsn,
-				.hdr.not_modified_since = request_lsns.not_modified_since,
+				.page.hdr.tag = T_NeonGetPageRequest,
+				.page.hdr.u.recepient.bufid = InvalidBuffer,
+				.page.hdr.lsn = request_lsns[i].request_lsn,
+				.page.hdr.not_modified_since = request_lsns[i].not_modified_since,
 				.page.rinfo = InfoFromSMgrRel(reln),
 				.page.forknum = forknum,
-				.page.blocknum = blocknum + i,
+				.page.blkno = blocknum + i,
 			};
-			communicator_send_request(get_dhard_number(InfoFromSMgrRel(reln), blocknum + i), &request);
+			communicator_send_request(get_shard_number(InfoFromSMgrRel(reln), blocknum + i), &request);
 		}
 	}
 	return false;
@@ -1401,8 +1401,6 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 static bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 {
-	BufferTag	tag;
-
 	switch (reln->smgr_relpersistence)
 	{
 		case 0:					/* probably shouldn't happen, but ignore it */
@@ -1419,16 +1417,22 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 
 	if (!lfc_cache_contains(InfoFromSMgrRel(reln), forknum, blocknum))
 	{
-		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonGetPageRequest,
-			.hdr.bufid = InvalidBuffer,
-			.hdr.lsn = request_lsns.request_lsn,
-			.hdr.not_modified_since = request_lsns.not_modified_since,
-			.page.rinfo = InfoFromSMgrRel(reln),
-			.page.forknum = forknum,
-			.page.blocknum = blocknum + i,
-		};
-		communicator_send_request(get_dhard_number(InfoFromSMgrRel(reln), blocknum), &request);
+		neon_request_lsns request_lsns;
+		neon_get_request_lsns(InfoFromSMgrRel(reln),
+							  forknum, blocknum,
+							  &request_lsns, 1, NULL);
+		{
+			NeonCommunicatorRequest request = {
+				.page.hdr.tag = T_NeonGetPageRequest,
+				.page.hdr.u.recepient.bufid = InvalidBuffer,
+				.page.hdr.lsn = request_lsns.request_lsn,
+				.page.hdr.not_modified_since = request_lsns.not_modified_since,
+				.page.rinfo = InfoFromSMgrRel(reln),
+				.page.forknum = forknum,
+				.page.blkno = blocknum
+			};
+			communicator_send_request(get_shard_number(InfoFromSMgrRel(reln), blocknum), &request);
+		}
 	}
 	return false;
 }
@@ -1472,8 +1476,6 @@ neon_writeback(SMgrRelation reln, ForkNumber forknum,
 	 */
 	neon_log(SmgrTrace, "writeback noop");
 
-	prefetch_pump_state();
-
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
 		mdwriteback(reln, forknum, blocknum, nblocks);
@@ -1483,14 +1485,14 @@ neon_writeback(SMgrRelation reln, ForkNumber forknum,
 static Buffer
 GetBufferId(void const* ptr)
 {
-	Buffer buf = (size_t)(ptr -  BufferBlocks)/BLCKSZ;
+	Buffer buf = (size_t)((char*)ptr -  BufferBlocks)/BLCKSZ;
 	Assert(buf < NBuffers);
 	return buf+1;
 }
 
 static void
 #if PG_MAJORVERSION_NUM < 16
-neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber base_blockno, neon_request_lsns *request_lsns,
+neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blocknum, neon_request_lsns *request_lsns,
 				  char **buffers, BlockNumber nblocks, const bits8 *mask)
 #else
 neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blocknum, neon_request_lsns *request_lsns,
@@ -1500,15 +1502,15 @@ neon_read_at_lsnv(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blocknum,
 	for (int i = 0; i < nblocks; i++)
 	{
 		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonGetPageRequest,
-			.hdr.bufid = GetBufferId(buffers[i]),
-			.hdr.lsn = request_lsns.request_lsn,
-			.hdr.not_modified_since = request_lsns.not_modified_since,
-			.page.rinfo = InfoFromSMgrRel(reln),
+			.page.hdr.tag = T_NeonGetPageRequest,
+			.page.hdr.u.recepient.bufid = GetBufferId(buffers[i]),
+			.page.hdr.lsn = request_lsns[i].request_lsn,
+			.page.hdr.not_modified_since = request_lsns[i].not_modified_since,
+			.page.rinfo = rinfo,
 			.page.forknum = forknum,
-			.page.blocknum = blocknum + i,
+			.page.blkno = blocknum + i,
 		};
-		(void)communicator_request(get_dhard_number(InfoFromSMgrRel(reln), blocknum + i), &request);
+		(void)communicator_request(get_shard_number(rinfo, blocknum + i), &request);
 	}
 }
 
@@ -1649,6 +1651,7 @@ static void
 neon_readv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		void **buffers, BlockNumber nblocks)
 {
+	bits8		read[PG_IOV_MAX / 8] = {0};
 	neon_request_lsns request_lsns[PG_IOV_MAX];
 	int			lfc_result;
 
@@ -1673,8 +1676,6 @@ neon_readv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		neon_log(ERROR, "Read request too large: %d is larger than max %d",
 				 nblocks, PG_IOV_MAX);
 
-	memset(read, 0, sizeof(read));
-
 	/* Try to read from local file cache */
 	lfc_result = lfc_readv_select(InfoFromSMgrRel(reln), forknum, blocknum, buffers,
 								  nblocks, read);
@@ -1973,10 +1974,10 @@ neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 
 	{
 		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonNblocksRequest,
-			.hdr.bufid = InvalidBuffer,
-			.hdr.lsn = request_lsns.request_lsn,
-			.hdr.not_modified_since = request_lsns.not_modified_since,
+			.nblocks.hdr.tag = T_NeonNblocksRequest,
+			.nblocks.hdr.u.recepient.bufid = InvalidBuffer,
+			.nblocks.hdr.lsn = request_lsns.request_lsn,
+			.nblocks.hdr.not_modified_since = request_lsns.not_modified_since,
 			.nblocks.rinfo = InfoFromSMgrRel(reln),
 			.nblocks.forknum = forknum,
 		};
@@ -1998,10 +1999,10 @@ neon_dbsize(Oid dbNode)
 
 	{
 		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonDbSizeRequest,
-			.hdr.bufid = InvalidBuffer,
-			.hdr.lsn = request_lsns.request_lsn,
-			.hdr.not_modified_since = request_lsns.not_modified_since,
+			.dbsize.hdr.tag = T_NeonDbSizeRequest,
+			.dbsize.hdr.u.recepient.bufid = InvalidBuffer,
+			.dbsize.hdr.lsn = request_lsns.request_lsn,
+			.dbsize.hdr.not_modified_since = request_lsns.not_modified_since,
 			.dbsize.dbNode = dbNode,
 		};
 		return communicator_request(0, &request);
@@ -2103,8 +2104,6 @@ neon_immedsync(SMgrRelation reln, ForkNumber forknum)
 
 	neon_log(SmgrTrace, "[NEON_SMGR] immedsync noop");
 
-	prefetch_pump_state();
-
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
 		mdimmedsync(reln, forknum);
@@ -2415,15 +2414,14 @@ neon_extend_rel_size(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
 		 * block, which is fine and expected.
 		 */
 		NeonCommunicatorRequest request = {
-			.hdr.tag = T_NeonNblocksRequest,
-			.hdr.bufid = InvalidBuffer,
-			.hdr.lsn = end_recptr,
-			.hdr.not_modified_since = end_recptr,
-			.nblocks.rinfo = InfoFromSMgrRel(reln),
+			.nblocks.hdr.tag = T_NeonNblocksRequest,
+			.nblocks.hdr.u.recepient.bufid = InvalidBuffer,
+			.nblocks.hdr.lsn = end_recptr,
+			.nblocks.hdr.not_modified_since = end_recptr,
+			.nblocks.rinfo = rinfo,
 			.nblocks.forknum = forknum,
 		};
 		relsize = (BlockNumber)communicator_request(0, &request);
-
 		relsize = Max(relsize, blkno + 1);
 
 		set_cached_relsize(rinfo, forknum, relsize);
@@ -2559,7 +2557,12 @@ neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id)
 	if (no_redo_needed)
 	{
 		SetLastWrittenLSNForBlock(end_recptr, rinfo, forknum, blkno);
-		lfc_evict(rinfo, forknum, blkno);
+		/*
+		 * Redo changes if page exists in LFC.
+		 * We should perform this check after assigning LwLSN to prevent
+		 * prefetching of some older version of the page by some other backend.
+		 */
+		no_redo_needed = !lfc_cache_contains(rinfo, forknum, blkno);
 	}
 
 	LWLockRelease(partitionLock);