RX: Avoid retrying calls on busy channels

When we receive an RX_PACKET_TYPE_BUSY packet, we currently ignore it.
This is a problem when the server has a long-running call on that same
call channel that we don't know about, since we will then keep
retrying the call on the same channel and keep getting
RX_PACKET_TYPE_BUSY responses.

Try to avoid this by returning the RX_CALL_TIMEOUT error when we get a
BUSY packet and there are other free call channels available on the
conn. When the application gets the error and retries the call, we
avoid using the same call channel again where possible. When all of
the call channels appear busy, we revert to effectively the old
behavior of retrying the call on the same channel until we get an
RX_CALL_DEAD (or similar) error.

FIXES 128671

Change-Id: I29b84aa36dcc856528c30d26811e5d6c4b78a1ca
Reviewed-on: http://gerrit.openafs.org/3431
Tested-by: Andrew Deason <adeason@sinenomine.net>
Reviewed-by: Jeffrey Altman <jaltman@openafs.org>
Tested-by: BuildBot <buildbot@rampaginggeek.com>
Reviewed-by: Derrick Brashear <shadow@dementia.org>

diff --git a/src/rx/rx.c b/src/rx/rx.c
index 0b75496..fa9c9a1 100644 (file)
--- a/src/rx/rx.c
+++ b/src/rx/rx.c
@@ -843,6 +843,7 @@ rx_NewConnection(afs_uint32 shost, u_short sport, u_short sservice,
     for (i = 0; i < RX_MAXCALLS; i++) {
        conn->twind[i] = rx_initSendWindow;
        conn->rwind[i] = rx_initReceiveWindow;
+       conn->lastBusy[i] = 0;
     }
 
     RXS_NewConnection(securityObject, conn);
@@ -1210,9 +1211,10 @@ rxi_WakeUpTransmitQueue(struct rx_call *call)
 struct rx_call *
 rx_NewCall(struct rx_connection *conn)
 {
-    int i, wait;
+    int i, wait, ignoreBusy = 1;
     struct rx_call *call;
     struct clock queueTime;
+    afs_uint32 leastBusy = 0;
     SPLVAR;
 
     clock_NewTime();
@@ -1263,9 +1265,25 @@ rx_NewCall(struct rx_connection *conn)
        for (i = 0; i < RX_MAXCALLS; i++) {
            call = conn->call[i];
            if (call) {
+               if (!ignoreBusy && conn->lastBusy[i] != leastBusy) {
+                   /* we're not ignoring busy call slots; only look at the
+                    * call slot that is the "least" busy */
+                   continue;
+               }
+
                if (call->state == RX_STATE_DALLY) {
                     MUTEX_ENTER(&call->lock);
                     if (call->state == RX_STATE_DALLY) {
+                       if (ignoreBusy && conn->lastBusy[i]) {
+                           /* if we're ignoring busy call slots, skip any ones that
+                            * have lastBusy set */
+                           if (leastBusy == 0 || conn->lastBusy[i] < leastBusy) {
+                               leastBusy = conn->lastBusy[i];
+                           }
+                           MUTEX_EXIT(&call->lock);
+                           continue;
+                       }
+
                         /*
                          * We are setting the state to RX_STATE_RESET to
                          * ensure that no one else will attempt to use this
@@ -1322,6 +1340,15 @@ rx_NewCall(struct rx_connection *conn)
                     MUTEX_EXIT(&call->lock);
                 }
            } else {
+               if (ignoreBusy && conn->lastBusy[i]) {
+                   /* if we're ignoring busy call slots, skip any ones that
+                    * have lastBusy set */
+                   if (leastBusy == 0 || conn->lastBusy[i] < leastBusy) {
+                       leastBusy = conn->lastBusy[i];
+                   }
+                   continue;
+               }
+
                 /* rxi_NewCall returns with mutex locked */
                call = rxi_NewCall(conn, i);
                 MUTEX_ENTER(&rx_refcnt_mutex);
@@ -1331,10 +1358,18 @@ rx_NewCall(struct rx_connection *conn)
            }
        }
        if (i < RX_MAXCALLS) {
+           conn->lastBusy[i] = 0;
            break;
        }
         if (!wait)
             continue;
+       if (leastBusy && ignoreBusy) {
+           /* we didn't find a useable call slot, but we did see at least one
+            * 'busy' slot; look again and only use a slot with the 'least
+            * busy time */
+           ignoreBusy = 0;
+           continue;
+       }
 
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->flags |= RX_CONN_MAKECALL_WAITING;
@@ -2173,6 +2208,11 @@ rx_EndCall(struct rx_call *call, afs_int32 rc)
         MUTEX_EXIT(&call->lock);
         MUTEX_ENTER(&conn->conn_call_lock);
         MUTEX_ENTER(&call->lock);
+
+       if (!(call->flags & RX_CALL_PEER_BUSY)) {
+           conn->lastBusy[call->channel] = 0;
+       }
+
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->flags |= RX_CONN_BUSY;
        if (conn->flags & RX_CONN_MAKECALL_WAITING) {
@@ -2769,6 +2809,86 @@ rxi_FindConnection(osi_socket socket, afs_uint32 host,
     return conn;
 }
 
+/**
+ * Timeout a call on a busy call channel if appropriate.
+ *
+ * @param[in] call The busy call.
+ *
+ * @pre 'call' is marked as busy (namely,
+ *      call->conn->lastBusy[call->channel] != 0)
+ *
+ * @pre call->lock is held
+ *
+ * @note call->lock is dropped and reacquired
+ */
+static void
+rxi_CheckBusy(struct rx_call *call)
+{
+    struct rx_connection *conn = call->conn;
+    int channel = call->channel;
+    int freechannel = 0;
+    int i;
+    afs_uint32 callNumber = *call->callNumber;
+
+    MUTEX_EXIT(&call->lock);
+
+    MUTEX_ENTER(&conn->conn_call_lock);
+
+    /* Are there any other call slots on this conn that we should try? Look for
+     * slots that are empty and are either non-busy, or were marked as busy
+     * longer than conn->secondsUntilDead seconds before this call started. */
+
+    for (i = 0; i < RX_MAXCALLS && !freechannel; i++) {
+       if (i == channel) {
+           /* only look at channels that aren't us */
+           continue;
+       }
+
+       if (conn->lastBusy[i]) {
+           /* if this channel looked busy too recently, don't look at it */
+           if (conn->lastBusy[i] >= call->startTime.sec) {
+               continue;
+           }
+           if (call->startTime.sec - conn->lastBusy[i] < conn->secondsUntilDead) {
+               continue;
+           }
+       }
+
+       if (conn->call[i]) {
+           struct rx_call *tcall = conn->call[i];
+           MUTEX_ENTER(&tcall->lock);
+           if (tcall->state == RX_STATE_DALLY) {
+               freechannel = 1;
+           }
+           MUTEX_EXIT(&tcall->lock);
+       } else {
+           freechannel = 1;
+       }
+    }
+
+    MUTEX_EXIT(&conn->conn_call_lock);
+
+    MUTEX_ENTER(&call->lock);
+
+    /* Since the call->lock and conn->conn_call_lock have been released it is
+     * possible that (1) the call may no longer be busy and/or (2) the call may
+     * have been reused by another waiting thread. Therefore, we must confirm
+     * that the call state has not changed when deciding whether or not to
+     * force this application thread to retry by forcing a Timeout error. */
+
+    if (freechannel && *call->callNumber == callNumber &&
+        (call->flags & RX_CALL_PEER_BUSY)) {
+       /* Since 'freechannel' is set, there exists another channel in this
+        * rx_conn that the application thread might be able to use. We know
+        * that we have the correct call since callNumber is unchanged, and we
+        * know that the call is still busy. So, set the call error state to
+        * RX_CALL_TIMEOUT so the application can retry the request, presumably
+        * on a less-busy call channel. */
+
+       rxi_CallError(call, RX_CALL_TIMEOUT);
+    }
+}
+
 /* There are two packet tracing routines available for testing and monitoring
  * Rx.  One is called just after every packet is received and the other is
  * called just before every packet is sent.  Received packets, have had their
@@ -3252,9 +3372,26 @@ rxi_ReceivePacket(struct rx_packet *np, osi_socket socket,
         MUTEX_EXIT(&rx_refcnt_mutex);
        return np;              /* xmitting; drop packet */
     }
-    case RX_PACKET_TYPE_BUSY:
-       /* XXXX */
-       break;
+    case RX_PACKET_TYPE_BUSY: {
+       struct clock busyTime;
+       clock_NewTime();
+       clock_GetTime(&busyTime);
+
+       MUTEX_EXIT(&call->lock);
+
+       MUTEX_ENTER(&conn->conn_call_lock);
+       MUTEX_ENTER(&call->lock);
+       conn->lastBusy[call->channel] = busyTime.sec;
+       call->flags |= RX_CALL_PEER_BUSY;
+       MUTEX_EXIT(&call->lock);
+       MUTEX_EXIT(&conn->conn_call_lock);
+
+       MUTEX_ENTER(&rx_refcnt_mutex);
+       conn->refCount--;
+       MUTEX_EXIT(&rx_refcnt_mutex);
+       return np;
+    }
+
     case RX_PACKET_TYPE_ACKALL:
        /* All packets acknowledged, so we can drop all packets previously
         * readied for sending */
@@ -3296,6 +3433,8 @@ rxi_ReceivePacket(struct rx_packet *np, osi_socket socket,
      * the packet will be delivered to the user before any get time is required
      * (if not, then the time won't actually be re-evaluated here). */
     call->lastReceiveTime = clock_Sec();
+    /* we've received a legit packet, so the channel is not busy */
+    call->flags &= ~RX_CALL_PEER_BUSY;
     MUTEX_EXIT(&call->lock);
     MUTEX_ENTER(&rx_refcnt_mutex);
     conn->refCount--;
@@ -5023,6 +5162,12 @@ rxi_ResetCall(struct rx_call *call, int newcall)
     }
     call->flags = 0;
 
+    if ((flags & RX_CALL_PEER_BUSY)) {
+       /* The call channel is still busy; resetting the call doesn't change
+        * that */
+       call->flags |= RX_CALL_PEER_BUSY;
+    }
+
     rxi_ClearReceiveQueue(call);
     /* why init the queue if you just emptied it? queue_Init(&call->rq); */
 
@@ -5640,6 +5785,11 @@ rxi_Start(struct rxevent *event,
        CALL_RELE(call, RX_CALL_REFCOUNT_RESEND);
         MUTEX_EXIT(&rx_refcnt_mutex);
        call->resendEvent = NULL;
+
+       if ((call->flags & RX_CALL_PEER_BUSY)) {
+           rxi_CheckBusy(call);
+       }
+
        if (queue_IsEmpty(&call->tq)) {
            /* Nothing to do */
            return;

diff --git a/src/rx/rx.h b/src/rx/rx.h
index deb50fb..d280fe9 100644 (file)
--- a/src/rx/rx.h
+++ b/src/rx/rx.h
@@ -241,6 +241,9 @@ struct rx_connection {
     afs_uint32 callNumber[RX_MAXCALLS];        /* Current call numbers */
     afs_uint32 rwind[RX_MAXCALLS];
     u_short twind[RX_MAXCALLS];
+    afs_uint32 lastBusy[RX_MAXCALLS]; /* timestamp of the last time we got an
+                                       * RX_PACKET_TYPE_BUSY packet for this
+                                       * call slot, or 0 if the slot is not busy */
     afs_uint32 serial;         /* Next outgoing packet serial number */
     afs_uint32 lastSerial;     /* # of last packet received, for computing skew */
     afs_int32 maxSerial;       /* largest serial number seen on incoming packets */
@@ -633,6 +636,8 @@ struct rx_call {
 #define RX_CALL_IOVEC_WAIT     16384   /* waiting thread is using an iovec */
 #define RX_CALL_HAVE_LAST      32768   /* Last packet has been received */
 #define RX_CALL_NEED_START     0x10000 /* tells rxi_Start to start again */
+#define RX_CALL_PEER_BUSY      0x20000 /* the last packet we received on this call was a
+                                         * BUSY packet; i.e. the channel for this call is busy */
 
 
 /* The structure of the data portion of an acknowledge packet: An acknowledge