On 06.04.2018 20:00, Konstantin Knizhnik wrote:
Attached please find new version of the patch with several bug fixes
+ support of more than one session pools associated with different ports.
Now it is possible to make postmaster listen several ports for
accepting pooled connections, while leaving main Postgres port for
dedicated backends.
Each session pool is intended to be used for particular database/user
combination.
Sorry, wrong patch was attached.
--
Konstantin Knizhnik
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 93c4bbf..dfc072c 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -194,6 +194,7 @@ char *namespace_search_path = NULL;
/* Local functions */
static void recomputeNamespacePath(void);
static void InitTempTableNamespace(void);
+static Oid GetTempTableNamespace(void);
static void RemoveTempRelations(Oid tempNamespaceId);
static void RemoveTempRelationsCallback(int code, Datum arg);
static void NamespaceCallback(Datum arg, int cacheid, uint32 hashvalue);
@@ -441,9 +442,7 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
if (strcmp(newRelation->schemaname, "pg_temp") == 0)
{
/* Initialize temp namespace if first time through */
- if (!OidIsValid(myTempNamespace))
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
/* use exact schema given */
namespaceId = get_namespace_oid(newRelation->schemaname, false);
@@ -452,9 +451,7 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
else if (newRelation->relpersistence == RELPERSISTENCE_TEMP)
{
/* Initialize temp namespace if first time through */
- if (!OidIsValid(myTempNamespace))
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
else
{
@@ -463,8 +460,7 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
if (activeTempCreationPending)
{
/* Need to initialize temp namespace */
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
namespaceId = activeCreationNamespace;
if (!OidIsValid(namespaceId))
@@ -2902,9 +2898,7 @@ LookupCreationNamespace(const char *nspname)
if (strcmp(nspname, "pg_temp") == 0)
{
/* Initialize temp namespace if first time through */
- if (!OidIsValid(myTempNamespace))
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
namespaceId = get_namespace_oid(nspname, false);
@@ -2967,9 +2961,7 @@ QualifiedNameGetCreationNamespace(List *names, char **objname_p)
if (strcmp(schemaname, "pg_temp") == 0)
{
/* Initialize temp namespace if first time through */
- if (!OidIsValid(myTempNamespace))
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
/* use exact schema given */
namespaceId = get_namespace_oid(schemaname, false);
@@ -2982,8 +2974,7 @@ QualifiedNameGetCreationNamespace(List *names, char **objname_p)
if (activeTempCreationPending)
{
/* Need to initialize temp namespace */
- InitTempTableNamespace();
- return myTempNamespace;
+ return GetTempTableNamespace();
}
namespaceId = activeCreationNamespace;
if (!OidIsValid(namespaceId))
@@ -3250,8 +3241,11 @@ void
SetTempNamespaceState(Oid tempNamespaceId, Oid tempToastNamespaceId)
{
/* Worker should not have created its own namespaces ... */
- Assert(myTempNamespace == InvalidOid);
- Assert(myTempToastNamespace == InvalidOid);
+ if (!ActiveSession)
+ {
+ Assert(myTempNamespace == InvalidOid);
+ Assert(myTempToastNamespace == InvalidOid);
+ }
Assert(myTempNamespaceSubID == InvalidSubTransactionId);
/* Assign same namespace OIDs that leader has */
@@ -3771,6 +3765,22 @@ recomputeNamespacePath(void)
list_free(oidlist);
}
+static Oid
+GetTempTableNamespace(void)
+{
+ if (ActiveSession)
+ {
+ if (!OidIsValid(ActiveSession->tempNamespace))
+ InitTempTableNamespace();
+ }
+ else
+ {
+ if (!OidIsValid(myTempNamespace))
+ InitTempTableNamespace();
+ }
+ return myTempNamespace;
+}
+
/*
* InitTempTableNamespace
* Initialize temp table namespace on first use in a particular backend
@@ -3782,8 +3792,6 @@ InitTempTableNamespace(void)
Oid namespaceId;
Oid toastspaceId;
- Assert(!OidIsValid(myTempNamespace));
-
/*
* First, do permission check to see if we are authorized to make temp
* tables. We use a nonstandard error message here since "databasename:
@@ -3822,7 +3830,10 @@ InitTempTableNamespace(void)
(errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
errmsg("cannot create temporary tables during a parallel operation")));
- snprintf(namespaceName, sizeof(namespaceName), "pg_temp_%d", MyBackendId);
+ if (ActiveSession)
+ snprintf(namespaceName, sizeof(namespaceName), "pg_temp_%d_%s", MyBackendId, ActiveSession->id);
+ else
+ snprintf(namespaceName, sizeof(namespaceName), "pg_temp_%d", MyBackendId);
namespaceId = get_namespace_oid(namespaceName, true);
if (!OidIsValid(namespaceId))
@@ -3854,8 +3865,10 @@ InitTempTableNamespace(void)
* it. (We assume there is no need to clean it out if it does exist, since
* dropping a parent table should make its toast table go away.)
*/
- snprintf(namespaceName, sizeof(namespaceName), "pg_toast_temp_%d",
- MyBackendId);
+ if (ActiveSession)
+ snprintf(namespaceName, sizeof(namespaceName), "pg_toast_temp_%d_%s", MyBackendId, ActiveSession->id);
+ else
+ snprintf(namespaceName, sizeof(namespaceName), "pg_toast_temp_%d", MyBackendId);
toastspaceId = get_namespace_oid(namespaceName, true);
if (!OidIsValid(toastspaceId))
@@ -3873,7 +3886,11 @@ InitTempTableNamespace(void)
*/
myTempNamespace = namespaceId;
myTempToastNamespace = toastspaceId;
-
+ if (ActiveSession)
+ {
+ ActiveSession->tempNamespace = namespaceId;
+ ActiveSession->tempToastNamespace = toastspaceId;
+ }
/* It should not be done already. */
AssertState(myTempNamespaceSubID == InvalidSubTransactionId);
myTempNamespaceSubID = GetCurrentSubTransactionId();
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
index cff49ba..b728ab1 100644
--- a/src/backend/catalog/storage.c
+++ b/src/backend/catalog/storage.c
@@ -25,6 +25,7 @@
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "catalog/catalog.h"
+#include "catalog/namespace.h"
#include "catalog/storage.h"
#include "catalog/storage_xlog.h"
#include "storage/freespace.h"
diff --git a/src/backend/commands/prepare.c b/src/backend/commands/prepare.c
index b945b15..8e8a737 100644
--- a/src/backend/commands/prepare.c
+++ b/src/backend/commands/prepare.c
@@ -813,3 +813,32 @@ build_regtype_array(Oid *param_types, int num_params)
result = construct_array(tmp_ary, num_params, REGTYPEOID, 4, true, 'i');
return PointerGetDatum(result);
}
+
+/*
+ * Drop all statements prepared in the specified session.
+ */
+void
+DropSessionPreparedStatements(char const* sessionId)
+{
+ HASH_SEQ_STATUS seq;
+ PreparedStatement *entry;
+ size_t idLen = strlen(sessionId);
+
+ /* nothing cached */
+ if (!prepared_queries)
+ return;
+
+ /* walk over cache */
+ hash_seq_init(&seq, prepared_queries);
+ while ((entry = hash_seq_search(&seq)) != NULL)
+ {
+ if (strncmp(entry->stmt_name, sessionId, idLen) == 0 && entry->stmt_name[idLen] == '.')
+ {
+ /* Release the plancache entry */
+ DropCachedPlan(entry->plansource);
+
+ /* Now we can remove the hash table entry */
+ hash_search(prepared_queries, entry->stmt_name, HASH_REMOVE, NULL);
+ }
+ }
+}
diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c
index a4f6d4d..7f40edb 100644
--- a/src/backend/libpq/pqcomm.c
+++ b/src/backend/libpq/pqcomm.c
@@ -1029,6 +1029,17 @@ pq_peekbyte(void)
}
/* --------------------------------
+ * pq_available_bytes - get number of buffered bytes available for reading.
+ *
+ * --------------------------------
+ */
+int
+pq_available_bytes(void)
+{
+ return PqRecvLength - PqRecvPointer;
+}
+
+/* --------------------------------
* pq_getbyte_if_available - get a single byte from connection,
* if available
*
diff --git a/src/backend/port/Makefile b/src/backend/port/Makefile
index aba1e92..56ec998 100644
--- a/src/backend/port/Makefile
+++ b/src/backend/port/Makefile
@@ -21,7 +21,7 @@ subdir = src/backend/port
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
-OBJS = atomics.o dynloader.o pg_sema.o pg_shmem.o $(TAS)
+OBJS = atomics.o dynloader.o pg_sema.o pg_shmem.o send_sock.o $(TAS)
ifeq ($(PORTNAME), win32)
SUBDIRS += win32
diff --git a/src/backend/port/send_sock.c b/src/backend/port/send_sock.c
new file mode 100644
index 0000000..5f3f929
--- /dev/null
+++ b/src/backend/port/send_sock.c
@@ -0,0 +1,151 @@
+/*-------------------------------------------------------------------------
+ *
+ * send_sock.c
+ * Send socket descriptor to another process
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/port/send_sock.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/wait.h>
+#include <time.h>
+#include <unistd.h>
+
+#ifdef WIN32
+typedef struct
+{
+ SOCKET origsocket;
+ WSAPROTOCOL_INFO wsainfo;
+} InheritableSocket;
+#endif
+
+/*
+ * Send socket descriptor "sock" to backend process through Unix socket "chan"
+ */
+int pg_send_sock(pgsocket chan, pgsocket sock, pid_t pid)
+{
+#ifdef WIN32
+ InheritableSocket dst;
+ size_t rc;
+ dst.origsocket = sock;
+ if (WSADuplicateSocket(sock, pid, &dst.wsainfo) != 0)
+ {
+ ereport(FATAL,
+ (errmsg("could not duplicate socket %d for use in backend: error code %d",
+ (int)sock, WSAGetLastError())));
+ return -1;
+ }
+ rc = send(chan, &dst, sizeof(dst), 0);
+ if (rc != sizeof(dst))
+ {
+ ereport(FATAL,
+ (errmsg("Failed to send inheritable socket: rc=%d, error code %d",
+ (int)rc, WSAGetLastError())));
+ return -1;
+ }
+ return 0;
+#else
+ struct msghdr msg = { 0 };
+ struct iovec io;
+ struct cmsghdr * cmsg;
+ char buf[CMSG_SPACE(sizeof(sock))];
+ memset(buf, '\0', sizeof(buf));
+
+ /* On Mac OS X, the struct iovec is needed, even if it points to minimal data */
+ io.iov_base = "";
+ io.iov_len = 1;
+
+ msg.msg_iov = &io;
+ msg.msg_iovlen = 1;
+ msg.msg_control = buf;
+ msg.msg_controllen = sizeof(buf);
+
+ cmsg = CMSG_FIRSTHDR(&msg);
+ cmsg->cmsg_level = SOL_SOCKET;
+ cmsg->cmsg_type = SCM_RIGHTS;
+ cmsg->cmsg_len = CMSG_LEN(sizeof(sock));
+
+ memcpy(CMSG_DATA(cmsg), &sock, sizeof(sock));
+ msg.msg_controllen = cmsg->cmsg_len;
+
+ if (sendmsg(chan, &msg, 0) < 0)
+ {
+ return -1;
+ }
+ return 0;
+#endif
+}
+
+
+/*
+ * Receive socket descriptor from postmaster process through Unix socket "chan"
+ */
+pgsocket pg_recv_sock(pgsocket chan)
+{
+#ifdef WIN32
+ InheritableSocket src;
+ SOCKET s;
+ size_t rc = recv(chan, &src, sizeof(src), 0);
+ if (rc != sizeof(src))
+ {
+ ereport(FATAL,
+ (errmsg("Failed to receive inheritable socket: rc=%d, error code %d",
+ (int)rc, WSAGetLastError())));
+ }
+ s = WSASocket(FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
+ FROM_PROTOCOL_INFO,
+ &src.wsainfo,
+ 0,
+ 0);
+ if (s == INVALID_SOCKET)
+ {
+ ereport(FATAL,
+ (errmsg("could not create inherited socket: error code %d\n",
+ WSAGetLastError())));
+ }
+
+ /*
+ * To make sure we don't get two references to the same socket, close
+ * the original one. (This would happen when inheritance actually
+ * works..
+ */
+ closesocket(src.origsocket);
+ return s;
+#else
+ struct msghdr msg = {0};
+ char c_buffer[256];
+ char m_buffer[256];
+ struct iovec io;
+ struct cmsghdr * cmsg;
+ pgsocket sock;
+
+ io.iov_base = m_buffer;
+ io.iov_len = sizeof(m_buffer);
+ msg.msg_iov = &io;
+ msg.msg_iovlen = 1;
+
+ msg.msg_control = c_buffer;
+ msg.msg_controllen = sizeof(c_buffer);
+
+ if (recvmsg(chan, &msg, 0) < 0)
+ {
+ return -1;
+ }
+
+ cmsg = CMSG_FIRSTHDR(&msg);
+ memcpy(&sock, CMSG_DATA(cmsg), sizeof(sock));
+
+ return sock;
+#endif
+}
diff --git a/src/backend/port/win32/socket.c b/src/backend/port/win32/socket.c
index f4356fe..7fd901f 100644
--- a/src/backend/port/win32/socket.c
+++ b/src/backend/port/win32/socket.c
@@ -726,3 +726,65 @@ pgwin32_socket_strerror(int err)
}
return wserrbuf;
}
+
+int pgwin32_socketpair(int domain, int type, int protocol, SOCKET socks[2])
+{
+ union {
+ struct sockaddr_in inaddr;
+ struct sockaddr addr;
+ } a;
+ SOCKET listener;
+ int e;
+ socklen_t addrlen = sizeof(a.inaddr);
+ DWORD flags = 0;
+ int reuse = 1;
+
+ socks[0] = socks[1] = -1;
+
+ listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+ if (listener == -1)
+ return SOCKET_ERROR;
+
+ memset(&a, 0, sizeof(a));
+ a.inaddr.sin_family = AF_INET;
+ a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ a.inaddr.sin_port = 0;
+
+ for (;;) {
+ if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
+ (char*) &reuse, (socklen_t) sizeof(reuse)) == -1)
+ break;
+ if (bind(listener, &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+ break;
+
+ memset(&a, 0, sizeof(a));
+ if (getsockname(listener, &a.addr, &addrlen) == SOCKET_ERROR)
+ break;
+ a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ a.inaddr.sin_family = AF_INET;
+
+ if (listen(listener, 1) == SOCKET_ERROR)
+ break;
+
+ socks[0] = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
+ if (socks[0] == -1)
+ break;
+ if (connect(socks[0], &a.addr, sizeof(a.inaddr)) == SOCKET_ERROR)
+ break;
+
+ socks[1] = accept(listener, NULL, NULL);
+ if (socks[1] == -1)
+ break;
+
+ closesocket(listener);
+ return 0;
+ }
+
+ e = WSAGetLastError();
+ closesocket(listener);
+ closesocket(socks[0]);
+ closesocket(socks[1]);
+ WSASetLastError(e);
+ socks[0] = socks[1] = -1;
+ return SOCKET_ERROR;
+}
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index f3ddf82..707b880 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -169,6 +169,7 @@ typedef struct bkend
pid_t pid; /* process id of backend */
int32 cancel_key; /* cancel key for cancels for this backend */
int child_slot; /* PMChildSlot for this backend, if any */
+ pgsocket session_send_sock; /* Write end of socket pipe to this backend used to send session socket descriptor to the backend process */
/*
* Flavor of backend or auxiliary process. Note that BACKEND_TYPE_WALSND
@@ -179,6 +180,8 @@ typedef struct bkend
bool dead_end; /* is it going to send an error and quit? */
bool bgworker_notify; /* gets bgworker start/stop notifications */
dlist_node elem; /* list link in BackendList */
+ int session_pool_id; /* identifier of backends session pool */
+ int worker_id; /* identifier of worker within session pool */
} Backend;
static dlist_head BackendList = DLIST_STATIC_INIT(BackendList);
@@ -189,7 +192,14 @@ static Backend *ShmemBackendArray;
BackgroundWorker *MyBgworkerEntry = NULL;
+typedef struct PostmasterSessionPool
+{
+ Backend** workers; /* pool backends */
+ int n_workers; /* number of launched worker backends in this pool so far */
+ int rr_index; /* index of current backends used to implement round-robin distribution of sessions through backends. */
+} PostmasterSessionPool;
+static PostmasterSessionPool SessionPools[MAX_SESSION_PORTS];
/* The socket number we are listening for connections on */
int PostPortNumber;
@@ -213,7 +223,7 @@ int ReservedBackends;
/* The socket(s) we're listening to. */
#define MAXLISTEN 64
-static pgsocket ListenSocket[MAXLISTEN];
+static pgsocket ListenSocket[MAX_SESSION_PORTS][MAXLISTEN];
/*
* Set by the -o option
@@ -411,8 +421,7 @@ static void BackendInitialize(Port *port);
static void BackendRun(Port *port) pg_attribute_noreturn();
static void ExitPostmaster(int status) pg_attribute_noreturn();
static int ServerLoop(void);
-static int BackendStartup(Port *port);
-static int ProcessStartupPacket(Port *port, bool SSLdone);
+static int BackendStartup(Port *port, int session_pool_id);
static void SendNegotiateProtocolVersion(List *unrecognized_protocol_options);
static void processCancelRequest(Port *port, void *pkt);
static int initMasks(fd_set *rmask);
@@ -485,8 +494,9 @@ typedef struct
{
Port port;
InheritableSocket portsocket;
+ InheritableSocket sessionsocket;
char DataDir[MAXPGPATH];
- pgsocket ListenSocket[MAXLISTEN];
+ pgsocket ListenSocket[MAX_SESSION_PORTS][MAXLISTEN];
int32 MyCancelKey;
int MyPMChildSlot;
#ifndef WIN32
@@ -577,7 +587,7 @@ PostmasterMain(int argc, char *argv[])
int status;
char *userDoption = NULL;
bool listen_addr_saved = false;
- int i;
+ int i, j;
char *output_config_variable = NULL;
MyProcPid = PostmasterPid = getpid();
@@ -990,8 +1000,9 @@ PostmasterMain(int argc, char *argv[])
* First, mark them all closed, and set up an on_proc_exit function that's
* charged with closing the sockets again at postmaster shutdown.
*/
- for (i = 0; i < MAXLISTEN; i++)
- ListenSocket[i] = PGINVALID_SOCKET;
+ for (i = 0; i <= SessionPoolPorts; i++)
+ for (j = 0; j < MAXLISTEN; j++)
+ ListenSocket[i][j] = PGINVALID_SOCKET;
on_proc_exit(CloseServerPorts, 0);
@@ -1019,33 +1030,35 @@ PostmasterMain(int argc, char *argv[])
{
char *curhost = (char *) lfirst(l);
- if (strcmp(curhost, "*") == 0)
- status = StreamServerPort(AF_UNSPEC, NULL,
- (unsigned short) PostPortNumber,
- NULL,
- ListenSocket, MAXLISTEN);
- else
- status = StreamServerPort(AF_UNSPEC, curhost,
- (unsigned short) PostPortNumber,
- NULL,
- ListenSocket, MAXLISTEN);
-
- if (status == STATUS_OK)
+ for (i = 0; i <= SessionPoolPorts; i++)
{
- success++;
- /* record the first successful host addr in lockfile */
- if (!listen_addr_saved)
+ if (strcmp(curhost, "*") == 0)
+ status = StreamServerPort(AF_UNSPEC, NULL,
+ (unsigned short) PostPortNumber + i,
+ NULL,
+ ListenSocket[i], MAXLISTEN);
+ else
+ status = StreamServerPort(AF_UNSPEC, curhost,
+ (unsigned short) PostPortNumber + i,
+ NULL,
+ ListenSocket[i], MAXLISTEN);
+
+ if (status == STATUS_OK)
{
- AddToDataDirLockFile(LOCK_FILE_LINE_LISTEN_ADDR, curhost);
- listen_addr_saved = true;
+ success++;
+ /* record the first successful host addr in lockfile */
+ if (!listen_addr_saved)
+ {
+ AddToDataDirLockFile(LOCK_FILE_LINE_LISTEN_ADDR, curhost);
+ listen_addr_saved = true;
+ }
}
+ else
+ ereport(WARNING,
+ (errmsg("could not create listen socket for \"%s\"",
+ curhost)));
}
- else
- ereport(WARNING,
- (errmsg("could not create listen socket for \"%s\"",
- curhost)));
}
-
if (!success && elemlist != NIL)
ereport(FATAL,
(errmsg("could not create any TCP/IP sockets")));
@@ -1056,7 +1069,7 @@ PostmasterMain(int argc, char *argv[])
#ifdef USE_BONJOUR
/* Register for Bonjour only if we opened TCP socket(s) */
- if (enable_bonjour && ListenSocket[0] != PGINVALID_SOCKET)
+ if (enable_bonjour && ListenSocket[0][0] != PGINVALID_SOCKET)
{
DNSServiceErrorType err;
@@ -1117,24 +1130,26 @@ PostmasterMain(int argc, char *argv[])
{
char *socketdir = (char *) lfirst(l);
- status = StreamServerPort(AF_UNIX, NULL,
- (unsigned short) PostPortNumber,
- socketdir,
- ListenSocket, MAXLISTEN);
-
- if (status == STATUS_OK)
+ for (i = 0; i <= SessionPoolPorts; i++)
{
- success++;
- /* record the first successful Unix socket in lockfile */
- if (success == 1)
- AddToDataDirLockFile(LOCK_FILE_LINE_SOCKET_DIR, socketdir);
+ status = StreamServerPort(AF_UNIX, NULL,
+ (unsigned short) PostPortNumber + i,
+ socketdir,
+ ListenSocket[i], MAXLISTEN);
+
+ if (status == STATUS_OK)
+ {
+ success++;
+ /* record the first successful Unix socket in lockfile */
+ if (success == 1)
+ AddToDataDirLockFile(LOCK_FILE_LINE_SOCKET_DIR, socketdir);
+ }
+ else
+ ereport(WARNING,
+ (errmsg("could not create Unix-domain socket in directory \"%s\"",
+ socketdir)));
}
- else
- ereport(WARNING,
- (errmsg("could not create Unix-domain socket in directory \"%s\"",
- socketdir)));
}
-
if (!success && elemlist != NIL)
ereport(FATAL,
(errmsg("could not create any Unix-domain sockets")));
@@ -1147,7 +1162,7 @@ PostmasterMain(int argc, char *argv[])
/*
* check that we have some socket to listen on
*/
- if (ListenSocket[0] == PGINVALID_SOCKET)
+ if (ListenSocket[0][0] == PGINVALID_SOCKET)
ereport(FATAL,
(errmsg("no socket created for listening")));
@@ -1379,7 +1394,7 @@ PostmasterMain(int argc, char *argv[])
static void
CloseServerPorts(int status, Datum arg)
{
- int i;
+ int i, j;
/*
* First, explicitly close all the socket FDs. We used to just let this
@@ -1387,12 +1402,15 @@ CloseServerPorts(int status, Datum arg)
* before we remove the postmaster.pid lockfile; otherwise there's a race
* condition if a new postmaster wants to re-use the TCP port number.
*/
- for (i = 0; i < MAXLISTEN; i++)
+ for (i = 0; i <= SessionPoolPorts; i++)
{
- if (ListenSocket[i] != PGINVALID_SOCKET)
+ for (j = 0; j < MAXLISTEN; j++)
{
- StreamClose(ListenSocket[i]);
- ListenSocket[i] = PGINVALID_SOCKET;
+ if (ListenSocket[i][j] != PGINVALID_SOCKET)
+ {
+ StreamClose(ListenSocket[i][j]);
+ ListenSocket[i][j] = PGINVALID_SOCKET;
+ }
}
}
@@ -1741,27 +1759,30 @@ ServerLoop(void)
*/
if (selres > 0)
{
- int i;
+ int i, j;
- for (i = 0; i < MAXLISTEN; i++)
+ for (i = 0; i <= SessionPoolPorts; i++)
{
- if (ListenSocket[i] == PGINVALID_SOCKET)
- break;
- if (FD_ISSET(ListenSocket[i], &rmask))
+ for (j = 0; j < MAXLISTEN; j++)
{
- Port *port;
-
- port = ConnCreate(ListenSocket[i]);
- if (port)
+ if (ListenSocket[i][j] == PGINVALID_SOCKET)
+ break;
+ if (FD_ISSET(ListenSocket[i][j], &rmask))
{
- BackendStartup(port);
-
- /*
- * We no longer need the open socket or port structure
- * in this process
- */
- StreamClose(port->sock);
- ConnFree(port);
+ Port *port;
+
+ port = ConnCreate(ListenSocket[i][j]);
+ if (port)
+ {
+ BackendStartup(port, i);
+
+ /*
+ * We no longer need the open socket or port structure
+ * in this process
+ */
+ StreamClose(port->sock);
+ ConnFree(port);
+ }
}
}
}
@@ -1913,20 +1934,23 @@ static int
initMasks(fd_set *rmask)
{
int maxsock = -1;
- int i;
+ int i, j;
FD_ZERO(rmask);
- for (i = 0; i < MAXLISTEN; i++)
+ for (i = 0; i <= SessionPoolPorts; i++)
{
- int fd = ListenSocket[i];
+ for (j = 0; j < MAXLISTEN; j++)
+ {
+ int fd = ListenSocket[i][j];
- if (fd == PGINVALID_SOCKET)
- break;
- FD_SET(fd, rmask);
+ if (fd == PGINVALID_SOCKET)
+ break;
+ FD_SET(fd, rmask);
- if (fd > maxsock)
- maxsock = fd;
+ if (fd > maxsock)
+ maxsock = fd;
+ }
}
return maxsock + 1;
@@ -1944,8 +1968,8 @@ initMasks(fd_set *rmask)
* send anything to the client, which would typically be appropriate
* if we detect a communications failure.)
*/
-static int
-ProcessStartupPacket(Port *port, bool SSLdone)
+int
+ProcessStartupPacket(Port *port, bool SSLdone, MemoryContext memctx)
{
int32 len;
void *buf;
@@ -1978,7 +2002,6 @@ ProcessStartupPacket(Port *port, bool SSLdone)
errmsg("invalid length of startup packet")));
return STATUS_ERROR;
}
-
/*
* Allocate at least the size of an old-style startup packet, plus one
* extra byte, and make sure all are zeroes. This ensures we will have
@@ -2043,7 +2066,7 @@ retry1:
#endif
/* regular startup packet, cancel, etc packet should follow... */
/* but not another SSL negotiation request */
- return ProcessStartupPacket(port, true);
+ return ProcessStartupPacket(port, true, memctx);
}
/* Could add additional special packet types here */
@@ -2073,7 +2096,7 @@ retry1:
* not worry about leaking this storage on failure, since we aren't in the
* postmaster process anymore.
*/
- oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ oldcontext = MemoryContextSwitchTo(memctx);
if (PG_PROTOCOL_MAJOR(proto) >= 3)
{
@@ -2099,7 +2122,6 @@ retry1:
if (valoffset >= len)
break; /* missing value, will complain below */
valptr = ((char *) buf) + valoffset;
-
if (strcmp(nameptr, "database") == 0)
port->database_name = pstrdup(valptr);
else if (strcmp(nameptr, "user") == 0)
@@ -2449,7 +2471,7 @@ ConnCreate(int serverFd)
ConnFree(port);
return NULL;
}
-
+ SessionPoolSock = PGINVALID_SOCKET;
/*
* Allocate GSSAPI specific state struct
*/
@@ -2498,7 +2520,7 @@ ConnFree(Port *conn)
void
ClosePostmasterPorts(bool am_syslogger)
{
- int i;
+ int i, j;
#ifndef WIN32
@@ -2515,12 +2537,15 @@ ClosePostmasterPorts(bool am_syslogger)
#endif
/* Close the listen sockets */
- for (i = 0; i < MAXLISTEN; i++)
+ for (i = 0; i < SessionPoolPorts; i++)
{
- if (ListenSocket[i] != PGINVALID_SOCKET)
+ for (j = 0; j < MAXLISTEN; j++)
{
- StreamClose(ListenSocket[i]);
- ListenSocket[i] = PGINVALID_SOCKET;
+ if (ListenSocket[i][j] != PGINVALID_SOCKET)
+ {
+ StreamClose(ListenSocket[i][j]);
+ ListenSocket[i][j] = PGINVALID_SOCKET;
+ }
}
}
@@ -3236,6 +3261,28 @@ CleanupBackgroundWorker(int pid,
}
/*
+ * Unlink backend from backend's list and free memory
+ */
+static void UnlinkBackend(Backend* bp)
+{
+ if (bp->bkend_type == BACKEND_TYPE_NORMAL
+ && bp->session_send_sock != PGINVALID_SOCKET)
+ {
+ PostmasterSessionPool* pool = &SessionPools[bp->session_pool_id];
+ Assert(pool->n_workers > bp->worker_id && pool->workers[bp->worker_id] == bp);
+ if (--pool->n_workers != 0)
+ {
+ pool->workers[bp->worker_id] = pool->workers[pool->n_workers];
+ pool->rr_index %= pool->n_workers;
+ }
+ closesocket(bp->session_send_sock);
+ elog(DEBUG2, "Cleanup backend %d", bp->pid);
+ }
+ dlist_delete(&bp->elem);
+ free(bp);
+}
+
+/*
* CleanupBackend -- cleanup after terminated backend.
*
* Remove all local state associated with backend.
@@ -3312,8 +3359,7 @@ CleanupBackend(int pid,
*/
BackgroundWorkerStopNotifications(bp->pid);
}
- dlist_delete(iter.cur);
- free(bp);
+ UnlinkBackend(bp);
break;
}
}
@@ -3415,8 +3461,7 @@ HandleChildCrash(int pid, int exitstatus, const char *procname)
ShmemBackendArrayRemove(bp);
#endif
}
- dlist_delete(iter.cur);
- free(bp);
+ UnlinkBackend(bp);
/* Keep looping so we can signal remaining backends */
}
else
@@ -4013,16 +4058,35 @@ TerminateChildren(int signal)
* Note: if you change this code, also consider StartAutovacuumWorker.
*/
static int
-BackendStartup(Port *port)
+BackendStartup(Port *port, int session_pool_id)
{
Backend *bn; /* for backend cleanup */
pid_t pid;
+ pgsocket session_pipe[2];
+ PostmasterSessionPool* pool = &SessionPools[session_pool_id];
+ bool dedicated_backend = SessionPoolSize == 0 || (SessionPoolPorts != 0 && session_pool_id == 0);
+
+ if (!dedicated_backend && pool->n_workers >= SessionPoolSize)
+ {
+ Backend* worker = pool->workers[pool->rr_index];
+ /* In case of session pooling instead of spawning new backend open new session at one of the existed backends. */
+ elog(DEBUG2, "Start new session in pool %d for socket %d at backend %d",
+ session_pool_id, port->sock, worker->pid);
+ /* Send connection socket to the worker backend */
+ if (pg_send_sock(worker->session_send_sock, port->sock, worker->pid) < 0)
+ elog(FATAL, "Failed to send session socket: %m");
+
+ pool->rr_index = (pool->rr_index + 1) % pool->n_workers; /* round-robin */
+
+ return STATUS_OK;
+ }
+
/*
* Create backend data structure. Better before the fork() so we can
* handle failure cleanly.
*/
- bn = (Backend *) malloc(sizeof(Backend));
+ bn = (Backend *) calloc(1, sizeof(Backend));
if (!bn)
{
ereport(LOG,
@@ -4030,7 +4094,6 @@ BackendStartup(Port *port)
errmsg("out of memory")));
return STATUS_ERROR;
}
-
/*
* Compute the cancel key that will be assigned to this backend. The
* backend will have its own copy in the forked-off process' value of
@@ -4063,12 +4126,28 @@ BackendStartup(Port *port)
/* Hasn't asked to be notified about any bgworkers yet */
bn->bgworker_notify = false;
+ /* Create socket pair for sending session sockets to the backend */
+ if (!dedicated_backend)
+ {
+ if (socketpair(AF_UNIX, SOCK_DGRAM, 0, session_pipe) < 0)
+ ereport(FATAL,
+ (errcode_for_file_access(),
+ errmsg_internal("could not create socket pair for launching sessions: %m")));
+#ifdef WIN32
+ SessionPoolSock = session_pipe[0];
+#endif
+ }
#ifdef EXEC_BACKEND
pid = backend_forkexec(port);
#else /* !EXEC_BACKEND */
pid = fork_process();
if (pid == 0) /* child */
{
+ if (!dedicated_backend)
+ {
+ SessionPoolSock = session_pipe[0]; /* Use this socket for receiving client session socket descriptor */
+ close(session_pipe[1]); /* Close unused end of the pipe */
+ }
free(bn);
/* Detangle from postmaster */
@@ -4110,9 +4189,22 @@ BackendStartup(Port *port)
* of backends.
*/
bn->pid = pid;
+ bn->session_send_sock = PGINVALID_SOCKET;
bn->bkend_type = BACKEND_TYPE_NORMAL; /* Can change later to WALSND */
dlist_push_head(&BackendList, &bn->elem);
+ if (!dedicated_backend)
+ {
+ bn->session_send_sock = session_pipe[1]; /* Use this socket for sending client session socket descriptor */
+ closesocket(session_pipe[0]); /* Close unused end of the pipe */
+ if (pool->workers == NULL)
+ pool->workers = (Backend**)malloc(sizeof(Backend*)*SessionPoolSize);
+ bn->worker_id = pool->n_workers++;
+ pool->workers[bn->worker_id] = bn;
+ bn->session_pool_id = session_pool_id;
+ elog(DEBUG2, "Start %d-th worker in session pool %d pid %d",
+ pool->n_workers, session_pool_id, pid);
+ }
#ifdef EXEC_BACKEND
if (!bn->dead_end)
ShmemBackendArrayAdd(bn);
@@ -4299,7 +4391,7 @@ BackendInitialize(Port *port)
* Receive the startup packet (which might turn out to be a cancel request
* packet).
*/
- status = ProcessStartupPacket(port, false);
+ status = ProcessStartupPacket(port, false, TopMemoryContext);
/*
* Stop here if it was bad or a cancel packet. ProcessStartupPacket
@@ -6033,6 +6125,9 @@ save_backend_variables(BackendParameters *param, Port *port,
if (!write_inheritable_socket(¶m->portsocket, port->sock, childPid))
return false;
+ if (!write_inheritable_socket(¶m->sessionsocket, SessionPoolSock, childPid))
+ return false;
+
strlcpy(param->DataDir, DataDir, MAXPGPATH);
memcpy(¶m->ListenSocket, &ListenSocket, sizeof(ListenSocket));
@@ -6265,6 +6360,7 @@ restore_backend_variables(BackendParameters *param, Port *port)
{
memcpy(port, ¶m->port, sizeof(Port));
read_inheritable_socket(&port->sock, ¶m->portsocket);
+ read_inheritable_socket(&SessionPoolSock, ¶m->sessionsocket);
SetDataDir(param->DataDir);
diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/latch.c
index e6706f7..0f62792 100644
--- a/src/backend/storage/ipc/latch.c
+++ b/src/backend/storage/ipc/latch.c
@@ -76,6 +76,7 @@ struct WaitEventSet
{
int nevents; /* number of registered events */
int nevents_space; /* maximum number of events in this set */
+ int free_events; /* L1-list of free events linked by "pos" and terminated by -1*/
/*
* Array, of nevents_space length, storing the definition of events this
@@ -129,9 +130,9 @@ static void drainSelfPipe(void);
#if defined(WAIT_USE_EPOLL)
static void WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action);
#elif defined(WAIT_USE_POLL)
-static void WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event);
+static void WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event, bool remove);
#elif defined(WAIT_USE_WIN32)
-static void WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event);
+static void WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event, bool remove);
#endif
static inline int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
@@ -562,6 +563,7 @@ CreateWaitEventSet(MemoryContext context, int nevents)
set->latch = NULL;
set->nevents_space = nevents;
+ set->free_events = -1;
#if defined(WAIT_USE_EPOLL)
#ifdef EPOLL_CLOEXEC
@@ -667,9 +669,11 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
void *user_data)
{
WaitEvent *event;
+ int free_event;
/* not enough space */
- Assert(set->nevents < set->nevents_space);
+ if (set->nevents == set->nevents_space)
+ return -1;
if (latch)
{
@@ -690,8 +694,19 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
if (fd == PGINVALID_SOCKET && (events & WL_SOCKET_MASK))
elog(ERROR, "cannot wait on socket event without a socket");
- event = &set->events[set->nevents];
- event->pos = set->nevents++;
+ free_event = set->free_events;
+ if (free_event >= 0)
+ {
+ event = &set->events[free_event];
+ set->free_events = event->pos;
+ event->pos = free_event;
+ }
+ else
+ {
+ event = &set->events[set->nevents];
+ event->pos = set->nevents;
+ }
+ set->nevents += 1;
event->fd = fd;
event->events = events;
event->user_data = user_data;
@@ -718,15 +733,38 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
#if defined(WAIT_USE_EPOLL)
WaitEventAdjustEpoll(set, event, EPOLL_CTL_ADD);
#elif defined(WAIT_USE_POLL)
- WaitEventAdjustPoll(set, event);
+ WaitEventAdjustPoll(set, event, false);
#elif defined(WAIT_USE_WIN32)
- WaitEventAdjustWin32(set, event);
+ WaitEventAdjustWin32(set, event, false);
#endif
return event->pos;
}
/*
+ * Remove event with specified socket descriptor
+ */
+void DeleteWaitEventFromSet(WaitEventSet *set, pgsocket fd)
+{
+ int i, n = set->nevents;
+ for (i = 0; i < n; i++)
+ {
+ WaitEvent *event = &set->events[i];
+ if (event->fd == fd)
+ {
+#if defined(WAIT_USE_EPOLL)
+ WaitEventAdjustEpoll(set, event, EPOLL_CTL_DEL);
+#elif defined(WAIT_USE_POLL)
+ WaitEventAdjustPoll(set, event, true);
+#elif defined(WAIT_USE_WIN32)
+ WaitEventAdjustWin32(set, event, true);
+#endif
+ break;
+ }
+ }
+}
+
+/*
* Change the event mask and, in the WL_LATCH_SET case, the latch associated
* with the WaitEvent.
*
@@ -774,9 +812,9 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch)
#if defined(WAIT_USE_EPOLL)
WaitEventAdjustEpoll(set, event, EPOLL_CTL_MOD);
#elif defined(WAIT_USE_POLL)
- WaitEventAdjustPoll(set, event);
+ WaitEventAdjustPoll(set, event, false);
#elif defined(WAIT_USE_WIN32)
- WaitEventAdjustWin32(set, event);
+ WaitEventAdjustWin32(set, event, false);
#endif
}
@@ -827,14 +865,33 @@ WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action)
ereport(ERROR,
(errcode_for_socket_access(),
errmsg("epoll_ctl() failed: %m")));
+
+ if (action == EPOLL_CTL_DEL)
+ {
+ int pos = event->pos;
+ event->fd = PGINVALID_SOCKET;
+ set->nevents -= 1;
+ event->pos = set->free_events;
+ set->free_events = pos;
+ }
}
#endif
#if defined(WAIT_USE_POLL)
static void
-WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event)
+WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event, bool remove)
{
- struct pollfd *pollfd = &set->pollfds[event->pos];
+ int pos = event->pos;
+ struct pollfd *pollfd = &set->pollfds[pos];
+
+ if (remove)
+ {
+ set->nevents -= 1;
+ *pollfd = set->pollfds[set->nevents];
+ set->events[pos] = set->events[set->nevents];
+ event->pos = pos;
+ return;
+ }
pollfd->revents = 0;
pollfd->fd = event->fd;
@@ -865,9 +922,25 @@ WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event)
#if defined(WAIT_USE_WIN32)
static void
-WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event)
+WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event, bool remove)
{
- HANDLE *handle = &set->handles[event->pos + 1];
+ int pos = event->pos;
+ HANDLE *handle = &set->handles[pos + 1];
+
+ if (remove)
+ {
+ Assert(event->fd != PGINVALID_SOCKET);
+
+ if (*handle != WSA_INVALID_EVENT)
+ WSACloseEvent(*handle);
+
+ set->nevents -= 1;
+ set->events[pos] = set->events[set->nevents];
+ *handle = set->handles[set->nevents + 1];
+ set->handles[set->nevents + 1] = WSA_INVALID_EVENT;
+ event->pos = pos;
+ return;
+ }
if (event->events == WL_LATCH_SET)
{
@@ -880,7 +953,7 @@ WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event)
}
else
{
- int flags = FD_CLOSE; /* always check for errors/EOF */
+ int flags = FD_CLOSE; /* always check for errors/EOF */
if (event->events & WL_SOCKET_READABLE)
flags |= FD_READ;
@@ -897,8 +970,8 @@ WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event)
WSAGetLastError());
}
if (WSAEventSelect(event->fd, *handle, flags) != 0)
- elog(ERROR, "failed to set up event for socket: error code %u",
- WSAGetLastError());
+ elog(ERROR, "failed to set up event for socket %p: error code %u",
+ event->fd, WSAGetLastError());
Assert(event->fd != PGINVALID_SOCKET);
}
@@ -1296,7 +1369,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
{
if (cur_event->reset)
{
- WaitEventAdjustWin32(set, cur_event);
+ WaitEventAdjustWin32(set, cur_event, false);
cur_event->reset = false;
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index ddc3ec8..08e18e4 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -40,6 +40,7 @@
#include "access/printtup.h"
#include "access/xact.h"
#include "catalog/pg_type.h"
+#include "catalog/namespace.h"
#include "commands/async.h"
#include "commands/prepare.h"
#include "libpq/libpq.h"
@@ -75,9 +76,9 @@
#include "utils/snapmgr.h"
#include "utils/timeout.h"
#include "utils/timestamp.h"
+#include "utils/builtins.h"
#include "mb/pg_wchar.h"
-
/* ----------------
* global variables
* ----------------
@@ -98,6 +99,10 @@ int max_stack_depth = 100;
/* wait N seconds to allow attach from a debugger */
int PostAuthDelay = 0;
+/* Local socket for redirecting sessions to the backends */
+pgsocket SessionPoolSock = PGINVALID_SOCKET;
+/* Pointer to the active session */
+SessionContext* ActiveSession;
/* ----------------
@@ -169,6 +174,12 @@ static ProcSignalReason RecoveryConflictReason;
static MemoryContext row_description_context = NULL;
static StringInfoData row_description_buf;
+static WaitEventSet* SessionPool; /* Set of all sessions sockets */
+static int64 SessionCount; /* Number of sessions */
+static Port* BackendPort; /* Reference to the original port of this backend created when this backend was launched.
+ * Session using this port may be already terminated, but since it is allocated in TopMemoryContext,
+ * its content is still valid and is used as template for ports of new sessions */
+
/* ----------------------------------------------------------------
* decls for routines only used in this file
* ----------------------------------------------------------------
@@ -194,6 +205,27 @@ static void log_disconnections(int code, Datum arg);
static void enable_statement_timeout(void);
static void disable_statement_timeout(void);
+/*
+ * Generate session ID unique within this backend
+ */
+static char* CreateSessionId(void)
+{
+ char buf[64];
+ pg_lltoa(++SessionCount, buf);
+ return pstrdup(buf);
+}
+
+/*
+ * Free all memory associated with session and delete session object itself
+ */
+static void DeleteSession(SessionContext* session)
+{
+ elog(DEBUG1, "Delete session %p, id=%s, memory context=%p", session, session->id, session->memory);
+ RestoreSessionGUCs(session);
+ ReleaseSessionGUCs(session);
+ MemoryContextDelete(session->memory);
+ free(session);
+}
/* ----------------------------------------------------------------
* routines to obtain user input
@@ -1232,6 +1264,12 @@ exec_parse_message(const char *query_string, /* string to execute */
bool save_log_statement_stats = log_statement_stats;
char msec_str[32];
+ if (ActiveSession && stmt_name[0] != '\0')
+ {
+ /* Make names of prepared statements unique for session in case of using internal session pool */
+ stmt_name = psprintf("%s.%s", ActiveSession->id, stmt_name);
+ }
+
/*
* Report query to various monitoring facilities.
*/
@@ -1503,6 +1541,12 @@ exec_bind_message(StringInfo input_message)
portal_name = pq_getmsgstring(input_message);
stmt_name = pq_getmsgstring(input_message);
+ if (ActiveSession && stmt_name[0] != '\0')
+ {
+ /* Make names of prepared statements unique for session in case of using internal session pool */
+ stmt_name = psprintf("%s.%s", ActiveSession->id, stmt_name);
+ }
+
ereport(DEBUG2,
(errmsg("bind %s to %s",
*portal_name ? portal_name : "<unnamed>",
@@ -2325,6 +2369,12 @@ exec_describe_statement_message(const char *stmt_name)
CachedPlanSource *psrc;
int i;
+ if (ActiveSession && stmt_name[0] != '\0')
+ {
+ /* Make names of prepared statements unique for session in case of using internal session pool */
+ stmt_name = psprintf("%s.%s", ActiveSession->id, stmt_name);
+ }
+
/*
* Start up a transaction command. (Note that this will normally change
* current memory context.) Nothing happens if we are already in one.
@@ -3603,7 +3653,6 @@ process_postgres_switches(int argc, char *argv[], GucContext ctx,
#endif
}
-
/* ----------------------------------------------------------------
* PostgresMain
* postgres main loop -- all backends, interactive or otherwise start here
@@ -3654,6 +3703,21 @@ PostgresMain(int argc, char *argv[],
progname)));
}
+ /* Assign session for this backend in case of session pooling */
+ if (SessionPoolSize != 0)
+ {
+ MemoryContext oldcontext;
+ ActiveSession = (SessionContext*)calloc(1, sizeof(SessionContext));
+ ActiveSession->memory = AllocSetContextCreate(TopMemoryContext,
+ "SessionMemoryContext",
+ ALLOCSET_DEFAULT_SIZES);
+ oldcontext = MemoryContextSwitchTo(ActiveSession->memory);
+ ActiveSession->id = CreateSessionId();
+ ActiveSession->port = MyProcPort;
+ BackendPort = MyProcPort;
+ MemoryContextSwitchTo(oldcontext);
+ }
+
/* Acquire configuration parameters, unless inherited from postmaster */
if (!IsUnderPostmaster)
{
@@ -3783,7 +3847,7 @@ PostgresMain(int argc, char *argv[],
* ... else we'd need to copy the Port data first. Also, subsidiary data
* such as the username isn't lost either; see ProcessStartupPacket().
*/
- if (PostmasterContext)
+ if (PostmasterContext && SessionPoolSize == 0)
{
MemoryContextDelete(PostmasterContext);
PostmasterContext = NULL;
@@ -4069,6 +4133,152 @@ PostgresMain(int argc, char *argv[],
ReadyForQuery(whereToSendOutput);
send_ready_for_query = false;
+
+ /*
+ * Here we perform multiplexing of client sessions if session pooling is enabled.
+ * As far as we perform transaction level pooling, rescheduling is done only when we are not in transaction.
+ */
+ if (SessionPoolSock != PGINVALID_SOCKET && !IsTransactionState() && pq_available_bytes() == 0)
+ {
+ WaitEvent ready_client;
+ if (SessionPool == NULL)
+ {
+ /* Construct wait event set if not constructed yet */
+ SessionPool = CreateWaitEventSet(TopMemoryContext, MaxSessions+3);
+ /* Add event to detect postmaster death */
+ AddWaitEventToSet(SessionPool, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, NULL, ActiveSession);
+ /* Add event for backends latch */
+ AddWaitEventToSet(SessionPool, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, ActiveSession);
+ /* Add event for accepting new sessions */
+ AddWaitEventToSet(SessionPool, WL_SOCKET_READABLE, SessionPoolSock, NULL, ActiveSession);
+ /* Add event for current session */
+ AddWaitEventToSet(SessionPool, WL_SOCKET_READABLE, MyProcPort->sock, NULL, ActiveSession);
+ }
+ ChooseSession:
+ DoingCommandRead = true;
+ /* Select which client session is ready to send new query */
+ if (WaitEventSetWait(SessionPool, -1, &ready_client, 1, PG_WAIT_CLIENT) != 1)
+ {
+ /* TODO: do some error recovery here */
+ elog(FATAL, "Failed to poll client sessions");
+ }
+ CHECK_FOR_INTERRUPTS();
+ DoingCommandRead = false;
+
+ if (ready_client.events & WL_POSTMASTER_DEATH)
+ ereport(FATAL,
+ (errcode(ERRCODE_ADMIN_SHUTDOWN),
+ errmsg("terminating connection due to unexpected postmaster exit")));
+
+ if (ready_client.events & WL_LATCH_SET)
+ {
+ ResetLatch(MyLatch);
+ ProcessClientReadInterrupt(true);
+ goto ChooseSession;
+ }
+
+ if (ready_client.fd == SessionPoolSock)
+ {
+ /* Here we handle case of attaching new session */
+ int status;
+ SessionContext* session;
+ StringInfoData buf;
+ Port* port;
+ pgsocket sock;
+ MemoryContext oldcontext;
+
+ sock = pg_recv_sock(SessionPoolSock);
+ if (sock == PGINVALID_SOCKET)
+ elog(FATAL, "Failed to receive session socket: %m");
+
+ session = (SessionContext*)calloc(1, sizeof(SessionContext));
+ session->memory = AllocSetContextCreate(TopMemoryContext,
+ "SessionMemoryContext",
+ ALLOCSET_DEFAULT_SIZES);
+ oldcontext = MemoryContextSwitchTo(session->memory);
+ port = palloc(sizeof(Port));
+ memcpy(port, BackendPort, sizeof(Port));
+
+ /*
+ * Receive the startup packet (which might turn out to be a cancel request
+ * packet).
+ */
+ port->sock = sock;
+ session->port = port;
+ session->id = CreateSessionId();
+
+ MyProcPort = port;
+ status = ProcessStartupPacket(port, false, session->memory);
+ MemoryContextSwitchTo(oldcontext);
+
+ /*
+ * TODO: Currently we assume that all sessions are accessing the same database under the same user.
+ * Just report an error if it is not true
+ */
+ if (strcmp(port->database_name, MyProcPort->database_name) ||
+ strcmp(port->user_name, MyProcPort->user_name))
+ {
+ elog(FATAL, "Failed to open session (dbname=%s user=%s) in backend %d (dbname=%s user=%s)",
+ port->database_name, port->user_name,
+ MyProcPid, MyProcPort->database_name, MyProcPort->user_name);
+ }
+ else if (status == STATUS_OK)
+ {
+ if (AddWaitEventToSet(SessionPool, WL_SOCKET_READABLE, sock, NULL, session) < 0)
+ {
+ elog(WARNING, "Too much pooled sessions: %d", MaxSessions);
+ }
+ else
+ {
+ elog(DEBUG2, "Start new session %d in backend %d for database %s user %s",
+ (int)sock, MyProcPid, port->database_name, port->user_name);
+ RestoreSessionGUCs(ActiveSession);
+ ActiveSession = session;
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ PerformAuthentication(MyProcPort);
+ CommitTransactionCommand();
+
+ /*
+ * Send GUC options to the client
+ */
+ BeginReportingGUCOptions();
+
+ /*
+ * Send this backend's cancellation info to the frontend.
+ */
+ pq_beginmessage(&buf, 'K');
+ pq_sendint32(&buf, (int32) MyProcPid);
+ pq_sendint32(&buf, (int32) MyCancelKey);
+ pq_endmessage(&buf);
+
+ /* Need not flush since ReadyForQuery will do it. */
+ send_ready_for_query = true;
+ continue;
+ }
+ }
+ /* Error while processing of startup package
+ * Reject this session and return back to listening sockets
+ */
+ DeleteSession(session);
+ elog(LOG, "Session startup failed");
+ closesocket(sock);
+ goto ChooseSession;
+ }
+ else
+ {
+ SessionContext* newSession = (SessionContext*)ready_client.user_data;
+ if (ActiveSession != newSession)
+ {
+ elog(DEBUG2, "Switch to session %d in backend %d", ready_client.fd, MyProcPid);
+ RestoreSessionGUCs(ActiveSession);
+ ActiveSession = newSession;
+ RestoreSessionGUCs(ActiveSession);
+ MyProcPort = ActiveSession->port;
+ SetTempNamespaceState(ActiveSession->tempNamespace, ActiveSession->tempToastNamespace);
+ }
+ }
+ }
}
/*
@@ -4350,6 +4560,39 @@ PostgresMain(int argc, char *argv[],
* it will fail to be called during other backend-shutdown
* scenarios.
*/
+
+ if (SessionPool)
+ {
+ /* In case of session pooling close the session, but do not terminate the backend
+ * even if there are not more sessions in this backend.
+ * The reason for keeping backend alive is to prevent redundant process launches if
+ * some client repeatedly open/close connection to the database.
+ * Maximal number of launched backends in case of connection pooling is intended to be
+ * optimal for this system and workload, so there are no reasons to try to reduce this number
+ * when there are no active sessions.
+ */
+ DeleteWaitEventFromSet(SessionPool, MyProcPort->sock);
+ elog(DEBUG1, "Close session %d in backend %d", MyProcPort->sock, MyProcPid);
+
+ pq_getmsgend(&input_message);
+ if (pq_is_reading_msg())
+ pq_endmsgread();
+
+ closesocket(MyProcPort->sock);
+ MyProcPort->sock = PGINVALID_SOCKET;
+ MyProcPort = NULL;
+
+ if (ActiveSession)
+ {
+ DropSessionPreparedStatements(ActiveSession->id);
+ DeleteSession(ActiveSession);
+ ActiveSession = NULL;
+ }
+ whereToSendOutput = DestRemote;
+ /* Need to perform rescheduling to some other session or accept new session */
+ goto ChooseSession;
+ }
+ elog(DEBUG1, "Terminate backend %d", MyProcPid);
proc_exit(0);
case 'd': /* copy data */
diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c
index 54fa4a3..14fd972 100644
--- a/src/backend/utils/init/globals.c
+++ b/src/backend/utils/init/globals.c
@@ -120,7 +120,10 @@ int maintenance_work_mem = 16384;
* register background workers.
*/
int NBuffers = 1000;
+int SessionPoolSize = 0;
+int SessionPoolPorts = 0;
int MaxConnections = 90;
+int MaxSessions = 1000;
int max_worker_processes = 8;
int max_parallel_workers = 8;
int MaxBackends = 0;
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index f9b3309..571c80f 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -65,7 +65,7 @@
static HeapTuple GetDatabaseTuple(const char *dbname);
static HeapTuple GetDatabaseTupleByOid(Oid dboid);
-static void PerformAuthentication(Port *port);
+void PerformAuthentication(Port *port);
static void CheckMyDatabase(const char *name, bool am_superuser);
static void InitCommunication(void);
static void ShutdownPostgres(int code, Datum arg);
@@ -180,7 +180,7 @@ GetDatabaseTupleByOid(Oid dboid)
*
* returns: nothing. Will not return at all if there's any failure.
*/
-static void
+void
PerformAuthentication(Port *port)
{
/* This should be set already, but let's make sure */
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 72f6be3..f82c2cb 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -1871,6 +1871,44 @@ static struct config_int ConfigureNamesInt[] =
},
{
+ {"max_sessions", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
+ gettext_noop("Sets the maximum number of client session."),
+ gettext_noop("Maximal number of client sessions which can be handled by one backend if session pooling is switched on. "
+ "So maximal number of client connections is session_pool_size*max_sessions")
+ },
+ &MaxSessions,
+ 1000, 1, INT_MAX,
+ NULL, NULL, NULL
+ },
+
+ {
+ {"session_pool_size", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
+ gettext_noop("Sets number of backends serving client sessions."),
+ gettext_noop("If non-zero then session pooling will be used: "
+ "client connections will be redirected to one of the backends and maximal number of backends is determined by this parameter."
+ "Launched backend are never terminated even in case of no active sessions.")
+ },
+ &SessionPoolSize,
+ 0, 0, INT_MAX,
+ NULL, NULL, NULL
+ },
+
+ {
+ {"session_pool_ports", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
+ gettext_noop("Number of session ports = number of session pools."),
+ gettext_noop("Number of extra parts which PostgreSQL will listen to accept client session. Each such port has separate session pool."
+ "It is intended that each port corresponds to some particular database/user combination, so that all backends in this session "
+ "pool will handle connection accessing this database. If session_pool_port is non zero then postmaster will always spawn dedicated (non-pooling) "
+ " backends at the main Postgres port. If session_pool_port is zero and session_pool_size is not zero, then sessions (pooled connection) will be also "
+ "accepted at main port. Session pool ports are allocatged sequentially: if Postgres main port is 5432 and session_pool_ports is 2, "
+ "then ports 5433 and 5434 will be used for connection pooling.")
+ },
+ &SessionPoolPorts,
+ 0, 0, MAX_SESSION_PORTS,
+ NULL, NULL, NULL
+ },
+
+ {
{"superuser_reserved_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
gettext_noop("Sets the number of connection slots reserved for superusers."),
NULL
@@ -5104,6 +5142,95 @@ NewGUCNestLevel(void)
}
/*
+ * Set GUCs for this session
+ */
+void
+RestoreSessionGUCs(SessionContext* session)
+{
+ SessionGUC* sg;
+ if (session == NULL)
+ return;
+ for (sg = session->gucs; sg != NULL; sg = sg->next)
+ {
+ void* old_extra = sg->var->extra;
+ sg->var->extra = sg->val.extra;
+ switch (sg->var->vartype)
+ {
+ case PGC_BOOL:
+ {
+ struct config_bool *conf = (struct config_bool*)sg->var;
+ bool oldval = *conf->variable;
+ *conf->variable = sg->val.val.boolval;
+ if (conf->assign_hook)
+ conf->assign_hook(sg->val.val.boolval, sg->val.extra);
+ sg->val.val.boolval = oldval;
+ break;
+ }
+ case PGC_INT:
+ {
+ struct config_int *conf = (struct config_int*)sg->var;
+ int oldval = *conf->variable;
+ *conf->variable = sg->val.val.intval;
+ if (conf->assign_hook)
+ conf->assign_hook(sg->val.val.intval, sg->val.extra);
+ sg->val.val.intval = oldval;
+ break;
+ }
+ case PGC_REAL:
+ {
+ struct config_real *conf = (struct config_real*)sg->var;
+ double oldval = *conf->variable;
+ *conf->variable = sg->val.val.realval;
+ if (conf->assign_hook)
+ conf->assign_hook(sg->val.val.realval, sg->val.extra);
+ sg->val.val.realval = oldval;
+ break;
+ }
+ case PGC_STRING:
+ {
+ struct config_string *conf = (struct config_string*)sg->var;
+ char* oldval = *conf->variable;
+ *conf->variable = sg->val.val.stringval;
+ if (conf->assign_hook)
+ conf->assign_hook(sg->val.val.stringval, sg->val.extra);
+ sg->val.val.stringval = oldval;
+ break;
+ }
+ case PGC_ENUM:
+ {
+ struct config_enum *conf = (struct config_enum*)sg->var;
+ int oldval = *conf->variable;
+ *conf->variable = sg->val.val.enumval;
+ if (conf->assign_hook)
+ conf->assign_hook(sg->val.val.enumval, sg->val.extra);
+ sg->val.val.enumval = oldval;
+ break;
+ }
+ }
+ sg->val.extra = old_extra;
+ }
+}
+
+/*
+ * Deallocate memory for session GUCs
+ */
+void
+ReleaseSessionGUCs(SessionContext* session)
+{
+ SessionGUC* sg;
+ for (sg = session->gucs; sg != NULL; sg = sg->next)
+ {
+ if (sg->val.extra)
+ set_extra_field(sg->var, &sg->val.extra, NULL);
+ if (sg->var->vartype == PGC_STRING)
+ {
+ struct config_string* conf = (struct config_string*)sg->var;
+ set_string_field(conf, &sg->val.val.stringval, NULL);
+ }
+ }
+}
+
+/*
* Do GUC processing at transaction or subtransaction commit or abort, or
* when exiting a function that has proconfig settings, or when undoing a
* transient assignment to some GUC variables. (The name is thus a bit of
@@ -5172,7 +5299,42 @@ AtEOXact_GUC(bool isCommit, int nestLevel)
else if (stack->state == GUC_SET)
{
/* we keep the current active value */
- discard_stack_value(gconf, &stack->prior);
+ if (ActiveSession)
+ {
+ SessionGUC* sg;
+ for (sg = ActiveSession->gucs; sg != NULL && sg->var != gconf; sg = sg->next);
+ if (sg == NULL)
+ {
+ sg = MemoryContextAllocZero(ActiveSession->memory,
+ sizeof(SessionGUC));
+ sg->var = gconf;
+ sg->next = ActiveSession->gucs;
+ ActiveSession->gucs = sg;
+ }
+ switch (gconf->vartype)
+ {
+ case PGC_BOOL:
+ sg->val.val.boolval = stack->prior.val.boolval;
+ break;
+ case PGC_INT:
+ sg->val.val.intval = stack->prior.val.intval;
+ break;
+ case PGC_REAL:
+ sg->val.val.realval = stack->prior.val.realval;
+ break;
+ case PGC_STRING:
+ sg->val.val.stringval = stack->prior.val.stringval;
+ break;
+ case PGC_ENUM:
+ sg->val.val.enumval = stack->prior.val.enumval;
+ break;
+ }
+ sg->val.extra = stack->prior.extra;
+ }
+ else
+ {
+ discard_stack_value(gconf, &stack->prior);
+ }
}
else /* must be GUC_LOCAL */
restorePrior = true;
@@ -5197,8 +5359,8 @@ AtEOXact_GUC(bool isCommit, int nestLevel)
case GUC_SET:
/* next level always becomes SET */
- discard_stack_value(gconf, &stack->prior);
- if (prev->state == GUC_SET_LOCAL)
+ discard_stack_value(gconf, &stack->prior);
+ if (prev->state == GUC_SET_LOCAL)
discard_stack_value(gconf, &prev->masked);
prev->state = GUC_SET;
break;
diff --git a/src/include/commands/prepare.h b/src/include/commands/prepare.h
index ffec029..cb5f8d4 100644
--- a/src/include/commands/prepare.h
+++ b/src/include/commands/prepare.h
@@ -56,5 +56,6 @@ extern TupleDesc FetchPreparedStatementResultDesc(PreparedStatement *stmt);
extern List *FetchPreparedStatementTargetList(PreparedStatement *stmt);
extern void DropAllPreparedStatements(void);
+extern void DropSessionPreparedStatements(char const* sessionId);
#endif /* PREPARE_H */
diff --git a/src/include/libpq/libpq.h b/src/include/libpq/libpq.h
index 2e7725d..9169b21 100644
--- a/src/include/libpq/libpq.h
+++ b/src/include/libpq/libpq.h
@@ -71,6 +71,7 @@ extern int pq_getbyte(void);
extern int pq_peekbyte(void);
extern int pq_getbyte_if_available(unsigned char *c);
extern int pq_putbytes(const char *s, size_t len);
+extern int pq_available_bytes(void);
/*
* prototypes for functions in be-secure.c
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index 54ee273..66d7e33 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -157,6 +157,9 @@ extern PGDLLIMPORT char *DataDir;
extern PGDLLIMPORT int NBuffers;
extern PGDLLIMPORT int MaxBackends;
extern PGDLLIMPORT int MaxConnections;
+extern PGDLLIMPORT int MaxSessions;
+extern PGDLLIMPORT int SessionPoolSize;
+extern PGDLLIMPORT int SessionPoolPorts;
extern PGDLLIMPORT int max_worker_processes;
extern int max_parallel_workers;
@@ -175,6 +178,8 @@ extern char pkglib_path[];
extern char postgres_exec_path[];
#endif
+#define MAX_SESSION_PORTS 8
+
/*
* done in storage/backendid.h for now.
*
@@ -420,6 +425,7 @@ extern void InitializeMaxBackends(void);
extern void InitPostgres(const char *in_dbname, Oid dboid, const char *username,
Oid useroid, char *out_dbname);
extern void BaseInit(void);
+extern void PerformAuthentication(struct Port *port);
/* in utils/init/miscinit.c */
extern bool IgnoreSystemIndexes;
diff --git a/src/include/port.h b/src/include/port.h
index 3e528fa..8a0ac98 100644
--- a/src/include/port.h
+++ b/src/include/port.h
@@ -41,6 +41,10 @@ typedef SOCKET pgsocket;
extern bool pg_set_noblock(pgsocket sock);
extern bool pg_set_block(pgsocket sock);
+/* send/receive socket descriptor */
+extern int pg_send_sock(pgsocket chan, pgsocket sock, pid_t pid);
+extern pgsocket pg_recv_sock(pgsocket chan);
+
/* Portable path handling for Unix/Win32 (in path.c) */
extern bool has_drive_prefix(const char *filename);
diff --git a/src/include/port/win32_port.h b/src/include/port/win32_port.h
index d31c28f..e667434 100644
--- a/src/include/port/win32_port.h
+++ b/src/include/port/win32_port.h
@@ -447,6 +447,7 @@ extern int pgkill(int pid, int sig);
#define select(n, r, w, e, timeout) pgwin32_select(n, r, w, e, timeout)
#define recv(s, buf, len, flags) pgwin32_recv(s, buf, len, flags)
#define send(s, buf, len, flags) pgwin32_send(s, buf, len, flags)
+#define socketpair(af, type, protocol, socks) pgwin32_socketpair(af, type, protocol, socks)
SOCKET pgwin32_socket(int af, int type, int protocol);
int pgwin32_bind(SOCKET s, struct sockaddr *addr, int addrlen);
@@ -456,6 +457,7 @@ int pgwin32_connect(SOCKET s, const struct sockaddr *name, int namelen);
int pgwin32_select(int nfds, fd_set *readfs, fd_set *writefds, fd_set *exceptfds, const struct timeval *timeout);
int pgwin32_recv(SOCKET s, char *buf, int len, int flags);
int pgwin32_send(SOCKET s, const void *buf, int len, int flags);
+int pgwin32_socketpair(int domain, int type, int protocol, SOCKET socks[2]);
const char *pgwin32_socket_strerror(int err);
int pgwin32_waitforsinglesocket(SOCKET s, int what, int timeout);
diff --git a/src/include/postmaster/postmaster.h b/src/include/postmaster/postmaster.h
index 1877eef..c9527c9 100644
--- a/src/include/postmaster/postmaster.h
+++ b/src/include/postmaster/postmaster.h
@@ -62,6 +62,9 @@ extern Size ShmemBackendArraySize(void);
extern void ShmemBackendArrayAllocation(void);
#endif
+struct Port;
+extern int ProcessStartupPacket(struct Port *port, bool SSLdone, MemoryContext memctx);
+
/*
* Note: MAX_BACKENDS is limited to 2^18-1 because that's the width reserved
* for buffer references in buf_internals.h. This limitation could be lifted
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
index a4bcb48..10f30d1 100644
--- a/src/include/storage/latch.h
+++ b/src/include/storage/latch.h
@@ -176,6 +176,8 @@ extern int WaitLatch(volatile Latch *latch, int wakeEvents, long timeout,
extern int WaitLatchOrSocket(volatile Latch *latch, int wakeEvents,
pgsocket sock, long timeout, uint32 wait_event_info);
+extern void DeleteWaitEventFromSet(WaitEventSet *set, pgsocket fd);
+
/*
* Unix implementation uses SIGUSR1 for inter-process signaling.
* Win32 doesn't need this.
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 5c19a61..11eded3 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -21,6 +21,7 @@
#include "storage/lock.h"
#include "storage/pg_sema.h"
#include "storage/proclist_types.h"
+#include "utils/guc_tables.h"
/*
* Each backend advertises up to PGPROC_MAX_CACHED_SUBXIDS TransactionIds
@@ -273,6 +274,29 @@ extern PGDLLIMPORT PROC_HDR *ProcGlobal;
extern PGPROC *PreparedXactProcs;
+typedef struct SessionGUC
+{
+ struct SessionGUC* next;
+ config_var_value val;
+ struct config_generic *var;
+} SessionGUC;
+
+/*
+ * Information associated with client session
+ */
+typedef struct SessionContext
+{
+ MemoryContext memory; /* memory context used for global session data (replacement of TopMemoryContext) */
+ struct Port* port; /* connection port */
+ char* id; /* session identifier used to construct unique prepared statement names */
+ Oid tempNamespace; /* temporary namespace */
+ Oid tempToastNamespace; /* temporary toast namespace */
+ SessionGUC* gucs;
+} SessionContext;
+
+
+extern PGDLLIMPORT SessionContext *ActiveSession;
+
/* Accessor for PGPROC given a pgprocno. */
#define GetPGProcByNumber(n) (&ProcGlobal->allProcs[(n)])
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 63b4e48..191eeaa 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -34,6 +34,7 @@ extern CommandDest whereToSendOutput;
extern PGDLLIMPORT const char *debug_query_string;
extern int max_stack_depth;
extern int PostAuthDelay;
+extern pgsocket SessionPoolSock;
/* GUC-configurable parameters */
diff --git a/src/include/utils/guc.h b/src/include/utils/guc.h
index 77daa5a..86e89e8 100644
--- a/src/include/utils/guc.h
+++ b/src/include/utils/guc.h
@@ -394,6 +394,12 @@ extern Size EstimateGUCStateSpace(void);
extern void SerializeGUCState(Size maxsize, char *start_address);
extern void RestoreGUCState(void *gucstate);
+/* Session polling support function */
+struct SessionContext;
+extern void RestoreSessionGUCs(struct SessionContext* session);
+extern void ReleaseSessionGUCs(struct SessionContext* session);
+
+
/* Support for messages reported from GUC check hooks */
extern PGDLLIMPORT char *GUC_check_errmsg_string;
