[PGPRO-4074] DeleteWaitEvent() for event sets.

Cherry-pick of c82e0d53f5.

Note: I've blindly adapted to kqueue, need to test it.

tags: multimaster, connpool.
(cherry picked from commit 434471857301439c8b2a48dd1594bc608c65d23a)

Author: arssher

src/backend/storage/ipc/latch.c

@@ -78,18 +78,36 @@
 #error "no wait set implementation available"
 #endif
 
+/*
+ * Connection pooler and mtm need to delete events from event set.
+ * As far as we have too preserve positions of all other events,
+ * we can not move events. So we have to maintain list of free events.
+ * But poll/WaitForMultipleObjects manipulates with array of listened events.
+ * That is why elements in pollfds and handle arrays should be stored without holes
+ * and we need to maintain mapping between them and WaitEventSet events.
+ * This mapping is stored in "permutation" array. Also we need backward mapping
+ * (from event to descriptors array) which is implemented using "index" field of WaitEvent.
+ */
+
 /* typedef in latch.h */
 struct WaitEventSet
 {
 	int			nevents;		/* number of registered events */
 	int			nevents_space;	/* maximum number of events in this set */
 
+	/*
+	 * L1-list of free events linked by "pos" and terminated by -1.
+	 */
+	int			free_events;
+
 	/*
 	 * Array, of nevents_space length, storing the definition of events this
 	 * set is waiting for.
 	 */
 	WaitEvent  *events;
 
+	int        *permutation;    /* indexes of used events (see comment above) */
+
 	/*
 	 * If WL_LATCH_SET is specified in any wait event, latch is a pointer to
 	 * said latch, and latch_pos the offset in the ->events array. This is
@@ -150,9 +168,9 @@ static void WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action
 #elif defined(WAIT_USE_KQUEUE)
 static void WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events);
 #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,
@@ -574,6 +592,7 @@ CreateWaitEventSet(MemoryContext context, int nevents)
 	 */
 	sz += MAXALIGN(sizeof(WaitEventSet));
 	sz += MAXALIGN(sizeof(WaitEvent) * nevents);
+	sz += MAXALIGN(sizeof(int) * nevents);
 
 #if defined(WAIT_USE_EPOLL)
 	sz += MAXALIGN(sizeof(struct epoll_event) * nevents);
@@ -594,23 +613,23 @@ CreateWaitEventSet(MemoryContext context, int nevents)
 	set->events = (WaitEvent *) data;
 	data += MAXALIGN(sizeof(WaitEvent) * nevents);
 
+	set->permutation = (int *) data;
+	data += MAXALIGN(sizeof(int) * nevents);
+
 #if defined(WAIT_USE_EPOLL)
 	set->epoll_ret_events = (struct epoll_event *) data;
-	data += MAXALIGN(sizeof(struct epoll_event) * nevents);
 #elif defined(WAIT_USE_KQUEUE)
 	set->kqueue_ret_events = (struct kevent *) data;
-	data += MAXALIGN(sizeof(struct kevent) * nevents);
 #elif defined(WAIT_USE_POLL)
 	set->pollfds = (struct pollfd *) data;
-	data += MAXALIGN(sizeof(struct pollfd) * nevents);
 #elif defined(WAIT_USE_WIN32)
-	set->handles = (HANDLE) data;
-	data += MAXALIGN(sizeof(HANDLE) * nevents);
+	set->handles = (HANDLE*) data;
 #endif
 
 	set->latch = NULL;
 	set->nevents_space = nevents;
 	set->exit_on_postmaster_death = false;
+	set->free_events = -1;
 
 #if defined(WAIT_USE_EPOLL)
 	if (!AcquireExternalFD())
@@ -702,12 +721,11 @@ FreeWaitEventSet(WaitEventSet *set)
 	close(set->kqueue_fd);
 	ReleaseExternalFD();
 #elif defined(WAIT_USE_WIN32)
-	WaitEvent  *cur_event;
+	int i;
 
-	for (cur_event = set->events;
-		 cur_event < (set->events + set->nevents);
-		 cur_event++)
+	for (i = 0; i < set->nevents; i++)
 	{
+		WaitEvent* cur_event = &set->events[set->permutation[i]];
 		if (cur_event->events & WL_LATCH_SET)
 		{
 			/* uses the latch's HANDLE */
@@ -720,7 +738,7 @@ FreeWaitEventSet(WaitEventSet *set)
 		{
 			/* Clean up the event object we created for the socket */
 			WSAEventSelect(cur_event->fd, NULL, 0);
-			WSACloseEvent(set->handles[cur_event->pos + 1]);
+			WSACloseEvent(set->handles[cur_event->index + 1]);
 		}
 	}
 #endif
@@ -761,6 +779,7 @@ 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);
@@ -790,8 +809,20 @@ 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->permutation[set->nevents] = event->pos;
+	event->index = set->nevents++;
 	event->fd = fd;
 	event->events = events;
 	event->user_data = user_data;
@@ -820,14 +851,54 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch,
 #elif defined(WAIT_USE_KQUEUE)
 	WaitEventAdjustKqueue(set, event, 0);
 #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 position in event set.
+ *
+ * 'pos' is the id returned by AddWaitEventToSet.
+ */
+void
+DeleteWaitEvent(WaitEventSet *set, int pos)
+{
+	WaitEvent  *event;
+#if defined(WAIT_USE_KQUEUE)
+	int			old_events;
+#endif
+
+	Assert(pos < set->nevents_space);
+	event = &set->events[pos];
+
+#if defined(WAIT_USE_EPOLL)
+	WaitEventAdjustEpoll(set, event, EPOLL_CTL_DEL);
+#elif defined(WAIT_USE_KQUEUE)
+	old_events = event->events;
+	event->events = 0;
+	WaitEventAdjustKqueue(set, event, old_events);
+#elif defined(WAIT_USE_POLL)
+	WaitEventAdjustPoll(set, event, true);
+#elif defined(WAIT_USE_WIN32)
+	WaitEventAdjustWin32(set, event, true);
+#endif
+	if (--set->nevents != 0)
+	{
+		set->permutation[event->index] = set->permutation[set->nevents];
+		set->events[set->permutation[set->nevents]].index = event->index;
+	}
+	event->fd = PGINVALID_SOCKET;
+	event->events = 0;
+	event->index = -1;
+	event->pos = set->free_events;
+	set->free_events = pos;
+}
+
+
 /*
  * Change the event mask and, in the WL_LATCH_SET case, the latch associated
  * with the WaitEvent.
@@ -842,7 +913,7 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch)
 	int			old_events;
 #endif
 
-	Assert(pos < set->nevents);
+	Assert(pos < set->nevents_space);
 
 	event = &set->events[pos];
 #if defined(WAIT_USE_KQUEUE)
@@ -884,9 +955,9 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch)
 #elif defined(WAIT_USE_KQUEUE)
 	WaitEventAdjustKqueue(set, event, old_events);
 #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
 }
 
@@ -933,7 +1004,20 @@ WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action)
 	 */
 	rc = epoll_ctl(set->epoll_fd, action, event->fd, &epoll_ev);
 
-	if (rc < 0)
+	/*
+	 * Skip throwing error in case of EPOLL_CTL_DEL. Upon connection error
+	 * libpq may or may not close the socket, so epfd can disappear.
+	 *
+	 * XXX it is not entirely clear which errnos should be checked
+	 * here. According to the mans I would say it is 'EBADF' (closed socket is
+	 * not valid, right?), any simple test on my 5.1.11 debian agrees with
+	 * that. However, msvs-6-3 bf machine with 2.6.32 spits out ENOENT (under
+	 * dmq) despite evidently correct usage (we don't DEL the same fd
+	 * twice). EINVAL was also historically checked here.
+	 */
+	if (rc < 0 &&
+		!(action == EPOLL_CTL_DEL &&
+		  (errno == EBADF || errno == EINVAL || errno == ENOENT)))
 		ereport(ERROR,
 				(errcode_for_socket_access(),
 		/* translator: %s is a syscall name, such as "poll()" */
@@ -944,11 +1028,16 @@ WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action)
 
 #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];
+	struct pollfd *pollfd = &set->pollfds[event->index];
+
+	if (remove)
+	{
+		*pollfd = set->pollfds[set->nevents - 1]; /* nevents is not decremented yet */
+		return;
+	}
 
-	pollfd->revents = 0;
 	pollfd->fd = event->fd;
 
 	/* prepare pollfd entry once */
@@ -1088,7 +1177,11 @@ WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events)
 		if (event->events == WL_POSTMASTER_DEATH &&
 			(errno == ESRCH || errno == EACCES))
 			set->report_postmaster_not_running = true;
-		else
+		/*
+		 * Like in WaitEventAdjustEpoll, don't throw if we are trying to
+		 * remove already closed socket. FIXME: ensure this check is right.
+		 */
+		else if (!(event->events == 0 && errno == EBADF ))
 			ereport(ERROR,
 					(errcode_for_socket_access(),
 			/* translator: %s is a syscall name, such as "poll()" */
@@ -1112,9 +1205,21 @@ WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events)
 
 #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];
+	HANDLE	   *handle = &set->handles[event->index + 1];
+
+	if (remove)
+	{
+		Assert(event->fd != PGINVALID_SOCKET);
+
+		if (*handle != WSA_INVALID_EVENT)
+			WSACloseEvent(*handle);
+
+		*handle = set->handles[set->nevents]; /* nevents is not decremented yet but we need to add 1 to the index */
+		set->handles[set->nevents] = WSA_INVALID_EVENT;
+		return;
+	}
 
 	if (event->events == WL_LATCH_SET)
 	{
@@ -1562,11 +1667,12 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 {
 	int			returned_events = 0;
 	int			rc;
-	WaitEvent  *cur_event;
-	struct pollfd *cur_pollfd;
+	int			i;
+	struct pollfd *cur_pollfd = set->pollfds;
+	WaitEvent* cur_event;
 
 	/* Sleep */
-	rc = poll(set->pollfds, set->nevents, (int) cur_timeout);
+	rc = poll(cur_pollfd, set->nevents, (int) cur_timeout);
 
 	/* Check return code */
 	if (rc < 0)
@@ -1589,15 +1695,13 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 		return -1;
 	}
 
-	for (cur_event = set->events, cur_pollfd = set->pollfds;
-		 cur_event < (set->events + set->nevents) &&
-		 returned_events < nevents;
-		 cur_event++, cur_pollfd++)
+	for (i = 0; i < set->nevents && returned_events < nevents; i++, cur_pollfd++)
 	{
 		/* no activity on this FD, skip */
 		if (cur_pollfd->revents == 0)
 			continue;
 
+		cur_event = &set->events[set->permutation[i]];
 		occurred_events->pos = cur_event->pos;
 		occurred_events->user_data = cur_event->user_data;
 		occurred_events->events = 0;
@@ -1688,17 +1792,25 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 					  WaitEvent *occurred_events, int nevents)
 {
 	int			returned_events = 0;
+	int			i;
 	DWORD		rc;
-	WaitEvent  *cur_event;
+	WaitEvent*	cur_event;
 
 	/* Reset any wait events that need it */
-	for (cur_event = set->events;
-		 cur_event < (set->events + set->nevents);
-		 cur_event++)
-	{
-		if (cur_event->reset)
-		{
-			WaitEventAdjustWin32(set, cur_event);
+	for (i = 0; i < set->nevents; i++)
+ 	{
+		cur_event = &set->events[set->permutation[i]];
+
+		/*
+		 * I have problem at Windows when SSPI connections "hanged" in WaitForMultipleObjects which
+		 * doesn't signal presence of input data (while it is possible to read this data from the socket).
+		 * Looks like "reset" logic is not completely correct (resetting event just after
+		 * receiveing presious read event). Reseting all read events fixes this problem.
+		 */
+		if (cur_event->events & WL_SOCKET_READABLE)
+		/* if (cur_event->reset) */
+ 		{
+			WaitEventAdjustWin32(set, cur_event, false);
 			cur_event->reset = false;
 		}
 
@@ -1764,7 +1876,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 	 * With an offset of one, due to the always present pgwin32_signal_event,
 	 * the handle offset directly corresponds to a wait event.
 	 */
-	cur_event = (WaitEvent *) &set->events[rc - WAIT_OBJECT_0 - 1];
+	cur_event = (WaitEvent *) &set->events[set->permutation[rc - WAIT_OBJECT_0 - 1]];
 
 	occurred_events->pos = cur_event->pos;
 	occurred_events->user_data = cur_event->user_data;
@@ -1805,7 +1917,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
 	else if (cur_event->events & WL_SOCKET_MASK)
 	{
 		WSANETWORKEVENTS resEvents;
-		HANDLE		handle = set->handles[cur_event->pos + 1];
+		HANDLE		handle = set->handles[cur_event->index + 1];
 
 		Assert(cur_event->fd);