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