Add a poll flag for one-shot callbacks in AsyncSocket

The original change had an issue.  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 tried to
re-register itself, it would hit a boost exception.  To avoid that,
delay the reset of CallbackFD::stream to the end of Invoke so that the
SharedStream object will not be destroyed as long as the callback is
re-registered.

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 {