Make VigorPollImpl FD-based callbacks non-periodic (take 3)

The previous attempt reveals another pre-existing issue.  An fd that is
closed by AsyncSocket may be reused right away.  The SharedStream for
that fd may still have an outstanding callback that needs to fire
before it would be destroyed.  If the destruction happens after the fd
is re-used and the new owner is using boost asio stream, the Release()
call would cancel the async read/write operation for the new owner.
One simple fix is to dup the fd when creating a SharedStream.  This
means hostd would need 2 fds for each running VM.  The hostd fd limit
is currently 4096 and I was able to power on 1K VMs.  If necessary, the
fd limit can be increased (the system limit is 32K).

Previous description:

The original change has an issue on Linux that is not observed in ESX.
When the read and write callbacks fire in parallel, both would release
the reference to the SharedStream, triggering its destructor.  Later
when the read callback tries to re-register itself, it would hit a
boost exception.  To avoid that, I delayed the reset of
CallbackFD::stream to the end of Invoke so that the SharedStream
object woudl not be destroyed as long as the callback is re-registered.

Original description:

The hard-to-reproduce AsyncSocket leak turns out to be a variation of
PR 1530561.  The fix for that bug only plugs the leak when the race is
between a VigorPollImpl callback that fires with an CanceledException
(which can be triggered by another callback registered on the same fd --
this is the boost behavior) and a thread trying to unregister the same
callback (e.g. upon detection of a disconnect).  There is still a
potential reference leak when the read and write callbacks run in
parallel, if a disconnection happens between the two asyncsocket
callbacks (the asyncsocket callbacks take the asyncsocket lock, so they
are mostly serialized).  The second asyncsocket callback handling the
disconnect error would unregister the first callback while the first
callback would be attempting to re-register itself at the VigorPollImpl
level (to provide the periodic behavior exposed by IVmdbPoll).  If the
unregister wins the race, it would return false because the callback
cannot be cancelled (it is running).  Then ReRegister would not find
the callback in the hash table, so the callback would not fire again,
which means the reference taken when registering the callback is leaked.

Applying the fix for PR 1530561 in the above scenario would not work
because an AsycnSocket callback may unregister itself and in that case
the reference is appropriately released.  Unless we save the thread ID
in Unregister, we would not be able to distinguish when the callback
needs to run in order to release the reference.  After considering
various options, I believe trying to support periodic fd-based
callbacks on top of a multi-threaded poll API is troublesome.  This
change adds a poll flag for one-shot callbacks and used that in
AsyncSocket with IVmdbPoll.  VigorPollImpl only support the one-shot
behavior.  In the long term I would like to retire IVmdbPoll and create
a new poll interface.  For now I would like to minimize churn while
fixing the memory leak.

diff --git a/open-vm-tools/lib/asyncsocket/asyncsocket.c b/open-vm-tools/lib/asyncsocket/asyncsocket.c
index bdea2f56dfbdca754dc4819a4a2db03cac0416a8..54c61d269cc2e22e0e04764142ab3586e8deb9e6 100644 (file)
--- a/open-vm-tools/lib/asyncsocket/asyncsocket.c
+++ b/open-vm-tools/lib/asyncsocket/asyncsocket.c
@@ -4754,22 +4754,22 @@ AsyncTCPSocketIPollRecvCallback(void *clientData)  // IN:
 
    AsyncTCPSocketLock(asock);
    if (asock->recvCbTimer) {
-      /* IVmdbPoll only has periodic callbacks. */
+      /* IVmdbPoll only has periodic timer callbacks. */
       AsyncTCPSocketIPollRemove(asock, FALSE, 0, asock->internalRecvFn);
       asock->recvCbTimer = FALSE;
    }
    lock = AsyncTCPSocketPollParams(asock)->lock;
-   asock->inIPollCb |= IN_IPOLL_RECV;
    if (asock->recvCb && asock->inBlockingRecv == 0) {
+      asock->inIPollCb |= IN_IPOLL_RECV;
+      AsyncTCPSocketRecvCallback(clientData);
+      asock->inIPollCb &= ~IN_IPOLL_RECV;
       /*
-       * There is no need to take a reference here -- the fact that this
-       * callback is running means AsyncsocketIPollRemove would not release a
-       * reference if it is called.
+       * Re-register the callback if it has not been canceled.  Lock may have
+       * been dropped to fire recv callback so re-check inBlockingRecv.
        */
-      int error = AsyncTCPSocketFillRecvBuffer(asock);
-
-      if (error == ASOCKERR_GENERIC || error == ASOCKERR_REMOTE_DISCONNECT) {
-         AsyncTCPSocketHandleError(asock, error);
+      if (asock->recvCb && asock->inBlockingRecv == 0) {
+         AsyncTCPSocketIPollAdd(asock, TRUE, POLL_FLAG_READ,
+                                asock->internalRecvFn, asock->fd);
       }
    } else {
       TCPSOCKLG0(asock, ("Skip recv because %s\n",
@@ -4777,19 +4777,11 @@ AsyncTCPSocketIPollRecvCallback(void *clientData)  // IN:
                                        : "recv callback is cancelled"));
    }
 
-   asock->inIPollCb &= ~IN_IPOLL_RECV;
-   if (asock->recvCb) {
-      AsyncTCPSocketUnlock(asock);
-   } else {
-      /*
-       * Callback has been unregistered.  Per above, we need to release the
-       * reference explicitly.
-       */
-      AsyncTCPSocketRelease(asock);
-      AsyncTCPSocketUnlock(asock);
-      if (lock != NULL) {
-         MXUser_DecRefRecLock(lock);
-      }
+   /* This is a one-shot callback so we always release the reference taken. */
+   AsyncTCPSocketRelease(asock);
+   AsyncTCPSocketUnlock(asock);
+   if (lock != NULL) {
+      MXUser_DecRefRecLock(lock);
    }
 #endif
 }
@@ -4897,21 +4889,15 @@ AsyncTCPSocketIPollSendCallback(void *clientData)  // IN:
    AsyncTCPSocketLock(s);
    s->inIPollCb |= IN_IPOLL_SEND;
    lock = AsyncTCPSocketPollParams(s)->lock;
+   if (s->sendCbTimer) {
+      /* IVmdbPoll only has periodic timer callback. */
+      AsyncTCPSocketIPollRemove(s, FALSE, 0, AsyncTCPSocketIPollSendCallback);
+      s->sendCbTimer = FALSE;
+   }
    if (s->sendCb) {
-      /*
-       * Unregister this callback as we want the non-periodic behavior.  There
-       * is no need to take a reference here -- the fact that this callback is
-       * running means AsyncsocketIPollRemove would not release a reference.
-       * We would release that reference at the end.
-       */
-      if (s->sendCbTimer) {
-         AsyncTCPSocketIPollRemove(s, FALSE, 0, AsyncTCPSocketIPollSendCallback);
-      } else {
-         AsyncTCPSocketIPollRemove(s, TRUE, POLL_FLAG_WRITE,
-                                   AsyncTCPSocketIPollSendCallback);
-      }
-
       AsyncTCPSocketSendCallback(s);
+   } else {
+      TCPSOCKLG0(s, ("cancelled send callback fired\n"));
    }
 
    s->inIPollCb &= ~IN_IPOLL_SEND;
@@ -5065,8 +5051,9 @@ AsyncTCPSocketIPollAdd(AsyncTCPSocket *asock,         // IN
    poll = AsyncTCPSocketPollParams(asock)->iPoll;
 
    if (socket) {
-      int pollFlags = (flags & POLL_FLAG_READ) != 0 ? VMDB_PRF_READ
-                                                    : VMDB_PRF_WRITE;
+      int pollFlags = VMDB_PRF_ONE_SHOT |
+                      ((flags & POLL_FLAG_READ) != 0 ? VMDB_PRF_READ
+                                                     : VMDB_PRF_WRITE);
 
       ret = poll->Register(poll, pollFlags, callback, asock, info);
    } else {
@@ -5124,8 +5111,9 @@ AsyncTCPSocketIPollRemove(AsyncTCPSocket *asock,         // IN
    poll = AsyncTCPSocketPollParams(asock)->iPoll;
 
    if (socket) {
-      int pollFlags = (flags & POLL_FLAG_READ) != 0 ? VMDB_PRF_READ
-                                                    : VMDB_PRF_WRITE;
+      int pollFlags = VMDB_PRF_ONE_SHOT |
+                      ((flags & POLL_FLAG_READ) != 0 ? VMDB_PRF_READ
+                                                     : VMDB_PRF_WRITE);
 
       ret = poll->Unregister(poll, pollFlags, callback, asock);
    } else {