2 * Copyright 2000, International Business Machines Corporation and others.
5 * This software has been released under the terms of the IBM Public
6 * License. For details, see the LICENSE file in the top-level source
7 * directory or online at http://www.openafs.org/dl/license10.html
10 /* RX: Extended Remote Procedure Call */
12 #include <afsconfig.h>
13 #include <afs/param.h>
16 # include "afs/sysincludes.h"
17 # include "afsincludes.h"
22 # ifdef AFS_LINUX20_ENV
23 # include "h/socket.h"
25 # include "netinet/in.h"
27 # include "netinet/ip6.h"
28 # include "inet/common.h"
30 # include "inet/ip_ire.h"
32 # include "afs/afs_args.h"
33 # include "afs/afs_osi.h"
34 # ifdef RX_KERNEL_TRACE
35 # include "rx_kcommon.h"
37 # if defined(AFS_AIX_ENV)
41 # undef RXDEBUG /* turn off debugging */
43 # if defined(AFS_SGI_ENV)
44 # include "sys/debug.h"
47 # include "afs/sysincludes.h"
48 # include "afsincludes.h"
49 # endif /* !UKERNEL */
50 # include "afs/lock.h"
51 # include "rx_kmutex.h"
52 # include "rx_kernel.h"
53 # define AFSOP_STOP_RXCALLBACK 210 /* Stop CALLBACK process */
54 # define AFSOP_STOP_AFS 211 /* Stop AFS process */
55 # define AFSOP_STOP_BKG 212 /* Stop BKG process */
56 extern afs_int32 afs_termState;
58 # include "sys/lockl.h"
59 # include "sys/lock_def.h"
60 # endif /* AFS_AIX41_ENV */
61 # include "afs/rxgen_consts.h"
66 # include <afs/afsutil.h>
67 # include <WINNT\afsreg.h>
75 #include <opr/queue.h>
76 #include <hcrypto/rand.h>
80 #include "rx_atomic.h"
81 #include "rx_globals.h"
83 #include "rx_internal.h"
90 #include "rx_packet.h"
91 #include "rx_server.h"
93 #include <afs/rxgen_consts.h>
96 #ifdef AFS_PTHREAD_ENV
98 int (*registerProgram) (pid_t, char *) = 0;
99 int (*swapNameProgram) (pid_t, const char *, char *) = 0;
102 int (*registerProgram) (PROCESS, char *) = 0;
103 int (*swapNameProgram) (PROCESS, const char *, char *) = 0;
107 /* Local static routines */
108 static void rxi_DestroyConnectionNoLock(struct rx_connection *conn);
109 static void rxi_ComputeRoundTripTime(struct rx_packet *, struct rx_ackPacket *,
110 struct rx_call *, struct rx_peer *,
112 static void rxi_Resend(struct rxevent *event, void *arg0, void *arg1,
114 static void rxi_SendDelayedAck(struct rxevent *event, void *call,
115 void *dummy, int dummy2);
116 static void rxi_SendDelayedCallAbort(struct rxevent *event, void *arg1,
117 void *dummy, int dummy2);
118 static void rxi_SendDelayedConnAbort(struct rxevent *event, void *arg1,
119 void *unused, int unused2);
120 static void rxi_ReapConnections(struct rxevent *unused, void *unused1,
121 void *unused2, int unused3);
122 static struct rx_packet *rxi_SendCallAbort(struct rx_call *call,
123 struct rx_packet *packet,
124 int istack, int force);
125 static void rxi_AckAll(struct rx_call *call);
126 static struct rx_connection
127 *rxi_FindConnection(osi_socket socket, afs_uint32 host, u_short port,
128 u_short serviceId, afs_uint32 cid,
129 afs_uint32 epoch, int type, u_int securityIndex,
130 int *unknownService);
131 static struct rx_packet
132 *rxi_ReceiveDataPacket(struct rx_call *call, struct rx_packet *np,
133 int istack, osi_socket socket,
134 afs_uint32 host, u_short port, int *tnop,
135 struct rx_call **newcallp);
136 static struct rx_packet
137 *rxi_ReceiveAckPacket(struct rx_call *call, struct rx_packet *np,
139 static struct rx_packet
140 *rxi_ReceiveResponsePacket(struct rx_connection *conn,
141 struct rx_packet *np, int istack);
142 static struct rx_packet
143 *rxi_ReceiveChallengePacket(struct rx_connection *conn,
144 struct rx_packet *np, int istack);
145 static void rxi_AttachServerProc(struct rx_call *call, osi_socket socket,
146 int *tnop, struct rx_call **newcallp);
147 static void rxi_ClearTransmitQueue(struct rx_call *call, int force);
148 static void rxi_ClearReceiveQueue(struct rx_call *call);
149 static void rxi_ResetCall(struct rx_call *call, int newcall);
150 static void rxi_ScheduleKeepAliveEvent(struct rx_call *call);
151 static void rxi_ScheduleNatKeepAliveEvent(struct rx_connection *conn);
152 static void rxi_ScheduleGrowMTUEvent(struct rx_call *call, int secs);
153 static void rxi_KeepAliveOn(struct rx_call *call);
154 static void rxi_GrowMTUOn(struct rx_call *call);
155 static void rxi_ChallengeOn(struct rx_connection *conn);
156 static int rxi_CheckCall(struct rx_call *call, int haveCTLock);
157 static void rxi_AckAllInTransmitQueue(struct rx_call *call);
158 static void rxi_CancelKeepAliveEvent(struct rx_call *call);
159 static void rxi_CancelDelayedAbortEvent(struct rx_call *call);
160 static void rxi_CancelGrowMTUEvent(struct rx_call *call);
161 static void update_nextCid(void);
163 #ifdef RX_ENABLE_LOCKS
165 rx_atomic_t rxi_start_aborted; /* rxi_start awoke after rxi_Send in error.*/
166 rx_atomic_t rxi_start_in_error;
168 #endif /* RX_ENABLE_LOCKS */
170 /* Constant delay time before sending an acknowledge of the last packet
171 * received. This is to avoid sending an extra acknowledge when the
172 * client is about to make another call, anyway, or the server is
175 * The lastAckDelay may not exceeed 400ms without causing peers to
176 * unecessarily timeout.
178 struct clock rx_lastAckDelay = {0, 400000};
180 /* Constant delay time before sending a soft ack when none was requested.
181 * This is to make sure we send soft acks before the sender times out,
182 * Normally we wait and send a hard ack when the receiver consumes the packet
184 * This value has been 100ms in all shipping versions of OpenAFS. Changing it
185 * will require changes to the peer's RTT calculations.
187 struct clock rx_softAckDelay = {0, 100000};
190 * rxi_rpc_peer_stat_cnt counts the total number of peer stat structures
191 * currently allocated within rx. This number is used to allocate the
192 * memory required to return the statistics when queried.
193 * Protected by the rx_rpc_stats mutex.
196 static unsigned int rxi_rpc_peer_stat_cnt;
199 * rxi_rpc_process_stat_cnt counts the total number of local process stat
200 * structures currently allocated within rx. The number is used to allocate
201 * the memory required to return the statistics when queried.
202 * Protected by the rx_rpc_stats mutex.
205 static unsigned int rxi_rpc_process_stat_cnt;
208 * rxi_busyChannelError is a boolean. It indicates whether or not RX_CALL_BUSY
209 * errors should be reported to the application when a call channel appears busy
210 * (inferred from the receipt of RX_PACKET_TYPE_BUSY packets on the channel),
211 * and there are other call channels in the connection that are not busy.
212 * If 0, we do not return errors upon receiving busy packets; we just keep
213 * trying on the same call channel until we hit a timeout.
215 static afs_int32 rxi_busyChannelError = 0;
217 rx_atomic_t rx_nWaiting = RX_ATOMIC_INIT(0);
218 rx_atomic_t rx_nWaited = RX_ATOMIC_INIT(0);
220 /* Incoming calls wait on this queue when there are no available
221 * server processes */
222 struct opr_queue rx_incomingCallQueue;
224 /* Server processes wait on this queue when there are no appropriate
225 * calls to process */
226 struct opr_queue rx_idleServerQueue;
228 #if !defined(offsetof)
229 #include <stddef.h> /* for definition of offsetof() */
232 #ifdef RX_ENABLE_LOCKS
233 afs_kmutex_t rx_atomic_mutex;
236 /* Forward prototypes */
237 static struct rx_call * rxi_NewCall(struct rx_connection *, int);
240 putConnection (struct rx_connection *conn) {
241 MUTEX_ENTER(&rx_refcnt_mutex);
243 MUTEX_EXIT(&rx_refcnt_mutex);
246 #ifdef AFS_PTHREAD_ENV
249 * Use procedural initialization of mutexes/condition variables
253 extern afs_kmutex_t rx_quota_mutex;
254 extern afs_kmutex_t rx_pthread_mutex;
255 extern afs_kmutex_t rx_packets_mutex;
256 extern afs_kmutex_t rx_refcnt_mutex;
257 extern afs_kmutex_t des_init_mutex;
258 extern afs_kmutex_t des_random_mutex;
260 extern afs_kmutex_t rx_clock_mutex;
261 extern afs_kmutex_t rxi_connCacheMutex;
262 extern afs_kmutex_t event_handler_mutex;
263 extern afs_kmutex_t listener_mutex;
264 extern afs_kmutex_t rx_if_init_mutex;
265 extern afs_kmutex_t rx_if_mutex;
267 extern afs_kcondvar_t rx_event_handler_cond;
268 extern afs_kcondvar_t rx_listener_cond;
271 static afs_kmutex_t epoch_mutex;
272 static afs_kmutex_t rx_init_mutex;
273 static afs_kmutex_t rx_debug_mutex;
274 static afs_kmutex_t rx_rpc_stats;
277 rxi_InitPthread(void)
279 MUTEX_INIT(&rx_quota_mutex, "quota", MUTEX_DEFAULT, 0);
280 MUTEX_INIT(&rx_pthread_mutex, "pthread", MUTEX_DEFAULT, 0);
281 MUTEX_INIT(&rx_packets_mutex, "packets", MUTEX_DEFAULT, 0);
282 MUTEX_INIT(&rx_refcnt_mutex, "refcnts", MUTEX_DEFAULT, 0);
284 MUTEX_INIT(&rx_clock_mutex, "clock", MUTEX_DEFAULT, 0);
285 MUTEX_INIT(&rxi_connCacheMutex, "conn cache", MUTEX_DEFAULT, 0);
286 MUTEX_INIT(&event_handler_mutex, "event handler", MUTEX_DEFAULT, 0);
287 MUTEX_INIT(&listener_mutex, "listener", MUTEX_DEFAULT, 0);
288 MUTEX_INIT(&rx_if_init_mutex, "if init", MUTEX_DEFAULT, 0);
289 MUTEX_INIT(&rx_if_mutex, "if", MUTEX_DEFAULT, 0);
291 MUTEX_INIT(&rx_stats_mutex, "stats", MUTEX_DEFAULT, 0);
292 MUTEX_INIT(&rx_atomic_mutex, "atomic", MUTEX_DEFAULT, 0);
293 MUTEX_INIT(&epoch_mutex, "epoch", MUTEX_DEFAULT, 0);
294 MUTEX_INIT(&rx_init_mutex, "init", MUTEX_DEFAULT, 0);
295 MUTEX_INIT(&rx_debug_mutex, "debug", MUTEX_DEFAULT, 0);
298 CV_INIT(&rx_event_handler_cond, "evhand", CV_DEFAULT, 0);
299 CV_INIT(&rx_listener_cond, "rxlisten", CV_DEFAULT, 0);
302 osi_Assert(pthread_key_create(&rx_thread_id_key, NULL) == 0);
303 osi_Assert(pthread_key_create(&rx_ts_info_key, NULL) == 0);
305 MUTEX_INIT(&rx_rpc_stats, "rx_rpc_stats", MUTEX_DEFAULT, 0);
306 MUTEX_INIT(&rx_freePktQ_lock, "rx_freePktQ_lock", MUTEX_DEFAULT, 0);
307 #ifdef RX_ENABLE_LOCKS
310 #endif /* RX_LOCKS_DB */
311 MUTEX_INIT(&freeSQEList_lock, "freeSQEList lock", MUTEX_DEFAULT, 0);
312 MUTEX_INIT(&rx_freeCallQueue_lock, "rx_freeCallQueue_lock", MUTEX_DEFAULT,
314 CV_INIT(&rx_waitingForPackets_cv, "rx_waitingForPackets_cv", CV_DEFAULT,
316 MUTEX_INIT(&rx_peerHashTable_lock, "rx_peerHashTable_lock", MUTEX_DEFAULT,
318 MUTEX_INIT(&rx_connHashTable_lock, "rx_connHashTable_lock", MUTEX_DEFAULT,
320 MUTEX_INIT(&rx_serverPool_lock, "rx_serverPool_lock", MUTEX_DEFAULT, 0);
322 MUTEX_INIT(&rxi_keyCreate_lock, "rxi_keyCreate_lock", MUTEX_DEFAULT, 0);
324 #endif /* RX_ENABLE_LOCKS */
327 pthread_once_t rx_once_init = PTHREAD_ONCE_INIT;
328 #define INIT_PTHREAD_LOCKS osi_Assert(pthread_once(&rx_once_init, rxi_InitPthread)==0)
330 * The rx_stats_mutex mutex protects the following global variables:
331 * rxi_lowConnRefCount
332 * rxi_lowPeerRefCount
341 * The rx_quota_mutex mutex protects the following global variables:
349 * The rx_freePktQ_lock protects the following global variables:
354 * The rx_packets_mutex mutex protects the following global variables:
362 * The rx_pthread_mutex mutex protects the following global variables:
363 * rxi_fcfs_thread_num
366 #define INIT_PTHREAD_LOCKS
370 /* Variables for handling the minProcs implementation. availProcs gives the
371 * number of threads available in the pool at this moment (not counting dudes
372 * executing right now). totalMin gives the total number of procs required
373 * for handling all minProcs requests. minDeficit is a dynamic variable
374 * tracking the # of procs required to satisfy all of the remaining minProcs
376 * For fine grain locking to work, the quota check and the reservation of
377 * a server thread has to come while rxi_availProcs and rxi_minDeficit
378 * are locked. To this end, the code has been modified under #ifdef
379 * RX_ENABLE_LOCKS so that quota checks and reservation occur at the
380 * same time. A new function, ReturnToServerPool() returns the allocation.
382 * A call can be on several queue's (but only one at a time). When
383 * rxi_ResetCall wants to remove the call from a queue, it has to ensure
384 * that no one else is touching the queue. To this end, we store the address
385 * of the queue lock in the call structure (under the call lock) when we
386 * put the call on a queue, and we clear the call_queue_lock when the
387 * call is removed from a queue (once the call lock has been obtained).
388 * This allows rxi_ResetCall to safely synchronize with others wishing
389 * to manipulate the queue.
392 #if defined(RX_ENABLE_LOCKS)
393 static afs_kmutex_t rx_rpc_stats;
396 /* We keep a "last conn pointer" in rxi_FindConnection. The odds are
397 ** pretty good that the next packet coming in is from the same connection
398 ** as the last packet, since we're send multiple packets in a transmit window.
400 struct rx_connection *rxLastConn = 0;
402 #ifdef RX_ENABLE_LOCKS
403 /* The locking hierarchy for rx fine grain locking is composed of these
406 * rx_connHashTable_lock - synchronizes conn creation, rx_connHashTable access
407 * also protects updates to rx_nextCid
408 * conn_call_lock - used to synchonize rx_EndCall and rx_NewCall
409 * call->lock - locks call data fields.
410 * These are independent of each other:
411 * rx_freeCallQueue_lock
416 * serverQueueEntry->lock
417 * rx_peerHashTable_lock - locked under rx_connHashTable_lock
419 * peer->lock - locks peer data fields.
420 * conn_data_lock - that more than one thread is not updating a conn data
421 * field at the same time.
432 * Do we need a lock to protect the peer field in the conn structure?
433 * conn->peer was previously a constant for all intents and so has no
434 * lock protecting this field. The multihomed client delta introduced
435 * a RX code change : change the peer field in the connection structure
436 * to that remote interface from which the last packet for this
437 * connection was sent out. This may become an issue if further changes
440 #define SET_CALL_QUEUE_LOCK(C, L) (C)->call_queue_lock = (L)
441 #define CLEAR_CALL_QUEUE_LOCK(C) (C)->call_queue_lock = NULL
443 /* rxdb_fileID is used to identify the lock location, along with line#. */
444 static int rxdb_fileID = RXDB_FILE_RX;
445 #endif /* RX_LOCKS_DB */
446 #else /* RX_ENABLE_LOCKS */
447 #define SET_CALL_QUEUE_LOCK(C, L)
448 #define CLEAR_CALL_QUEUE_LOCK(C)
449 #endif /* RX_ENABLE_LOCKS */
450 struct rx_serverQueueEntry *rx_waitForPacket = 0;
452 /* ------------Exported Interfaces------------- */
454 /* Initialize rx. A port number may be mentioned, in which case this
455 * becomes the default port number for any service installed later.
456 * If 0 is provided for the port number, a random port will be chosen
457 * by the kernel. Whether this will ever overlap anything in
458 * /etc/services is anybody's guess... Returns 0 on success, -1 on
463 rx_atomic_t rxinit_status = RX_ATOMIC_INIT(1);
466 rx_InitHost(u_int host, u_int port)
473 char *htable, *ptable;
478 if (!rx_atomic_test_and_clear_bit(&rxinit_status, 0))
479 return 0; /* already started */
485 if (afs_winsockInit() < 0)
491 * Initialize anything necessary to provide a non-premptive threading
494 rxi_InitializeThreadSupport();
497 /* Allocate and initialize a socket for client and perhaps server
500 rx_socket = rxi_GetHostUDPSocket(host, (u_short) port);
501 if (rx_socket == OSI_NULLSOCKET) {
504 #if defined(RX_ENABLE_LOCKS) && defined(KERNEL)
507 #endif /* RX_LOCKS_DB */
508 MUTEX_INIT(&rx_stats_mutex, "rx_stats_mutex", MUTEX_DEFAULT, 0);
509 MUTEX_INIT(&rx_quota_mutex, "rx_quota_mutex", MUTEX_DEFAULT, 0);
510 MUTEX_INIT(&rx_atomic_mutex, "rx_atomic_mutex", MUTEX_DEFAULT, 0);
511 MUTEX_INIT(&rx_pthread_mutex, "rx_pthread_mutex", MUTEX_DEFAULT, 0);
512 MUTEX_INIT(&rx_packets_mutex, "rx_packets_mutex", MUTEX_DEFAULT, 0);
513 MUTEX_INIT(&rx_refcnt_mutex, "rx_refcnt_mutex", MUTEX_DEFAULT, 0);
514 MUTEX_INIT(&rx_rpc_stats, "rx_rpc_stats", MUTEX_DEFAULT, 0);
515 MUTEX_INIT(&rx_freePktQ_lock, "rx_freePktQ_lock", MUTEX_DEFAULT, 0);
516 MUTEX_INIT(&freeSQEList_lock, "freeSQEList lock", MUTEX_DEFAULT, 0);
517 MUTEX_INIT(&rx_freeCallQueue_lock, "rx_freeCallQueue_lock", MUTEX_DEFAULT,
519 CV_INIT(&rx_waitingForPackets_cv, "rx_waitingForPackets_cv", CV_DEFAULT,
521 MUTEX_INIT(&rx_peerHashTable_lock, "rx_peerHashTable_lock", MUTEX_DEFAULT,
523 MUTEX_INIT(&rx_connHashTable_lock, "rx_connHashTable_lock", MUTEX_DEFAULT,
525 MUTEX_INIT(&rx_serverPool_lock, "rx_serverPool_lock", MUTEX_DEFAULT, 0);
526 #if defined(AFS_HPUX110_ENV)
528 rx_sleepLock = alloc_spinlock(LAST_HELD_ORDER - 10, "rx_sleepLock");
529 #endif /* AFS_HPUX110_ENV */
530 #endif /* RX_ENABLE_LOCKS && KERNEL */
533 rx_connDeadTime = 12;
534 rx_tranquil = 0; /* reset flag */
535 rxi_ResetStatistics();
536 htable = osi_Alloc(rx_hashTableSize * sizeof(struct rx_connection *));
537 PIN(htable, rx_hashTableSize * sizeof(struct rx_connection *)); /* XXXXX */
538 memset(htable, 0, rx_hashTableSize * sizeof(struct rx_connection *));
539 ptable = osi_Alloc(rx_hashTableSize * sizeof(struct rx_peer *));
540 PIN(ptable, rx_hashTableSize * sizeof(struct rx_peer *)); /* XXXXX */
541 memset(ptable, 0, rx_hashTableSize * sizeof(struct rx_peer *));
543 /* Malloc up a bunch of packets & buffers */
545 opr_queue_Init(&rx_freePacketQueue);
546 rxi_NeedMorePackets = FALSE;
547 rx_nPackets = 0; /* rx_nPackets is managed by rxi_MorePackets* */
549 /* enforce a minimum number of allocated packets */
550 if (rx_extraPackets < rxi_nSendFrags * rx_maxSendWindow)
551 rx_extraPackets = rxi_nSendFrags * rx_maxSendWindow;
553 /* allocate the initial free packet pool */
554 #ifdef RX_ENABLE_TSFPQ
555 rxi_MorePacketsTSFPQ(rx_extraPackets + RX_MAX_QUOTA + 2, RX_TS_FPQ_FLUSH_GLOBAL, 0);
556 #else /* RX_ENABLE_TSFPQ */
557 rxi_MorePackets(rx_extraPackets + RX_MAX_QUOTA + 2); /* fudge */
558 #endif /* RX_ENABLE_TSFPQ */
565 #if defined(AFS_NT40_ENV) && !defined(AFS_PTHREAD_ENV)
566 tv.tv_sec = clock_now.sec;
567 tv.tv_usec = clock_now.usec;
568 srand((unsigned int)tv.tv_usec);
575 #if defined(KERNEL) && !defined(UKERNEL)
576 /* Really, this should never happen in a real kernel */
579 struct sockaddr_in addr;
581 int addrlen = sizeof(addr);
583 socklen_t addrlen = sizeof(addr);
585 if (getsockname((intptr_t)rx_socket, (struct sockaddr *)&addr, &addrlen)) {
587 osi_Free(htable, rx_hashTableSize * sizeof(struct rx_connection *));
590 rx_port = addr.sin_port;
593 rx_stats.minRtt.sec = 9999999;
594 if (RAND_bytes(&rx_epoch, sizeof(rx_epoch)) != 1)
596 rx_epoch = (rx_epoch & ~0x40000000) | 0x80000000;
597 if (RAND_bytes(&rx_nextCid, sizeof(rx_nextCid)) != 1)
599 rx_nextCid &= RX_CIDMASK;
600 MUTEX_ENTER(&rx_quota_mutex);
601 rxi_dataQuota += rx_extraQuota; /* + extra pkts caller asked to rsrv */
602 MUTEX_EXIT(&rx_quota_mutex);
603 /* *Slightly* random start time for the cid. This is just to help
604 * out with the hashing function at the peer */
605 rx_nextCid = ((tv.tv_sec ^ tv.tv_usec) << RX_CIDSHIFT);
606 rx_connHashTable = (struct rx_connection **)htable;
607 rx_peerHashTable = (struct rx_peer **)ptable;
609 rx_hardAckDelay.sec = 0;
610 rx_hardAckDelay.usec = 100000; /* 100 milliseconds */
612 rxevent_Init(20, rxi_ReScheduleEvents);
614 /* Initialize various global queues */
615 opr_queue_Init(&rx_idleServerQueue);
616 opr_queue_Init(&rx_incomingCallQueue);
617 opr_queue_Init(&rx_freeCallQueue);
619 #if defined(AFS_NT40_ENV) && !defined(KERNEL)
620 /* Initialize our list of usable IP addresses. */
624 /* Start listener process (exact function is dependent on the
625 * implementation environment--kernel or user space) */
629 rx_atomic_clear_bit(&rxinit_status, 0);
636 return rx_InitHost(htonl(INADDR_ANY), port);
642 * The rxi_rto functions implement a TCP (RFC2988) style algorithm for
643 * maintaing the round trip timer.
648 * Start a new RTT timer for a given call and packet.
650 * There must be no resendEvent already listed for this call, otherwise this
651 * will leak events - intended for internal use within the RTO code only
654 * the RX call to start the timer for
655 * @param[in] lastPacket
656 * a flag indicating whether the last packet has been sent or not
658 * @pre call must be locked before calling this function
662 rxi_rto_startTimer(struct rx_call *call, int lastPacket, int istack)
664 struct clock now, retryTime;
669 clock_Add(&retryTime, &call->rto);
671 /* If we're sending the last packet, and we're the client, then the server
672 * may wait for an additional 400ms before returning the ACK, wait for it
673 * rather than hitting a timeout */
674 if (lastPacket && call->conn->type == RX_CLIENT_CONNECTION)
675 clock_Addmsec(&retryTime, 400);
677 CALL_HOLD(call, RX_CALL_REFCOUNT_RESEND);
678 call->resendEvent = rxevent_Post(&retryTime, &now, rxi_Resend,
683 * Cancel an RTT timer for a given call.
687 * the RX call to cancel the timer for
689 * @pre call must be locked before calling this function
694 rxi_rto_cancel(struct rx_call *call)
696 if (call->resendEvent != NULL) {
697 rxevent_Cancel(&call->resendEvent);
698 CALL_RELE(call, RX_CALL_REFCOUNT_RESEND);
703 * Tell the RTO timer that we have sent a packet.
705 * If the timer isn't already running, then start it. If the timer is running,
709 * the RX call that the packet has been sent on
710 * @param[in] lastPacket
711 * A flag which is true if this is the last packet for the call
713 * @pre The call must be locked before calling this function
718 rxi_rto_packet_sent(struct rx_call *call, int lastPacket, int istack)
720 if (call->resendEvent)
723 rxi_rto_startTimer(call, lastPacket, istack);
727 * Tell the RTO timer that we have received an new ACK message
729 * This function should be called whenever a call receives an ACK that
730 * acknowledges new packets. Whatever happens, we stop the current timer.
731 * If there are unacked packets in the queue which have been sent, then
732 * we restart the timer from now. Otherwise, we leave it stopped.
735 * the RX call that the ACK has been received on
739 rxi_rto_packet_acked(struct rx_call *call, int istack)
741 struct opr_queue *cursor;
743 rxi_rto_cancel(call);
745 if (opr_queue_IsEmpty(&call->tq))
748 for (opr_queue_Scan(&call->tq, cursor)) {
749 struct rx_packet *p = opr_queue_Entry(cursor, struct rx_packet, entry);
750 if (p->header.seq > call->tfirst + call->twind)
753 if (!(p->flags & RX_PKTFLAG_ACKED) && p->flags & RX_PKTFLAG_SENT) {
754 rxi_rto_startTimer(call, p->header.flags & RX_LAST_PACKET, istack);
762 * Set an initial round trip timeout for a peer connection
764 * @param[in] secs The timeout to set in seconds
768 rx_rto_setPeerTimeoutSecs(struct rx_peer *peer, int secs) {
769 peer->rtt = secs * 8000;
773 * Enables or disables the busy call channel error (RX_CALL_BUSY).
775 * @param[in] onoff Non-zero to enable busy call channel errors.
777 * @pre Neither rx_Init nor rx_InitHost have been called yet
780 rx_SetBusyChannelError(afs_int32 onoff)
782 osi_Assert(rx_atomic_test_bit(&rxinit_status, 0));
783 rxi_busyChannelError = onoff ? 1 : 0;
787 * Set a delayed ack event on the specified call for the given time
789 * @param[in] call - the call on which to set the event
790 * @param[in] offset - the delay from now after which the event fires
793 rxi_PostDelayedAckEvent(struct rx_call *call, struct clock *offset)
795 struct clock now, when;
799 clock_Add(&when, offset);
801 if (call->delayedAckEvent && clock_Gt(&call->delayedAckTime, &when)) {
802 /* The event we're cancelling already has a reference, so we don't
804 rxevent_Cancel(&call->delayedAckEvent);
805 call->delayedAckEvent = rxevent_Post(&when, &now, rxi_SendDelayedAck,
808 call->delayedAckTime = when;
809 } else if (!call->delayedAckEvent) {
810 CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
811 call->delayedAckEvent = rxevent_Post(&when, &now,
814 call->delayedAckTime = when;
819 rxi_CancelDelayedAckEvent(struct rx_call *call)
821 if (call->delayedAckEvent) {
822 rxevent_Cancel(&call->delayedAckEvent);
823 CALL_RELE(call, RX_CALL_REFCOUNT_DELAY);
827 /* called with unincremented nRequestsRunning to see if it is OK to start
828 * a new thread in this service. Could be "no" for two reasons: over the
829 * max quota, or would prevent others from reaching their min quota.
831 #ifdef RX_ENABLE_LOCKS
832 /* This verion of QuotaOK reserves quota if it's ok while the
833 * rx_serverPool_lock is held. Return quota using ReturnToServerPool().
836 QuotaOK(struct rx_service *aservice)
838 /* check if over max quota */
839 if (aservice->nRequestsRunning >= aservice->maxProcs) {
843 /* under min quota, we're OK */
844 /* otherwise, can use only if there are enough to allow everyone
845 * to go to their min quota after this guy starts.
848 MUTEX_ENTER(&rx_quota_mutex);
849 if ((aservice->nRequestsRunning < aservice->minProcs)
850 || (rxi_availProcs > rxi_minDeficit)) {
851 aservice->nRequestsRunning++;
852 /* just started call in minProcs pool, need fewer to maintain
854 if (aservice->nRequestsRunning <= aservice->minProcs)
857 MUTEX_EXIT(&rx_quota_mutex);
860 MUTEX_EXIT(&rx_quota_mutex);
866 ReturnToServerPool(struct rx_service *aservice)
868 aservice->nRequestsRunning--;
869 MUTEX_ENTER(&rx_quota_mutex);
870 if (aservice->nRequestsRunning < aservice->minProcs)
873 MUTEX_EXIT(&rx_quota_mutex);
876 #else /* RX_ENABLE_LOCKS */
878 QuotaOK(struct rx_service *aservice)
881 /* under min quota, we're OK */
882 if (aservice->nRequestsRunning < aservice->minProcs)
885 /* check if over max quota */
886 if (aservice->nRequestsRunning >= aservice->maxProcs)
889 /* otherwise, can use only if there are enough to allow everyone
890 * to go to their min quota after this guy starts.
892 MUTEX_ENTER(&rx_quota_mutex);
893 if (rxi_availProcs > rxi_minDeficit)
895 MUTEX_EXIT(&rx_quota_mutex);
898 #endif /* RX_ENABLE_LOCKS */
901 /* Called by rx_StartServer to start up lwp's to service calls.
902 NExistingProcs gives the number of procs already existing, and which
903 therefore needn't be created. */
905 rxi_StartServerProcs(int nExistingProcs)
907 struct rx_service *service;
912 /* For each service, reserve N processes, where N is the "minimum"
913 * number of processes that MUST be able to execute a request in parallel,
914 * at any time, for that process. Also compute the maximum difference
915 * between any service's maximum number of processes that can run
916 * (i.e. the maximum number that ever will be run, and a guarantee
917 * that this number will run if other services aren't running), and its
918 * minimum number. The result is the extra number of processes that
919 * we need in order to provide the latter guarantee */
920 for (i = 0; i < RX_MAX_SERVICES; i++) {
922 service = rx_services[i];
923 if (service == (struct rx_service *)0)
925 nProcs += service->minProcs;
926 diff = service->maxProcs - service->minProcs;
930 nProcs += maxdiff; /* Extra processes needed to allow max number requested to run in any given service, under good conditions */
931 nProcs -= nExistingProcs; /* Subtract the number of procs that were previously created for use as server procs */
932 for (i = 0; i < nProcs; i++) {
933 rxi_StartServerProc(rx_ServerProc, rx_stackSize);
939 /* This routine is only required on Windows */
941 rx_StartClientThread(void)
943 #ifdef AFS_PTHREAD_ENV
945 pid = pthread_self();
946 #endif /* AFS_PTHREAD_ENV */
948 #endif /* AFS_NT40_ENV */
950 /* This routine must be called if any services are exported. If the
951 * donateMe flag is set, the calling process is donated to the server
954 rx_StartServer(int donateMe)
956 struct rx_service *service;
962 /* Start server processes, if necessary (exact function is dependent
963 * on the implementation environment--kernel or user space). DonateMe
964 * will be 1 if there is 1 pre-existing proc, i.e. this one. In this
965 * case, one less new proc will be created rx_StartServerProcs.
967 rxi_StartServerProcs(donateMe);
969 /* count up the # of threads in minProcs, and add set the min deficit to
970 * be that value, too.
972 for (i = 0; i < RX_MAX_SERVICES; i++) {
973 service = rx_services[i];
974 if (service == (struct rx_service *)0)
976 MUTEX_ENTER(&rx_quota_mutex);
977 rxi_totalMin += service->minProcs;
978 /* below works even if a thread is running, since minDeficit would
979 * still have been decremented and later re-incremented.
981 rxi_minDeficit += service->minProcs;
982 MUTEX_EXIT(&rx_quota_mutex);
985 /* Turn on reaping of idle server connections */
986 rxi_ReapConnections(NULL, NULL, NULL, 0);
995 #ifdef AFS_PTHREAD_ENV
997 pid = afs_pointer_to_int(pthread_self());
998 #else /* AFS_PTHREAD_ENV */
1000 LWP_CurrentProcess(&pid);
1001 #endif /* AFS_PTHREAD_ENV */
1003 sprintf(name, "srv_%d", ++nProcs);
1004 if (registerProgram)
1005 (*registerProgram) (pid, name);
1007 #endif /* AFS_NT40_ENV */
1008 rx_ServerProc(NULL); /* Never returns */
1010 #ifdef RX_ENABLE_TSFPQ
1011 /* no use leaving packets around in this thread's local queue if
1012 * it isn't getting donated to the server thread pool.
1014 rxi_FlushLocalPacketsTSFPQ();
1015 #endif /* RX_ENABLE_TSFPQ */
1019 /* Create a new client connection to the specified service, using the
1020 * specified security object to implement the security model for this
1022 struct rx_connection *
1023 rx_NewConnection(afs_uint32 shost, u_short sport, u_short sservice,
1024 struct rx_securityClass *securityObject,
1025 int serviceSecurityIndex)
1028 struct rx_connection *conn;
1033 dpf(("rx_NewConnection(host %x, port %u, service %u, securityObject %p, "
1034 "serviceSecurityIndex %d)\n",
1035 ntohl(shost), ntohs(sport), sservice, securityObject,
1036 serviceSecurityIndex));
1038 /* Vasilsi said: "NETPRI protects Cid and Alloc", but can this be true in
1039 * the case of kmem_alloc? */
1040 conn = rxi_AllocConnection();
1041 #ifdef RX_ENABLE_LOCKS
1042 MUTEX_INIT(&conn->conn_call_lock, "conn call lock", MUTEX_DEFAULT, 0);
1043 MUTEX_INIT(&conn->conn_data_lock, "conn data lock", MUTEX_DEFAULT, 0);
1044 CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
1047 MUTEX_ENTER(&rx_connHashTable_lock);
1048 conn->type = RX_CLIENT_CONNECTION;
1049 conn->epoch = rx_epoch;
1050 conn->cid = rx_nextCid;
1052 conn->peer = rxi_FindPeer(shost, sport, 1);
1053 conn->serviceId = sservice;
1054 conn->securityObject = securityObject;
1055 conn->securityData = (void *) 0;
1056 conn->securityIndex = serviceSecurityIndex;
1057 rx_SetConnDeadTime(conn, rx_connDeadTime);
1058 rx_SetConnSecondsUntilNatPing(conn, 0);
1059 conn->ackRate = RX_FAST_ACK_RATE;
1060 conn->nSpecific = 0;
1061 conn->specific = NULL;
1062 conn->challengeEvent = NULL;
1063 conn->delayedAbortEvent = NULL;
1064 conn->abortCount = 0;
1066 for (i = 0; i < RX_MAXCALLS; i++) {
1067 conn->twind[i] = rx_initSendWindow;
1068 conn->rwind[i] = rx_initReceiveWindow;
1069 conn->lastBusy[i] = 0;
1072 RXS_NewConnection(securityObject, conn);
1074 CONN_HASH(shost, sport, conn->cid, conn->epoch, RX_CLIENT_CONNECTION);
1076 conn->refCount++; /* no lock required since only this thread knows... */
1077 conn->next = rx_connHashTable[hashindex];
1078 rx_connHashTable[hashindex] = conn;
1079 if (rx_stats_active)
1080 rx_atomic_inc(&rx_stats.nClientConns);
1081 MUTEX_EXIT(&rx_connHashTable_lock);
1087 * Ensure a connection's timeout values are valid.
1089 * @param[in] conn The connection to check
1091 * @post conn->secondUntilDead <= conn->idleDeadTime <= conn->hardDeadTime,
1092 * unless idleDeadTime and/or hardDeadTime are not set
1096 rxi_CheckConnTimeouts(struct rx_connection *conn)
1098 /* a connection's timeouts must have the relationship
1099 * deadTime <= idleDeadTime <= hardDeadTime. Otherwise, for example, a
1100 * total loss of network to a peer may cause an idle timeout instead of a
1101 * dead timeout, simply because the idle timeout gets hit first. Also set
1102 * a minimum deadTime of 6, just to ensure it doesn't get set too low. */
1103 /* this logic is slightly complicated by the fact that
1104 * idleDeadTime/hardDeadTime may not be set at all, but it's not too bad.
1106 conn->secondsUntilDead = MAX(conn->secondsUntilDead, 6);
1107 if (conn->idleDeadTime) {
1108 conn->idleDeadTime = MAX(conn->idleDeadTime, conn->secondsUntilDead);
1110 if (conn->hardDeadTime) {
1111 if (conn->idleDeadTime) {
1112 conn->hardDeadTime = MAX(conn->idleDeadTime, conn->hardDeadTime);
1114 conn->hardDeadTime = MAX(conn->secondsUntilDead, conn->hardDeadTime);
1120 rx_SetConnDeadTime(struct rx_connection *conn, int seconds)
1122 /* The idea is to set the dead time to a value that allows several
1123 * keepalives to be dropped without timing out the connection. */
1124 conn->secondsUntilDead = seconds;
1125 rxi_CheckConnTimeouts(conn);
1126 conn->secondsUntilPing = conn->secondsUntilDead / 6;
1130 rx_SetConnHardDeadTime(struct rx_connection *conn, int seconds)
1132 conn->hardDeadTime = seconds;
1133 rxi_CheckConnTimeouts(conn);
1137 rx_SetConnIdleDeadTime(struct rx_connection *conn, int seconds)
1139 conn->idleDeadTime = seconds;
1140 rxi_CheckConnTimeouts(conn);
1143 int rxi_lowPeerRefCount = 0;
1144 int rxi_lowConnRefCount = 0;
1147 * Cleanup a connection that was destroyed in rxi_DestroyConnectioNoLock.
1148 * NOTE: must not be called with rx_connHashTable_lock held.
1151 rxi_CleanupConnection(struct rx_connection *conn)
1153 /* Notify the service exporter, if requested, that this connection
1154 * is being destroyed */
1155 if (conn->type == RX_SERVER_CONNECTION && conn->service->destroyConnProc)
1156 (*conn->service->destroyConnProc) (conn);
1158 /* Notify the security module that this connection is being destroyed */
1159 RXS_DestroyConnection(conn->securityObject, conn);
1161 /* If this is the last connection using the rx_peer struct, set its
1162 * idle time to now. rxi_ReapConnections will reap it if it's still
1163 * idle (refCount == 0) after rx_idlePeerTime (60 seconds) have passed.
1165 MUTEX_ENTER(&rx_peerHashTable_lock);
1166 if (conn->peer->refCount < 2) {
1167 conn->peer->idleWhen = clock_Sec();
1168 if (conn->peer->refCount < 1) {
1169 conn->peer->refCount = 1;
1170 if (rx_stats_active) {
1171 MUTEX_ENTER(&rx_stats_mutex);
1172 rxi_lowPeerRefCount++;
1173 MUTEX_EXIT(&rx_stats_mutex);
1177 conn->peer->refCount--;
1178 MUTEX_EXIT(&rx_peerHashTable_lock);
1180 if (rx_stats_active)
1182 if (conn->type == RX_SERVER_CONNECTION)
1183 rx_atomic_dec(&rx_stats.nServerConns);
1185 rx_atomic_dec(&rx_stats.nClientConns);
1188 if (conn->specific) {
1190 for (i = 0; i < conn->nSpecific; i++) {
1191 if (conn->specific[i] && rxi_keyCreate_destructor[i])
1192 (*rxi_keyCreate_destructor[i]) (conn->specific[i]);
1193 conn->specific[i] = NULL;
1195 free(conn->specific);
1197 conn->specific = NULL;
1198 conn->nSpecific = 0;
1199 #endif /* !KERNEL */
1201 MUTEX_DESTROY(&conn->conn_call_lock);
1202 MUTEX_DESTROY(&conn->conn_data_lock);
1203 CV_DESTROY(&conn->conn_call_cv);
1205 rxi_FreeConnection(conn);
1208 /* Destroy the specified connection */
1210 rxi_DestroyConnection(struct rx_connection *conn)
1212 MUTEX_ENTER(&rx_connHashTable_lock);
1213 rxi_DestroyConnectionNoLock(conn);
1214 /* conn should be at the head of the cleanup list */
1215 if (conn == rx_connCleanup_list) {
1216 rx_connCleanup_list = rx_connCleanup_list->next;
1217 MUTEX_EXIT(&rx_connHashTable_lock);
1218 rxi_CleanupConnection(conn);
1220 #ifdef RX_ENABLE_LOCKS
1222 MUTEX_EXIT(&rx_connHashTable_lock);
1224 #endif /* RX_ENABLE_LOCKS */
1228 rxi_DestroyConnectionNoLock(struct rx_connection *conn)
1230 struct rx_connection **conn_ptr;
1232 struct rx_packet *packet;
1239 MUTEX_ENTER(&conn->conn_data_lock);
1240 MUTEX_ENTER(&rx_refcnt_mutex);
1241 if (conn->refCount > 0)
1244 if (rx_stats_active) {
1245 MUTEX_ENTER(&rx_stats_mutex);
1246 rxi_lowConnRefCount++;
1247 MUTEX_EXIT(&rx_stats_mutex);
1251 if ((conn->refCount > 0) || (conn->flags & RX_CONN_BUSY)) {
1252 /* Busy; wait till the last guy before proceeding */
1253 MUTEX_EXIT(&rx_refcnt_mutex);
1254 MUTEX_EXIT(&conn->conn_data_lock);
1259 /* If the client previously called rx_NewCall, but it is still
1260 * waiting, treat this as a running call, and wait to destroy the
1261 * connection later when the call completes. */
1262 if ((conn->type == RX_CLIENT_CONNECTION)
1263 && (conn->flags & (RX_CONN_MAKECALL_WAITING|RX_CONN_MAKECALL_ACTIVE))) {
1264 conn->flags |= RX_CONN_DESTROY_ME;
1265 MUTEX_EXIT(&conn->conn_data_lock);
1269 MUTEX_EXIT(&rx_refcnt_mutex);
1270 MUTEX_EXIT(&conn->conn_data_lock);
1272 /* Check for extant references to this connection */
1273 MUTEX_ENTER(&conn->conn_call_lock);
1274 for (i = 0; i < RX_MAXCALLS; i++) {
1275 struct rx_call *call = conn->call[i];
1278 if (conn->type == RX_CLIENT_CONNECTION) {
1279 MUTEX_ENTER(&call->lock);
1280 if (call->delayedAckEvent) {
1281 /* Push the final acknowledgment out now--there
1282 * won't be a subsequent call to acknowledge the
1283 * last reply packets */
1284 rxi_CancelDelayedAckEvent(call);
1285 if (call->state == RX_STATE_PRECALL
1286 || call->state == RX_STATE_ACTIVE) {
1287 rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
1292 MUTEX_EXIT(&call->lock);
1296 MUTEX_EXIT(&conn->conn_call_lock);
1298 #ifdef RX_ENABLE_LOCKS
1300 if (MUTEX_TRYENTER(&conn->conn_data_lock)) {
1301 MUTEX_EXIT(&conn->conn_data_lock);
1303 /* Someone is accessing a packet right now. */
1307 #endif /* RX_ENABLE_LOCKS */
1310 /* Don't destroy the connection if there are any call
1311 * structures still in use */
1312 MUTEX_ENTER(&conn->conn_data_lock);
1313 conn->flags |= RX_CONN_DESTROY_ME;
1314 MUTEX_EXIT(&conn->conn_data_lock);
1319 if (conn->natKeepAliveEvent) {
1320 rxi_NatKeepAliveOff(conn);
1323 if (conn->delayedAbortEvent) {
1324 rxevent_Cancel(&conn->delayedAbortEvent);
1325 packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
1327 MUTEX_ENTER(&conn->conn_data_lock);
1328 rxi_SendConnectionAbort(conn, packet, 0, 1);
1329 MUTEX_EXIT(&conn->conn_data_lock);
1330 rxi_FreePacket(packet);
1334 /* Remove from connection hash table before proceeding */
1336 &rx_connHashTable[CONN_HASH
1337 (peer->host, peer->port, conn->cid, conn->epoch,
1339 for (; *conn_ptr; conn_ptr = &(*conn_ptr)->next) {
1340 if (*conn_ptr == conn) {
1341 *conn_ptr = conn->next;
1345 /* if the conn that we are destroying was the last connection, then we
1346 * clear rxLastConn as well */
1347 if (rxLastConn == conn)
1350 /* Make sure the connection is completely reset before deleting it. */
1351 /* get rid of pending events that could zap us later */
1352 rxevent_Cancel(&conn->challengeEvent);
1353 rxevent_Cancel(&conn->checkReachEvent);
1354 rxevent_Cancel(&conn->natKeepAliveEvent);
1356 /* Add the connection to the list of destroyed connections that
1357 * need to be cleaned up. This is necessary to avoid deadlocks
1358 * in the routines we call to inform others that this connection is
1359 * being destroyed. */
1360 conn->next = rx_connCleanup_list;
1361 rx_connCleanup_list = conn;
1364 /* Externally available version */
1366 rx_DestroyConnection(struct rx_connection *conn)
1371 rxi_DestroyConnection(conn);
1376 rx_GetConnection(struct rx_connection *conn)
1381 MUTEX_ENTER(&rx_refcnt_mutex);
1383 MUTEX_EXIT(&rx_refcnt_mutex);
1387 #ifdef RX_ENABLE_LOCKS
1388 /* Wait for the transmit queue to no longer be busy.
1389 * requires the call->lock to be held */
1391 rxi_WaitforTQBusy(struct rx_call *call) {
1392 while (!call->error && (call->flags & RX_CALL_TQ_BUSY)) {
1393 call->flags |= RX_CALL_TQ_WAIT;
1395 MUTEX_ASSERT(&call->lock);
1396 CV_WAIT(&call->cv_tq, &call->lock);
1398 if (call->tqWaiters == 0) {
1399 call->flags &= ~RX_CALL_TQ_WAIT;
1406 rxi_WakeUpTransmitQueue(struct rx_call *call)
1408 if (call->tqWaiters || (call->flags & RX_CALL_TQ_WAIT)) {
1409 dpf(("call %"AFS_PTR_FMT" has %d waiters and flags %d\n",
1410 call, call->tqWaiters, call->flags));
1411 #ifdef RX_ENABLE_LOCKS
1412 MUTEX_ASSERT(&call->lock);
1413 CV_BROADCAST(&call->cv_tq);
1414 #else /* RX_ENABLE_LOCKS */
1415 osi_rxWakeup(&call->tq);
1416 #endif /* RX_ENABLE_LOCKS */
1420 /* Start a new rx remote procedure call, on the specified connection.
1421 * If wait is set to 1, wait for a free call channel; otherwise return
1422 * 0. Maxtime gives the maximum number of seconds this call may take,
1423 * after rx_NewCall returns. After this time interval, a call to any
1424 * of rx_SendData, rx_ReadData, etc. will fail with RX_CALL_TIMEOUT.
1425 * For fine grain locking, we hold the conn_call_lock in order to
1426 * to ensure that we don't get signalle after we found a call in an active
1427 * state and before we go to sleep.
1430 rx_NewCall(struct rx_connection *conn)
1432 int i, wait, ignoreBusy = 1;
1433 struct rx_call *call;
1434 struct clock queueTime;
1435 afs_uint32 leastBusy = 0;
1439 dpf(("rx_NewCall(conn %"AFS_PTR_FMT")\n", conn));
1442 clock_GetTime(&queueTime);
1444 * Check if there are others waiting for a new call.
1445 * If so, let them go first to avoid starving them.
1446 * This is a fairly simple scheme, and might not be
1447 * a complete solution for large numbers of waiters.
1449 * makeCallWaiters keeps track of the number of
1450 * threads waiting to make calls and the
1451 * RX_CONN_MAKECALL_WAITING flag bit is used to
1452 * indicate that there are indeed calls waiting.
1453 * The flag is set when the waiter is incremented.
1454 * It is only cleared when makeCallWaiters is 0.
1455 * This prevents us from accidently destroying the
1456 * connection while it is potentially about to be used.
1458 MUTEX_ENTER(&conn->conn_call_lock);
1459 MUTEX_ENTER(&conn->conn_data_lock);
1460 while (conn->flags & RX_CONN_MAKECALL_ACTIVE) {
1461 conn->flags |= RX_CONN_MAKECALL_WAITING;
1462 conn->makeCallWaiters++;
1463 MUTEX_EXIT(&conn->conn_data_lock);
1465 #ifdef RX_ENABLE_LOCKS
1466 CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
1470 MUTEX_ENTER(&conn->conn_data_lock);
1471 conn->makeCallWaiters--;
1472 if (conn->makeCallWaiters == 0)
1473 conn->flags &= ~RX_CONN_MAKECALL_WAITING;
1476 /* We are now the active thread in rx_NewCall */
1477 conn->flags |= RX_CONN_MAKECALL_ACTIVE;
1478 MUTEX_EXIT(&conn->conn_data_lock);
1483 for (i = 0; i < RX_MAXCALLS; i++) {
1484 call = conn->call[i];
1486 if (!ignoreBusy && conn->lastBusy[i] != leastBusy) {
1487 /* we're not ignoring busy call slots; only look at the
1488 * call slot that is the "least" busy */
1492 if (call->state == RX_STATE_DALLY) {
1493 MUTEX_ENTER(&call->lock);
1494 if (call->state == RX_STATE_DALLY) {
1495 if (ignoreBusy && conn->lastBusy[i]) {
1496 /* if we're ignoring busy call slots, skip any ones that
1497 * have lastBusy set */
1498 if (leastBusy == 0 || conn->lastBusy[i] < leastBusy) {
1499 leastBusy = conn->lastBusy[i];
1501 MUTEX_EXIT(&call->lock);
1506 * We are setting the state to RX_STATE_RESET to
1507 * ensure that no one else will attempt to use this
1508 * call once we drop the conn->conn_call_lock and
1509 * call->lock. We must drop the conn->conn_call_lock
1510 * before calling rxi_ResetCall because the process
1511 * of clearing the transmit queue can block for an
1512 * extended period of time. If we block while holding
1513 * the conn->conn_call_lock, then all rx_EndCall
1514 * processing will block as well. This has a detrimental
1515 * effect on overall system performance.
1517 call->state = RX_STATE_RESET;
1518 (*call->callNumber)++;
1519 MUTEX_EXIT(&conn->conn_call_lock);
1520 CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
1521 rxi_ResetCall(call, 0);
1522 if (MUTEX_TRYENTER(&conn->conn_call_lock))
1526 * If we failed to be able to safely obtain the
1527 * conn->conn_call_lock we will have to drop the
1528 * call->lock to avoid a deadlock. When the call->lock
1529 * is released the state of the call can change. If it
1530 * is no longer RX_STATE_RESET then some other thread is
1533 MUTEX_EXIT(&call->lock);
1534 MUTEX_ENTER(&conn->conn_call_lock);
1535 MUTEX_ENTER(&call->lock);
1537 if (call->state == RX_STATE_RESET)
1541 * If we get here it means that after dropping
1542 * the conn->conn_call_lock and call->lock that
1543 * the call is no longer ours. If we can't find
1544 * a free call in the remaining slots we should
1545 * not go immediately to RX_CONN_MAKECALL_WAITING
1546 * because by dropping the conn->conn_call_lock
1547 * we have given up synchronization with rx_EndCall.
1548 * Instead, cycle through one more time to see if
1549 * we can find a call that can call our own.
1551 CALL_RELE(call, RX_CALL_REFCOUNT_BEGIN);
1554 MUTEX_EXIT(&call->lock);
1557 if (ignoreBusy && conn->lastBusy[i]) {
1558 /* if we're ignoring busy call slots, skip any ones that
1559 * have lastBusy set */
1560 if (leastBusy == 0 || conn->lastBusy[i] < leastBusy) {
1561 leastBusy = conn->lastBusy[i];
1566 /* rxi_NewCall returns with mutex locked */
1567 call = rxi_NewCall(conn, i);
1568 CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
1572 if (i < RX_MAXCALLS) {
1573 conn->lastBusy[i] = 0;
1574 call->flags &= ~RX_CALL_PEER_BUSY;
1579 if (leastBusy && ignoreBusy) {
1580 /* we didn't find a useable call slot, but we did see at least one
1581 * 'busy' slot; look again and only use a slot with the 'least
1587 MUTEX_ENTER(&conn->conn_data_lock);
1588 conn->flags |= RX_CONN_MAKECALL_WAITING;
1589 conn->makeCallWaiters++;
1590 MUTEX_EXIT(&conn->conn_data_lock);
1592 #ifdef RX_ENABLE_LOCKS
1593 CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
1597 MUTEX_ENTER(&conn->conn_data_lock);
1598 conn->makeCallWaiters--;
1599 if (conn->makeCallWaiters == 0)
1600 conn->flags &= ~RX_CONN_MAKECALL_WAITING;
1601 MUTEX_EXIT(&conn->conn_data_lock);
1603 /* Client is initially in send mode */
1604 call->state = RX_STATE_ACTIVE;
1605 call->error = conn->error;
1607 call->app.mode = RX_MODE_ERROR;
1609 call->app.mode = RX_MODE_SENDING;
1611 #ifdef AFS_RXERRQ_ENV
1612 /* remember how many network errors the peer has when we started, so if
1613 * more errors are encountered after the call starts, we know the other endpoint won't be
1614 * responding to us */
1615 call->neterr_gen = rx_atomic_read(&conn->peer->neterrs);
1618 /* remember start time for call in case we have hard dead time limit */
1619 call->queueTime = queueTime;
1620 clock_GetTime(&call->startTime);
1621 call->app.bytesSent = 0;
1622 call->app.bytesRcvd = 0;
1624 /* Turn on busy protocol. */
1625 rxi_KeepAliveOn(call);
1627 /* Attempt MTU discovery */
1628 rxi_GrowMTUOn(call);
1631 * We are no longer the active thread in rx_NewCall
1633 MUTEX_ENTER(&conn->conn_data_lock);
1634 conn->flags &= ~RX_CONN_MAKECALL_ACTIVE;
1635 MUTEX_EXIT(&conn->conn_data_lock);
1638 * Wake up anyone else who might be giving us a chance to
1639 * run (see code above that avoids resource starvation).
1641 #ifdef RX_ENABLE_LOCKS
1642 if (call->flags & (RX_CALL_TQ_BUSY | RX_CALL_TQ_CLEARME)) {
1643 osi_Panic("rx_NewCall call about to be used without an empty tq");
1646 CV_BROADCAST(&conn->conn_call_cv);
1650 MUTEX_EXIT(&conn->conn_call_lock);
1651 MUTEX_EXIT(&call->lock);
1654 dpf(("rx_NewCall(call %"AFS_PTR_FMT")\n", call));
1659 rxi_HasActiveCalls(struct rx_connection *aconn)
1662 struct rx_call *tcall;
1666 for (i = 0; i < RX_MAXCALLS; i++) {
1667 if ((tcall = aconn->call[i])) {
1668 if ((tcall->state == RX_STATE_ACTIVE)
1669 || (tcall->state == RX_STATE_PRECALL)) {
1680 rxi_GetCallNumberVector(struct rx_connection *aconn,
1681 afs_int32 * aint32s)
1684 struct rx_call *tcall;
1688 MUTEX_ENTER(&aconn->conn_call_lock);
1689 for (i = 0; i < RX_MAXCALLS; i++) {
1690 if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
1691 aint32s[i] = aconn->callNumber[i] + 1;
1693 aint32s[i] = aconn->callNumber[i];
1695 MUTEX_EXIT(&aconn->conn_call_lock);
1701 rxi_SetCallNumberVector(struct rx_connection *aconn,
1702 afs_int32 * aint32s)
1705 struct rx_call *tcall;
1709 MUTEX_ENTER(&aconn->conn_call_lock);
1710 for (i = 0; i < RX_MAXCALLS; i++) {
1711 if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
1712 aconn->callNumber[i] = aint32s[i] - 1;
1714 aconn->callNumber[i] = aint32s[i];
1716 MUTEX_EXIT(&aconn->conn_call_lock);
1721 /* Advertise a new service. A service is named locally by a UDP port
1722 * number plus a 16-bit service id. Returns (struct rx_service *) 0
1725 char *serviceName; Name for identification purposes (e.g. the
1726 service name might be used for probing for
1729 rx_NewServiceHost(afs_uint32 host, u_short port, u_short serviceId,
1730 char *serviceName, struct rx_securityClass **securityObjects,
1731 int nSecurityObjects,
1732 afs_int32(*serviceProc) (struct rx_call * acall))
1734 osi_socket socket = OSI_NULLSOCKET;
1735 struct rx_service *tservice;
1741 if (serviceId == 0) {
1743 "rx_NewService: service id for service %s is not non-zero.\n",
1750 "rx_NewService: A non-zero port must be specified on this call if a non-zero port was not provided at Rx initialization (service %s).\n",
1758 tservice = rxi_AllocService();
1761 MUTEX_INIT(&tservice->svc_data_lock, "svc data lock", MUTEX_DEFAULT, 0);
1763 for (i = 0; i < RX_MAX_SERVICES; i++) {
1764 struct rx_service *service = rx_services[i];
1766 if (port == service->servicePort && host == service->serviceHost) {
1767 if (service->serviceId == serviceId) {
1768 /* The identical service has already been
1769 * installed; if the caller was intending to
1770 * change the security classes used by this
1771 * service, he/she loses. */
1773 "rx_NewService: tried to install service %s with service id %d, which is already in use for service %s\n",
1774 serviceName, serviceId, service->serviceName);
1776 rxi_FreeService(tservice);
1779 /* Different service, same port: re-use the socket
1780 * which is bound to the same port */
1781 socket = service->socket;
1784 if (socket == OSI_NULLSOCKET) {
1785 /* If we don't already have a socket (from another
1786 * service on same port) get a new one */
1787 socket = rxi_GetHostUDPSocket(host, port);
1788 if (socket == OSI_NULLSOCKET) {
1790 rxi_FreeService(tservice);
1795 service->socket = socket;
1796 service->serviceHost = host;
1797 service->servicePort = port;
1798 service->serviceId = serviceId;
1799 service->serviceName = serviceName;
1800 service->nSecurityObjects = nSecurityObjects;
1801 service->securityObjects = securityObjects;
1802 service->minProcs = 0;
1803 service->maxProcs = 1;
1804 service->idleDeadTime = 60;
1805 service->connDeadTime = rx_connDeadTime;
1806 service->executeRequestProc = serviceProc;
1807 service->checkReach = 0;
1808 service->nSpecific = 0;
1809 service->specific = NULL;
1810 rx_services[i] = service; /* not visible until now */
1816 rxi_FreeService(tservice);
1817 (osi_Msg "rx_NewService: cannot support > %d services\n",
1822 /* Set configuration options for all of a service's security objects */
1825 rx_SetSecurityConfiguration(struct rx_service *service,
1826 rx_securityConfigVariables type,
1830 for (i = 0; i<service->nSecurityObjects; i++) {
1831 if (service->securityObjects[i]) {
1832 RXS_SetConfiguration(service->securityObjects[i], NULL, type,
1840 rx_NewService(u_short port, u_short serviceId, char *serviceName,
1841 struct rx_securityClass **securityObjects, int nSecurityObjects,
1842 afs_int32(*serviceProc) (struct rx_call * acall))
1844 return rx_NewServiceHost(htonl(INADDR_ANY), port, serviceId, serviceName, securityObjects, nSecurityObjects, serviceProc);
1847 /* Generic request processing loop. This routine should be called
1848 * by the implementation dependent rx_ServerProc. If socketp is
1849 * non-null, it will be set to the file descriptor that this thread
1850 * is now listening on. If socketp is null, this routine will never
1853 rxi_ServerProc(int threadID, struct rx_call *newcall, osi_socket * socketp)
1855 struct rx_call *call;
1857 struct rx_service *tservice = NULL;
1864 call = rx_GetCall(threadID, tservice, socketp);
1865 if (socketp && *socketp != OSI_NULLSOCKET) {
1866 /* We are now a listener thread */
1872 if (afs_termState == AFSOP_STOP_RXCALLBACK) {
1873 #ifdef RX_ENABLE_LOCKS
1875 #endif /* RX_ENABLE_LOCKS */
1876 afs_termState = AFSOP_STOP_AFS;
1877 afs_osi_Wakeup(&afs_termState);
1878 #ifdef RX_ENABLE_LOCKS
1880 #endif /* RX_ENABLE_LOCKS */
1885 /* if server is restarting( typically smooth shutdown) then do not
1886 * allow any new calls.
1889 if (rx_tranquil && (call != NULL)) {
1893 MUTEX_ENTER(&call->lock);
1895 rxi_CallError(call, RX_RESTARTING);
1896 rxi_SendCallAbort(call, (struct rx_packet *)0, 0, 0);
1898 MUTEX_EXIT(&call->lock);
1903 tservice = call->conn->service;
1905 if (tservice->beforeProc)
1906 (*tservice->beforeProc) (call);
1908 code = tservice->executeRequestProc(call);
1910 if (tservice->afterProc)
1911 (*tservice->afterProc) (call, code);
1913 rx_EndCall(call, code);
1915 if (tservice->postProc)
1916 (*tservice->postProc) (code);
1918 if (rx_stats_active) {
1919 MUTEX_ENTER(&rx_stats_mutex);
1921 MUTEX_EXIT(&rx_stats_mutex);
1928 rx_WakeupServerProcs(void)
1930 struct rx_serverQueueEntry *np, *tqp;
1931 struct opr_queue *cursor;
1935 MUTEX_ENTER(&rx_serverPool_lock);
1937 #ifdef RX_ENABLE_LOCKS
1938 if (rx_waitForPacket)
1939 CV_BROADCAST(&rx_waitForPacket->cv);
1940 #else /* RX_ENABLE_LOCKS */
1941 if (rx_waitForPacket)
1942 osi_rxWakeup(rx_waitForPacket);
1943 #endif /* RX_ENABLE_LOCKS */
1944 MUTEX_ENTER(&freeSQEList_lock);
1945 for (np = rx_FreeSQEList; np; np = tqp) {
1946 tqp = *(struct rx_serverQueueEntry **)np;
1947 #ifdef RX_ENABLE_LOCKS
1948 CV_BROADCAST(&np->cv);
1949 #else /* RX_ENABLE_LOCKS */
1951 #endif /* RX_ENABLE_LOCKS */
1953 MUTEX_EXIT(&freeSQEList_lock);
1954 for (opr_queue_Scan(&rx_idleServerQueue, cursor)) {
1955 np = opr_queue_Entry(cursor, struct rx_serverQueueEntry, entry);
1956 #ifdef RX_ENABLE_LOCKS
1957 CV_BROADCAST(&np->cv);
1958 #else /* RX_ENABLE_LOCKS */
1960 #endif /* RX_ENABLE_LOCKS */
1962 MUTEX_EXIT(&rx_serverPool_lock);
1967 * One thing that seems to happen is that all the server threads get
1968 * tied up on some empty or slow call, and then a whole bunch of calls
1969 * arrive at once, using up the packet pool, so now there are more
1970 * empty calls. The most critical resources here are server threads
1971 * and the free packet pool. The "doreclaim" code seems to help in
1972 * general. I think that eventually we arrive in this state: there
1973 * are lots of pending calls which do have all their packets present,
1974 * so they won't be reclaimed, are multi-packet calls, so they won't
1975 * be scheduled until later, and thus are tying up most of the free
1976 * packet pool for a very long time.
1978 * 1. schedule multi-packet calls if all the packets are present.
1979 * Probably CPU-bound operation, useful to return packets to pool.
1980 * Do what if there is a full window, but the last packet isn't here?
1981 * 3. preserve one thread which *only* runs "best" calls, otherwise
1982 * it sleeps and waits for that type of call.
1983 * 4. Don't necessarily reserve a whole window for each thread. In fact,
1984 * the current dataquota business is badly broken. The quota isn't adjusted
1985 * to reflect how many packets are presently queued for a running call.
1986 * So, when we schedule a queued call with a full window of packets queued
1987 * up for it, that *should* free up a window full of packets for other 2d-class
1988 * calls to be able to use from the packet pool. But it doesn't.
1990 * NB. Most of the time, this code doesn't run -- since idle server threads
1991 * sit on the idle server queue and are assigned by "...ReceivePacket" as soon
1992 * as a new call arrives.
1994 /* Sleep until a call arrives. Returns a pointer to the call, ready
1995 * for an rx_Read. */
1996 #ifdef RX_ENABLE_LOCKS
1998 rx_GetCall(int tno, struct rx_service *cur_service, osi_socket * socketp)
2000 struct rx_serverQueueEntry *sq;
2001 struct rx_call *call = (struct rx_call *)0;
2002 struct rx_service *service = NULL;
2004 MUTEX_ENTER(&freeSQEList_lock);
2006 if ((sq = rx_FreeSQEList)) {
2007 rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
2008 MUTEX_EXIT(&freeSQEList_lock);
2009 } else { /* otherwise allocate a new one and return that */
2010 MUTEX_EXIT(&freeSQEList_lock);
2011 sq = rxi_Alloc(sizeof(struct rx_serverQueueEntry));
2012 MUTEX_INIT(&sq->lock, "server Queue lock", MUTEX_DEFAULT, 0);
2013 CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
2016 MUTEX_ENTER(&rx_serverPool_lock);
2017 if (cur_service != NULL) {
2018 ReturnToServerPool(cur_service);
2021 if (!opr_queue_IsEmpty(&rx_incomingCallQueue)) {
2022 struct rx_call *tcall, *choice2 = NULL;
2023 struct opr_queue *cursor;
2025 /* Scan for eligible incoming calls. A call is not eligible
2026 * if the maximum number of calls for its service type are
2027 * already executing */
2028 /* One thread will process calls FCFS (to prevent starvation),
2029 * while the other threads may run ahead looking for calls which
2030 * have all their input data available immediately. This helps
2031 * keep threads from blocking, waiting for data from the client. */
2032 for (opr_queue_Scan(&rx_incomingCallQueue, cursor)) {
2033 tcall = opr_queue_Entry(cursor, struct rx_call, entry);
2035 service = tcall->conn->service;
2036 if (!QuotaOK(service)) {
2039 MUTEX_ENTER(&rx_pthread_mutex);
2040 if (tno == rxi_fcfs_thread_num
2041 || opr_queue_IsEnd(&rx_incomingCallQueue, cursor)) {
2042 MUTEX_EXIT(&rx_pthread_mutex);
2043 /* If we're the fcfs thread , then we'll just use
2044 * this call. If we haven't been able to find an optimal
2045 * choice, and we're at the end of the list, then use a
2046 * 2d choice if one has been identified. Otherwise... */
2047 call = (choice2 ? choice2 : tcall);
2048 service = call->conn->service;
2050 MUTEX_EXIT(&rx_pthread_mutex);
2051 if (!opr_queue_IsEmpty(&tcall->rq)) {
2052 struct rx_packet *rp;
2053 rp = opr_queue_First(&tcall->rq, struct rx_packet,
2055 if (rp->header.seq == 1) {
2057 || (rp->header.flags & RX_LAST_PACKET)) {
2059 } else if (rxi_2dchoice && !choice2
2060 && !(tcall->flags & RX_CALL_CLEARED)
2061 && (tcall->rprev > rxi_HardAckRate)) {
2071 ReturnToServerPool(service);
2077 opr_queue_Remove(&call->entry);
2078 MUTEX_EXIT(&rx_serverPool_lock);
2079 MUTEX_ENTER(&call->lock);
2081 if (call->flags & RX_CALL_WAIT_PROC) {
2082 call->flags &= ~RX_CALL_WAIT_PROC;
2083 rx_atomic_dec(&rx_nWaiting);
2086 if (call->state != RX_STATE_PRECALL || call->error) {
2087 MUTEX_EXIT(&call->lock);
2088 MUTEX_ENTER(&rx_serverPool_lock);
2089 ReturnToServerPool(service);
2094 if (opr_queue_IsEmpty(&call->rq)
2095 || opr_queue_First(&call->rq, struct rx_packet, entry)->header.seq != 1)
2096 rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
2098 CLEAR_CALL_QUEUE_LOCK(call);
2101 /* If there are no eligible incoming calls, add this process
2102 * to the idle server queue, to wait for one */
2106 *socketp = OSI_NULLSOCKET;
2108 sq->socketp = socketp;
2109 opr_queue_Append(&rx_idleServerQueue, &sq->entry);
2110 #ifndef AFS_AIX41_ENV
2111 rx_waitForPacket = sq;
2112 #endif /* AFS_AIX41_ENV */
2114 CV_WAIT(&sq->cv, &rx_serverPool_lock);
2116 if (afs_termState == AFSOP_STOP_RXCALLBACK) {
2117 MUTEX_EXIT(&rx_serverPool_lock);
2118 return (struct rx_call *)0;
2121 } while (!(call = sq->newcall)
2122 && !(socketp && *socketp != OSI_NULLSOCKET));
2123 MUTEX_EXIT(&rx_serverPool_lock);
2125 MUTEX_ENTER(&call->lock);
2131 MUTEX_ENTER(&freeSQEList_lock);
2132 *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
2133 rx_FreeSQEList = sq;
2134 MUTEX_EXIT(&freeSQEList_lock);
2137 clock_GetTime(&call->startTime);
2138 call->state = RX_STATE_ACTIVE;
2139 call->app.mode = RX_MODE_RECEIVING;
2140 #ifdef RX_KERNEL_TRACE
2141 if (ICL_SETACTIVE(afs_iclSetp)) {
2142 int glockOwner = ISAFS_GLOCK();
2145 afs_Trace3(afs_iclSetp, CM_TRACE_WASHERE, ICL_TYPE_STRING,
2146 __FILE__, ICL_TYPE_INT32, __LINE__, ICL_TYPE_POINTER,
2153 rxi_calltrace(RX_CALL_START, call);
2154 dpf(("rx_GetCall(port=%d, service=%d) ==> call %"AFS_PTR_FMT"\n",
2155 call->conn->service->servicePort, call->conn->service->serviceId,
2158 MUTEX_EXIT(&call->lock);
2159 CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
2161 dpf(("rx_GetCall(socketp=%p, *socketp=0x%x)\n", socketp, *socketp));
2166 #else /* RX_ENABLE_LOCKS */
2168 rx_GetCall(int tno, struct rx_service *cur_service, osi_socket * socketp)
2170 struct rx_serverQueueEntry *sq;
2171 struct rx_call *call = (struct rx_call *)0, *choice2;
2172 struct rx_service *service = NULL;
2176 MUTEX_ENTER(&freeSQEList_lock);
2178 if ((sq = rx_FreeSQEList)) {
2179 rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
2180 MUTEX_EXIT(&freeSQEList_lock);
2181 } else { /* otherwise allocate a new one and return that */
2182 MUTEX_EXIT(&freeSQEList_lock);
2183 sq = rxi_Alloc(sizeof(struct rx_serverQueueEntry));
2184 MUTEX_INIT(&sq->lock, "server Queue lock", MUTEX_DEFAULT, 0);
2185 CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
2187 MUTEX_ENTER(&sq->lock);
2189 if (cur_service != NULL) {
2190 cur_service->nRequestsRunning--;
2191 MUTEX_ENTER(&rx_quota_mutex);
2192 if (cur_service->nRequestsRunning < cur_service->minProcs)
2195 MUTEX_EXIT(&rx_quota_mutex);
2197 if (!opr_queue_IsEmpty(&rx_incomingCallQueue)) {
2198 struct rx_call *tcall;
2199 struct opr_queue *cursor;
2200 /* Scan for eligible incoming calls. A call is not eligible
2201 * if the maximum number of calls for its service type are
2202 * already executing */
2203 /* One thread will process calls FCFS (to prevent starvation),
2204 * while the other threads may run ahead looking for calls which
2205 * have all their input data available immediately. This helps
2206 * keep threads from blocking, waiting for data from the client. */
2207 choice2 = (struct rx_call *)0;
2208 for (opr_queue_Scan(&rx_incomingCallQueue, cursor)) {
2209 tcall = opr_queue_Entry(cursor, struct rx_call, entry);
2210 service = tcall->conn->service;
2211 if (QuotaOK(service)) {
2212 MUTEX_ENTER(&rx_pthread_mutex);
2213 /* XXX - If tcall->entry.next is NULL, then we're no longer
2214 * on a queue at all. This shouldn't happen. */
2215 if (tno == rxi_fcfs_thread_num || !tcall->entry.next) {
2216 MUTEX_EXIT(&rx_pthread_mutex);
2217 /* If we're the fcfs thread, then we'll just use
2218 * this call. If we haven't been able to find an optimal
2219 * choice, and we're at the end of the list, then use a
2220 * 2d choice if one has been identified. Otherwise... */
2221 call = (choice2 ? choice2 : tcall);
2222 service = call->conn->service;
2224 MUTEX_EXIT(&rx_pthread_mutex);
2225 if (!opr_queue_IsEmpty(&tcall->rq)) {
2226 struct rx_packet *rp;
2227 rp = opr_queue_First(&tcall->rq, struct rx_packet,
2229 if (rp->header.seq == 1
2231 || (rp->header.flags & RX_LAST_PACKET))) {
2233 } else if (rxi_2dchoice && !choice2
2234 && !(tcall->flags & RX_CALL_CLEARED)
2235 && (tcall->rprev > rxi_HardAckRate)) {
2248 opr_queue_Remove(&call->entry);
2249 /* we can't schedule a call if there's no data!!! */
2250 /* send an ack if there's no data, if we're missing the
2251 * first packet, or we're missing something between first
2252 * and last -- there's a "hole" in the incoming data. */
2253 if (opr_queue_IsEmpty(&call->rq)
2254 || opr_queue_First(&call->rq, struct rx_packet, entry)->header.seq != 1
2255 || call->rprev != opr_queue_Last(&call->rq, struct rx_packet, entry)->header.seq)
2256 rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
2258 call->flags &= (~RX_CALL_WAIT_PROC);
2259 service->nRequestsRunning++;
2260 /* just started call in minProcs pool, need fewer to maintain
2262 MUTEX_ENTER(&rx_quota_mutex);
2263 if (service->nRequestsRunning <= service->minProcs)
2266 MUTEX_EXIT(&rx_quota_mutex);
2267 rx_atomic_dec(&rx_nWaiting);
2268 /* MUTEX_EXIT(&call->lock); */
2270 /* If there are no eligible incoming calls, add this process
2271 * to the idle server queue, to wait for one */
2274 *socketp = OSI_NULLSOCKET;
2276 sq->socketp = socketp;
2277 opr_queue_Append(&rx_idleServerQueue, &sq->entry);
2281 if (afs_termState == AFSOP_STOP_RXCALLBACK) {
2283 rxi_Free(sq, sizeof(struct rx_serverQueueEntry));
2284 return (struct rx_call *)0;
2287 } while (!(call = sq->newcall)
2288 && !(socketp && *socketp != OSI_NULLSOCKET));
2290 MUTEX_EXIT(&sq->lock);
2292 MUTEX_ENTER(&freeSQEList_lock);
2293 *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
2294 rx_FreeSQEList = sq;
2295 MUTEX_EXIT(&freeSQEList_lock);
2298 clock_GetTime(&call->startTime);
2299 call->state = RX_STATE_ACTIVE;
2300 call->app.mode = RX_MODE_RECEIVING;
2301 #ifdef RX_KERNEL_TRACE
2302 if (ICL_SETACTIVE(afs_iclSetp)) {
2303 int glockOwner = ISAFS_GLOCK();
2306 afs_Trace3(afs_iclSetp, CM_TRACE_WASHERE, ICL_TYPE_STRING,
2307 __FILE__, ICL_TYPE_INT32, __LINE__, ICL_TYPE_POINTER,
2314 rxi_calltrace(RX_CALL_START, call);
2315 dpf(("rx_GetCall(port=%d, service=%d) ==> call %p\n",
2316 call->conn->service->servicePort, call->conn->service->serviceId,
2319 dpf(("rx_GetCall(socketp=%p, *socketp=0x%x)\n", socketp, *socketp));
2326 #endif /* RX_ENABLE_LOCKS */
2330 /* Establish a procedure to be called when a packet arrives for a
2331 * call. This routine will be called at most once after each call,
2332 * and will also be called if there is an error condition on the or
2333 * the call is complete. Used by multi rx to build a selection
2334 * function which determines which of several calls is likely to be a
2335 * good one to read from.
2336 * NOTE: the way this is currently implemented it is probably only a
2337 * good idea to (1) use it immediately after a newcall (clients only)
2338 * and (2) only use it once. Other uses currently void your warranty
2341 rx_SetArrivalProc(struct rx_call *call,
2342 void (*proc) (struct rx_call * call,
2345 void * handle, int arg)
2347 call->arrivalProc = proc;
2348 call->arrivalProcHandle = handle;
2349 call->arrivalProcArg = arg;
2352 /* Call is finished (possibly prematurely). Return rc to the peer, if
2353 * appropriate, and return the final error code from the conversation
2357 rx_EndCall(struct rx_call *call, afs_int32 rc)
2359 struct rx_connection *conn = call->conn;
2363 dpf(("rx_EndCall(call %"AFS_PTR_FMT" rc %d error %d abortCode %d)\n",
2364 call, rc, call->error, call->abortCode));
2367 MUTEX_ENTER(&call->lock);
2369 if (rc == 0 && call->error == 0) {
2370 call->abortCode = 0;
2371 call->abortCount = 0;
2374 call->arrivalProc = (void (*)())0;
2375 if (rc && call->error == 0) {
2376 rxi_CallError(call, rc);
2377 call->app.mode = RX_MODE_ERROR;
2378 /* Send an abort message to the peer if this error code has
2379 * only just been set. If it was set previously, assume the
2380 * peer has already been sent the error code or will request it
2382 rxi_SendCallAbort(call, (struct rx_packet *)0, 0, 0);
2384 if (conn->type == RX_SERVER_CONNECTION) {
2385 /* Make sure reply or at least dummy reply is sent */
2386 if (call->app.mode == RX_MODE_RECEIVING) {
2387 MUTEX_EXIT(&call->lock);
2388 rxi_WriteProc(call, 0, 0);
2389 MUTEX_ENTER(&call->lock);
2391 if (call->app.mode == RX_MODE_SENDING) {
2392 MUTEX_EXIT(&call->lock);
2393 rxi_FlushWrite(call);
2394 MUTEX_ENTER(&call->lock);
2396 rxi_calltrace(RX_CALL_END, call);
2397 /* Call goes to hold state until reply packets are acknowledged */
2398 if (call->tfirst + call->nSoftAcked < call->tnext) {
2399 call->state = RX_STATE_HOLD;
2401 call->state = RX_STATE_DALLY;
2402 rxi_ClearTransmitQueue(call, 0);
2403 rxi_rto_cancel(call);
2404 rxi_CancelKeepAliveEvent(call);
2406 } else { /* Client connection */
2408 /* Make sure server receives input packets, in the case where
2409 * no reply arguments are expected */
2411 if ((call->app.mode == RX_MODE_SENDING)
2412 || (call->app.mode == RX_MODE_RECEIVING && call->rnext == 1)) {
2413 MUTEX_EXIT(&call->lock);
2414 (void)rxi_ReadProc(call, &dummy, 1);
2415 MUTEX_ENTER(&call->lock);
2418 /* If we had an outstanding delayed ack, be nice to the server
2419 * and force-send it now.
2421 if (call->delayedAckEvent) {
2422 rxi_CancelDelayedAckEvent(call);
2423 rxi_SendDelayedAck(NULL, call, NULL, 0);
2426 /* We need to release the call lock since it's lower than the
2427 * conn_call_lock and we don't want to hold the conn_call_lock
2428 * over the rx_ReadProc call. The conn_call_lock needs to be held
2429 * here for the case where rx_NewCall is perusing the calls on
2430 * the connection structure. We don't want to signal until
2431 * rx_NewCall is in a stable state. Otherwise, rx_NewCall may
2432 * have checked this call, found it active and by the time it
2433 * goes to sleep, will have missed the signal.
2435 MUTEX_EXIT(&call->lock);
2436 MUTEX_ENTER(&conn->conn_call_lock);
2437 MUTEX_ENTER(&call->lock);
2439 if (!(call->flags & RX_CALL_PEER_BUSY)) {
2440 conn->lastBusy[call->channel] = 0;
2443 MUTEX_ENTER(&conn->conn_data_lock);
2444 conn->flags |= RX_CONN_BUSY;
2445 if (conn->flags & RX_CONN_MAKECALL_WAITING) {
2446 MUTEX_EXIT(&conn->conn_data_lock);
2447 #ifdef RX_ENABLE_LOCKS
2448 CV_BROADCAST(&conn->conn_call_cv);
2453 #ifdef RX_ENABLE_LOCKS
2455 MUTEX_EXIT(&conn->conn_data_lock);
2457 #endif /* RX_ENABLE_LOCKS */
2458 call->state = RX_STATE_DALLY;
2460 error = call->error;
2462 /* currentPacket, nLeft, and NFree must be zeroed here, because
2463 * ResetCall cannot: ResetCall may be called at splnet(), in the
2464 * kernel version, and may interrupt the macros rx_Read or
2465 * rx_Write, which run at normal priority for efficiency. */
2466 if (call->app.currentPacket) {
2467 #ifdef RX_TRACK_PACKETS
2468 call->app.currentPacket->flags &= ~RX_PKTFLAG_CP;
2470 rxi_FreePacket(call->app.currentPacket);
2471 call->app.currentPacket = (struct rx_packet *)0;
2474 call->app.nLeft = call->app.nFree = call->app.curlen = 0;
2476 /* Free any packets from the last call to ReadvProc/WritevProc */
2477 #ifdef RXDEBUG_PACKET
2479 #endif /* RXDEBUG_PACKET */
2480 rxi_FreePackets(0, &call->app.iovq);
2481 MUTEX_EXIT(&call->lock);
2483 CALL_RELE(call, RX_CALL_REFCOUNT_BEGIN);
2484 if (conn->type == RX_CLIENT_CONNECTION) {
2485 MUTEX_ENTER(&conn->conn_data_lock);
2486 conn->flags &= ~RX_CONN_BUSY;
2487 MUTEX_EXIT(&conn->conn_data_lock);
2488 MUTEX_EXIT(&conn->conn_call_lock);
2492 * Map errors to the local host's errno.h format.
2494 error = ntoh_syserr_conv(error);
2496 /* If the caller said the call failed with some error, we had better
2497 * return an error code. */
2498 osi_Assert(!rc || error);
2502 #if !defined(KERNEL)
2504 /* Call this routine when shutting down a server or client (especially
2505 * clients). This will allow Rx to gracefully garbage collect server
2506 * connections, and reduce the number of retries that a server might
2507 * make to a dead client.
2508 * This is not quite right, since some calls may still be ongoing and
2509 * we can't lock them to destroy them. */
2513 struct rx_connection **conn_ptr, **conn_end;
2516 if (rx_atomic_test_and_set_bit(&rxinit_status, 0))
2517 return; /* Already shutdown. */
2519 rxi_DeleteCachedConnections();
2520 if (rx_connHashTable) {
2521 MUTEX_ENTER(&rx_connHashTable_lock);
2522 for (conn_ptr = &rx_connHashTable[0], conn_end =
2523 &rx_connHashTable[rx_hashTableSize]; conn_ptr < conn_end;
2525 struct rx_connection *conn, *next;
2526 for (conn = *conn_ptr; conn; conn = next) {
2528 if (conn->type == RX_CLIENT_CONNECTION) {
2529 MUTEX_ENTER(&rx_refcnt_mutex);
2531 MUTEX_EXIT(&rx_refcnt_mutex);
2532 #ifdef RX_ENABLE_LOCKS
2533 rxi_DestroyConnectionNoLock(conn);
2534 #else /* RX_ENABLE_LOCKS */
2535 rxi_DestroyConnection(conn);
2536 #endif /* RX_ENABLE_LOCKS */
2540 #ifdef RX_ENABLE_LOCKS
2541 while (rx_connCleanup_list) {
2542 struct rx_connection *conn;
2543 conn = rx_connCleanup_list;
2544 rx_connCleanup_list = rx_connCleanup_list->next;
2545 MUTEX_EXIT(&rx_connHashTable_lock);
2546 rxi_CleanupConnection(conn);
2547 MUTEX_ENTER(&rx_connHashTable_lock);
2549 MUTEX_EXIT(&rx_connHashTable_lock);
2550 #endif /* RX_ENABLE_LOCKS */
2555 afs_winsockCleanup();
2561 /* if we wakeup packet waiter too often, can get in loop with two
2562 AllocSendPackets each waking each other up (from ReclaimPacket calls) */
2564 rxi_PacketsUnWait(void)
2566 if (!rx_waitingForPackets) {
2570 if (rxi_OverQuota(RX_PACKET_CLASS_SEND)) {
2571 return; /* still over quota */
2574 rx_waitingForPackets = 0;
2575 #ifdef RX_ENABLE_LOCKS
2576 CV_BROADCAST(&rx_waitingForPackets_cv);
2578 osi_rxWakeup(&rx_waitingForPackets);
2584 /* ------------------Internal interfaces------------------------- */
2586 /* Return this process's service structure for the
2587 * specified socket and service */
2588 static struct rx_service *
2589 rxi_FindService(osi_socket socket, u_short serviceId)
2591 struct rx_service **sp;
2592 for (sp = &rx_services[0]; *sp; sp++) {
2593 if ((*sp)->serviceId == serviceId && (*sp)->socket == socket)
2599 #ifdef RXDEBUG_PACKET
2600 #ifdef KDUMP_RX_LOCK
2601 static struct rx_call_rx_lock *rx_allCallsp = 0;
2603 static struct rx_call *rx_allCallsp = 0;
2605 #endif /* RXDEBUG_PACKET */
2607 /* Allocate a call structure, for the indicated channel of the
2608 * supplied connection. The mode and state of the call must be set by
2609 * the caller. Returns the call with mutex locked. */
2610 static struct rx_call *
2611 rxi_NewCall(struct rx_connection *conn, int channel)
2613 struct rx_call *call;
2614 #ifdef RX_ENABLE_LOCKS
2615 struct rx_call *cp; /* Call pointer temp */
2616 struct opr_queue *cursor;
2619 dpf(("rxi_NewCall(conn %"AFS_PTR_FMT", channel %d)\n", conn, channel));
2621 /* Grab an existing call structure, or allocate a new one.
2622 * Existing call structures are assumed to have been left reset by
2624 MUTEX_ENTER(&rx_freeCallQueue_lock);
2626 #ifdef RX_ENABLE_LOCKS
2628 * EXCEPT that the TQ might not yet be cleared out.
2629 * Skip over those with in-use TQs.
2632 for (opr_queue_Scan(&rx_freeCallQueue, cursor)) {
2633 cp = opr_queue_Entry(cursor, struct rx_call, entry);
2634 if (!(cp->flags & RX_CALL_TQ_BUSY)) {
2640 #else /* RX_ENABLE_LOCKS */
2641 if (!opr_queue_IsEmpty(&rx_freeCallQueue)) {
2642 call = opr_queue_First(&rx_freeCallQueue, struct rx_call, entry);
2643 #endif /* RX_ENABLE_LOCKS */
2644 opr_queue_Remove(&call->entry);
2645 if (rx_stats_active)
2646 rx_atomic_dec(&rx_stats.nFreeCallStructs);
2647 MUTEX_EXIT(&rx_freeCallQueue_lock);
2648 MUTEX_ENTER(&call->lock);
2649 CLEAR_CALL_QUEUE_LOCK(call);
2650 #ifdef RX_ENABLE_LOCKS
2651 /* Now, if TQ wasn't cleared earlier, do it now. */
2652 rxi_WaitforTQBusy(call);
2653 if (call->flags & RX_CALL_TQ_CLEARME) {
2654 rxi_ClearTransmitQueue(call, 1);
2655 /*queue_Init(&call->tq);*/
2657 #endif /* RX_ENABLE_LOCKS */
2658 /* Bind the call to its connection structure */
2660 rxi_ResetCall(call, 1);
2663 call = rxi_Alloc(sizeof(struct rx_call));
2664 #ifdef RXDEBUG_PACKET
2665 call->allNextp = rx_allCallsp;
2666 rx_allCallsp = call;
2668 rx_atomic_inc_and_read(&rx_stats.nCallStructs);
2669 #else /* RXDEBUG_PACKET */
2670 rx_atomic_inc(&rx_stats.nCallStructs);
2671 #endif /* RXDEBUG_PACKET */
2673 MUTEX_EXIT(&rx_freeCallQueue_lock);
2674 MUTEX_INIT(&call->lock, "call lock", MUTEX_DEFAULT, NULL);
2675 MUTEX_ENTER(&call->lock);
2676 CV_INIT(&call->cv_twind, "call twind", CV_DEFAULT, 0);
2677 CV_INIT(&call->cv_rq, "call rq", CV_DEFAULT, 0);
2678 CV_INIT(&call->cv_tq, "call tq", CV_DEFAULT, 0);
2680 /* Initialize once-only items */
2681 opr_queue_Init(&call->tq);
2682 opr_queue_Init(&call->rq);
2683 opr_queue_Init(&call->app.iovq);
2684 #ifdef RXDEBUG_PACKET
2685 call->rqc = call->tqc = call->iovqc = 0;
2686 #endif /* RXDEBUG_PACKET */
2687 /* Bind the call to its connection structure (prereq for reset) */
2689 rxi_ResetCall(call, 1);
2691 call->channel = channel;
2692 call->callNumber = &conn->callNumber[channel];
2693 call->rwind = conn->rwind[channel];
2694 call->twind = conn->twind[channel];
2695 /* Note that the next expected call number is retained (in
2696 * conn->callNumber[i]), even if we reallocate the call structure
2698 conn->call[channel] = call;
2699 /* if the channel's never been used (== 0), we should start at 1, otherwise
2700 * the call number is valid from the last time this channel was used */
2701 if (*call->callNumber == 0)
2702 *call->callNumber = 1;
2707 /* A call has been inactive long enough that so we can throw away
2708 * state, including the call structure, which is placed on the call
2711 * call->lock amd rx_refcnt_mutex are held upon entry.
2712 * haveCTLock is set when called from rxi_ReapConnections.
2714 * return 1 if the call is freed, 0 if not.
2717 rxi_FreeCall(struct rx_call *call, int haveCTLock)
2719 int channel = call->channel;
2720 struct rx_connection *conn = call->conn;
2721 u_char state = call->state;
2724 * We are setting the state to RX_STATE_RESET to
2725 * ensure that no one else will attempt to use this
2726 * call once we drop the refcnt lock. We must drop
2727 * the refcnt lock before calling rxi_ResetCall
2728 * because it cannot be held across acquiring the
2729 * freepktQ lock. NewCall does the same.
2731 call->state = RX_STATE_RESET;
2732 MUTEX_EXIT(&rx_refcnt_mutex);
2733 rxi_ResetCall(call, 0);
2735 if (MUTEX_TRYENTER(&conn->conn_call_lock))
2737 if (state == RX_STATE_DALLY || state == RX_STATE_HOLD)
2738 (*call->callNumber)++;
2740 if (call->conn->call[channel] == call)
2741 call->conn->call[channel] = 0;
2742 MUTEX_EXIT(&conn->conn_call_lock);
2745 * We couldn't obtain the conn_call_lock so we can't
2746 * disconnect the call from the connection. Set the
2747 * call state to dally so that the call can be reused.
2749 MUTEX_ENTER(&rx_refcnt_mutex);
2750 call->state = RX_STATE_DALLY;
2754 MUTEX_ENTER(&rx_freeCallQueue_lock);
2755 SET_CALL_QUEUE_LOCK(call, &rx_freeCallQueue_lock);
2756 #ifdef RX_ENABLE_LOCKS
2757 /* A call may be free even though its transmit queue is still in use.
2758 * Since we search the call list from head to tail, put busy calls at
2759 * the head of the list, and idle calls at the tail.
2761 if (call->flags & RX_CALL_TQ_BUSY)
2762 opr_queue_Prepend(&rx_freeCallQueue, &call->entry);
2764 opr_queue_Append(&rx_freeCallQueue, &call->entry);
2765 #else /* RX_ENABLE_LOCKS */
2766 opr_queue_Append(&rx_freeCallQueue, &call->entry);
2767 #endif /* RX_ENABLE_LOCKS */
2768 if (rx_stats_active)
2769 rx_atomic_inc(&rx_stats.nFreeCallStructs);
2770 MUTEX_EXIT(&rx_freeCallQueue_lock);
2772 /* Destroy the connection if it was previously slated for
2773 * destruction, i.e. the Rx client code previously called
2774 * rx_DestroyConnection (client connections), or
2775 * rxi_ReapConnections called the same routine (server
2776 * connections). Only do this, however, if there are no
2777 * outstanding calls. Note that for fine grain locking, there appears
2778 * to be a deadlock in that rxi_FreeCall has a call locked and
2779 * DestroyConnectionNoLock locks each call in the conn. But note a
2780 * few lines up where we have removed this call from the conn.
2781 * If someone else destroys a connection, they either have no
2782 * call lock held or are going through this section of code.
2784 MUTEX_ENTER(&conn->conn_data_lock);
2785 if (conn->flags & RX_CONN_DESTROY_ME && !(conn->flags & RX_CONN_MAKECALL_WAITING)) {
2786 MUTEX_ENTER(&rx_refcnt_mutex);
2788 MUTEX_EXIT(&rx_refcnt_mutex);
2789 MUTEX_EXIT(&conn->conn_data_lock);
2790 #ifdef RX_ENABLE_LOCKS
2792 rxi_DestroyConnectionNoLock(conn);
2794 rxi_DestroyConnection(conn);
2795 #else /* RX_ENABLE_LOCKS */
2796 rxi_DestroyConnection(conn);
2797 #endif /* RX_ENABLE_LOCKS */
2799 MUTEX_EXIT(&conn->conn_data_lock);
2801 MUTEX_ENTER(&rx_refcnt_mutex);
2805 rx_atomic_t rxi_Allocsize = RX_ATOMIC_INIT(0);
2806 rx_atomic_t rxi_Alloccnt = RX_ATOMIC_INIT(0);
2809 rxi_Alloc(size_t size)
2813 if (rx_stats_active) {
2814 rx_atomic_add(&rxi_Allocsize, (int) size);
2815 rx_atomic_inc(&rxi_Alloccnt);
2819 #if defined(KERNEL) && !defined(UKERNEL) && defined(AFS_FBSD80_ENV)
2820 afs_osi_Alloc_NoSleep(size);
2825 osi_Panic("rxi_Alloc error");
2831 rxi_Free(void *addr, size_t size)
2833 if (rx_stats_active) {
2834 rx_atomic_sub(&rxi_Allocsize, (int) size);
2835 rx_atomic_dec(&rxi_Alloccnt);
2837 osi_Free(addr, size);
2841 rxi_SetPeerMtu(struct rx_peer *peer, afs_uint32 host, afs_uint32 port, int mtu)
2843 struct rx_peer **peer_ptr = NULL, **peer_end = NULL;
2844 struct rx_peer *next = NULL;
2848 MUTEX_ENTER(&rx_peerHashTable_lock);
2850 peer_ptr = &rx_peerHashTable[0];
2851 peer_end = &rx_peerHashTable[rx_hashTableSize];
2854 for ( ; peer_ptr < peer_end; peer_ptr++) {
2857 for ( ; peer; peer = next) {
2859 if (host == peer->host)
2864 hashIndex = PEER_HASH(host, port);
2865 for (peer = rx_peerHashTable[hashIndex]; peer; peer = peer->next) {
2866 if ((peer->host == host) && (peer->port == port))
2871 MUTEX_ENTER(&rx_peerHashTable_lock);
2876 MUTEX_EXIT(&rx_peerHashTable_lock);
2878 MUTEX_ENTER(&peer->peer_lock);
2879 /* We don't handle dropping below min, so don't */
2880 mtu = MAX(mtu, RX_MIN_PACKET_SIZE);
2881 peer->ifMTU=MIN(mtu, peer->ifMTU);
2882 peer->natMTU = rxi_AdjustIfMTU(peer->ifMTU);
2883 /* if we tweaked this down, need to tune our peer MTU too */
2884 peer->MTU = MIN(peer->MTU, peer->natMTU);
2885 /* if we discovered a sub-1500 mtu, degrade */
2886 if (peer->ifMTU < OLD_MAX_PACKET_SIZE)
2887 peer->maxDgramPackets = 1;
2888 /* We no longer have valid peer packet information */
2889 if (peer->maxPacketSize + RX_HEADER_SIZE > peer->ifMTU)
2890 peer->maxPacketSize = 0;
2891 MUTEX_EXIT(&peer->peer_lock);
2893 MUTEX_ENTER(&rx_peerHashTable_lock);
2895 if (host && !port) {
2897 /* pick up where we left off */
2901 MUTEX_EXIT(&rx_peerHashTable_lock);
2904 #ifdef AFS_RXERRQ_ENV
2906 rxi_SetPeerDead(struct sock_extended_err *err, afs_uint32 host, afs_uint16 port)
2908 int hashIndex = PEER_HASH(host, port);
2909 struct rx_peer *peer;
2911 MUTEX_ENTER(&rx_peerHashTable_lock);
2913 for (peer = rx_peerHashTable[hashIndex]; peer; peer = peer->next) {
2914 if (peer->host == host && peer->port == port) {
2920 MUTEX_EXIT(&rx_peerHashTable_lock);
2923 rx_atomic_inc(&peer->neterrs);
2924 MUTEX_ENTER(&peer->peer_lock);
2925 peer->last_err_origin = RX_NETWORK_ERROR_ORIGIN_ICMP;
2926 peer->last_err_type = err->ee_type;
2927 peer->last_err_code = err->ee_code;
2928 MUTEX_EXIT(&peer->peer_lock);
2930 MUTEX_ENTER(&rx_peerHashTable_lock);
2932 MUTEX_EXIT(&rx_peerHashTable_lock);
2937 rxi_ProcessNetError(struct sock_extended_err *err, afs_uint32 addr, afs_uint16 port)
2939 # ifdef AFS_ADAPT_PMTU
2940 if (err->ee_errno == EMSGSIZE && err->ee_info >= 68) {
2941 rxi_SetPeerMtu(NULL, addr, port, err->ee_info - RX_IPUDP_SIZE);
2945 if (err->ee_origin == SO_EE_ORIGIN_ICMP && err->ee_type == ICMP_DEST_UNREACH) {
2946 switch (err->ee_code) {
2947 case ICMP_NET_UNREACH:
2948 case ICMP_HOST_UNREACH:
2949 case ICMP_PORT_UNREACH:
2952 rxi_SetPeerDead(err, addr, port);
2959 rxi_TranslateICMP(int type, int code)
2962 case ICMP_DEST_UNREACH:
2964 case ICMP_NET_UNREACH:
2965 return "Destination Net Unreachable";
2966 case ICMP_HOST_UNREACH:
2967 return "Destination Host Unreachable";
2968 case ICMP_PROT_UNREACH:
2969 return "Destination Protocol Unreachable";
2970 case ICMP_PORT_UNREACH:
2971 return "Destination Port Unreachable";
2973 return "Destination Net Prohibited";
2975 return "Destination Host Prohibited";
2981 #endif /* AFS_RXERRQ_ENV */
2984 * Get the last network error for a connection
2986 * A "network error" here means an error retrieved from ICMP, or some other
2987 * mechanism outside of Rx that informs us of errors in network reachability.
2989 * If a peer associated with the given Rx connection has received a network
2990 * error recently, this function allows the caller to know what error
2991 * specifically occurred. This can be useful to know, since e.g. ICMP errors
2992 * can cause calls to that peer to be quickly aborted. So, this function can
2993 * help see why a call was aborted due to network errors.
2995 * If we have received traffic from a peer since the last network error, we
2996 * treat that peer as if we had not received an network error for it.
2998 * @param[in] conn The Rx connection to examine
2999 * @param[out] err_origin The origin of the last network error (e.g. ICMP);
3000 * one of the RX_NETWORK_ERROR_ORIGIN_* constants
3001 * @param[out] err_type The type of the last error
3002 * @param[out] err_code The code of the last error
3003 * @param[out] msg Human-readable error message, if applicable; NULL otherwise
3005 * @return If we have an error
3006 * @retval -1 No error to get; 'out' params are undefined
3007 * @retval 0 We have an error; 'out' params contain the last error
3010 rx_GetNetworkError(struct rx_connection *conn, int *err_origin, int *err_type,
3011 int *err_code, const char **msg)
3013 #ifdef AFS_RXERRQ_ENV
3014 struct rx_peer *peer = conn->peer;
3015 if (rx_atomic_read(&peer->neterrs)) {
3016 MUTEX_ENTER(&peer->peer_lock);
3017 *err_origin = peer->last_err_origin;
3018 *err_type = peer->last_err_type;
3019 *err_code = peer->last_err_code;
3020 MUTEX_EXIT(&peer->peer_lock);
3023 if (*err_origin == RX_NETWORK_ERROR_ORIGIN_ICMP) {
3024 *msg = rxi_TranslateICMP(*err_type, *err_code);
3033 /* Find the peer process represented by the supplied (host,port)
3034 * combination. If there is no appropriate active peer structure, a
3035 * new one will be allocated and initialized
3038 rxi_FindPeer(afs_uint32 host, u_short port, int create)
3042 hashIndex = PEER_HASH(host, port);
3043 MUTEX_ENTER(&rx_peerHashTable_lock);
3044 for (pp = rx_peerHashTable[hashIndex]; pp; pp = pp->next) {
3045 if ((pp->host == host) && (pp->port == port))
3050 pp = rxi_AllocPeer(); /* This bzero's *pp */
3051 pp->host = host; /* set here or in InitPeerParams is zero */
3053 #ifdef AFS_RXERRQ_ENV
3054 rx_atomic_set(&pp->neterrs, 0);
3056 MUTEX_INIT(&pp->peer_lock, "peer_lock", MUTEX_DEFAULT, 0);
3057 opr_queue_Init(&pp->rpcStats);
3058 pp->next = rx_peerHashTable[hashIndex];
3059 rx_peerHashTable[hashIndex] = pp;
3060 rxi_InitPeerParams(pp);
3061 if (rx_stats_active)
3062 rx_atomic_inc(&rx_stats.nPeerStructs);
3068 MUTEX_EXIT(&rx_peerHashTable_lock);
3073 /* Find the connection at (host, port) started at epoch, and with the
3074 * given connection id. Creates the server connection if necessary.
3075 * The type specifies whether a client connection or a server
3076 * connection is desired. In both cases, (host, port) specify the
3077 * peer's (host, pair) pair. Client connections are not made
3078 * automatically by this routine. The parameter socket gives the
3079 * socket descriptor on which the packet was received. This is used,
3080 * in the case of server connections, to check that *new* connections
3081 * come via a valid (port, serviceId). Finally, the securityIndex
3082 * parameter must match the existing index for the connection. If a
3083 * server connection is created, it will be created using the supplied
3084 * index, if the index is valid for this service */
3085 static struct rx_connection *
3086 rxi_FindConnection(osi_socket socket, afs_uint32 host,
3087 u_short port, u_short serviceId, afs_uint32 cid,
3088 afs_uint32 epoch, int type, u_int securityIndex,
3089 int *unknownService)
3091 int hashindex, flag, i;
3092 struct rx_connection *conn;
3093 *unknownService = 0;
3094 hashindex = CONN_HASH(host, port, cid, epoch, type);
3095 MUTEX_ENTER(&rx_connHashTable_lock);
3096 rxLastConn ? (conn = rxLastConn, flag = 0) : (conn =
3097 rx_connHashTable[hashindex],
3100 if ((conn->type == type) && ((cid & RX_CIDMASK) == conn->cid)
3101 && (epoch == conn->epoch)) {
3102 struct rx_peer *pp = conn->peer;
3103 if (securityIndex != conn->securityIndex) {
3104 /* this isn't supposed to happen, but someone could forge a packet
3105 * like this, and there seems to be some CM bug that makes this
3106 * happen from time to time -- in which case, the fileserver
3108 MUTEX_EXIT(&rx_connHashTable_lock);
3109 return (struct rx_connection *)0;
3111 if (pp->host == host && pp->port == port)
3113 if (type == RX_CLIENT_CONNECTION && pp->port == port)
3115 /* So what happens when it's a callback connection? */
3116 if ( /*type == RX_CLIENT_CONNECTION && */
3117 (conn->epoch & 0x80000000))
3121 /* the connection rxLastConn that was used the last time is not the
3122 ** one we are looking for now. Hence, start searching in the hash */
3124 conn = rx_connHashTable[hashindex];
3129 struct rx_service *service;
3130 if (type == RX_CLIENT_CONNECTION) {
3131 MUTEX_EXIT(&rx_connHashTable_lock);
3132 return (struct rx_connection *)0;
3134 service = rxi_FindService(socket, serviceId);
3135 if (!service || (securityIndex >= service->nSecurityObjects)
3136 || (service->securityObjects[securityIndex] == 0)) {
3137 MUTEX_EXIT(&rx_connHashTable_lock);
3138 *unknownService = 1;
3139 return (struct rx_connection *)0;
3141 conn = rxi_AllocConnection(); /* This bzero's the connection */
3142 MUTEX_INIT(&conn->conn_call_lock, "conn call lock", MUTEX_DEFAULT, 0);
3143 MUTEX_INIT(&conn->conn_data_lock, "conn data lock", MUTEX_DEFAULT, 0);
3144 CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
3145 conn->next = rx_connHashTable[hashindex];
3146 rx_connHashTable[hashindex] = conn;
3147 conn->peer = rxi_FindPeer(host, port, 1);
3148 conn->type = RX_SERVER_CONNECTION;
3149 conn->lastSendTime = clock_Sec(); /* don't GC immediately */
3150 conn->epoch = epoch;
3151 conn->cid = cid & RX_CIDMASK;
3152 conn->ackRate = RX_FAST_ACK_RATE;
3153 conn->service = service;
3154 conn->serviceId = serviceId;
3155 conn->securityIndex = securityIndex;
3156 conn->securityObject = service->securityObjects[securityIndex];
3157 conn->nSpecific = 0;
3158 conn->specific = NULL;
3159 rx_SetConnDeadTime(conn, service->connDeadTime);
3160 rx_SetConnIdleDeadTime(conn, service->idleDeadTime);
3161 for (i = 0; i < RX_MAXCALLS; i++) {
3162 conn->twind[i] = rx_initSendWindow;
3163 conn->rwind[i] = rx_initReceiveWindow;
3165 /* Notify security object of the new connection */
3166 RXS_NewConnection(conn->securityObject, conn);
3167 /* XXXX Connection timeout? */
3168 if (service->newConnProc)
3169 (*service->newConnProc) (conn);
3170 if (rx_stats_active)
3171 rx_atomic_inc(&rx_stats.nServerConns);
3174 MUTEX_ENTER(&rx_refcnt_mutex);
3176 MUTEX_EXIT(&rx_refcnt_mutex);
3178 rxLastConn = conn; /* store this connection as the last conn used */
3179 MUTEX_EXIT(&rx_connHashTable_lock);
3184 * Timeout a call on a busy call channel if appropriate.
3186 * @param[in] call The busy call.
3188 * @pre 'call' is marked as busy (namely,
3189 * call->conn->lastBusy[call->channel] != 0)
3191 * @pre call->lock is held
3192 * @pre rxi_busyChannelError is nonzero
3194 * @note call->lock is dropped and reacquired
3197 rxi_CheckBusy(struct rx_call *call)
3199 struct rx_connection *conn = call->conn;
3200 int channel = call->channel;
3201 int freechannel = 0;
3204 MUTEX_EXIT(&call->lock);
3206 MUTEX_ENTER(&conn->conn_call_lock);
3208 /* Are there any other call slots on this conn that we should try? Look for
3209 * slots that are empty and are either non-busy, or were marked as busy
3210 * longer than conn->secondsUntilDead seconds before this call started. */
3212 for (i = 0; i < RX_MAXCALLS && !freechannel; i++) {
3214 /* only look at channels that aren't us */
3218 if (conn->lastBusy[i]) {
3219 /* if this channel looked busy too recently, don't look at it */
3220 if (conn->lastBusy[i] >= call->startTime.sec) {
3223 if (call->startTime.sec - conn->lastBusy[i] < conn->secondsUntilDead) {
3228 if (conn->call[i]) {
3229 struct rx_call *tcall = conn->call[i];
3230 MUTEX_ENTER(&tcall->lock);
3231 if (tcall->state == RX_STATE_DALLY) {
3234 MUTEX_EXIT(&tcall->lock);
3240 MUTEX_ENTER(&call->lock);
3242 /* Since the call->lock has been released it is possible that the call may
3243 * no longer be busy (the call channel cannot have been reallocated as we
3244 * haven't dropped the conn_call_lock) Therefore, we must confirm
3245 * that the call state has not changed when deciding whether or not to
3246 * force this application thread to retry by forcing a Timeout error. */
3248 if (freechannel && (call->flags & RX_CALL_PEER_BUSY)) {
3249 /* Since 'freechannel' is set, there exists another channel in this
3250 * rx_conn that the application thread might be able to use. We know
3251 * that we have the correct call since callNumber is unchanged, and we
3252 * know that the call is still busy. So, set the call error state to
3253 * rxi_busyChannelError so the application can retry the request,
3254 * presumably on a less-busy call channel. */
3256 rxi_CallError(call, RX_CALL_BUSY);
3258 MUTEX_EXIT(&conn->conn_call_lock);
3262 * Abort the call if the server is over the busy threshold. This
3263 * can be used without requiring a call structure be initialised,
3264 * or connected to a particular channel
3267 rxi_AbortIfServerBusy(osi_socket socket, struct rx_connection *conn,
3268 struct rx_packet *np)
3270 if ((rx_BusyThreshold > 0) &&
3271 (rx_atomic_read(&rx_nWaiting) > rx_BusyThreshold)) {
3272 rxi_SendRawAbort(socket, conn->peer->host, conn->peer->port,
3273 rx_BusyError, np, 0);
3274 if (rx_stats_active)
3275 rx_atomic_inc(&rx_stats.nBusies);
3282 static_inline struct rx_call *
3283 rxi_ReceiveClientCall(struct rx_packet *np, struct rx_connection *conn)
3286 struct rx_call *call;
3288 channel = np->header.cid & RX_CHANNELMASK;
3289 MUTEX_ENTER(&conn->conn_call_lock);
3290 call = conn->call[channel];
3291 if (!call || conn->callNumber[channel] != np->header.callNumber) {
3292 MUTEX_EXIT(&conn->conn_call_lock);
3293 if (rx_stats_active)
3294 rx_atomic_inc(&rx_stats.spuriousPacketsRead);
3298 MUTEX_ENTER(&call->lock);
3299 MUTEX_EXIT(&conn->conn_call_lock);
3301 if ((call->state == RX_STATE_DALLY)
3302 && np->header.type == RX_PACKET_TYPE_ACK) {
3303 if (rx_stats_active)
3304 rx_atomic_inc(&rx_stats.ignorePacketDally);
3305 MUTEX_EXIT(&call->lock);
3312 static_inline struct rx_call *
3313 rxi_ReceiveServerCall(osi_socket socket, struct rx_packet *np,
3314 struct rx_connection *conn)
3317 struct rx_call *call;
3319 channel = np->header.cid & RX_CHANNELMASK;
3320 MUTEX_ENTER(&conn->conn_call_lock);
3321 call = conn->call[channel];
3324 if (rxi_AbortIfServerBusy(socket, conn, np)) {
3325 MUTEX_EXIT(&conn->conn_call_lock);
3329 call = rxi_NewCall(conn, channel); /* returns locked call */
3330 *call->callNumber = np->header.callNumber;
3331 MUTEX_EXIT(&conn->conn_call_lock);
3333 call->state = RX_STATE_PRECALL;
3334 clock_GetTime(&call->queueTime);
3335 call->app.bytesSent = 0;
3336 call->app.bytesRcvd = 0;
3337 rxi_KeepAliveOn(call);
3342 if (np->header.callNumber == conn->callNumber[channel]) {
3343 MUTEX_ENTER(&call->lock);
3344 MUTEX_EXIT(&conn->conn_call_lock);
3348 if (np->header.callNumber < conn->callNumber[channel]) {
3349 MUTEX_EXIT(&conn->conn_call_lock);
3350 if (rx_stats_active)
3351 rx_atomic_inc(&rx_stats.spuriousPacketsRead);
3355 MUTEX_ENTER(&call->lock);
3356 MUTEX_EXIT(&conn->conn_call_lock);
3358 /* Wait until the transmit queue is idle before deciding
3359 * whether to reset the current call. Chances are that the
3360 * call will be in ether DALLY or HOLD state once the TQ_BUSY
3363 #ifdef RX_ENABLE_LOCKS
3364 if (call->state == RX_STATE_ACTIVE && !call->error) {
3365 rxi_WaitforTQBusy(call);
3366 /* If we entered error state while waiting,
3367 * must call rxi_CallError to permit rxi_ResetCall
3368 * to processed when the tqWaiter count hits zero.
3371 rxi_CallError(call, call->error);
3372 MUTEX_EXIT(&call->lock);
3376 #endif /* RX_ENABLE_LOCKS */
3377 /* If the new call cannot be taken right now send a busy and set
3378 * the error condition in this call, so that it terminates as
3379 * quickly as possible */
3380 if (call->state == RX_STATE_ACTIVE) {
3381 rxi_CallError(call, RX_CALL_DEAD);
3382 rxi_SendSpecial(call, conn, NULL, RX_PACKET_TYPE_BUSY,
3384 MUTEX_EXIT(&call->lock);
3388 if (rxi_AbortIfServerBusy(socket, conn, np)) {
3389 MUTEX_EXIT(&call->lock);
3393 rxi_ResetCall(call, 0);
3394 /* The conn_call_lock is not held but no one else should be
3395 * using this call channel while we are processing this incoming
3396 * packet. This assignment should be safe.
3398 *call->callNumber = np->header.callNumber;
3399 call->state = RX_STATE_PRECALL;
3400 clock_GetTime(&call->queueTime);
3401 call->app.bytesSent = 0;
3402 call->app.bytesRcvd = 0;
3403 rxi_KeepAliveOn(call);
3409 /* There are two packet tracing routines available for testing and monitoring
3410 * Rx. One is called just after every packet is received and the other is
3411 * called just before every packet is sent. Received packets, have had their
3412 * headers decoded, and packets to be sent have not yet had their headers
3413 * encoded. Both take two parameters: a pointer to the packet and a sockaddr
3414 * containing the network address. Both can be modified. The return value, if
3415 * non-zero, indicates that the packet should be dropped. */
3417 int (*rx_justReceived) (struct rx_packet *, struct sockaddr_in *) = 0;
3418 int (*rx_almostSent) (struct rx_packet *, struct sockaddr_in *) = 0;
3420 /* A packet has been received off the interface. Np is the packet, socket is
3421 * the socket number it was received from (useful in determining which service
3422 * this packet corresponds to), and (host, port) reflect the host,port of the
3423 * sender. This call returns the packet to the caller if it is finished with
3424 * it, rather than de-allocating it, just as a small performance hack */
3427 rxi_ReceivePacket(struct rx_packet *np, osi_socket socket,
3428 afs_uint32 host, u_short port, int *tnop,
3429 struct rx_call **newcallp)
3431 struct rx_call *call;
3432 struct rx_connection *conn;
3434 int unknownService = 0;
3438 struct rx_packet *tnp;
3441 /* We don't print out the packet until now because (1) the time may not be
3442 * accurate enough until now in the lwp implementation (rx_Listener only gets
3443 * the time after the packet is read) and (2) from a protocol point of view,
3444 * this is the first time the packet has been seen */
3445 packetType = (np->header.type > 0 && np->header.type < RX_N_PACKET_TYPES)
3446 ? rx_packetTypes[np->header.type - 1] : "*UNKNOWN*";
3447 dpf(("R %d %s: %x.%d.%d.%d.%d.%d.%d flags %d, packet %"AFS_PTR_FMT"\n",
3448 np->header.serial, packetType, ntohl(host), ntohs(port), np->header.serviceId,
3449 np->header.epoch, np->header.cid, np->header.callNumber,
3450 np->header.seq, np->header.flags, np));
3453 /* Account for connectionless packets */
3454 if (rx_stats_active &&
3455 ((np->header.type == RX_PACKET_TYPE_VERSION) ||
3456 (np->header.type == RX_PACKET_TYPE_DEBUG))) {
3457 struct rx_peer *peer;
3459 /* Try to look up the peer structure, but don't create one */
3460 peer = rxi_FindPeer(host, port, 0);
3462 /* Since this may not be associated with a connection, it may have
3463 * no refCount, meaning we could race with ReapConnections
3466 if (peer && (peer->refCount > 0)) {
3467 #ifdef AFS_RXERRQ_ENV
3468 if (rx_atomic_read(&peer->neterrs)) {
3469 rx_atomic_set(&peer->neterrs, 0);
3472 MUTEX_ENTER(&peer->peer_lock);
3473 peer->bytesReceived += np->length;
3474 MUTEX_EXIT(&peer->peer_lock);
3478 if (np->header.type == RX_PACKET_TYPE_VERSION) {
3479 return rxi_ReceiveVersionPacket(np, socket, host, port, 1);
3482 if (np->header.type == RX_PACKET_TYPE_DEBUG) {
3483 return rxi_ReceiveDebugPacket(np, socket, host, port, 1);
3486 /* If an input tracer function is defined, call it with the packet and
3487 * network address. Note this function may modify its arguments. */
3488 if (rx_justReceived) {
3489 struct sockaddr_in addr;
3491 addr.sin_family = AF_INET;
3492 addr.sin_port = port;
3493 addr.sin_addr.s_addr = host;
3494 memset(&addr.sin_zero, 0, sizeof(addr.sin_zero));
3495 #ifdef STRUCT_SOCKADDR_HAS_SA_LEN
3496 addr.sin_len = sizeof(addr);
3497 #endif /* AFS_OSF_ENV */
3498 drop = (*rx_justReceived) (np, &addr);
3499 /* drop packet if return value is non-zero */
3502 port = addr.sin_port; /* in case fcn changed addr */
3503 host = addr.sin_addr.s_addr;
3507 /* If packet was not sent by the client, then *we* must be the client */
3508 type = ((np->header.flags & RX_CLIENT_INITIATED) != RX_CLIENT_INITIATED)
3509 ? RX_CLIENT_CONNECTION : RX_SERVER_CONNECTION;
3511 /* Find the connection (or fabricate one, if we're the server & if
3512 * necessary) associated with this packet */
3514 rxi_FindConnection(socket, host, port, np->header.serviceId,
3515 np->header.cid, np->header.epoch, type,
3516 np->header.securityIndex, &unknownService);
3518 /* To avoid having 2 connections just abort at each other,
3519 don't abort an abort. */
3521 if (unknownService && (np->header.type != RX_PACKET_TYPE_ABORT))
3522 rxi_SendRawAbort(socket, host, port, RX_INVALID_OPERATION,
3527 #ifdef AFS_RXERRQ_ENV
3528 if (rx_atomic_read(&conn->peer->neterrs)) {
3529 rx_atomic_set(&conn->peer->neterrs, 0);
3533 /* If we're doing statistics, then account for the incoming packet */
3534 if (rx_stats_active) {
3535 MUTEX_ENTER(&conn->peer->peer_lock);
3536 conn->peer->bytesReceived += np->length;
3537 MUTEX_EXIT(&conn->peer->peer_lock);
3540 /* If the connection is in an error state, send an abort packet and ignore
3541 * the incoming packet */
3543 /* Don't respond to an abort packet--we don't want loops! */
3544 MUTEX_ENTER(&conn->conn_data_lock);
3545 if (np->header.type != RX_PACKET_TYPE_ABORT)
3546 np = rxi_SendConnectionAbort(conn, np, 1, 0);
3547 putConnection(conn);
3548 MUTEX_EXIT(&conn->conn_data_lock);
3552 /* Check for connection-only requests (i.e. not call specific). */
3553 if (np->header.callNumber == 0) {
3554 switch (np->header.type) {
3555 case RX_PACKET_TYPE_ABORT: {
3556 /* What if the supplied error is zero? */
3557 afs_int32 errcode = ntohl(rx_GetInt32(np, 0));
3558 dpf(("rxi_ReceivePacket ABORT rx_GetInt32 = %d\n", errcode));
3559 rxi_ConnectionError(conn, errcode);
3560 putConnection(conn);
3563 case RX_PACKET_TYPE_CHALLENGE:
3564 tnp = rxi_ReceiveChallengePacket(conn, np, 1);
3565 putConnection(conn);
3567 case RX_PACKET_TYPE_RESPONSE:
3568 tnp = rxi_ReceiveResponsePacket(conn, np, 1);
3569 putConnection(conn);
3571 case RX_PACKET_TYPE_PARAMS:
3572 case RX_PACKET_TYPE_PARAMS + 1:
3573 case RX_PACKET_TYPE_PARAMS + 2:
3574 /* ignore these packet types for now */
3575 putConnection(conn);
3579 /* Should not reach here, unless the peer is broken: send an
3581 rxi_ConnectionError(conn, RX_PROTOCOL_ERROR);
3582 MUTEX_ENTER(&conn->conn_data_lock);
3583 tnp = rxi_SendConnectionAbort(conn, np, 1, 0);
3584 putConnection(conn);
3585 MUTEX_EXIT(&conn->conn_data_lock);
3590 if (type == RX_SERVER_CONNECTION)
3591 call = rxi_ReceiveServerCall(socket, np, conn);
3593 call = rxi_ReceiveClientCall(np, conn);
3596 putConnection(conn);
3600 MUTEX_ASSERT(&call->lock);
3601 /* Set remote user defined status from packet */
3602 call->remoteStatus = np->header.userStatus;
3604 /* Now do packet type-specific processing */
3605 switch (np->header.type) {
3606 case RX_PACKET_TYPE_DATA:
3607 /* If we're a client, and receiving a response, then all the packets
3608 * we transmitted packets are implicitly acknowledged. */
3609 if (type == RX_CLIENT_CONNECTION && !opr_queue_IsEmpty(&call->tq))
3610 rxi_AckAllInTransmitQueue(call);
3612 np = rxi_ReceiveDataPacket(call, np, 1, socket, host, port, tnop,
3615 case RX_PACKET_TYPE_ACK:
3616 /* Respond immediately to ack packets requesting acknowledgement
3618 if (np->header.flags & RX_REQUEST_ACK) {
3620 (void)rxi_SendCallAbort(call, 0, 1, 0);
3622 (void)rxi_SendAck(call, 0, np->header.serial,
3623 RX_ACK_PING_RESPONSE, 1);
3625 np = rxi_ReceiveAckPacket(call, np, 1);
3627 case RX_PACKET_TYPE_ABORT: {
3628 /* An abort packet: reset the call, passing the error up to the user. */
3629 /* What if error is zero? */
3630 /* What if the error is -1? the application will treat it as a timeout. */
3631 afs_int32 errdata = ntohl(*(afs_int32 *) rx_DataOf(np));
3632 dpf(("rxi_ReceivePacket ABORT rx_DataOf = %d\n", errdata));
3633 rxi_CallError(call, errdata);
3634 MUTEX_EXIT(&call->lock);
3635 putConnection(conn);
3636 return np; /* xmitting; drop packet */
3638 case RX_PACKET_TYPE_BUSY: {
3639 struct clock busyTime;
3641 clock_GetTime(&busyTime);
3643 MUTEX_EXIT(&call->lock);
3645 MUTEX_ENTER(&conn->conn_call_lock);
3646 MUTEX_ENTER(&call->lock);
3647 conn->lastBusy[call->channel] = busyTime.sec;
3648 call->flags |= RX_CALL_PEER_BUSY;
3649 MUTEX_EXIT(&call->lock);
3650 MUTEX_EXIT(&conn->conn_call_lock);
3652 putConnection(conn);
3656 case RX_PACKET_TYPE_ACKALL:
3657 /* All packets acknowledged, so we can drop all packets previously
3658 * readied for sending */
3659 rxi_AckAllInTransmitQueue(call);
3662 /* Should not reach here, unless the peer is broken: send an abort
3664 rxi_CallError(call, RX_PROTOCOL_ERROR);
3665 np = rxi_SendCallAbort(call, np, 1, 0);
3668 /* Note when this last legitimate packet was received, for keep-alive
3669 * processing. Note, we delay getting the time until now in the hope that
3670 * the packet will be delivered to the user before any get time is required
3671 * (if not, then the time won't actually be re-evaluated here). */
3672 call->lastReceiveTime = clock_Sec();
3673 /* we've received a legit packet, so the channel is not busy */
3674 call->flags &= ~RX_CALL_PEER_BUSY;
3675 MUTEX_EXIT(&call->lock);
3676 putConnection(conn);
3680 /* return true if this is an "interesting" connection from the point of view
3681 of someone trying to debug the system */
3683 rxi_IsConnInteresting(struct rx_connection *aconn)
3686 struct rx_call *tcall;
3688 if (aconn->flags & (RX_CONN_MAKECALL_WAITING | RX_CONN_DESTROY_ME))
3691 for (i = 0; i < RX_MAXCALLS; i++) {
3692 tcall = aconn->call[i];
3694 if ((tcall->state == RX_STATE_PRECALL)
3695 || (tcall->state == RX_STATE_ACTIVE))
3697 if ((tcall->app.mode == RX_MODE_SENDING)
3698 || (tcall->app.mode == RX_MODE_RECEIVING))
3706 /* if this is one of the last few packets AND it wouldn't be used by the
3707 receiving call to immediately satisfy a read request, then drop it on
3708 the floor, since accepting it might prevent a lock-holding thread from
3709 making progress in its reading. If a call has been cleared while in
3710 the precall state then ignore all subsequent packets until the call
3711 is assigned to a thread. */
3714 TooLow(struct rx_packet *ap, struct rx_call *acall)
3718 MUTEX_ENTER(&rx_quota_mutex);
3719 if (((ap->header.seq != 1) && (acall->flags & RX_CALL_CLEARED)
3720 && (acall->state == RX_STATE_PRECALL))
3721 || ((rx_nFreePackets < rxi_dataQuota + 2)
3722 && !((ap->header.seq < acall->rnext + rx_initSendWindow)
3723 && (acall->flags & RX_CALL_READER_WAIT)))) {
3726 MUTEX_EXIT(&rx_quota_mutex);
3732 * Clear the attach wait flag on a connection and proceed.
3734 * Any processing waiting for a connection to be attached should be
3735 * unblocked. We clear the flag and do any other needed tasks.
3738 * the conn to unmark waiting for attach
3740 * @pre conn's conn_data_lock must be locked before calling this function
3744 rxi_ConnClearAttachWait(struct rx_connection *conn)
3746 /* Indicate that rxi_CheckReachEvent is no longer running by
3747 * clearing the flag. Must be atomic under conn_data_lock to
3748 * avoid a new call slipping by: rxi_CheckConnReach holds
3749 * conn_data_lock while checking RX_CONN_ATTACHWAIT.
3751 conn->flags &= ~RX_CONN_ATTACHWAIT;
3752 if (conn->flags & RX_CONN_NAT_PING) {
3753 conn->flags &= ~RX_CONN_NAT_PING;
3754 rxi_ScheduleNatKeepAliveEvent(conn);
3759 rxi_CheckReachEvent(struct rxevent *event, void *arg1, void *arg2, int dummy)
3761 struct rx_connection *conn = arg1;
3762 struct rx_call *acall = arg2;
3763 struct rx_call *call = acall;
3764 struct clock when, now;
3767 MUTEX_ENTER(&conn->conn_data_lock);
3770 rxevent_Put(&conn->checkReachEvent);
3772 waiting = conn->flags & RX_CONN_ATTACHWAIT;
3774 putConnection(conn);
3776 MUTEX_EXIT(&conn->conn_data_lock);
3780 MUTEX_ENTER(&conn->conn_call_lock);
3781 MUTEX_ENTER(&conn->conn_data_lock);
3782 for (i = 0; i < RX_MAXCALLS; i++) {