rx-thread-id-startup-20030303

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

/*
 * Copyright 2000, International Business Machines Corporation and others.
 * All Rights Reserved.
 * 
 * This software has been released under the terms of the IBM Public
 * License.  For details, see the LICENSE file in the top-level source
 * directory or online at http://www.openafs.org/dl/license10.html
 */

/* RX:  Extended Remote Procedure Call */

#include <afsconfig.h>
#ifdef  KERNEL
#include "afs/param.h"
#else
#include <afs/param.h>
#endif

RCSID("$Header$");

#ifdef KERNEL
#include "afs/sysincludes.h"
#include "afsincludes.h"
#ifndef UKERNEL
#include "h/types.h"
#include "h/time.h"
#include "h/stat.h"
#ifdef  AFS_OSF_ENV
#include <net/net_globals.h>
#endif  /* AFS_OSF_ENV */
#ifdef AFS_LINUX20_ENV
#include "h/socket.h"
#endif
#include "netinet/in.h"
#include "afs/afs_args.h"
#include "afs/afs_osi.h"
#if     (defined(AFS_AUX_ENV) || defined(AFS_AIX_ENV))
#include "h/systm.h"
#endif
#ifdef RXDEBUG
#undef RXDEBUG      /* turn off debugging */
#endif /* RXDEBUG */
#if defined(AFS_SGI_ENV)
#include "sys/debug.h"
#endif
#include "afsint.h"
#ifdef  AFS_ALPHA_ENV
#undef kmem_alloc
#undef kmem_free
#undef mem_alloc
#undef mem_free
#undef register
#endif  /* AFS_ALPHA_ENV */
#else /* !UKERNEL */
#include "afs/sysincludes.h"
#include "afsincludes.h"
#endif /* !UKERNEL */
#include "afs/lock.h"
#include "rx_kmutex.h"
#include "rx_kernel.h"
#include "rx_clock.h"
#include "rx_queue.h"
#include "rx.h"
#include "rx_globals.h"
#include "rx_trace.h"
#define AFSOP_STOP_RXCALLBACK   210     /* Stop CALLBACK process */
#define AFSOP_STOP_AFS          211     /* Stop AFS process */
#define AFSOP_STOP_BKG          212     /* Stop BKG process */
#include "afsint.h"
extern afs_int32 afs_termState;
#ifdef AFS_AIX41_ENV
#include "sys/lockl.h"
#include "sys/lock_def.h"
#endif /* AFS_AIX41_ENV */
# include "rxgen_consts.h"
#else /* KERNEL */
# include <sys/types.h>
# include <errno.h>
#ifdef AFS_NT40_ENV
# include <stdlib.h>
# include <fcntl.h>
# include <afsutil.h>
#else
# include <sys/socket.h>
# include <sys/file.h>
# include <netdb.h>
# include <sys/stat.h>
# include <netinet/in.h>
# include <sys/time.h>
#endif
#ifdef HAVE_STRING_H
#include <string.h>
#else
#ifdef HAVE_STRINGS_H
#include <strings.h>
#endif
#endif
# include "rx.h"
# include "rx_user.h"
# include "rx_clock.h"
# include "rx_queue.h"
# include "rx_globals.h"
# include "rx_trace.h"
# include <afs/rxgen_consts.h>
#endif /* KERNEL */

int (*registerProgram)() = 0;
int (*swapNameProgram)() = 0;

/* Local static routines */
static void rxi_DestroyConnectionNoLock(register struct rx_connection *conn);
#ifdef RX_ENABLE_LOCKS
static void rxi_SetAcksInTransmitQueue(register struct rx_call *call);
#endif

#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
struct rx_tq_debug {
    afs_int32 rxi_start_aborted; /* rxi_start awoke after rxi_Send in error. */
    afs_int32 rxi_start_in_error;
} rx_tq_debug;
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */

/*
 * rxi_rpc_peer_stat_cnt counts the total number of peer stat structures
 * currently allocated within rx.  This number is used to allocate the
 * memory required to return the statistics when queried.
 */

static unsigned int rxi_rpc_peer_stat_cnt;

/*
 * rxi_rpc_process_stat_cnt counts the total number of local process stat
 * structures currently allocated within rx.  The number is used to allocate
 * the memory required to return the statistics when queried.
 */

static unsigned int rxi_rpc_process_stat_cnt;

#if !defined(offsetof)
#include <stddef.h>     /* for definition of offsetof() */
#endif

#ifdef AFS_PTHREAD_ENV
#include <assert.h>

/*
 * Use procedural initialization of mutexes/condition variables
 * to ease NT porting
 */

extern pthread_mutex_t rxkad_stats_mutex;
extern pthread_mutex_t des_init_mutex;
extern pthread_mutex_t des_random_mutex;
extern pthread_mutex_t rx_clock_mutex;
extern pthread_mutex_t rxi_connCacheMutex;
extern pthread_mutex_t rx_event_mutex;
extern pthread_mutex_t osi_malloc_mutex;
extern pthread_mutex_t event_handler_mutex;
extern pthread_mutex_t listener_mutex;
extern pthread_mutex_t rx_if_init_mutex;
extern pthread_mutex_t rx_if_mutex;
extern pthread_mutex_t rxkad_client_uid_mutex;
extern pthread_mutex_t rxkad_random_mutex;

extern pthread_cond_t rx_event_handler_cond;
extern pthread_cond_t rx_listener_cond;

static pthread_mutex_t epoch_mutex;
static pthread_mutex_t rx_init_mutex;
static pthread_mutex_t rx_debug_mutex;

static void rxi_InitPthread(void) {
    assert(pthread_mutex_init(&rx_clock_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rxi_connCacheMutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rx_init_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&epoch_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rx_event_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&des_init_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&des_random_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&osi_malloc_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&event_handler_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&listener_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rx_if_init_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rx_if_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rxkad_client_uid_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rxkad_random_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rxkad_stats_mutex,
                              (const pthread_mutexattr_t*)0)==0);
    assert(pthread_mutex_init(&rx_debug_mutex,
                              (const pthread_mutexattr_t*)0)==0);

    assert(pthread_cond_init(&rx_event_handler_cond,
                              (const pthread_condattr_t*)0)==0);
    assert(pthread_cond_init(&rx_listener_cond,
                              (const pthread_condattr_t*)0)==0);
    assert(pthread_key_create(&rx_thread_id_key, NULL) == 0);
}

pthread_once_t rx_once_init = PTHREAD_ONCE_INIT;
#define INIT_PTHREAD_LOCKS \
assert(pthread_once(&rx_once_init, rxi_InitPthread)==0);
/*
 * The rx_stats_mutex mutex protects the following global variables:
 * rxi_dataQuota
 * rxi_minDeficit
 * rxi_availProcs
 * rxi_totalMin
 * rxi_lowConnRefCount
 * rxi_lowPeerRefCount
 * rxi_nCalls
 * rxi_Alloccnt
 * rxi_Allocsize
 * rx_nFreePackets
 * rx_tq_debug
 * rx_stats
 */
#else
#define INIT_PTHREAD_LOCKS
#endif


/* Variables for handling the minProcs implementation.  availProcs gives the
 * number of threads available in the pool at this moment (not counting dudes
 * executing right now).  totalMin gives the total number of procs required
 * for handling all minProcs requests.  minDeficit is a dynamic variable
 * tracking the # of procs required to satisfy all of the remaining minProcs
 * demands.
 * For fine grain locking to work, the quota check and the reservation of
 * a server thread has to come while rxi_availProcs and rxi_minDeficit
 * are locked. To this end, the code has been modified under #ifdef
 * RX_ENABLE_LOCKS so that quota checks and reservation occur at the
 * same time. A new function, ReturnToServerPool() returns the allocation.
 * 
 * A call can be on several queue's (but only one at a time). When
 * rxi_ResetCall wants to remove the call from a queue, it has to ensure
 * that no one else is touching the queue. To this end, we store the address
 * of the queue lock in the call structure (under the call lock) when we
 * put the call on a queue, and we clear the call_queue_lock when the
 * call is removed from a queue (once the call lock has been obtained).
 * This allows rxi_ResetCall to safely synchronize with others wishing
 * to manipulate the queue.
 */

#ifdef RX_ENABLE_LOCKS
static afs_kmutex_t rx_rpc_stats;
void rxi_StartUnlocked();
#endif

/* We keep a "last conn pointer" in rxi_FindConnection. The odds are 
** pretty good that the next packet coming in is from the same connection 
** as the last packet, since we're send multiple packets in a transmit window.
*/
struct rx_connection *rxLastConn = 0; 

#ifdef RX_ENABLE_LOCKS
/* The locking hierarchy for rx fine grain locking is composed of these
 * tiers:
 *
 * rx_connHashTable_lock - synchronizes conn creation, rx_connHashTable access
 * conn_call_lock - used to synchonize rx_EndCall and rx_NewCall
 * call->lock - locks call data fields.
 * These are independent of each other:
 *      rx_freeCallQueue_lock
 *      rxi_keyCreate_lock
 * rx_serverPool_lock
 * freeSQEList_lock
 *
 * serverQueueEntry->lock
 * rx_rpc_stats
 * rx_peerHashTable_lock - locked under rx_connHashTable_lock
 * peer->lock - locks peer data fields.
 * conn_data_lock - that more than one thread is not updating a conn data
 *                  field at the same time.
 * rx_freePktQ_lock
 *
 * lowest level:
 *      multi_handle->lock
 *      rxevent_lock
 *      rx_stats_mutex
 *
 * Do we need a lock to protect the peer field in the conn structure?
 *      conn->peer was previously a constant for all intents and so has no
 *      lock protecting this field. The multihomed client delta introduced
 *      a RX code change : change the peer field in the connection structure
 *      to that remote inetrface from which the last packet for this
 *      connection was sent out. This may become an issue if further changes
 *      are made.
 */
#define SET_CALL_QUEUE_LOCK(C, L) (C)->call_queue_lock = (L)
#define CLEAR_CALL_QUEUE_LOCK(C) (C)->call_queue_lock = NULL
#ifdef RX_LOCKS_DB
/* rxdb_fileID is used to identify the lock location, along with line#. */
static int rxdb_fileID = RXDB_FILE_RX;
#endif /* RX_LOCKS_DB */
#else /* RX_ENABLE_LOCKS */
#define SET_CALL_QUEUE_LOCK(C, L)
#define CLEAR_CALL_QUEUE_LOCK(C)
#endif /* RX_ENABLE_LOCKS */
struct rx_serverQueueEntry *rx_waitForPacket = 0;

/* ------------Exported Interfaces------------- */

/* This function allows rxkad to set the epoch to a suitably random number
 * which rx_NewConnection will use in the future.  The principle purpose is to
 * get rxnull connections to use the same epoch as the rxkad connections do, at
 * least once the first rxkad connection is established.  This is important now
 * that the host/port addresses aren't used in FindConnection: the uniqueness
 * of epoch/cid matters and the start time won't do. */

#ifdef AFS_PTHREAD_ENV
/*
 * This mutex protects the following global variables:
 * rx_epoch
 */

#define LOCK_EPOCH assert(pthread_mutex_lock(&epoch_mutex)==0);
#define UNLOCK_EPOCH assert(pthread_mutex_unlock(&epoch_mutex)==0);
#else
#define LOCK_EPOCH
#define UNLOCK_EPOCH
#endif /* AFS_PTHREAD_ENV */

void rx_SetEpoch (afs_uint32 epoch)
{
    LOCK_EPOCH
    rx_epoch = epoch;
    UNLOCK_EPOCH
}

/* Initialize rx.  A port number may be mentioned, in which case this
 * becomes the default port number for any service installed later.
 * If 0 is provided for the port number, a random port will be chosen
 * by the kernel.  Whether this will ever overlap anything in
 * /etc/services is anybody's guess...  Returns 0 on success, -1 on
 * error. */
static int rxinit_status = 1;
#ifdef AFS_PTHREAD_ENV
/*
 * This mutex protects the following global variables:
 * rxinit_status
 */

#define LOCK_RX_INIT assert(pthread_mutex_lock(&rx_init_mutex)==0);
#define UNLOCK_RX_INIT assert(pthread_mutex_unlock(&rx_init_mutex)==0);
#else
#define LOCK_RX_INIT
#define UNLOCK_RX_INIT
#endif

int rx_Init(u_int port)
{
#ifdef KERNEL
    osi_timeval_t tv;
#else /* KERNEL */
    struct timeval tv;
#endif /* KERNEL */
    char *htable, *ptable;
    int tmp_status;

#if defined(AFS_DJGPP_ENV) && !defined(DEBUG)
    __djgpp_set_quiet_socket(1);
#endif

    SPLVAR;

    INIT_PTHREAD_LOCKS
    LOCK_RX_INIT
    if (rxinit_status == 0) {
        tmp_status = rxinit_status;
        UNLOCK_RX_INIT
        return tmp_status; /* Already started; return previous error code. */
    }

#ifdef AFS_NT40_ENV
    if (afs_winsockInit()<0)
        return -1;
#endif

#ifndef KERNEL
    /*
     * Initialize anything necessary to provide a non-premptive threading
     * environment.
     */
    rxi_InitializeThreadSupport();
#endif

    /* Allocate and initialize a socket for client and perhaps server
     * connections. */

    rx_socket = rxi_GetUDPSocket((u_short)port); 
    if (rx_socket == OSI_NULLSOCKET) {
        UNLOCK_RX_INIT
        return RX_ADDRINUSE;
    }
    

#ifdef  RX_ENABLE_LOCKS
#ifdef RX_LOCKS_DB
    rxdb_init();
#endif /* RX_LOCKS_DB */
    MUTEX_INIT(&rx_stats_mutex, "rx_stats_mutex",MUTEX_DEFAULT,0);
    MUTEX_INIT(&rx_rpc_stats, "rx_rpc_stats",MUTEX_DEFAULT,0);
    MUTEX_INIT(&rx_freePktQ_lock, "rx_freePktQ_lock",MUTEX_DEFAULT,0);
    MUTEX_INIT(&freeSQEList_lock, "freeSQEList lock",MUTEX_DEFAULT,0);
    MUTEX_INIT(&rx_freeCallQueue_lock, "rx_freeCallQueue_lock",
               MUTEX_DEFAULT,0);
    CV_INIT(&rx_waitingForPackets_cv, "rx_waitingForPackets_cv",CV_DEFAULT, 0);
    MUTEX_INIT(&rx_peerHashTable_lock,"rx_peerHashTable_lock",MUTEX_DEFAULT,0);
    MUTEX_INIT(&rx_connHashTable_lock,"rx_connHashTable_lock",MUTEX_DEFAULT,0);
    MUTEX_INIT(&rx_serverPool_lock, "rx_serverPool_lock", MUTEX_DEFAULT, 0);
#ifndef KERNEL
    MUTEX_INIT(&rxi_keyCreate_lock, "rxi_keyCreate_lock", MUTEX_DEFAULT, 0);
#endif /* !KERNEL */
#if defined(KERNEL) && defined(AFS_HPUX110_ENV)
    if ( !uniprocessor )
      rx_sleepLock = alloc_spinlock(LAST_HELD_ORDER-10, "rx_sleepLock");
#endif /* KERNEL && AFS_HPUX110_ENV */
#else /* RX_ENABLE_LOCKS */
#if defined(KERNEL) && defined(AFS_GLOBAL_SUNLOCK) && !defined(AFS_HPUX_ENV) && !defined(AFS_OBSD_ENV)
    mutex_init(&afs_rxglobal_lock, "afs_rxglobal_lock", MUTEX_DEFAULT, NULL);
#endif /* AFS_GLOBAL_SUNLOCK */
#endif /* RX_ENABLE_LOCKS */

    rxi_nCalls = 0;
    rx_connDeadTime = 12;
    rx_tranquil     = 0;        /* reset flag */
    memset((char *)&rx_stats, 0, sizeof(struct rx_stats));
    htable = (char *)
        osi_Alloc(rx_hashTableSize*sizeof(struct rx_connection *));
    PIN(htable, rx_hashTableSize*sizeof(struct rx_connection *));  /* XXXXX */
    memset(htable, 0, rx_hashTableSize*sizeof(struct rx_connection *));
    ptable =  (char *) osi_Alloc(rx_hashTableSize*sizeof(struct rx_peer *));   
    PIN(ptable, rx_hashTableSize*sizeof(struct rx_peer *));       /* XXXXX */
    memset(ptable, 0, rx_hashTableSize*sizeof(struct rx_peer *));

    /* Malloc up a bunch of packets & buffers */
    rx_nFreePackets = 0;
    rx_nPackets = rx_extraPackets + RX_MAX_QUOTA + 2;   /* fudge */
    queue_Init(&rx_freePacketQueue);
    rxi_NeedMorePackets = FALSE;
    rxi_MorePackets(rx_nPackets);
    rx_CheckPackets();

    NETPRI;
    AFS_RXGLOCK();

    clock_Init();

#if defined(AFS_NT40_ENV) && !defined(AFS_PTHREAD_ENV)
    tv.tv_sec = clock_now.sec;
    tv.tv_usec = clock_now.usec;
    srand((unsigned int) tv.tv_usec);
#else
    osi_GetTime(&tv);
#endif
    if (port) {
        rx_port = port;
    } else {
#if defined(KERNEL) && !defined(UKERNEL)
        /* Really, this should never happen in a real kernel */
        rx_port = 0;
#else
        struct sockaddr_in addr;
        int addrlen = sizeof(addr);
        if (getsockname((int)rx_socket, (struct sockaddr *) &addr, &addrlen)) {
            rx_Finalize();
            return -1;
        }
        rx_port = addr.sin_port;
#endif
    }
    rx_stats.minRtt.sec = 9999999;
#ifdef  KERNEL
    rx_SetEpoch (tv.tv_sec | 0x80000000);
#else
    rx_SetEpoch (tv.tv_sec);            /* Start time of this package, rxkad
                                         * will provide a randomer value. */
#endif
    MUTEX_ENTER(&rx_stats_mutex);
    rxi_dataQuota += rx_extraQuota;     /* + extra pkts caller asked to rsrv */
    MUTEX_EXIT(&rx_stats_mutex);
    /* *Slightly* random start time for the cid.  This is just to help
     * out with the hashing function at the peer */
    rx_nextCid = ((tv.tv_sec ^ tv.tv_usec) << RX_CIDSHIFT);
    rx_connHashTable = (struct rx_connection **) htable;
    rx_peerHashTable = (struct rx_peer **) ptable;

    rx_lastAckDelay.sec = 0;
    rx_lastAckDelay.usec = 400000; /* 400 milliseconds */
    rx_hardAckDelay.sec = 0;
    rx_hardAckDelay.usec = 100000; /* 100 milliseconds */
    rx_softAckDelay.sec = 0;
    rx_softAckDelay.usec = 100000; /* 100 milliseconds */

    rxevent_Init(20, rxi_ReScheduleEvents);

    /* Initialize various global queues */
    queue_Init(&rx_idleServerQueue);
    queue_Init(&rx_incomingCallQueue);
    queue_Init(&rx_freeCallQueue);

#if defined(AFS_NT40_ENV) && !defined(KERNEL)
    /* Initialize our list of usable IP addresses. */
    rx_GetIFInfo();
#endif

    /* Start listener process (exact function is dependent on the
     * implementation environment--kernel or user space) */
    rxi_StartListener();

    AFS_RXGUNLOCK();
    USERPRI;
    tmp_status = rxinit_status = 0;
    UNLOCK_RX_INIT
    return tmp_status;
}

/* called with unincremented nRequestsRunning to see if it is OK to start
 * a new thread in this service.  Could be "no" for two reasons: over the
 * max quota, or would prevent others from reaching their min quota.
 */
#ifdef RX_ENABLE_LOCKS
/* This verion of QuotaOK reserves quota if it's ok while the
 * rx_serverPool_lock is held.  Return quota using ReturnToServerPool().
 */
static int QuotaOK(register struct rx_service *aservice)
{
    /* check if over max quota */
    if (aservice->nRequestsRunning >= aservice->maxProcs) {
        return 0;
    }

    /* under min quota, we're OK */
    /* otherwise, can use only if there are enough to allow everyone
     * to go to their min quota after this guy starts.
     */
    MUTEX_ENTER(&rx_stats_mutex);
    if ((aservice->nRequestsRunning < aservice->minProcs) ||
         (rxi_availProcs > rxi_minDeficit)) {
        aservice->nRequestsRunning++;
        /* just started call in minProcs pool, need fewer to maintain
         * guarantee */
        if (aservice->nRequestsRunning <= aservice->minProcs)
            rxi_minDeficit--;
        rxi_availProcs--;
        MUTEX_EXIT(&rx_stats_mutex);
        return 1;
    }
    MUTEX_EXIT(&rx_stats_mutex);

    return 0;
}

static void ReturnToServerPool(register struct rx_service *aservice)
{
    aservice->nRequestsRunning--;
    MUTEX_ENTER(&rx_stats_mutex);
    if (aservice->nRequestsRunning < aservice->minProcs) rxi_minDeficit++;
    rxi_availProcs++;
    MUTEX_EXIT(&rx_stats_mutex);
}

#else /* RX_ENABLE_LOCKS */
static int QuotaOK(register struct rx_service *aservice)
{
    int rc=0;
    /* under min quota, we're OK */
    if (aservice->nRequestsRunning < aservice->minProcs) return 1;

    /* check if over max quota */
    if (aservice->nRequestsRunning >= aservice->maxProcs) return 0;

    /* otherwise, can use only if there are enough to allow everyone
     * to go to their min quota after this guy starts.
     */
    if (rxi_availProcs > rxi_minDeficit) rc = 1;
    return rc;
}
#endif /* RX_ENABLE_LOCKS */

#ifndef KERNEL
/* Called by rx_StartServer to start up lwp's to service calls.
   NExistingProcs gives the number of procs already existing, and which
   therefore needn't be created. */
void rxi_StartServerProcs(int nExistingProcs)
{
    register struct rx_service *service;
    register int i;
    int maxdiff = 0;
    int nProcs = 0;

    /* For each service, reserve N processes, where N is the "minimum"
       number of processes that MUST be able to execute a request in parallel,
       at any time, for that process.  Also compute the maximum difference
       between any service's maximum number of processes that can run
       (i.e. the maximum number that ever will be run, and a guarantee
       that this number will run if other services aren't running), and its
       minimum number.  The result is the extra number of processes that
       we need in order to provide the latter guarantee */
    for (i=0; i<RX_MAX_SERVICES; i++) {
        int diff;
        service = rx_services[i];
        if (service == (struct rx_service *) 0) break;
        nProcs += service->minProcs;
        diff = service->maxProcs - service->minProcs;
        if (diff > maxdiff) maxdiff = diff;
    }
    nProcs += maxdiff; /* Extra processes needed to allow max number requested to run in any given service, under good conditions */
    nProcs -= nExistingProcs; /* Subtract the number of procs that were previously created for use as server procs */
    for (i = 0; i<nProcs; i++) {
        rxi_StartServerProc(rx_ServerProc, rx_stackSize);
    }
}
#endif /* KERNEL */

/* This routine must be called if any services are exported.  If the
 * donateMe flag is set, the calling process is donated to the server
 * process pool */
void rx_StartServer(int donateMe)
{
    register struct rx_service *service;
    register int i, nProcs=0;
    SPLVAR;
    clock_NewTime();

    NETPRI;
    AFS_RXGLOCK();
    /* Start server processes, if necessary (exact function is dependent
     * on the implementation environment--kernel or user space).  DonateMe
     * will be 1 if there is 1 pre-existing proc, i.e. this one.  In this
     * case, one less new proc will be created rx_StartServerProcs.
     */
    rxi_StartServerProcs(donateMe);

    /* count up the # of threads in minProcs, and add set the min deficit to
     * be that value, too.
     */
    for (i=0; i<RX_MAX_SERVICES; i++) {
        service = rx_services[i];
        if (service == (struct rx_service *) 0) break;
        MUTEX_ENTER(&rx_stats_mutex);
        rxi_totalMin += service->minProcs;
        /* below works even if a thread is running, since minDeficit would
         * still have been decremented and later re-incremented.
         */
        rxi_minDeficit += service->minProcs;
        MUTEX_EXIT(&rx_stats_mutex);
    }

    /* Turn on reaping of idle server connections */
    rxi_ReapConnections();

    AFS_RXGUNLOCK();
    USERPRI;

    if (donateMe) {
#ifndef AFS_NT40_ENV
#ifndef KERNEL
        char name[32];
#ifdef AFS_PTHREAD_ENV
        pid_t pid;
        pid = (pid_t) pthread_self();
#else /* AFS_PTHREAD_ENV */
        PROCESS pid;
        LWP_CurrentProcess(&pid);
#endif /* AFS_PTHREAD_ENV */

        sprintf(name,"srv_%d", ++nProcs);
        if (registerProgram)
            (*registerProgram)(pid, name);
#endif /* KERNEL */
#endif /* AFS_NT40_ENV */
        rx_ServerProc(); /* Never returns */
    }
    return;
}

/* Create a new client connection to the specified service, using the
 * specified security object to implement the security model for this
 * connection. */
struct rx_connection *rx_NewConnection(register afs_uint32 shost, 
        u_short sport, u_short sservice, 
        register struct rx_securityClass *securityObject, int serviceSecurityIndex)
{
    int hashindex;
    afs_int32 cid;
    register struct rx_connection *conn;

    SPLVAR;

    clock_NewTime();
    dpf(("rx_NewConnection(host %x, port %u, service %u, securityObject %x, serviceSecurityIndex %d)\n",
          shost, sport, sservice, securityObject, serviceSecurityIndex));

    /* Vasilsi said: "NETPRI protects Cid and Alloc", but can this be true in
     * the case of kmem_alloc? */
    conn = rxi_AllocConnection();
#ifdef  RX_ENABLE_LOCKS
    MUTEX_INIT(&conn->conn_call_lock, "conn call lock",MUTEX_DEFAULT,0);
    MUTEX_INIT(&conn->conn_data_lock, "conn call lock",MUTEX_DEFAULT,0);
    CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
#endif
    NETPRI;
    AFS_RXGLOCK();
    MUTEX_ENTER(&rx_connHashTable_lock);
    cid = (rx_nextCid += RX_MAXCALLS);
    conn->type = RX_CLIENT_CONNECTION;
    conn->cid = cid;
    conn->epoch = rx_epoch;
    conn->peer = rxi_FindPeer(shost, sport, 0, 1);
    conn->serviceId = sservice;
    conn->securityObject = securityObject;
    /* This doesn't work in all compilers with void (they're buggy), so fake it
     * with VOID */
    conn->securityData = (VOID *) 0;
    conn->securityIndex = serviceSecurityIndex;
    rx_SetConnDeadTime(conn, rx_connDeadTime);
    conn->ackRate = RX_FAST_ACK_RATE;
    conn->nSpecific = 0;
    conn->specific = NULL;
    conn->challengeEvent = NULL;
    conn->delayedAbortEvent = NULL;
    conn->abortCount = 0;
    conn->error = 0;

    RXS_NewConnection(securityObject, conn);
    hashindex = CONN_HASH(shost, sport, conn->cid, conn->epoch, RX_CLIENT_CONNECTION);
    
    conn->refCount++; /* no lock required since only this thread knows... */
    conn->next = rx_connHashTable[hashindex];
    rx_connHashTable[hashindex] = conn;
    MUTEX_ENTER(&rx_stats_mutex);
    rx_stats.nClientConns++;
    MUTEX_EXIT(&rx_stats_mutex);

    MUTEX_EXIT(&rx_connHashTable_lock);
    AFS_RXGUNLOCK();
    USERPRI;
    return conn;
}

void rx_SetConnDeadTime(register struct rx_connection *conn, register int seconds)
{
    /* The idea is to set the dead time to a value that allows several
     * keepalives to be dropped without timing out the connection. */
    conn->secondsUntilDead = MAX(seconds, 6);
    conn->secondsUntilPing = conn->secondsUntilDead/6;
}

int rxi_lowPeerRefCount = 0;
int rxi_lowConnRefCount = 0;

/*
 * Cleanup a connection that was destroyed in rxi_DestroyConnectioNoLock.
 * NOTE: must not be called with rx_connHashTable_lock held.
 */
void rxi_CleanupConnection(struct rx_connection *conn)
{
    /* Notify the service exporter, if requested, that this connection
     * is being destroyed */
    if (conn->type == RX_SERVER_CONNECTION && conn->service->destroyConnProc)
      (*conn->service->destroyConnProc)(conn);

    /* Notify the security module that this connection is being destroyed */
    RXS_DestroyConnection(conn->securityObject, conn);

    /* If this is the last connection using the rx_peer struct, set its
     * idle time to now. rxi_ReapConnections will reap it if it's still
     * idle (refCount == 0) after rx_idlePeerTime (60 seconds) have passed.
     */
    MUTEX_ENTER(&rx_peerHashTable_lock);
    if (--conn->peer->refCount <= 0) {
        conn->peer->idleWhen = clock_Sec();
        if (conn->peer->refCount < 0) {
            conn->peer->refCount = 0; 
            MUTEX_ENTER(&rx_stats_mutex);
            rxi_lowPeerRefCount ++;
            MUTEX_EXIT(&rx_stats_mutex);
        }
    }
    MUTEX_EXIT(&rx_peerHashTable_lock);

    MUTEX_ENTER(&rx_stats_mutex);
    if (conn->type == RX_SERVER_CONNECTION)
      rx_stats.nServerConns--;
    else
      rx_stats.nClientConns--;
    MUTEX_EXIT(&rx_stats_mutex);

#ifndef KERNEL
    if (conn->specific) {
        int i;
        for (i = 0 ; i < conn->nSpecific ; i++) {
            if (conn->specific[i] && rxi_keyCreate_destructor[i])
                (*rxi_keyCreate_destructor[i])(conn->specific[i]);
            conn->specific[i] = NULL;
        }
        free(conn->specific);
    }
    conn->specific = NULL;
    conn->nSpecific = 0;
#endif /* !KERNEL */

    MUTEX_DESTROY(&conn->conn_call_lock);
    MUTEX_DESTROY(&conn->conn_data_lock);
    CV_DESTROY(&conn->conn_call_cv);
        
    rxi_FreeConnection(conn);
}

/* Destroy the specified connection */
void rxi_DestroyConnection(register struct rx_connection *conn)
{
    MUTEX_ENTER(&rx_connHashTable_lock);
    rxi_DestroyConnectionNoLock(conn);
    /* conn should be at the head of the cleanup list */
    if (conn == rx_connCleanup_list) {
        rx_connCleanup_list = rx_connCleanup_list->next;
        MUTEX_EXIT(&rx_connHashTable_lock);
        rxi_CleanupConnection(conn);
    }
#ifdef RX_ENABLE_LOCKS
    else {
        MUTEX_EXIT(&rx_connHashTable_lock);
    }
#endif /* RX_ENABLE_LOCKS */
}
    
static void rxi_DestroyConnectionNoLock(register struct rx_connection *conn)
{
    register struct rx_connection **conn_ptr;
    register int havecalls = 0;
    struct rx_packet *packet;
    int i;
    SPLVAR;

    clock_NewTime();

    NETPRI;
    MUTEX_ENTER(&conn->conn_data_lock);
    if (conn->refCount > 0)
        conn->refCount--;
    else {
        MUTEX_ENTER(&rx_stats_mutex);
        rxi_lowConnRefCount++;
        MUTEX_EXIT(&rx_stats_mutex);
    }

    if ((conn->refCount > 0) || (conn->flags & RX_CONN_BUSY)) {
        /* Busy; wait till the last guy before proceeding */
        MUTEX_EXIT(&conn->conn_data_lock);
        USERPRI;
        return;
    }

    /* If the client previously called rx_NewCall, but it is still
     * waiting, treat this as a running call, and wait to destroy the
     * connection later when the call completes. */
    if ((conn->type == RX_CLIENT_CONNECTION) &&
        (conn->flags & RX_CONN_MAKECALL_WAITING)) {
        conn->flags |= RX_CONN_DESTROY_ME;
        MUTEX_EXIT(&conn->conn_data_lock);
        USERPRI;
        return;
    }
    MUTEX_EXIT(&conn->conn_data_lock);

    /* Check for extant references to this connection */
    for (i = 0; i<RX_MAXCALLS; i++) {
        register struct rx_call *call = conn->call[i];
        if (call) {
            havecalls = 1;
            if (conn->type == RX_CLIENT_CONNECTION) {
                MUTEX_ENTER(&call->lock);
                if (call->delayedAckEvent) {
                    /* Push the final acknowledgment out now--there
                     * won't be a subsequent call to acknowledge the
                     * last reply packets */
                    rxevent_Cancel(call->delayedAckEvent, call,
                                   RX_CALL_REFCOUNT_DELAY);
                    if (call->state == RX_STATE_PRECALL ||
                        call->state == RX_STATE_ACTIVE) {
                        rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);
                    } else {
                        rxi_AckAll(NULL, call, 0);
                    }
                }
                MUTEX_EXIT(&call->lock);
            }
        }
    }
#ifdef RX_ENABLE_LOCKS
    if (!havecalls) {
        if (MUTEX_TRYENTER(&conn->conn_data_lock)) {
            MUTEX_EXIT(&conn->conn_data_lock);
        }
        else {
            /* Someone is accessing a packet right now. */
            havecalls = 1;
        }
    }
#endif /* RX_ENABLE_LOCKS */

    if (havecalls) {
        /* Don't destroy the connection if there are any call
         * structures still in use */
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->flags |= RX_CONN_DESTROY_ME;
        MUTEX_EXIT(&conn->conn_data_lock);
        USERPRI;
        return;
    }

    if (conn->delayedAbortEvent) {
        rxevent_Cancel(conn->delayedAbortEvent, (struct rx_call *)0, 0);
        packet = rxi_AllocPacket(RX_PACKET_CLASS_SPECIAL);
        if (packet) {
            MUTEX_ENTER(&conn->conn_data_lock);
            rxi_SendConnectionAbort(conn, packet, 0, 1);
            MUTEX_EXIT(&conn->conn_data_lock);
            rxi_FreePacket(packet);
        }
    }

    /* Remove from connection hash table before proceeding */
    conn_ptr = & rx_connHashTable[ CONN_HASH(peer->host, peer->port, conn->cid,
                                             conn->epoch, conn->type) ];
    for ( ; *conn_ptr; conn_ptr = &(*conn_ptr)->next) {
        if (*conn_ptr == conn) {
            *conn_ptr = conn->next;
            break;
        }
    }
    /* if the conn that we are destroying was the last connection, then we
    * clear rxLastConn as well */
    if ( rxLastConn == conn )
        rxLastConn = 0;

    /* Make sure the connection is completely reset before deleting it. */
    /* get rid of pending events that could zap us later */
    if (conn->challengeEvent)
        rxevent_Cancel(conn->challengeEvent, (struct rx_call*)0, 0);
    if (conn->checkReachEvent)
        rxevent_Cancel(conn->checkReachEvent, (struct rx_call*)0, 0);
 
    /* Add the connection to the list of destroyed connections that
     * need to be cleaned up. This is necessary to avoid deadlocks
     * in the routines we call to inform others that this connection is
     * being destroyed. */
    conn->next = rx_connCleanup_list;
    rx_connCleanup_list = conn;
}

/* Externally available version */
void rx_DestroyConnection(register struct rx_connection *conn) 
{
    SPLVAR;

    NETPRI;
    AFS_RXGLOCK();
    rxi_DestroyConnection (conn);
    AFS_RXGUNLOCK();
    USERPRI;
}

/* Start a new rx remote procedure call, on the specified connection.
 * If wait is set to 1, wait for a free call channel; otherwise return
 * 0.  Maxtime gives the maximum number of seconds this call may take,
 * after rx_MakeCall returns.  After this time interval, a call to any
 * of rx_SendData, rx_ReadData, etc. will fail with RX_CALL_TIMEOUT.
 * For fine grain locking, we hold the conn_call_lock in order to 
 * to ensure that we don't get signalle after we found a call in an active
 * state and before we go to sleep.
 */
struct rx_call *rx_NewCall(register struct rx_connection *conn)
{
    register int i;
    register struct rx_call *call;
    struct clock queueTime;
    SPLVAR;

    clock_NewTime();
    dpf (("rx_MakeCall(conn %x)\n", conn));

    NETPRI;
    clock_GetTime(&queueTime);
    AFS_RXGLOCK();
    MUTEX_ENTER(&conn->conn_call_lock);

    /*
     * Check if there are others waiting for a new call.
     * If so, let them go first to avoid starving them.
     * This is a fairly simple scheme, and might not be
     * a complete solution for large numbers of waiters.
     */
    if (conn->makeCallWaiters) {
#ifdef  RX_ENABLE_LOCKS
        CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
#else
        osi_rxSleep(conn);
#endif
    }

    for (;;) {
        for (i=0; i<RX_MAXCALLS; i++) {
            call = conn->call[i];
            if (call) {
                MUTEX_ENTER(&call->lock);
                if (call->state == RX_STATE_DALLY) {
                    rxi_ResetCall(call, 0);
                    (*call->callNumber)++;
                    break;
                }
                MUTEX_EXIT(&call->lock);
            }
            else {
                call = rxi_NewCall(conn, i);
                break;
            }
        }
        if (i < RX_MAXCALLS) {
            break;
        }
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->flags |= RX_CONN_MAKECALL_WAITING;
        MUTEX_EXIT(&conn->conn_data_lock);

        conn->makeCallWaiters++;
#ifdef  RX_ENABLE_LOCKS
        CV_WAIT(&conn->conn_call_cv, &conn->conn_call_lock);
#else
        osi_rxSleep(conn);
#endif
        conn->makeCallWaiters--;
    }
    /*
     * Wake up anyone else who might be giving us a chance to
     * run (see code above that avoids resource starvation).
     */
#ifdef  RX_ENABLE_LOCKS
    CV_BROADCAST(&conn->conn_call_cv);
#else
    osi_rxWakeup(conn);
#endif

    CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);

    /* Client is initially in send mode */
    call->state = RX_STATE_ACTIVE;
    call->mode = RX_MODE_SENDING;

    /* remember start time for call in case we have hard dead time limit */
    call->queueTime = queueTime;
    clock_GetTime(&call->startTime);
    hzero(call->bytesSent);
    hzero(call->bytesRcvd);

    /* Turn on busy protocol. */
    rxi_KeepAliveOn(call);

    MUTEX_EXIT(&call->lock);
    MUTEX_EXIT(&conn->conn_call_lock);
    AFS_RXGUNLOCK();
    USERPRI;

#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
    /* Now, if TQ wasn't cleared earlier, do it now. */
    AFS_RXGLOCK();
    MUTEX_ENTER(&call->lock);
    while (call->flags & RX_CALL_TQ_BUSY) {
        call->flags |= RX_CALL_TQ_WAIT;
#ifdef RX_ENABLE_LOCKS
        CV_WAIT(&call->cv_tq, &call->lock);
#else /* RX_ENABLE_LOCKS */
        osi_rxSleep(&call->tq);
#endif /* RX_ENABLE_LOCKS */
    }
    if (call->flags & RX_CALL_TQ_CLEARME) {
        rxi_ClearTransmitQueue(call, 0);
        queue_Init(&call->tq);
    }
    MUTEX_EXIT(&call->lock);
    AFS_RXGUNLOCK();
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */

    return call;
}

int rxi_HasActiveCalls(register struct rx_connection *aconn)
{
    register int i;
    register struct rx_call *tcall;
    SPLVAR;

    NETPRI;
    for(i=0; i<RX_MAXCALLS; i++) {
      if ((tcall = aconn->call[i])) {
        if ((tcall->state == RX_STATE_ACTIVE) 
            || (tcall->state == RX_STATE_PRECALL)) {
          USERPRI;
          return 1;
        }
      }
    }
    USERPRI;
    return 0;
}

int rxi_GetCallNumberVector(register struct rx_connection *aconn, 
        register afs_int32 *aint32s)
{
    register int i;
    register struct rx_call *tcall;
    SPLVAR;

    NETPRI;
    for(i=0; i<RX_MAXCALLS; i++) {
        if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
            aint32s[i] = aconn->callNumber[i]+1;
        else
            aint32s[i] = aconn->callNumber[i];
    }
    USERPRI;
    return 0;
}

int rxi_SetCallNumberVector(register struct rx_connection *aconn, 
        register afs_int32 *aint32s)
{
    register int i;
    register struct rx_call *tcall;
    SPLVAR;

    NETPRI;
    for(i=0; i<RX_MAXCALLS; i++) {
        if ((tcall = aconn->call[i]) && (tcall->state == RX_STATE_DALLY))
            aconn->callNumber[i] = aint32s[i] - 1;
        else
            aconn->callNumber[i] = aint32s[i];
    }
    USERPRI;
    return 0;
}

/* Advertise a new service.  A service is named locally by a UDP port
 * number plus a 16-bit service id.  Returns (struct rx_service *) 0
 * on a failure. 
 *
     char *serviceName;  Name for identification purposes (e.g. the
                         service name might be used for probing for
                         statistics) */
struct rx_service *rx_NewService(u_short port, u_short serviceId, 
        char *serviceName, 
        struct rx_securityClass **securityObjects,
        int nSecurityObjects, afs_int32 (*serviceProc)(struct rx_call *acall))
{    
    osi_socket socket = OSI_NULLSOCKET;
    register struct rx_service *tservice;    
    register int i;
    SPLVAR;

    clock_NewTime();

    if (serviceId == 0) {
        (osi_Msg "rx_NewService:  service id for service %s is not non-zero.\n",
         serviceName);
        return 0;
    }
    if (port == 0) {
        if (rx_port == 0) {
            (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);
            return 0;
        }
        port = rx_port;
        socket = rx_socket;
    }

    tservice = rxi_AllocService();
    NETPRI;
    AFS_RXGLOCK();
    for (i = 0; i<RX_MAX_SERVICES; i++) {
        register struct rx_service *service = rx_services[i];
        if (service) {
            if (port == service->servicePort) {
                if (service->serviceId == serviceId) {
                    /* The identical service has already been
                     * installed; if the caller was intending to
                     * change the security classes used by this
                     * service, he/she loses. */
                    (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);
                    AFS_RXGUNLOCK();
                    USERPRI;
                    rxi_FreeService(tservice);
                    return service;
                }
                /* Different service, same port: re-use the socket
                 * which is bound to the same port */
                socket = service->socket;
            }
        } else {
            if (socket == OSI_NULLSOCKET) {
                /* If we don't already have a socket (from another
                 * service on same port) get a new one */
                socket = rxi_GetUDPSocket(port);
                if (socket == OSI_NULLSOCKET) {
                    AFS_RXGUNLOCK();
                    USERPRI;
                    rxi_FreeService(tservice);
                    return 0;
                }
            }
            service = tservice;
            service->socket = socket;
            service->servicePort = port;
            service->serviceId = serviceId;
            service->serviceName = serviceName;
            service->nSecurityObjects = nSecurityObjects;
            service->securityObjects = securityObjects;
            service->minProcs = 0;
            service->maxProcs = 1;
            service->idleDeadTime = 60;
            service->connDeadTime = rx_connDeadTime;
            service->executeRequestProc = serviceProc;
            service->checkReach = 0;
            rx_services[i] = service;   /* not visible until now */
            AFS_RXGUNLOCK();
            USERPRI;
            return service;
        }
    }
    AFS_RXGUNLOCK();
    USERPRI;
    rxi_FreeService(tservice);
    (osi_Msg "rx_NewService: cannot support > %d services\n", RX_MAX_SERVICES);
    return 0;
}

/* Generic request processing loop. This routine should be called
 * by the implementation dependent rx_ServerProc. If socketp is
 * non-null, it will be set to the file descriptor that this thread
 * is now listening on. If socketp is null, this routine will never
 * returns. */
void rxi_ServerProc(int threadID, struct rx_call *newcall, osi_socket *socketp)
{
    register struct rx_call *call;
    register afs_int32 code;
    register struct rx_service *tservice = NULL;

    for (;;) {
        if (newcall) {
            call = newcall;
            newcall = NULL;
        } else {
            call = rx_GetCall(threadID, tservice, socketp);
            if (socketp && *socketp != OSI_NULLSOCKET) {
                /* We are now a listener thread */
                return;
            }
        }

        /* if server is restarting( typically smooth shutdown) then do not
         * allow any new calls.
         */

        if ( rx_tranquil && (call != NULL) ) {
            SPLVAR;

            NETPRI;
            AFS_RXGLOCK();
            MUTEX_ENTER(&call->lock);

            rxi_CallError(call, RX_RESTARTING);
            rxi_SendCallAbort(call, (struct rx_packet *) 0, 0, 0);

            MUTEX_EXIT(&call->lock);
            AFS_RXGUNLOCK();
            USERPRI;
        }

#ifdef  KERNEL
        if (afs_termState == AFSOP_STOP_RXCALLBACK) {
#ifdef RX_ENABLE_LOCKS
            AFS_GLOCK();
#endif /* RX_ENABLE_LOCKS */
            afs_termState = AFSOP_STOP_AFS;
            afs_osi_Wakeup(&afs_termState);
#ifdef RX_ENABLE_LOCKS
            AFS_GUNLOCK();
#endif /* RX_ENABLE_LOCKS */
            return;
        }
#endif

        tservice = call->conn->service;

        if (tservice->beforeProc) (*tservice->beforeProc)(call);

        code = call->conn->service->executeRequestProc(call);

        if (tservice->afterProc) (*tservice->afterProc)(call, code);

        rx_EndCall(call, code);
        MUTEX_ENTER(&rx_stats_mutex);
        rxi_nCalls++;
        MUTEX_EXIT(&rx_stats_mutex);
    }
}


void rx_WakeupServerProcs(void)
{
    struct rx_serverQueueEntry *np, *tqp;
    SPLVAR;

    NETPRI;
    AFS_RXGLOCK();
    MUTEX_ENTER(&rx_serverPool_lock);

#ifdef RX_ENABLE_LOCKS
    if (rx_waitForPacket)
        CV_BROADCAST(&rx_waitForPacket->cv);
#else /* RX_ENABLE_LOCKS */
    if (rx_waitForPacket)
        osi_rxWakeup(rx_waitForPacket);
#endif /* RX_ENABLE_LOCKS */
    MUTEX_ENTER(&freeSQEList_lock);
    for (np = rx_FreeSQEList; np; np = tqp) {
      tqp = *(struct rx_serverQueueEntry **)np;
#ifdef RX_ENABLE_LOCKS
      CV_BROADCAST(&np->cv);
#else /* RX_ENABLE_LOCKS */
      osi_rxWakeup(np);
#endif /* RX_ENABLE_LOCKS */
    }
    MUTEX_EXIT(&freeSQEList_lock);
    for (queue_Scan(&rx_idleServerQueue, np, tqp, rx_serverQueueEntry)) {
#ifdef RX_ENABLE_LOCKS
      CV_BROADCAST(&np->cv);
#else /* RX_ENABLE_LOCKS */
      osi_rxWakeup(np);
#endif /* RX_ENABLE_LOCKS */
    }
    MUTEX_EXIT(&rx_serverPool_lock);
    AFS_RXGUNLOCK();
    USERPRI;
}

/* meltdown:
 * One thing that seems to happen is that all the server threads get
 * tied up on some empty or slow call, and then a whole bunch of calls
 * arrive at once, using up the packet pool, so now there are more 
 * empty calls.  The most critical resources here are server threads
 * and the free packet pool.  The "doreclaim" code seems to help in
 * general.  I think that eventually we arrive in this state: there
 * are lots of pending calls which do have all their packets present,
 * so they won't be reclaimed, are multi-packet calls, so they won't
 * be scheduled until later, and thus are tying up most of the free 
 * packet pool for a very long time.
 * future options:
 * 1.  schedule multi-packet calls if all the packets are present.  
 * Probably CPU-bound operation, useful to return packets to pool. 
 * Do what if there is a full window, but the last packet isn't here?
 * 3.  preserve one thread which *only* runs "best" calls, otherwise
 * it sleeps and waits for that type of call.
 * 4.  Don't necessarily reserve a whole window for each thread.  In fact, 
 * the current dataquota business is badly broken.  The quota isn't adjusted
 * to reflect how many packets are presently queued for a running call.
 * So, when we schedule a queued call with a full window of packets queued
 * up for it, that *should* free up a window full of packets for other 2d-class
 * calls to be able to use from the packet pool.  But it doesn't.
 *
 * NB.  Most of the time, this code doesn't run -- since idle server threads
 * sit on the idle server queue and are assigned by "...ReceivePacket" as soon
 * as a new call arrives.
 */
/* Sleep until a call arrives.  Returns a pointer to the call, ready
 * for an rx_Read. */
#ifdef RX_ENABLE_LOCKS
struct rx_call *rx_GetCall(int tno, struct rx_service *cur_service, osi_socket *socketp)
{
    struct rx_serverQueueEntry *sq;
    register struct rx_call *call = (struct rx_call *) 0;
    struct rx_service *service = NULL;
    SPLVAR;

    MUTEX_ENTER(&freeSQEList_lock);

    if ((sq = rx_FreeSQEList)) {
        rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
        MUTEX_EXIT(&freeSQEList_lock);
    } else {    /* otherwise allocate a new one and return that */
        MUTEX_EXIT(&freeSQEList_lock);
        sq = (struct rx_serverQueueEntry *) rxi_Alloc(sizeof(struct rx_serverQueueEntry));
        MUTEX_INIT(&sq->lock, "server Queue lock",MUTEX_DEFAULT,0);
        CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
    }

    MUTEX_ENTER(&rx_serverPool_lock);
    if (cur_service != NULL) {
        ReturnToServerPool(cur_service);
    }
    while (1) {
        if (queue_IsNotEmpty(&rx_incomingCallQueue)) {
            register struct rx_call *tcall, *ncall, *choice2 = NULL;

            /* Scan for eligible incoming calls.  A call is not eligible
             * if the maximum number of calls for its service type are
             * already executing */
            /* One thread will process calls FCFS (to prevent starvation),
             * while the other threads may run ahead looking for calls which
             * have all their input data available immediately.  This helps 
             * keep threads from blocking, waiting for data from the client. */
            for (queue_Scan(&rx_incomingCallQueue, tcall, ncall, rx_call)) {
                service = tcall->conn->service;
                if (!QuotaOK(service)) {
                    continue;
                }
                if (tno==rxi_fcfs_thread_num || !tcall->queue_item_header.next  ) {
                    /* If we're the fcfs thread , then  we'll just use 
                     * this call. If we haven't been able to find an optimal 
                     * choice, and we're at the end of the list, then use a 
                     * 2d choice if one has been identified.  Otherwise... */
                    call = (choice2 ? choice2 : tcall);
                    service = call->conn->service;
                } else if (!queue_IsEmpty(&tcall->rq)) {
                    struct rx_packet *rp;
                    rp = queue_First(&tcall->rq, rx_packet);
                    if (rp->header.seq == 1) {
                        if (!meltdown_1pkt ||
                            (rp->header.flags & RX_LAST_PACKET)) {
                            call = tcall;
                        } else if (rxi_2dchoice && !choice2 &&
                                   !(tcall->flags & RX_CALL_CLEARED) &&
                                   (tcall->rprev > rxi_HardAckRate)) {
                            choice2 = tcall;
                        } else rxi_md2cnt++;
                    }
                }
                if (call) {
                    break;
                } else {
                    ReturnToServerPool(service);
                }
            }
        }

        if (call) {
            queue_Remove(call);
            MUTEX_EXIT(&rx_serverPool_lock);
            MUTEX_ENTER(&call->lock);

            if (call->state != RX_STATE_PRECALL || call->error) {
                MUTEX_EXIT(&call->lock);
                MUTEX_ENTER(&rx_serverPool_lock);
                ReturnToServerPool(service);
                call = NULL;
                continue;
            }

            if (queue_IsEmpty(&call->rq) ||
                queue_First(&call->rq, rx_packet)->header.seq != 1)
                rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);

            CLEAR_CALL_QUEUE_LOCK(call);
            call->flags &= ~RX_CALL_WAIT_PROC;
            MUTEX_ENTER(&rx_stats_mutex);
            rx_nWaiting--;
            MUTEX_EXIT(&rx_stats_mutex);
            break;
        }
        else {
            /* If there are no eligible incoming calls, add this process
             * to the idle server queue, to wait for one */
            sq->newcall = 0;
            sq->tno = tno;
            if (socketp) {
                *socketp = OSI_NULLSOCKET;
            }
            sq->socketp = socketp;
            queue_Append(&rx_idleServerQueue, sq);
#ifndef AFS_AIX41_ENV
            rx_waitForPacket = sq;
#endif /* AFS_AIX41_ENV */
            do {
                CV_WAIT(&sq->cv, &rx_serverPool_lock);
#ifdef  KERNEL
                if (afs_termState == AFSOP_STOP_RXCALLBACK) {
                    MUTEX_EXIT(&rx_serverPool_lock);
                    return (struct rx_call *)0;
                }
#endif
            } while (!(call = sq->newcall) &&
                     !(socketp && *socketp != OSI_NULLSOCKET));
            MUTEX_EXIT(&rx_serverPool_lock);
            if (call) {
                MUTEX_ENTER(&call->lock);
            }
            break;
        }
    }

    MUTEX_ENTER(&freeSQEList_lock);
    *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
    rx_FreeSQEList = sq;
    MUTEX_EXIT(&freeSQEList_lock);

    if (call) {
        clock_GetTime(&call->startTime);
        call->state = RX_STATE_ACTIVE;
        call->mode = RX_MODE_RECEIVING;

        rxi_calltrace(RX_CALL_START, call);
        dpf(("rx_GetCall(port=%d, service=%d) ==> call %x\n", 
             call->conn->service->servicePort, 
             call->conn->service->serviceId, call));

        CALL_HOLD(call, RX_CALL_REFCOUNT_BEGIN);
        MUTEX_EXIT(&call->lock);
    } else {
        dpf(("rx_GetCall(socketp=0x%x, *socketp=0x%x)\n", socketp, *socketp));
    }

    return call;
}
#else /* RX_ENABLE_LOCKS */
struct rx_call *rx_GetCall(int tno, struct rx_service *cur_service, osi_socket *socketp)
{
    struct rx_serverQueueEntry *sq;
    register struct rx_call *call = (struct rx_call *) 0, *choice2;
    struct rx_service *service = NULL;
    SPLVAR;

    NETPRI;
    AFS_RXGLOCK();
    MUTEX_ENTER(&freeSQEList_lock);

    if ((sq = rx_FreeSQEList)) {
        rx_FreeSQEList = *(struct rx_serverQueueEntry **)sq;
        MUTEX_EXIT(&freeSQEList_lock);
    } else {    /* otherwise allocate a new one and return that */
        MUTEX_EXIT(&freeSQEList_lock);
        sq = (struct rx_serverQueueEntry *) rxi_Alloc(sizeof(struct rx_serverQueueEntry));
        MUTEX_INIT(&sq->lock, "server Queue lock",MUTEX_DEFAULT,0);
        CV_INIT(&sq->cv, "server Queue lock", CV_DEFAULT, 0);
    }
    MUTEX_ENTER(&sq->lock);

    if (cur_service != NULL) {
        cur_service->nRequestsRunning--;
        if (cur_service->nRequestsRunning < cur_service->minProcs)
            rxi_minDeficit++;
        rxi_availProcs++;
    }
    if (queue_IsNotEmpty(&rx_incomingCallQueue)) {
        register struct rx_call *tcall, *ncall;
        /* Scan for eligible incoming calls.  A call is not eligible
         * if the maximum number of calls for its service type are
         * already executing */
        /* One thread will process calls FCFS (to prevent starvation),
         * while the other threads may run ahead looking for calls which
         * have all their input data available immediately.  This helps 
         * keep threads from blocking, waiting for data from the client. */
        choice2 = (struct rx_call *) 0;
        for (queue_Scan(&rx_incomingCallQueue, tcall, ncall, rx_call)) {
          service = tcall->conn->service;
          if (QuotaOK(service)) {
             if (tno==rxi_fcfs_thread_num || !tcall->queue_item_header.next  ) {
                 /* If we're the fcfs thread, then  we'll just use 
                  * this call. If we haven't been able to find an optimal 
                  * choice, and we're at the end of the list, then use a 
                  * 2d choice if one has been identified.  Otherwise... */
                 call = (choice2 ? choice2 : tcall);
                 service = call->conn->service;
             } else if (!queue_IsEmpty(&tcall->rq)) {
                 struct rx_packet *rp;
                 rp = queue_First(&tcall->rq, rx_packet);
                 if (rp->header.seq == 1
                     && (!meltdown_1pkt ||
                         (rp->header.flags & RX_LAST_PACKET))) {
                     call = tcall;
                 } else if (rxi_2dchoice && !choice2 &&
                            !(tcall->flags & RX_CALL_CLEARED) &&
                            (tcall->rprev > rxi_HardAckRate)) {
                     choice2 = tcall;
                 } else rxi_md2cnt++;
             }
          }
          if (call) 
             break;
        }
      }

    if (call) {
        queue_Remove(call);
        /* we can't schedule a call if there's no data!!! */
        /* send an ack if there's no data, if we're missing the
         * first packet, or we're missing something between first 
         * and last -- there's a "hole" in the incoming data. */
        if (queue_IsEmpty(&call->rq) ||
            queue_First(&call->rq, rx_packet)->header.seq != 1 ||
            call->rprev != queue_Last(&call->rq, rx_packet)->header.seq)
          rxi_SendAck(call, 0, 0, RX_ACK_DELAY, 0);

        call->flags &= (~RX_CALL_WAIT_PROC);
        service->nRequestsRunning++;
        /* just started call in minProcs pool, need fewer to maintain
         * guarantee */
        if (service->nRequestsRunning <= service->minProcs)
            rxi_minDeficit--;
        rxi_availProcs--;
        rx_nWaiting--;
        /* MUTEX_EXIT(&call->lock); */
    }
    else {
        /* If there are no eligible incoming calls, add this process
         * to the idle server queue, to wait for one */
        sq->newcall = 0;
        if (socketp) {
            *socketp = OSI_NULLSOCKET;
        }
        sq->socketp = socketp;
        queue_Append(&rx_idleServerQueue, sq);
        do {
            osi_rxSleep(sq);
#ifdef  KERNEL
                if (afs_termState == AFSOP_STOP_RXCALLBACK) {
                    AFS_RXGUNLOCK();
                    USERPRI;
                    return (struct rx_call *)0;
                }
#endif
        } while (!(call = sq->newcall) &&
                 !(socketp && *socketp != OSI_NULLSOCKET));
    }
    MUTEX_EXIT(&sq->lock);

    MUTEX_ENTER(&freeSQEList_lock);
    *(struct rx_serverQueueEntry **)sq = rx_FreeSQEList;
    rx_FreeSQEList = sq;
    MUTEX_EXIT(&freeSQEList_lock);

    if (call) {
        clock_GetTime(&call->startTime);
        call->state = RX_STATE_ACTIVE;
        call->mode = RX_MODE_RECEIVING;

        rxi_calltrace(RX_CALL_START, call);
        dpf(("rx_GetCall(port=%d, service=%d) ==> call %x\n", 
         call->conn->service->servicePort, 
         call->conn->service->serviceId, call));
    } else {
        dpf(("rx_GetCall(socketp=0x%x, *socketp=0x%x)\n", socketp, *socketp));
    }

    AFS_RXGUNLOCK();
    USERPRI;

    return call;
}
#endif /* RX_ENABLE_LOCKS */



/* Establish a procedure to be called when a packet arrives for a
 * call.  This routine will be called at most once after each call,
 * and will also be called if there is an error condition on the or
 * the call is complete.  Used by multi rx to build a selection
 * function which determines which of several calls is likely to be a
 * good one to read from.  
 * NOTE: the way this is currently implemented it is probably only a
 * good idea to (1) use it immediately after a newcall (clients only)
 * and (2) only use it once.  Other uses currently void your warranty
 */
void rx_SetArrivalProc(register struct rx_call *call, 
        register VOID (*proc)(register struct rx_call *call,
        register struct multi_handle *mh, register int index),
        register VOID *handle, register VOID *arg)
{
    call->arrivalProc = proc;
    call->arrivalProcHandle = handle;
    call->arrivalProcArg = arg;
}

/* Call is finished (possibly prematurely).  Return rc to the peer, if
 * appropriate, and return the final error code from the conversation
 * to the caller */

afs_int32 rx_EndCall(register struct rx_call *call, afs_int32 rc)
{
    register struct rx_connection *conn = call->conn;
    register struct rx_service *service;
    register struct rx_packet *tp; /* Temporary packet pointer */
    register struct rx_packet *nxp; /* Next packet pointer, for queue_Scan */
    afs_int32 error;
    SPLVAR;

    dpf(("rx_EndCall(call %x)\n", call));

    NETPRI;
    AFS_RXGLOCK();
    MUTEX_ENTER(&call->lock);

    if (rc == 0 && call->error == 0) {
        call->abortCode = 0;
        call->abortCount = 0;
    }

    call->arrivalProc = (VOID (*)()) 0;
    if (rc && call->error == 0) {
        rxi_CallError(call, rc);
        /* Send an abort message to the peer if this error code has
         * only just been set.  If it was set previously, assume the
         * peer has already been sent the error code or will request it 
         */
        rxi_SendCallAbort(call, (struct rx_packet *) 0, 0, 0);
    }
    if (conn->type == RX_SERVER_CONNECTION) {
        /* Make sure reply or at least dummy reply is sent */
        if (call->mode == RX_MODE_RECEIVING) {
            rxi_WriteProc(call, 0, 0);
        }
        if (call->mode == RX_MODE_SENDING) {
            rxi_FlushWrite(call);
        }
        service = conn->service;
        rxi_calltrace(RX_CALL_END, call);
        /* Call goes to hold state until reply packets are acknowledged */
        if (call->tfirst + call->nSoftAcked < call->tnext) {
            call->state = RX_STATE_HOLD;
        } else {
            call->state = RX_STATE_DALLY;
            rxi_ClearTransmitQueue(call, 0);
            rxevent_Cancel(call->resendEvent, call, RX_CALL_REFCOUNT_RESEND);
            rxevent_Cancel(call->keepAliveEvent, call, RX_CALL_REFCOUNT_ALIVE);
        }
    }
    else { /* Client connection */
        char dummy;
        /* Make sure server receives input packets, in the case where
         * no reply arguments are expected */
        if ((call->mode == RX_MODE_SENDING)
         || (call->mode == RX_MODE_RECEIVING && call->rnext == 1)) {
            (void) rxi_ReadProc(call, &dummy, 1);
        }

        /* If we had an outstanding delayed ack, be nice to the server
         * and force-send it now.
         */
        if (call->delayedAckEvent) {
            rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
            call->delayedAckEvent = NULL;
            rxi_SendDelayedAck(NULL, call, NULL);
        }

        /* We need to release the call lock since it's lower than the
         * conn_call_lock and we don't want to hold the conn_call_lock
         * over the rx_ReadProc call. The conn_call_lock needs to be held
         * here for the case where rx_NewCall is perusing the calls on
         * the connection structure. We don't want to signal until
         * rx_NewCall is in a stable state. Otherwise, rx_NewCall may
         * have checked this call, found it active and by the time it
         * goes to sleep, will have missed the signal.
         */
        MUTEX_EXIT(&call->lock);
        MUTEX_ENTER(&conn->conn_call_lock);
        MUTEX_ENTER(&call->lock);
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->flags |= RX_CONN_BUSY;
        if (conn->flags & RX_CONN_MAKECALL_WAITING) {
            conn->flags &= (~RX_CONN_MAKECALL_WAITING);
            MUTEX_EXIT(&conn->conn_data_lock);
#ifdef  RX_ENABLE_LOCKS
            CV_BROADCAST(&conn->conn_call_cv);
#else
            osi_rxWakeup(conn);
#endif
        }
#ifdef RX_ENABLE_LOCKS
        else {
            MUTEX_EXIT(&conn->conn_data_lock);
        }
#endif /* RX_ENABLE_LOCKS */
        call->state = RX_STATE_DALLY;
    }
    error = call->error;

    /* currentPacket, nLeft, and NFree must be zeroed here, because
     * ResetCall cannot: ResetCall may be called at splnet(), in the
     * kernel version, and may interrupt the macros rx_Read or
     * rx_Write, which run at normal priority for efficiency. */
    if (call->currentPacket) {
        rxi_FreePacket(call->currentPacket);
        call->currentPacket = (struct rx_packet *) 0;
        call->nLeft = call->nFree = call->curlen = 0;
    }
    else
        call->nLeft = call->nFree = call->curlen = 0;

    /* Free any packets from the last call to ReadvProc/WritevProc */
    for (queue_Scan(&call->iovq, tp, nxp, rx_packet)) {
        queue_Remove(tp);
        rxi_FreePacket(tp);
    }

    CALL_RELE(call, RX_CALL_REFCOUNT_BEGIN);
    MUTEX_EXIT(&call->lock);
    if (conn->type == RX_CLIENT_CONNECTION) {
        MUTEX_EXIT(&conn->conn_call_lock);
        conn->flags &= ~RX_CONN_BUSY;
    }
    AFS_RXGUNLOCK();
    USERPRI;
    /*
     * Map errors to the local host's errno.h format.
     */
    error = ntoh_syserr_conv(error);
    return error;
}

#if !defined(KERNEL)

/* Call this routine when shutting down a server or client (especially
 * clients).  This will allow Rx to gracefully garbage collect server
 * connections, and reduce the number of retries that a server might
 * make to a dead client.
 * This is not quite right, since some calls may still be ongoing and
 * we can't lock them to destroy them. */
void rx_Finalize(void)
{
    register struct rx_connection **conn_ptr, **conn_end;

    INIT_PTHREAD_LOCKS
    LOCK_RX_INIT
    if (rxinit_status == 1) {
        UNLOCK_RX_INIT
        return; /* Already shutdown. */
    }
    rxi_DeleteCachedConnections();
    if (rx_connHashTable) {
        MUTEX_ENTER(&rx_connHashTable_lock);
        for (conn_ptr = &rx_connHashTable[0], 
             conn_end = &rx_connHashTable[rx_hashTableSize]; 
             conn_ptr < conn_end; conn_ptr++) {
            struct rx_connection *conn, *next;
            for (conn = *conn_ptr; conn; conn = next) {
                next = conn->next;
                if (conn->type == RX_CLIENT_CONNECTION) {
                    /* MUTEX_ENTER(&conn->conn_data_lock); when used in kernel */
                    conn->refCount++;
                    /* MUTEX_EXIT(&conn->conn_data_lock); when used in kernel */
#ifdef RX_ENABLE_LOCKS
                    rxi_DestroyConnectionNoLock(conn);
#else /* RX_ENABLE_LOCKS */
                    rxi_DestroyConnection(conn);
#endif /* RX_ENABLE_LOCKS */
                }
            }
        }
#ifdef RX_ENABLE_LOCKS
        while (rx_connCleanup_list) {
            struct rx_connection *conn;
            conn = rx_connCleanup_list;
            rx_connCleanup_list = rx_connCleanup_list->next;
            MUTEX_EXIT(&rx_connHashTable_lock);
            rxi_CleanupConnection(conn);
            MUTEX_ENTER(&rx_connHashTable_lock);
        }
        MUTEX_EXIT(&rx_connHashTable_lock);
#endif /* RX_ENABLE_LOCKS */
    }
    rxi_flushtrace();

    rxinit_status = 1;
    UNLOCK_RX_INIT
}
#endif

/* if we wakeup packet waiter too often, can get in loop with two
    AllocSendPackets each waking each other up (from ReclaimPacket calls) */
void rxi_PacketsUnWait(void)
{
    if (!rx_waitingForPackets) {
        return;
    }
#ifdef KERNEL
    if (rxi_OverQuota(RX_PACKET_CLASS_SEND)) {
        return;                                     /* still over quota */
    }
#endif /* KERNEL */
    rx_waitingForPackets = 0;
#ifdef  RX_ENABLE_LOCKS
    CV_BROADCAST(&rx_waitingForPackets_cv);
#else
    osi_rxWakeup(&rx_waitingForPackets);
#endif
    return;
}


/* ------------------Internal interfaces------------------------- */

/* Return this process's service structure for the
 * specified socket and service */
struct rx_service *rxi_FindService(register osi_socket socket, 
        register u_short serviceId)
{
    register struct rx_service **sp;    
    for (sp = &rx_services[0]; *sp; sp++) {
        if ((*sp)->serviceId == serviceId && (*sp)->socket == socket) 
          return *sp;
    }
    return 0;
}

/* Allocate a call structure, for the indicated channel of the
 * supplied connection.  The mode and state of the call must be set by
 * the caller. Returns the call with mutex locked. */
struct rx_call *rxi_NewCall(register struct rx_connection *conn,
        register int channel)
{
    register struct rx_call *call;
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
    register struct rx_call *cp;        /* Call pointer temp */
    register struct rx_call *nxp;       /* Next call pointer, for queue_Scan */
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */

    /* Grab an existing call structure, or allocate a new one.
     * Existing call structures are assumed to have been left reset by
     * rxi_FreeCall */
    MUTEX_ENTER(&rx_freeCallQueue_lock);

#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
    /*
     * EXCEPT that the TQ might not yet be cleared out.
     * Skip over those with in-use TQs.
     */
    call = NULL;
    for (queue_Scan(&rx_freeCallQueue, cp, nxp, rx_call)) {
        if (!(cp->flags & RX_CALL_TQ_BUSY)) {
            call = cp;
            break;
        }
    }
    if (call) {
#else /* AFS_GLOBAL_RXLOCK_KERNEL */
    if (queue_IsNotEmpty(&rx_freeCallQueue)) {
        call = queue_First(&rx_freeCallQueue, rx_call);
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        queue_Remove(call);
        MUTEX_ENTER(&rx_stats_mutex);
        rx_stats.nFreeCallStructs--;
        MUTEX_EXIT(&rx_stats_mutex);
        MUTEX_EXIT(&rx_freeCallQueue_lock);
        MUTEX_ENTER(&call->lock);
        CLEAR_CALL_QUEUE_LOCK(call);
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
        /* Now, if TQ wasn't cleared earlier, do it now. */
        if (call->flags & RX_CALL_TQ_CLEARME) {
            rxi_ClearTransmitQueue(call, 0);
            queue_Init(&call->tq);
        }
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        /* Bind the call to its connection structure */
        call->conn = conn;
        rxi_ResetCall(call, 1);
    }
    else {
        call = (struct rx_call *) rxi_Alloc(sizeof(struct rx_call));

        MUTEX_EXIT(&rx_freeCallQueue_lock);
        MUTEX_INIT(&call->lock, "call lock", MUTEX_DEFAULT, NULL);
        MUTEX_ENTER(&call->lock);
        CV_INIT(&call->cv_twind, "call twind", CV_DEFAULT, 0);
        CV_INIT(&call->cv_rq, "call rq", CV_DEFAULT, 0);
        CV_INIT(&call->cv_tq, "call tq", CV_DEFAULT, 0);

        MUTEX_ENTER(&rx_stats_mutex);
        rx_stats.nCallStructs++;
        MUTEX_EXIT(&rx_stats_mutex);
        /* Initialize once-only items */
        queue_Init(&call->tq);
        queue_Init(&call->rq);
        queue_Init(&call->iovq);
        /* Bind the call to its connection structure (prereq for reset) */
        call->conn = conn;
        rxi_ResetCall(call, 1);
    }
    call->channel = channel;
    call->callNumber = &conn->callNumber[channel];
    /* Note that the next expected call number is retained (in
     * conn->callNumber[i]), even if we reallocate the call structure
     */
    conn->call[channel] = call;
    /* if the channel's never been used (== 0), we should start at 1, otherwise
        the call number is valid from the last time this channel was used */
    if (*call->callNumber == 0) *call->callNumber = 1;

    return call;
}

/* A call has been inactive long enough that so we can throw away
 * state, including the call structure, which is placed on the call
 * free list.
 * Call is locked upon entry.
 * haveCTLock set if called from rxi_ReapConnections
 */
#ifdef RX_ENABLE_LOCKS
void rxi_FreeCall(register struct rx_call *call, int haveCTLock)
#else /* RX_ENABLE_LOCKS */
void rxi_FreeCall(register struct rx_call *call)
#endif /* RX_ENABLE_LOCKS */
{
    register int channel = call->channel;
    register struct rx_connection *conn = call->conn;


    if (call->state == RX_STATE_DALLY || call->state == RX_STATE_HOLD)
      (*call->callNumber)++;
    rxi_ResetCall(call, 0);
    call->conn->call[channel] = (struct rx_call *) 0;

    MUTEX_ENTER(&rx_freeCallQueue_lock);
    SET_CALL_QUEUE_LOCK(call, &rx_freeCallQueue_lock);
#ifdef AFS_GLOBAL_RXLOCK_KERNEL
    /* A call may be free even though its transmit queue is still in use.
     * Since we search the call list from head to tail, put busy calls at
     * the head of the list, and idle calls at the tail.
     */
    if (call->flags & RX_CALL_TQ_BUSY)
        queue_Prepend(&rx_freeCallQueue, call);
    else
        queue_Append(&rx_freeCallQueue, call);
#else /* AFS_GLOBAL_RXLOCK_KERNEL */
    queue_Append(&rx_freeCallQueue, call);
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
    MUTEX_ENTER(&rx_stats_mutex);
    rx_stats.nFreeCallStructs++;
    MUTEX_EXIT(&rx_stats_mutex);

    MUTEX_EXIT(&rx_freeCallQueue_lock);
 
    /* Destroy the connection if it was previously slated for
     * destruction, i.e. the Rx client code previously called
     * rx_DestroyConnection (client connections), or
     * rxi_ReapConnections called the same routine (server
     * connections).  Only do this, however, if there are no
     * outstanding calls. Note that for fine grain locking, there appears
     * to be a deadlock in that rxi_FreeCall has a call locked and
     * DestroyConnectionNoLock locks each call in the conn. But note a
     * few lines up where we have removed this call from the conn.
     * If someone else destroys a connection, they either have no
     * call lock held or are going through this section of code.
     */
    if (conn->flags & RX_CONN_DESTROY_ME) {
        MUTEX_ENTER(&conn->conn_data_lock);
        conn->refCount++;
        MUTEX_EXIT(&conn->conn_data_lock);
#ifdef RX_ENABLE_LOCKS
        if (haveCTLock)
            rxi_DestroyConnectionNoLock(conn);
        else
            rxi_DestroyConnection(conn);
#else /* RX_ENABLE_LOCKS */
        rxi_DestroyConnection(conn);
#endif /* RX_ENABLE_LOCKS */
    }
}

afs_int32 rxi_Alloccnt = 0, rxi_Allocsize = 0;
char *rxi_Alloc(register size_t size)
{
    register char *p;

#if defined(AFS_AIX41_ENV) && defined(KERNEL)
    /* Grab the AFS filesystem lock. See afs/osi.h for the lock
     * implementation.
     */
    int glockOwner = ISAFS_GLOCK();
    if (!glockOwner)
        AFS_GLOCK();
#endif
    MUTEX_ENTER(&rx_stats_mutex);
    rxi_Alloccnt++; rxi_Allocsize += size;
    MUTEX_EXIT(&rx_stats_mutex);
#if     (defined(AFS_AIX32_ENV) || defined(AFS_HPUX_ENV)) && !defined(AFS_HPUX100_ENV) && defined(KERNEL)
    if (size > AFS_SMALLOCSIZ) {
        p = (char *) osi_AllocMediumSpace(size);
    } else
        p = (char *) osi_AllocSmall(size, 1);
#if defined(AFS_AIX41_ENV) && defined(KERNEL)
    if (!glockOwner)
        AFS_GUNLOCK();
#endif
#else
    p = (char *) osi_Alloc(size);
#endif
    if (!p) osi_Panic("rxi_Alloc error");
    memset(p, 0, size);
    return p;
}

void rxi_Free(void *addr, register size_t size)
{
#if defined(AFS_AIX41_ENV) && defined(KERNEL)
    /* Grab the AFS filesystem lock. See afs/osi.h for the lock
     * implementation.
     */
    int glockOwner = ISAFS_GLOCK();
    if (!glockOwner)
        AFS_GLOCK();
#endif
    MUTEX_ENTER(&rx_stats_mutex);
    rxi_Alloccnt--; rxi_Allocsize -= size;
    MUTEX_EXIT(&rx_stats_mutex);
#if     (defined(AFS_AIX32_ENV) || defined(AFS_HPUX_ENV)) && !defined(AFS_HPUX100_ENV) && defined(KERNEL)
    if (size > AFS_SMALLOCSIZ)
        osi_FreeMediumSpace(addr);
    else
        osi_FreeSmall(addr);
#if defined(AFS_AIX41_ENV) && defined(KERNEL)
    if (!glockOwner)
        AFS_GUNLOCK();
#endif
#else
    osi_Free(addr, size);
#endif    
}

/* Find the peer process represented by the supplied (host,port)
 * combination.  If there is no appropriate active peer structure, a
 * new one will be allocated and initialized 
 * The origPeer, if set, is a pointer to a peer structure on which the
 * refcount will be be decremented. This is used to replace the peer
 * structure hanging off a connection structure */
struct rx_peer *rxi_FindPeer(register afs_uint32 host, 
        register u_short port, struct rx_peer *origPeer, int create)
{
    register struct rx_peer *pp;
    int hashIndex;
    hashIndex = PEER_HASH(host, port);
    MUTEX_ENTER(&rx_peerHashTable_lock);
    for (pp = rx_peerHashTable[hashIndex]; pp; pp = pp->next) {
        if ((pp->host == host) && (pp->port == port)) break;
    }
    if (!pp) {
        if (create) {
            pp = rxi_AllocPeer(); /* This bzero's *pp */
            pp->host = host;      /* set here or in InitPeerParams is zero */
            pp->port = port;
            MUTEX_INIT(&pp->peer_lock, "peer_lock", MUTEX_DEFAULT, 0);
            queue_Init(&pp->congestionQueue);
            queue_Init(&pp->rpcStats);
            pp->next = rx_peerHashTable[hashIndex];
            rx_peerHashTable[hashIndex] = pp;
            rxi_InitPeerParams(pp);
            MUTEX_ENTER(&rx_stats_mutex);
            rx_stats.nPeerStructs++;
            MUTEX_EXIT(&rx_stats_mutex);
        }
    }
    if (pp && create) {
        pp->refCount++;
    }
    if ( origPeer)
        origPeer->refCount--;
    MUTEX_EXIT(&rx_peerHashTable_lock);
    return pp;
}


/* Find the connection at (host, port) started at epoch, and with the
 * given connection id.  Creates the server connection if necessary.
 * The type specifies whether a client connection or a server
 * connection is desired.  In both cases, (host, port) specify the
 * peer's (host, pair) pair.  Client connections are not made
 * automatically by this routine.  The parameter socket gives the
 * socket descriptor on which the packet was received.  This is used,
 * in the case of server connections, to check that *new* connections
 * come via a valid (port, serviceId).  Finally, the securityIndex
 * parameter must match the existing index for the connection.  If a
 * server connection is created, it will be created using the supplied
 * index, if the index is valid for this service */
struct rx_connection *rxi_FindConnection(osi_socket socket, 
        register afs_int32 host, register u_short port, u_short serviceId, 
        afs_uint32 cid, afs_uint32 epoch, int type, u_int securityIndex)
{
    int hashindex, flag;
    register struct rx_connection *conn;
    hashindex = CONN_HASH(host, port, cid, epoch, type);
    MUTEX_ENTER(&rx_connHashTable_lock);
    rxLastConn ? (conn = rxLastConn, flag = 0) :
                 (conn = rx_connHashTable[hashindex], flag = 1);
    for (; conn; ) {
      if ((conn->type == type) && ((cid&RX_CIDMASK) == conn->cid) 
          && (epoch == conn->epoch)) {
        register struct rx_peer *pp = conn->peer;
        if (securityIndex != conn->securityIndex) {
            /* this isn't supposed to happen, but someone could forge a packet
               like this, and there seems to be some CM bug that makes this
               happen from time to time -- in which case, the fileserver
               asserts. */  
            MUTEX_EXIT(&rx_connHashTable_lock);
            return (struct rx_connection *) 0;
        }
        if (pp->host == host && pp->port == port)
            break;
        if (type == RX_CLIENT_CONNECTION && pp->port == port)
            break;
        if (type == RX_CLIENT_CONNECTION && (conn->epoch & 0x80000000))
            break;
      }
      if ( !flag )
      {
        /* the connection rxLastConn that was used the last time is not the
        ** one we are looking for now. Hence, start searching in the hash */
        flag = 1;
        conn = rx_connHashTable[hashindex];
      }
      else
        conn = conn->next;
    }
    if (!conn) {
        struct rx_service *service;
        if (type == RX_CLIENT_CONNECTION) {
            MUTEX_EXIT(&rx_connHashTable_lock);
            return (struct rx_connection *) 0;
        }
        service = rxi_FindService(socket, serviceId);
        if (!service || (securityIndex >= service->nSecurityObjects) 
            || (service->securityObjects[securityIndex] == 0)) {
            MUTEX_EXIT(&rx_connHashTable_lock);
            return (struct rx_connection *) 0;
        }
        conn = rxi_AllocConnection(); /* This bzero's the connection */
        MUTEX_INIT(&conn->conn_call_lock, "conn call lock",
                   MUTEX_DEFAULT,0);
        MUTEX_INIT(&conn->conn_data_lock, "conn data lock",
                   MUTEX_DEFAULT,0);
        CV_INIT(&conn->conn_call_cv, "conn call cv", CV_DEFAULT, 0);
        conn->next = rx_connHashTable[hashindex];
        rx_connHashTable[hashindex] = conn;
        conn->peer = rxi_FindPeer(host, port, 0, 1);
        conn->type = RX_SERVER_CONNECTION;
        conn->lastSendTime = clock_Sec();   /* don't GC immediately */
        conn->epoch = epoch;
        conn->cid = cid & RX_CIDMASK;
        /* conn->serial = conn->lastSerial = 0; */
        /* conn->timeout = 0; */
        conn->ackRate = RX_FAST_ACK_RATE;
        conn->service = service;
        conn->serviceId = serviceId;
        conn->securityIndex = securityIndex;
        conn->securityObject = service->securityObjects[securityIndex];
        conn->nSpecific = 0;
        conn->specific = NULL;
        rx_SetConnDeadTime(conn, service->connDeadTime);
        rx_SetConnIdleDeadTime(conn, service->idleDeadTime);
        /* Notify security object of the new connection */
        RXS_NewConnection(conn->securityObject, conn);
        /* XXXX Connection timeout? */
        if (service->newConnProc) (*service->newConnProc)(conn);
        MUTEX_ENTER(&rx_stats_mutex);
        rx_stats.nServerConns++;
        MUTEX_EXIT(&rx_stats_mutex);
    }

    MUTEX_ENTER(&conn->conn_data_lock);
    conn->refCount++;
    MUTEX_EXIT(&conn->conn_data_lock);

    rxLastConn = conn;  /* store this connection as the last conn used */
    MUTEX_EXIT(&rx_connHashTable_lock);
    return conn;
}

/* There are two packet tracing routines available for testing and monitoring
 * Rx.  One is called just after every packet is received and the other is
 * called just before every packet is sent.  Received packets, have had their
 * headers decoded, and packets to be sent have not yet had their headers
 * encoded.  Both take two parameters: a pointer to the packet and a sockaddr
 * containing the network address.  Both can be modified.  The return value, if
 * non-zero, indicates that the packet should be dropped.  */

int (*rx_justReceived)() = 0;
int (*rx_almostSent)() = 0;

/* A packet has been received off the interface.  Np is the packet, socket is
 * the socket number it was received from (useful in determining which service
 * this packet corresponds to), and (host, port) reflect the host,port of the
 * sender.  This call returns the packet to the caller if it is finished with
 * it, rather than de-allocating it, just as a small performance hack */

struct rx_packet *rxi_ReceivePacket(register struct rx_packet *np, 
        osi_socket socket, afs_uint32 host, u_short port, 
        int *tnop, struct rx_call **newcallp)
{
    register struct rx_call *call;
    register struct rx_connection *conn;
    int channel;
    afs_uint32 currentCallNumber;
    int type;
    int skew;
#ifdef RXDEBUG
    char *packetType;
#endif
    struct rx_packet *tnp;

#ifdef RXDEBUG
/* We don't print out the packet until now because (1) the time may not be
 * accurate enough until now in the lwp implementation (rx_Listener only gets
 * the time after the packet is read) and (2) from a protocol point of view,
 * this is the first time the packet has been seen */
    packetType = (np->header.type > 0 && np->header.type < RX_N_PACKET_TYPES)
        ? rx_packetTypes[np->header.type-1]: "*UNKNOWN*";
    dpf(("R %d %s: %x.%d.%d.%d.%d.%d.%d flags %d, packet %x",
         np->header.serial, packetType, host, port, np->header.serviceId,
         np->header.epoch, np->header.cid, np->header.callNumber, 
         np->header.seq, np->header.flags, np));
#endif

    if(np->header.type == RX_PACKET_TYPE_VERSION) {
      return rxi_ReceiveVersionPacket(np,socket,host,port, 1);
    }

    if (np->header.type == RX_PACKET_TYPE_DEBUG) {
        return rxi_ReceiveDebugPacket(np, socket, host, port, 1);
    }
#ifdef RXDEBUG
    /* If an input tracer function is defined, call it with the packet and
     * network address.  Note this function may modify its arguments. */
    if (rx_justReceived) {
        struct sockaddr_in addr;
        int drop;
        addr.sin_family = AF_INET;
        addr.sin_port = port;
        addr.sin_addr.s_addr = host;
#ifdef STRUCT_SOCKADDR_HAS_SA_LEN 
        addr.sin_len = sizeof(addr);
#endif  /* AFS_OSF_ENV */
        drop = (*rx_justReceived) (np, &addr);
        /* drop packet if return value is non-zero */
        if (drop) return np;
        port = addr.sin_port;           /* in case fcn changed addr */
        host = addr.sin_addr.s_addr;
    }
#endif

    /* If packet was not sent by the client, then *we* must be the client */
    type = ((np->header.flags&RX_CLIENT_INITIATED) != RX_CLIENT_INITIATED)
        ? RX_CLIENT_CONNECTION : RX_SERVER_CONNECTION;

    /* Find the connection (or fabricate one, if we're the server & if
     * necessary) associated with this packet */
    conn = rxi_FindConnection(socket, host, port, np->header.serviceId,
                              np->header.cid, np->header.epoch, type, 
                              np->header.securityIndex);

    if (!conn) {
      /* If no connection found or fabricated, just ignore the packet.
       * (An argument could be made for sending an abort packet for
       * the conn) */
      return np;
    }   

    MUTEX_ENTER(&conn->conn_data_lock);
    if (conn->maxSerial < np->header.serial)
        conn->maxSerial = np->header.serial;
    MUTEX_EXIT(&conn->conn_data_lock);

    /* If the connection is in an error state, send an abort packet and ignore
     * the incoming packet */
    if (conn->error) {
        /* Don't respond to an abort packet--we don't want loops! */
        MUTEX_ENTER(&conn->conn_data_lock);
        if (np->header.type != RX_PACKET_TYPE_ABORT)
            np = rxi_SendConnectionAbort(conn, np, 1, 0);
        conn->refCount--;
        MUTEX_EXIT(&conn->conn_data_lock);
        return np;
    }

    /* Check for connection-only requests (i.e. not call specific). */
    if (np->header.callNumber == 0) {
        switch (np->header.type) {
            case RX_PACKET_TYPE_ABORT:
                /* What if the supplied error is zero? */
                rxi_ConnectionError(conn, ntohl(rx_GetInt32(np,0)));
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return np;
            case RX_PACKET_TYPE_CHALLENGE:
                tnp = rxi_ReceiveChallengePacket(conn, np, 1);
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return tnp;
            case RX_PACKET_TYPE_RESPONSE:
                tnp = rxi_ReceiveResponsePacket(conn, np, 1);
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return tnp;
            case RX_PACKET_TYPE_PARAMS:
            case RX_PACKET_TYPE_PARAMS+1:
            case RX_PACKET_TYPE_PARAMS+2:
                /* ignore these packet types for now */
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return np;


            default:
                /* Should not reach here, unless the peer is broken: send an
                 * abort packet */
                rxi_ConnectionError(conn, RX_PROTOCOL_ERROR);
                MUTEX_ENTER(&conn->conn_data_lock);
                tnp = rxi_SendConnectionAbort(conn, np, 1, 0);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return tnp;
        }
    }

    channel = np->header.cid & RX_CHANNELMASK;
    call = conn->call[channel];
#ifdef  RX_ENABLE_LOCKS
    if (call)
        MUTEX_ENTER(&call->lock);
    /* Test to see if call struct is still attached to conn. */
    if (call != conn->call[channel]) {
        if (call)
            MUTEX_EXIT(&call->lock);
        if (type == RX_SERVER_CONNECTION) {
            call = conn->call[channel];
            /* If we started with no call attached and there is one now,
             * another thread is also running this routine and has gotten
             * the connection channel. We should drop this packet in the tests
             * below. If there was a call on this connection and it's now
             * gone, then we'll be making a new call below.
             * If there was previously a call and it's now different then
             * the old call was freed and another thread running this routine
             * has created a call on this channel. One of these two threads
             * has a packet for the old call and the code below handles those
             * cases.
             */
            if (call)
                MUTEX_ENTER(&call->lock);
        }
        else {
            /* This packet can't be for this call. If the new call address is
             * 0 then no call is running on this channel. If there is a call
             * then, since this is a client connection we're getting data for
             * it must be for the previous call.
             */
            MUTEX_ENTER(&rx_stats_mutex);
            rx_stats.spuriousPacketsRead++;
            MUTEX_EXIT(&rx_stats_mutex);
            MUTEX_ENTER(&conn->conn_data_lock);
            conn->refCount--;
            MUTEX_EXIT(&conn->conn_data_lock);
            return np;
        }
    }
#endif
    currentCallNumber = conn->callNumber[channel];

    if (type == RX_SERVER_CONNECTION) { /* We're the server */
        if (np->header.callNumber < currentCallNumber) {
            MUTEX_ENTER(&rx_stats_mutex);
            rx_stats.spuriousPacketsRead++;
            MUTEX_EXIT(&rx_stats_mutex);
#ifdef  RX_ENABLE_LOCKS
            if (call)
                MUTEX_EXIT(&call->lock);
#endif
            MUTEX_ENTER(&conn->conn_data_lock);
            conn->refCount--;
            MUTEX_EXIT(&conn->conn_data_lock);
            return np;
        }
        if (!call) {
            MUTEX_ENTER(&conn->conn_call_lock);
            call = rxi_NewCall(conn, channel);
            MUTEX_EXIT(&conn->conn_call_lock);
            *call->callNumber = np->header.callNumber;
            call->state = RX_STATE_PRECALL;
            clock_GetTime(&call->queueTime);
            hzero(call->bytesSent);
            hzero(call->bytesRcvd);
            rxi_KeepAliveOn(call);
        }
        else if (np->header.callNumber != currentCallNumber) {
            /* Wait until the transmit queue is idle before deciding
             * whether to reset the current call. Chances are that the
             * call will be in ether DALLY or HOLD state once the TQ_BUSY
             * flag is cleared.
             */
#ifdef AFS_GLOBAL_RXLOCK_KERNEL
            while ((call->state == RX_STATE_ACTIVE) &&
                   (call->flags & RX_CALL_TQ_BUSY)) {
                call->flags |= RX_CALL_TQ_WAIT;
#ifdef RX_ENABLE_LOCKS
                CV_WAIT(&call->cv_tq, &call->lock);
#else /* RX_ENABLE_LOCKS */
                osi_rxSleep(&call->tq);
#endif /* RX_ENABLE_LOCKS */
            }
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
            /* If the new call cannot be taken right now send a busy and set
             * the error condition in this call, so that it terminates as
             * quickly as possible */
            if (call->state == RX_STATE_ACTIVE) {
                struct rx_packet *tp;

                rxi_CallError(call, RX_CALL_DEAD);
                tp = rxi_SendSpecial(call, conn, np, RX_PACKET_TYPE_BUSY, NULL, 0, 1);
                MUTEX_EXIT(&call->lock);
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return tp;
            }
            rxi_ResetCall(call, 0);
            *call->callNumber = np->header.callNumber;
            call->state = RX_STATE_PRECALL;
            clock_GetTime(&call->queueTime);
            hzero(call->bytesSent);
            hzero(call->bytesRcvd);
            /*
             * If the number of queued calls exceeds the overload
             * threshold then abort this call.
             */
            if ((rx_BusyThreshold > 0) && (rx_nWaiting > rx_BusyThreshold)) {
                struct rx_packet *tp;

                rxi_CallError(call, rx_BusyError);
                tp = rxi_SendCallAbort(call, np, 1, 0);
                MUTEX_EXIT(&call->lock);
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return tp;
            }
            rxi_KeepAliveOn(call);
        }
        else {
            /* Continuing call; do nothing here. */
        }
    } else { /* we're the client */
        /* Ignore all incoming acknowledgements for calls in DALLY state */
        if ( call && (call->state == RX_STATE_DALLY) 
         && (np->header.type == RX_PACKET_TYPE_ACK)) {
            MUTEX_ENTER(&rx_stats_mutex);
            rx_stats.ignorePacketDally++;
            MUTEX_EXIT(&rx_stats_mutex);
#ifdef  RX_ENABLE_LOCKS
            if (call) {
                MUTEX_EXIT(&call->lock);
            }
#endif
            MUTEX_ENTER(&conn->conn_data_lock);
            conn->refCount--;
            MUTEX_EXIT(&conn->conn_data_lock);
            return np;
        }
        
        /* Ignore anything that's not relevant to the current call.  If there
         * isn't a current call, then no packet is relevant. */
        if (!call || (np->header.callNumber != currentCallNumber)) {
            MUTEX_ENTER(&rx_stats_mutex);
            rx_stats.spuriousPacketsRead++;
            MUTEX_EXIT(&rx_stats_mutex);
#ifdef  RX_ENABLE_LOCKS
            if (call) {
                MUTEX_EXIT(&call->lock);
            }
#endif
            MUTEX_ENTER(&conn->conn_data_lock);
            conn->refCount--;
            MUTEX_EXIT(&conn->conn_data_lock);
            return np;  
        }
        /* If the service security object index stamped in the packet does not
         * match the connection's security index, ignore the packet */
        if (np->header.securityIndex != conn->securityIndex) {
#ifdef  RX_ENABLE_LOCKS
            MUTEX_EXIT(&call->lock);
#endif
            MUTEX_ENTER(&conn->conn_data_lock);
            conn->refCount--;       
            MUTEX_EXIT(&conn->conn_data_lock);
            return np;
        }

        /* If we're receiving the response, then all transmit packets are
         * implicitly acknowledged.  Get rid of them. */
        if (np->header.type == RX_PACKET_TYPE_DATA) {
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
            /* XXX Hack. Because we must release the global rx lock when
             * sending packets (osi_NetSend) we drop all acks while we're
             * traversing the tq in rxi_Start sending packets out because
             * packets may move to the freePacketQueue as result of being here!
             * So we drop these packets until we're safely out of the
             * traversing. Really ugly! 
             * For fine grain RX locking, we set the acked field in the
             * packets and let rxi_Start remove them from the transmit queue.
             */
            if (call->flags & RX_CALL_TQ_BUSY) {
#ifdef  RX_ENABLE_LOCKS
                rxi_SetAcksInTransmitQueue(call);
#else
                conn->refCount--;
                return np;              /* xmitting; drop packet */
#endif
            }
            else {
                rxi_ClearTransmitQueue(call, 0);
            }
#else /* AFS_GLOBAL_RXLOCK_KERNEL */
            rxi_ClearTransmitQueue(call, 0);
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        } else {
          if (np->header.type == RX_PACKET_TYPE_ACK) {
        /* now check to see if this is an ack packet acknowledging that the
         * server actually *lost* some hard-acked data.  If this happens we
         * ignore this packet, as it may indicate that the server restarted in
         * the middle of a call.  It is also possible that this is an old ack
         * packet.  We don't abort the connection in this case, because this
         * *might* just be an old ack packet.  The right way to detect a server
         * restart in the midst of a call is to notice that the server epoch
         * changed, btw.  */
        /* XXX I'm not sure this is exactly right, since tfirst **IS**
         * XXX unacknowledged.  I think that this is off-by-one, but
         * XXX I don't dare change it just yet, since it will
         * XXX interact badly with the server-restart detection 
         * XXX code in receiveackpacket.  */
            if (ntohl(rx_GetInt32(np, FIRSTACKOFFSET)) < call->tfirst) {
                MUTEX_ENTER(&rx_stats_mutex);
                rx_stats.spuriousPacketsRead++;
                MUTEX_EXIT(&rx_stats_mutex);
                MUTEX_EXIT(&call->lock);
                MUTEX_ENTER(&conn->conn_data_lock);
                conn->refCount--;
                MUTEX_EXIT(&conn->conn_data_lock);
                return np;
            }
          }
        } /* else not a data packet */
    }

    osirx_AssertMine(&call->lock, "rxi_ReceivePacket middle");
    /* Set remote user defined status from packet */
    call->remoteStatus = np->header.userStatus;

    /* Note the gap between the expected next packet and the actual
     * packet that arrived, when the new packet has a smaller serial number
     * than expected.  Rioses frequently reorder packets all by themselves,
     * so this will be quite important with very large window sizes.
     * Skew is checked against 0 here to avoid any dependence on the type of
     * inPacketSkew (which may be unsigned).  In C, -1 > (unsigned) 0 is always
     * true! 
     * The inPacketSkew should be a smoothed running value, not just a maximum.  MTUXXX
     * see CalculateRoundTripTime for an example of how to keep smoothed values.
     * I think using a beta of 1/8 is probably appropriate.  93.04.21
     */
    MUTEX_ENTER(&conn->conn_data_lock);
    skew = conn->lastSerial - np->header.serial;
    conn->lastSerial = np->header.serial;
    MUTEX_EXIT(&conn->conn_data_lock);
    if (skew > 0) {
      register struct rx_peer *peer;
      peer = conn->peer;
      if (skew > peer->inPacketSkew) {
        dpf (("*** In skew changed from %d to %d\n", peer->inPacketSkew, skew));
        peer->inPacketSkew = skew;
      }
    }

    /* Now do packet type-specific processing */
    switch (np->header.type) {
        case RX_PACKET_TYPE_DATA:
            np = rxi_ReceiveDataPacket(call, np, 1, socket, host, port,
                                       tnop, newcallp);
            break;
        case RX_PACKET_TYPE_ACK:
            /* Respond immediately to ack packets requesting acknowledgement
             * (ping packets) */
            if (np->header.flags & RX_REQUEST_ACK) {
                if (call->error)
                    (void) rxi_SendCallAbort(call, 0, 1, 0);
                else
                    (void) rxi_SendAck(call, 0, np->header.serial,
                                       RX_ACK_PING_RESPONSE, 1);
            }
            np = rxi_ReceiveAckPacket(call, np, 1);
            break;
        case RX_PACKET_TYPE_ABORT:
            /* An abort packet: reset the connection, passing the error up to
             * the user */
            /* What if error is zero? */
            rxi_CallError(call, ntohl(*(afs_int32 *)rx_DataOf(np)));
            break;
        case RX_PACKET_TYPE_BUSY:
            /* XXXX */
            break;
        case RX_PACKET_TYPE_ACKALL:
            /* All packets acknowledged, so we can drop all packets previously
             * readied for sending */
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
            /* XXX Hack. We because we can't release the global rx lock when
             * sending packets (osi_NetSend) we drop all ack pkts while we're
             * traversing the tq in rxi_Start sending packets out because
             * packets may move to the freePacketQueue as result of being
             * here! So we drop these packets until we're safely out of the
             * traversing. Really ugly! 
             * For fine grain RX locking, we set the acked field in the packets
             * and let rxi_Start remove the packets from the transmit queue.
             */
            if (call->flags & RX_CALL_TQ_BUSY) {
#ifdef  RX_ENABLE_LOCKS
                rxi_SetAcksInTransmitQueue(call);
                break;
#else /* RX_ENABLE_LOCKS */
                conn->refCount--;
                return np;              /* xmitting; drop packet */
#endif /* RX_ENABLE_LOCKS */
            }
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
            rxi_ClearTransmitQueue(call, 0);
            break;
        default:
            /* Should not reach here, unless the peer is broken: send an abort
             * packet */
            rxi_CallError(call, RX_PROTOCOL_ERROR);
            np = rxi_SendCallAbort(call, np, 1, 0);
            break;
    };
    /* Note when this last legitimate packet was received, for keep-alive
     * processing.  Note, we delay getting the time until now in the hope that
     * the packet will be delivered to the user before any get time is required
     * (if not, then the time won't actually be re-evaluated here). */
    call->lastReceiveTime = clock_Sec();
    MUTEX_EXIT(&call->lock);
    MUTEX_ENTER(&conn->conn_data_lock);
    conn->refCount--;
    MUTEX_EXIT(&conn->conn_data_lock);
    return np;
}

/* return true if this is an "interesting" connection from the point of view
    of someone trying to debug the system */
int rxi_IsConnInteresting(struct rx_connection *aconn)
{
    register int i;
    register struct rx_call *tcall;

    if (aconn->flags & (RX_CONN_MAKECALL_WAITING | RX_CONN_DESTROY_ME))
        return 1;
    for(i=0;i<RX_MAXCALLS;i++) {
        tcall = aconn->call[i];
        if (tcall) {
            if ((tcall->state == RX_STATE_PRECALL) || (tcall->state == RX_STATE_ACTIVE))
                return 1;
            if ((tcall->mode == RX_MODE_SENDING) || (tcall->mode == RX_MODE_RECEIVING))
                return 1;
        }
    }
    return 0;
}

#ifdef KERNEL
/* if this is one of the last few packets AND it wouldn't be used by the
   receiving call to immediately satisfy a read request, then drop it on
   the floor, since accepting it might prevent a lock-holding thread from
   making progress in its reading. If a call has been cleared while in
   the precall state then ignore all subsequent packets until the call
   is assigned to a thread. */

static int TooLow(struct rx_packet *ap, struct rx_call *acall)
{
    int rc=0;
    MUTEX_ENTER(&rx_stats_mutex);
    if (((ap->header.seq != 1) &&
         (acall->flags & RX_CALL_CLEARED) &&
         (acall->state == RX_STATE_PRECALL)) ||
        ((rx_nFreePackets < rxi_dataQuota+2) &&
         !( (ap->header.seq < acall->rnext+rx_initSendWindow) 
           && (acall->flags & RX_CALL_READER_WAIT)))) {
        rc = 1;
    }
    MUTEX_EXIT(&rx_stats_mutex);
    return rc;
}
#endif /* KERNEL */

static void rxi_CheckReachEvent(struct rxevent *event, 
        struct rx_connection *conn, struct rx_call *acall)
{
    struct rx_call *call = acall;
    struct clock when;
    int i, waiting;

    MUTEX_ENTER(&conn->conn_data_lock);
    conn->checkReachEvent = NULL;
    waiting = conn->flags & RX_CONN_ATTACHWAIT;
    if (event) conn->refCount--;
    MUTEX_EXIT(&conn->conn_data_lock);

    if (waiting) {
        if (!call) {
            MUTEX_ENTER(&conn->conn_call_lock);
            MUTEX_ENTER(&conn->conn_data_lock);
            for (i=0; i<RX_MAXCALLS; i++) {
                struct rx_call *tc = conn->call[i];
                if (tc && tc->state == RX_STATE_PRECALL) {
                    call = tc;
                    break;
                }
            }
            if (!call)
                /* Indicate that rxi_CheckReachEvent is no longer running by
                 * clearing the flag.  Must be atomic under conn_data_lock to
                 * avoid a new call slipping by: rxi_CheckConnReach holds
                 * conn_data_lock while checking RX_CONN_ATTACHWAIT.
                 */
                conn->flags &= ~RX_CONN_ATTACHWAIT;
            MUTEX_EXIT(&conn->conn_data_lock);
            MUTEX_EXIT(&conn->conn_call_lock);
        }

        if (call) {
            if (call != acall) MUTEX_ENTER(&call->lock);
            rxi_SendAck(call, NULL, 0, RX_ACK_PING, 0);
            if (call != acall) MUTEX_EXIT(&call->lock);

            clock_GetTime(&when);
            when.sec += RX_CHECKREACH_TIMEOUT;
            MUTEX_ENTER(&conn->conn_data_lock);
            if (!conn->checkReachEvent) {
                conn->refCount++;
                conn->checkReachEvent =
                    rxevent_Post(&when, rxi_CheckReachEvent, conn, NULL);
            }
            MUTEX_EXIT(&conn->conn_data_lock);
        }
    }
}

static int rxi_CheckConnReach(struct rx_connection *conn, struct rx_call *call)
{
    struct rx_service *service = conn->service;
    struct rx_peer *peer = conn->peer;
    afs_uint32 now, lastReach;

    if (service->checkReach == 0)
        return 0;

    now = clock_Sec();
    MUTEX_ENTER(&peer->peer_lock);
    lastReach = peer->lastReachTime;
    MUTEX_EXIT(&peer->peer_lock);
    if (now - lastReach < RX_CHECKREACH_TTL)
        return 0;

    MUTEX_ENTER(&conn->conn_data_lock);
    if (conn->flags & RX_CONN_ATTACHWAIT) {
        MUTEX_EXIT(&conn->conn_data_lock);
        return 1;
    }
    conn->flags |= RX_CONN_ATTACHWAIT;
    MUTEX_EXIT(&conn->conn_data_lock);
    if (!conn->checkReachEvent)
        rxi_CheckReachEvent(NULL, conn, call);

    return 1;
}

/* try to attach call, if authentication is complete */
static void TryAttach(register struct rx_call *acall, 
        register osi_socket socket, register int *tnop, 
        register struct rx_call **newcallp, int reachOverride)
{
    struct rx_connection *conn = acall->conn;

    if (conn->type==RX_SERVER_CONNECTION && acall->state==RX_STATE_PRECALL) {
        /* Don't attach until we have any req'd. authentication. */
        if (RXS_CheckAuthentication(conn->securityObject, conn) == 0) {
            if (reachOverride || rxi_CheckConnReach(conn, acall) == 0)
                rxi_AttachServerProc(acall, socket, tnop, newcallp);
                /* Note:  this does not necessarily succeed; there
                 * may not any proc available
                 */
        }
        else {
            rxi_ChallengeOn(acall->conn);
        }
    }
}

/* A data packet has been received off the interface.  This packet is
 * appropriate to the call (the call is in the right state, etc.).  This
 * routine can return a packet to the caller, for re-use */

struct rx_packet *rxi_ReceiveDataPacket(register struct rx_call *call, 
        register struct rx_packet *np, int istack, osi_socket socket, 
        afs_uint32 host, u_short port, int *tnop, struct rx_call **newcallp)
{
    int ackNeeded = 0;  /* 0 means no, otherwise ack_reason */
    int newPackets = 0;
    int didHardAck = 0;
    int haveLast = 0;
    afs_uint32 seq, serial, flags;
    int isFirst;
    struct rx_packet *tnp;
    struct clock when;
    MUTEX_ENTER(&rx_stats_mutex);
    rx_stats.dataPacketsRead++;
    MUTEX_EXIT(&rx_stats_mutex);

#ifdef KERNEL
    /* If there are no packet buffers, drop this new packet, unless we can find
     * packet buffers from inactive calls */
    if (!call->error &&
        (rxi_OverQuota(RX_PACKET_CLASS_RECEIVE) || TooLow(np, call))) {
        MUTEX_ENTER(&rx_freePktQ_lock);
        rxi_NeedMorePackets = TRUE;
        MUTEX_EXIT(&rx_freePktQ_lock);
        MUTEX_ENTER(&rx_stats_mutex);
        rx_stats.noPacketBuffersOnRead++;
        MUTEX_EXIT(&rx_stats_mutex);
        call->rprev = np->header.serial;
        rxi_calltrace(RX_TRACE_DROP, call);
        dpf (("packet %x dropped on receipt - quota problems", np));
        if (rxi_doreclaim)
            rxi_ClearReceiveQueue(call);
        clock_GetTime(&when);
        clock_Add(&when, &rx_softAckDelay);
        if (!call->delayedAckEvent ||
            clock_Gt(&call->delayedAckEvent->eventTime, &when)) {
            rxevent_Cancel(call->delayedAckEvent, call,
                           RX_CALL_REFCOUNT_DELAY);
            CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
            call->delayedAckEvent = rxevent_Post(&when, rxi_SendDelayedAck,
                                                 call, 0);
        }
        /* we've damaged this call already, might as well do it in. */
        return np;
    }
#endif /* KERNEL */

    /*
     * New in AFS 3.5, if the RX_JUMBO_PACKET flag is set then this
     * packet is one of several packets transmitted as a single
     * datagram. Do not send any soft or hard acks until all packets
     * in a jumbogram have been processed. Send negative acks right away.
     */
    for (isFirst = 1 , tnp = NULL ; isFirst || tnp ; isFirst = 0 ) {
        /* tnp is non-null when there are more packets in the
         * current jumbo gram */
        if (tnp) {
            if (np)
                rxi_FreePacket(np);
            np = tnp;
        }

        seq = np->header.seq;
        serial = np->header.serial;
        flags = np->header.flags;

        /* If the call is in an error state, send an abort message */
        if (call->error)
            return rxi_SendCallAbort(call, np, istack, 0);

        /* The RX_JUMBO_PACKET is set in all but the last packet in each
         * AFS 3.5 jumbogram. */
        if (flags & RX_JUMBO_PACKET) {
            tnp = rxi_SplitJumboPacket(np, host, port, isFirst);
        } else {
            tnp = NULL;
        }

        if (np->header.spare != 0) {
            MUTEX_ENTER(&call->conn->conn_data_lock);
            call->conn->flags |= RX_CONN_USING_PACKET_CKSUM;
            MUTEX_EXIT(&call->conn->conn_data_lock);
        }

        /* The usual case is that this is the expected next packet */
        if (seq == call->rnext) {

            /* Check to make sure it is not a duplicate of one already queued */
            if (queue_IsNotEmpty(&call->rq) 
                && queue_First(&call->rq, rx_packet)->header.seq == seq) {
                MUTEX_ENTER(&rx_stats_mutex);
                rx_stats.dupPacketsRead++;
                MUTEX_EXIT(&rx_stats_mutex);
                dpf (("packet %x dropped on receipt - duplicate", np));
                rxevent_Cancel(call->delayedAckEvent, call,
                               RX_CALL_REFCOUNT_DELAY);
                np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
                ackNeeded = 0;
                call->rprev = seq;
                continue;
            }

            /* It's the next packet. Stick it on the receive queue
             * for this call. Set newPackets to make sure we wake
             * the reader once all packets have been processed */
            queue_Prepend(&call->rq, np);
            call->nSoftAcks++;
            np = NULL; /* We can't use this anymore */
            newPackets = 1;

            /* If an ack is requested then set a flag to make sure we
             * send an acknowledgement for this packet */
            if (flags & RX_REQUEST_ACK) {
                ackNeeded = RX_ACK_REQUESTED;
            }

            /* Keep track of whether we have received the last packet */
            if (flags & RX_LAST_PACKET) {
                call->flags |= RX_CALL_HAVE_LAST;
                haveLast = 1;
            }

            /* Check whether we have all of the packets for this call */
            if (call->flags & RX_CALL_HAVE_LAST) {
                afs_uint32 tseq;                /* temporary sequence number */
                struct rx_packet *tp;   /* Temporary packet pointer */
                struct rx_packet *nxp;  /* Next pointer, for queue_Scan */

                for (tseq = seq, queue_Scan(&call->rq, tp, nxp, rx_packet)) {
                    if (tseq != tp->header.seq)
                        break;
                    if (tp->header.flags & RX_LAST_PACKET) {
                        call->flags |= RX_CALL_RECEIVE_DONE;
                        break;
                    }
                    tseq++;
                }
            }

            /* Provide asynchronous notification for those who want it
             * (e.g. multi rx) */
            if (call->arrivalProc) {
                (*call->arrivalProc)(call, call->arrivalProcHandle,
                                     (int) call->arrivalProcArg);
                call->arrivalProc = (VOID (*)()) 0;
            }

            /* Update last packet received */
            call->rprev = seq;

            /* If there is no server process serving this call, grab
             * one, if available. We only need to do this once. If a
             * server thread is available, this thread becomes a server
             * thread and the server thread becomes a listener thread. */
            if (isFirst) {
                TryAttach(call, socket, tnop, newcallp, 0);
            }
        }       
        /* This is not the expected next packet. */
        else {
            /* Determine whether this is a new or old packet, and if it's
             * a new one, whether it fits into the current receive window.
             * Also figure out whether the packet was delivered in sequence.
             * We use the prev variable to determine whether the new packet
             * is the successor of its immediate predecessor in the
             * receive queue, and the missing flag to determine whether
             * any of this packets predecessors are missing.  */

            afs_uint32 prev;            /* "Previous packet" sequence number */
            struct rx_packet *tp;       /* Temporary packet pointer */
            struct rx_packet *nxp;      /* Next pointer, for queue_Scan */
            int missing;                /* Are any predecessors missing? */

            /* If the new packet's sequence number has been sent to the
             * application already, then this is a duplicate */
            if (seq < call->rnext) {
                MUTEX_ENTER(&rx_stats_mutex);
                rx_stats.dupPacketsRead++;
                MUTEX_EXIT(&rx_stats_mutex);
                rxevent_Cancel(call->delayedAckEvent, call,
                               RX_CALL_REFCOUNT_DELAY);
                np = rxi_SendAck(call, np, serial, RX_ACK_DUPLICATE, istack);
                ackNeeded = 0;
                call->rprev = seq;
                continue;
            }

            /* If the sequence number is greater than what can be
             * accomodated by the current window, then send a negative
             * acknowledge and drop the packet */
            if ((call->rnext + call->rwind) <= seq) {
                rxevent_Cancel(call->delayedAckEvent, call,
                               RX_CALL_REFCOUNT_DELAY);
                np = rxi_SendAck(call, np, serial,
                                 RX_ACK_EXCEEDS_WINDOW, istack);
                ackNeeded = 0;
                call->rprev = seq;
                continue;
            }

            /* Look for the packet in the queue of old received packets */
            for (prev = call->rnext - 1, missing = 0,
                 queue_Scan(&call->rq, tp, nxp, rx_packet)) {
                /*Check for duplicate packet */
                if (seq == tp->header.seq) {
                    MUTEX_ENTER(&rx_stats_mutex);
                    rx_stats.dupPacketsRead++;
                    MUTEX_EXIT(&rx_stats_mutex);
                    rxevent_Cancel(call->delayedAckEvent, call,
                                   RX_CALL_REFCOUNT_DELAY);
                    np = rxi_SendAck(call, np, serial, 
                                     RX_ACK_DUPLICATE, istack);
                    ackNeeded = 0;
                    call->rprev = seq;
                    goto nextloop;
                }
                /* If we find a higher sequence packet, break out and
                 * insert the new packet here. */
                if (seq < tp->header.seq) break;
                /* Check for missing packet */
                if (tp->header.seq != prev+1) {
                    missing = 1;
                }

                prev = tp->header.seq;
            }

            /* Keep track of whether we have received the last packet. */
            if (flags & RX_LAST_PACKET) {
                call->flags |= RX_CALL_HAVE_LAST;
            }

            /* It's within the window: add it to the the receive queue.
             * tp is left by the previous loop either pointing at the
             * packet before which to insert the new packet, or at the
             * queue head if the queue is empty or the packet should be
             * appended. */
            queue_InsertBefore(tp, np);
            call->nSoftAcks++;
            np = NULL;

            /* Check whether we have all of the packets for this call */
            if ((call->flags & RX_CALL_HAVE_LAST)
             && !(call->flags & RX_CALL_RECEIVE_DONE)) {
                afs_uint32 tseq;                /* temporary sequence number */

                for (tseq = call->rnext,
                     queue_Scan(&call->rq, tp, nxp, rx_packet)) {
                    if (tseq != tp->header.seq)
                        break;
                    if (tp->header.flags & RX_LAST_PACKET) {
                        call->flags |= RX_CALL_RECEIVE_DONE;
                        break;
                    }
                    tseq++;
                }
            }

            /* We need to send an ack of the packet is out of sequence, 
             * or if an ack was requested by the peer. */
            if (seq != prev+1 || missing || (flags & RX_REQUEST_ACK)) {
                ackNeeded = RX_ACK_OUT_OF_SEQUENCE;
            }

            /* Acknowledge the last packet for each call */
            if (flags & RX_LAST_PACKET) {
                haveLast = 1;
            }

            call->rprev = seq;
        }
nextloop:;
    }

    if (newPackets) {
        /*
         * If the receiver is waiting for an iovec, fill the iovec
         * using the data from the receive queue */
        if (call->flags & RX_CALL_IOVEC_WAIT) {
            didHardAck = rxi_FillReadVec(call, serial); 
            /* the call may have been aborted */
            if (call->error) {
                return NULL;
            }
            if (didHardAck) {
                ackNeeded = 0;
            }
        }

        /* Wakeup the reader if any */
        if ((call->flags & RX_CALL_READER_WAIT) &&
            (!(call->flags & RX_CALL_IOVEC_WAIT) || !(call->iovNBytes) ||
             (call->iovNext >= call->iovMax) ||
             (call->flags & RX_CALL_RECEIVE_DONE))) {
            call->flags &= ~RX_CALL_READER_WAIT;
#ifdef  RX_ENABLE_LOCKS
            CV_BROADCAST(&call->cv_rq);
#else
            osi_rxWakeup(&call->rq);
#endif
        }
    }

    /*
     * Send an ack when requested by the peer, or once every
     * rxi_SoftAckRate packets until the last packet has been
     * received. Always send a soft ack for the last packet in
     * the server's reply. */
    if (ackNeeded) {
        rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
        np = rxi_SendAck(call, np, serial, ackNeeded, istack);
    } else if (call->nSoftAcks > (u_short)rxi_SoftAckRate) {
        rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
        np = rxi_SendAck(call, np, serial, RX_ACK_IDLE, istack);
    } else if (call->nSoftAcks) {
        clock_GetTime(&when);
        if (haveLast && !(flags & RX_CLIENT_INITIATED)) {
            clock_Add(&when, &rx_lastAckDelay);
        } else {
            clock_Add(&when, &rx_softAckDelay);
        }
        if (!call->delayedAckEvent ||
            clock_Gt(&call->delayedAckEvent->eventTime, &when)) {
            rxevent_Cancel(call->delayedAckEvent, call,
                           RX_CALL_REFCOUNT_DELAY);
            CALL_HOLD(call, RX_CALL_REFCOUNT_DELAY);
            call->delayedAckEvent = rxevent_Post(&when, rxi_SendDelayedAck,
                                                 call, 0);
        }
    } else if (call->flags & RX_CALL_RECEIVE_DONE) {
        rxevent_Cancel(call->delayedAckEvent, call, RX_CALL_REFCOUNT_DELAY);
    }

    return np;
}

#ifdef  ADAPT_WINDOW
static void rxi_ComputeRate();
#endif

static void rxi_UpdatePeerReach(struct rx_connection *conn, struct rx_call *acall)
{
    struct rx_peer *peer = conn->peer;

    MUTEX_ENTER(&peer->peer_lock);
    peer->lastReachTime = clock_Sec();
    MUTEX_EXIT(&peer->peer_lock);

    MUTEX_ENTER(&conn->conn_data_lock);
    if (conn->flags & RX_CONN_ATTACHWAIT) {
        int i;

        conn->flags &= ~RX_CONN_ATTACHWAIT;
        MUTEX_EXIT(&conn->conn_data_lock);

        for (i=0; i<RX_MAXCALLS; i++) {
            struct rx_call *call = conn->call[i];
            if (call) {
                if (call != acall) MUTEX_ENTER(&call->lock);
                /* tnop can be null if newcallp is null */
                TryAttach(call, (osi_socket) -1, NULL, NULL, 1);
                if (call != acall) MUTEX_EXIT(&call->lock);
            }
        }
    } else
        MUTEX_EXIT(&conn->conn_data_lock);
}

/* rxi_ComputePeerNetStats
 *
 * Called exclusively by rxi_ReceiveAckPacket to compute network link
 * estimates (like RTT and throughput) based on ack packets.  Caller
 * must ensure that the packet in question is the right one (i.e.
 * serial number matches).
 */
static void
rxi_ComputePeerNetStats(struct rx_call *call, struct rx_packet *p,
        struct rx_ackPacket *ap, struct rx_packet *np)
{
    struct rx_peer *peer = call->conn->peer;

    /* Use RTT if not delayed by client. */
    if (ap->reason != RX_ACK_DELAY)
        rxi_ComputeRoundTripTime(p, &p->timeSent, peer);
#ifdef ADAPT_WINDOW
    rxi_ComputeRate(peer, call, p, np, ap->reason);
#endif
}

/* The real smarts of the whole thing.  */
struct rx_packet *rxi_ReceiveAckPacket(register struct rx_call *call, 
        struct rx_packet *np, int istack)
{
    struct rx_ackPacket *ap;
    int nAcks;
    register struct rx_packet *tp;
    register struct rx_packet *nxp;     /* Next packet pointer for queue_Scan */
    register struct rx_connection *conn = call->conn;
    struct rx_peer *peer = conn->peer;
    afs_uint32 first;
    afs_uint32 serial;
    /* because there are CM's that are bogus, sending weird values for this. */
    afs_uint32 skew = 0;
    int nbytes;
    int missing;
    int acked;
    int nNacked = 0;
    int newAckCount = 0;
    u_short maxMTU = 0;  /* Set if peer supports AFS 3.4a jumbo datagrams */
    int maxDgramPackets = 0; /* Set if peer supports AFS 3.5 jumbo datagrams */

    MUTEX_ENTER(&rx_stats_mutex);
    rx_stats.ackPacketsRead++;
    MUTEX_EXIT(&rx_stats_mutex);
    ap = (struct rx_ackPacket *) rx_DataOf(np);
    nbytes = rx_Contiguous(np) - ((ap->acks) - (u_char *)ap);
    if (nbytes < 0)
      return np;       /* truncated ack packet */

    /* depends on ack packet struct */
    nAcks = MIN((unsigned)nbytes, (unsigned) ap->nAcks);
    first = ntohl(ap->firstPacket);
    serial = ntohl(ap->serial);
    /* temporarily disabled -- needs to degrade over time 
       skew = ntohs(ap->maxSkew); */

    /* Ignore ack packets received out of order */
    if (first < call->tfirst) {
        return np;
    }

    if (np->header.flags & RX_SLOW_START_OK) {
        call->flags |= RX_CALL_SLOW_START_OK;
    }

    if (ap->reason == RX_ACK_PING_RESPONSE)
        rxi_UpdatePeerReach(conn, call);
    
#ifdef RXDEBUG
    if (rx_Log) {
      fprintf( rx_Log, 
              "RACK: reason %x previous %u seq %u serial %u skew %d first %u",
              ap->reason, ntohl(ap->previousPacket), 
              (unsigned int) np->header.seq, (unsigned int) serial, 
              (unsigned int) skew, ntohl(ap->firstPacket));
        if (nAcks) {
            int offset;
            for (offset = 0; offset < nAcks; offset++) 
                putc(ap->acks[offset] == RX_ACK_TYPE_NACK? '-' : '*', rx_Log);
        }
        putc('\n', rx_Log);
    }
#endif

    /* Update the outgoing packet skew value to the latest value of
     * the peer's incoming packet skew value.  The ack packet, of
     * course, could arrive out of order, but that won't affect things
     * much */
    MUTEX_ENTER(&peer->peer_lock);
    peer->outPacketSkew = skew;

    /* Check for packets that no longer need to be transmitted, and
     * discard them.  This only applies to packets positively
     * acknowledged as having been sent to the peer's upper level.
     * All other packets must be retained.  So only packets with
     * sequence numbers < ap->firstPacket are candidates. */
    for (queue_Scan(&call->tq, tp, nxp, rx_packet)) {
        if (tp->header.seq >= first) break;
        call->tfirst = tp->header.seq + 1;
        if (serial && (tp->header.serial == serial ||
                       tp->firstSerial == serial))
            rxi_ComputePeerNetStats(call, tp, ap, np);
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
    /* XXX Hack. Because we have to release the global rx lock when sending
     * packets (osi_NetSend) we drop all acks while we're traversing the tq
     * in rxi_Start sending packets out because packets may move to the
     * freePacketQueue as result of being here! So we drop these packets until
     * we're safely out of the traversing. Really ugly! 
     * To make it even uglier, if we're using fine grain locking, we can
     * set the ack bits in the packets and have rxi_Start remove the packets
     * when it's done transmitting.
     */
        if (!(tp->flags & RX_PKTFLAG_ACKED)) {
            newAckCount++;
        }
        if (call->flags & RX_CALL_TQ_BUSY) {
#ifdef RX_ENABLE_LOCKS
            tp->flags |= RX_PKTFLAG_ACKED;
            call->flags |= RX_CALL_TQ_SOME_ACKED;
#else /* RX_ENABLE_LOCKS */
            break;
#endif /* RX_ENABLE_LOCKS */
        } else
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        {
        queue_Remove(tp);
        rxi_FreePacket(tp); /* rxi_FreePacket mustn't wake up anyone, preemptively. */
        }
    }

#ifdef ADAPT_WINDOW
    /* Give rate detector a chance to respond to ping requests */
    if (ap->reason == RX_ACK_PING_RESPONSE) {
        rxi_ComputeRate(peer, call, 0, np, ap->reason);
    }
#endif

    /* N.B. we don't turn off any timers here.  They'll go away by themselves, anyway */
   
   /* Now go through explicit acks/nacks and record the results in
    * the waiting packets.  These are packets that can't be released
    * yet, even with a positive acknowledge.  This positive
    * acknowledge only means the packet has been received by the
    * peer, not that it will be retained long enough to be sent to
    * the peer's upper level.  In addition, reset the transmit timers
    * of any missing packets (those packets that must be missing
    * because this packet was out of sequence) */

    call->nSoftAcked = 0;
    for (missing = 0, queue_Scan(&call->tq, tp, nxp, rx_packet)) {
        /* Update round trip time if the ack was stimulated on receipt
         * of this packet */
#ifdef AFS_GLOBAL_RXLOCK_KERNEL
#ifdef RX_ENABLE_LOCKS
        if (tp->header.seq >= first)
#endif /* RX_ENABLE_LOCKS */
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
            if (serial && (tp->header.serial == serial ||
                           tp->firstSerial == serial))
                rxi_ComputePeerNetStats(call, tp, ap, np);

        /* Set the acknowledge flag per packet based on the
         * information in the ack packet. An acknowlegded packet can
         * be downgraded when the server has discarded a packet it
         * soacked previously, or when an ack packet is received
         * out of sequence. */
        if (tp->header.seq < first) {
            /* Implicit ack information */
            if (!(tp->flags & RX_PKTFLAG_ACKED)) {
                newAckCount++;
            }
            tp->flags |= RX_PKTFLAG_ACKED;
        }
        else if (tp->header.seq < first + nAcks) {
            /* Explicit ack information:  set it in the packet appropriately */
            if (ap->acks[tp->header.seq - first] == RX_ACK_TYPE_ACK) {
                if (!(tp->flags & RX_PKTFLAG_ACKED)) {
                    newAckCount++;
                    tp->flags |= RX_PKTFLAG_ACKED;
                }
                if (missing) {
                    nNacked++;
                } else {
                    call->nSoftAcked++;
                }
            } else {
                tp->flags &= ~RX_PKTFLAG_ACKED;
                missing = 1;
            }
        }
        else {
            tp->flags &= ~RX_PKTFLAG_ACKED;
            missing = 1;
        }

        /* If packet isn't yet acked, and it has been transmitted at least 
         * once, reset retransmit time using latest timeout 
         * ie, this should readjust the retransmit timer for all outstanding 
         * packets...  So we don't just retransmit when we should know better*/

        if (!(tp->flags & RX_PKTFLAG_ACKED) && !clock_IsZero(&tp->retryTime)) {
          tp->retryTime = tp->timeSent;
          clock_Add(&tp->retryTime, &peer->timeout);
          /* shift by eight because one quarter-sec ~ 256 milliseconds */
          clock_Addmsec(&(tp->retryTime), ((afs_uint32) tp->backoff) << 8);
        }
    }

    /* If the window has been extended by this acknowledge packet,
     * then wakeup a sender waiting in alloc for window space, or try
     * sending packets now, if he's been sitting on packets due to
     * lack of window space */
    if (call->tnext < (call->tfirst + call->twind))  {
#ifdef  RX_ENABLE_LOCKS
        CV_SIGNAL(&call->cv_twind);
#else
        if (call->flags & RX_CALL_WAIT_WINDOW_ALLOC) {
            call->flags &= ~RX_CALL_WAIT_WINDOW_ALLOC;
            osi_rxWakeup(&call->twind);
        }
#endif
        if (call->flags & RX_CALL_WAIT_WINDOW_SEND) {
            call->flags &= ~RX_CALL_WAIT_WINDOW_SEND;
        }
    }

    /* if the ack packet has a receivelen field hanging off it,
     * update our state */
    if ( np->length >= rx_AckDataSize(ap->nAcks) + 2*sizeof(afs_int32)) {
      afs_uint32 tSize;

      /* If the ack packet has a "recommended" size that is less than 
       * what I am using now, reduce my size to match */
      rx_packetread(np, rx_AckDataSize(ap->nAcks)+sizeof(afs_int32),
                    sizeof(afs_int32), &tSize);
      tSize = (afs_uint32) ntohl(tSize);
      peer->natMTU = rxi_AdjustIfMTU(MIN(tSize, peer->ifMTU));

      /* Get the maximum packet size to send to this peer */
      rx_packetread(np, rx_AckDataSize(ap->nAcks), sizeof(afs_int32),
                    &tSize);
      tSize = (afs_uint32)ntohl(tSize);
      tSize = (afs_uint32)MIN(tSize, rx_MyMaxSendSize);
      tSize = rxi_AdjustMaxMTU(peer->natMTU, tSize);

      /* sanity check - peer might have restarted with different params.
       * If peer says "send less", dammit, send less...  Peer should never 
       * be unable to accept packets of the size that prior AFS versions would
       * send without asking.  */
      if (peer->maxMTU != tSize) {
          peer->maxMTU = tSize;
          peer->MTU = MIN(tSize, peer->MTU);
          call->MTU = MIN(call->MTU, tSize);
          peer->congestSeq++;
      }

      if ( np->length == rx_AckDataSize(ap->nAcks) +3*sizeof(afs_int32)) {
          /* AFS 3.4a */
          rx_packetread(np, rx_AckDataSize(ap->nAcks)+2*sizeof(afs_int32),
                        sizeof(afs_int32), &tSize);
          tSize = (afs_uint32) ntohl(tSize);  /* peer's receive window, if it's */
          if (tSize < call->twind) {       /* smaller than our send */
              call->twind = tSize;         /* window, we must send less... */
              call->ssthresh = MIN(call->twind, call->ssthresh);
          }

          /* Only send jumbograms to 3.4a fileservers. 3.3a RX gets the
           * network MTU confused with the loopback MTU. Calculate the
           * maximum MTU here for use in the slow start code below.
           */
          maxMTU = peer->maxMTU;
          /* Did peer restart with older RX version? */
          if (peer->maxDgramPackets > 1) {
              peer->maxDgramPackets = 1;
          }
      } else if ( np->length >= rx_AckDataSize(ap->nAcks) +4*sizeof(afs_int32)) {
          /* AFS 3.5 */
          rx_packetread(np, rx_AckDataSize(ap->nAcks)+2*sizeof(afs_int32),
                        sizeof(afs_int32), &tSize);
          tSize = (afs_uint32) ntohl(tSize);
          /*
           * As of AFS 3.5 we set the send window to match the receive window. 
           */
          if (tSize < call->twind) {
              call->twind = tSize;
              call->ssthresh = MIN(call->twind, call->ssthresh);
          } else if (tSize > call->twind) {
              call->twind = tSize;
          }

          /*
           * As of AFS 3.5, a jumbogram is more than one fixed size
           * packet transmitted in a single UDP datagram. If the remote
           * MTU is smaller than our local MTU then never send a datagram
           * larger than the natural MTU.
           */
          rx_packetread(np, rx_AckDataSize(ap->nAcks)+3*sizeof(afs_int32),
                        sizeof(afs_int32), &tSize);
          maxDgramPackets = (afs_uint32) ntohl(tSize);
          maxDgramPackets = MIN(maxDgramPackets, rxi_nDgramPackets);
          maxDgramPackets = MIN(maxDgramPackets,
                                (int)(peer->ifDgramPackets));
          maxDgramPackets = MIN(maxDgramPackets, tSize);
          if (maxDgramPackets > 1) {
            peer->maxDgramPackets = maxDgramPackets;
            call->MTU = RX_JUMBOBUFFERSIZE+RX_HEADER_SIZE;
          } else {
            peer->maxDgramPackets = 1;
            call->MTU = peer->natMTU;
          }
       } else if (peer->maxDgramPackets > 1) {
          /* Restarted with lower version of RX */
          peer->maxDgramPackets = 1;
       }
    } else if (peer->maxDgramPackets > 1 ||
               peer->maxMTU != OLD_MAX_PACKET_SIZE) {
        /* Restarted with lower version of RX */
        peer->maxMTU = OLD_MAX_PACKET_SIZE;
        peer->natMTU = OLD_MAX_PACKET_SIZE;
        peer->MTU = OLD_MAX_PACKET_SIZE;
        peer->maxDgramPackets = 1;
        peer->nDgramPackets = 1;
        peer->congestSeq++;
        call->MTU = OLD_MAX_PACKET_SIZE;
    }

    if (nNacked) {
        /*
         * Calculate how many datagrams were successfully received after
         * the first missing packet and adjust the negative ack counter
         * accordingly.
         */
        call->nAcks = 0;
        call->nNacks++;
        nNacked = (nNacked + call->nDgramPackets - 1) / call->nDgramPackets;
        if (call->nNacks < nNacked) {
            call->nNacks = nNacked;
        }
    } else {
        if (newAckCount) {
            call->nAcks++;
        }
        call->nNacks = 0;
    }

    if (call->flags & RX_CALL_FAST_RECOVER) {
        if (nNacked) {
            call->cwind = MIN((int)(call->cwind + 1), rx_maxSendWindow);
        } else {
            call->flags &= ~RX_CALL_FAST_RECOVER;
            call->cwind = call->nextCwind;
            call->nextCwind = 0;
            call->nAcks = 0;
        }
        call->nCwindAcks = 0;
    }
    else if (nNacked && call->nNacks >= (u_short)rx_nackThreshold) {
        /* Three negative acks in a row trigger congestion recovery */
#ifdef  AFS_GLOBAL_RXLOCK_KERNEL
        MUTEX_EXIT(&peer->peer_lock);
        if (call->flags & RX_CALL_FAST_RECOVER_WAIT) {
                /* someone else is waiting to start recovery */
                return np;
        }
        call->flags |= RX_CALL_FAST_RECOVER_WAIT;
        while (call->flags & RX_CALL_TQ_BUSY) {
            call->flags |= RX_CALL_TQ_WAIT;
#ifdef RX_ENABLE_LOCKS
            CV_WAIT(&call->cv_tq, &call->lock);
#else /* RX_ENABLE_LOCKS */
            osi_rxSleep(&call->tq);
#endif /* RX_ENABLE_LOCKS */
        }
        MUTEX_ENTER(&peer->peer_lock);
#endif /* AFS_GLOBAL_RXLOCK_KERNEL */
        call->flags &= ~RX_CALL_FAST_RECOVER_WAIT;
        call->flags |= RX_CALL_FAST_RECOVER;
        call->ssthresh = MAX(4, MIN((int)call->cwind, (int)call->twind))>>1;
        call->cwind = MIN((int)(call->ssthresh + rx_nackThreshold),
                          rx_maxSendWindow);
        call->nDgramPackets = MAX(2, (int)call->nDgramPackets)>>1;
        call->nextCwind = call->ssthresh;
        call->nAcks = 0;
        call->nNacks = 0;
        peer->MTU = call->MTU;
        peer->cwind = call->nextCwind;
        peer->nDgramPackets = call->nDgramPackets;
        peer->congestSeq++;
        call->congestSeq = peer->congestSeq;
        /* Reset the resend times on the packets that were nacked
         * so we will retransmit as soon as the window permits*/