rx-tq-busy-20070516

[openafs.git] / src / rx / rx.c

diff --git a/src/rx/rx.c b/src/rx/rx.c
index 4934de3..df35abc 100644 (file)
--- a/src/rx/rx.c
+++ b/src/rx/rx.c
@@ -1039,10 +1039,28 @@ rx_GetConnection(register struct rx_connection *conn)
     USERPRI;
 }
 
+/* Wait for the transmit queue to no longer be busy. 
+ * requires the call->lock to be held */
+static void rxi_WaitforTQBusy(struct rx_call *call) {
+    while (call->flags & RX_CALL_TQ_BUSY) {
+       call->flags |= RX_CALL_TQ_WAIT;
+       call->tqWaiters++;
+#ifdef RX_ENABLE_LOCKS
+       osirx_AssertMine(&call->lock, "rxi_WaitforTQ lock");
+       CV_WAIT(&call->cv_tq, &call->lock);
+#else /* RX_ENABLE_LOCKS */
+       osi_rxSleep(&call->tq);
+#endif /* RX_ENABLE_LOCKS */
+       call->tqWaiters--;
+       if (call->tqWaiters == 0) {
+           call->flags &= ~RX_CALL_TQ_WAIT;
+       }
+    }
+}
 /* Start a new rx remote procedure call, on the specified connection.
  * If wait is set to 1, wait for a free call channel; otherwise return
  * 0.  Maxtime gives the maximum number of seconds this call may take,
- * after rx_MakeCall returns.  After this time interval, a call to any
+ * after rx_NewCall returns.  After this time interval, a call to any
  * of rx_SendData, rx_ReadData, etc. will fail with RX_CALL_TIMEOUT.
  * For fine grain locking, we hold the conn_call_lock in order to 
  * to ensure that we don't get signalle after we found a call in an active
@@ -1057,7 +1075,7 @@ rx_NewCall(register struct rx_connection *conn)
     SPLVAR;
 
     clock_NewTime();
-    dpf(("rx_MakeCall(conn %x)\n", conn));
+    dpf(("rx_NewCall(conn %x)\n", conn));
 
     NETPRI;
     clock_GetTime(&queueTime);
@@ -1164,20 +1182,7 @@ rx_NewCall(register struct rx_connection *conn)
 #ifdef AFS_GLOBAL_RXLOCK_KERNEL
     /* Now, if TQ wasn't cleared earlier, do it now. */
     MUTEX_ENTER(&call->lock);
-    while (call->flags & RX_CALL_TQ_BUSY) {
-       call->flags |= RX_CALL_TQ_WAIT;
-       call->tqWaiters++;
-#ifdef RX_ENABLE_LOCKS
-       osirx_AssertMine(&call->lock, "rxi_Start lock4");
-       CV_WAIT(&call->cv_tq, &call->lock);
-#else /* RX_ENABLE_LOCKS */
-       osi_rxSleep(&call->tq);
-#endif /* RX_ENABLE_LOCKS */
-       call->tqWaiters--;
-       if (call->tqWaiters == 0) {
-           call->flags &= ~RX_CALL_TQ_WAIT;
-       }
-    }
+    rxi_WaitforTQBusy(call);
     if (call->flags & RX_CALL_TQ_CLEARME) {
        rxi_ClearTransmitQueue(call, 0);
        queue_Init(&call->tq);
@@ -1185,6 +1190,7 @@ rx_NewCall(register struct rx_connection *conn)
     MUTEX_EXIT(&call->lock);
 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
 
+    dpf(("rx_NewCall(call %x)\n", call));
     return call;
 }
 
@@ -2080,6 +2086,8 @@ rxi_NewCall(register struct rx_connection *conn, register int channel)
     register struct rx_call *nxp;      /* Next call pointer, for queue_Scan */
 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
 
+    dpf(("rxi_NewCall(conn %x, channel %d)\n", conn, channel));
+
     /* Grab an existing call structure, or allocate a new one.
      * Existing call structures are assumed to have been left reset by
      * rxi_FreeCall */
@@ -3352,9 +3360,11 @@ rxi_ReceiveDataPacket(register struct rx_call *call,
 
            /* We need to send an ack of the packet is out of sequence, 
             * or if an ack was requested by the peer. */
-           if (seq != prev + 1 || missing || (flags & RX_REQUEST_ACK)) {
+           if (seq != prev + 1 || missing) {
                ackNeeded = RX_ACK_OUT_OF_SEQUENCE;
-           }
+           } else if (flags & RX_REQUEST_ACK) {
+               ackNeeded = RX_ACK_REQUESTED;
+            }
 
            /* Acknowledge the last packet for each call */
            if (flags & RX_LAST_PACKET) {
@@ -3886,19 +3896,7 @@ rxi_ReceiveAckPacket(register struct rx_call *call, struct rx_packet *np,
            return np;
        }
        call->flags |= RX_CALL_FAST_RECOVER_WAIT;
-       while (call->flags & RX_CALL_TQ_BUSY) {
-           call->flags |= RX_CALL_TQ_WAIT;
-           call->tqWaiters++;
-#ifdef RX_ENABLE_LOCKS
-           osirx_AssertMine(&call->lock, "rxi_Start lock2");
-           CV_WAIT(&call->cv_tq, &call->lock);
-#else /* RX_ENABLE_LOCKS */
-           osi_rxSleep(&call->tq);
-#endif /* RX_ENABLE_LOCKS */
-           call->tqWaiters--;
-           if (call->tqWaiters == 0)
-               call->flags &= ~RX_CALL_TQ_WAIT;
-       }
+       rxi_WaitforTQBusy(call);
        MUTEX_ENTER(&peer->peer_lock);
 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        call->flags &= ~RX_CALL_FAST_RECOVER_WAIT;
@@ -4463,6 +4461,8 @@ rxi_ResetCall(register struct rx_call *call, register int newcall)
     register struct rx_peer *peer;
     struct rx_packet *packet;
 
+    dpf(("rxi_ResetCall(call %x, newcall %d)\n", call, newcall));
+
     /* Notify anyone who is waiting for asynchronous packet arrival */
     if (call->arrivalProc) {
        (*call->arrivalProc) (call, call->arrivalProcHandle,
@@ -5104,19 +5104,7 @@ rxi_Start(struct rxevent *event, register struct rx_call *call,
            return;
        }
        call->flags |= RX_CALL_FAST_RECOVER_WAIT;
-       while (call->flags & RX_CALL_TQ_BUSY) {
-           call->flags |= RX_CALL_TQ_WAIT;
-           call->tqWaiters++;
-#ifdef RX_ENABLE_LOCKS
-           osirx_AssertMine(&call->lock, "rxi_Start lock1");
-           CV_WAIT(&call->cv_tq, &call->lock);
-#else /* RX_ENABLE_LOCKS */
-           osi_rxSleep(&call->tq);
-#endif /* RX_ENABLE_LOCKS */
-           call->tqWaiters--;
-           if (call->tqWaiters == 0)
-               call->flags &= ~RX_CALL_TQ_WAIT;
-       }
+       rxi_WaitforTQBusy(call);
 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        call->flags &= ~RX_CALL_FAST_RECOVER_WAIT;
        call->flags |= RX_CALL_FAST_RECOVER;
@@ -5631,6 +5619,7 @@ rxi_SendDelayedCallAbort(struct rxevent *event, register struct rx_call *call,
                            (char *)&error, sizeof(error), 0);
        rxi_FreePacket(packet);
     }
+    CALL_RELE(call, RX_CALL_REFCOUNT_ABORT);
     MUTEX_EXIT(&call->lock);
 }