Port recent changes from EE

Author: knizhnik

src/backend/port/send_sock.c

@@ -82,8 +82,11 @@ pg_send_sock(pgsocket chan, pgsocket sock, pid_t pid)
     memcpy(CMSG_DATA(cmsg), &sock, sizeof(sock));
     msg.msg_controllen = cmsg->cmsg_len;
 
-    if (sendmsg(chan, &msg, 0) < 0)
-		return PGINVALID_SOCKET;
+    while (sendmsg(chan, &msg, 0) < 0)
+	{
+		if (errno != EINTR)
+			return PGINVALID_SOCKET;
+	}
 
 	return 0;
 #endif
@@ -142,8 +145,11 @@ pg_recv_sock(pgsocket chan)
     msg.msg_control = c_buffer;
     msg.msg_controllen = sizeof(c_buffer);
 
-    if (recvmsg(chan, &msg, 0) < 0)
-		return PGINVALID_SOCKET;
+    while (recvmsg(chan, &msg, 0) < 0)
+	{
+		if (errno != EINTR)
+			return PGINVALID_SOCKET;
+	}
 
     cmsg = CMSG_FIRSTHDR(&msg);
 	if (!cmsg)

src/backend/storage/file/fd.c

@@ -116,7 +116,7 @@
  * the number of open files.  (This appears to be true on most if not
  * all platforms as of Feb 2004.)
  */
-#define NUM_RESERVED_FDS		10
+#define NUM_RESERVED_FDS		20
 
 /*
  * If we have fewer than this many usable FDs after allowing for the reserved
@@ -276,7 +276,6 @@ static int	nextTempTableSpace = 0;
  * Insert		   - put a file at the front of the Lru ring
  * LruInsert	   - put a file at the front of the Lru ring and open it
  * ReleaseLruFile  - Release an fd by closing the last entry in the Lru ring
- * ReleaseLruFiles - Release fd(s) until we're under the max_safe_fds limit
  * AllocateVfd	   - grab a free (or new) file record (from VfdArray)
  * FreeVfd		   - free a file record
  *
@@ -304,7 +303,6 @@ static void LruDelete(File file);
 static void Insert(File file);
 static int	LruInsert(File file);
 static bool ReleaseLruFile(void);
-static void ReleaseLruFiles(void);
 static File AllocateVfd(void);
 static void FreeVfd(File file);
 
@@ -1176,7 +1174,7 @@ ReleaseLruFile(void)
  * Release kernel FDs as needed to get under the max_safe_fds limit.
  * After calling this, it's OK to try to open another file.
  */
-static void
+void
 ReleaseLruFiles(void)
 {
 	while (nfile + numAllocatedDescs >= max_safe_fds)

src/backend/storage/ipc/ipc.c

@@ -304,6 +304,13 @@ atexit_callback(void)
 void
 on_proc_exit(pg_on_exit_callback function, Datum arg)
 {
+	int i = on_proc_exit_index;
+
+	while (--i >= 0)
+	{
+		if (on_proc_exit_list[i].function == function && on_proc_exit_list[i].arg == arg)
+			return;
+	}
 	if (on_proc_exit_index >= MAX_ON_EXITS)
 		ereport(FATAL,
 				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),

src/backend/tcop/postgres.c

@@ -3675,10 +3675,13 @@ process_postgres_switches(int argc, char *argv[], GucContext ctx,
 #endif
 }
 
-#define ACTIVE_SESSION_MAGIC  0xDEFA1234U
-#define REMOVED_SESSION_MAGIC 0xDEADDEEDU
+#define ACTIVE_SESSION_MAGIC    0xDEFA1234U
+#define REMOVED_SESSION_MAGIC   0xDEADDEEDU
+#define MIN_FREE_FDS            10
+#define DESCRIPTORS_PER_SESSION 2
 
 static int nActiveSessions = 0;
+static int maxActiveSessions = 0;
 
 static SessionContext *
 CreateSession(void)
@@ -3694,6 +3697,23 @@ CreateSession(void)
 	session->magic = ACTIVE_SESSION_MAGIC;
 	session->eventPos = -1;
 	nActiveSessions += 1;
+	if (nActiveSessions > maxActiveSessions)
+	{
+		int new_max_safe_fds = max_safe_fds - (nActiveSessions - maxActiveSessions)*DESCRIPTORS_PER_SESSION;
+		if (new_max_safe_fds >= MIN_FREE_FDS)
+		{
+			max_safe_fds = new_max_safe_fds;
+			/* Ensure that we have enough free descriptors to establish new session.
+			 * Unlike fd.c, which throws away lest recently used file descriptors
+			 * only when open() call is failed, we prefer more conservative (pessimistic)
+			 * aporach here.
+			 */
+			ReleaseLruFiles();
+		}
+		else
+			elog(WARNING, "Too few free file desriptors %d for %d sessions", new_max_safe_fds, maxActiveSessions);
+		maxActiveSessions = nActiveSessions;
+	}
 	return session;
 }
 
@@ -3723,15 +3743,15 @@ SwitchToSession(SessionContext *session)
 	RestoreSessionGUCs(ActiveSession);
 	ActiveSession = session;
 
-	MyProcPort = ActiveSession->port;
-	SetTempNamespaceState(ActiveSession->tempNamespace,
-						  ActiveSession->tempToastNamespace);
+	MyProcPort = session->port;
+	SetTempNamespaceState(session->tempNamespace,
+						  session->tempToastNamespace);
 	pq_set_current_state(session->port->pqcomm_state, session->port,
 						 session->eventSet);
 	whereToSendOutput = DestRemote;
 
-	RestoreSessionGUCs(ActiveSession);
-	LoadSessionVariables(ActiveSession);
+	RestoreSessionGUCs(session);
+	LoadSessionVariables(session);
 }
 
 static void
@@ -4369,11 +4389,12 @@ PostgresMain(int argc, char *argv[],
 					pgsocket sock;
 					MemoryContext oldcontext;
 
+					session = CreateSession();
+
 					sock = pg_recv_sock(SessionPoolSock);
 					if (sock == PGINVALID_SOCKET)
 						elog(ERROR, "Failed to receive session socket: %m");
 
-					session = CreateSession();
 
 					/* Initialize port and wait event set for this session */
 					oldcontext = MemoryContextSwitchTo(session->memory);
@@ -4436,11 +4457,10 @@ PostgresMain(int argc, char *argv[],
 					pq_sendint(&buf, (int32) MyProcPid, 4);
 					pq_sendint(&buf, (int32) MyCancelKey, 4);
 					pq_endmessage(&buf);
-
 					/* Need not flush since ReadyForQuery will do it. */
-					send_ready_for_query = true;
 
-					continue;
+					ReadyForQuery(whereToSendOutput);
+					goto ChooseSession;
 				}
 				else
 				{

src/include/storage/fd.h

@@ -138,6 +138,7 @@ extern int	durable_rename(const char *oldfile, const char *newfile, int loglevel
 extern int	durable_unlink(const char *fname, int loglevel);
 extern int	durable_link_or_rename(const char *oldfile, const char *newfile, int loglevel);
 extern void SyncDataDirectory(void);
+extern void ReleaseLruFiles(void);
 
 /* Filename components */
 #define PG_TEMP_FILES_DIR "pgsql_tmp"