63ed36604daa3edbd3a737722509507acd3bed47
[openafs.git] / src / rx / rx.c
1 /*
2  * Copyright 2000, International Business Machines Corporation and others.
3  * All Rights Reserved.
4  * 
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
8  */
9
10 /* RX:  Extended Remote Procedure Call */
11
12 #include <afsconfig.h>
13 #ifdef  KERNEL
14 #include "afs/param.h"
15 #else
16 #include <afs/param.h>
17 #endif
18
19
20 #ifdef KERNEL
21 #include "afs/sysincludes.h"
22 #include "afsincludes.h"
23 #ifndef UKERNEL
24 #include "h/types.h"
25 #include "h/time.h"
26 #include "h/stat.h"
27 #ifdef  AFS_OSF_ENV
28 #include <net/net_globals.h>
29 #endif /* AFS_OSF_ENV */
30 #ifdef AFS_LINUX20_ENV
31 #include "h/socket.h"
32 #endif
33 #include "netinet/in.h"
34 #ifdef AFS_SUN57_ENV
35 #include "inet/common.h"
36 #include "inet/ip.h"
37 #include "inet/ip_ire.h"
38 #endif
39 #include "afs/afs_args.h"
40 #include "afs/afs_osi.h"
41 #ifdef RX_KERNEL_TRACE
42 #include "rx_kcommon.h"
43 #endif
44 #if     (defined(AFS_AUX_ENV) || defined(AFS_AIX_ENV))
45 #include "h/systm.h"
46 #endif
47 #ifdef RXDEBUG
48 #undef RXDEBUG                  /* turn off debugging */
49 #endif /* RXDEBUG */
50 #if defined(AFS_SGI_ENV)
51 #include "sys/debug.h"
52 #endif
53 #include "afsint.h"
54 #ifdef  AFS_OSF_ENV
55 #undef kmem_alloc
56 #undef kmem_free
57 #undef mem_alloc
58 #undef mem_free
59 #undef register
60 #endif /* AFS_OSF_ENV */
61 #else /* !UKERNEL */
62 #include "afs/sysincludes.h"
63 #include "afsincludes.h"
64 #endif /* !UKERNEL */
65 #include "afs/lock.h"
66 #include "rx_kmutex.h"
67 #include "rx_kernel.h"
68 #include "rx_clock.h"
69 #include "rx_queue.h"
70 #include "rx.h"
71 #include "rx_globals.h"
72 #include "rx_trace.h"
73 #define AFSOP_STOP_RXCALLBACK   210     /* Stop CALLBACK process */
74 #define AFSOP_STOP_AFS          211     /* Stop AFS process */
75 #define AFSOP_STOP_BKG          212     /* Stop BKG process */
76 #include "afsint.h"
77 extern afs_int32 afs_termState;
78 #ifdef AFS_AIX41_ENV
79 #include "sys/lockl.h"
80 #include "sys/lock_def.h"
81 #endif /* AFS_AIX41_ENV */
82 # include "rxgen_consts.h"
83 #else /* KERNEL */
84 # include <sys/types.h>
85 # include <string.h>
86 # include <stdarg.h>
87 # include <errno.h>
88 #ifdef AFS_NT40_ENV
89 # include <stdlib.h>
90 # include <fcntl.h>
91 # include <afs/afsutil.h>
92 # include <WINNT\afsreg.h>
93 #else
94 # include <sys/socket.h>
95 # include <sys/file.h>
96 # include <netdb.h>
97 # include <sys/stat.h>
98 # include <netinet/in.h>
99 # include <sys/time.h>
100 #endif
101 # include "rx.h"
102 # include "rx_user.h"
103 # include "rx_clock.h"
104 # include "rx_queue.h"
105 # include "rx_globals.h"
106 # include "rx_trace.h"
107 # include <afs/rxgen_consts.h>
108 #endif /* KERNEL */
109
110 #ifndef KERNEL
111 #ifdef AFS_PTHREAD_ENV
112 #ifndef AFS_NT40_ENV
113 int (*registerProgram) (pid_t, char *) = 0;
114 int (*swapNameProgram) (pid_t, const char *, char *) = 0;
115 #endif
116 #else
117 int (*registerProgram) (PROCESS, char *) = 0;
118 int (*swapNameProgram) (PROCESS, const char *, char *) = 0;
119 #endif
120 #endif
121
122 /* Local static routines */
123 static void rxi_DestroyConnectionNoLock(struct rx_connection *conn);
124 #ifdef RX_ENABLE_LOCKS
125 static void rxi_SetAcksInTransmitQueue(struct rx_call *call);
126 #endif
127
128 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
129 struct rx_tq_debug {
130     afs_int32 rxi_start_aborted;        /* rxi_start awoke after rxi_Send in error. */
131     afs_int32 rxi_start_in_error;
132 } rx_tq_debug;
133 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
134
135 /*
136  * rxi_rpc_peer_stat_cnt counts the total number of peer stat structures
137  * currently allocated within rx.  This number is used to allocate the
138  * memory required to return the statistics when queried.
139  */
140
141 static unsigned int rxi_rpc_peer_stat_cnt;
142
143 /*
144  * rxi_rpc_process_stat_cnt counts the total number of local process stat
145  * structures currently allocated within rx.  The number is used to allocate
146  * the memory required to return the statistics when queried.
147  */
148
149 static unsigned int rxi_rpc_process_stat_cnt;
150
151 #if !defined(offsetof)
152 #include <stddef.h>             /* for definition of offsetof() */
153 #endif
154
155 #ifdef AFS_PTHREAD_ENV
156 #include <assert.h>
157
158 /*
159  * Use procedural initialization of mutexes/condition variables
160  * to ease NT porting
161  */
162
163 extern afs_kmutex_t rx_stats_mutex;
164 extern afs_kmutex_t rx_waiting_mutex;
165 extern afs_kmutex_t rx_quota_mutex;
166 extern afs_kmutex_t rx_pthread_mutex;
167 extern afs_kmutex_t rx_packets_mutex;
168 extern afs_kmutex_t des_init_mutex;
169 extern afs_kmutex_t des_random_mutex;
170 extern afs_kmutex_t rx_clock_mutex;
171 extern afs_kmutex_t rxi_connCacheMutex;
172 extern afs_kmutex_t rx_event_mutex;
173 extern afs_kmutex_t osi_malloc_mutex;
174 extern afs_kmutex_t event_handler_mutex;
175 extern afs_kmutex_t listener_mutex;
176 extern afs_kmutex_t rx_if_init_mutex;
177 extern afs_kmutex_t rx_if_mutex;
178 extern afs_kmutex_t rxkad_client_uid_mutex;
179 extern afs_kmutex_t rxkad_random_mutex;
180
181 extern afs_kcondvar_t rx_event_handler_cond;
182 extern afs_kcondvar_t rx_listener_cond;
183
184 static afs_kmutex_t epoch_mutex;
185 static afs_kmutex_t rx_init_mutex;
186 static afs_kmutex_t rx_debug_mutex;
187 static afs_kmutex_t rx_rpc_stats;
188
189 static void
190 rxi_InitPthread(void)
191 {
192     MUTEX_INIT(&rx_clock_mutex, "clock", MUTEX_DEFAULT, 0);
193     MUTEX_INIT(&rx_stats_mutex, "stats", MUTEX_DEFAULT, 0);
194     MUTEX_INIT(&rx_waiting_mutex, "waiting", MUTEX_DEFAULT, 0);
195     MUTEX_INIT(&rx_quota_mutex, "quota", MUTEX_DEFAULT, 0);
196     MUTEX_INIT(&rx_pthread_mutex, "pthread", MUTEX_DEFAULT, 0);
197     MUTEX_INIT(&rx_packets_mutex, "packets", MUTEX_DEFAULT, 0);
198     MUTEX_INIT(&epoch_mutex, "epoch", MUTEX_DEFAULT, 0);
199     MUTEX_INIT(&rx_init_mutex, "init", MUTEX_DEFAULT, 0);
200     MUTEX_INIT(&rx_event_mutex, "event", MUTEX_DEFAULT, 0);
201     MUTEX_INIT(&des_init_mutex, "des", MUTEX_DEFAULT, 0);
202     MUTEX_INIT(&des_random_mutex, "random", MUTEX_DEFAULT, 0);
203     MUTEX_INIT(&osi_malloc_mutex, "malloc", MUTEX_DEFAULT, 0);
204     MUTEX_INIT(&event_handler_mutex, "event handler", MUTEX_DEFAULT, 0);
205     MUTEX_INIT(&rxi_connCacheMutex, "conn cache", MUTEX_DEFAULT, 0);
206     MUTEX_INIT(&listener_mutex, "listener", MUTEX_DEFAULT, 0);
207     MUTEX_INIT(&rx_if_init_mutex, "if init", MUTEX_DEFAULT, 0);
208     MUTEX_INIT(&rx_if_mutex, "if", MUTEX_DEFAULT, 0);
209     MUTEX_INIT(&rxkad_client_uid_mutex, "uid", MUTEX_DEFAULT, 0);
210     MUTEX_INIT(&rxkad_random_mutex, "rxkad random", MUTEX_DEFAULT, 0);
211     MUTEX_INIT(&rx_debug_mutex, "debug", MUTEX_DEFAULT, 0);
212
213     assert(pthread_cond_init
214            (&rx_event_handler_cond, (const pthread_condattr_t *)0) == 0);
215     assert(pthread_cond_init(&rx_listener_cond, (const pthread_condattr_t *)0)
216            == 0);
217     assert(pthread_key_create(&rx_thread_id_key, NULL) == 0);
218     assert(pthread_key_create(&rx_ts_info_key, NULL) == 0);
219  
220     rxkad_global_stats_init();
221
222     MUTEX_INIT(&rx_rpc_stats, "rx_rpc_stats", MUTEX_DEFAULT, 0);
223     MUTEX_INIT(&rx_freePktQ_lock, "rx_freePktQ_lock", MUTEX_DEFAULT, 0);
224 #ifdef  RX_ENABLE_LOCKS
225 #ifdef RX_LOCKS_DB
226     rxdb_init();
227 #endif /* RX_LOCKS_DB */
228     MUTEX_INIT(&freeSQEList_lock, "freeSQEList lock", MUTEX_DEFAULT, 0);
229     MUTEX_INIT(&rx_freeCallQueue_lock, "rx_freeCallQueue_lock", MUTEX_DEFAULT,
230                0);
231     CV_INIT(&rx_waitingForPackets_cv, "rx_waitingForPackets_cv", CV_DEFAULT,
232             0);
233     MUTEX_INIT(&rx_peerHashTable_lock, "rx_peerHashTable_lock", MUTEX_DEFAULT,
234                0);
235     MUTEX_INIT(&rx_connHashTable_lock, "rx_connHashTable_lock", MUTEX_DEFAULT,
236                0);
237     MUTEX_INIT(&rx_serverPool_lock, "rx_serverPool_lock", MUTEX_DEFAULT, 0);
238     MUTEX_INIT(&rxi_keyCreate_lock, "rxi_keyCreate_lock", MUTEX_DEFAULT, 0);
239 #endif /* RX_ENABLE_LOCKS */
240 }
241
242 pthread_once_t rx_once_init = PTHREAD_ONCE_INIT;
243 #define INIT_PTHREAD_LOCKS \
244 assert(pthread_once(&rx_once_init, rxi_InitPthread)==0)
245 /*
246  * The rx_stats_mutex mutex protects the following global variables:
247  * rxi_lowConnRefCount
248  * rxi_lowPeerRefCount
249  * rxi_nCalls
250  * rxi_Alloccnt
251  * rxi_Allocsize
252  * rx_tq_debug
253  * rx_stats
254  */
255
256 /*
257  * The rx_quota_mutex mutex protects the following global variables:
258  * rxi_dataQuota
259  * rxi_minDeficit
260  * rxi_availProcs
261  * rxi_totalMin
262  */
263
264 /* 
265  * The rx_freePktQ_lock protects the following global variables:
266  * rx_nFreePackets 
267  */
268
269 /*
270  * The rx_packets_mutex mutex protects the following global variables:
271  * rx_nPackets
272  * rx_TSFPQLocalMax
273  * rx_TSFPQGlobSize
274  * rx_TSFPQMaxProcs
275  */
276
277 /*
278  * The rx_pthread_mutex mutex protects the following global variables:
279  * rxi_pthread_hinum
280  */
281 #else
282 #define INIT_PTHREAD_LOCKS
283 #endif
284
285
286 /* Variables for handling the minProcs implementation.  availProcs gives the
287  * number of threads available in the pool at this moment (not counting dudes
288  * executing right now).  totalMin gives the total number of procs required
289  * for handling all minProcs requests.  minDeficit is a dynamic variable
290  * tracking the # of procs required to satisfy all of the remaining minProcs
291  * demands.
292  * For fine grain locking to work, the quota check and the reservation of
293  * a server thread has to come while rxi_availProcs and rxi_minDeficit
294  * are locked. To this end, the code has been modified under #ifdef
295  * RX_ENABLE_LOCKS so that quota checks and reservation occur at the
296  * same time. A new function, ReturnToServerPool() returns the allocation.
297  * 
298  * A call can be on several queue's (but only one at a time). When
299  * rxi_ResetCall wants to remove the call from a queue, it has to ensure
300  * that no one else is touching the queue. To this end, we store the address
301  * of the queue lock in the call structure (under the call lock) when we
302  * put the call on a queue, and we clear the call_queue_lock when the
303  * call is removed from a queue (once the call lock has been obtained).
304  * This allows rxi_ResetCall to safely synchronize with others wishing
305  * to manipulate the queue.
306  */
307
308 #if defined(RX_ENABLE_LOCKS) && defined(KERNEL)
309 static afs_kmutex_t rx_rpc_stats;
310 void rxi_StartUnlocked(struct rxevent *event, void *call,
311                        void *arg1, int istack);
312 #endif
313
314 /* We keep a "last conn pointer" in rxi_FindConnection. The odds are 
315 ** pretty good that the next packet coming in is from the same connection 
316 ** as the last packet, since we're send multiple packets in a transmit window.
317 */
318 struct rx_connection *rxLastConn = 0;
319
320 #ifdef RX_ENABLE_LOCKS
321 /* The locking hierarchy for rx fine grain locking is composed of these
322  * tiers:
323  *
324  * rx_connHashTable_lock - synchronizes conn creation, rx_connHashTable access
325  * conn_call_lock - used to synchonize rx_EndCall and rx_NewCall
326  * call->lock - locks call data fields.
327  * These are independent of each other:
328  *      rx_freeCallQueue_lock
329  *      rxi_keyCreate_lock
330  * rx_serverPool_lock
331  * freeSQEList_lock
332  *
333  * serverQueueEntry->lock
334  * rx_rpc_stats
335  * rx_peerHashTable_lock - locked under rx_connHashTable_lock
336  * peer->lock - locks peer data fields.
337  * conn_data_lock - that more than one thread is not updating a conn data
338  *                  field at the same time.
339  * rx_freePktQ_lock
340  *
341  * lowest level:
342  *      multi_handle->lock
343  *      rxevent_lock
344  *      rx_stats_mutex
345  *
346  * Do we need a lock to protect the peer field in the conn structure?
347  *      conn->peer was previously a constant for all intents and so has no
348  *      lock protecting this field. The multihomed client delta introduced
349  *      a RX code change : change the peer field in the connection structure
350  *      to that remote inetrface from which the last packet for this
351  *      connection was sent out. This may become an issue if further changes
352  *      are made.
353  */
354 #define SET_CALL_QUEUE_LOCK(C, L) (C)->call_queue_lock = (L)
355 #define CLEAR_CALL_QUEUE_LOCK(C) (C)->call_queue_lock = NULL
356 #ifdef RX_LOCKS_DB
357 /* rxdb_fileID is used to identify the lock location, along with line#. */
358 static int rxdb_fileID = RXDB_FILE_RX;
359 #endif /* RX_LOCKS_DB */
360 #else /* RX_ENABLE_LOCKS */
361 #define SET_CALL_QUEUE_LOCK(C, L)
362 #define CLEAR_CALL_QUEUE_LOCK(C)
363 #endif /* RX_ENABLE_LOCKS */
364 struct rx_serverQueueEntry *rx_waitForPacket = 0;
365 struct rx_serverQueueEntry *rx_waitingForPacket = 0;
366
367 /* ------------Exported Interfaces------------- */
368
369 /* This function allows rxkad to set the epoch to a suitably random number
370  * which rx_NewConnection will use in the future.  The principle purpose is to
371  * get rxnull connections to use the same epoch as the rxkad connections do, at
372  * least once the first rxkad connection is established.  This is important now
373  * that the host/port addresses aren't used in FindConnection: the uniqueness
374  * of epoch/cid matters and the start time won't do. */
375
376 #ifdef AFS_PTHREAD_ENV
377 /*
378  * This mutex protects the following global variables:
379  * rx_epoch
380  */
381
382 #define LOCK_EPOCH MUTEX_ENTER(&epoch_mutex)
383 #define UNLOCK_EPOCH MUTEX_EXIT(&epoch_mutex)
384 #else
385 #define LOCK_EPOCH
386 #define UNLOCK_EPOCH
387 #endif /* AFS_PTHREAD_ENV */
388
389 void
390 rx_SetEpoch(afs_uint32 epoch)
391 {
392     LOCK_EPOCH;
393     rx_epoch = epoch;
394     UNLOCK_EPOCH;
395 }
396
397 /* Initialize rx.  A port number may be mentioned, in which case this
398  * becomes the default port number for any service installed later.
399  * If 0 is provided for the port number, a random port will be chosen
400  * by the kernel.  Whether this will ever overlap anything in
401  * /etc/services is anybody's guess...  Returns 0 on success, -1 on
402  * error. */
403 #ifndef AFS_NT40_ENV
404 static
405 #endif
406 int rxinit_status = 1;
407 #ifdef AFS_PTHREAD_ENV
408 /*
409  * This mutex protects the following global variables:
410  * rxinit_status
411  */
412
413 #define LOCK_RX_INIT MUTEX_ENTER(&rx_init_mutex)
414 #define UNLOCK_RX_INIT MUTEX_EXIT(&rx_init_mutex)
415 #else
416 #define LOCK_RX_INIT
417 #define UNLOCK_RX_INIT
418 #endif
419
420 int
421 rx_InitHost(u_int host, u_int port)
422 {
423 #ifdef KERNEL
424     osi_timeval_t tv;
425 #else /* KERNEL */
426     struct timeval tv;
427 #endif /* KERNEL */
428     char *htable, *ptable;
429     int tmp_status;
430     
431     SPLVAR;
432     
433     INIT_PTHREAD_LOCKS;
434     LOCK_RX_INIT;
435     if (rxinit_status == 0) {
436         tmp_status = rxinit_status;
437         UNLOCK_RX_INIT;
438         return tmp_status;      /* Already started; return previous error code. */
439     }
440 #ifdef RXDEBUG
441     rxi_DebugInit();
442 #endif
443 #ifdef AFS_NT40_ENV
444     if (afs_winsockInit() < 0)
445         return -1;
446 #endif
447
448 #ifndef KERNEL
449     /*
450      * Initialize anything necessary to provide a non-premptive threading
451      * environment.
452      */
453     rxi_InitializeThreadSupport();
454 #endif
455
456     /* Allocate and initialize a socket for client and perhaps server
457      * connections. */
458
459     rx_socket = rxi_GetHostUDPSocket(host, (u_short) port);
460     if (rx_socket == OSI_NULLSOCKET) {
461         UNLOCK_RX_INIT;
462         return RX_ADDRINUSE;
463     }
464 #if defined(RX_ENABLE_LOCKS) && defined(KERNEL)
465 #ifdef RX_LOCKS_DB
466     rxdb_init();
467 #endif /* RX_LOCKS_DB */
468     MUTEX_INIT(&rx_stats_mutex, "rx_stats_mutex", MUTEX_DEFAULT, 0);
469     MUTEX_INIT(&rx_waiting_mutex, "rx_waiting_mutex", MUTEX_DEFAULT, 0);
470     MUTEX_INIT(&rx_quota_mutex, "rx_quota_mutex", MUTEX_DEFAULT, 0);
471     MUTEX_INIT(&rx_pthread_mutex, "rx_pthread_mutex", MUTEX_DEFAULT, 0);
472     MUTEX_INIT(&rx_packets_mutex, "rx_packets_mutex", MUTEX_DEFAULT, 0);
473     MUTEX_INIT(&rx_rpc_stats, "rx_rpc_stats", MUTEX_DEFAULT, 0);
474     MUTEX_INIT(&rx_freePktQ_lock, "rx_freePktQ_lock", MUTEX_DEFAULT, 0);
475     MUTEX_INIT(&freeSQEList_lock, "freeSQEList lock", MUTEX_DEFAULT, 0);
476     MUTEX_INIT(&rx_freeCallQueue_lock, "rx_freeCallQueue_lock", MUTEX_DEFAULT,
477                0);
478     CV_INIT(&rx_waitingForPackets_cv, "rx_waitingForPackets_cv", CV_DEFAULT,
479             0);
480     MUTEX_INIT(&rx_peerHashTable_lock, "rx_peerHashTable_lock", MUTEX_DEFAULT,
481                0);
482     MUTEX_INIT(&rx_connHashTable_lock, "rx_connHashTable_lock", MUTEX_DEFAULT,
483                0);
484     MUTEX_INIT(&rx_serverPool_lock, "rx_serverPool_lock", MUTEX_DEFAULT, 0);
485 #if defined(AFS_HPUX110_ENV)
486     if (!uniprocessor)
487         rx_sleepLock = alloc_spinlock(LAST_HELD_ORDER - 10, "rx_sleepLock");
488 #endif /* AFS_HPUX110_ENV */
489 #endif /* RX_ENABLE_LOCKS && KERNEL */
490
491     rxi_nCalls = 0;
492     rx_connDeadTime = 12;
493     rx_tranquil = 0;            /* reset flag */
494     memset((char *)&rx_stats, 0, sizeof(struct rx_statistics));
495     htable = (char *)
496         osi_Alloc(rx_hashTableSize * sizeof(struct rx_connection *));
497     PIN(htable, rx_hashTableSize * sizeof(struct rx_connection *));     /* XXXXX */
498     memset(htable, 0, rx_hashTableSize * sizeof(struct rx_connection *));
499     ptable = (char *)osi_Alloc(rx_hashTableSize * sizeof(struct rx_peer *));
500     PIN(ptable, rx_hashTableSize * sizeof(struct rx_peer *));   /* XXXXX */
501     memset(ptable, 0, rx_hashTableSize * sizeof(struct rx_peer *));
502
503     /* Malloc up a bunch of packets & buffers */
504     rx_nFreePackets = 0;
505     queue_Init(&rx_freePacketQueue);
506     rxi_NeedMorePackets = FALSE;
507 #ifdef RX_ENABLE_TSFPQ
508     rx_nPackets = 0;    /* in TSFPQ version, rx_nPackets is managed by rxi_MorePackets* */
509     rxi_MorePacketsTSFPQ(rx_extraPackets + RX_MAX_QUOTA + 2, RX_TS_FPQ_FLUSH_GLOBAL, 0);
510 #else /* RX_ENABLE_TSFPQ */
511     rx_nPackets = rx_extraPackets + RX_MAX_QUOTA + 2;   /* fudge */
512     rxi_MorePackets(rx_nPackets);
513 #endif /* RX_ENABLE_TSFPQ */
514     rx_CheckPackets();
515
516     NETPRI;
517
518     clock_Init();
519
520 #if defined(AFS_NT40_ENV) && !defined(AFS_PTHREAD_ENV)
521     tv.tv_sec = clock_now.sec;
522     tv.tv_usec = clock_now.usec;
523     srand((unsigned int)tv.tv_usec);
524 #else
525     osi_GetTime(&tv);
526 #endif
527     if (port) {
528         rx_port = port;
529     } else {
530 #if defined(KERNEL) && !defined(UKERNEL)
531         /* Really, this should never happen in a real kernel */
532         rx_port = 0;
533 #else
534         struct sockaddr_in addr;
535 #ifdef AFS_NT40_ENV
536         int addrlen = sizeof(addr);
537 #else
538         socklen_t addrlen = sizeof(addr);
539 #endif
540         if (getsockname((int)rx_socket, (struct sockaddr *)&addr, &addrlen)) {
541             rx_Finalize();
542             return -1;
543         }
544         rx_port = addr.sin_port;
545 #endif
546     }
547     rx_stats.minRtt.sec = 9999999;
548 #ifdef  KERNEL
549     rx_SetEpoch(tv.tv_sec | 0x80000000);
550 #else
551     rx_SetEpoch(tv.tv_sec);     /* Start time of this package, rxkad
552                                  * will provide a randomer value. */
553 #endif
554     MUTEX_ENTER(&rx_quota_mutex);
555     rxi_dataQuota += rx_extraQuota; /* + extra pkts caller asked to rsrv */
556     MUTEX_EXIT(&rx_quota_mutex);
557     /* *Slightly* random start time for the cid.  This is just to help
558      * out with the hashing function at the peer */
559     rx_nextCid = ((tv.tv_sec ^ tv.tv_usec) << RX_CIDSHIFT);
560     rx_connHashTable = (struct rx_connection **)htable;
561     rx_peerHashTable = (struct rx_peer **)ptable;
562
563     rx_lastAckDelay.sec = 0;
564     rx_lastAckDelay.usec = 400000;      /* 400 milliseconds */
565     rx_hardAckDelay.sec = 0;
566     rx_hardAckDelay.usec = 100000;      /* 100 milliseconds */
567     rx_softAckDelay.sec = 0;
568     rx_softAckDelay.usec = 100000;      /* 100 milliseconds */
569
570     rxevent_Init(20, rxi_ReScheduleEvents);
571
572     /* Initialize various global queues */
573     queue_Init(&rx_idleServerQueue);
574     queue_Init(&rx_incomingCallQueue);
575     queue_Init(&rx_freeCallQueue);
576
577 #if defined(AFS_NT40_ENV) && !defined(KERNEL)
578     /* Initialize our list of usable IP addresses. */
579     rx_GetIFInfo();
580 #endif
581
582     /* Start listener process (exact function is dependent on the
583      * implementation environment--kernel or user space) */
584     rxi_StartListener();
585
586     USERPRI;
587     tmp_status = rxinit_status = 0;
588     UNLOCK_RX_INIT;
589     return tmp_status;
590 }
591
592 int
593 rx_Init(u_int port)
594 {
595     return rx_InitHost(htonl(INADDR_ANY), port);
596 }
597
598 /* called with unincremented nRequestsRunning to see if it is OK to start
599  * a new thread in this service.  Could be "no" for two reasons: over the
600  * max quota, or would prevent others from reaching their min quota.
601  */
602 #ifdef RX_ENABLE_LOCKS
603 /* This verion of QuotaOK reserves quota if it's ok while the
604  * rx_serverPool_lock is held.  Return quota using ReturnToServerPool().
605  */
606 static int
607 QuotaOK(struct rx_service *aservice)
608 {
609     /* check if over max quota */
610     if (aservice->nRequestsRunning >= aservice->maxProcs) {
611         return 0;
612     }
613
614     /* under min quota, we're OK */
615     /* otherwise, can use only if there are enough to allow everyone
616      * to go to their min quota after this guy starts.
617      */
618
619     MUTEX_ENTER(&rx_quota_mutex);
620     if ((aservice->nRequestsRunning < aservice->minProcs)
621         || (rxi_availProcs > rxi_minDeficit)) {
622         aservice->nRequestsRunning++;
623         /* just started call in minProcs pool, need fewer to maintain
624          * guarantee */
625         if (aservice->nRequestsRunning <= aservice->minProcs)
626             rxi_minDeficit--;
627         rxi_availProcs--;
628         MUTEX_EXIT(&rx_quota_mutex);
629         return 1;
630     }
631     MUTEX_EXIT(&rx_quota_mutex);
632
633     return 0;
634 }
635
636 static void
637 ReturnToServerPool(struct rx_service *aservice)
638 {
639     aservice->nRequestsRunning--;
640     MUTEX_ENTER(&rx_quota_mutex);
641     if (aservice->nRequestsRunning < aservice->minProcs)
642         rxi_minDeficit++;
643     rxi_availProcs++;
644     MUTEX_EXIT(&rx_quota_mutex);
645 }
646
647 #else /* RX_ENABLE_LOCKS */
648 static int
649 QuotaOK(struct rx_service *aservice)
650 {
651     int rc = 0;
652     /* under min quota, we're OK */
653     if (aservice->nRequestsRunning < aservice->minProcs)
654         return 1;
655
656     /* check if over max quota */
657     if (aservice->nRequestsRunning >= aservice->maxProcs)
658         return 0;
659
660     /* otherwise, can use only if there are enough to allow everyone
661      * to go to their min quota after this guy starts.
662      */
663     if (rxi_availProcs > rxi_minDeficit)
664         rc = 1;
665     return rc;
666 }
667 #endif /* RX_ENABLE_LOCKS */
668
669 #ifndef KERNEL
670 /* Called by rx_StartServer to start up lwp's to service calls.
671    NExistingProcs gives the number of procs already existing, and which
672    therefore needn't be created. */
673 void
674 rxi_StartServerProcs(int nExistingProcs)
675 {
676     struct rx_service *service;
677     int i;
678     int maxdiff = 0;
679     int nProcs = 0;
680
681     /* For each service, reserve N processes, where N is the "minimum"
682      * number of processes that MUST be able to execute a request in parallel,
683      * at any time, for that process.  Also compute the maximum difference
684      * between any service's maximum number of processes that can run
685      * (i.e. the maximum number that ever will be run, and a guarantee
686      * that this number will run if other services aren't running), and its
687      * minimum number.  The result is the extra number of processes that
688      * we need in order to provide the latter guarantee */
689     for (i = 0; i < RX_MAX_SERVICES; i++) {
690         int diff;
691         service = rx_services[i];
692         if (service == (struct rx_service *)0)
693             break;
694         nProcs += service->minProcs;
695         diff = service->maxProcs - service->minProcs;
696         if (diff > maxdiff)
697             maxdiff = diff;
698     }
699     nProcs += maxdiff;          /* Extra processes needed to allow max number requested to run in any given service, under good conditions */
700     nProcs -= nExistingProcs;   /* Subtract the number of procs that were previously created for use as server procs */
701     for (i = 0; i < nProcs; i++) {
702         rxi_StartServerProc(rx_ServerProc, rx_stackSize);
703     }
704 }
705 #endif /* KERNEL */
706
707 #ifdef AFS_NT40_ENV
708 /* This routine is only required on Windows */
709 void
710 rx_StartClientThread(void)
711 {
712 #ifdef AFS_PTHREAD_ENV
713     pthread_t pid;
714     pid = pthread_self();
715 #endif /* AFS_PTHREAD_ENV */
716 }
717 #endif /* AFS_NT40_ENV */
718
719 /* This routine must be called if any services are exported.  If the
720  * donateMe flag is set, the calling process is donated to the server
721  * process pool */
722 void
723 rx_StartServer(int donateMe)
724 {
725     struct rx_service *service;
726     int i;
727     SPLVAR;
728     clock_NewTime();
729
730     NETPRI;
731     /* Start server processes, if necessary (exact function is dependent
732      * on the implementation environment--kernel or user space).  DonateMe
733      * will be 1 if there is 1 pre-existing proc, i.e. this one.  In this
734      * case, one less new proc will be created rx_StartServerProcs.
735      */
736     rxi_StartServerProcs(donateMe);
737
738     /* count up the # of threads in minProcs, and add set the min deficit to
739      * be that value, too.
740      */
741     for (i = 0; i < RX_MAX_SERVICES; i++) {
742         service = rx_services[i];
743         if (service == (struct rx_service *)0)
744             break;
745         MUTEX_ENTER(&rx_quota_mutex);
746         rxi_totalMin += service->minProcs;
747         /* below works even if a thread is running, since minDeficit would
748          * still have been decremented and later re-incremented.
749          */
750         rxi_minDeficit += service->minProcs;
751         MUTEX_EXIT(&rx_quota_mutex);
752     }
753
754     /* Turn on reaping of idle server connections */
755     rxi_ReapConnections(NULL, NULL, NULL);
756
757     USERPRI;
758
759     if (donateMe) {
760 #ifndef AFS_NT40_ENV
761 #ifndef KERNEL
762         char name[32];
763         static int nProcs;
764 #ifdef AFS_PTHREAD_ENV
765         pid_t pid;
766         pid = (pid_t) pthread_self();
767 #else /* AFS_PTHREAD_ENV */
768         PROCESS pid;
769         LWP_CurrentProcess(&pid);
770 #endif /* AFS_PTHREAD_ENV */
771
772         sprintf(name, "srv_%d", ++nProcs);
773         if (registerProgram)
774             (*registerProgram) (pid, name);
775 #endif /* KERNEL */
776 #endif /* AFS_NT40_ENV */
777         rx_ServerProc(NULL);    /* Never returns */
778     }
779 #ifdef RX_ENABLE_TSFPQ
780     /* no use leaving packets around in this thread's local queue if
781      * it isn't getting donated to the server thread pool. 
782      */
783     rxi_FlushLocalPacketsTSFPQ();
784 #endif /* RX_ENABLE_TSFPQ */
785     return;
786 }
787
788 /* Create a new client connection to the specified service, using the
789  * specified security object to implement the security model for this
790  * connection. */
791 struct rx_connection *
792 rx_NewConnection(afs_uint32 shost, u_short sport, u_short sservice,
793                  struct rx_securityClass *securityObject,
794                  int serviceSecurityIndex)
795 {
796     int hashindex, i;
797     afs_int32 cid;
798     struct rx_connection *conn;
799
800     SPLVAR;
801
802     clock_NewTime();
803     dpf(("rx_NewConnection(host %x, port %u, service %u, securityObject %x, serviceSecurityIndex %d)\n", ntohl(shost), ntohs(sport), sservice, securityObject, serviceSecurityIndex));
804
805     /* Vasilsi said: "NETPRI protects Cid and Alloc", but can this be true in
806      * the case of kmem_alloc? */
807     conn = rxi_AllocConnection();
808 #ifdef  RX_ENABLE_LOCKS
809     MUTEX_INIT(&conn->conn_call_lock, "conn call lock", MUTEX_DEFAULT, 0);
810     MUTEX_INIT(&conn->conn_data_lock, "conn data lock", MUTEX_DEFAULT, 0);
811     CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
812 #endif
813     NETPRI;
814     MUTEX_ENTER(&rx_connHashTable_lock);
815     cid = (rx_nextCid += RX_MAXCALLS);
816     conn->type = RX_CLIENT_CONNECTION;
817     conn->cid = cid;
818     conn->epoch = rx_epoch;
819     conn->peer = rxi_FindPeer(shost, sport, 0, 1);
820     conn->serviceId = sservice;
821     conn->securityObject = securityObject;
822     conn->securityData = (void *) 0;
823     conn->securityIndex = serviceSecurityIndex;
824     rx_SetConnDeadTime(conn, rx_connDeadTime);
825     conn->ackRate = RX_FAST_ACK_RATE;
826     conn->nSpecific = 0;
827     conn->specific = NULL;
828     conn->challengeEvent = NULL;
829     conn->delayedAbortEvent = NULL;
830     conn->abortCount = 0;
831     conn->error = 0;
832     for (i = 0; i < RX_MAXCALLS; i++) {
833         conn->twind[i] = rx_initSendWindow;
834         conn->rwind[i] = rx_initReceiveWindow;
835     }
836
837     RXS_NewConnection(securityObject, conn);
838     hashindex =
839         CONN_HASH(shost, sport, conn->cid, conn->epoch, RX_CLIENT_CONNECTION);
840
841     conn->refCount++;           /* no lock required since only this thread knows... */
842     conn->next = rx_connHashTable[hashindex];
843     rx_connHashTable[hashindex] = conn;
844     if (rx_stats_active)
845         rx_MutexIncrement(rx_stats.nClientConns, rx_stats_mutex);
846     MUTEX_EXIT(&rx_connHashTable_lock);
847     USERPRI;
848     return conn;
849 }
850
851 void
852 rx_SetConnDeadTime(struct rx_connection *conn, int seconds)
853 {
854     /* The idea is to set the dead time to a value that allows several
855      * keepalives to be dropped without timing out the connection. */
856     conn->secondsUntilDead = MAX(seconds, 6);
857     conn->secondsUntilPing = conn->secondsUntilDead / 6;
858 }
859
860 int rxi_lowPeerRefCount = 0;
861 int rxi_lowConnRefCount = 0;
862
863 /*
864  * Cleanup a connection that was destroyed in rxi_DestroyConnectioNoLock.
865  * NOTE: must not be called with rx_connHashTable_lock held.
866  */
867 void
868 rxi_CleanupConnection(struct rx_connection *conn)
869 {
870     /* Notify the service exporter, if requested, that this connection
871      * is being destroyed */
872     if (conn->type == RX_SERVER_CONNECTION && conn->service->destroyConnProc)
873         (*conn->service->destroyConnProc) (conn);
874
875     /* Notify the security module that this connection is being destroyed */
876     RXS_DestroyConnection(conn->securityObject, conn);
877
878     /* If this is the last connection using the rx_peer struct, set its
879      * idle time to now. rxi_ReapConnections will reap it if it's still
880      * idle (refCount == 0) after rx_idlePeerTime (60 seconds) have passed.
881      */
882     MUTEX_ENTER(&rx_peerHashTable_lock);
883     if (conn->peer->refCount < 2) {
884         conn->peer->idleWhen = clock_Sec();
885         if (conn->peer->refCount < 1) {
886             conn->peer->refCount = 1;
887             if (rx_stats_active) {
888                 MUTEX_ENTER(&rx_stats_mutex);
889                 rxi_lowPeerRefCount++;
890                 MUTEX_EXIT(&rx_stats_mutex);
891             }
892         }
893     }
894     conn->peer->refCount--;
895     MUTEX_EXIT(&rx_peerHashTable_lock);
896
897     if (rx_stats_active)
898     {
899         if (conn->type == RX_SERVER_CONNECTION)
900             rx_MutexDecrement(rx_stats.nServerConns, rx_stats_mutex);
901         else
902             rx_MutexDecrement(rx_stats.nClientConns, rx_stats_mutex);
903     }
904 #ifndef KERNEL
905     if (conn->specific) {
906         int i;
907         for (i = 0; i < conn->nSpecific; i++) {
908             if (conn->specific[i] && rxi_keyCreate_destructor[i])
909                 (*rxi_keyCreate_destructor[i]) (conn->specific[i]);
910             conn->specific[i] = NULL;
911         }
912         free(conn->specific);
913     }
914     conn->specific = NULL;
915     conn->nSpecific = 0;
916 #endif /* !KERNEL */
917
918     MUTEX_DESTROY(&conn->conn_call_lock);
919     MUTEX_DESTROY(&conn->conn_data_lock);
920     CV_DESTROY(&conn->conn_call_cv);
921
922     rxi_FreeConnection(conn);
923 }
924
925 /* Destroy the specified connection */
926 void
927 rxi_DestroyConnection(struct rx_connection *conn)
928 {
929     MUTEX_ENTER(&rx_connHashTable_lock);
930     rxi_DestroyConnectionNoLock(conn);
931     /* conn should be at the head of the cleanup list */
932     if (conn == rx_connCleanup_list) {
933         rx_connCleanup_list = rx_connCleanup_list->next;
934         MUTEX_EXIT(&rx_connHashTable_lock);
935         rxi_CleanupConnection(conn);
936     }
937 #ifdef RX_ENABLE_LOCKS
938     else {
939         MUTEX_EXIT(&rx_connHashTable_lock);
940     }
941 #endif /* RX_ENABLE_LOCKS */
942 }
943
944 static void
945 rxi_DestroyConnectionNoLock(struct rx_connection *conn)
946 {
947     struct rx_connection **conn_ptr;
948     int havecalls = 0;
949     struct rx_packet *packet;
950     int i;
951     SPLVAR;
952
953     clock_NewTime();
954
955     NETPRI;
956     MUTEX_ENTER(&conn->conn_data_lock);
957     if (conn->refCount > 0)
958         conn->refCount--;
959     else {
960         if (rx_stats_active) {
961             MUTEX_ENTER(&rx_stats_mutex);
962             rxi_lowConnRefCount++;
963             MUTEX_EXIT(&rx_stats_mutex);
964         }
965     }
966
967     if ((conn->refCount > 0) || (conn->flags & RX_CONN_BUSY)) {
968         /* Busy; wait till the last guy before proceeding */
969         MUTEX_EXIT(&conn->conn_data_lock);
970         USERPRI;
971         return;
972     }
973
974     /* If the client previously called rx_NewCall, but it is still
975      * waiting, treat this as a running call, and wait to destroy the
976      * connection later when the call completes. */
977     if ((conn->type == RX_CLIENT_CONNECTION)
978         && (conn->flags & RX_CONN_MAKECALL_WAITING)) {
979         conn->flags |= RX_CONN_DESTROY_ME;
980         MUTEX_EXIT(&conn->conn_data_lock);
981         USERPRI;
982         return;
983     }
984     MUTEX_EXIT(&conn->conn_data_lock);
985
986     /* Check for extant references to this connection */
987     for (i = 0; i < RX_MAXCALLS; i++) {
988         struct rx_call *call = conn->call[i];
989         if (call) {
990             havecalls = 1;
991             if (conn->type == RX_CLIENT_CONNECTION) {
992                 MUTEX_ENTER(&call->lock);
993                 if (call->delayedAckEvent) {
994                     /* Push the final acknowledgment out now--there
995                      * won't be a subsequent call to acknowledge the
996                      * last reply packets */
997                     rxevent_Cancel(call->delayedAckEvent, call,
998                                    RX_CALL_REFCOUNT_DELAY);
999                     if (call->state == RX_STATE_PRECALL
1000                         || call->state == RX_STATE_ACTIVE) {
1001                         rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
1002                     } else {
1003                         rxi_AckAll(NULL, call, 0);
1004                     }
1005                 }
1006                 MUTEX_EXIT(&call->lock);
1007             }
1008         }
1009     }
1010 #ifdef RX_ENABLE_LOCKS
1011     if (!havecalls) {
1012         if (MUTEX_TRYENTER(&conn->conn_data_lock)) {
1013             MUTEX_EXIT(&conn->conn_data_lock);
1014         } else {
1015             /* Someone is accessing a packet right now. */
1016             havecalls = 1;
1017         }
1018     }
1019 #endif /* RX_ENABLE_LOCKS */
1020
1021     if (havecalls) {
1022         /* Don't destroy the connection if there are any call
1023          * structures still in use */
1024         MUTEX_ENTER(&conn->conn_data_lock);
1025         conn->flags |= RX_CONN_DESTROY_ME;
1026         MUTEX_EXIT(&conn->conn_data_lock);
1027         USERPRI;
1028         return;
1029     }
1030
1031     if (conn->delayedAbortEvent) {
1032         rxevent_Cancel(conn->delayedAbortEvent, (struct rx_call *)0, 0);
1033         packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
1034         if (packet) {
1035             MUTEX_ENTER(&conn->conn_data_lock);
1036             rxi_SendConnectionAbort(conn, packet, 0, 1);
1037             MUTEX_EXIT(&conn->conn_data_lock);
1038             rxi_FreePacket(packet);
1039         }
1040     }
1041
1042     /* Remove from connection hash table before proceeding */
1043     conn_ptr =
1044         &rx_connHashTable[CONN_HASH
1045                           (peer->host, peer->port, conn->cid, conn->epoch,
1046                            conn->type)];
1047     for (; *conn_ptr; conn_ptr = &(*conn_ptr)->next) {
1048         if (*conn_ptr == conn) {
1049             *conn_ptr = conn->next;
1050             break;
1051         }
1052     }
1053     /* if the conn that we are destroying was the last connection, then we
1054      * clear rxLastConn as well */
1055     if (rxLastConn == conn)
1056         rxLastConn = 0;
1057
1058     /* Make sure the connection is completely reset before deleting it. */
1059     /* get rid of pending events that could zap us later */
1060     if (conn->challengeEvent)
1061         rxevent_Cancel(conn->challengeEvent, (struct rx_call *)0, 0);
1062     if (conn->checkReachEvent)
1063         rxevent_Cancel(conn->checkReachEvent, (struct rx_call *)0, 0);
1064
1065     /* Add the connection to the list of destroyed connections that
1066      * need to be cleaned up. This is necessary to avoid deadlocks
1067      * in the routines we call to inform others that this connection is
1068      * being destroyed. */
1069     conn->next = rx_connCleanup_list;
1070     rx_connCleanup_list = conn;
1071 }
1072
1073 /* Externally available version */
1074 void
1075 rx_DestroyConnection(struct rx_connection *conn)
1076 {
1077     SPLVAR;
1078
1079     NETPRI;
1080     rxi_DestroyConnection(conn);
1081     USERPRI;
1082 }
1083
1084 void
1085 rx_GetConnection(struct rx_connection *conn)
1086 {
1087     SPLVAR;
1088
1089     NETPRI;
1090     MUTEX_ENTER(&conn->conn_data_lock);
1091     conn->refCount++;
1092     MUTEX_EXIT(&conn->conn_data_lock);
1093     USERPRI;
1094 }
1095
1096 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
1097 /* Wait for the transmit queue to no longer be busy. 
1098  * requires the call->lock to be held */
1099 static void rxi_WaitforTQBusy(struct rx_call *call) {
1100     while (call->flags & RX_CALL_TQ_BUSY) {
1101         call->flags |= RX_CALL_TQ_WAIT;
1102         call->tqWaiters++;
1103 #ifdef RX_ENABLE_LOCKS
1104         osirx_AssertMine(&call->lock, "rxi_WaitforTQ lock");
1105         CV_WAIT(&call->cv_tq, &call->lock);
1106 #else /* RX_ENABLE_LOCKS */
1107         osi_rxSleep(&call->tq);
1108 #endif /* RX_ENABLE_LOCKS */
1109         call->tqWaiters--;
1110         if (call->tqWaiters == 0) {
1111             call->flags &= ~RX_CALL_TQ_WAIT;
1112         }
1113     }
1114 }
1115 #endif
1116
1117 /* Start a new rx remote procedure call, on the specified connection.
1118  * If wait is set to 1, wait for a free call channel; otherwise return
1119  * 0.  Maxtime gives the maximum number of seconds this call may take,
1120  * after rx_NewCall returns.  After this time interval, a call to any
1121  * of rx_SendData, rx_ReadData, etc. will fail with RX_CALL_TIMEOUT.
1122  * For fine grain locking, we hold the conn_call_lock in order to 
1123  * to ensure that we don't get signalle after we found a call in an active
1124  * state and before we go to sleep.
1125  */
1126 struct rx_call *
1127 rx_NewCall(struct rx_connection *conn)
1128 {
1129     int i;
1130     struct rx_call *call;
1131     struct clock queueTime;
1132     SPLVAR;
1133
1134     clock_NewTime();
1135     dpf(("rx_NewCall(conn %x)\n", conn));
1136
1137     NETPRI;
1138     clock_GetTime(&queueTime);
1139     MUTEX_ENTER(&conn->conn_call_lock);
1140
1141     /*
1142      * Check if there are others waiting for a new call.
1143      * If so, let them go first to avoid starving them.
1144      * This is a fairly simple scheme, and might not be
1145      * a complete solution for large numbers of waiters.
1146      * 
1147      * makeCallWaiters keeps track of the number of 
1148      * threads waiting to make calls and the 
1149      * RX_CONN_MAKECALL_WAITING flag bit is used to 
1150      * indicate that there are indeed calls waiting.
1151      * The flag is set when the waiter is incremented.
1152      * It is only cleared in rx_EndCall when 
1153      * makeCallWaiters is 0.  This prevents us from 
1154      * accidently destroying the connection while it
1155      * is potentially about to be used.
1156      */
1157     MUTEX_ENTER(&conn->conn_data_lock);
1158     if (conn->makeCallWaiters) {
1159         conn->flags |= RX_CONN_MAKECALL_WAITING;
1160         conn->makeCallWaiters++;
1161         MUTEX_EXIT(&conn->conn_data_lock);
1162
1163 #ifdef  RX_ENABLE_LOCKS
1164         CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
1165 #else
1166         osi_rxSleep(conn);
1167 #endif
1168         MUTEX_ENTER(&conn->conn_data_lock);
1169         conn->makeCallWaiters--;
1170     } 
1171     MUTEX_EXIT(&conn->conn_data_lock);
1172
1173     for (;;) {
1174         for (i = 0; i < RX_MAXCALLS; i++) {
1175             call = conn->call[i];
1176             if (call) {
1177                 MUTEX_ENTER(&call->lock);
1178                 if (call->state == RX_STATE_DALLY) {
1179                     rxi_ResetCall(call, 0);
1180                     (*call->callNumber)++;
1181                     break;
1182                 }
1183                 MUTEX_EXIT(&call->lock);
1184             } else {
1185                 call = rxi_NewCall(conn, i);
1186                 break;
1187             }
1188         }
1189         if (i < RX_MAXCALLS) {
1190             break;
1191         }
1192         MUTEX_ENTER(&conn->conn_data_lock);
1193         conn->flags |= RX_CONN_MAKECALL_WAITING;
1194         conn->makeCallWaiters++;
1195         MUTEX_EXIT(&conn->conn_data_lock);
1196
1197 #ifdef  RX_ENABLE_LOCKS
1198         CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
1199 #else
1200         osi_rxSleep(conn);
1201 #endif
1202         MUTEX_ENTER(&conn->conn_data_lock);
1203         conn->makeCallWaiters--;
1204         MUTEX_EXIT(&conn->conn_data_lock);
1205     }
1206     /*
1207      * Wake up anyone else who might be giving us a chance to
1208      * run (see code above that avoids resource starvation).
1209      */
1210 #ifdef  RX_ENABLE_LOCKS
1211     CV_BROADCAST(&conn->conn_call_cv);
1212 #else
1213     osi_rxWakeup(conn);
1214 #endif
1215
1216     CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
1217
1218     /* Client is initially in send mode */
1219     call->state = RX_STATE_ACTIVE;
1220     call->error = conn->error;
1221     if (call->error)
1222         call->mode = RX_MODE_ERROR;
1223     else
1224         call->mode = RX_MODE_SENDING;
1225     
1226     /* remember start time for call in case we have hard dead time limit */
1227     call->queueTime = queueTime;
1228     clock_GetTime(&call->startTime);
1229     hzero(call->bytesSent);
1230     hzero(call->bytesRcvd);
1231
1232     /* Turn on busy protocol. */
1233     rxi_KeepAliveOn(call);
1234
1235     MUTEX_EXIT(&call->lock);
1236     MUTEX_EXIT(&conn->conn_call_lock);
1237     USERPRI;
1238
1239 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
1240     /* Now, if TQ wasn't cleared earlier, do it now. */
1241     MUTEX_ENTER(&call->lock);
1242     rxi_WaitforTQBusy(call);
1243     if (call->flags & RX_CALL_TQ_CLEARME) {
1244         rxi_ClearTransmitQueue(call, 1);
1245         /*queue_Init(&call->tq);*/
1246     }
1247     MUTEX_EXIT(&call->lock);
1248 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
1249
1250     dpf(("rx_NewCall(call %x)\n", call));
1251     return call;
1252 }
1253
1254 int
1255 rxi_HasActiveCalls(struct rx_connection *aconn)
1256 {
1257     int i;
1258     struct rx_call *tcall;
1259     SPLVAR;
1260
1261     NETPRI;
1262     for (i = 0; i < RX_MAXCALLS; i++) {
1263         if ((tcall = aconn->call[i])) {
1264             if ((tcall->state == RX_STATE_ACTIVE)
1265                 || (tcall->state == RX_STATE_PRECALL)) {
1266                 USERPRI;
1267                 return 1;
1268             }
1269         }
1270     }
1271     USERPRI;
1272     return 0;
1273 }
1274
1275 int
1276 rxi_GetCallNumberVector(struct rx_connection *aconn,
1277                         afs_int32 * aint32s)
1278 {
1279     int i;
1280     struct rx_call *tcall;
1281     SPLVAR;
1282
1283     NETPRI;
1284     for (i = 0; i < RX_MAXCALLS; i++) {
1285         if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
1286             aint32s[i] = aconn->callNumber[i] + 1;
1287         else
1288             aint32s[i] = aconn->callNumber[i];
1289     }
1290     USERPRI;
1291     return 0;
1292 }
1293
1294 int
1295 rxi_SetCallNumberVector(struct rx_connection *aconn,
1296                         afs_int32 * aint32s)
1297 {
1298     int i;
1299     struct rx_call *tcall;
1300     SPLVAR;
1301
1302     NETPRI;
1303     for (i = 0; i < RX_MAXCALLS; i++) {
1304         if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
1305             aconn->callNumber[i] = aint32s[i] - 1;
1306         else
1307             aconn->callNumber[i] = aint32s[i];
1308     }
1309     USERPRI;
1310     return 0;
1311 }
1312
1313 /* Advertise a new service.  A service is named locally by a UDP port
1314  * number plus a 16-bit service id.  Returns (struct rx_service *) 0
1315  * on a failure. 
1316  *
1317      char *serviceName;  Name for identification purposes (e.g. the
1318                          service name might be used for probing for
1319                          statistics) */
1320 struct rx_service *
1321 rx_NewServiceHost(afs_uint32 host, u_short port, u_short serviceId, 
1322                   char *serviceName, struct rx_securityClass **securityObjects,
1323                   int nSecurityObjects, 
1324                   afs_int32(*serviceProc) (struct rx_call * acall))
1325 {
1326     osi_socket socket = OSI_NULLSOCKET;
1327     struct rx_service *tservice;
1328     int i;
1329     SPLVAR;
1330
1331     clock_NewTime();
1332
1333     if (serviceId == 0) {
1334         (osi_Msg
1335          "rx_NewService:  service id for service %s is not non-zero.\n",
1336          serviceName);
1337         return 0;
1338     }
1339     if (port == 0) {
1340         if (rx_port == 0) {
1341             (osi_Msg
1342              "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",
1343              serviceName);
1344             return 0;
1345         }
1346         port = rx_port;
1347         socket = rx_socket;
1348     }
1349
1350     tservice = rxi_AllocService();
1351     NETPRI;
1352     for (i = 0; i < RX_MAX_SERVICES; i++) {
1353         struct rx_service *service = rx_services[i];
1354         if (service) {
1355             if (port == service->servicePort && host == service->serviceHost) {
1356                 if (service->serviceId == serviceId) {
1357                     /* The identical service has already been
1358                      * installed; if the caller was intending to
1359                      * change the security classes used by this
1360                      * service, he/she loses. */
1361                     (osi_Msg
1362                      "rx_NewService: tried to install service %s with service id %d, which is already in use for service %s\n",
1363                      serviceName, serviceId, service->serviceName);
1364                     USERPRI;
1365                     rxi_FreeService(tservice);
1366                     return service;
1367                 }
1368                 /* Different service, same port: re-use the socket
1369                  * which is bound to the same port */
1370                 socket = service->socket;
1371             }
1372         } else {
1373             if (socket == OSI_NULLSOCKET) {
1374                 /* If we don't already have a socket (from another
1375                  * service on same port) get a new one */
1376                 socket = rxi_GetHostUDPSocket(host, port);
1377                 if (socket == OSI_NULLSOCKET) {
1378                     USERPRI;
1379                     rxi_FreeService(tservice);
1380                     return 0;
1381                 }
1382             }
1383             service = tservice;
1384             service->socket = socket;
1385             service->serviceHost = host;
1386             service->servicePort = port;
1387             service->serviceId = serviceId;
1388             service->serviceName = serviceName;
1389             service->nSecurityObjects = nSecurityObjects;
1390             service->securityObjects = securityObjects;
1391             service->minProcs = 0;
1392             service->maxProcs = 1;
1393             service->idleDeadTime = 60;
1394             service->idleDeadErr = 0;
1395             service->connDeadTime = rx_connDeadTime;
1396             service->executeRequestProc = serviceProc;
1397             service->checkReach = 0;
1398             rx_services[i] = service;   /* not visible until now */
1399             USERPRI;
1400             return service;
1401         }
1402     }
1403     USERPRI;
1404     rxi_FreeService(tservice);
1405     (osi_Msg "rx_NewService: cannot support > %d services\n",
1406      RX_MAX_SERVICES);
1407     return 0;
1408 }
1409
1410 /* Set configuration options for all of a service's security objects */
1411
1412 afs_int32 
1413 rx_SetSecurityConfiguration(struct rx_service *service, 
1414                             rx_securityConfigVariables type,
1415                             void *value)
1416 {
1417     int i;
1418     for (i = 0; i<service->nSecurityObjects; i++) {
1419         if (service->securityObjects[i]) {
1420             RXS_SetConfiguration(service->securityObjects[i], NULL, type, 
1421                                  value, NULL);
1422         }
1423     }
1424     return 0;
1425 }
1426
1427 struct rx_service *
1428 rx_NewService(u_short port, u_short serviceId, char *serviceName,
1429               struct rx_securityClass **securityObjects, int nSecurityObjects,
1430               afs_int32(*serviceProc) (struct rx_call * acall))
1431 {
1432     return rx_NewServiceHost(htonl(INADDR_ANY), port, serviceId, serviceName, securityObjects, nSecurityObjects, serviceProc);
1433 }
1434
1435 /* Generic request processing loop. This routine should be called
1436  * by the implementation dependent rx_ServerProc. If socketp is
1437  * non-null, it will be set to the file descriptor that this thread
1438  * is now listening on. If socketp is null, this routine will never
1439  * returns. */
1440 void
1441 rxi_ServerProc(int threadID, struct rx_call *newcall, osi_socket * socketp)
1442 {
1443     struct rx_call *call;
1444     afs_int32 code;
1445     struct rx_service *tservice = NULL;
1446
1447     for (;;) {
1448         if (newcall) {
1449             call = newcall;
1450             newcall = NULL;
1451         } else {
1452             call = rx_GetCall(threadID, tservice, socketp);
1453             if (socketp && *socketp != OSI_NULLSOCKET) {
1454                 /* We are now a listener thread */
1455                 return;
1456             }
1457         }
1458
1459         /* if server is restarting( typically smooth shutdown) then do not
1460          * allow any new calls.
1461          */
1462
1463         if (rx_tranquil && (call != NULL)) {
1464             SPLVAR;
1465
1466             NETPRI;
1467             MUTEX_ENTER(&call->lock);
1468
1469             rxi_CallError(call, RX_RESTARTING);
1470             rxi_SendCallAbort(call, (struct rx_packet *)0, 0, 0);
1471
1472             MUTEX_EXIT(&call->lock);
1473             USERPRI;
1474         }
1475 #ifdef  KERNEL
1476         if (afs_termState == AFSOP_STOP_RXCALLBACK) {
1477 #ifdef RX_ENABLE_LOCKS
1478             AFS_GLOCK();
1479 #endif /* RX_ENABLE_LOCKS */
1480             afs_termState = AFSOP_STOP_AFS;
1481             afs_osi_Wakeup(&afs_termState);
1482 #ifdef RX_ENABLE_LOCKS
1483             AFS_GUNLOCK();
1484 #endif /* RX_ENABLE_LOCKS */
1485             return;
1486         }
1487 #endif
1488
1489         tservice = call->conn->service;
1490
1491         if (tservice->beforeProc)
1492             (*tservice->beforeProc) (call);
1493
1494         code = call->conn->service->executeRequestProc(call);
1495
1496         if (tservice->afterProc)
1497             (*tservice->afterProc) (call, code);
1498
1499         rx_EndCall(call, code);
1500         if (rx_stats_active) {
1501             MUTEX_ENTER(&rx_stats_mutex);
1502             rxi_nCalls++;
1503             MUTEX_EXIT(&rx_stats_mutex);
1504         }
1505     }
1506 }
1507
1508
1509 void
1510 rx_WakeupServerProcs(void)
1511 {
1512     struct rx_serverQueueEntry *np, *tqp;
1513     SPLVAR;
1514
1515     NETPRI;
1516     MUTEX_ENTER(&rx_serverPool_lock);
1517
1518 #ifdef RX_ENABLE_LOCKS
1519     if (rx_waitForPacket)
1520         CV_BROADCAST(&rx_waitForPacket->cv);
1521 #else /* RX_ENABLE_LOCKS */
1522     if (rx_waitForPacket)
1523         osi_rxWakeup(rx_waitForPacket);
1524 #endif /* RX_ENABLE_LOCKS */
1525     MUTEX_ENTER(&freeSQEList_lock);
1526     for (np = rx_FreeSQEList; np; np = tqp) {
1527         tqp = *(struct rx_serverQueueEntry **)np;
1528 #ifdef RX_ENABLE_LOCKS
1529         CV_BROADCAST(&np->cv);
1530 #else /* RX_ENABLE_LOCKS */
1531         osi_rxWakeup(np);
1532 #endif /* RX_ENABLE_LOCKS */
1533     }
1534     MUTEX_EXIT(&freeSQEList_lock);
1535     for (queue_Scan(&rx_idleServerQueue, np, tqp, rx_serverQueueEntry)) {
1536 #ifdef RX_ENABLE_LOCKS
1537         CV_BROADCAST(&np->cv);
1538 #else /* RX_ENABLE_LOCKS */
1539         osi_rxWakeup(np);
1540 #endif /* RX_ENABLE_LOCKS */
1541     }
1542     MUTEX_EXIT(&rx_serverPool_lock);
1543     USERPRI;
1544 }
1545
1546 /* meltdown:
1547  * One thing that seems to happen is that all the server threads get
1548  * tied up on some empty or slow call, and then a whole bunch of calls
1549  * arrive at once, using up the packet pool, so now there are more 
1550  * empty calls.  The most critical resources here are server threads
1551  * and the free packet pool.  The "doreclaim" code seems to help in
1552  * general.  I think that eventually we arrive in this state: there
1553  * are lots of pending calls which do have all their packets present,
1554  * so they won't be reclaimed, are multi-packet calls, so they won't
1555  * be scheduled until later, and thus are tying up most of the free 
1556  * packet pool for a very long time.
1557  * future options:
1558  * 1.  schedule multi-packet calls if all the packets are present.  
1559  * Probably CPU-bound operation, useful to return packets to pool. 
1560  * Do what if there is a full window, but the last packet isn't here?
1561  * 3.  preserve one thread which *only* runs "best" calls, otherwise
1562  * it sleeps and waits for that type of call.
1563  * 4.  Don't necessarily reserve a whole window for each thread.  In fact, 
1564  * the current dataquota business is badly broken.  The quota isn't adjusted
1565  * to reflect how many packets are presently queued for a running call.
1566  * So, when we schedule a queued call with a full window of packets queued
1567  * up for it, that *should* free up a window full of packets for other 2d-class
1568  * calls to be able to use from the packet pool.  But it doesn't.
1569  *
1570  * NB.  Most of the time, this code doesn't run -- since idle server threads
1571  * sit on the idle server queue and are assigned by "...ReceivePacket" as soon
1572  * as a new call arrives.
1573  */
1574 /* Sleep until a call arrives.  Returns a pointer to the call, ready
1575  * for an rx_Read. */
1576 #ifdef RX_ENABLE_LOCKS
1577 struct rx_call *
1578 rx_GetCall(int tno, struct rx_service *cur_service, osi_socket * socketp)
1579 {
1580     struct rx_serverQueueEntry *sq;
1581     struct rx_call *call = (struct rx_call *)0;
1582     struct rx_service *service = NULL;
1583     SPLVAR;
1584
1585     MUTEX_ENTER(&freeSQEList_lock);
1586
1587     if ((sq = rx_FreeSQEList)) {
1588         rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
1589         MUTEX_EXIT(&freeSQEList_lock);
1590     } else {                    /* otherwise allocate a new one and return that */
1591         MUTEX_EXIT(&freeSQEList_lock);
1592         sq = (struct rx_serverQueueEntry *)
1593             rxi_Alloc(sizeof(struct rx_serverQueueEntry));
1594         MUTEX_INIT(&sq->lock, "server Queue lock", MUTEX_DEFAULT, 0);
1595         CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
1596     }
1597
1598     MUTEX_ENTER(&rx_serverPool_lock);
1599     if (cur_service != NULL) {
1600         ReturnToServerPool(cur_service);
1601     }
1602     while (1) {
1603         if (queue_IsNotEmpty(&rx_incomingCallQueue)) {
1604             struct rx_call *tcall, *ncall, *choice2 = NULL;
1605
1606             /* Scan for eligible incoming calls.  A call is not eligible
1607              * if the maximum number of calls for its service type are
1608              * already executing */
1609             /* One thread will process calls FCFS (to prevent starvation),
1610              * while the other threads may run ahead looking for calls which
1611              * have all their input data available immediately.  This helps 
1612              * keep threads from blocking, waiting for data from the client. */
1613             for (queue_Scan(&rx_incomingCallQueue, tcall, ncall, rx_call)) {
1614                 service = tcall->conn->service;
1615                 if (!QuotaOK(service)) {
1616                     continue;
1617                 }
1618                 MUTEX_ENTER(&rx_pthread_mutex);
1619                 if (tno == rxi_fcfs_thread_num
1620                     || !tcall->queue_item_header.next) {
1621                     MUTEX_EXIT(&rx_pthread_mutex);
1622                     /* If we're the fcfs thread , then  we'll just use 
1623                      * this call. If we haven't been able to find an optimal 
1624                      * choice, and we're at the end of the list, then use a 
1625                      * 2d choice if one has been identified.  Otherwise... */
1626                     call = (choice2 ? choice2 : tcall);
1627                     service = call->conn->service;
1628                 } else {
1629                     MUTEX_EXIT(&rx_pthread_mutex);
1630                     if (!queue_IsEmpty(&tcall->rq)) {
1631                         struct rx_packet *rp;
1632                         rp = queue_First(&tcall->rq, rx_packet);
1633                         if (rp->header.seq == 1) {
1634                             if (!meltdown_1pkt
1635                                 || (rp->header.flags & RX_LAST_PACKET)) {
1636                                 call = tcall;
1637                             } else if (rxi_2dchoice && !choice2
1638                                        && !(tcall->flags & RX_CALL_CLEARED)
1639                                        && (tcall->rprev > rxi_HardAckRate)) {
1640                                 choice2 = tcall;
1641                             } else
1642                                 rxi_md2cnt++;
1643                         }
1644                     }
1645                 }
1646                 if (call) {
1647                     break;
1648                 } else {
1649                     ReturnToServerPool(service);
1650                 }
1651             }
1652         }
1653
1654         if (call) {
1655             queue_Remove(call);
1656             MUTEX_EXIT(&rx_serverPool_lock);
1657             MUTEX_ENTER(&call->lock);
1658
1659             if (call->flags & RX_CALL_WAIT_PROC) {
1660                 call->flags &= ~RX_CALL_WAIT_PROC;
1661                 MUTEX_ENTER(&rx_waiting_mutex);
1662                 rx_nWaiting--;
1663                 MUTEX_EXIT(&rx_waiting_mutex);
1664             }
1665
1666             if (call->state != RX_STATE_PRECALL || call->error) {
1667                 MUTEX_EXIT(&call->lock);
1668                 MUTEX_ENTER(&rx_serverPool_lock);
1669                 ReturnToServerPool(service);
1670                 call = NULL;
1671                 continue;
1672             }
1673
1674             if (queue_IsEmpty(&call->rq)
1675                 || queue_First(&call->rq, rx_packet)->header.seq != 1)
1676                 rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
1677
1678             CLEAR_CALL_QUEUE_LOCK(call);
1679             break;
1680         } else {
1681             /* If there are no eligible incoming calls, add this process
1682              * to the idle server queue, to wait for one */
1683             sq->newcall = 0;
1684             sq->tno = tno;
1685             if (socketp) {
1686                 *socketp = OSI_NULLSOCKET;
1687             }
1688             sq->socketp = socketp;
1689             queue_Append(&rx_idleServerQueue, sq);
1690 #ifndef AFS_AIX41_ENV
1691             rx_waitForPacket = sq;
1692 #else
1693             rx_waitingForPacket = sq;
1694 #endif /* AFS_AIX41_ENV */
1695             do {
1696                 CV_WAIT(&sq->cv, &rx_serverPool_lock);
1697 #ifdef  KERNEL
1698                 if (afs_termState == AFSOP_STOP_RXCALLBACK) {
1699                     MUTEX_EXIT(&rx_serverPool_lock);
1700                     return (struct rx_call *)0;
1701                 }
1702 #endif
1703             } while (!(call = sq->newcall)
1704                      && !(socketp && *socketp != OSI_NULLSOCKET));
1705             MUTEX_EXIT(&rx_serverPool_lock);
1706             if (call) {
1707                 MUTEX_ENTER(&call->lock);
1708             }
1709             break;
1710         }
1711     }
1712
1713     MUTEX_ENTER(&freeSQEList_lock);
1714     *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
1715     rx_FreeSQEList = sq;
1716     MUTEX_EXIT(&freeSQEList_lock);
1717
1718     if (call) {
1719         clock_GetTime(&call->startTime);
1720         call->state = RX_STATE_ACTIVE;
1721         call->mode = RX_MODE_RECEIVING;
1722 #ifdef RX_KERNEL_TRACE
1723         if (ICL_SETACTIVE(afs_iclSetp)) {
1724             int glockOwner = ISAFS_GLOCK();
1725             if (!glockOwner)
1726                 AFS_GLOCK();
1727             afs_Trace3(afs_iclSetp, CM_TRACE_WASHERE, ICL_TYPE_STRING,
1728                        __FILE__, ICL_TYPE_INT32, __LINE__, ICL_TYPE_POINTER,
1729                        call);
1730             if (!glockOwner)
1731                 AFS_GUNLOCK();
1732         }
1733 #endif
1734
1735         rxi_calltrace(RX_CALL_START, call);
1736         dpf(("rx_GetCall(port=%d, service=%d) ==> call %x\n",
1737              call->conn->service->servicePort, call->conn->service->serviceId,
1738              call));
1739
1740         CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
1741         MUTEX_EXIT(&call->lock);
1742     } else {
1743         dpf(("rx_GetCall(socketp=0x%x, *socketp=0x%x)\n", socketp, *socketp));
1744     }
1745
1746     return call;
1747 }
1748 #else /* RX_ENABLE_LOCKS */
1749 struct rx_call *
1750 rx_GetCall(int tno, struct rx_service *cur_service, osi_socket * socketp)
1751 {
1752     struct rx_serverQueueEntry *sq;
1753     struct rx_call *call = (struct rx_call *)0, *choice2;
1754     struct rx_service *service = NULL;
1755     SPLVAR;
1756
1757     NETPRI;
1758     MUTEX_ENTER(&freeSQEList_lock);
1759
1760     if ((sq = rx_FreeSQEList)) {
1761         rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
1762         MUTEX_EXIT(&freeSQEList_lock);
1763     } else {                    /* otherwise allocate a new one and return that */
1764         MUTEX_EXIT(&freeSQEList_lock);
1765         sq = (struct rx_serverQueueEntry *)
1766             rxi_Alloc(sizeof(struct rx_serverQueueEntry));
1767         MUTEX_INIT(&sq->lock, "server Queue lock", MUTEX_DEFAULT, 0);
1768         CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
1769     }
1770     MUTEX_ENTER(&sq->lock);
1771
1772     if (cur_service != NULL) {
1773         cur_service->nRequestsRunning--;
1774         if (cur_service->nRequestsRunning < cur_service->minProcs)
1775             rxi_minDeficit++;
1776         rxi_availProcs++;
1777     }
1778     if (queue_IsNotEmpty(&rx_incomingCallQueue)) {
1779         struct rx_call *tcall, *ncall;
1780         /* Scan for eligible incoming calls.  A call is not eligible
1781          * if the maximum number of calls for its service type are
1782          * already executing */
1783         /* One thread will process calls FCFS (to prevent starvation),
1784          * while the other threads may run ahead looking for calls which
1785          * have all their input data available immediately.  This helps 
1786          * keep threads from blocking, waiting for data from the client. */
1787         choice2 = (struct rx_call *)0;
1788         for (queue_Scan(&rx_incomingCallQueue, tcall, ncall, rx_call)) {
1789             service = tcall->conn->service;
1790             if (QuotaOK(service)) {
1791                 MUTEX_ENTER(&rx_pthread_mutex);
1792                 if (tno == rxi_fcfs_thread_num
1793                     || !tcall->queue_item_header.next) {
1794                     MUTEX_EXIT(&rx_pthread_mutex);
1795                     /* If we're the fcfs thread, then  we'll just use 
1796                      * this call. If we haven't been able to find an optimal 
1797                      * choice, and we're at the end of the list, then use a 
1798                      * 2d choice if one has been identified.  Otherwise... */
1799                     call = (choice2 ? choice2 : tcall);
1800                     service = call->conn->service;
1801                 } else {
1802                     MUTEX_EXIT(&rx_pthread_mutex);
1803                     if (!queue_IsEmpty(&tcall->rq)) {
1804                         struct rx_packet *rp;
1805                         rp = queue_First(&tcall->rq, rx_packet);
1806                         if (rp->header.seq == 1
1807                             && (!meltdown_1pkt
1808                                 || (rp->header.flags & RX_LAST_PACKET))) {
1809                             call = tcall;
1810                         } else if (rxi_2dchoice && !choice2
1811                                    && !(tcall->flags & RX_CALL_CLEARED)
1812                                    && (tcall->rprev > rxi_HardAckRate)) {
1813                             choice2 = tcall;
1814                         } else
1815                             rxi_md2cnt++;
1816                     }
1817                 }
1818             }
1819             if (call)
1820                 break;
1821         }
1822     }
1823
1824     if (call) {
1825         queue_Remove(call);
1826         /* we can't schedule a call if there's no data!!! */
1827         /* send an ack if there's no data, if we're missing the
1828          * first packet, or we're missing something between first 
1829          * and last -- there's a "hole" in the incoming data. */
1830         if (queue_IsEmpty(&call->rq)
1831             || queue_First(&call->rq, rx_packet)->header.seq != 1
1832             || call->rprev != queue_Last(&call->rq, rx_packet)->header.seq)
1833             rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
1834
1835         call->flags &= (~RX_CALL_WAIT_PROC);
1836         service->nRequestsRunning++;
1837         /* just started call in minProcs pool, need fewer to maintain
1838          * guarantee */
1839         if (service->nRequestsRunning <= service->minProcs)
1840             rxi_minDeficit--;
1841         rxi_availProcs--;
1842         rx_nWaiting--;
1843         /* MUTEX_EXIT(&call->lock); */
1844     } else {
1845         /* If there are no eligible incoming calls, add this process
1846          * to the idle server queue, to wait for one */
1847         sq->newcall = 0;
1848         if (socketp) {
1849             *socketp = OSI_NULLSOCKET;
1850         }
1851         sq->socketp = socketp;
1852         queue_Append(&rx_idleServerQueue, sq);
1853         do {
1854             osi_rxSleep(sq);
1855 #ifdef  KERNEL
1856             if (afs_termState == AFSOP_STOP_RXCALLBACK) {
1857                 USERPRI;
1858                 rxi_Free(sq, sizeof(struct rx_serverQueueEntry));
1859                 return (struct rx_call *)0;
1860             }
1861 #endif
1862         } while (!(call = sq->newcall)
1863                  && !(socketp && *socketp != OSI_NULLSOCKET));
1864     }
1865     MUTEX_EXIT(&sq->lock);
1866
1867     MUTEX_ENTER(&freeSQEList_lock);
1868     *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
1869     rx_FreeSQEList = sq;
1870     MUTEX_EXIT(&freeSQEList_lock);
1871
1872     if (call) {
1873         clock_GetTime(&call->startTime);
1874         call->state = RX_STATE_ACTIVE;
1875         call->mode = RX_MODE_RECEIVING;
1876 #ifdef RX_KERNEL_TRACE
1877         if (ICL_SETACTIVE(afs_iclSetp)) {
1878             int glockOwner = ISAFS_GLOCK();
1879             if (!glockOwner)
1880                 AFS_GLOCK();
1881             afs_Trace3(afs_iclSetp, CM_TRACE_WASHERE, ICL_TYPE_STRING,
1882                        __FILE__, ICL_TYPE_INT32, __LINE__, ICL_TYPE_POINTER,
1883                        call);
1884             if (!glockOwner)
1885                 AFS_GUNLOCK();
1886         }
1887 #endif
1888
1889         rxi_calltrace(RX_CALL_START, call);
1890         dpf(("rx_GetCall(port=%d, service=%d) ==> call %x\n",
1891              call->conn->service->servicePort, call->conn->service->serviceId,
1892              call));
1893     } else {
1894         dpf(("rx_GetCall(socketp=0x%x, *socketp=0x%x)\n", socketp, *socketp));
1895     }
1896
1897     USERPRI;
1898
1899     return call;
1900 }
1901 #endif /* RX_ENABLE_LOCKS */
1902
1903
1904
1905 /* Establish a procedure to be called when a packet arrives for a
1906  * call.  This routine will be called at most once after each call,
1907  * and will also be called if there is an error condition on the or
1908  * the call is complete.  Used by multi rx to build a selection
1909  * function which determines which of several calls is likely to be a
1910  * good one to read from.  
1911  * NOTE: the way this is currently implemented it is probably only a
1912  * good idea to (1) use it immediately after a newcall (clients only)
1913  * and (2) only use it once.  Other uses currently void your warranty
1914  */
1915 void
1916 rx_SetArrivalProc(struct rx_call *call,
1917                   void (*proc) (struct rx_call * call,
1918                                         void * mh,
1919                                         int index),
1920                   void * handle, int arg)
1921 {
1922     call->arrivalProc = proc;
1923     call->arrivalProcHandle = handle;
1924     call->arrivalProcArg = arg;
1925 }
1926
1927 /* Call is finished (possibly prematurely).  Return rc to the peer, if
1928  * appropriate, and return the final error code from the conversation
1929  * to the caller */
1930
1931 afs_int32
1932 rx_EndCall(struct rx_call *call, afs_int32 rc)
1933 {
1934     struct rx_connection *conn = call->conn;
1935     struct rx_service *service;
1936     afs_int32 error;
1937     SPLVAR;
1938
1939
1940
1941     dpf(("rx_EndCall(call %x rc %d error %d abortCode %d)\n", call, rc, call->error, call->abortCode));
1942
1943     NETPRI;
1944     MUTEX_ENTER(&call->lock);
1945
1946     if (rc == 0 && call->error == 0) {
1947         call->abortCode = 0;
1948         call->abortCount = 0;
1949     }
1950
1951     call->arrivalProc = (void (*)())0;
1952     if (rc && call->error == 0) {
1953         rxi_CallError(call, rc);
1954         /* Send an abort message to the peer if this error code has
1955          * only just been set.  If it was set previously, assume the
1956          * peer has already been sent the error code or will request it 
1957          */
1958         rxi_SendCallAbort(call, (struct rx_packet *)0, 0, 0);
1959     }
1960     if (conn->type == RX_SERVER_CONNECTION) {
1961         /* Make sure reply or at least dummy reply is sent */
1962         if (call->mode == RX_MODE_RECEIVING) {
1963             rxi_WriteProc(call, 0, 0);
1964         }
1965         if (call->mode == RX_MODE_SENDING) {
1966             rxi_FlushWrite(call);
1967         }
1968         service = conn->service;
1969         rxi_calltrace(RX_CALL_END, call);
1970         /* Call goes to hold state until reply packets are acknowledged */
1971         if (call->tfirst + call->nSoftAcked < call->tnext) {
1972             call->state = RX_STATE_HOLD;
1973         } else {
1974             call->state = RX_STATE_DALLY;
1975             rxi_ClearTransmitQueue(call, 0);
1976             rxevent_Cancel(call->resendEvent, call, RX_CALL_REFCOUNT_RESEND);
1977             rxevent_Cancel(call->keepAliveEvent, call,
1978                            RX_CALL_REFCOUNT_ALIVE);
1979         }
1980     } else {                    /* Client connection */
1981         char dummy;
1982         /* Make sure server receives input packets, in the case where
1983          * no reply arguments are expected */
1984         if ((call->mode == RX_MODE_SENDING)
1985             || (call->mode == RX_MODE_RECEIVING && call->rnext == 1)) {
1986             (void)rxi_ReadProc(call, &dummy, 1);
1987         }
1988
1989         /* If we had an outstanding delayed ack, be nice to the server
1990          * and force-send it now.
1991          */
1992         if (call->delayedAckEvent) {
1993             rxevent_Cancel(call->delayedAckEvent, call,
1994                            RX_CALL_REFCOUNT_DELAY);
1995             call->delayedAckEvent = NULL;
1996             rxi_SendDelayedAck(NULL, call, NULL);
1997         }
1998
1999         /* We need to release the call lock since it's lower than the
2000          * conn_call_lock and we don't want to hold the conn_call_lock
2001          * over the rx_ReadProc call. The conn_call_lock needs to be held
2002          * here for the case where rx_NewCall is perusing the calls on
2003          * the connection structure. We don't want to signal until
2004          * rx_NewCall is in a stable state. Otherwise, rx_NewCall may
2005          * have checked this call, found it active and by the time it
2006          * goes to sleep, will have missed the signal.
2007          *
2008          * Do not clear the RX_CONN_MAKECALL_WAITING flag as long as
2009          * there are threads waiting to use the conn object.
2010          */
2011         MUTEX_EXIT(&call->lock);
2012         MUTEX_ENTER(&conn->conn_call_lock);
2013         MUTEX_ENTER(&call->lock);
2014         MUTEX_ENTER(&conn->conn_data_lock);
2015         conn->flags |= RX_CONN_BUSY;
2016         if (conn->flags & RX_CONN_MAKECALL_WAITING) {
2017             if (conn->makeCallWaiters == 0)
2018                 conn->flags &= (~RX_CONN_MAKECALL_WAITING);
2019             MUTEX_EXIT(&conn->conn_data_lock);
2020 #ifdef  RX_ENABLE_LOCKS
2021             CV_BROADCAST(&conn->conn_call_cv);
2022 #else
2023             osi_rxWakeup(conn);
2024 #endif
2025         }
2026 #ifdef RX_ENABLE_LOCKS
2027         else {
2028             MUTEX_EXIT(&conn->conn_data_lock);
2029         }
2030 #endif /* RX_ENABLE_LOCKS */
2031         call->state = RX_STATE_DALLY;
2032     }
2033     error = call->error;
2034
2035     /* currentPacket, nLeft, and NFree must be zeroed here, because
2036      * ResetCall cannot: ResetCall may be called at splnet(), in the
2037      * kernel version, and may interrupt the macros rx_Read or
2038      * rx_Write, which run at normal priority for efficiency. */
2039     if (call->currentPacket) {
2040         call->currentPacket->flags &= ~RX_PKTFLAG_CP;
2041         rxi_FreePacket(call->currentPacket);
2042         call->currentPacket = (struct rx_packet *)0;
2043     }
2044         
2045     call->nLeft = call->nFree = call->curlen = 0;
2046
2047     /* Free any packets from the last call to ReadvProc/WritevProc */
2048 #ifdef RXDEBUG_PACKET
2049     call->iovqc -=
2050 #endif /* RXDEBUG_PACKET */
2051         rxi_FreePackets(0, &call->iovq);
2052
2053     CALL_RELE(call, RX_CALL_REFCOUNT_BEGIN);
2054     MUTEX_EXIT(&call->lock);
2055     if (conn->type == RX_CLIENT_CONNECTION) {
2056         MUTEX_EXIT(&conn->conn_call_lock);
2057         conn->flags &= ~RX_CONN_BUSY;
2058     }
2059     USERPRI;
2060     /*
2061      * Map errors to the local host's errno.h format.
2062      */
2063     error = ntoh_syserr_conv(error);
2064     return error;
2065 }
2066
2067 #if !defined(KERNEL)
2068
2069 /* Call this routine when shutting down a server or client (especially
2070  * clients).  This will allow Rx to gracefully garbage collect server
2071  * connections, and reduce the number of retries that a server might
2072  * make to a dead client.
2073  * This is not quite right, since some calls may still be ongoing and
2074  * we can't lock them to destroy them. */
2075 void
2076 rx_Finalize(void)
2077 {
2078     struct rx_connection **conn_ptr, **conn_end;
2079
2080     INIT_PTHREAD_LOCKS;
2081     LOCK_RX_INIT;
2082     if (rxinit_status == 1) {
2083         UNLOCK_RX_INIT;
2084         return;                 /* Already shutdown. */
2085     }
2086     rxi_DeleteCachedConnections();
2087     if (rx_connHashTable) {
2088         MUTEX_ENTER(&rx_connHashTable_lock);
2089         for (conn_ptr = &rx_connHashTable[0], conn_end =
2090              &rx_connHashTable[rx_hashTableSize]; conn_ptr < conn_end;
2091              conn_ptr++) {
2092             struct rx_connection *conn, *next;
2093             for (conn = *conn_ptr; conn; conn = next) {
2094                 next = conn->next;
2095                 if (conn->type == RX_CLIENT_CONNECTION) {
2096                     /* MUTEX_ENTER(&conn->conn_data_lock); when used in kernel */
2097                     conn->refCount++;
2098                     /* MUTEX_EXIT(&conn->conn_data_lock); when used in kernel */
2099 #ifdef RX_ENABLE_LOCKS
2100                     rxi_DestroyConnectionNoLock(conn);
2101 #else /* RX_ENABLE_LOCKS */
2102                     rxi_DestroyConnection(conn);
2103 #endif /* RX_ENABLE_LOCKS */
2104                 }
2105             }
2106         }
2107 #ifdef RX_ENABLE_LOCKS
2108         while (rx_connCleanup_list) {
2109             struct rx_connection *conn;
2110             conn = rx_connCleanup_list;
2111             rx_connCleanup_list = rx_connCleanup_list->next;
2112             MUTEX_EXIT(&rx_connHashTable_lock);
2113             rxi_CleanupConnection(conn);
2114             MUTEX_ENTER(&rx_connHashTable_lock);
2115         }
2116         MUTEX_EXIT(&rx_connHashTable_lock);
2117 #endif /* RX_ENABLE_LOCKS */
2118     }
2119     rxi_flushtrace();
2120
2121 #ifdef AFS_NT40_ENV
2122     afs_winsockCleanup();
2123 #endif
2124
2125     rxinit_status = 1;
2126     UNLOCK_RX_INIT;
2127 }
2128 #endif
2129
2130 /* if we wakeup packet waiter too often, can get in loop with two
2131     AllocSendPackets each waking each other up (from ReclaimPacket calls) */
2132 void
2133 rxi_PacketsUnWait(void)
2134 {
2135     if (!rx_waitingForPackets) {
2136         return;
2137     }
2138 #ifdef KERNEL
2139     if (rxi_OverQuota(RX_PACKET_CLASS_SEND)) {
2140         return;                 /* still over quota */
2141     }
2142 #endif /* KERNEL */
2143     rx_waitingForPackets = 0;
2144 #ifdef  RX_ENABLE_LOCKS
2145     CV_BROADCAST(&rx_waitingForPackets_cv);
2146 #else
2147     osi_rxWakeup(&rx_waitingForPackets);
2148 #endif
2149     return;
2150 }
2151
2152
2153 /* ------------------Internal interfaces------------------------- */
2154
2155 /* Return this process's service structure for the
2156  * specified socket and service */
2157 struct rx_service *
2158 rxi_FindService(osi_socket socket, u_short serviceId)
2159 {
2160     struct rx_service **sp;
2161     for (sp = &rx_services[0]; *sp; sp++) {
2162         if ((*sp)->serviceId == serviceId && (*sp)->socket == socket)
2163             return *sp;
2164     }
2165     return 0;
2166 }
2167
2168 #ifdef DEBUG
2169 #ifdef KDUMP_RX_LOCK
2170 static struct rx_call_rx_lock *rx_allCallsp = 0;
2171 #else
2172 static struct rx_call *rx_allCallsp = 0;
2173 #endif
2174 #endif /* DEBUG */
2175
2176 /* Allocate a call structure, for the indicated channel of the
2177  * supplied connection.  The mode and state of the call must be set by
2178  * the caller. Returns the call with mutex locked. */
2179 struct rx_call *
2180 rxi_NewCall(struct rx_connection *conn, int channel)
2181 {
2182     struct rx_call *call;
2183 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
2184     struct rx_call *cp; /* Call pointer temp */
2185     struct rx_call *nxp;        /* Next call pointer, for queue_Scan */
2186 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2187
2188     dpf(("rxi_NewCall(conn %x, channel %d)\n", conn, channel));
2189
2190     /* Grab an existing call structure, or allocate a new one.
2191      * Existing call structures are assumed to have been left reset by
2192      * rxi_FreeCall */
2193     MUTEX_ENTER(&rx_freeCallQueue_lock);
2194
2195 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
2196     /*
2197      * EXCEPT that the TQ might not yet be cleared out.
2198      * Skip over those with in-use TQs.
2199      */
2200     call = NULL;
2201     for (queue_Scan(&rx_freeCallQueue, cp, nxp, rx_call)) {
2202         if (!(cp->flags & RX_CALL_TQ_BUSY)) {
2203             call = cp;
2204             break;
2205         }
2206     }
2207     if (call) {
2208 #else /* AFS_GLOBAL_RXLOCK_KERNEL */
2209     if (queue_IsNotEmpty(&rx_freeCallQueue)) {
2210         call = queue_First(&rx_freeCallQueue, rx_call);
2211 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2212         queue_Remove(call);
2213         if (rx_stats_active)
2214             rx_MutexDecrement(rx_stats.nFreeCallStructs, rx_stats_mutex);
2215         MUTEX_EXIT(&rx_freeCallQueue_lock);
2216         MUTEX_ENTER(&call->lock);
2217         CLEAR_CALL_QUEUE_LOCK(call);
2218 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
2219         /* Now, if TQ wasn't cleared earlier, do it now. */
2220         if (call->flags & RX_CALL_TQ_CLEARME) {
2221             rxi_ClearTransmitQueue(call, 1);
2222             /*queue_Init(&call->tq);*/
2223         }
2224 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2225         /* Bind the call to its connection structure */
2226         call->conn = conn;
2227         rxi_ResetCall(call, 1);
2228     } else {
2229
2230         call = (struct rx_call *)rxi_Alloc(sizeof(struct rx_call));
2231 #ifdef RXDEBUG_PACKET
2232         call->allNextp = rx_allCallsp;
2233         rx_allCallsp = call;
2234         call->call_id = 
2235 #endif /* RXDEBUG_PACKET */
2236             rx_MutexIncrement(rx_stats.nCallStructs, rx_stats_mutex);
2237         
2238         MUTEX_EXIT(&rx_freeCallQueue_lock);
2239         MUTEX_INIT(&call->lock, "call lock", MUTEX_DEFAULT, NULL);
2240         MUTEX_ENTER(&call->lock);
2241         CV_INIT(&call->cv_twind, "call twind", CV_DEFAULT, 0);
2242         CV_INIT(&call->cv_rq, "call rq", CV_DEFAULT, 0);
2243         CV_INIT(&call->cv_tq, "call tq", CV_DEFAULT, 0);
2244
2245         /* Initialize once-only items */
2246         queue_Init(&call->tq);
2247         queue_Init(&call->rq);
2248         queue_Init(&call->iovq);
2249 #ifdef RXDEBUG_PACKET
2250         call->rqc = call->tqc = call->iovqc = 0;
2251 #endif /* RXDEBUG_PACKET */
2252         /* Bind the call to its connection structure (prereq for reset) */
2253         call->conn = conn;
2254         rxi_ResetCall(call, 1);
2255     }
2256     call->channel = channel;
2257     call->callNumber = &conn->callNumber[channel];
2258     call->rwind = conn->rwind[channel];
2259     call->twind = conn->twind[channel];
2260     /* Note that the next expected call number is retained (in
2261      * conn->callNumber[i]), even if we reallocate the call structure
2262      */
2263     conn->call[channel] = call;
2264     /* if the channel's never been used (== 0), we should start at 1, otherwise
2265      * the call number is valid from the last time this channel was used */
2266     if (*call->callNumber == 0)
2267         *call->callNumber = 1;
2268
2269     return call;
2270 }
2271
2272 /* A call has been inactive long enough that so we can throw away
2273  * state, including the call structure, which is placed on the call
2274  * free list.
2275  * Call is locked upon entry.
2276  * haveCTLock set if called from rxi_ReapConnections
2277  */
2278 #ifdef RX_ENABLE_LOCKS
2279 void
2280 rxi_FreeCall(struct rx_call *call, int haveCTLock)
2281 #else /* RX_ENABLE_LOCKS */
2282 void
2283 rxi_FreeCall(struct rx_call *call)
2284 #endif                          /* RX_ENABLE_LOCKS */
2285 {
2286     int channel = call->channel;
2287     struct rx_connection *conn = call->conn;
2288
2289
2290     if (call->state == RX_STATE_DALLY || call->state == RX_STATE_HOLD)
2291         (*call->callNumber)++;
2292     rxi_ResetCall(call, 0);
2293     call->conn->call[channel] = (struct rx_call *)0;
2294
2295     MUTEX_ENTER(&rx_freeCallQueue_lock);
2296     SET_CALL_QUEUE_LOCK(call, &rx_freeCallQueue_lock);
2297 #ifdef AFS_GLOBAL_RXLOCK_KERNEL
2298     /* A call may be free even though its transmit queue is still in use.
2299      * Since we search the call list from head to tail, put busy calls at
2300      * the head of the list, and idle calls at the tail.
2301      */
2302     if (call->flags & RX_CALL_TQ_BUSY)
2303         queue_Prepend(&rx_freeCallQueue, call);
2304     else
2305         queue_Append(&rx_freeCallQueue, call);
2306 #else /* AFS_GLOBAL_RXLOCK_KERNEL */
2307     queue_Append(&rx_freeCallQueue, call);
2308 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2309     if (rx_stats_active)
2310         rx_MutexIncrement(rx_stats.nFreeCallStructs, rx_stats_mutex);
2311     MUTEX_EXIT(&rx_freeCallQueue_lock);
2312
2313     /* Destroy the connection if it was previously slated for
2314      * destruction, i.e. the Rx client code previously called
2315      * rx_DestroyConnection (client connections), or
2316      * rxi_ReapConnections called the same routine (server
2317      * connections).  Only do this, however, if there are no
2318      * outstanding calls. Note that for fine grain locking, there appears
2319      * to be a deadlock in that rxi_FreeCall has a call locked and
2320      * DestroyConnectionNoLock locks each call in the conn. But note a
2321      * few lines up where we have removed this call from the conn.
2322      * If someone else destroys a connection, they either have no
2323      * call lock held or are going through this section of code.
2324      */
2325     if (conn->flags & RX_CONN_DESTROY_ME && !(conn->flags & RX_CONN_MAKECALL_WAITING)) {
2326         MUTEX_ENTER(&conn->conn_data_lock);
2327         conn->refCount++;
2328         MUTEX_EXIT(&conn->conn_data_lock);
2329 #ifdef RX_ENABLE_LOCKS
2330         if (haveCTLock)
2331             rxi_DestroyConnectionNoLock(conn);
2332         else
2333             rxi_DestroyConnection(conn);
2334 #else /* RX_ENABLE_LOCKS */
2335         rxi_DestroyConnection(conn);
2336 #endif /* RX_ENABLE_LOCKS */
2337     }
2338 }
2339
2340 afs_int32 rxi_Alloccnt = 0, rxi_Allocsize = 0;
2341 char *
2342 rxi_Alloc(size_t size)
2343 {
2344     char *p;
2345
2346     if (rx_stats_active)
2347         rx_MutexAdd1Increment2(rxi_Allocsize, (afs_int32)size, rxi_Alloccnt, rx_stats_mutex);
2348
2349 p = (char *)
2350 #if defined(KERNEL) && !defined(UKERNEL) && defined(AFS_FBSD80_ENV)
2351   afs_osi_Alloc_NoSleep(size);
2352 #else
2353   osi_Alloc(size);
2354 #endif
2355     if (!p)
2356         osi_Panic("rxi_Alloc error");
2357     memset(p, 0, size);
2358     return p;
2359 }
2360
2361 void
2362 rxi_Free(void *addr, size_t size)
2363 {
2364     if (rx_stats_active)
2365         rx_MutexAdd1Decrement2(rxi_Allocsize, -(afs_int32)size, rxi_Alloccnt, rx_stats_mutex);
2366     osi_Free(addr, size);
2367 }
2368
2369 void 
2370 rxi_SetPeerMtu(afs_uint32 host, afs_uint32 port, int mtu)
2371 {
2372     struct rx_peer **peer_ptr, **peer_end;
2373     int hashIndex;
2374
2375     MUTEX_ENTER(&rx_peerHashTable_lock);
2376     if (port == 0) {
2377        for (peer_ptr = &rx_peerHashTable[0], peer_end =
2378                 &rx_peerHashTable[rx_hashTableSize]; peer_ptr < peer_end;
2379             peer_ptr++) {
2380            struct rx_peer *peer, *next;
2381            for (peer = *peer_ptr; peer; peer = next) {
2382                next = peer->next;
2383                if (host == peer->host) {
2384                    MUTEX_ENTER(&peer->peer_lock);
2385                    peer->ifMTU=MIN(mtu, peer->ifMTU);
2386                    peer->natMTU = rxi_AdjustIfMTU(peer->ifMTU);
2387                    MUTEX_EXIT(&peer->peer_lock);
2388                }
2389            }
2390        }
2391     } else {
2392        struct rx_peer *peer;
2393        hashIndex = PEER_HASH(host, port);
2394        for (peer = rx_peerHashTable[hashIndex]; peer; peer = peer->next) {
2395            if ((peer->host == host) && (peer->port == port)) {
2396                MUTEX_ENTER(&peer->peer_lock);
2397                peer->ifMTU=MIN(mtu, peer->ifMTU);
2398                peer->natMTU = rxi_AdjustIfMTU(peer->ifMTU);
2399                MUTEX_EXIT(&peer->peer_lock);
2400            }
2401        }
2402     }
2403     MUTEX_EXIT(&rx_peerHashTable_lock);
2404 }
2405
2406 /* Find the peer process represented by the supplied (host,port)
2407  * combination.  If there is no appropriate active peer structure, a
2408  * new one will be allocated and initialized 
2409  * The origPeer, if set, is a pointer to a peer structure on which the
2410  * refcount will be be decremented. This is used to replace the peer
2411  * structure hanging off a connection structure */
2412 struct rx_peer *
2413 rxi_FindPeer(afs_uint32 host, u_short port,
2414              struct rx_peer *origPeer, int create)
2415 {
2416     struct rx_peer *pp;
2417     int hashIndex;
2418     hashIndex = PEER_HASH(host, port);
2419     MUTEX_ENTER(&rx_peerHashTable_lock);
2420     for (pp = rx_peerHashTable[hashIndex]; pp; pp = pp->next) {
2421         if ((pp->host == host) && (pp->port == port))
2422             break;
2423     }
2424     if (!pp) {
2425         if (create) {
2426             pp = rxi_AllocPeer();       /* This bzero's *pp */
2427             pp->host = host;    /* set here or in InitPeerParams is zero */
2428             pp->port = port;
2429             MUTEX_INIT(&pp->peer_lock, "peer_lock", MUTEX_DEFAULT, 0);
2430             queue_Init(&pp->congestionQueue);
2431             queue_Init(&pp->rpcStats);
2432             pp->next = rx_peerHashTable[hashIndex];
2433             rx_peerHashTable[hashIndex] = pp;
2434             rxi_InitPeerParams(pp);
2435             if (rx_stats_active)
2436                 rx_MutexIncrement(rx_stats.nPeerStructs, rx_stats_mutex);
2437         }
2438     }
2439     if (pp && create) {
2440         pp->refCount++;
2441     }
2442     if (origPeer)
2443         origPeer->refCount--;
2444     MUTEX_EXIT(&rx_peerHashTable_lock);
2445     return pp;
2446 }
2447
2448
2449 /* Find the connection at (host, port) started at epoch, and with the
2450  * given connection id.  Creates the server connection if necessary.
2451  * The type specifies whether a client connection or a server
2452  * connection is desired.  In both cases, (host, port) specify the
2453  * peer's (host, pair) pair.  Client connections are not made
2454  * automatically by this routine.  The parameter socket gives the
2455  * socket descriptor on which the packet was received.  This is used,
2456  * in the case of server connections, to check that *new* connections
2457  * come via a valid (port, serviceId).  Finally, the securityIndex
2458  * parameter must match the existing index for the connection.  If a
2459  * server connection is created, it will be created using the supplied
2460  * index, if the index is valid for this service */
2461 struct rx_connection *
2462 rxi_FindConnection(osi_socket socket, afs_int32 host,
2463                    u_short port, u_short serviceId, afs_uint32 cid,
2464                    afs_uint32 epoch, int type, u_int securityIndex)
2465 {
2466     int hashindex, flag, i;
2467     struct rx_connection *conn;
2468     hashindex = CONN_HASH(host, port, cid, epoch, type);
2469     MUTEX_ENTER(&rx_connHashTable_lock);
2470     rxLastConn ? (conn = rxLastConn, flag = 0) : (conn =
2471                                                   rx_connHashTable[hashindex],
2472                                                   flag = 1);
2473     for (; conn;) {
2474         if ((conn->type == type) && ((cid & RX_CIDMASK) == conn->cid)
2475             && (epoch == conn->epoch)) {
2476             struct rx_peer *pp = conn->peer;
2477             if (securityIndex != conn->securityIndex) {
2478                 /* this isn't supposed to happen, but someone could forge a packet
2479                  * like this, and there seems to be some CM bug that makes this
2480                  * happen from time to time -- in which case, the fileserver
2481                  * asserts. */
2482                 MUTEX_EXIT(&rx_connHashTable_lock);
2483                 return (struct rx_connection *)0;
2484             }
2485             if (pp->host == host && pp->port == port)
2486                 break;
2487             if (type == RX_CLIENT_CONNECTION && pp->port == port)
2488                 break;
2489             /* So what happens when it's a callback connection? */
2490             if (                /*type == RX_CLIENT_CONNECTION && */
2491                    (conn->epoch & 0x80000000))
2492                 break;
2493         }
2494         if (!flag) {
2495             /* the connection rxLastConn that was used the last time is not the
2496              ** one we are looking for now. Hence, start searching in the hash */
2497             flag = 1;
2498             conn = rx_connHashTable[hashindex];
2499         } else
2500             conn = conn->next;
2501     }
2502     if (!conn) {
2503         struct rx_service *service;
2504         if (type == RX_CLIENT_CONNECTION) {
2505             MUTEX_EXIT(&rx_connHashTable_lock);
2506             return (struct rx_connection *)0;
2507         }
2508         service = rxi_FindService(socket, serviceId);
2509         if (!service || (securityIndex >= service->nSecurityObjects)
2510             || (service->securityObjects[securityIndex] == 0)) {
2511             MUTEX_EXIT(&rx_connHashTable_lock);
2512             return (struct rx_connection *)0;
2513         }
2514         conn = rxi_AllocConnection();   /* This bzero's the connection */
2515         MUTEX_INIT(&conn->conn_call_lock, "conn call lock", MUTEX_DEFAULT, 0);
2516         MUTEX_INIT(&conn->conn_data_lock, "conn data lock", MUTEX_DEFAULT, 0);
2517         CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
2518         conn->next = rx_connHashTable[hashindex];
2519         rx_connHashTable[hashindex] = conn;
2520         conn->peer = rxi_FindPeer(host, port, 0, 1);
2521         conn->type = RX_SERVER_CONNECTION;
2522         conn->lastSendTime = clock_Sec();       /* don't GC immediately */
2523         conn->epoch = epoch;
2524         conn->cid = cid & RX_CIDMASK;
2525         /* conn->serial = conn->lastSerial = 0; */
2526         /* conn->timeout = 0; */
2527         conn->ackRate = RX_FAST_ACK_RATE;
2528         conn->service = service;
2529         conn->serviceId = serviceId;
2530         conn->securityIndex = securityIndex;
2531         conn->securityObject = service->securityObjects[securityIndex];
2532         conn->nSpecific = 0;
2533         conn->specific = NULL;
2534         rx_SetConnDeadTime(conn, service->connDeadTime);
2535         rx_SetConnIdleDeadTime(conn, service->idleDeadTime);
2536         rx_SetServerConnIdleDeadErr(conn, service->idleDeadErr);
2537         for (i = 0; i < RX_MAXCALLS; i++) {
2538             conn->twind[i] = rx_initSendWindow;
2539             conn->rwind[i] = rx_initReceiveWindow;
2540         }
2541         /* Notify security object of the new connection */
2542         RXS_NewConnection(conn->securityObject, conn);
2543         /* XXXX Connection timeout? */
2544         if (service->newConnProc)
2545             (*service->newConnProc) (conn);
2546         if (rx_stats_active)
2547             rx_MutexIncrement(rx_stats.nServerConns, rx_stats_mutex);
2548     }
2549
2550     MUTEX_ENTER(&conn->conn_data_lock);
2551     conn->refCount++;
2552     MUTEX_EXIT(&conn->conn_data_lock);
2553
2554     rxLastConn = conn;          /* store this connection as the last conn used */
2555     MUTEX_EXIT(&rx_connHashTable_lock);
2556     return conn;
2557 }
2558
2559 /* There are two packet tracing routines available for testing and monitoring
2560  * Rx.  One is called just after every packet is received and the other is
2561  * called just before every packet is sent.  Received packets, have had their
2562  * headers decoded, and packets to be sent have not yet had their headers
2563  * encoded.  Both take two parameters: a pointer to the packet and a sockaddr
2564  * containing the network address.  Both can be modified.  The return value, if
2565  * non-zero, indicates that the packet should be dropped.  */
2566
2567 int (*rx_justReceived) (struct rx_packet *, struct sockaddr_in *) = 0;
2568 int (*rx_almostSent) (struct rx_packet *, struct sockaddr_in *) = 0;
2569
2570 /* A packet has been received off the interface.  Np is the packet, socket is
2571  * the socket number it was received from (useful in determining which service
2572  * this packet corresponds to), and (host, port) reflect the host,port of the
2573  * sender.  This call returns the packet to the caller if it is finished with
2574  * it, rather than de-allocating it, just as a small performance hack */
2575
2576 struct rx_packet *
2577 rxi_ReceivePacket(struct rx_packet *np, osi_socket socket,
2578                   afs_uint32 host, u_short port, int *tnop,
2579                   struct rx_call **newcallp)
2580 {
2581     struct rx_call *call;
2582     struct rx_connection *conn;
2583     int channel;
2584     afs_uint32 currentCallNumber;
2585     int type;
2586     int skew;
2587 #ifdef RXDEBUG
2588     char *packetType;
2589 #endif
2590     struct rx_packet *tnp;
2591
2592 #ifdef RXDEBUG
2593 /* We don't print out the packet until now because (1) the time may not be
2594  * accurate enough until now in the lwp implementation (rx_Listener only gets
2595  * the time after the packet is read) and (2) from a protocol point of view,
2596  * this is the first time the packet has been seen */
2597     packetType = (np->header.type > 0 && np->header.type < RX_N_PACKET_TYPES)
2598         ? rx_packetTypes[np->header.type - 1] : "*UNKNOWN*";
2599     dpf(("R %d %s: %x.%d.%d.%d.%d.%d.%d flags %d, packet %x",
2600          np->header.serial, packetType, ntohl(host), ntohs(port), np->header.serviceId,
2601          np->header.epoch, np->header.cid, np->header.callNumber,
2602          np->header.seq, np->header.flags, np));
2603 #endif
2604
2605     if (np->header.type == RX_PACKET_TYPE_VERSION) {
2606         return rxi_ReceiveVersionPacket(np, socket, host, port, 1);
2607     }
2608
2609     if (np->header.type == RX_PACKET_TYPE_DEBUG) {
2610         return rxi_ReceiveDebugPacket(np, socket, host, port, 1);
2611     }
2612 #ifdef RXDEBUG
2613     /* If an input tracer function is defined, call it with the packet and
2614      * network address.  Note this function may modify its arguments. */
2615     if (rx_justReceived) {
2616         struct sockaddr_in addr;
2617         int drop;
2618         addr.sin_family = AF_INET;
2619         addr.sin_port = port;
2620         addr.sin_addr.s_addr = host;
2621 #ifdef STRUCT_SOCKADDR_HAS_SA_LEN
2622         addr.sin_len = sizeof(addr);
2623 #endif /* AFS_OSF_ENV */
2624         drop = (*rx_justReceived) (np, &addr);
2625         /* drop packet if return value is non-zero */
2626         if (drop)
2627             return np;
2628         port = addr.sin_port;   /* in case fcn changed addr */
2629         host = addr.sin_addr.s_addr;
2630     }
2631 #endif
2632
2633     /* If packet was not sent by the client, then *we* must be the client */
2634     type = ((np->header.flags & RX_CLIENT_INITIATED) != RX_CLIENT_INITIATED)
2635         ? RX_CLIENT_CONNECTION : RX_SERVER_CONNECTION;
2636
2637     /* Find the connection (or fabricate one, if we're the server & if
2638      * necessary) associated with this packet */
2639     conn =
2640         rxi_FindConnection(socket, host, port, np->header.serviceId,
2641                            np->header.cid, np->header.epoch, type,
2642                            np->header.securityIndex);
2643
2644     if (!conn) {
2645         /* If no connection found or fabricated, just ignore the packet.
2646          * (An argument could be made for sending an abort packet for
2647          * the conn) */
2648         return np;
2649     }
2650
2651     MUTEX_ENTER(&conn->conn_data_lock);
2652     if (conn->maxSerial < np->header.serial)
2653         conn->maxSerial = np->header.serial;
2654     MUTEX_EXIT(&conn->conn_data_lock);
2655
2656     /* If the connection is in an error state, send an abort packet and ignore
2657      * the incoming packet */
2658     if (conn->error) {
2659         /* Don't respond to an abort packet--we don't want loops! */
2660         MUTEX_ENTER(&conn->conn_data_lock);
2661         if (np->header.type != RX_PACKET_TYPE_ABORT)
2662             np = rxi_SendConnectionAbort(conn, np, 1, 0);
2663         conn->refCount--;
2664         MUTEX_EXIT(&conn->conn_data_lock);
2665         return np;
2666     }
2667
2668     /* Check for connection-only requests (i.e. not call specific). */
2669     if (np->header.callNumber == 0) {
2670         switch (np->header.type) {
2671         case RX_PACKET_TYPE_ABORT: {
2672             /* What if the supplied error is zero? */
2673             afs_int32 errcode = ntohl(rx_GetInt32(np, 0));
2674             dpf(("rxi_ReceivePacket ABORT rx_GetInt32 = %d", errcode));
2675             rxi_ConnectionError(conn, errcode);
2676             MUTEX_ENTER(&conn->conn_data_lock);
2677             conn->refCount--;
2678             MUTEX_EXIT(&conn->conn_data_lock);
2679             return np;
2680         }
2681         case RX_PACKET_TYPE_CHALLENGE:
2682             tnp = rxi_ReceiveChallengePacket(conn, np, 1);
2683             MUTEX_ENTER(&conn->conn_data_lock);
2684             conn->refCount--;
2685             MUTEX_EXIT(&conn->conn_data_lock);
2686             return tnp;
2687         case RX_PACKET_TYPE_RESPONSE:
2688             tnp = rxi_ReceiveResponsePacket(conn, np, 1);
2689             MUTEX_ENTER(&conn->conn_data_lock);
2690             conn->refCount--;
2691             MUTEX_EXIT(&conn->conn_data_lock);
2692             return tnp;
2693         case RX_PACKET_TYPE_PARAMS:
2694         case RX_PACKET_TYPE_PARAMS + 1:
2695         case RX_PACKET_TYPE_PARAMS + 2:
2696             /* ignore these packet types for now */
2697             MUTEX_ENTER(&conn->conn_data_lock);
2698             conn->refCount--;
2699             MUTEX_EXIT(&conn->conn_data_lock);
2700             return np;
2701
2702
2703         default:
2704             /* Should not reach here, unless the peer is broken: send an
2705              * abort packet */
2706             rxi_ConnectionError(conn, RX_PROTOCOL_ERROR);
2707             MUTEX_ENTER(&conn->conn_data_lock);
2708             tnp = rxi_SendConnectionAbort(conn, np, 1, 0);
2709             conn->refCount--;
2710             MUTEX_EXIT(&conn->conn_data_lock);
2711             return tnp;
2712         }
2713     }
2714
2715     channel = np->header.cid & RX_CHANNELMASK;
2716     call = conn->call[channel];
2717 #ifdef  RX_ENABLE_LOCKS
2718     if (call)
2719         MUTEX_ENTER(&call->lock);
2720     /* Test to see if call struct is still attached to conn. */
2721     if (call != conn->call[channel]) {
2722         if (call)
2723             MUTEX_EXIT(&call->lock);
2724         if (type == RX_SERVER_CONNECTION) {
2725             call = conn->call[channel];
2726             /* If we started with no call attached and there is one now,
2727              * another thread is also running this routine and has gotten
2728              * the connection channel. We should drop this packet in the tests
2729              * below. If there was a call on this connection and it's now
2730              * gone, then we'll be making a new call below.
2731              * If there was previously a call and it's now different then
2732              * the old call was freed and another thread running this routine
2733              * has created a call on this channel. One of these two threads
2734              * has a packet for the old call and the code below handles those
2735              * cases.
2736              */
2737             if (call)
2738                 MUTEX_ENTER(&call->lock);
2739         } else {
2740             /* This packet can't be for this call. If the new call address is
2741              * 0 then no call is running on this channel. If there is a call
2742              * then, since this is a client connection we're getting data for
2743              * it must be for the previous call.
2744              */
2745             if (rx_stats_active)
2746                 rx_MutexIncrement(rx_stats.spuriousPacketsRead, rx_stats_mutex);
2747             MUTEX_ENTER(&conn->conn_data_lock);
2748             conn->refCount--;
2749             MUTEX_EXIT(&conn->conn_data_lock);
2750             return np;
2751         }
2752     }
2753 #endif
2754     currentCallNumber = conn->callNumber[channel];
2755
2756     if (type == RX_SERVER_CONNECTION) { /* We're the server */
2757         if (np->header.callNumber < currentCallNumber) {
2758             if (rx_stats_active)
2759                 rx_MutexIncrement(rx_stats.spuriousPacketsRead, rx_stats_mutex);
2760 #ifdef  RX_ENABLE_LOCKS
2761             if (call)
2762                 MUTEX_EXIT(&call->lock);
2763 #endif
2764             MUTEX_ENTER(&conn->conn_data_lock);
2765             conn->refCount--;
2766             MUTEX_EXIT(&conn->conn_data_lock);
2767             return np;
2768         }
2769         if (!call) {
2770             MUTEX_ENTER(&conn->conn_call_lock);
2771             call = rxi_NewCall(conn, channel);
2772             MUTEX_EXIT(&conn->conn_call_lock);
2773             *call->callNumber = np->header.callNumber;
2774 #ifdef RXDEBUG
2775             if (np->header.callNumber == 0) 
2776                 dpf(("RecPacket call 0 %d %s: %x.%u.%u.%u.%u.%u.%u flags %d, packet %lx resend %d.%0.3d len %d", np->header.serial, rx_packetTypes[np->header.type - 1], ntohl(conn->peer->host), ntohs(conn->peer->port), np->header.serial, np->header.epoch, np->header.cid, np->header.callNumber, np->header.seq, np->header.flags, (unsigned long)np, np->retryTime.sec, np->retryTime.usec / 1000, np->length));
2777 #endif
2778             call->state = RX_STATE_PRECALL;
2779             clock_GetTime(&call->queueTime);
2780             hzero(call->bytesSent);
2781             hzero(call->bytesRcvd);
2782             /*
2783              * If the number of queued calls exceeds the overload
2784              * threshold then abort this call.
2785              */
2786             if ((rx_BusyThreshold > 0) && (rx_nWaiting > rx_BusyThreshold)) {
2787                 struct rx_packet *tp;
2788                 
2789                 rxi_CallError(call, rx_BusyError);
2790                 tp = rxi_SendCallAbort(call, np, 1, 0);
2791                 MUTEX_EXIT(&call->lock);
2792                 MUTEX_ENTER(&conn->conn_data_lock);
2793                 conn->refCount--;
2794                 MUTEX_EXIT(&conn->conn_data_lock);
2795                 if (rx_stats_active)
2796                     rx_MutexIncrement(rx_stats.nBusies, rx_stats_mutex);
2797                 return tp;
2798             }
2799             rxi_KeepAliveOn(call);
2800         } else if (np->header.callNumber != currentCallNumber) {
2801             /* Wait until the transmit queue is idle before deciding
2802              * whether to reset the current call. Chances are that the
2803              * call will be in ether DALLY or HOLD state once the TQ_BUSY
2804              * flag is cleared.
2805              */
2806 #ifdef AFS_GLOBAL_RXLOCK_KERNEL
2807             while ((call->state == RX_STATE_ACTIVE)
2808                    && (call->flags & RX_CALL_TQ_BUSY)) {
2809                 call->flags |= RX_CALL_TQ_WAIT;
2810                 call->tqWaiters++;
2811 #ifdef RX_ENABLE_LOCKS
2812                 osirx_AssertMine(&call->lock, "rxi_Start lock3");
2813                 CV_WAIT(&call->cv_tq, &call->lock);
2814 #else /* RX_ENABLE_LOCKS */
2815                 osi_rxSleep(&call->tq);
2816 #endif /* RX_ENABLE_LOCKS */
2817                 call->tqWaiters--;
2818                 if (call->tqWaiters == 0)
2819                     call->flags &= ~RX_CALL_TQ_WAIT;
2820             }
2821 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2822             /* If the new call cannot be taken right now send a busy and set
2823              * the error condition in this call, so that it terminates as
2824              * quickly as possible */
2825             if (call->state == RX_STATE_ACTIVE) {
2826                 struct rx_packet *tp;
2827
2828                 rxi_CallError(call, RX_CALL_DEAD);
2829                 tp = rxi_SendSpecial(call, conn, np, RX_PACKET_TYPE_BUSY,
2830                                      NULL, 0, 1);
2831                 MUTEX_EXIT(&call->lock);
2832                 MUTEX_ENTER(&conn->conn_data_lock);
2833                 conn->refCount--;
2834                 MUTEX_EXIT(&conn->conn_data_lock);
2835                 return tp;
2836             }
2837             rxi_ResetCall(call, 0);
2838             *call->callNumber = np->header.callNumber;
2839 #ifdef RXDEBUG
2840             if (np->header.callNumber == 0) 
2841                 dpf(("RecPacket call 0 %d %s: %x.%u.%u.%u.%u.%u.%u flags %d, packet %lx resend %d.%0.3d len %d", np->header.serial, rx_packetTypes[np->header.type - 1], ntohl(conn->peer->host), ntohs(conn->peer->port), np->header.serial, np->header.epoch, np->header.cid, np->header.callNumber, np->header.seq, np->header.flags, (unsigned long)np, np->retryTime.sec, np->retryTime.usec / 1000, np->length));
2842 #endif
2843             call->state = RX_STATE_PRECALL;
2844             clock_GetTime(&call->queueTime);
2845             hzero(call->bytesSent);
2846             hzero(call->bytesRcvd);
2847             /*
2848              * If the number of queued calls exceeds the overload
2849              * threshold then abort this call.
2850              */
2851             if ((rx_BusyThreshold > 0) && (rx_nWaiting > rx_BusyThreshold)) {
2852                 struct rx_packet *tp;
2853
2854                 rxi_CallError(call, rx_BusyError);
2855                 tp = rxi_SendCallAbort(call, np, 1, 0);
2856                 MUTEX_EXIT(&call->lock);
2857                 MUTEX_ENTER(&conn->conn_data_lock);
2858                 conn->refCount--;
2859                 MUTEX_EXIT(&conn->conn_data_lock);
2860                 if (rx_stats_active)
2861                     rx_MutexIncrement(rx_stats.nBusies, rx_stats_mutex);
2862                 return tp;
2863             }
2864             rxi_KeepAliveOn(call);
2865         } else {
2866             /* Continuing call; do nothing here. */
2867         }
2868     } else {                    /* we're the client */
2869         /* Ignore all incoming acknowledgements for calls in DALLY state */
2870         if (call && (call->state == RX_STATE_DALLY)
2871             && (np->header.type == RX_PACKET_TYPE_ACK)) {
2872             if (rx_stats_active)
2873                 rx_MutexIncrement(rx_stats.ignorePacketDally, rx_stats_mutex);
2874 #ifdef  RX_ENABLE_LOCKS
2875             if (call) {
2876                 MUTEX_EXIT(&call->lock);
2877             }
2878 #endif
2879             MUTEX_ENTER(&conn->conn_data_lock);
2880             conn->refCount--;
2881             MUTEX_EXIT(&conn->conn_data_lock);
2882             return np;
2883         }
2884
2885         /* Ignore anything that's not relevant to the current call.  If there
2886          * isn't a current call, then no packet is relevant. */
2887         if (!call || (np->header.callNumber != currentCallNumber)) {
2888             if (rx_stats_active)
2889                 rx_MutexIncrement(rx_stats.spuriousPacketsRead, rx_stats_mutex);
2890 #ifdef  RX_ENABLE_LOCKS
2891             if (call) {
2892                 MUTEX_EXIT(&call->lock);
2893             }
2894 #endif
2895             MUTEX_ENTER(&conn->conn_data_lock);
2896             conn->refCount--;
2897             MUTEX_EXIT(&conn->conn_data_lock);
2898             return np;
2899         }
2900         /* If the service security object index stamped in the packet does not
2901          * match the connection's security index, ignore the packet */
2902         if (np->header.securityIndex != conn->securityIndex) {
2903 #ifdef  RX_ENABLE_LOCKS
2904             MUTEX_EXIT(&call->lock);
2905 #endif
2906             MUTEX_ENTER(&conn->conn_data_lock);
2907             conn->refCount--;
2908             MUTEX_EXIT(&conn->conn_data_lock);
2909             return np;
2910         }
2911
2912         /* If we're receiving the response, then all transmit packets are
2913          * implicitly acknowledged.  Get rid of them. */
2914         if (np->header.type == RX_PACKET_TYPE_DATA) {
2915 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
2916             /* XXX Hack. Because we must release the global rx lock when
2917              * sending packets (osi_NetSend) we drop all acks while we're
2918              * traversing the tq in rxi_Start sending packets out because
2919              * packets may move to the freePacketQueue as result of being here!
2920              * So we drop these packets until we're safely out of the
2921              * traversing. Really ugly! 
2922              * For fine grain RX locking, we set the acked field in the
2923              * packets and let rxi_Start remove them from the transmit queue.
2924              */
2925             if (call->flags & RX_CALL_TQ_BUSY) {
2926 #ifdef  RX_ENABLE_LOCKS
2927                 rxi_SetAcksInTransmitQueue(call);
2928 #else
2929                 conn->refCount--;
2930                 return np;      /* xmitting; drop packet */
2931 #endif
2932             } else {
2933                 rxi_ClearTransmitQueue(call, 0);
2934             }
2935 #else /* AFS_GLOBAL_RXLOCK_KERNEL */
2936             rxi_ClearTransmitQueue(call, 0);
2937 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
2938         } else {
2939             if (np->header.type == RX_PACKET_TYPE_ACK) {
2940                 /* now check to see if this is an ack packet acknowledging that the
2941                  * server actually *lost* some hard-acked data.  If this happens we
2942                  * ignore this packet, as it may indicate that the server restarted in
2943                  * the middle of a call.  It is also possible that this is an old ack
2944                  * packet.  We don't abort the connection in this case, because this
2945                  * *might* just be an old ack packet.  The right way to detect a server
2946                  * restart in the midst of a call is to notice that the server epoch
2947                  * changed, btw.  */
2948                 /* XXX I'm not sure this is exactly right, since tfirst **IS**
2949                  * XXX unacknowledged.  I think that this is off-by-one, but
2950                  * XXX I don't dare change it just yet, since it will
2951                  * XXX interact badly with the server-restart detection 
2952                  * XXX code in receiveackpacket.  */
2953                 if (ntohl(rx_GetInt32(np, FIRSTACKOFFSET)) < call->tfirst) {
2954                     if (rx_stats_active)
2955                         rx_MutexIncrement(rx_stats.spuriousPacketsRead, rx_stats_mutex);
2956                     MUTEX_EXIT(&call->lock);
2957                     MUTEX_ENTER(&conn->conn_data_lock);
2958                     conn->refCount--;
2959                     MUTEX_EXIT(&conn->conn_data_lock);
2960                     return np;
2961                 }
2962             }
2963         }                       /* else not a data packet */
2964     }
2965
2966     osirx_AssertMine(&call->lock, "rxi_ReceivePacket middle");
2967     /* Set remote user defined status from packet */
2968     call->remoteStatus = np->header.userStatus;
2969
2970     /* Note the gap between the expected next packet and the actual
2971      * packet that arrived, when the new packet has a smaller serial number
2972      * than expected.  Rioses frequently reorder packets all by themselves,
2973      * so this will be quite important with very large window sizes.
2974      * Skew is checked against 0 here to avoid any dependence on the type of
2975      * inPacketSkew (which may be unsigned).  In C, -1 > (unsigned) 0 is always
2976      * true! 
2977      * The inPacketSkew should be a smoothed running value, not just a maximum.  MTUXXX
2978      * see CalculateRoundTripTime for an example of how to keep smoothed values.
2979      * I think using a beta of 1/8 is probably appropriate.  93.04.21
2980      */
2981     MUTEX_ENTER(&conn->conn_data_lock);
2982     skew = conn->lastSerial - np->header.serial;
2983     conn->lastSerial = np->header.serial;
2984     MUTEX_EXIT(&conn->conn_data_lock);
2985     if (skew > 0) {
2986         struct rx_peer *peer;
2987         peer = conn->peer;
2988         if (skew > peer->inPacketSkew) {
2989             dpf(("*** In skew changed from %d to %d\n", peer->inPacketSkew,
2990                  skew));
2991             peer->inPacketSkew = skew;
2992         }
2993     }
2994
2995     /* Now do packet type-specific processing */
2996     switch (np->header.type) {
2997     case RX_PACKET_TYPE_DATA:
2998         np = rxi_ReceiveDataPacket(call, np, 1, socket, host, port, tnop,
2999                                    newcallp);
3000         break;
3001     case RX_PACKET_TYPE_ACK:
3002         /* Respond immediately to ack packets requesting acknowledgement
3003          * (ping packets) */
3004         if (np->header.flags & RX_REQUEST_ACK) {
3005             if (call->error)
3006                 (void)rxi_SendCallAbort(call, 0, 1, 0);
3007             else
3008                 (void)rxi_SendAck(call, 0, np->header.serial,
3009                                   RX_ACK_PING_RESPONSE, 1);
3010         }
3011         np = rxi_ReceiveAckPacket(call, np, 1);
3012         break;
3013     case RX_PACKET_TYPE_ABORT: {
3014         /* An abort packet: reset the call, passing the error up to the user. */
3015         /* What if error is zero? */
3016         /* What if the error is -1? the application will treat it as a timeout. */
3017         afs_int32 errdata = ntohl(*(afs_int32 *) rx_DataOf(np));
3018         dpf(("rxi_ReceivePacket ABORT rx_DataOf = %d", errdata));
3019         rxi_CallError(call, errdata);
3020         MUTEX_EXIT(&call->lock);
3021         MUTEX_ENTER(&conn->conn_data_lock);
3022         conn->refCount--;
3023         MUTEX_EXIT(&conn->conn_data_lock);
3024         return np;              /* xmitting; drop packet */
3025     }
3026     case RX_PACKET_TYPE_BUSY:
3027         /* XXXX */
3028         break;
3029     case RX_PACKET_TYPE_ACKALL:
3030         /* All packets acknowledged, so we can drop all packets previously
3031          * readied for sending */
3032 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
3033         /* XXX Hack. We because we can't release the global rx lock when
3034          * sending packets (osi_NetSend) we drop all ack pkts while we're
3035          * traversing the tq in rxi_Start sending packets out because
3036          * packets may move to the freePacketQueue as result of being
3037          * here! So we drop these packets until we're safely out of the
3038          * traversing. Really ugly! 
3039          * For fine grain RX locking, we set the acked field in the packets
3040          * and let rxi_Start remove the packets from the transmit queue.
3041          */
3042         if (call->flags & RX_CALL_TQ_BUSY) {
3043 #ifdef  RX_ENABLE_LOCKS
3044             rxi_SetAcksInTransmitQueue(call);
3045             break;
3046 #else /* RX_ENABLE_LOCKS */
3047             MUTEX_EXIT(&call->lock);
3048             MUTEX_ENTER(&conn->conn_data_lock);
3049             conn->refCount--;
3050             MUTEX_EXIT(&conn->conn_data_lock);
3051             return np;          /* xmitting; drop packet */
3052 #endif /* RX_ENABLE_LOCKS */
3053         }
3054 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
3055         rxi_ClearTransmitQueue(call, 0);
3056         rxevent_Cancel(call->keepAliveEvent, call, RX_CALL_REFCOUNT_ALIVE);
3057         break;
3058     default:
3059         /* Should not reach here, unless the peer is broken: send an abort
3060          * packet */
3061         rxi_CallError(call, RX_PROTOCOL_ERROR);
3062         np = rxi_SendCallAbort(call, np, 1, 0);
3063         break;
3064     };
3065     /* Note when this last legitimate packet was received, for keep-alive
3066      * processing.  Note, we delay getting the time until now in the hope that
3067      * the packet will be delivered to the user before any get time is required
3068      * (if not, then the time won't actually be re-evaluated here). */
3069     call->lastReceiveTime = clock_Sec();
3070     MUTEX_EXIT(&call->lock);
3071     MUTEX_ENTER(&conn->conn_data_lock);
3072     conn->refCount--;
3073     MUTEX_EXIT(&conn->conn_data_lock);
3074     return np;
3075 }
3076
3077 /* return true if this is an "interesting" connection from the point of view
3078     of someone trying to debug the system */
3079 int
3080 rxi_IsConnInteresting(struct rx_connection *aconn)
3081 {
3082     int i;
3083     struct rx_call *tcall;
3084
3085     if (aconn->flags & (RX_CONN_MAKECALL_WAITING | RX_CONN_DESTROY_ME))
3086         return 1;
3087     for (i = 0; i < RX_MAXCALLS; i++) {
3088         tcall = aconn->call[i];
3089         if (tcall) {
3090             if ((tcall->state == RX_STATE_PRECALL)
3091                 || (tcall->state == RX_STATE_ACTIVE))
3092                 return 1;
3093             if ((tcall->mode == RX_MODE_SENDING)
3094                 || (tcall->mode == RX_MODE_RECEIVING))
3095                 return 1;
3096         }
3097     }
3098     return 0;
3099 }
3100
3101 #ifdef KERNEL
3102 /* if this is one of the last few packets AND it wouldn't be used by the
3103    receiving call to immediately satisfy a read request, then drop it on
3104    the floor, since accepting it might prevent a lock-holding thread from
3105    making progress in its reading. If a call has been cleared while in
3106    the precall state then ignore all subsequent packets until the call
3107    is assigned to a thread. */
3108
3109 static int
3110 TooLow(struct rx_packet *ap, struct rx_call *acall)
3111 {
3112     int rc = 0;
3113
3114     MUTEX_ENTER(&rx_quota_mutex);
3115     if (((ap->header.seq != 1) && (acall->flags & RX_CALL_CLEARED)
3116          && (acall->state == RX_STATE_PRECALL))
3117         || ((rx_nFreePackets < rxi_dataQuota + 2)
3118             && !((ap->header.seq < acall->rnext + rx_initSendWindow)
3119                  && (acall->flags & RX_CALL_READER_WAIT)))) {
3120         rc = 1;
3121     }
3122     MUTEX_EXIT(&rx_quota_mutex);
3123     return rc;
3124 }
3125 #endif /* KERNEL */
3126
3127 static void
3128 rxi_CheckReachEvent(struct rxevent *event, void *arg1, void *arg2)
3129 {
3130     struct rx_connection *conn = arg1;
3131     struct rx_call *acall = arg2;
3132     struct rx_call *call = acall;
3133     struct clock when, now;
3134     int i, waiting;
3135
3136     MUTEX_ENTER(&conn->conn_data_lock);
3137     conn->checkReachEvent = NULL;
3138     waiting = conn->flags & RX_CONN_ATTACHWAIT;
3139     if (event)
3140         conn->refCount--;
3141     MUTEX_EXIT(&conn->conn_data_lock);
3142
3143     if (waiting) {
3144         if (!call) {
3145             MUTEX_ENTER(&conn->conn_call_lock);
3146             MUTEX_ENTER(&conn->conn_data_lock);
3147             for (i = 0; i < RX_MAXCALLS; i++) {
3148                 struct rx_call *tc = conn->call[i];
3149                 if (tc && tc->state == RX_STATE_PRECALL) {
3150                     call = tc;
3151                     break;
3152                 }
3153             }
3154             if (!call)
3155                 /* Indicate that rxi_CheckReachEvent is no longer running by
3156                  * clearing the flag.  Must be atomic under conn_data_lock to
3157                  * avoid a new call slipping by: rxi_CheckConnReach holds
3158                  * conn_data_lock while checking RX_CONN_ATTACHWAIT.
3159                  */
3160                 conn->flags &= ~RX_CONN_ATTACHWAIT;
3161             MUTEX_EXIT(&conn->conn_data_lock);
3162             MUTEX_EXIT(&conn->conn_call_lock);
3163         }
3164
3165         if (call) {
3166             if (call != acall)
3167                 MUTEX_ENTER(&call->lock);
3168             rxi_SendAck(call, NULL, 0, RX_ACK_PING, 0);
3169             if (call != acall)
3170                 MUTEX_EXIT(&call->lock);
3171
3172             clock_GetTime(&now);
3173             when = now;
3174             when.sec += RX_CHECKREACH_TIMEOUT;
3175             MUTEX_ENTER(&conn->conn_data_lock);
3176             if (!conn->checkReachEvent) {
3177                 conn->refCount++;
3178                 conn->checkReachEvent =
3179                     rxevent_PostNow(&when, &now, rxi_CheckReachEvent, conn, 
3180                                     NULL);
3181             }
3182             MUTEX_EXIT(&conn->conn_data_lock);
3183         }
3184     }
3185 }
3186
3187 static int
3188 rxi_CheckConnReach(struct rx_connection *conn, struct rx_call *call)
3189 {
3190     struct rx_service *service = conn->service;
3191     struct rx_peer *peer = conn->peer;
3192     afs_uint32 now, lastReach;
3193
3194     if (service->checkReach == 0)
3195         return 0;
3196
3197     now = clock_Sec();
3198     MUTEX_ENTER(&peer->peer_lock);
3199     lastReach = peer->lastReachTime;
3200     MUTEX_EXIT(&peer->peer_lock);
3201     if (now - lastReach < RX_CHECKREACH_TTL)
3202         return 0;
3203
3204     MUTEX_ENTER(&conn->conn_data_lock);
3205     if (conn->flags & RX_CONN_ATTACHWAIT) {
3206         MUTEX_EXIT(&conn->conn_data_lock);
3207         return 1;
3208     }
3209     conn->flags |= RX_CONN_ATTACHWAIT;
3210     MUTEX_EXIT(&conn->conn_data_lock);
3211     if (!conn->checkReachEvent)
3212         rxi_CheckReachEvent(NULL, conn, call);
3213
3214     return 1;
3215 }
3216
3217 /* try to attach call, if authentication is complete */
3218 static void
3219 TryAttach(struct rx_call *acall, osi_socket socket,
3220           int *tnop, struct rx_call **newcallp,
3221           int reachOverride)
3222 {
3223     struct rx_connection *conn = acall->conn;
3224
3225     if (conn->type == RX_SERVER_CONNECTION
3226         && acall->state == RX_STATE_PRECALL) {
3227         /* Don't attach until we have any req'd. authentication. */
3228         if (RXS_CheckAuthentication(conn->securityObject, conn) == 0) {
3229             if (reachOverride || rxi_CheckConnReach(conn, acall) == 0)
3230                 rxi_AttachServerProc(acall, socket, tnop, newcallp);
3231             /* Note:  this does not necessarily succeed; there
3232              * may not any proc available
3233              */
3234         } else {
3235             rxi_ChallengeOn(acall->conn);
3236         }
3237     }
3238 }
3239
3240 /* A data packet has been received off the interface.  This packet is
3241  * appropriate to the call (the call is in the right state, etc.).  This
3242  * routine can return a packet to the caller, for re-use */
3243
3244 struct rx_packet *
3245 rxi_ReceiveDataPacket(struct rx_call *call,
3246                       struct rx_packet *np, int istack,
3247                       osi_socket socket, afs_uint32 host, u_short port,
3248                       int *tnop, struct rx_call **newcallp)
3249 {
3250     int ackNeeded = 0;          /* 0 means no, otherwise ack_reason */
3251     int newPackets = 0;
3252     int didHardAck = 0;
3253     int haveLast = 0;
3254     afs_uint32 seq; 
3255     afs_uint32 serial=0, flags=0;
3256     int isFirst;
3257     struct rx_packet *tnp;
3258     struct clock when, now;
3259     if (rx_stats_active)
3260         rx_MutexIncrement(rx_stats.dataPacketsRead, rx_stats_mutex);
3261
3262 #ifdef KERNEL
3263     /* If there are no packet buffers, drop this new packet, unless we can find
3264      * packet buffers from inactive calls */
3265     if (!call->error
3266         && (rxi_OverQuota(RX_PACKET_CLASS_RECEIVE) || TooLow(np, call))) {
3267         MUTEX_ENTER(&rx_freePktQ_lock);
3268         rxi_NeedMorePackets = TRUE;
3269         MUTEX_EXIT(&rx_freePktQ_lock);
3270         if (rx_stats_active)
3271             rx_MutexIncrement(rx_stats.noPacketBuffersOnRead, rx_stats_mutex);
3272         call->rprev = np->header.serial;
3273         rxi_calltrace(RX_TRACE_DROP, call);
3274         dpf(("packet %x dropped on receipt - quota problems", np));
3275         if (rxi_doreclaim)
3276             rxi_ClearReceiveQueue(call);
3277         clock_GetTime(&now);
3278         when = now;
3279         clock_Add(&when, &rx_softAckDelay);
3280         if (!call->delayedAckEvent
3281             || clock_Gt(&call->delayedAckEvent->eventTime, &when)) {
3282             rxevent_Cancel(call->delayedAckEvent, call,
3283                            RX_CALL_REFCOUNT_DELAY);
3284             CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
3285             call->delayedAckEvent =
3286                 rxevent_PostNow(&when, &now, rxi_SendDelayedAck, call, 0);
3287         }
3288         /* we've damaged this call already, might as well do it in. */
3289         return np;
3290     }
3291 #endif /* KERNEL */
3292
3293     /*
3294      * New in AFS 3.5, if the RX_JUMBO_PACKET flag is set then this
3295      * packet is one of several packets transmitted as a single
3296      * datagram. Do not send any soft or hard acks until all packets
3297      * in a jumbogram have been processed. Send negative acks right away.
3298      */
3299     for (isFirst = 1, tnp = NULL; isFirst || tnp; isFirst = 0) {
3300         /* tnp is non-null when there are more packets in the
3301          * current jumbo gram */
3302         if (tnp) {
3303             if (np)
3304                 rxi_FreePacket(np);
3305             np = tnp;
3306         }
3307
3308         seq = np->header.seq;
3309         serial = np->header.serial;
3310         flags = np->header.flags;
3311
3312         /* If the call is in an error state, send an abort message */
3313         if (call->error)
3314             return rxi_SendCallAbort(call, np, istack, 0);
3315
3316         /* The RX_JUMBO_PACKET is set in all but the last packet in each
3317          * AFS 3.5 jumbogram. */
3318         if (flags & RX_JUMBO_PACKET) {
3319             tnp = rxi_SplitJumboPacket(np, host, port, isFirst);
3320         } else {
3321             tnp = NULL;
3322         }
3323
3324         if (np->header.spare != 0) {
3325             MUTEX_ENTER(&call->conn->conn_data_lock);
3326             call->conn->flags |= RX_CONN_USING_PACKET_CKSUM;
3327             MUTEX_EXIT(&call->conn->conn_data_lock);
3328         }
3329
3330         /* The usual case is that this is the expected next packet */
3331         if (seq == call->rnext) {
3332
3333             /* Check to make sure it is not a duplicate of one already queued */
3334             if (queue_IsNotEmpty(&call->rq)
3335                 && queue_First(&call->rq, rx_packet)->header.seq == seq) {
3336                 if (rx_stats_active)
3337                     rx_MutexIncrement(rx_stats.dupPacketsRead, rx_stats_mutex);
3338                 dpf(("packet %x dropped on receipt - duplicate", np));
3339                 rxevent_Cancel(call->delayedAckEvent, call,
3340                                RX_CALL_REFCOUNT_DELAY);
3341                 np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
3342                 ackNeeded = 0;
3343                 call->rprev = seq;
3344                 continue;
3345             }
3346
3347             /* It's the next packet. Stick it on the receive queue
3348              * for this call. Set newPackets to make sure we wake
3349              * the reader once all packets have been processed */
3350             np->flags |= RX_PKTFLAG_RQ;
3351             queue_Prepend(&call->rq, np);
3352 #ifdef RXDEBUG_PACKET
3353             call->rqc++;
3354 #endif /* RXDEBUG_PACKET */
3355             call->nSoftAcks++;
3356             np = NULL;          /* We can't use this anymore */
3357             newPackets = 1;
3358
3359             /* If an ack is requested then set a flag to make sure we
3360              * send an acknowledgement for this packet */
3361             if (flags & RX_REQUEST_ACK) {
3362                 ackNeeded = RX_ACK_REQUESTED;
3363             }
3364
3365             /* Keep track of whether we have received the last packet */
3366             if (flags & RX_LAST_PACKET) {
3367                 call->flags |= RX_CALL_HAVE_LAST;
3368                 haveLast = 1;
3369             }
3370
3371             /* Check whether we have all of the packets for this call */
3372             if (call->flags & RX_CALL_HAVE_LAST) {
3373                 afs_uint32 tseq;        /* temporary sequence number */
3374                 struct rx_packet *tp;   /* Temporary packet pointer */
3375                 struct rx_packet *nxp;  /* Next pointer, for queue_Scan */
3376
3377                 for (tseq = seq, queue_Scan(&call->rq, tp, nxp, rx_packet)) {
3378                     if (tseq != tp->header.seq)
3379                         break;
3380                     if (tp->header.flags & RX_LAST_PACKET) {
3381                         call->flags |= RX_CALL_RECEIVE_DONE;
3382                         break;
3383                     }
3384                     tseq++;
3385                 }
3386             }
3387
3388             /* Provide asynchronous notification for those who want it
3389              * (e.g. multi rx) */
3390             if (call->arrivalProc) {
3391                 (*call->arrivalProc) (call, call->arrivalProcHandle,
3392                                       call->arrivalProcArg);
3393                 call->arrivalProc = (void (*)())0;
3394             }
3395
3396             /* Update last packet received */
3397             call->rprev = seq;
3398
3399             /* If there is no server process serving this call, grab
3400              * one, if available. We only need to do this once. If a
3401              * server thread is available, this thread becomes a server
3402              * thread and the server thread becomes a listener thread. */
3403             if (isFirst) {
3404                 TryAttach(call, socket, tnop, newcallp, 0);
3405             }
3406         }
3407         /* This is not the expected next packet. */
3408         else {
3409             /* Determine whether this is a new or old packet, and if it's
3410              * a new one, whether it fits into the current receive window.
3411              * Also figure out whether the packet was delivered in sequence.
3412              * We use the prev variable to determine whether the new packet
3413              * is the successor of its immediate predecessor in the
3414              * receive queue, and the missing flag to determine whether
3415              * any of this packets predecessors are missing.  */
3416
3417             afs_uint32 prev;    /* "Previous packet" sequence number */
3418             struct rx_packet *tp;       /* Temporary packet pointer */
3419             struct rx_packet *nxp;      /* Next pointer, for queue_Scan */
3420             int missing;        /* Are any predecessors missing? */
3421
3422             /* If the new packet's sequence number has been sent to the
3423              * application already, then this is a duplicate */
3424             if (seq < call->rnext) {
3425                 if (rx_stats_active)
3426                     rx_MutexIncrement(rx_stats.dupPacketsRead, rx_stats_mutex);
3427                 rxevent_Cancel(call->delayedAckEvent, call,
3428                                RX_CALL_REFCOUNT_DELAY);
3429                 np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
3430                 ackNeeded = 0;
3431                 call->rprev = seq;
3432                 continue;
3433             }
3434
3435             /* If the sequence number is greater than what can be
3436              * accomodated by the current window, then send a negative
3437              * acknowledge and drop the packet */
3438             if ((call->rnext + call->rwind) <= seq) {
3439                 rxevent_Cancel(call->delayedAckEvent, call,
3440                                RX_CALL_REFCOUNT_DELAY);
3441                 np = rxi_SendAck(call, np, serial, RX_ACK_EXCEEDS_WINDOW,
3442                                  istack);
3443                 ackNeeded = 0;
3444                 call->rprev = seq;
3445                 continue;
3446             }
3447
3448             /* Look for the packet in the queue of old received packets */
3449             for (prev = call->rnext - 1, missing =
3450                  0, queue_Scan(&call->rq, tp, nxp, rx_packet)) {
3451                 /*Check for duplicate packet */
3452                 if (seq == tp->header.seq) {
3453                     if (rx_stats_active)
3454                         rx_MutexIncrement(rx_stats.dupPacketsRead, rx_stats_mutex);
3455                     rxevent_Cancel(call->delayedAckEvent, call,
3456                                    RX_CALL_REFCOUNT_DELAY);
3457                     np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE,
3458                                      istack);
3459                     ackNeeded = 0;
3460                     call->rprev = seq;
3461                     goto nextloop;
3462                 }
3463                 /* If we find a higher sequence packet, break out and
3464                  * insert the new packet here. */
3465                 if (seq < tp->header.seq)
3466                     break;
3467                 /* Check for missing packet */
3468                 if (tp->header.seq != prev + 1) {
3469                     missing = 1;
3470                 }
3471
3472                 prev = tp->header.seq;
3473             }
3474
3475             /* Keep track of whether we have received the last packet. */
3476             if (flags & RX_LAST_PACKET) {
3477                 call->flags |= RX_CALL_HAVE_LAST;
3478             }
3479
3480             /* It's within the window: add it to the the receive queue.
3481              * tp is left by the previous loop either pointing at the
3482              * packet before which to insert the new packet, or at the
3483              * queue head if the queue is empty or the packet should be
3484              * appended. */
3485             np->flags |= RX_PKTFLAG_RQ;
3486 #ifdef RXDEBUG_PACKET
3487             call->rqc++;
3488 #endif /* RXDEBUG_PACKET */
3489             queue_InsertBefore(tp, np);
3490             call->nSoftAcks++;
3491             np = NULL;
3492
3493             /* Check whether we have all of the packets for this call */
3494             if ((call->flags & RX_CALL_HAVE_LAST)
3495                 && !(call->flags & RX_CALL_RECEIVE_DONE)) {
3496                 afs_uint32 tseq;        /* temporary sequence number */
3497
3498                 for (tseq =
3499                      call->rnext, queue_Scan(&call->rq, tp, nxp, rx_packet)) {
3500                     if (tseq != tp->header.seq)
3501                         break;
3502                     if (tp->header.flags & RX_LAST_PACKET) {
3503                         call->flags |= RX_CALL_RECEIVE_DONE;
3504                         break;
3505                     }
3506                     tseq++;
3507                 }
3508             }
3509
3510             /* We need to send an ack of the packet is out of sequence, 
3511              * or if an ack was requested by the peer. */
3512             if (seq != prev + 1 || missing) {
3513                 ackNeeded = RX_ACK_OUT_OF_SEQUENCE;
3514             } else if (flags & RX_REQUEST_ACK) {
3515                 ackNeeded = RX_ACK_REQUESTED;
3516             }
3517
3518             /* Acknowledge the last packet for each call */
3519             if (flags & RX_LAST_PACKET) {
3520                 haveLast = 1;
3521             }
3522
3523             call->rprev = seq;
3524         }
3525       nextloop:;
3526     }
3527
3528     if (newPackets) {
3529         /*
3530          * If the receiver is waiting for an iovec, fill the iovec
3531          * using the data from the receive queue */
3532         if (call->flags & RX_CALL_IOVEC_WAIT) {
3533             didHardAck = rxi_FillReadVec(call, serial);
3534             /* the call may have been aborted */
3535             if (call->error) {
3536                 return NULL;
3537             }
3538             if (didHardAck) {
3539                 ackNeeded = 0;
3540             }
3541         }
3542
3543         /* Wakeup the reader if any */
3544         if ((call->flags & RX_CALL_READER_WAIT)
3545             && (!(call->flags & RX_CALL_IOVEC_WAIT) || !(call->iovNBytes)
3546                 || (call->iovNext >= call->iovMax)
3547                 || (call->flags & RX_CALL_RECEIVE_DONE))) {
3548             call->flags &= ~RX_CALL_READER_WAIT;
3549 #ifdef  RX_ENABLE_LOCKS
3550             CV_BROADCAST(&call->cv_rq);
3551 #else
3552             osi_rxWakeup(&call->rq);
3553 #endif
3554         }
3555     }
3556
3557     /*
3558      * Send an ack when requested by the peer, or once every
3559      * rxi_SoftAckRate packets until the last packet has been
3560      * received. Always send a soft ack for the last packet in
3561      * the server's reply. */
3562     if (ackNeeded) {
3563         rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
3564         np = rxi_SendAck(call, np, serial, ackNeeded, istack);
3565     } else if (call->nSoftAcks > (u_short) rxi_SoftAckRate) {
3566         rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
3567         np = rxi_SendAck(call, np, serial, RX_ACK_IDLE, istack);
3568     } else if (call->nSoftAcks) {
3569         clock_GetTime(&now);
3570         when = now;
3571         if (haveLast && !(flags & RX_CLIENT_INITIATED)) {
3572             clock_Add(&when, &rx_lastAckDelay);
3573         } else {
3574             clock_Add(&when, &rx_softAckDelay);
3575         }
3576         if (!call->delayedAckEvent
3577             || clock_Gt(&call->delayedAckEvent->eventTime, &when)) {
3578             rxevent_Cancel(call->delayedAckEvent, call,
3579                            RX_CALL_REFCOUNT_DELAY);
3580             CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
3581             call->delayedAckEvent =
3582                 rxevent_PostNow(&when, &now, rxi_SendDelayedAck, call, 0);
3583         }
3584     } else if (call->flags & RX_CALL_RECEIVE_DONE) {
3585         rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
3586     }
3587
3588     return np;
3589 }
3590
3591 #ifdef  ADAPT_WINDOW
3592 static void rxi_ComputeRate();
3593 #endif
3594
3595 static void
3596 rxi_UpdatePeerReach(struct rx_connection *conn, struct rx_call *acall)
3597 {
3598     struct rx_peer *peer = conn->peer;
3599
3600     MUTEX_ENTER(&peer->peer_lock);
3601     peer->lastReachTime = clock_Sec();
3602     MUTEX_EXIT(&peer->peer_lock);
3603
3604     MUTEX_ENTER(&conn->conn_data_lock);
3605     if (conn->flags & RX_CONN_ATTACHWAIT) {
3606         int i;
3607
3608         conn->flags &= ~RX_CONN_ATTACHWAIT;
3609         MUTEX_EXIT(&conn->conn_data_lock);
3610
3611         for (i = 0; i < RX_MAXCALLS; i++) {
3612             struct rx_call *call = conn->call[i];
3613             if (call) {
3614                 if (call != acall)
3615                     MUTEX_ENTER(&call->lock);
3616                 /* tnop can be null if newcallp is null */
3617                 TryAttach(call, (osi_socket) - 1, NULL, NULL, 1);
3618                 if (call != acall)
3619                     MUTEX_EXIT(&call->lock);
3620             }
3621         }
3622     } else
3623         MUTEX_EXIT(&conn->conn_data_lock);
3624 }
3625
3626 static const char *
3627 rx_ack_reason(int reason)
3628 {
3629     switch (reason) {
3630     case RX_ACK_REQUESTED:
3631         return "requested";
3632     case RX_ACK_DUPLICATE:
3633         return "duplicate";
3634     case RX_ACK_OUT_OF_SEQUENCE:
3635         return "sequence";
3636     case RX_ACK_EXCEEDS_WINDOW:
3637         return "window";
3638     case RX_ACK_NOSPACE:
3639         return "nospace";
3640     case RX_ACK_PING:
3641         return "ping";
3642     case RX_ACK_PING_RESPONSE:
3643         return "response";
3644     case RX_ACK_DELAY:
3645         return "delay";
3646     case RX_ACK_IDLE:
3647         return "idle";
3648     default:
3649         return "unknown!!";
3650     }
3651 }
3652
3653
3654 /* rxi_ComputePeerNetStats
3655  *
3656  * Called exclusively by rxi_ReceiveAckPacket to compute network link
3657  * estimates (like RTT and throughput) based on ack packets.  Caller
3658  * must ensure that the packet in question is the right one (i.e.
3659  * serial number matches).
3660  */
3661 static void
3662 rxi_ComputePeerNetStats(struct rx_call *call, struct rx_packet *p,
3663                         struct rx_ackPacket *ap, struct rx_packet *np)
3664 {
3665     struct rx_peer *peer = call->conn->peer;
3666
3667     /* Use RTT if not delayed by client. */
3668     if (ap->reason != RX_ACK_DELAY)
3669         rxi_ComputeRoundTripTime(p, &p->timeSent, peer);
3670 #ifdef ADAPT_WINDOW
3671     rxi_ComputeRate(peer, call, p, np, ap->reason);
3672 #endif
3673 }
3674
3675 /* The real smarts of the whole thing.  */
3676 struct rx_packet *
3677 rxi_ReceiveAckPacket(struct rx_call *call, struct rx_packet *np,
3678                      int istack)
3679 {
3680     struct rx_ackPacket *ap;
3681     int nAcks;
3682     struct rx_packet *tp;
3683     struct rx_packet *nxp;      /* Next packet pointer for queue_Scan */
3684     struct rx_connection *conn = call->conn;
3685     struct rx_peer *peer = conn->peer;
3686     afs_uint32 first;
3687     afs_uint32 serial;
3688     /* because there are CM's that are bogus, sending weird values for this. */
3689     afs_uint32 skew = 0;
3690     int nbytes;
3691     int missing;
3692     int acked;
3693     int nNacked = 0;
3694     int newAckCount = 0;
3695     u_short maxMTU = 0;         /* Set if peer supports AFS 3.4a jumbo datagrams */
3696     int maxDgramPackets = 0;    /* Set if peer supports AFS 3.5 jumbo datagrams */
3697
3698     if (rx_stats_active)
3699         rx_MutexIncrement(rx_stats.ackPacketsRead, rx_stats_mutex);
3700     ap = (struct rx_ackPacket *)rx_DataOf(np);
3701     nbytes = rx_Contiguous(np) - (int)((ap->acks) - (u_char *) ap);
3702     if (nbytes < 0)
3703         return np;              /* truncated ack packet */
3704
3705     /* depends on ack packet struct */
3706     nAcks = MIN((unsigned)nbytes, (unsigned)ap->nAcks);
3707     first = ntohl(ap->firstPacket);
3708     serial = ntohl(ap->serial);
3709     /* temporarily disabled -- needs to degrade over time 
3710      * skew = ntohs(ap->maxSkew); */
3711
3712     /* Ignore ack packets received out of order */
3713     if (first < call->tfirst) {
3714         return np;
3715     }
3716
3717     if (np->header.flags & RX_SLOW_START_OK) {
3718         call->flags |= RX_CALL_SLOW_START_OK;
3719     }
3720
3721     if (ap->reason == RX_ACK_PING_RESPONSE)
3722         rxi_UpdatePeerReach(conn, call);
3723
3724 #ifdef RXDEBUG
3725 #ifdef AFS_NT40_ENV
3726     if (rxdebug_active) {
3727         char msg[512];
3728         size_t len;
3729
3730         len = _snprintf(msg, sizeof(msg),
3731                         "tid[%d] RACK: reason %s serial %u previous %u seq %u skew %d first %u acks %u space %u ",
3732                          GetCurrentThreadId(), rx_ack_reason(ap->reason), 
3733                          ntohl(ap->serial), ntohl(ap->previousPacket),
3734                          (unsigned int)np->header.seq, (unsigned int)skew, 
3735                          ntohl(ap->firstPacket), ap->nAcks, ntohs(ap->bufferSpace) );
3736         if (nAcks) {
3737             int offset;
3738
3739             for (offset = 0; offset < nAcks && len < sizeof(msg); offset++) 
3740                 msg[len++] = (ap->acks[offset] == RX_ACK_TYPE_NACK ? '-' : '*');
3741         }
3742         msg[len++]='\n';
3743         msg[len] = '\0';
3744         OutputDebugString(msg);
3745     }
3746 #else /* AFS_NT40_ENV */
3747     if (rx_Log) {
3748         fprintf(rx_Log,
3749                 "RACK: reason %x previous %u seq %u serial %u skew %d first %u",
3750                 ap->reason, ntohl(ap->previousPacket),
3751                 (unsigned int)np->header.seq, (unsigned int)serial,
3752                 (unsigned int)skew, ntohl(ap->firstPacket));
3753         if (nAcks) {
3754             int offset;
3755             for (offset = 0; offset < nAcks; offset++)
3756                 putc(ap->acks[offset] == RX_ACK_TYPE_NACK ? '-' : '*',
3757                      rx_Log);
3758         }
3759         putc('\n', rx_Log);
3760     }
3761 #endif /* AFS_NT40_ENV */
3762 #endif
3763
3764     /* Update the outgoing packet skew value to the latest value of
3765      * the peer's incoming packet skew value.  The ack packet, of
3766      * course, could arrive out of order, but that won't affect things
3767      * much */
3768     MUTEX_ENTER(&peer->peer_lock);
3769     peer->outPacketSkew = skew;
3770
3771     /* Check for packets that no longer need to be transmitted, and
3772      * discard them.  This only applies to packets positively
3773      * acknowledged as having been sent to the peer's upper level.
3774      * All other packets must be retained.  So only packets with
3775      * sequence numbers < ap->firstPacket are candidates. */
3776     for (queue_Scan(&call->tq, tp, nxp, rx_packet)) {
3777         if (tp->header.seq >= first)
3778             break;
3779         call->tfirst = tp->header.seq + 1;
3780         if (serial
3781             && (tp->header.serial == serial || tp->firstSerial == serial))
3782             rxi_ComputePeerNetStats(call, tp, ap, np);
3783         if (!(tp->flags & RX_PKTFLAG_ACKED)) {
3784             newAckCount++;
3785         }
3786 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
3787         /* XXX Hack. Because we have to release the global rx lock when sending
3788          * packets (osi_NetSend) we drop all acks while we're traversing the tq
3789          * in rxi_Start sending packets out because packets may move to the
3790          * freePacketQueue as result of being here! So we drop these packets until
3791          * we're safely out of the traversing. Really ugly! 
3792          * To make it even uglier, if we're using fine grain locking, we can
3793          * set the ack bits in the packets and have rxi_Start remove the packets
3794          * when it's done transmitting.
3795          */
3796         if (call->flags & RX_CALL_TQ_BUSY) {
3797 #ifdef RX_ENABLE_LOCKS
3798             tp->flags |= RX_PKTFLAG_ACKED;
3799             call->flags |= RX_CALL_TQ_SOME_ACKED;
3800 #else /* RX_ENABLE_LOCKS */
3801             break;
3802 #endif /* RX_ENABLE_LOCKS */
3803         } else
3804 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
3805         {
3806             queue_Remove(tp);
3807             tp->flags &= ~RX_PKTFLAG_TQ;
3808 #ifdef RXDEBUG_PACKET
3809             call->tqc--;
3810 #endif /* RXDEBUG_PACKET */
3811             rxi_FreePacket(tp); /* rxi_FreePacket mustn't wake up anyone, preemptively. */
3812         }
3813     }
3814
3815 #ifdef ADAPT_WINDOW
3816     /* Give rate detector a chance to respond to ping requests */
3817     if (ap->reason == RX_ACK_PING_RESPONSE) {
3818         rxi_ComputeRate(peer, call, 0, np, ap->reason);
3819     }
3820 #endif
3821
3822     /* N.B. we don't turn off any timers here.  They'll go away by themselves, anyway */
3823
3824     /* Now go through explicit acks/nacks and record the results in
3825      * the waiting packets.  These are packets that can't be released
3826      * yet, even with a positive acknowledge.  This positive
3827      * acknowledge only means the packet has been received by the
3828      * peer, not that it will be retained long enough to be sent to
3829      * the peer's upper level.  In addition, reset the transmit timers
3830      * of any missing packets (those packets that must be missing
3831      * because this packet was out of sequence) */
3832
3833     call->nSoftAcked = 0;
3834     for (missing = 0, queue_Scan(&call->tq, tp, nxp, rx_packet)) {
3835         /* Update round trip time if the ack was stimulated on receipt
3836          * of this packet */
3837 #ifdef AFS_GLOBAL_RXLOCK_KERNEL
3838 #ifdef RX_ENABLE_LOCKS
3839         if (tp->header.seq >= first)
3840 #endif /* RX_ENABLE_LOCKS */
3841 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
3842             if (serial
3843                 && (tp->header.serial == serial || tp->firstSerial == serial))
3844                 rxi_ComputePeerNetStats(call, tp, ap, np);
3845
3846         /* Set the acknowledge flag per packet based on the
3847          * information in the ack packet. An acknowlegded packet can
3848          * be downgraded when the server has discarded a packet it
3849          * soacked previously, or when an ack packet is received
3850          * out of sequence. */
3851         if (tp->header.seq < first) {
3852             /* Implicit ack information */
3853             if (!(tp->flags & RX_PKTFLAG_ACKED)) {
3854                 newAckCount++;
3855             }
3856             tp->flags |= RX_PKTFLAG_ACKED;
3857         } else if (tp->header.seq < first + nAcks) {
3858             /* Explicit ack information:  set it in the packet appropriately */
3859             if (ap->acks[tp->header.seq - first] == RX_ACK_TYPE_ACK) {
3860                 if (!(tp->flags & RX_PKTFLAG_ACKED)) {
3861                     newAckCount++;
3862                     tp->flags |= RX_PKTFLAG_ACKED;
3863                 }
3864                 if (missing) {
3865                     nNacked++;
3866                 } else {
3867                     call->nSoftAcked++;
3868                 }
3869             } else /* RX_ACK_TYPE_NACK */ {
3870                 tp->flags &= ~RX_PKTFLAG_ACKED;
3871                 missing = 1;
3872             }
3873         } else {
3874             tp->flags &= ~RX_PKTFLAG_ACKED;
3875             missing = 1;
3876         }
3877
3878         /* If packet isn't yet acked, and it has been transmitted at least 
3879          * once, reset retransmit time using latest timeout 
3880          * ie, this should readjust the retransmit timer for all outstanding 
3881          * packets...  So we don't just retransmit when we should know better*/
3882
3883         if (!(tp->flags & RX_PKTFLAG_ACKED) && !clock_IsZero(&tp->retryTime)) {
3884             tp->retryTime = tp->timeSent;
3885             clock_Add(&tp->retryTime, &peer->timeout);
3886             /* shift by eight because one quarter-sec ~ 256 milliseconds */
3887             clock_Addmsec(&(tp->retryTime), ((afs_uint32) tp->backoff) << 8);
3888         }
3889     }
3890
3891     /* If the window has been extended by this acknowledge packet,
3892      * then wakeup a sender waiting in alloc for window space, or try
3893      * sending packets now, if he's been sitting on packets due to
3894      * lack of window space */
3895     if (call->tnext < (call->tfirst + call->twind)) {
3896 #ifdef  RX_ENABLE_LOCKS
3897         CV_SIGNAL(&call->cv_twind);
3898 #else
3899         if (call->flags & RX_CALL_WAIT_WINDOW_ALLOC) {
3900             call->flags &= ~RX_CALL_WAIT_WINDOW_ALLOC;
3901             osi_rxWakeup(&call->twind);
3902         }
3903 #endif
3904         if (call->flags & RX_CALL_WAIT_WINDOW_SEND) {
3905             call->flags &= ~RX_CALL_WAIT_WINDOW_SEND;
3906         }
3907     }
3908
3909     /* if the ack packet has a receivelen field hanging off it,
3910      * update our state */
3911     if (np->length >= rx_AckDataSize(ap->nAcks) + 2 * sizeof(afs_int32)) {
3912         afs_uint32 tSize;
3913
3914         /* If the ack packet has a "recommended" size that is less than 
3915          * what I am using now, reduce my size to match */
3916         rx_packetread(np, rx_AckDataSize(ap->nAcks) + sizeof(afs_int32),
3917                       (int)sizeof(afs_int32), &tSize);
3918         tSize = (afs_uint32) ntohl(tSize);
3919         peer->natMTU = rxi_AdjustIfMTU(MIN(tSize, peer->ifMTU));
3920
3921         /* Get the maximum packet size to send to this peer */
3922         rx_packetread(np, rx_AckDataSize(ap->nAcks), (int)sizeof(afs_int32),
3923                       &tSize);
3924         tSize = (afs_uint32) ntohl(tSize);
3925         tSize = (afs_uint32) MIN(tSize, rx_MyMaxSendSize);
3926         tSize = rxi_AdjustMaxMTU(peer->natMTU, tSize);
3927
3928         /* sanity check - peer might have restarted with different params.
3929          * If peer says "send less", dammit, send less...  Peer should never 
3930          * be unable to accept packets of the size that prior AFS versions would
3931          * send without asking.  */
3932         if (peer->maxMTU != tSize) {
3933             if (peer->maxMTU > tSize) /* possible cong., maxMTU decreased */
3934                 peer->congestSeq++;
3935             peer->maxMTU = tSize;
3936             peer->MTU = MIN(tSize, peer->MTU);
3937             call->MTU = MIN(call->MTU, tSize);
3938         }
3939
3940         if (np->length == rx_AckDataSize(ap->nAcks) + 3 * sizeof(afs_int32)) {
3941             /* AFS 3.4a */
3942             rx_packetread(np,
3943                           rx_AckDataSize(ap->nAcks) + 2 * sizeof(afs_int32),
3944                           (int)sizeof(afs_int32), &tSize);
3945             tSize = (afs_uint32) ntohl(tSize);  /* peer's receive window, if it's */
3946             if (tSize < call->twind) {  /* smaller than our send */
3947                 call->twind = tSize;    /* window, we must send less... */
3948                 call->ssthresh = MIN(call->twind, call->ssthresh);
3949                 call->conn->twind[call->channel] = call->twind;
3950             }
3951
3952             /* Only send jumbograms to 3.4a fileservers. 3.3a RX gets the
3953              * network MTU confused with the loopback MTU. Calculate the
3954              * maximum MTU here for use in the slow start code below.
3955              */
3956             maxMTU = peer->maxMTU;
3957             /* Did peer restart with older RX version? */
3958             if (peer->maxDgramPackets > 1) {
3959                 peer->maxDgramPackets = 1;
3960             }
3961         } else if (np->length >=
3962                    rx_AckDataSize(ap->nAcks) + 4 * sizeof(afs_int32)) {
3963             /* AFS 3.5 */
3964             rx_packetread(np,
3965                           rx_AckDataSize(ap->nAcks) + 2 * sizeof(afs_int32),
3966                           sizeof(afs_int32), &tSize);
3967             tSize = (afs_uint32) ntohl(tSize);
3968             /*
3969              * As of AFS 3.5 we set the send window to match the receive window. 
3970              */
3971             if (tSize < call->twind) {
3972                 call->twind = tSize;
3973                 call->conn->twind[call->channel] = call->twind;
3974                 call->ssthresh = MIN(call->twind, call->ssthresh);
3975             } else if (tSize > call->twind) {
3976                 call->twind = tSize;
3977                 call->conn->twind[call->channel] = call->twind;
3978             }
3979
3980             /*
3981              * As of AFS 3.5, a jumbogram is more than one fixed size
3982              * packet transmitted in a single UDP datagram. If the remote
3983              * MTU is smaller than our local MTU then never send a datagram
3984              * larger than the natural MTU.
3985              */
3986             rx_packetread(np,
3987                           rx_AckDataSize(ap->nAcks) + 3 * sizeof(afs_int32),
3988                           sizeof(afs_int32), &tSize);
3989             maxDgramPackets = (afs_uint32) ntohl(tSize);
3990             maxDgramPackets = MIN(maxDgramPackets, rxi_nDgramPackets);
3991             maxDgramPackets =
3992                 MIN(maxDgramPackets, (int)(peer->ifDgramPackets));
3993             maxDgramPackets = MIN(maxDgramPackets, tSize);
3994             if (maxDgramPackets > 1) {
3995                 peer->maxDgramPackets = maxDgramPackets;
3996                 call->MTU = RX_JUMBOBUFFERSIZE + RX_HEADER_SIZE;
3997             } else {
3998                 peer->maxDgramPackets = 1;
3999                 call->MTU = peer->natMTU;
4000             }
4001         } else if (peer->maxDgramPackets > 1) {
4002             /* Restarted with lower version of RX */
4003             peer->maxDgramPackets = 1;
4004         }
4005     } else if (peer->maxDgramPackets > 1
4006                || peer->maxMTU != OLD_MAX_PACKET_SIZE) {
4007         /* Restarted with lower version of RX */
4008         peer->maxMTU = OLD_MAX_PACKET_SIZE;
4009         peer->natMTU = OLD_MAX_PACKET_SIZE;
4010         peer->MTU = OLD_MAX_PACKET_SIZE;
4011         peer->maxDgramPackets = 1;
4012         peer->nDgramPackets = 1;
4013         peer->congestSeq++;
4014         call->MTU = OLD_MAX_PACKET_SIZE;
4015     }
4016
4017     if (nNacked) {
4018         /*
4019          * Calculate how many datagrams were successfully received after
4020          * the first missing packet and adjust the negative ack counter
4021          * accordingly.
4022          */
4023         call->nAcks = 0;
4024         call->nNacks++;
4025         nNacked = (nNacked + call->nDgramPackets - 1) / call->nDgramPackets;
4026         if (call->nNacks < nNacked) {
4027             call->nNacks = nNacked;
4028         }
4029     } else {
4030         call->nAcks += newAckCount;
4031         call->nNacks = 0;
4032     }
4033
4034     if (call->flags & RX_CALL_FAST_RECOVER) {
4035         if (nNacked) {
4036             call->cwind = MIN((int)(call->cwind + 1), rx_maxSendWindow);
4037         } else {
4038             call->flags &= ~RX_CALL_FAST_RECOVER;
4039             call->cwind = call->nextCwind;
4040             call->nextCwind = 0;
4041             call->nAcks = 0;
4042         }
4043         call->nCwindAcks = 0;
4044     } else if (nNacked && call->nNacks >= (u_short) rx_nackThreshold) {
4045         /* Three negative acks in a row trigger congestion recovery */
4046 #ifdef  AFS_GLOBAL_RXLOCK_KERNEL
4047         MUTEX_EXIT(&peer->peer_lock);
4048         if (call->flags & RX_CALL_FAST_RECOVER_WAIT) {
4049             /* someone else is waiting to start recovery */
4050             return np;
4051         }
4052         call->flags |= RX_CALL_FAST_RECOVER_WAIT;
4053         rxi_WaitforTQBusy(call);
4054         MUTEX_ENTER(&peer->peer_lock);
4055 #endif /* AFS_GLOBAL_RXLOCK_KERNEL */
4056         call->flags &= ~RX_CALL_FAST_RECOVER_WAIT;
4057         call->flags |= RX_CALL_FAST_RECOVER;
4058         call->ssthresh = MAX(4, MIN((int)call->cwind, (int)call->twind)) >> 1;
4059         call->cwind =
4060             MIN((int)(call->ssthresh + rx_nackThreshold), rx_maxSendWindow);
4061         call->nDgramPackets = MAX(2, (int)call->nDgramPackets) >> 1;
4062         call->nextCwind = call->ssthresh;
4063         call->nAcks = 0;
4064         call->nNacks = 0;
4065         peer->MTU = call->MTU;
4066         peer->cwind = call->nextCwind;
4067         peer->nDgramPackets = call->nDgramPackets;
4068         peer->congestSeq++;
4069         call->congestSeq = peer->congestSeq;
4070         /* Reset the resend times on the packets that were nacked
4071          * so we will retransmit as soon as the window permits*/
4072         for (acked = 0, queue_ScanBackwards(&call->tq, tp, nxp, rx_packet)) {
4073             if (acked) {
4074                 if (!(tp->flags & RX_PKTFLAG_ACKED)) {
4075                     clock_Zero(&tp->retryTime);
4076                 }
4077             } else if (tp->flags & RX_PKTFLAG_ACKED) {
4078                 acked = 1;
4079             }
4080         }
4081     } else {
4082         /* If cwind is smaller than ssthresh, then increase
4083          * the window one packet for each ack we receive (exponential
4084          * growth).
4085          * If cwind is greater than or equal to ssthresh then increase
4086          * the congestion window by one packet for each cwind acks we
4087          * receive (linear growth).  */
4088         if (call->cwind < call->ssthresh) {
4089             call->cwind =
4090                 MIN((int)call->ssthresh, (int)(call->cwind + newAckCount));
4091             call->nCwindAcks = 0;
4092         } else {
4093             call->nCwindAcks += newAckCount;
4094             if (call->nCwindAcks >= call->cwind) {
4095                 call->nCwindAcks = 0;
4096                 call->cwind = MIN((int)(call->cwind + 1), rx_maxSendWindow);
4097             }
4098         }
4099         /*
4100          * If we have received several acknowledgements in a row then
4101          * it is time to increase the size of our datagrams
4102          */
4103         if ((int)call->nAcks > rx_nDgramThreshold) {
4104             if (peer->maxDgramPackets > 1) {
4105                 if (call->nDgramPackets < peer->maxDgramPackets) {
4106                     call->nDgramPackets++;
4107                 }
4108                 call->MTU = RX_HEADER_SIZE + RX_JUMBOBUFFERSIZE;
4109             } else if (call->MTU < peer->maxMTU) {
4110                 call->MTU += peer->natMTU;
4111                 call->MTU = MIN(call->MTU, peer->maxMTU);
4112             }
4113             call->nAcks = 0;
4114         }
4115     }
4116
4117     MUTEX_EXIT(&peer->peer_lock);       /* rxi_Start will lock peer. */
4118
4119     /* Servers need to hold the call until all response packets have
4120      * been acknowledged. Soft acks are good enough since clients
4121      * are not allowed to clear their receive queues. */
4122     if (call->state == RX_STATE_HOLD
4123         && call->tfirst + call->nSoftAcked >= call->tnext) {
4124         call->state = RX_STATE_DALLY;
4125         rxi_ClearTransmitQueue(call, 0);
4126         rxevent_Cancel(call->keepAliveEvent, call, RX_CALL_REFCOUNT_ALIVE);
4127     } else if (!queue_IsEmpty(&call->tq)) {
4128         rxi_Start(0, call, 0, istack);
4129     }
4130     return np;
4131 }
4132
4133 /* Received a response to a challenge packet */
4134 struct rx_packet *
4135 rxi_ReceiveResponsePacket(struct rx_connection *conn,
4136                           struct rx_packet *np, int istack)
4137 {
4138     int error;
4139
4140     /* Ignore the packet if we're the client */
4141     if (conn->type == RX_CLIENT_CONNECTION)
4142         return np;
4143
4144     /* If already authenticated, ignore the packet (it's probably a retry) */
4145     if (RXS_CheckAuthentication(conn->securityObject, conn) == 0)
4146         return np;
4147
4148     /* Otherwise, have the security object evaluate the response packet */
4149     error = RXS_CheckResponse(conn->securityObject, conn, np);
4150     if (error) {
4151         /* If the response is invalid, reset the connection, sending
4152          * an abort to the peer */
4153 #ifndef KERNEL
4154         rxi_Delay(1);
4155 #endif
4156         rxi_ConnectionError(conn, error);
4157         MUTEX_ENTER(&conn->conn_data_lock);
4158         np = rxi_SendConnectionAbort(conn, np, istack, 0);
4159         MUTEX_EXIT(&conn->conn_data_lock);
4160         return np;
4161     } else {
4162         /* If the response is valid, any calls waiting to attach
4163          * servers can now do so */
4164         int i;
4165
4166         for (i = 0; i < RX_MAXCALLS; i++) {
4167             struct rx_call *call = conn->call[i];
4168             if (call) {
4169                 MUTEX_ENTER(&call->lock);
4170                 if (call->state == RX_STATE_PRECALL)
4171                     rxi_AttachServerProc(call, (osi_socket) - 1, NULL, NULL);
4172                 /* tnop can be null if newcallp is null */
4173                 MUTEX_EXIT(&call->lock);
4174             }
4175         }
4176
4177         /* Update the peer reachability information, just in case
4178          * some calls went into attach-wait while we were waiting
4179          * for authentication..
4180          */
4181         rxi_UpdatePeerReach(conn, NULL);
4182     }
4183     return np;
4184 }
4185
4186 /* A client has received an authentication challenge: the security
4187  * object is asked to cough up a respectable response packet to send
4188  * back to the server.  The server is responsible for retrying the
4189  * challenge if it fails to get a response. */
4190
4191 struct rx_packet *
4192 rxi_ReceiveChallengePacket(struct rx_connection *conn,
4193                            struct rx_packet *np, int istack)
4194 {
4195     int error;
4196
4197     /* Ignore the challenge if we're the server */
4198     if (conn->type == RX_SERVER_CONNECTION)
4199         return np;
4200
4201     /* Ignore the challenge if the connection is otherwise idle; someone's
4202      * trying to use us as an oracle. */
4203     if (!rxi_HasActiveCalls(conn))
4204         return np;
4205
4206     /* Send the security object the challenge packet.  It is expected to fill
4207      * in the response. */
4208     error = RXS_GetResponse(conn->securityObject, conn, np);
4209
4210     /* If the security object is unable to return a valid response, reset the
4211      * connection and send an abort to the peer.  Otherwise send the response
4212      * packet to the peer connection. */
4213     if (error) {
4214         rxi_ConnectionError(conn, error);
4215         MUTEX_ENTER(&conn->conn_data_lock);
4216         np = rxi_SendConnectionAbort(conn, np, istack, 0);
4217         MUTEX_EXIT(&conn->conn_data_lock);
4218     } else {
4219         np = rxi_SendSpecial((struct rx_call *)0, conn, np,
4220                              RX_PACKET_TYPE_RESPONSE, NULL, -1, istack);
4221     }
4222     return np;
4223 }
4224
4225
4226 /* Find an available server process to service the current request in
4227  * the given call structure.  If one isn't available, queue up this
4228  * call so it eventually gets one */
4229 void
4230 rxi_AttachServerProc(struct rx_call *call,
4231                      osi_socket socket, int *tnop,
4232                      struct rx_call **newcallp)
4233 {
4234     struct rx_serverQueueEntry *sq;
4235     struct rx_service *service = call->conn->service;
4236     int haveQuota = 0;
4237
4238     /* May already be attached */
4239     if (call->state == RX_STATE_ACTIVE)
4240         return;
4241
4242     MUTEX_ENTER(&rx_serverPool_lock);
4243
4244     haveQuota = QuotaOK(service);
4245     if ((!haveQuota) || queue_IsEmpty(&rx_idleServerQueue)) {
4246         /* If there are no processes available to service this call,
4247          * put the call on the incoming call queue (unless it's
4248          * already on the queue).
4249          */
4250 #ifdef RX_ENABLE_LOCKS
4251         if (haveQuota)
4252             ReturnToServerPool(service);
4253 #endif /* RX_ENABLE_LOCKS */
4254
4255         if (!(call->flags & RX_CALL_WAIT_PROC)) {
4256             call->flags |= RX_CALL_WAIT_PROC;
4257             MUTEX_ENTER(&rx_waiting_mutex);
4258             rx_nWaiting++;
4259             rx_nWaited++;
4260             MUTEX_EXIT(&rx_waiting_mutex);
4261             rxi_calltrace(RX_CALL_ARRIVAL, call);
4262             SET_CALL_QUEUE_LOCK(call, &rx_serverPool_lock);
4263             queue_Append(&rx_incomingCallQueue, call);
4264         }
4265     } else {
4266         sq = queue_First(&rx_idleServerQueue, rx_serverQueueEntry);
4267
4268         /* If hot threads are enabled, and both newcallp and sq->socketp
4269          * are non-null, then this thread will process the call, and the
4270          * idle server thread will start listening on this threads socket.
4271          */
4272         queue_Remove(sq);
4273         if (rx_enable_hot_thread && newcallp && sq->socketp) {
4274             *newcallp = call;
4275             *tnop = sq->tno;
4276             *sq->socketp = socket;
4277             clock_GetTime(&call->startTime);
4278             CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
4279         } else {
4280             sq->newcall = call;
4281         }
4282         if (call->flags & RX_CALL_WAIT_PROC) {
4283             /* Conservative:  I don't think this should happen */
4284             call->flags &= ~RX_CALL_WAIT_PROC;
4285             if (queue_IsOnQueue(call)) {
4286                 queue_Remove(call);
4287                 
4288                 MUTEX_ENTER(&rx_waiting_mutex);
4289                 rx_nWaiting--;
4290                 MUTEX_EXIT(&rx_waiting_mutex);
4291             }
4292         }
4293         call->state = RX_STATE_ACTIVE;
4294         call->mode = RX_MODE_RECEIVING;
4295 #ifdef RX_KERNEL_TRACE
4296         {
4297             int glockOwner = ISAFS_GLOCK();
4298             if (!glockOwner)
4299                 AFS_GLOCK();
4300             afs_Trace3(afs_iclSetp, CM_TRACE_WASHERE, ICL_TYPE_STRING,
4301                        __FILE__, ICL_TYPE_INT32, __LINE__, ICL_TYPE_POINTER,
4302                        call);
4303             if (!glockOwner)
4304                 AFS_GUNLOCK();
4305         }
4306 #endif
4307         if (call->flags & RX_CALL_CLEARED) {
4308             /* send an ack now to start the packet flow up again */
4309             call->flags &= ~RX_CALL_CLEARED;
4310             rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
4311         }
4312 #ifdef  RX_ENABLE_LOCKS
4313         CV_SIGNAL(&sq->cv);
4314 #else
4315         service->nRequestsRunning++;
4316         if (service->nRequestsRunning <= service->minProcs)
4317             rxi_minDeficit--;
4318         rxi_availProcs--;
4319         osi_rxWakeup(sq);
4320 #endif
4321     }
4322     MUTEX_EXIT(&rx_serverPool_lock);
4323 }
4324
4325 /* Delay the sending of an acknowledge event for a short while, while
4326  * a new call is being prepared (in the case of a client) or a reply
4327  * is being prepared (in the case of a server).  Rather than sending
4328  * an ack packet, an ACKALL packet is sent. */
4329 void
4330 rxi_AckAll(struct rxevent *event, struct rx_call *call, char *dummy)
4331 {
4332 #ifdef RX_ENABLE_LOCKS
4333     if (event) {
4334         MUTEX_ENTER(&call->lock);
4335         call->delayedAckEvent = NULL;
4336         CALL_RELE(call, RX_CALL_REFCOUNT_ACKALL);
4337     }
4338     rxi_SendSpecial(call, call->conn, (struct rx_packet *)0,
4339                     RX_PACKET_TYPE_ACKALL, NULL, 0, 0);
4340     if (event)
4341         MUTEX_EXIT(&call->lock);
4342 #else /* RX_ENABLE_LOCKS */
4343     if (event)
4344         call->delayedAckEvent = NULL;
4345     rxi_SendSpecial(call, call->conn, (struct rx_packet *)0,
4346                     RX_PACKET_TYPE_ACKALL, NULL, 0, 0);
4347 #endif /* RX_ENABLE_LOCKS */
4348 }
4349
4350 void
4351 rxi_SendDelayedAck(struct rxevent *event, void *arg1, void *unused)
4352 {
4353     struct rx_call *call = arg1;
4354 #ifdef RX_ENABLE_LOCKS
4355     if (event) {
4356         MUTEX_ENTER(&call->lock);
4357   &nbs