io_uring support

From a7f51cad2800fa9a26b65e5278e49e0c8fb5f9d6 Mon Sep 17 00:00:00 2001
From: erthalion <9erthalion6@gmail.com>
Date: Sat, 17 Aug 2019 21:22:34 +0200
Subject: [PATCH v1] io uring

POC for io_uring support on pg_prewarm example.
---
 configure                       |  75 +++++++++++++++++++++
 contrib/pg_prewarm/pg_prewarm.c |  39 +++++++++++
 src/backend/storage/file/fd.c   | 113 ++++++++++++++++++++++++++++++++
 src/backend/storage/smgr/md.c   |  64 ++++++++++++++++++
 src/backend/storage/smgr/smgr.c |  40 +++++++++++
 src/backend/utils/misc/guc.c    |  10 +++
 src/include/pg_config.h.in      |   3 +
 src/include/storage/fd.h        |  12 ++++
 src/include/storage/md.h        |   4 ++
 src/include/storage/smgr.h      |   6 ++
 10 files changed, 366 insertions(+)

diff --git a/configure b/configure
index 2c98e80c19..bb26e2dd21 100755
--- a/configure
+++ b/configure
@@ -700,6 +700,7 @@ LD
 LDFLAGS_SL
 LDFLAGS_EX
 with_zlib
+with_liburing
 with_system_tzdata
 with_libxslt
 with_libxml
@@ -863,6 +864,7 @@ with_libxml
 with_libxslt
 with_system_tzdata
 with_zlib
+with_liburing
 with_gnu_ld
 enable_largefile
 enable_float4_byval
@@ -1569,6 +1571,7 @@ Optional Packages:
   --with-system-tzdata=DIR
                           use system time zone data in DIR
   --without-zlib          do not use Zlib
+  --without-liburing      do not use liburing
   --with-gnu-ld           assume the C compiler uses GNU ld [default=no]
 
 Some influential environment variables:
@@ -8302,6 +8305,25 @@ else
 
 fi
 
+# Check whether --with-liburing was given.
+if test "${with_liburing+set}" = set; then :
+  withval=$with_liburing;
+  case $withval in
+    yes)
+      :
+      ;;
+    no)
+      :
+      ;;
+    *)
+      as_fn_error $? "no argument expected for --with-liburing option" "$LINENO" 5
+      ;;
+  esac
+
+else
+  with_liburing=yes
+
+fi
 
 
 
@@ -11795,6 +11817,59 @@ fi
 
 fi
 
+if test "$with_liburing" = yes; then
+  { $as_echo "$as_me:${as_lineno-$LINENO}: checking for io_uring_queue_init in -luring" >&5
+$as_echo_n "checking for io_uring_queue_init in -luring... " >&6; }
+if ${ac_cv_luring_init+:} false; then :
+  $as_echo_n "(cached) " >&6
+else
+  ac_check_lib_save_LIBS=$LIBS
+LIBS="-luring  $LIBS"
+cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h.  */
+
+/* Override any GCC internal prototype to avoid an error.
+   Use char because int might match the return type of a GCC
+   builtin and then its argument prototype would still apply.  */
+#ifdef __cplusplus
+extern "C"
+#endif
+char io_uring_queue_init ();
+int
+main ()
+{
+return io_uring_queue_init ();
+  ;
+  return 0;
+}
+_ACEOF
+if ac_fn_c_try_link "$LINENO"; then :
+  ac_cv_luring_init=yes
+else
+  ac_cv_luring_init=no
+fi
+rm -f core conftest.err conftest.$ac_objext \
+    conftest$ac_exeext conftest.$ac_ext
+LIBS=$ac_check_lib_save_LIBS
+fi
+{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $ac_cv_luring_init" >&5
+$as_echo "$ac_cv_luring_init" >&6; }
+if test "x$ac_cv_luring_init" = xyes; then :
+  cat >>confdefs.h <<_ACEOF
+#define HAVE_LIBURING 1
+_ACEOF
+
+  LIBS="-luring $LIBS"
+
+else
+  as_fn_error $? "io uring library not found
+If you have liburing already installed, see config.log for details on the
+failure.  It is possible the compiler isn't looking in the proper directory.
+Use --without-liburing to disable ip uring support." "$LINENO" 5
+fi
+
+fi
+
 if test "$enable_spinlocks" = yes; then
 
 $as_echo "#define HAVE_SPINLOCKS 1" >>confdefs.h
diff --git a/contrib/pg_prewarm/pg_prewarm.c b/contrib/pg_prewarm/pg_prewarm.c
index f3deb47a97..58f2dc02de 100644
--- a/contrib/pg_prewarm/pg_prewarm.c
+++ b/contrib/pg_prewarm/pg_prewarm.c
@@ -33,6 +33,7 @@ typedef enum
 {
 	PREWARM_PREFETCH,
 	PREWARM_READ,
+	PREWARM_ASYNC_READ,
 	PREWARM_BUFFER
 } PrewarmType;
 
@@ -84,6 +85,8 @@ pg_prewarm(PG_FUNCTION_ARGS)
 		ptype = PREWARM_PREFETCH;
 	else if (strcmp(ttype, "read") == 0)
 		ptype = PREWARM_READ;
+	else if (strcmp(ttype, "asyncread") == 0)
+		ptype = PREWARM_ASYNC_READ;
 	else if (strcmp(ttype, "buffer") == 0)
 		ptype = PREWARM_BUFFER;
 	else
@@ -182,6 +185,42 @@ pg_prewarm(PG_FUNCTION_ARGS)
 			++blocks_done;
 		}
 	}
+	else if (ptype == PREWARM_ASYNC_READ)
+	{
+#ifdef HAVE_LIBURING
+		int chunk = 0, chunk_size = async_queue_depth - 1;
+		int64 start = 0, stop = 0;
+
+		while (stop <= last_block)
+		{
+			start = first_block + chunk * chunk_size;
+			stop = start + chunk_size;
+
+			for (block = start; block <= stop; ++block)
+			{
+				CHECK_FOR_INTERRUPTS();
+				smgrqueueread(rel->rd_smgr, forkNumber, block, blockbuffer.data);
+			}
+
+			smgrsubmitread(rel->rd_smgr, forkNumber, block);
+
+			for (block = start; block <= stop; ++block)
+			{
+				BlockNumber readBlock;
+
+				CHECK_FOR_INTERRUPTS();
+				readBlock = smgrwaitread(rel->rd_smgr, forkNumber, block);
+				++blocks_done;
+			}
+
+			chunk++;
+		}
+#else
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("async read is not supported by this build")));
+#endif
+	}
 	else if (ptype == PREWARM_BUFFER)
 	{
 		/*
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index a76112d6cd..859e6039e7 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -79,6 +79,10 @@
 #include <sys/resource.h>		/* for getrlimit */
 #endif
 
+#ifdef HAVE_LIBURING
+#include "liburing.h"
+#endif
+
 #include "miscadmin.h"
 #include "access/xact.h"
 #include "access/xlog.h"
@@ -101,6 +105,9 @@
 #define PG_FLUSH_DATA_WORKS 1
 #endif
 
+
+int			async_queue_depth = 64;
+
 /*
  * We must leave some file descriptors free for system(), the dynamic loader,
  * and other code that tries to open files without consulting fd.c.  This
@@ -258,6 +265,9 @@ static Oid *tempTableSpaces = NULL;
 static int	numTempTableSpaces = -1;
 static int	nextTempTableSpace = 0;
 
+#ifdef HAVE_LIBURING
+struct io_uring 	ring;
+#endif
 
 /*--------------------
  *
@@ -801,6 +811,15 @@ InitFileAccess(void)
 
 	/* register proc-exit hook to ensure temp files are dropped at exit */
 	on_proc_exit(AtProcExit_Files, 0);
+
+#ifdef HAVE_LIBURING
+	int returnCode = io_uring_queue_init(async_queue_depth, &ring, 0);
+	if (returnCode < 0)
+		ereport(FATAL,
+				(errcode(ERRCODE_SYSTEM_ERROR),
+				 errmsg("Cannot init io uring async_queue_depth %d, %s",
+					    async_queue_depth, strerror(-returnCode))));
+#endif
 }
 
 /*
@@ -1912,6 +1931,96 @@ retry:
 	return returnCode;
 }
 
+int
+FileQueueRead(File file, char *buffer, int amount, off_t offset, uint32 id)
+{
+#ifdef HAVE_LIBURING
+	int				returnCode;
+	io_data		   *data;
+	struct io_uring_sqe *sqe;
+
+	Vfd		   *vfdP;
+
+	Assert(FileIsValid(file));
+
+	DO_DB(elog(LOG, "FileQueueRead: %d (%s) " INT64_FORMAT " %d %p",
+			   file, VfdCache[file].fileName,
+			   (int64) offset,
+			   amount, buffer));
+
+	returnCode = FileAccess(file);
+	if (returnCode < 0)
+		return returnCode;
+
+	vfdP = &VfdCache[file];
+
+	data = (io_data *) palloc(sizeof(io_data));
+	data->id = id;
+	data->ioVector.iov_base = buffer;
+	data->ioVector.iov_len = amount;
+
+	sqe = io_uring_get_sqe(&ring);
+	if (sqe != NULL)
+	{
+		io_uring_prep_readv(sqe, vfdP->fd, &data->ioVector, 1, offset);
+		io_uring_sqe_set_data(sqe, data);
+
+		return 0;
+	}
+	else
+	{
+		ereport(FATAL,
+				(errcode(ERRCODE_SYSTEM_ERROR),
+				 errmsg("Cannot get sqe, %s", strerror(-returnCode))));
+	}
+#else
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("async read is not supported")));
+#endif
+}
+
+int
+FileSubmitRead()
+{
+#ifdef HAVE_LIBURING
+	int			returnCode;
+	returnCode = io_uring_submit(&ring);
+	if (returnCode < 0)
+		return returnCode;
+
+	return 0;
+#else
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("async read is not supported")));
+#endif
+}
+
+io_data *
+FileWaitRead()
+{
+#ifdef HAVE_LIBURING
+	int			returnCode;
+	struct io_uring_cqe *cqe = NULL;
+
+	returnCode = io_uring_wait_cqe(&ring, &cqe);
+	if (returnCode < 0)
+	{
+		io_data	*data = (io_data *) palloc(sizeof(io_data));
+		data->returnCode = returnCode;
+		return data;
+	}
+
+	io_uring_cqe_seen(&ring, cqe);
+	return io_uring_cqe_get_data(cqe);
+#else
+	ereport(ERROR,
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+			 errmsg("async read is not supported")));
+#endif
+}
+
 int
 FileWrite(File file, char *buffer, int amount, off_t offset,
 		  uint32 wait_event_info)
@@ -2797,6 +2906,10 @@ static void
 AtProcExit_Files(int code, Datum arg)
 {
 	CleanupTempFiles(false, true);
+
+#ifdef HAVE_LIBURING
+	io_uring_queue_exit(&ring);
+#endif
 }
 
 /*
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 07f3c93d3f..1b988e051c 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -663,6 +663,70 @@ mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	}
 }
 
+/*
+ *	mdqueueread() -- Queue a read for the specified block from a relation.
+ */
+void
+mdqueueread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+	   char *buffer)
+{
+	off_t		seekpos;
+	MdfdVec    *v;
+
+	v = _mdfd_getseg(reln, forknum, blocknum, false,
+					 EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+	seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+	if (FileQueueRead(v->mdfd_vfd, buffer, BLCKSZ, seekpos, blocknum) < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not queue read for block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+}
+
+/*
+ *	mdsubmitread() -- Submit all queued reads for the specified block from a
+ *	relation.
+ */
+void
+mdsubmitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+{
+	MdfdVec    *v;
+	v = _mdfd_getseg(reln, forknum, blocknum, false,
+					 EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+	if (FileSubmitRead() < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not submit reads for block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+}
+
+/*
+ *	mdwaitread() -- Wait completion of a queued read for the specified block
+ *	from a relation.
+ */
+BlockNumber
+mdwaitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+{
+	MdfdVec    *v;
+	io_data    *data;
+	v = _mdfd_getseg(reln, forknum, blocknum, false,
+					 EXTENSION_FAIL | EXTENSION_CREATE_RECOVERY);
+
+	data = FileWaitRead();
+	if (data->returnCode < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not wait read for block %u in file \"%s\": %m",
+						blocknum, FilePathName(v->mdfd_vfd))));
+	else
+		return (BlockNumber) data->id;
+}
+
 /*
  *	mdwrite() -- Write the supplied block at the appropriate location.
  *
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index b0d9f21e68..6d73d30db4 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -53,6 +53,12 @@ typedef struct f_smgr
 								  BlockNumber blocknum);
 	void		(*smgr_read) (SMgrRelation reln, ForkNumber forknum,
 							  BlockNumber blocknum, char *buffer);
+	void		(*smgr_queue_read) (SMgrRelation reln, ForkNumber forknum,
+							  BlockNumber blocknum, char *buffer);
+	void		(*smgr_submit_read) (SMgrRelation reln, ForkNumber forknum,
+							  BlockNumber blocknum);
+	BlockNumber	(*smgr_wait_read) (SMgrRelation reln, ForkNumber forknum,
+							  BlockNumber blocknum);
 	void		(*smgr_write) (SMgrRelation reln, ForkNumber forknum,
 							   BlockNumber blocknum, char *buffer, bool skipFsync);
 	void		(*smgr_writeback) (SMgrRelation reln, ForkNumber forknum,
@@ -76,6 +82,9 @@ static const f_smgr smgrsw[] = {
 		.smgr_extend = mdextend,
 		.smgr_prefetch = mdprefetch,
 		.smgr_read = mdread,
+		.smgr_queue_read = mdqueueread,
+		.smgr_submit_read = mdsubmitread,
+		.smgr_wait_read = mdwaitread,
 		.smgr_write = mdwrite,
 		.smgr_writeback = mdwriteback,
 		.smgr_nblocks = mdnblocks,
@@ -565,6 +574,37 @@ smgrread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	smgrsw[reln->smgr_which].smgr_read(reln, forknum, blocknum, buffer);
 }
 
+/*
+ *	smgrqueueread() -- queue a read for a particular block from a relation into
+ *					   the supplied buffer.
+ */
+void
+smgrqueueread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+		 char *buffer)
+{
+	smgrsw[reln->smgr_which].smgr_queue_read(reln, forknum, blocknum, buffer);
+}
+
+/*
+ *	smgrsubmitread() -- submit all reads for a particular block from a relation
+ *						into the supplied buffer.
+ */
+void
+smgrsubmitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+{
+	smgrsw[reln->smgr_which].smgr_submit_read(reln, forknum, blocknum);
+}
+
+/*
+ *	smgrwaitread() -- wait a reads for a particular block from a relation into
+ *					  the supplied buffer.
+ */
+BlockNumber
+smgrwaitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+{
+	return smgrsw[reln->smgr_which].smgr_wait_read(reln, forknum, blocknum);
+}
+
 /*
  *	smgrwrite() -- Write the supplied buffer out.
  *
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 90ffd89339..956d6cfc02 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -2338,6 +2338,16 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"async_queue_depth", PGC_POSTMASTER, RESOURCES_KERNEL,
+			gettext_noop("Queue depth"),
+			NULL
+		},
+		&async_queue_depth,
+		64, 25, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	/*
 	 * See also CheckRequiredParameterValues() if this parameter changes
 	 */
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index 512213aa32..21c682f8f0 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -377,6 +377,9 @@
 /* Define to 1 if you have the `z' library (-lz). */
 #undef HAVE_LIBZ
 
+/* Define to 1 if you have the `uring' library (-luring). */
+#undef HAVE_LIBURING
+
 /* Define to 1 if the system has the type `locale_t'. */
 #undef HAVE_LOCALE_T
 
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index d2a8c52044..dcc9336f10 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -47,6 +47,7 @@ typedef int File;
 
 /* GUC parameter */
 extern PGDLLIMPORT int max_files_per_process;
+extern PGDLLIMPORT int async_queue_depth;
 extern PGDLLIMPORT bool data_sync_retry;
 
 /*
@@ -67,6 +68,13 @@ extern int	max_safe_fds;
 #define FILE_POSSIBLY_DELETED(err)	((err) == ENOENT || (err) == EACCES)
 #endif
 
+typedef struct io_data
+{
+	struct iovec 	 ioVector;
+	uint32		 	 id;
+	int				 returnCode;
+} io_data;
+
 /*
  * prototypes for functions in fd.c
  */
@@ -78,6 +86,10 @@ extern File OpenTemporaryFile(bool interXact);
 extern void FileClose(File file);
 extern int	FilePrefetch(File file, off_t offset, int amount, uint32 wait_event_info);
 extern int	FileRead(File file, char *buffer, int amount, off_t offset, uint32 wait_event_info);
+extern int	FileQueueRead(File file, char *buffer, int amount, off_t offset,
+						  uint32 id);
+extern int	FileSubmitRead();
+extern io_data *FileWaitRead();
 extern int	FileWrite(File file, char *buffer, int amount, off_t offset, uint32 wait_event_info);
 extern int	FileSync(File file, uint32 wait_event_info);
 extern off_t FileSize(File file);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index c0f05e23ff..1048853232 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -32,6 +32,10 @@ extern void mdprefetch(SMgrRelation reln, ForkNumber forknum,
 					   BlockNumber blocknum);
 extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 				   char *buffer);
+extern void mdqueueread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+				   char *buffer);
+extern void mdsubmitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum);
+extern BlockNumber mdwaitread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum);
 extern void mdwrite(SMgrRelation reln, ForkNumber forknum,
 					BlockNumber blocknum, char *buffer, bool skipFsync);
 extern void mdwriteback(SMgrRelation reln, ForkNumber forknum,
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index d286c8c7b1..0e51f26460 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -97,6 +97,12 @@ extern void smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
 					 BlockNumber blocknum, char *buffer);
+extern void smgrqueueread(SMgrRelation reln, ForkNumber forknum,
+					 BlockNumber blocknum, char *buffer);
+extern void smgrsubmitread(SMgrRelation reln, ForkNumber forknum,
+					 BlockNumber blocknum);
+extern BlockNumber smgrwaitread(SMgrRelation reln, ForkNumber forknum,
+					 BlockNumber blocknum);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,
 					  BlockNumber blocknum, char *buffer, bool skipFsync);
 extern void smgrwriteback(SMgrRelation reln, ForkNumber forknum,
-- 
2.21.0