Replace not-thread safe LWlocks in LFC with pthread sync primitives

pgxn/neon/neon.c

@@ -56,6 +56,7 @@ uint32		WAIT_EVENT_NEON_LFC_MAINTENANCE;
 uint32		WAIT_EVENT_NEON_LFC_READ;
 uint32		WAIT_EVENT_NEON_LFC_TRUNCATE;
 uint32		WAIT_EVENT_NEON_LFC_WRITE;
+uint32		WAIT_EVENT_NEON_LFC_CV_WAIT;
 uint32		WAIT_EVENT_NEON_PS_STARTING;
 uint32		WAIT_EVENT_NEON_PS_CONFIGURING;
 uint32		WAIT_EVENT_NEON_PS_SEND;
@@ -538,6 +539,7 @@ neon_shmem_startup_hook(void)
 	WAIT_EVENT_NEON_LFC_READ = WaitEventExtensionNew("Neon/FileCache_Read");
 	WAIT_EVENT_NEON_LFC_TRUNCATE = WaitEventExtensionNew("Neon/FileCache_Truncate");
 	WAIT_EVENT_NEON_LFC_WRITE = WaitEventExtensionNew("Neon/FileCache_Write");
+	WAIT_EVENT_NEON_LFC_CV_WAIT = WaitEventExtensionNew("Neon/FileCache_CvWait");
 	WAIT_EVENT_NEON_PS_STARTING = WaitEventExtensionNew("Neon/PS_Starting");
 	WAIT_EVENT_NEON_PS_CONFIGURING = WaitEventExtensionNew("Neon/PS_Configuring");
 	WAIT_EVENT_NEON_PS_SEND = WaitEventExtensionNew("Neon/PS_SendIO");

pgxn/neon/neon.h

@@ -28,6 +28,7 @@ extern uint32		WAIT_EVENT_NEON_LFC_MAINTENANCE;
 extern uint32		WAIT_EVENT_NEON_LFC_READ;
 extern uint32		WAIT_EVENT_NEON_LFC_TRUNCATE;
 extern uint32		WAIT_EVENT_NEON_LFC_WRITE;
+extern uint32		WAIT_EVENT_NEON_LFC_CV_WAIT;
 extern uint32		WAIT_EVENT_NEON_PS_STARTING;
 extern uint32		WAIT_EVENT_NEON_PS_CONFIGURING;
 extern uint32		WAIT_EVENT_NEON_PS_SEND;
@@ -38,6 +39,7 @@ extern uint32		WAIT_EVENT_NEON_WAL_DL;
 #define WAIT_EVENT_NEON_LFC_READ		WAIT_EVENT_BUFFILE_READ
 #define WAIT_EVENT_NEON_LFC_TRUNCATE	WAIT_EVENT_BUFFILE_TRUNCATE
 #define WAIT_EVENT_NEON_LFC_WRITE		WAIT_EVENT_BUFFILE_WRITE
+#define WAIT_EVENT_NEON_LFC_CV_WAIT 	WAIT_EVENT_BUFFILE_READ
 #define WAIT_EVENT_NEON_PS_STARTING		PG_WAIT_EXTENSION
 #define WAIT_EVENT_NEON_PS_CONFIGURING	PG_WAIT_EXTENSION
 #define WAIT_EVENT_NEON_PS_SEND			PG_WAIT_EXTENSION

pgxn/neon/pagestore_client.h

@@ -50,11 +50,18 @@ typedef enum
 
 typedef uint64 NeonRequestId;
 
+
 /* base struct for c-style inheritance */
 typedef struct
 {
 	NeonMessageTag tag;
-	NeonRequestId reqid;
+	union {
+		struct {
+			int		procno; /* process number */
+			Buffer	bufid;  /* InvalidBuffer for prefetch */
+		} recepient;
+		NeonRequestId reqid; /* two fields above temporary replace reqid, just to preserve protocol */
+	} u;
 	XLogRecPtr	lsn;
 	XLogRecPtr	not_modified_since;
 } NeonMessage;
@@ -140,6 +147,30 @@ typedef struct
 	int			segno;
 } NeonGetSlruSegmentRequest;
 
+typedef union {
+	NeonRequest hdr;
+	NeonNblocksRequest	exists;
+	NeonNblocksRequest	nblocks;
+	NeonDbSizeRequest	dbsize;
+	NeonGetPageRequest	page;
+	NeonGetSlruSegmentRequest	slru;
+} NeonCommunicatorRequest;
+
+
+typedef struct
+{
+	NeonMessageTag tag;
+	int64          value;
+} NeonCommunicatorResponse;
+
+typedef struct
+{
+	uint64			write_pos;
+	pthread_mutex_t	mutex;
+	pthread_cond_t	cond;
+	NeonCommunicatorRequest* requests;
+} NeonCommunicatorChannel;
+
 /* supertype of all the Neon*Response structs below */
 typedef NeonMessage NeonResponse;
 
@@ -161,6 +192,7 @@ typedef struct
 	char		page[FLEXIBLE_ARRAY_MEMBER];
 } NeonGetPageResponse;
 
+
 #define PS_GETPAGERESPONSE_SIZE (MAXALIGN(offsetof(NeonGetPageResponse, page) + BLCKSZ))
 
 typedef struct
@@ -184,57 +216,22 @@ typedef struct
 } NeonGetSlruSegmentResponse;
 
 
-extern StringInfoData nm_pack_request(NeonRequest *msg);
+extern void nm_pack_request(StringInfo s, NeonRequest *msg);
 extern NeonResponse *nm_unpack_response(StringInfo s);
-extern char *nm_to_string(NeonMessage *msg);
 
 /*
  * API
  */
 
-typedef uint16 shardno_t;
-
-typedef struct
-{
-	/*
-	 * Send this request to the PageServer associated with this shard.
-	 */
-	bool		(*send) (shardno_t  shard_no, NeonRequest * request);
-	/*
-	 * Blocking read for the next response of this shard.
-	 *
-	 * When a CANCEL signal is handled, the connection state will be
-	 * unmodified.
-	 */
-	NeonResponse *(*receive) (shardno_t shard_no);
-	/*
-	 * Try get the next response from the TCP buffers, if any.
-	 * Returns NULL when the data is not yet available. 
-	 */
-	NeonResponse *(*try_receive) (shardno_t shard_no);
-	/*
-	 * Make sure all requests are sent to PageServer.
-	 */
-	bool		(*flush) (shardno_t shard_no);
-	/*
-	 * Disconnect from this pageserver shard.
-	 */
-	void        (*disconnect) (shardno_t shard_no);
-} page_server_api;
-
-extern void prefetch_on_ps_disconnect(void);
-
-extern page_server_api *page_server;
-
 extern char *page_server_connstring;
-extern int	flush_every_n_requests;
 extern int	readahead_buffer_size;
 extern char *neon_timeline;
 extern char *neon_tenant;
 extern int32 max_cluster_size;
 extern int  neon_protocol_version;
+extern bool lfc_store_prefetch_result;
 
-extern shardno_t get_shard_number(BufferTag* tag);
+extern int get_shard_number(NRelFileInfo rinfo, BlockNumber blocknum);
 
 extern const f_smgr *smgr_neon(ProcNumber backend, NRelFileInfo rinfo);
 extern void smgr_init_neon(void);
@@ -301,14 +298,17 @@ extern bool lfc_cache_contains(NRelFileInfo rinfo, ForkNumber forkNum,
 							   BlockNumber blkno);
 extern int lfc_cache_containsv(NRelFileInfo rinfo, ForkNumber forkNum,
 							   BlockNumber blkno, int nblocks, bits8 *bitmap);
-extern void lfc_evict(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno);
 extern void lfc_init(void);
+extern bool lfc_prefetch(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
+						 const void* buffer, XLogRecPtr lsn);
+extern bool lfc_enabled(void);
+
 
 static inline bool
 lfc_read(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 		 void *buffer)
 {
-	bits8		rv = 0;
+	bits8		rv = 1;
 	return lfc_readv_select(rinfo, forkNum, blkno, &buffer, 1, &rv) == 1;
 }
 
@@ -319,4 +319,9 @@ lfc_write(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	return lfc_writev(rinfo, forkNum, blkno, &buffer, 1);
 }
 
+
+extern void communicator_send_request(int shard, NeonCommunicatorRequest* req);
+extern int64 communicator_receive_response(void);
+extern int64 communicator_request(int shard, NeonCommunicatorRequest* req);
+
 #endif