Merge branch 'conn_pool' of https://github.com/postgrespro/postgresql.builtin_pool into conn_pool

Author: knizhnik

src/backend/catalog/pg_db_role_setting.c

@@ -16,6 +16,7 @@
 #include "catalog/indexing.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_db_role_setting.h"
+#include "storage/proc.h"
 #include "utils/fmgroids.h"
 #include "utils/rel.h"
 #include "utils/tqual.h"
@@ -33,6 +34,7 @@ AlterSetting(Oid databaseid, Oid roleid, VariableSetStmt *setstmt)
 
 	/* Get the old tuple, if any. */
 
+	IsDedicatedBackend = true;
 	rel = heap_open(DbRoleSettingRelationId, RowExclusiveLock);
 	ScanKeyInit(&scankey[0],
 				Anum_pg_db_role_setting_setdatabase,

src/backend/commands/copy.c

@@ -2412,7 +2412,7 @@ CopyFrom(CopyState cstate)
 		 * registers the snapshot it uses.
 		 */
 		InvalidateCatalogSnapshot();
-		if (!ThereAreNoPriorRegisteredSnapshots() || !ThereAreNoReadyPortals())
+		if (!ThereAreNoPriorRegisteredSnapshots() || (SessionPoolSize == 0 && !ThereAreNoReadyPortals()))
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
 					 errmsg("cannot perform FREEZE because of prior transaction activity")));

src/backend/postmaster/postmaster.c

@@ -178,9 +178,9 @@ typedef struct bkend
 	 */
 	int			bkend_type;
 	bool		dead_end;		/* is it going to send an error and quit? */
-	bool		bgworker_notify;	/* gets bgworker start/stop notifications */
+	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         session_pool_id;/* identifier of backends session pool */
 	int         worker_id;      /* identifier of worker within session pool */
 } Backend;
 
@@ -3273,11 +3273,18 @@ static void UnlinkBackend(Backend* bp)
 		if (--pool->n_workers != 0)
 		{
 			pool->workers[bp->worker_id] = pool->workers[pool->n_workers];
+			pool->workers[bp->worker_id]->worker_id = bp->worker_id;
 			pool->rr_index %= pool->n_workers;
 		}
 		closesocket(bp->session_send_sock);
+		bp->session_send_sock = PGINVALID_SOCKET;
 		elog(DEBUG2, "Cleanup backend %d", bp->pid);
 	}
+}
+
+static void DeleteBackend(Backend* bp)
+{
+	UnlinkBackend(bp);
 	dlist_delete(&bp->elem);
 	free(bp);
 }
@@ -3359,7 +3366,7 @@ CleanupBackend(int pid,
 				 */
 				BackgroundWorkerStopNotifications(bp->pid);
 			}
-			UnlinkBackend(bp);
+			DeleteBackend(bp);
 			break;
 		}
 	}
@@ -3461,7 +3468,7 @@ HandleChildCrash(int pid, int exitstatus, const char *procname)
 				ShmemBackendArrayRemove(bp);
 #endif
 			}
-			UnlinkBackend(bp);
+			DeleteBackend(bp);
 			/* Keep looping so we can signal remaining backends */
 		}
 		else
@@ -4066,19 +4073,25 @@ BackendStartup(Port *port, int session_pool_id)
 	PostmasterSessionPool* pool = &SessionPools[session_pool_id];
 	bool dedicated_backend = SessionPoolSize == 0 || (SessionPoolPorts != 0 && session_pool_id == 0);
 
-	if (!dedicated_backend && pool->n_workers >= SessionPoolSize)
+	/* In case of session pooling instead of spawning new backend open new session at one of the existed backends. */
+	while (!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);
+		pool->rr_index = (pool->rr_index + 1) % pool->n_workers; /* round-robin */
+
 		/* 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 */
+		{
+			elog(LOG, "Failed to send session socket: %m");
+			UnlinkBackend(worker);
+		}
+		else
+		{
+			elog(DEBUG2, "Start new session in pool %d for socket %d at backend %d",
+				 session_pool_id, port->sock, worker->pid);
 
-		return STATUS_OK;
+			return STATUS_OK;
+		}
 	}
 
 

src/backend/tcop/postgres.c

@@ -77,6 +77,7 @@
 #include "utils/timeout.h"
 #include "utils/timestamp.h"
 #include "utils/builtins.h"
+#include "utils/varlena.h"
 #include "mb/pg_wchar.h"
 
 /* ----------------
@@ -103,7 +104,7 @@ int			PostAuthDelay = 0;
 pgsocket    SessionPoolSock = PGINVALID_SOCKET;
 /* Pointer to the active session */
 SessionContext* ActiveSession;
-
+bool        IsDedicatedBackend;
 
 /* ----------------
  *		private variables
@@ -179,7 +180,7 @@ 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 */
-
+static int             nPooledSessions;/* Number of pooled sessions in this backend */
 static bool            IdleInTransactionSessionError;
 
 /* ----------------------------------------------------------------
@@ -214,6 +215,7 @@ static char* CreateSessionId(void)
 {
 	char buf[64];
 	pg_lltoa(++SessionCount, buf);
+	nPooledSessions += 1;
 	return pstrdup(buf);
 }
 
@@ -223,6 +225,7 @@ static char* CreateSessionId(void)
 static void DeleteSession(SessionContext* session)
 {
 	elog(DEBUG1, "Delete session %p, id=%s,  memory context=%p", session, session->id, session->memory);
+	nPooledSessions -= 1;
 	FreeWaitEventSet(session->eventSet);
 	RestoreSessionGUCs(session);
 	ReleaseSessionGUCs(session);
@@ -2960,7 +2963,7 @@ ProcessInterrupts(void)
 		LockErrorCleanup();
 		/* don't send to client, we already know the connection to be dead. */
 		whereToSendOutput = DestNone;
-		ereport(FATAL,
+		ereport(ActiveSession ? ERROR : FATAL,
 				(errcode(ERRCODE_CONNECTION_FAILURE),
 				 errmsg("connection to client lost")));
 	}
@@ -3646,6 +3649,35 @@ process_postgres_switches(int argc, char *argv[], GucContext ctx,
 #endif
 }
 
+static bool
+IsDedicatedDatabase(char const* dbname)
+{
+	List       *namelist;
+	ListCell   *l;
+	char       *databases;
+	bool       found = false;
+
+    /* Need a modifiable copy of namespace_search_path string */
+	databases = pstrdup(DedicatedDatabases);
+
+	if (!SplitIdentifierString(databases, ',', &namelist)) {
+		elog(ERROR, "invalid list syntax");
+	}
+	foreach(l, namelist)
+	{
+		char *curname = (char *) lfirst(l);
+		if (strcmp(curname, dbname) == 0)
+		{
+			found = true;
+			break;
+		}
+	}
+	list_free(namelist);
+	pfree(databases);
+
+	return found;
+}
+
 /* ----------------------------------------------------------------
  * PostgresMain
  *	   postgres main loop -- all backends, interactive or otherwise start here
@@ -3696,6 +3728,14 @@ PostgresMain(int argc, char *argv[],
 							progname)));
 	}
 
+	/* Serve all conections to "postgres" database by dedicated backends */
+	if (SessionPoolSize != 0 && IsDedicatedDatabase(dbname))
+	{
+		elog(LOG, "Backend is dedicated");
+		SessionPoolSize = 0;
+		closesocket(SessionPoolSock);
+		SessionPoolSock = PGINVALID_SOCKET;
+	}
 	/* Assign session for this backend in case of session pooling */
 	if (SessionPoolSize != 0)
 	{
@@ -3978,7 +4018,8 @@ PostgresMain(int argc, char *argv[],
 		pq_comm_reset();
 
 		/* Report the error to the client and/or server log */
-		EmitErrorReport();
+		if (MyProcPort)
+			EmitErrorReport();
 
 		/*
 		 * Make sure debug_query_string gets reset before we possibly clobber
@@ -4223,12 +4264,12 @@ PostgresMain(int argc, char *argv[],
 					 * 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))
+					if (strcmp(port->database_name, BackendPort->database_name) ||
+						strcmp(port->user_name, BackendPort->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);
+							 MyProcPid, BackendPort->database_name, BackendPort->user_name);
 					}
 					else if (status == STATUS_OK)
 					{
@@ -4246,6 +4287,7 @@ PostgresMain(int argc, char *argv[],
 							StartTransactionCommand();
 							PerformAuthentication(MyProcPort);
 							CommitTransactionCommand();
+							SetTempNamespaceState(InvalidOid, InvalidOid);
 
 							/*
 							 * Send GUC options to the client
@@ -4330,6 +4372,8 @@ PostgresMain(int argc, char *argv[],
 		 */
 		if (ConfigReloadPending)
 		{
+			if (ActiveSession && RestartPoolerOnReload)
+				proc_exit(0);
 			ConfigReloadPending = false;
 			ProcessConfigFile(PGC_SIGHUP);
 		}
@@ -4570,7 +4614,7 @@ PostgresMain(int argc, char *argv[],
 				 * scenarios.
 				 */
 
-				if (SessionPool)
+				if (SessionPool && (!IsDedicatedBackend || SessionCount > 1))
 				{
 				  CloseSession:
 					/* In case of session pooling close the session, but do not terminate the backend

src/backend/utils/cache/relcache.c

@@ -74,6 +74,7 @@
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rowsecurity.h"
 #include "storage/lmgr.h"
+#include "storage/proc.h"
 #include "storage/smgr.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
@@ -2059,6 +2060,13 @@ RelationIdGetRelation(Oid relationId)
 				RelationClearRelation(rd, true);
 			Assert(rd->rd_isvalid);
 		}
+		/*
+		 * In case of session pooling, relation descriptor can be constructed by some other session,
+		 * so we need to recheck rd_islocaltemp value
+		 */
+		if (ActiveSession && RELATION_IS_OTHER_TEMP(rd) && isTempOrTempToastNamespace(rd->rd_rel->relnamespace))
+			rd->rd_islocaltemp = true;
+
 		return rd;
 	}
 

src/backend/utils/init/globals.c

@@ -140,3 +140,6 @@ int			VacuumPageDirty = 0;
 
 int			VacuumCostBalance = 0;	/* working state for vacuum */
 bool		VacuumCostActive = false;
+
+bool        RestartPoolerOnReload = false;
+char       *DedicatedDatabases;