#endif /* KERNEL */
#include <opr/queue.h>
+#include <hcrypto/rand.h>
#include "rx.h"
#include "rx_clock.h"
static struct rx_connection
*rxi_FindConnection(osi_socket socket, afs_uint32 host, u_short port,
u_short serviceId, afs_uint32 cid,
- afs_uint32 epoch, int type, u_int securityIndex);
+ afs_uint32 epoch, int type, u_int securityIndex,
+ int *unknownService);
static struct rx_packet
*rxi_ReceiveDataPacket(struct rx_call *call, struct rx_packet *np,
int istack, osi_socket socket,
static void rxi_KeepAliveOn(struct rx_call *call);
static void rxi_GrowMTUOn(struct rx_call *call);
static void rxi_ChallengeOn(struct rx_connection *conn);
-
-#ifdef RX_ENABLE_LOCKS
static int rxi_CheckCall(struct rx_call *call, int haveCTLock);
-static void rxi_SetAcksInTransmitQueue(struct rx_call *call);
-#else
-static int rxi_CheckCall(struct rx_call *call);
-#endif
+static void rxi_AckAllInTransmitQueue(struct rx_call *call);
+static void rxi_CancelKeepAliveEvent(struct rx_call *call);
+static void rxi_CancelDelayedAbortEvent(struct rx_call *call);
+static void rxi_CancelGrowMTUEvent(struct rx_call *call);
+static void update_nextCid(void);
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
struct rx_tq_debug {
rx_atomic_t rxi_start_aborted; /* rxi_start awoke after rxi_Send in error.*/
rx_atomic_t rxi_start_in_error;
} rx_tq_debug;
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
/* Constant delay time before sending an acknowledge of the last packet
* received. This is to avoid sending an extra acknowledge when the
* tiers:
*
* rx_connHashTable_lock - synchronizes conn creation, rx_connHashTable access
+ * also protects updates to rx_nextCid
* conn_call_lock - used to synchonize rx_EndCall and rx_NewCall
* call->lock - locks call data fields.
* These are independent of each other:
#define CLEAR_CALL_QUEUE_LOCK(C)
#endif /* RX_ENABLE_LOCKS */
struct rx_serverQueueEntry *rx_waitForPacket = 0;
-struct rx_serverQueueEntry *rx_waitingForPacket = 0;
/* ------------Exported Interfaces------------- */
-/* This function allows rxkad to set the epoch to a suitably random number
- * which rx_NewConnection will use in the future. The principle purpose is to
- * get rxnull connections to use the same epoch as the rxkad connections do, at
- * least once the first rxkad connection is established. This is important now
- * that the host/port addresses aren't used in FindConnection: the uniqueness
- * of epoch/cid matters and the start time won't do. */
-
-#ifdef AFS_PTHREAD_ENV
-/*
- * This mutex protects the following global variables:
- * rx_epoch
- */
-
-#define LOCK_EPOCH MUTEX_ENTER(&epoch_mutex)
-#define UNLOCK_EPOCH MUTEX_EXIT(&epoch_mutex)
-#else
-#define LOCK_EPOCH
-#define UNLOCK_EPOCH
-#endif /* AFS_PTHREAD_ENV */
-
-void
-rx_SetEpoch(afs_uint32 epoch)
-{
- LOCK_EPOCH;
- rx_epoch = epoch;
- UNLOCK_EPOCH;
-}
-
/* Initialize rx. A port number may be mentioned, in which case this
* becomes the default port number for any service installed later.
* If 0 is provided for the port number, a random port will be chosen
#ifndef AFS_NT40_ENV
static
#endif
-int rxinit_status = 1;
-#ifdef AFS_PTHREAD_ENV
-/*
- * This mutex protects the following global variables:
- * rxinit_status
- */
-
-#define LOCK_RX_INIT MUTEX_ENTER(&rx_init_mutex)
-#define UNLOCK_RX_INIT MUTEX_EXIT(&rx_init_mutex)
-#else
-#define LOCK_RX_INIT
-#define UNLOCK_RX_INIT
-#endif
+rx_atomic_t rxinit_status = RX_ATOMIC_INIT(1);
int
rx_InitHost(u_int host, u_int port)
struct timeval tv;
#endif /* KERNEL */
char *htable, *ptable;
- int tmp_status;
SPLVAR;
INIT_PTHREAD_LOCKS;
- LOCK_RX_INIT;
- if (rxinit_status == 0) {
- tmp_status = rxinit_status;
- UNLOCK_RX_INIT;
- return tmp_status; /* Already started; return previous error code. */
- }
+ if (!rx_atomic_test_and_clear_bit(&rxinit_status, 0))
+ return 0; /* already started */
+
#ifdef RXDEBUG
rxi_DebugInit();
#endif
rx_socket = rxi_GetHostUDPSocket(host, (u_short) port);
if (rx_socket == OSI_NULLSOCKET) {
- UNLOCK_RX_INIT;
return RX_ADDRINUSE;
}
#if defined(RX_ENABLE_LOCKS) && defined(KERNEL)
#endif
}
rx_stats.minRtt.sec = 9999999;
-#ifdef KERNEL
- rx_SetEpoch(tv.tv_sec | 0x80000000);
-#else
- rx_SetEpoch(tv.tv_sec); /* Start time of this package, rxkad
- * will provide a randomer value. */
-#endif
+ if (RAND_bytes(&rx_epoch, sizeof(rx_epoch)) != 1)
+ return -1;
+ rx_epoch = (rx_epoch & ~0x40000000) | 0x80000000;
+ if (RAND_bytes(&rx_nextCid, sizeof(rx_nextCid)) != 1)
+ return -1;
+ rx_nextCid &= RX_CIDMASK;
MUTEX_ENTER(&rx_quota_mutex);
rxi_dataQuota += rx_extraQuota; /* + extra pkts caller asked to rsrv */
MUTEX_EXIT(&rx_quota_mutex);
rx_GetIFInfo();
#endif
-#if defined(RXK_LISTENER_ENV) || !defined(KERNEL)
/* Start listener process (exact function is dependent on the
* implementation environment--kernel or user space) */
rxi_StartListener();
-#endif
USERPRI;
- tmp_status = rxinit_status = 0;
- UNLOCK_RX_INIT;
- return tmp_status;
+ rx_atomic_clear_bit(&rxinit_status, 0);
+ return 0;
}
int
static_inline void
rxi_rto_cancel(struct rx_call *call)
{
- rxevent_Cancel(&call->resendEvent, call, RX_CALL_REFCOUNT_RESEND);
+ if (call->resendEvent != NULL) {
+ rxevent_Cancel(&call->resendEvent);
+ CALL_RELE(call, RX_CALL_REFCOUNT_RESEND);
+ }
}
/*!
void
rx_SetBusyChannelError(afs_int32 onoff)
{
- osi_Assert(rxinit_status != 0);
+ osi_Assert(rx_atomic_test_bit(&rxinit_status, 0));
rxi_busyChannelError = onoff ? 1 : 0;
}
when = now;
clock_Add(&when, offset);
- if (!call->delayedAckEvent
- || clock_Gt(&call->delayedAckTime, &when)) {
+ if (call->delayedAckEvent && clock_Gt(&call->delayedAckTime, &when)) {
+ /* The event we're cancelling already has a reference, so we don't
+ * need a new one */
+ rxevent_Cancel(&call->delayedAckEvent);
+ call->delayedAckEvent = rxevent_Post(&when, &now, rxi_SendDelayedAck,
+ call, NULL, 0);
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ call->delayedAckTime = when;
+ } else if (!call->delayedAckEvent) {
CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
-
call->delayedAckEvent = rxevent_Post(&when, &now,
rxi_SendDelayedAck,
call, NULL, 0);
}
}
+void
+rxi_CancelDelayedAckEvent(struct rx_call *call)
+{
+ if (call->delayedAckEvent) {
+ rxevent_Cancel(&call->delayedAckEvent);
+ CALL_RELE(call, RX_CALL_REFCOUNT_DELAY);
+ }
+}
+
/* called with unincremented nRequestsRunning to see if it is OK to start
* a new thread in this service. Could be "no" for two reasons: over the
* max quota, or would prevent others from reaching their min quota.
int serviceSecurityIndex)
{
int hashindex, i;
- afs_int32 cid;
struct rx_connection *conn;
SPLVAR;
#endif
NETPRI;
MUTEX_ENTER(&rx_connHashTable_lock);
- cid = (rx_nextCid += RX_MAXCALLS);
conn->type = RX_CLIENT_CONNECTION;
- conn->cid = cid;
conn->epoch = rx_epoch;
- conn->peer = rxi_FindPeer(shost, sport, 0, 1);
+ conn->cid = rx_nextCid;
+ update_nextCid();
+ conn->peer = rxi_FindPeer(shost, sport, 1);
conn->serviceId = sservice;
conn->securityObject = securityObject;
conn->securityData = (void *) 0;
/* Push the final acknowledgment out now--there
* won't be a subsequent call to acknowledge the
* last reply packets */
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
if (call->state == RX_STATE_PRECALL
|| call->state == RX_STATE_ACTIVE) {
rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
}
if (conn->delayedAbortEvent) {
- rxevent_Cancel(&conn->delayedAbortEvent, NULL, 0);
+ rxevent_Cancel(&conn->delayedAbortEvent);
packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
if (packet) {
MUTEX_ENTER(&conn->conn_data_lock);
/* Make sure the connection is completely reset before deleting it. */
/* get rid of pending events that could zap us later */
- rxevent_Cancel(&conn->challengeEvent, NULL, 0);
- rxevent_Cancel(&conn->checkReachEvent, NULL, 0);
- rxevent_Cancel(&conn->natKeepAliveEvent, NULL, 0);
+ rxevent_Cancel(&conn->challengeEvent);
+ rxevent_Cancel(&conn->checkReachEvent);
+ rxevent_Cancel(&conn->natKeepAliveEvent);
/* Add the connection to the list of destroyed connections that
* need to be cleaned up. This is necessary to avoid deadlocks
USERPRI;
}
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
/* Wait for the transmit queue to no longer be busy.
* requires the call->lock to be held */
void
while (!call->error && (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");
+ MUTEX_ASSERT(&call->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;
dpf(("call %"AFS_PTR_FMT" has %d waiters and flags %d\n",
call, call->tqWaiters, call->flags));
#ifdef RX_ENABLE_LOCKS
- osirx_AssertMine(&call->lock, "rxi_Start start");
+ MUTEX_ASSERT(&call->lock);
CV_BROADCAST(&call->cv_tq);
#else /* RX_ENABLE_LOCKS */
osi_rxWakeup(&call->tq);
/* remember start time for call in case we have hard dead time limit */
call->queueTime = queueTime;
clock_GetTime(&call->startTime);
- call->bytesSent = 0;
- call->bytesRcvd = 0;
+ call->app.bytesSent = 0;
+ call->app.bytesRcvd = 0;
/* Turn on busy protocol. */
rxi_KeepAliveOn(call);
* run (see code above that avoids resource starvation).
*/
#ifdef RX_ENABLE_LOCKS
+ if (call->flags & (RX_CALL_TQ_BUSY | RX_CALL_TQ_CLEARME)) {
+ osi_Panic("rx_NewCall call about to be used without an empty tq");
+ }
+
CV_BROADCAST(&conn->conn_call_cv);
#else
osi_rxWakeup(conn);
#endif
MUTEX_EXIT(&conn->conn_call_lock);
-
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
- if (call->flags & (RX_CALL_TQ_BUSY | RX_CALL_TQ_CLEARME)) {
- osi_Panic("rx_NewCall call about to be used without an empty tq");
- }
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
-
MUTEX_EXIT(&call->lock);
USERPRI;
tservice = rxi_AllocService();
NETPRI;
-#ifdef RX_ENABLE_LOCKS
MUTEX_INIT(&tservice->svc_data_lock, "svc data lock", MUTEX_DEFAULT, 0);
-#endif
for (i = 0; i < RX_MAX_SERVICES; i++) {
struct rx_service *service = rx_services[i];
opr_queue_Append(&rx_idleServerQueue, &sq->entry);
#ifndef AFS_AIX41_ENV
rx_waitForPacket = sq;
-#else
- rx_waitingForPacket = sq;
#endif /* AFS_AIX41_ENV */
do {
CV_WAIT(&sq->cv, &rx_serverPool_lock);
call->state = RX_STATE_DALLY;
rxi_ClearTransmitQueue(call, 0);
rxi_rto_cancel(call);
- rxevent_Cancel(&call->keepAliveEvent, call,
- RX_CALL_REFCOUNT_ALIVE);
+ rxi_CancelKeepAliveEvent(call);
}
} else { /* Client connection */
char dummy;
* and force-send it now.
*/
if (call->delayedAckEvent) {
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
rxi_SendDelayedAck(NULL, call, NULL, 0);
}
* Map errors to the local host's errno.h format.
*/
error = ntoh_syserr_conv(error);
+
+ /* If the caller said the call failed with some error, we had better
+ * return an error code. */
+ osi_Assert(!rc || error);
return error;
}
struct rx_connection **conn_ptr, **conn_end;
INIT_PTHREAD_LOCKS;
- LOCK_RX_INIT;
- if (rxinit_status == 1) {
- UNLOCK_RX_INIT;
+ if (rx_atomic_test_and_set_bit(&rxinit_status, 0))
return; /* Already shutdown. */
- }
+
rxi_DeleteCachedConnections();
if (rx_connHashTable) {
MUTEX_ENTER(&rx_connHashTable_lock);
afs_winsockCleanup();
#endif
- rxinit_status = 1;
- UNLOCK_RX_INIT;
}
#endif
rxi_NewCall(struct rx_connection *conn, int channel)
{
struct rx_call *call;
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
struct rx_call *cp; /* Call pointer temp */
struct opr_queue *cursor;
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif
dpf(("rxi_NewCall(conn %"AFS_PTR_FMT", channel %d)\n", conn, channel));
* rxi_FreeCall */
MUTEX_ENTER(&rx_freeCallQueue_lock);
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
/*
* EXCEPT that the TQ might not yet be cleared out.
* Skip over those with in-use TQs.
}
}
if (call) {
-#else /* AFS_GLOBAL_RXLOCK_KERNEL */
+#else /* RX_ENABLE_LOCKS */
if (!opr_queue_IsEmpty(&rx_freeCallQueue)) {
call = opr_queue_First(&rx_freeCallQueue, struct rx_call, entry);
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
opr_queue_Remove(&call->entry);
if (rx_stats_active)
rx_atomic_dec(&rx_stats.nFreeCallStructs);
MUTEX_EXIT(&rx_freeCallQueue_lock);
MUTEX_ENTER(&call->lock);
CLEAR_CALL_QUEUE_LOCK(call);
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
/* Now, if TQ wasn't cleared earlier, do it now. */
rxi_WaitforTQBusy(call);
if (call->flags & RX_CALL_TQ_CLEARME) {
rxi_ClearTransmitQueue(call, 1);
/*queue_Init(&call->tq);*/
}
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
/* Bind the call to its connection structure */
call->conn = conn;
rxi_ResetCall(call, 1);
MUTEX_ENTER(&rx_freeCallQueue_lock);
SET_CALL_QUEUE_LOCK(call, &rx_freeCallQueue_lock);
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
/* A call may be free even though its transmit queue is still in use.
* Since we search the call list from head to tail, put busy calls at
* the head of the list, and idle calls at the tail.
opr_queue_Prepend(&rx_freeCallQueue, &call->entry);
else
opr_queue_Append(&rx_freeCallQueue, &call->entry);
-#else /* AFS_GLOBAL_RXLOCK_KERNEL */
+#else /* RX_ENABLE_LOCKS */
opr_queue_Append(&rx_freeCallQueue, &call->entry);
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
if (rx_stats_active)
rx_atomic_inc(&rx_stats.nFreeCallStructs);
MUTEX_EXIT(&rx_freeCallQueue_lock);
#ifdef AFS_RXERRQ_ENV
static void
-rxi_SetPeerDead(afs_uint32 host, afs_uint16 port)
+rxi_SetPeerDead(struct sock_extended_err *err, afs_uint32 host, afs_uint16 port)
{
int hashIndex = PEER_HASH(host, port);
struct rx_peer *peer;
for (peer = rx_peerHashTable[hashIndex]; peer; peer = peer->next) {
if (peer->host == host && peer->port == port) {
+ peer->refCount++;
break;
}
}
+ MUTEX_EXIT(&rx_peerHashTable_lock);
+
if (peer) {
rx_atomic_inc(&peer->neterrs);
- }
+ MUTEX_ENTER(&peer->peer_lock);
+ peer->last_err_origin = RX_NETWORK_ERROR_ORIGIN_ICMP;
+ peer->last_err_type = err->ee_type;
+ peer->last_err_code = err->ee_code;
+ MUTEX_EXIT(&peer->peer_lock);
- MUTEX_EXIT(&rx_peerHashTable_lock);
+ MUTEX_ENTER(&rx_peerHashTable_lock);
+ peer->refCount--;
+ MUTEX_EXIT(&rx_peerHashTable_lock);
+ }
}
void
case ICMP_PORT_UNREACH:
case ICMP_NET_ANO:
case ICMP_HOST_ANO:
- rxi_SetPeerDead(addr, port);
+ rxi_SetPeerDead(err, addr, port);
break;
}
}
}
+
+static const char *
+rxi_TranslateICMP(int type, int code)
+{
+ switch (type) {
+ case ICMP_DEST_UNREACH:
+ switch (code) {
+ case ICMP_NET_UNREACH:
+ return "Destination Net Unreachable";
+ case ICMP_HOST_UNREACH:
+ return "Destination Host Unreachable";
+ case ICMP_PROT_UNREACH:
+ return "Destination Protocol Unreachable";
+ case ICMP_PORT_UNREACH:
+ return "Destination Port Unreachable";
+ case ICMP_NET_ANO:
+ return "Destination Net Prohibited";
+ case ICMP_HOST_ANO:
+ return "Destination Host Prohibited";
+ }
+ break;
+ }
+ return NULL;
+}
#endif /* AFS_RXERRQ_ENV */
+/**
+ * Get the last network error for a connection
+ *
+ * A "network error" here means an error retrieved from ICMP, or some other
+ * mechanism outside of Rx that informs us of errors in network reachability.
+ *
+ * If a peer associated with the given Rx connection has received a network
+ * error recently, this function allows the caller to know what error
+ * specifically occurred. This can be useful to know, since e.g. ICMP errors
+ * can cause calls to that peer to be quickly aborted. So, this function can
+ * help see why a call was aborted due to network errors.
+ *
+ * If we have received traffic from a peer since the last network error, we
+ * treat that peer as if we had not received an network error for it.
+ *
+ * @param[in] conn The Rx connection to examine
+ * @param[out] err_origin The origin of the last network error (e.g. ICMP);
+ * one of the RX_NETWORK_ERROR_ORIGIN_* constants
+ * @param[out] err_type The type of the last error
+ * @param[out] err_code The code of the last error
+ * @param[out] msg Human-readable error message, if applicable; NULL otherwise
+ *
+ * @return If we have an error
+ * @retval -1 No error to get; 'out' params are undefined
+ * @retval 0 We have an error; 'out' params contain the last error
+ */
+int
+rx_GetNetworkError(struct rx_connection *conn, int *err_origin, int *err_type,
+ int *err_code, const char **msg)
+{
+#ifdef AFS_RXERRQ_ENV
+ struct rx_peer *peer = conn->peer;
+ if (rx_atomic_read(&peer->neterrs)) {
+ MUTEX_ENTER(&peer->peer_lock);
+ *err_origin = peer->last_err_origin;
+ *err_type = peer->last_err_type;
+ *err_code = peer->last_err_code;
+ MUTEX_EXIT(&peer->peer_lock);
+
+ *msg = NULL;
+ if (*err_origin == RX_NETWORK_ERROR_ORIGIN_ICMP) {
+ *msg = rxi_TranslateICMP(*err_type, *err_code);
+ }
+
+ return 0;
+ }
+#endif
+ return -1;
+}
+
/* Find the peer process represented by the supplied (host,port)
* combination. If there is no appropriate active peer structure, a
* new one will be allocated and initialized
- * The origPeer, if set, is a pointer to a peer structure on which the
- * refcount will be be decremented. This is used to replace the peer
- * structure hanging off a connection structure */
+ */
struct rx_peer *
-rxi_FindPeer(afs_uint32 host, u_short port,
- struct rx_peer *origPeer, int create)
+rxi_FindPeer(afs_uint32 host, u_short port, int create)
{
struct rx_peer *pp;
int hashIndex;
if (pp && create) {
pp->refCount++;
}
- if (origPeer)
- origPeer->refCount--;
MUTEX_EXIT(&rx_peerHashTable_lock);
return pp;
}
static struct rx_connection *
rxi_FindConnection(osi_socket socket, afs_uint32 host,
u_short port, u_short serviceId, afs_uint32 cid,
- afs_uint32 epoch, int type, u_int securityIndex)
+ afs_uint32 epoch, int type, u_int securityIndex,
+ int *unknownService)
{
int hashindex, flag, i;
struct rx_connection *conn;
+ *unknownService = 0;
hashindex = CONN_HASH(host, port, cid, epoch, type);
MUTEX_ENTER(&rx_connHashTable_lock);
rxLastConn ? (conn = rxLastConn, flag = 0) : (conn =
if (!service || (securityIndex >= service->nSecurityObjects)
|| (service->securityObjects[securityIndex] == 0)) {
MUTEX_EXIT(&rx_connHashTable_lock);
+ *unknownService = 1;
return (struct rx_connection *)0;
}
conn = rxi_AllocConnection(); /* This bzero's the connection */
CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
conn->next = rx_connHashTable[hashindex];
rx_connHashTable[hashindex] = conn;
- conn->peer = rxi_FindPeer(host, port, 0, 1);
+ conn->peer = rxi_FindPeer(host, port, 1);
conn->type = RX_SERVER_CONNECTION;
conn->lastSendTime = clock_Sec(); /* don't GC immediately */
conn->epoch = epoch;
int channel = call->channel;
int freechannel = 0;
int i;
- afs_uint32 callNumber;
MUTEX_EXIT(&call->lock);
MUTEX_ENTER(&conn->conn_call_lock);
- callNumber = *call->callNumber;
/* 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
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
+ /* Since the call->lock has been released it is possible that the call may
+ * no longer be busy (the call channel cannot have been reallocated as we
+ * haven't dropped the conn_call_lock) 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)) {
+ if (freechannel && (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
MUTEX_EXIT(&conn->conn_call_lock);
}
+/*!
+ * Abort the call if the server is over the busy threshold. This
+ * can be used without requiring a call structure be initialised,
+ * or connected to a particular channel
+ */
+static_inline int
+rxi_AbortIfServerBusy(osi_socket socket, struct rx_connection *conn,
+ struct rx_packet *np)
+{
+ if ((rx_BusyThreshold > 0) &&
+ (rx_atomic_read(&rx_nWaiting) > rx_BusyThreshold)) {
+ rxi_SendRawAbort(socket, conn->peer->host, conn->peer->port,
+ rx_BusyError, np, 0);
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.nBusies);
+ return 1;
+ }
+
+ return 0;
+}
+
+static_inline struct rx_call *
+rxi_ReceiveClientCall(struct rx_packet *np, struct rx_connection *conn)
+{
+ int channel;
+ struct rx_call *call;
+
+ channel = np->header.cid & RX_CHANNELMASK;
+ MUTEX_ENTER(&conn->conn_call_lock);
+ call = conn->call[channel];
+ if (!call || conn->callNumber[channel] != np->header.callNumber) {
+ MUTEX_EXIT(&conn->conn_call_lock);
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.spuriousPacketsRead);
+ return NULL;
+ }
+
+ MUTEX_ENTER(&call->lock);
+ MUTEX_EXIT(&conn->conn_call_lock);
+
+ if ((call->state == RX_STATE_DALLY)
+ && np->header.type == RX_PACKET_TYPE_ACK) {
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.ignorePacketDally);
+ MUTEX_EXIT(&call->lock);
+ return NULL;
+ }
+
+ return call;
+}
+
+static_inline struct rx_call *
+rxi_ReceiveServerCall(osi_socket socket, struct rx_packet *np,
+ struct rx_connection *conn)
+{
+ int channel;
+ struct rx_call *call;
+
+ channel = np->header.cid & RX_CHANNELMASK;
+ MUTEX_ENTER(&conn->conn_call_lock);
+ call = conn->call[channel];
+
+ if (!call) {
+ if (rxi_AbortIfServerBusy(socket, conn, np)) {
+ MUTEX_EXIT(&conn->conn_call_lock);
+ return NULL;
+ }
+
+ call = rxi_NewCall(conn, channel); /* returns locked call */
+ *call->callNumber = np->header.callNumber;
+ MUTEX_EXIT(&conn->conn_call_lock);
+
+ call->state = RX_STATE_PRECALL;
+ clock_GetTime(&call->queueTime);
+ call->app.bytesSent = 0;
+ call->app.bytesRcvd = 0;
+ rxi_KeepAliveOn(call);
+
+ return call;
+ }
+
+ if (np->header.callNumber == conn->callNumber[channel]) {
+ MUTEX_ENTER(&call->lock);
+ MUTEX_EXIT(&conn->conn_call_lock);
+ return call;
+ }
+
+ if (np->header.callNumber < conn->callNumber[channel]) {
+ MUTEX_EXIT(&conn->conn_call_lock);
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.spuriousPacketsRead);
+ return NULL;
+ }
+
+ MUTEX_ENTER(&call->lock);
+ MUTEX_EXIT(&conn->conn_call_lock);
+
+ /* Wait until the transmit queue is idle before deciding
+ * whether to reset the current call. Chances are that the
+ * call will be in ether DALLY or HOLD state once the TQ_BUSY
+ * flag is cleared.
+ */
+#ifdef RX_ENABLE_LOCKS
+ if (call->state == RX_STATE_ACTIVE && !call->error) {
+ rxi_WaitforTQBusy(call);
+ /* If we entered error state while waiting,
+ * must call rxi_CallError to permit rxi_ResetCall
+ * to processed when the tqWaiter count hits zero.
+ */
+ if (call->error) {
+ rxi_CallError(call, call->error);
+ MUTEX_EXIT(&call->lock);
+ return NULL;
+ }
+ }
+#endif /* RX_ENABLE_LOCKS */
+ /* If the new call cannot be taken right now send a busy and set
+ * the error condition in this call, so that it terminates as
+ * quickly as possible */
+ if (call->state == RX_STATE_ACTIVE) {
+ rxi_CallError(call, RX_CALL_DEAD);
+ rxi_SendSpecial(call, conn, NULL, RX_PACKET_TYPE_BUSY,
+ NULL, 0, 1);
+ MUTEX_EXIT(&call->lock);
+ return NULL;
+ }
+
+ if (rxi_AbortIfServerBusy(socket, conn, np)) {
+ MUTEX_EXIT(&call->lock);
+ return NULL;
+ }
+
+ rxi_ResetCall(call, 0);
+ /* The conn_call_lock is not held but no one else should be
+ * using this call channel while we are processing this incoming
+ * packet. This assignment should be safe.
+ */
+ *call->callNumber = np->header.callNumber;
+ call->state = RX_STATE_PRECALL;
+ clock_GetTime(&call->queueTime);
+ call->app.bytesSent = 0;
+ call->app.bytesRcvd = 0;
+ rxi_KeepAliveOn(call);
+
+ return call;
+}
+
+
/* 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
{
struct rx_call *call;
struct rx_connection *conn;
- int channel;
- afs_uint32 currentCallNumber;
int type;
+ int unknownService = 0;
#ifdef RXDEBUG
char *packetType;
#endif
struct rx_peer *peer;
/* Try to look up the peer structure, but don't create one */
- peer = rxi_FindPeer(host, port, 0, 0);
+ peer = rxi_FindPeer(host, port, 0);
/* Since this may not be associated with a connection, it may have
* no refCount, meaning we could race with ReapConnections
conn =
rxi_FindConnection(socket, host, port, np->header.serviceId,
np->header.cid, np->header.epoch, type,
- np->header.securityIndex);
+ np->header.securityIndex, &unknownService);
/* To avoid having 2 connections just abort at each other,
don't abort an abort. */
if (!conn) {
- if (np->header.type != RX_PACKET_TYPE_ABORT)
+ if (unknownService && (np->header.type != RX_PACKET_TYPE_ABORT))
rxi_SendRawAbort(socket, host, port, RX_INVALID_OPERATION,
np, 0);
return np;
}
}
- channel = np->header.cid & RX_CHANNELMASK;
- MUTEX_ENTER(&conn->conn_call_lock);
- call = conn->call[channel];
+ if (type == RX_SERVER_CONNECTION)
+ call = rxi_ReceiveServerCall(socket, np, conn);
+ else
+ call = rxi_ReceiveClientCall(np, conn);
- if (call) {
- MUTEX_ENTER(&call->lock);
- currentCallNumber = conn->callNumber[channel];
- MUTEX_EXIT(&conn->conn_call_lock);
- } else if (type == RX_SERVER_CONNECTION) { /* No call allocated */
- call = conn->call[channel];
- if (call) {
- MUTEX_ENTER(&call->lock);
- currentCallNumber = conn->callNumber[channel];
- MUTEX_EXIT(&conn->conn_call_lock);
- } else {
- call = rxi_NewCall(conn, channel); /* returns locked call */
- *call->callNumber = currentCallNumber = np->header.callNumber;
- MUTEX_EXIT(&conn->conn_call_lock);
-#ifdef RXDEBUG
- if (np->header.callNumber == 0)
- dpf(("RecPacket call 0 %d %s: %x.%u.%u.%u.%u.%u.%u flags %d, packet %"AFS_PTR_FMT" len %d\n",
- np->header.serial, rx_packetTypes[np->header.type - 1], ntohl(conn->peer->host), ntohs(conn->peer->port),
- np->header.serial, np->header.epoch, np->header.cid, np->header.callNumber, np->header.seq,
- np->header.flags, np, np->length));
-#endif
- call->state = RX_STATE_PRECALL;
- clock_GetTime(&call->queueTime);
- call->bytesSent = 0;
- call->bytesRcvd = 0;
- /*
- * If the number of queued calls exceeds the overload
- * threshold then abort this call.
- */
- if ((rx_BusyThreshold > 0) &&
- (rx_atomic_read(&rx_nWaiting) > rx_BusyThreshold)) {
- struct rx_packet *tp;
-
- rxi_CallError(call, rx_BusyError);
- tp = rxi_SendCallAbort(call, np, 1, 0);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.nBusies);
- return tp;
- }
- rxi_KeepAliveOn(call);
- }
- } else { /* RX_CLIENT_CONNECTION and No call allocated */
- /* This packet can't be for this call. If the new call address is
- * 0 then no call is running on this channel. If there is a call
- * then, since this is a client connection we're getting data for
- * it must be for the previous call.
- */
- MUTEX_EXIT(&conn->conn_call_lock);
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.spuriousPacketsRead);
+ if (call == NULL) {
putConnection(conn);
- return np;
- }
-
- /* There is a non-NULL locked call at this point */
- if (type == RX_SERVER_CONNECTION) { /* We're the server */
- if (np->header.callNumber < currentCallNumber) {
- MUTEX_EXIT(&call->lock);
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.spuriousPacketsRead);
- putConnection(conn);
- return np;
- } else if (np->header.callNumber != currentCallNumber) {
- /* Wait until the transmit queue is idle before deciding
- * whether to reset the current call. Chances are that the
- * call will be in ether DALLY or HOLD state once the TQ_BUSY
- * flag is cleared.
- */
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
- if (call->state == RX_STATE_ACTIVE) {
- rxi_WaitforTQBusy(call);
- /*
- * If we entered error state while waiting,
- * must call rxi_CallError to permit rxi_ResetCall
- * to processed when the tqWaiter count hits zero.
- */
- if (call->error) {
- rxi_CallError(call, call->error);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np;
- }
- }
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
- /* If the new call cannot be taken right now send a busy and set
- * the error condition in this call, so that it terminates as
- * quickly as possible */
- if (call->state == RX_STATE_ACTIVE) {
- struct rx_packet *tp;
-
- rxi_CallError(call, RX_CALL_DEAD);
- tp = rxi_SendSpecial(call, conn, np, RX_PACKET_TYPE_BUSY,
- NULL, 0, 1);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return tp;
- }
- rxi_ResetCall(call, 0);
- /*
- * The conn_call_lock is not held but no one else should be
- * using this call channel while we are processing this incoming
- * packet. This assignment should be safe.
- */
- *call->callNumber = np->header.callNumber;
-#ifdef RXDEBUG
- if (np->header.callNumber == 0)
- dpf(("RecPacket call 0 %d %s: %x.%u.%u.%u.%u.%u.%u flags %d, packet %"AFS_PTR_FMT" len %d\n",
- np->header.serial, rx_packetTypes[np->header.type - 1], ntohl(conn->peer->host), ntohs(conn->peer->port),
- np->header.serial, np->header.epoch, np->header.cid, np->header.callNumber, np->header.seq,
- np->header.flags, np, np->length));
-#endif
- call->state = RX_STATE_PRECALL;
- clock_GetTime(&call->queueTime);
- call->bytesSent = 0;
- call->bytesRcvd = 0;
- /*
- * If the number of queued calls exceeds the overload
- * threshold then abort this call.
- */
- if ((rx_BusyThreshold > 0) &&
- (rx_atomic_read(&rx_nWaiting) > rx_BusyThreshold)) {
- struct rx_packet *tp;
-
- rxi_CallError(call, rx_BusyError);
- tp = rxi_SendCallAbort(call, np, 1, 0);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.nBusies);
- return tp;
- }
- rxi_KeepAliveOn(call);
- } else {
- /* Continuing call; do nothing here. */
- }
- } else { /* we're the client */
- /* Ignore all incoming acknowledgements for calls in DALLY state */
- if ((call->state == RX_STATE_DALLY)
- && (np->header.type == RX_PACKET_TYPE_ACK)) {
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.ignorePacketDally);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np;
- }
-
- /* Ignore anything that's not relevant to the current call. If there
- * isn't a current call, then no packet is relevant. */
- if (np->header.callNumber != currentCallNumber) {
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.spuriousPacketsRead);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np;
- }
- /* If the service security object index stamped in the packet does not
- * match the connection's security index, ignore the packet */
- if (np->header.securityIndex != conn->securityIndex) {
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np;
- }
-
- /* If we're receiving the response, then all transmit packets are
- * implicitly acknowledged. Get rid of them. */
- if (np->header.type == RX_PACKET_TYPE_DATA) {
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
- /* XXX Hack. Because we must release the global rx lock when
- * sending packets (osi_NetSend) we drop all acks while we're
- * traversing the tq in rxi_Start sending packets out because
- * packets may move to the freePacketQueue as result of being here!
- * So we drop these packets until we're safely out of the
- * traversing. Really ugly!
- * For fine grain RX locking, we set the acked field in the
- * packets and let rxi_Start remove them from the transmit queue.
- */
- if (call->flags & RX_CALL_TQ_BUSY) {
-#ifdef RX_ENABLE_LOCKS
- rxi_SetAcksInTransmitQueue(call);
-#else
- putConnection(conn);
- return np; /* xmitting; drop packet */
-#endif
- } else {
- rxi_ClearTransmitQueue(call, 0);
- }
-#else /* AFS_GLOBAL_RXLOCK_KERNEL */
- rxi_ClearTransmitQueue(call, 0);
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
- } else {
- if (np->header.type == RX_PACKET_TYPE_ACK) {
- /* now check to see if this is an ack packet acknowledging that the
- * server actually *lost* some hard-acked data. If this happens we
- * ignore this packet, as it may indicate that the server restarted in
- * the middle of a call. It is also possible that this is an old ack
- * packet. We don't abort the connection in this case, because this
- * *might* just be an old ack packet. The right way to detect a server
- * restart in the midst of a call is to notice that the server epoch
- * changed, btw. */
- /* XXX I'm not sure this is exactly right, since tfirst **IS**
- * XXX unacknowledged. I think that this is off-by-one, but
- * XXX I don't dare change it just yet, since it will
- * XXX interact badly with the server-restart detection
- * XXX code in receiveackpacket. */
- if (ntohl(rx_GetInt32(np, FIRSTACKOFFSET)) < call->tfirst) {
- if (rx_stats_active)
- rx_atomic_inc(&rx_stats.spuriousPacketsRead);
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np;
- }
- }
- } /* else not a data packet */
+ return np;
}
- osirx_AssertMine(&call->lock, "rxi_ReceivePacket middle");
+ MUTEX_ASSERT(&call->lock);
/* Set remote user defined status from packet */
call->remoteStatus = np->header.userStatus;
/* Now do packet type-specific processing */
switch (np->header.type) {
case RX_PACKET_TYPE_DATA:
+ /* If we're a client, and receiving a response, then all the packets
+ * we transmitted packets are implicitly acknowledged. */
+ if (type == RX_CLIENT_CONNECTION && !opr_queue_IsEmpty(&call->tq))
+ rxi_AckAllInTransmitQueue(call);
+
np = rxi_ReceiveDataPacket(call, np, 1, socket, host, port, tnop,
newcallp);
break;
case RX_PACKET_TYPE_ACKALL:
/* All packets acknowledged, so we can drop all packets previously
* readied for sending */
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
- /* XXX Hack. We because we can't release the global rx lock when
- * sending packets (osi_NetSend) we drop all ack pkts while we're
- * traversing the tq in rxi_Start sending packets out because
- * packets may move to the freePacketQueue as result of being
- * here! So we drop these packets until we're safely out of the
- * traversing. Really ugly!
- * For fine grain RX locking, we set the acked field in the packets
- * and let rxi_Start remove the packets from the transmit queue.
- */
- if (call->flags & RX_CALL_TQ_BUSY) {
-#ifdef RX_ENABLE_LOCKS
- rxi_SetAcksInTransmitQueue(call);
- break;
-#else /* RX_ENABLE_LOCKS */
- MUTEX_EXIT(&call->lock);
- putConnection(conn);
- return np; /* xmitting; drop packet */
-#endif /* RX_ENABLE_LOCKS */
- }
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
- rxi_ClearTransmitQueue(call, 0);
+ rxi_AckAllInTransmitQueue(call);
break;
default:
/* Should not reach here, unless the peer is broken: send an abort
MUTEX_ENTER(&conn->conn_data_lock);
- if (event) {
- rxevent_Put(conn->checkReachEvent);
- conn->checkReachEvent = NULL;
- }
+ if (event)
+ rxevent_Put(&conn->checkReachEvent);
waiting = conn->flags & RX_CONN_ATTACHWAIT;
if (event) {
if (rx_stats_active)
rx_atomic_inc(&rx_stats.dupPacketsRead);
dpf(("packet %"AFS_PTR_FMT" dropped on receipt - duplicate\n", np));
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
ackNeeded = 0;
call->rprev = seq;
if (seq < call->rnext) {
if (rx_stats_active)
rx_atomic_inc(&rx_stats.dupPacketsRead);
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
ackNeeded = 0;
call->rprev = seq;
* accomodated by the current window, then send a negative
* acknowledge and drop the packet */
if ((call->rnext + call->rwind) <= seq) {
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, RX_ACK_EXCEEDS_WINDOW,
istack);
ackNeeded = 0;
if (seq == tp->header.seq) {
if (rx_stats_active)
rx_atomic_inc(&rx_stats.dupPacketsRead);
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE,
istack);
ackNeeded = 0;
* received. Always send a soft ack for the last packet in
* the server's reply. */
if (ackNeeded) {
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, ackNeeded, istack);
} else if (call->nSoftAcks > (u_short) rxi_SoftAckRate) {
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
np = rxi_SendAck(call, np, serial, RX_ACK_IDLE, istack);
} else if (call->nSoftAcks) {
if (haveLast && !(flags & RX_CLIENT_INITIATED))
else
rxi_PostDelayedAckEvent(call, &rx_softAckDelay);
} else if (call->flags & RX_CALL_RECEIVE_DONE) {
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
}
return np;
rxi_ComputeRoundTripTime(tp, ap, call, peer, &now);
}
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
- /* XXX Hack. Because we have to release the global rx lock when sending
+#ifdef RX_ENABLE_LOCKS
+ /* XXX Hack. Because we have to release the global call lock when sending
* packets (osi_NetSend) we drop all acks while we're traversing the tq
* in rxi_Start sending packets out because packets may move to the
* freePacketQueue as result of being here! So we drop these packets until
* when it's done transmitting.
*/
if (call->flags & RX_CALL_TQ_BUSY) {
-#ifdef RX_ENABLE_LOCKS
tp->flags |= RX_PKTFLAG_ACKED;
call->flags |= RX_CALL_TQ_SOME_ACKED;
-#else /* RX_ENABLE_LOCKS */
- break;
-#endif /* RX_ENABLE_LOCKS */
} else
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
{
opr_queue_Remove(&tp->entry);
#ifdef RX_TRACK_PACKETS
&& call->tfirst + call->nSoftAcked >= call->tnext) {
call->state = RX_STATE_DALLY;
rxi_ClearTransmitQueue(call, 0);
- rxevent_Cancel(&call->keepAliveEvent, call, RX_CALL_REFCOUNT_ALIVE);
+ rxi_CancelKeepAliveEvent(call);
} else if (!opr_queue_IsEmpty(&call->tq)) {
rxi_Start(call, istack);
}
return np;
}
+/**
+ * Schedule a connection abort to be sent after some delay.
+ *
+ * @param[in] conn The connection to send the abort on.
+ * @param[in] msec The number of milliseconds to wait before sending.
+ *
+ * @pre conn_data_lock must be held
+ */
+static void
+rxi_SendConnectionAbortLater(struct rx_connection *conn, int msec)
+{
+ struct clock when, now;
+ if (!conn->error) {
+ return;
+ }
+ if (!conn->delayedAbortEvent) {
+ clock_GetTime(&now);
+ when = now;
+ clock_Addmsec(&when, msec);
+ conn->delayedAbortEvent =
+ rxevent_Post(&when, &now, rxi_SendDelayedConnAbort, conn, NULL, 0);
+ }
+}
+
/* Received a response to a challenge packet */
static struct rx_packet *
rxi_ReceiveResponsePacket(struct rx_connection *conn,
error = RXS_CheckResponse(conn->securityObject, conn, np);
if (error) {
/* If the response is invalid, reset the connection, sending
- * an abort to the peer */
-#ifndef KERNEL
- rxi_Delay(1);
-#endif
+ * an abort to the peer. Send the abort with a 1 second delay,
+ * to avoid a peer hammering us by constantly recreating a
+ * connection with bad credentials. */
rxi_ConnectionError(conn, error);
MUTEX_ENTER(&conn->conn_data_lock);
- np = rxi_SendConnectionAbort(conn, np, istack, 0);
+ rxi_SendConnectionAbortLater(conn, 1000);
MUTEX_EXIT(&conn->conn_data_lock);
return np;
} else {
#ifdef RX_ENABLE_LOCKS
if (event) {
MUTEX_ENTER(&call->lock);
- if (event == call->delayedAckEvent) {
- rxevent_Put(call->delayedAckEvent);
- call->delayedAckEvent = NULL;
- }
+ if (event == call->delayedAckEvent)
+ rxevent_Put(&call->delayedAckEvent);
CALL_RELE(call, RX_CALL_REFCOUNT_DELAY);
}
(void)rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
if (event)
MUTEX_EXIT(&call->lock);
#else /* RX_ENABLE_LOCKS */
- if (event) {
- rxevent_Put(call->delayedAckEvent);
- call->delayedAckEvent = NULL;
- }
+ if (event)
+ rxevent_Put(&call->delayedAckEvent);
(void)rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
#endif /* RX_ENABLE_LOCKS */
}
-
#ifdef RX_ENABLE_LOCKS
/* Set ack in all packets in transmit queue. rxi_Start will deal with
* clearing them out.
}
#endif /* RX_ENABLE_LOCKS */
+/*!
+ * Acknowledge the whole transmit queue.
+ *
+ * If we're running without locks, or the transmit queue isn't busy, then
+ * we can just clear the queue now. Otherwise, we have to mark all of the
+ * packets as acknowledged, and let rxi_Start clear it later on
+ */
+static void
+rxi_AckAllInTransmitQueue(struct rx_call *call)
+{
+#ifdef RX_ENABLE_LOCKS
+ if (call->flags & RX_CALL_TQ_BUSY) {
+ rxi_SetAcksInTransmitQueue(call);
+ return;
+ }
+#endif
+ rxi_ClearTransmitQueue(call, 0);
+}
/* Clear out the transmit queue for the current call (all packets have
* been received by peer) */
static void
rxi_ClearTransmitQueue(struct rx_call *call, int force)
{
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
struct opr_queue *cursor;
if (!force && (call->flags & RX_CALL_TQ_BUSY)) {
int someAcked = 0;
call->flags |= RX_CALL_TQ_SOME_ACKED;
}
} else {
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
#ifdef RXDEBUG_PACKET
call->tqc -=
#endif /* RXDEBUG_PACKET */
rxi_FreePackets(0, &call->tq);
rxi_WakeUpTransmitQueue(call);
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
call->flags &= ~RX_CALL_TQ_CLEARME;
}
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif
rxi_rto_cancel(call);
call->tfirst = call->tnext; /* implicitly acknowledge all data already sent */
if (force || rxi_callAbortThreshhold == 0
|| call->abortCount < rxi_callAbortThreshhold) {
- if (call->delayedAbortEvent) {
- rxevent_Cancel(&call->delayedAbortEvent, call,
- RX_CALL_REFCOUNT_ABORT);
- }
+ rxi_CancelDelayedAbortEvent(call);
error = htonl(cerror);
call->abortCount++;
packet =
return packet;
}
+static void
+rxi_CancelDelayedAbortEvent(struct rx_call *call)
+{
+ if (call->delayedAbortEvent) {
+ rxevent_Cancel(&call->delayedAbortEvent);
+ CALL_RELE(call, RX_CALL_REFCOUNT_ABORT);
+ }
+}
+
/* Send an abort packet for the specified connection. Packet is an
* optional pointer to a packet that can be used to send the abort.
* Once the number of abort messages reaches the threshhold, an
struct rx_packet *packet, int istack, int force)
{
afs_int32 error;
- struct clock when, now;
if (!conn->error)
return packet;
if (force || rxi_connAbortThreshhold == 0
|| conn->abortCount < rxi_connAbortThreshhold) {
- rxevent_Cancel(&conn->delayedAbortEvent, NULL, 0);
+ rxevent_Cancel(&conn->delayedAbortEvent);
error = htonl(conn->error);
conn->abortCount++;
MUTEX_EXIT(&conn->conn_data_lock);
RX_PACKET_TYPE_ABORT, (char *)&error,
sizeof(error), istack);
MUTEX_ENTER(&conn->conn_data_lock);
- } else if (!conn->delayedAbortEvent) {
- clock_GetTime(&now);
- when = now;
- clock_Addmsec(&when, rxi_connAbortDelay);
- conn->delayedAbortEvent =
- rxevent_Post(&when, &now, rxi_SendDelayedConnAbort, conn, NULL, 0);
+ } else {
+ rxi_SendConnectionAbortLater(conn, rxi_connAbortDelay);
}
return packet;
}
dpf(("rxi_ConnectionError conn %"AFS_PTR_FMT" error %d\n", conn, error));
MUTEX_ENTER(&conn->conn_data_lock);
- rxevent_Cancel(&conn->challengeEvent, NULL, 0);
- rxevent_Cancel(&conn->natKeepAliveEvent, NULL, 0);
+ rxevent_Cancel(&conn->challengeEvent);
+ rxevent_Cancel(&conn->natKeepAliveEvent);
if (conn->checkReachEvent) {
- rxevent_Cancel(&conn->checkReachEvent, NULL, 0);
+ rxevent_Cancel(&conn->checkReachEvent);
conn->flags &= ~(RX_CONN_ATTACHWAIT|RX_CONN_NAT_PING);
putConnection(conn);
}
void
rxi_CallError(struct rx_call *call, afs_int32 error)
{
-#ifdef DEBUG
- osirx_AssertMine(&call->lock, "rxi_CallError");
-#endif
+ MUTEX_ASSERT(&call->lock);
dpf(("rxi_CallError call %"AFS_PTR_FMT" error %d call->error %d\n", call, error, call->error));
if (call->error)
error = call->error;
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
if (!((call->flags & RX_CALL_TQ_BUSY) || (call->tqWaiters > 0))) {
rxi_ResetCall(call, 0);
}
int flags;
struct rx_peer *peer;
struct rx_packet *packet;
-#ifdef DEBUG
- osirx_AssertMine(&call->lock, "rxi_ResetCall");
-#endif
+
+ MUTEX_ASSERT(&call->lock);
dpf(("rxi_ResetCall(call %"AFS_PTR_FMT", newcall %d)\n", call, newcall));
/* Notify anyone who is waiting for asynchronous packet arrival */
}
- rxevent_Cancel(&call->growMTUEvent, call, RX_CALL_REFCOUNT_MTU);
+ rxi_CancelGrowMTUEvent(call);
if (call->delayedAbortEvent) {
- rxevent_Cancel(&call->delayedAbortEvent, call, RX_CALL_REFCOUNT_ABORT);
+ rxi_CancelDelayedAbortEvent(call);
packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
if (packet) {
rxi_SendCallAbort(call, packet, 0, 1);
MUTEX_EXIT(&peer->peer_lock);
flags = call->flags;
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
rxi_WaitforTQBusy(call);
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
rxi_ClearTransmitQueue(call, 1);
if (call->tqWaiters || (flags & RX_CALL_TQ_WAIT)) {
}
#endif /* RX_ENABLE_LOCKS */
- rxi_KeepAliveOff(call);
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelKeepAliveEvent(call);
+ rxi_CancelDelayedAckEvent(call);
}
/* Send an acknowledge for the indicated packet (seq,serial) of the
/* Since we're about to send a data packet to the peer, it's
* safe to nuke any scheduled end-of-packets ack */
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
MUTEX_EXIT(&call->lock);
CALL_HOLD(call, RX_CALL_REFCOUNT_SEND);
* the listener or event threads
*/
if ((list[len - 1]->header.flags & RX_LAST_PACKET)
- || call->app.mode == RX_MODE_RECEIVING || call->app.mode == RX_MODE_EOF
+ || (call->flags & RX_CALL_FLUSH)
|| (call->flags & RX_CALL_FAST_RECOVER)) {
/* Check for the case where the current list contains
* an acked packet. Since we always send retransmissions
* event pending. */
if (event == call->resendEvent) {
CALL_RELE(call, RX_CALL_REFCOUNT_RESEND);
- rxevent_Put(call->resendEvent);
- call->resendEvent = NULL;
+ rxevent_Put(&call->resendEvent);
}
rxi_CheckPeerDead(call);
int maxXmitPackets;
if (call->error) {
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
if (rx_stats_active)
rx_atomic_inc(&rx_tq_debug.rxi_start_in_error);
#endif
* But check whether we're here recursively, and let the other guy
* do the work.
*/
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
if (!(call->flags & RX_CALL_TQ_BUSY)) {
call->flags |= RX_CALL_TQ_BUSY;
do {
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
restart:
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
call->flags &= ~RX_CALL_NEED_START;
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
nXmitPackets = 0;
maxXmitPackets = MIN(call->twind, call->cwind);
for (opr_queue_Scan(&call->tq, cursor)) {
istack);
}
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
if (call->error) {
/* We went into the error state while sending packets. Now is
* the time to reset the call. This will also inform the using
rxi_CallError(call, call->error);
return;
}
-#ifdef RX_ENABLE_LOCKS
+
if (call->flags & RX_CALL_TQ_SOME_ACKED) {
int missing;
call->flags &= ~RX_CALL_TQ_SOME_ACKED;
if (!missing)
call->flags |= RX_CALL_TQ_CLEARME;
}
-#endif /* RX_ENABLE_LOCKS */
if (call->flags & RX_CALL_TQ_CLEARME)
rxi_ClearTransmitQueue(call, 1);
} while (call->flags & RX_CALL_NEED_START);
/*
* TQ references no longer protected by this flag; they must remain
- * protected by the global lock.
+ * protected by the call lock.
*/
call->flags &= ~RX_CALL_TQ_BUSY;
rxi_WakeUpTransmitQueue(call);
} else {
call->flags |= RX_CALL_NEED_START;
}
-#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
+#endif /* RX_ENABLE_LOCKS */
} else {
rxi_rto_cancel(call);
}
/* Since we're about to send SOME sort of packet to the peer, it's
* safe to nuke any scheduled end-of-packets ack */
- rxevent_Cancel(&call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
/* Actually send the packet, filling in more connection-specific fields */
MUTEX_EXIT(&call->lock);
* may be freed!
* haveCTLock Set if calling from rxi_ReapConnections
*/
-#ifdef RX_ENABLE_LOCKS
static int
rxi_CheckCall(struct rx_call *call, int haveCTLock)
-#else /* RX_ENABLE_LOCKS */
-static int
-rxi_CheckCall(struct rx_call *call)
-#endif /* RX_ENABLE_LOCKS */
{
struct rx_connection *conn = call->conn;
afs_uint32 now;
return -1;
}
-#ifdef AFS_GLOBAL_RXLOCK_KERNEL
+#ifdef RX_ENABLE_LOCKS
if (call->flags & RX_CALL_TQ_BUSY) {
/* Call is active and will be reset by rxi_Start if it's
* in an error state.
} else {
#ifdef RX_ENABLE_LOCKS
/* Cancel pending events */
- rxevent_Cancel(&call->delayedAckEvent, call,
- RX_CALL_REFCOUNT_DELAY);
+ rxi_CancelDelayedAckEvent(call);
rxi_rto_cancel(call);
- rxevent_Cancel(&call->keepAliveEvent, call,
- RX_CALL_REFCOUNT_ALIVE);
- rxevent_Cancel(&call->growMTUEvent, call,
- RX_CALL_REFCOUNT_MTU);
+ rxi_CancelKeepAliveEvent(call);
+ rxi_CancelGrowMTUEvent(call);
MUTEX_ENTER(&rx_refcnt_mutex);
/* if rxi_FreeCall returns 1 it has freed the call */
if (call->refCount == 0 &&
MUTEX_ENTER(&rx_refcnt_mutex);
/* Only reschedule ourselves if the connection would not be destroyed */
if (conn->refCount <= 1) {
- rxevent_Put(conn->natKeepAliveEvent);
- conn->natKeepAliveEvent = NULL;
+ rxevent_Put(&conn->natKeepAliveEvent);
MUTEX_EXIT(&rx_refcnt_mutex);
MUTEX_EXIT(&conn->conn_data_lock);
rx_DestroyConnection(conn); /* drop the reference for this */
} else {
conn->refCount--; /* drop the reference for this */
MUTEX_EXIT(&rx_refcnt_mutex);
- rxevent_Put(conn->natKeepAliveEvent);
- conn->natKeepAliveEvent = NULL;
+ rxevent_Put(&conn->natKeepAliveEvent);
rxi_ScheduleNatKeepAliveEvent(conn);
MUTEX_EXIT(&conn->conn_data_lock);
}
CALL_RELE(call, RX_CALL_REFCOUNT_ALIVE);
MUTEX_ENTER(&call->lock);
- if (event == call->keepAliveEvent) {
- rxevent_Put(call->keepAliveEvent);
- call->keepAliveEvent = NULL;
- }
+ if (event == call->keepAliveEvent)
+ rxevent_Put(&call->keepAliveEvent);
now = clock_Sec();
-#ifdef RX_ENABLE_LOCKS
if (rxi_CheckCall(call, 0)) {
MUTEX_EXIT(&call->lock);
return;
}
-#else /* RX_ENABLE_LOCKS */
- if (rxi_CheckCall(call))
- return;
-#endif /* RX_ENABLE_LOCKS */
/* Don't try to keep alive dallying calls */
if (call->state == RX_STATE_DALLY) {
CALL_RELE(call, RX_CALL_REFCOUNT_MTU);
MUTEX_ENTER(&call->lock);
- if (event == call->growMTUEvent) {
- rxevent_Put(call->growMTUEvent);
- call->growMTUEvent = NULL;
- }
+ if (event == call->growMTUEvent)
+ rxevent_Put(&call->growMTUEvent);
-#ifdef RX_ENABLE_LOCKS
if (rxi_CheckCall(call, 0)) {
MUTEX_EXIT(&call->lock);
return;
}
-#else /* RX_ENABLE_LOCKS */
- if (rxi_CheckCall(call))
- return;
-#endif /* RX_ENABLE_LOCKS */
/* Don't bother with dallying calls */
if (call->state == RX_STATE_DALLY) {
}
static void
+rxi_CancelKeepAliveEvent(struct rx_call *call) {
+ if (call->keepAliveEvent) {
+ rxevent_Cancel(&call->keepAliveEvent);
+ CALL_RELE(call, RX_CALL_REFCOUNT_ALIVE);
+ }
+}
+
+static void
rxi_ScheduleGrowMTUEvent(struct rx_call *call, int secs)
{
if (!call->growMTUEvent) {
}
}
-/* N.B. rxi_KeepAliveOff: is defined earlier as a macro */
+static void
+rxi_CancelGrowMTUEvent(struct rx_call *call)
+{
+ if (call->growMTUEvent) {
+ rxevent_Cancel(&call->growMTUEvent);
+ CALL_RELE(call, RX_CALL_REFCOUNT_MTU);
+ }
+}
+
+/*
+ * Increment the counter for the next connection ID, handling overflow.
+ */
+static void
+update_nextCid(void)
+{
+ /* Overflow is technically undefined behavior; avoid it. */
+ if (rx_nextCid > MAX_AFS_INT32 - (1 << RX_CIDSHIFT))
+ rx_nextCid = -1 * ((MAX_AFS_INT32 / RX_CIDSHIFT) * RX_CIDSHIFT);
+ else
+ rx_nextCid += 1 << RX_CIDSHIFT;
+}
+
static void
rxi_KeepAliveOn(struct rx_call *call)
{
rxi_ScheduleKeepAliveEvent(call);
}
-/*
- * Solely in order that callers not need to include rx_call.h
- */
void
rx_KeepAliveOff(struct rx_call *call)
{
- rxi_KeepAliveOff(call);
+ MUTEX_ENTER(&call->lock);
+ rxi_CancelKeepAliveEvent(call);
+ MUTEX_EXIT(&call->lock);
}
+
void
rx_KeepAliveOn(struct rx_call *call)
{
+ MUTEX_ENTER(&call->lock);
rxi_KeepAliveOn(call);
+ MUTEX_EXIT(&call->lock);
}
static void
struct rx_packet *packet;
MUTEX_ENTER(&conn->conn_data_lock);
- rxevent_Put(conn->delayedAbortEvent);
- conn->delayedAbortEvent = NULL;
+ rxevent_Put(&conn->delayedAbortEvent);
error = htonl(conn->error);
conn->abortCount++;
MUTEX_EXIT(&conn->conn_data_lock);
struct rx_packet *packet;
MUTEX_ENTER(&call->lock);
- rxevent_Put(call->delayedAbortEvent);
- call->delayedAbortEvent = NULL;
+ rxevent_Put(&call->delayedAbortEvent);
error = htonl(call->error);
call->abortCount++;
packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
{
struct rx_connection *conn = arg0;
- if (event) {
- rxevent_Put(conn->challengeEvent);
- conn->challengeEvent = NULL;
- }
+ if (event)
+ rxevent_Put(&conn->challengeEvent);
+
+ /* If there are no active calls it is not worth re-issuing the
+ * challenge. If the client issues another call on this connection
+ * the challenge can be requested at that time.
+ */
+ if (!rxi_HasActiveCalls(conn))
+ return;
if (RXS_CheckAuthentication(conn->securityObject, conn) != 0) {
struct rx_packet *packet;
int unused3)
{
struct clock now, when;
+ struct rxevent *event;
clock_GetTime(&now);
/* Find server connection structures that haven't been used for
code = MUTEX_TRYENTER(&call->lock);
if (!code)
continue;
-#ifdef RX_ENABLE_LOCKS
result = rxi_CheckCall(call, 1);
-#else /* RX_ENABLE_LOCKS */
- result = rxi_CheckCall(call);
-#endif /* RX_ENABLE_LOCKS */
MUTEX_EXIT(&call->lock);
if (result == -2) {
/* If CheckCall freed the call, it might
when = now;
when.sec += RX_REAP_TIME; /* Check every RX_REAP_TIME seconds */
- rxevent_Put(rxevent_Post(&when, &now, rxi_ReapConnections, 0, NULL, 0));
+ event = rxevent_Post(&when, &now, rxi_ReapConnections, 0, NULL, 0);
+ rxevent_Put(&event);
}
struct rx_serverQueueEntry *sq;
#endif /* KERNEL */
- LOCK_RX_INIT;
- if (rxinit_status == 1) {
- UNLOCK_RX_INIT;
+ if (rx_atomic_test_and_set_bit(&rxinit_status, 0))
return; /* Already shutdown. */
- }
+
#ifndef KERNEL
rx_port = 0;
#ifndef AFS_PTHREAD_ENV
rxi_StopListener();
#endif /* AFS_PTHREAD_ENV */
shutdown_rxevent();
- rx_SetEpoch(0);
+ rx_epoch = 0;
#ifndef AFS_PTHREAD_ENV
#ifndef AFS_USE_GETTIMEOFDAY
clock_UnInit();
rxi_dataQuota = RX_MAX_QUOTA;
rxi_availProcs = rxi_totalMin = rxi_minDeficit = 0;
MUTEX_EXIT(&rx_quota_mutex);
- rxinit_status = 1;
- UNLOCK_RX_INIT;
}
-#ifdef RX_ENABLE_LOCKS
-void
-osirx_AssertMine(afs_kmutex_t * lockaddr, char *msg)
-{
- if (!MUTEX_ISMINE(lockaddr))
- osi_Panic("Lock not held: %s", msg);
-}
-#endif /* RX_ENABLE_LOCKS */
-
#ifndef KERNEL
/*
if (rxInterface == -1)
return;
- peer = rxi_FindPeer(peerHost, peerPort, 0, 0);
+ peer = rxi_FindPeer(peerHost, peerPort, 0);
if (!peer)
return;
if (rpcop_stat == NULL)
return NULL;
- peer = rxi_FindPeer(peerHost, peerPort, 0, 0);
+ peer = rxi_FindPeer(peerHost, peerPort, 0);
if (!peer)
return NULL;
for (opr_queue_Scan(&processStats, cursor)) {
unsigned int num_funcs = 0, i;
struct rx_interface_stat *rpc_stat
- = opr_queue_Entry(rpc_stat, struct rx_interface_stat, entry);
+ = opr_queue_Entry(cursor, struct rx_interface_stat, entry);
num_funcs = rpc_stat->stats[0].func_total;
for (i = 0; i < num_funcs; i++) {
RXDPRINTF(RXDPRINTOUT, "%s - call=0x%p, id=%u, state=%u, mode=%u, conn=%p, epoch=%u, cid=%u, callNum=%u, connFlags=0x%x, flags=0x%x, "
"rqc=%u,%u, tqc=%u,%u, iovqc=%u,%u, "
"lstatus=%u, rstatus=%u, error=%d, timeout=%u, "
- "resendEvent=%d, timeoutEvt=%d, keepAliveEvt=%d, delayedAckEvt=%d, delayedAbortEvt=%d, abortCode=%d, abortCount=%d, "
+ "resendEvent=%d, keepAliveEvt=%d, delayedAckEvt=%d, delayedAbortEvt=%d, abortCode=%d, abortCount=%d, "
"lastSendTime=%u, lastRecvTime=%u, lastSendData=%u"
#ifdef RX_ENABLE_LOCKS
", refCount=%u"
c->callNumber?*c->callNumber:0, c->conn?c->conn->flags:0, c->flags,
(afs_uint32)c->rqc, (afs_uint32)rqc, (afs_uint32)c->tqc, (afs_uint32)tqc, (afs_uint32)c->iovqc, (afs_uint32)iovqc,
(afs_uint32)c->localStatus, (afs_uint32)c->remoteStatus, c->error, c->timeout,
- c->resendEvent?1:0, c->timeoutEvent?1:0, c->keepAliveEvent?1:0, c->delayedAckEvent?1:0, c->delayedAbortEvent?1:0,
+ c->resendEvent?1:0, c->keepAliveEvent?1:0, c->delayedAckEvent?1:0, c->delayedAbortEvent?1:0,
c->abortCode, c->abortCount, c->lastSendTime, c->lastReceiveTime, c->lastSendData
#ifdef RX_ENABLE_LOCKS
, (afs_uint32)c->refCount
git://git.openafs.org / openafs.git / blobdiff