<commit_before>/*
 *
 * Copyright (C) 2000 <NAME> (<EMAIL>)
 * Copyright (C) 2001 <NAME> (<EMAIL>) and 
 *                    <NAME> (<EMAIL>).
 * Copyright (C) 2001 <NAME> (<EMAIL>), 
 *                    <NAME> (<EMAIL>) and
 *                    <NAME> (<EMAIL>).
 *
 *  Permission is hereby granted, free of charge, to any person obtaining
 *  a copy of this software and associated documentation files (the
 *  "Software"), to deal in the Software without restriction, including
 *  without limitation the rights to use, copy, modify, merge, publish,
 *  distribute, sublicense, and/or sell copies of the Software, and to
 *  permit persons to whom the Software is furnished to do so, subject to
 *  the following conditions:
 *
 *  The above copyright notice and this permission notice shall be
 *  included in all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 *  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 *  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 *  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 *  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 *  OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 *  WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 *
 */


// Chord's RPC manager is designed to perform flow control on all
// Chord RPCs. It maintains a cache of hosts (i.e. IP addresses)
// that it has connected to and maintains statistics about latency
// and such to those hosts. It uses these statistics to calculate
// optimal window sizes and delays.

// #define __J__ 1

#include "dhash_prot.h"
#include "chord_util.h"
#include "location.h"

long outbytes;
// UTILITY FUNCTIONS

const int shortstats (getenv ("SHORT_STATS") ? 1 : 0);
const int aclnttrace (getenv ("ACLNT_TRACE")
		      ? atoi (getenv ("ACLNT_TRACE")) : 0);
const bool aclnttime (getenv ("ACLNT_TIME"));

static inline const char *
tracetime ()
{
  static str buf ("");
  if (aclnttime) {
    timespec ts;
    clock_gettime (CLOCK_REALTIME, &ts);
    buf = strbuf (" %d.%06d", int (ts.tv_sec), int (ts.tv_nsec/1000));
  }
  return buf;
}

static
int partition (float *A, int p, int r)
{
  float x = A[p];
  int i = p - 1;
  int j = r + 1;

  while (1) {
    do {  j -= 1; } while (A[j] > x);
    do {  i += 1; } while (A[i] < x);
    if (i >= j)
      return j;
    float tmp = A[i];
    A[i] = A[j];
    A[j] = tmp;     
  }
}

static float
myselect (float *A, int p, int r, int i)
{
  if (p == r)
    return A[p];
  int q = partition (A, p, r);
  int k = q - p + 1;
  if (i <= k)
    return myselect(A, p, q, i);
  else
    return myselect(A, q+1, r, i-k);
}

// -----------------------------------------------------
hostinfo::hostinfo (const net_address &r)
  : host (r.hostname), rpcdelay (0), nrpc (0), maxdelay (0),
    a_lat (0.0), a_var (0.0)
{
}

// -----------------------------------------------------

rpc_manager::rpc_manager (ptr<u_int32_t> _nrcv)
  : nrpc (0), nrpcfailed (0), nsent (0), npending (0), nrcv (_nrcv)
{
  int dgram_fd = inetsocket (SOCK_DGRAM);
  if (dgram_fd < 0) fatal << "Failed to allocate dgram socket\n";
  dgram_xprt = axprt_dgram::alloc (dgram_fd, sizeof(sockaddr), 230000);
  if (!dgram_xprt) fatal << "Failed to allocate dgram xprt\n";
}

float
rpc_manager::get_a_lat (ptr<location> l)
{
  return 0.0;
}

float
rpc_manager::get_a_var (ptr<location> l)
{
  return 0.0;
}

float
rpc_manager::get_avg_lat ()
{
  return 0.0;
}

float
rpc_manager::get_avg_var ()
{
  return 0.0;
}

long
rpc_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long fake_seqno = 0)
{
  ref<aclnt> c = aclnt::alloc (dgram_xprt, prog, 
			       (sockaddr *)&(l->saddr));
  
  c->call (procno, in, out, wrap (this, &rpc_manager::doRPCcb, cb)); 
  return 0;
}

void
rpc_manager::doRPCcb (aclnt_cb realcb, clnt_stat err)
{
  if (err)
    nrpcfailed++;
  else
    nrpc++;
  
  (realcb) (err);
}

// -----------------------------------------------------
long
tcp_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long fake_seqno = 0)
{
  // hack to avoid limit on wrap()'s number of arguments
  RPC_delay_args *args = New RPC_delay_args (l, prog, procno,
					     in, out, cb);

  // wierd: tcpconnect wants the address in NBO, and port in HBO
  tcpconnect (l->saddr.sin_addr, ntohs (l->saddr.sin_port),
  	      wrap (this, &tcp_manager::doRPC_tcp_connect_cb, args));
  return 0;
}

void
tcp_manager::doRPC_tcp_connect_cb (RPC_delay_args *args, int fd)
{
  if (fd < 0) {
    warn << "locationtable: connect failed: " << strerror (errno) << "\n";
    (args->cb) (RPC_CANTSEND);
    return;
  }

  struct linger li;
  li.l_onoff = 1;
  li.l_linger = 0;
  setsockopt (fd, SOL_SOCKET, SO_LINGER, (char *) &li, sizeof (li));

  ptr<axprt_stream> stream_xprt = axprt_stream::alloc (fd);
  assert (stream_xprt);
  ptr<aclnt> c = aclnt::alloc (stream_xprt, args->prog);

  c->call (args->procno, args->in, args->out, 
	   wrap (this, &tcp_manager::doRPC_tcp_cleanup, args, fd));

}

void
tcp_manager::doRPC_tcp_cleanup (RPC_delay_args *args, int fd, clnt_stat err)
{
  (*args->cb)(err);
  tcp_abort (fd);
  delete args;
}

// -----------------------------------------------------
stp_manager::stp_manager (ptr<u_int32_t> _nrcv)
  : rpc_manager (_nrcv),
    a_lat (0.0),
    a_var (0.0),
    avg_lat (0.0),
    rpcdelay (0),
    seqno (0),
    cwind (1.0),
    cwind_ewma (1.0),
    ssthresh (6.0),
    left (0),
    cwind_cum (0.0),
    num_cwind_samples (0),
    num_qed (0),
    idle_timer (NULL),
    max_host_cache (100)
{
  delaycb (1, 0, wrap (this, &stp_manager::ratecb));
  reset_idle_timer ();
  st = getusec ();
  fake_seqno = -1;
}

stp_manager::~stp_manager ()
{
  if (idle_timer)
    timecb_remove (idle_timer);
}

hostinfo *
stp_manager::lookup_host (const net_address &r)
{
  hostinfo *h = hosts[r.hostname];
  if (!h) {
    if (hosts.size () > max_host_cache) {
      hostinfo *o = hostlru.first;
      hostlru.remove (o);
      delete (o);
    }
    h = New hostinfo (r);
    hostlru.insert_tail (h);
    hosts.insert (h);
  } else {
    // record recent access
    hostlru.remove (h);
    hostlru.insert_tail (h);
  }
  assert (h);
  return h;
}

void
stp_manager::ratecb () {
#if 0
  warnx << "sent " << nsent << " RPCs in the last second\n";
  warnx << "received " << *nrcv << " RPCs in the last second\n";
  warnx << npending << " RPCs are outstanding\n";
#endif
  // do something if nsent (+ nrcv) is too high xxx?

  warn << "sent " << (outbytes - nsent) << " bytes in the last second\n";
  delaycb (1, 0, wrap (this, &stp_manager::ratecb));
  nsent = outbytes;
  *nrcv = 0;
}

float
stp_manager::get_a_lat (ptr<location> l)
{
  hostinfo *h = lookup_host (l->addr);
  return h->a_lat;
}

float
stp_manager::get_a_var (ptr<location> l)
{
  hostinfo *h = lookup_host (l->addr);
  return h->a_var;
}

float
stp_manager::get_avg_lat ()
{
  return a_lat;
}

float
stp_manager::get_avg_var ()
{
  return a_var;
}

long
stp_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long _fake_seqno = 0)
{
  reset_idle_timer ();
  if (left + cwind < seqno) {
    RPC_delay_args *args = New RPC_delay_args (l, prog, procno,
					       in, out, cb);
    args->fake_seqno = fake_seqno;
    enqueue_rpc (args);
    fake_seqno--;
    return args->fake_seqno;
  } else {

    ref<aclnt> c = aclnt::alloc (dgram_xprt, prog, 
				 (sockaddr *)&(l->saddr));
    rpc_state *C = New rpc_state (l, cb, getusec (), seqno, prog.progno);
   
    C->b = rpccb_chord::alloc (c, 
			       wrap (this, &stp_manager::doRPCcb, c, C),
			       wrap (this, &stp_manager::timeout, C),
			       in,
			       out,
			       procno, 
			       (sockaddr *)&(l->saddr));
    
    sent_elm *s = New sent_elm (seqno);
    sent_Q.insert_tail (s);
    
    long sec, nsec;
    hostinfo *h = lookup_host (l->addr);
    setup_rexmit_timer (h, &sec, &nsec);

    C->b->send (sec, nsec);
    seqno++;
    nsent++;

    if (_fake_seqno) {
      C->rexmit_ID = _fake_seqno;
    } else {
      C->rexmit_ID = C->seqno;
    }

    user_rexmit_table.insert (C);
    return seqno;
  }
}

void
stp_manager::timeout (rpc_state *C)
{
#if 0
  u_int64_t now = getusec ();
  warn << "stp_manager::timeout\n"; 
  warn << "\t**now " << now << "\n";
  warn << "\tID " << C->ID << "\n";
  warn << "\ts " << C->s << "\n";
  warn << "\tprogno " << C->progno << "\n";
  warn << "\tseqno " << C->seqno << "\n";
  warn << "\trexmits " << C->rexmits << "\n";
#endif

#ifdef __J__
  warnx << gettime() << " TIMEOUT " << 1 + C->seqno << " cwind " << (int)cwind << " ssthresh " << (int)ssthresh << "\n";
#endif

  C->s = getusec ();
  rpc_done (-1);
}

void
stp_manager::doRPCcb (ref<aclnt> c, rpc_state *C, clnt_stat err)
{
  if (err) {
    nrpcfailed++;
    warn << "RPC failure: " << err << " destined for " << C->ID << "\n";
  } else {
    if (C->s > 0) {
      u_int64_t now = getusec ();
      // prevent overflow, caused by time reversal
      if (now >= C->s) {
	u_int64_t lat = now - C->s;
	update_latency (C->loc, lat, (C->progno == DHASH_PROGRAM));
      } else {
	warn << "*** Ignoring timewarp: C->s " << C->s << " > now " << now << "\n";
	warnx << " " << now       << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
      }
    }
  }

  (C->cb) (err);

  rpc_done (C->seqno);
  user_rexmit_table.remove (C);
  delete C;

}

void
stp_manager::rpc_done (long acked_seqno)
{
  // XXX acked_seqno
  if (acked_seqno < 0) {
    update_cwind (-1);
    return;
  }

#ifdef __JJ__
  warnx << gettime() << " rpc_done " << 1 + acked_seqno << "\n";
#endif

  // XXX this is all just debugging stuff... 
  //     put i suspect it has a reasonable
  //     performance impact.
  // --josh
  if (acked_seqno > 0) {
    acked_seq.push_back (acked_seqno);
    acked_time.push_back ((getusec () - st)/1000000.0);
    if (acked_seq.size () > 10000) acked_seq.pop_front ();
    if (acked_time.size () > 10000) acked_time.pop_front ();
  }

  // Ideally sent_Q should be a lru ordered hash table.
  // this might clean up the clode slight and provide
  // a slight performance boost
  // --josh
  sent_elm *s = sent_Q.first;
  while (s) {
    if (s->seqno == acked_seqno) {
      sent_Q.remove (s);
      delete s;
      break;
    }
    s = sent_Q.next (s);
  }
  
  update_cwind (acked_seqno);

  while (Q.first && (left + cwind >= seqno) ) {
    //int qsize = (num_qed > 100) ? 100 :  num_qed;
    //    int next = (int)(qsize*((float)random()/(float)RAND_MAX));
    RPC_delay_args *args = Q.first;
    //    for (int i = 0; (args) && (i < next); i++)
    //    args = Q.next (args);
    assert (args);
    Q.remove (args);

    doRPC (args->l, args->prog,
	   args->procno, 
	   args->in,
	   args->out,
	   args->cb,
	   args->fake_seqno);
    delete args;
    num_qed--;
  }

}

void
stp_manager::reset_idle_timer ()
{
  if (idle_timer) timecb_remove (idle_timer);
  idle_timer = delaycb (5, 0, wrap (this, &stp_manager::idle));
}

void
stp_manager::idle () 
{
  cwind = 1.0;
  cwind_ewma = 1.0;
  ssthresh = 6;
  idle_timer = NULL;
}

void
stp_manager::rexmit (long seqno)
{
  rpc_state *C = user_rexmit_table[seqno];
  assert (C);
  C->b->user_rexmit ();
}
void
stp_manager::update_cwind (int seq) 
{
  if (seq >= 0) {
    if (cwind < ssthresh) 
      cwind += 1.0; //slow start
    else
      cwind += 1.0/cwind; //AI
    if (seq == left) {
      if (sent_Q.first) 
	left = sent_Q.first->seqno;<commit_msg>change an assert to a warn until I can fix a fast retransmit bug<commit_after>/*
 *
 * Copyright (C) 2000 <NAME> (<EMAIL>)
 * Copyright (C) 2001 <NAME> (<EMAIL>) and 
 *                    <NAME> (<EMAIL>).
 * Copyright (C) 2001 <NAME> (<EMAIL>), 
 *                    <NAME> (<EMAIL>) and
 *                    <NAME> (<EMAIL>).
 *
 *  Permission is hereby granted, free of charge, to any person obtaining
 *  a copy of this software and associated documentation files (the
 *  "Software"), to deal in the Software without restriction, including
 *  without limitation the rights to use, copy, modify, merge, publish,
 *  distribute, sublicense, and/or sell copies of the Software, and to
 *  permit persons to whom the Software is furnished to do so, subject to
 *  the following conditions:
 *
 *  The above copyright notice and this permission notice shall be
 *  included in all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 *  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 *  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 *  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 *  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 *  OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 *  WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 *
 */


// Chord's RPC manager is designed to perform flow control on all
// Chord RPCs. It maintains a cache of hosts (i.e. IP addresses)
// that it has connected to and maintains statistics about latency
// and such to those hosts. It uses these statistics to calculate
// optimal window sizes and delays.

// #define __J__ 1

#include "dhash_prot.h"
#include "chord_util.h"
#include "location.h"

long outbytes;
// UTILITY FUNCTIONS

const int shortstats (getenv ("SHORT_STATS") ? 1 : 0);
const int aclnttrace (getenv ("ACLNT_TRACE")
		      ? atoi (getenv ("ACLNT_TRACE")) : 0);
const bool aclnttime (getenv ("ACLNT_TIME"));

static inline const char *
tracetime ()
{
  static str buf ("");
  if (aclnttime) {
    timespec ts;
    clock_gettime (CLOCK_REALTIME, &ts);
    buf = strbuf (" %d.%06d", int (ts.tv_sec), int (ts.tv_nsec/1000));
  }
  return buf;
}

static
int partition (float *A, int p, int r)
{
  float x = A[p];
  int i = p - 1;
  int j = r + 1;

  while (1) {
    do {  j -= 1; } while (A[j] > x);
    do {  i += 1; } while (A[i] < x);
    if (i >= j)
      return j;
    float tmp = A[i];
    A[i] = A[j];
    A[j] = tmp;     
  }
}

static float
myselect (float *A, int p, int r, int i)
{
  if (p == r)
    return A[p];
  int q = partition (A, p, r);
  int k = q - p + 1;
  if (i <= k)
    return myselect(A, p, q, i);
  else
    return myselect(A, q+1, r, i-k);
}

// -----------------------------------------------------
hostinfo::hostinfo (const net_address &r)
  : host (r.hostname), rpcdelay (0), nrpc (0), maxdelay (0),
    a_lat (0.0), a_var (0.0)
{
}

// -----------------------------------------------------

rpc_manager::rpc_manager (ptr<u_int32_t> _nrcv)
  : nrpc (0), nrpcfailed (0), nsent (0), npending (0), nrcv (_nrcv)
{
  int dgram_fd = inetsocket (SOCK_DGRAM);
  if (dgram_fd < 0) fatal << "Failed to allocate dgram socket\n";
  dgram_xprt = axprt_dgram::alloc (dgram_fd, sizeof(sockaddr), 230000);
  if (!dgram_xprt) fatal << "Failed to allocate dgram xprt\n";
}

float
rpc_manager::get_a_lat (ptr<location> l)
{
  return 0.0;
}

float
rpc_manager::get_a_var (ptr<location> l)
{
  return 0.0;
}

float
rpc_manager::get_avg_lat ()
{
  return 0.0;
}

float
rpc_manager::get_avg_var ()
{
  return 0.0;
}

long
rpc_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long fake_seqno = 0)
{
  ref<aclnt> c = aclnt::alloc (dgram_xprt, prog, 
			       (sockaddr *)&(l->saddr));
  
  c->call (procno, in, out, wrap (this, &rpc_manager::doRPCcb, cb)); 
  return 0;
}

void
rpc_manager::doRPCcb (aclnt_cb realcb, clnt_stat err)
{
  if (err)
    nrpcfailed++;
  else
    nrpc++;
  
  (realcb) (err);
}

// -----------------------------------------------------
long
tcp_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long fake_seqno = 0)
{
  // hack to avoid limit on wrap()'s number of arguments
  RPC_delay_args *args = New RPC_delay_args (l, prog, procno,
					     in, out, cb);

  // wierd: tcpconnect wants the address in NBO, and port in HBO
  tcpconnect (l->saddr.sin_addr, ntohs (l->saddr.sin_port),
  	      wrap (this, &tcp_manager::doRPC_tcp_connect_cb, args));
  return 0;
}

void
tcp_manager::doRPC_tcp_connect_cb (RPC_delay_args *args, int fd)
{
  if (fd < 0) {
    warn << "locationtable: connect failed: " << strerror (errno) << "\n";
    (args->cb) (RPC_CANTSEND);
    return;
  }

  struct linger li;
  li.l_onoff = 1;
  li.l_linger = 0;
  setsockopt (fd, SOL_SOCKET, SO_LINGER, (char *) &li, sizeof (li));

  ptr<axprt_stream> stream_xprt = axprt_stream::alloc (fd);
  assert (stream_xprt);
  ptr<aclnt> c = aclnt::alloc (stream_xprt, args->prog);

  c->call (args->procno, args->in, args->out, 
	   wrap (this, &tcp_manager::doRPC_tcp_cleanup, args, fd));

}

void
tcp_manager::doRPC_tcp_cleanup (RPC_delay_args *args, int fd, clnt_stat err)
{
  (*args->cb)(err);
  tcp_abort (fd);
  delete args;
}

// -----------------------------------------------------
stp_manager::stp_manager (ptr<u_int32_t> _nrcv)
  : rpc_manager (_nrcv),
    a_lat (0.0),
    a_var (0.0),
    avg_lat (0.0),
    rpcdelay (0),
    seqno (0),
    cwind (1.0),
    cwind_ewma (1.0),
    ssthresh (6.0),
    left (0),
    cwind_cum (0.0),
    num_cwind_samples (0),
    num_qed (0),
    idle_timer (NULL),
    max_host_cache (100)
{
  delaycb (1, 0, wrap (this, &stp_manager::ratecb));
  reset_idle_timer ();
  st = getusec ();
  fake_seqno = -1;
}

stp_manager::~stp_manager ()
{
  if (idle_timer)
    timecb_remove (idle_timer);
}

hostinfo *
stp_manager::lookup_host (const net_address &r)
{
  hostinfo *h = hosts[r.hostname];
  if (!h) {
    if (hosts.size () > max_host_cache) {
      hostinfo *o = hostlru.first;
      hostlru.remove (o);
      delete (o);
    }
    h = New hostinfo (r);
    hostlru.insert_tail (h);
    hosts.insert (h);
  } else {
    // record recent access
    hostlru.remove (h);
    hostlru.insert_tail (h);
  }
  assert (h);
  return h;
}

void
stp_manager::ratecb () {
#if 0
  warnx << "sent " << nsent << " RPCs in the last second\n";
  warnx << "received " << *nrcv << " RPCs in the last second\n";
  warnx << npending << " RPCs are outstanding\n";
#endif
  // do something if nsent (+ nrcv) is too high xxx?

  warn << "sent " << (outbytes - nsent) << " bytes in the last second\n";
  delaycb (1, 0, wrap (this, &stp_manager::ratecb));
  nsent = outbytes;
  *nrcv = 0;
}

float
stp_manager::get_a_lat (ptr<location> l)
{
  hostinfo *h = lookup_host (l->addr);
  return h->a_lat;
}

float
stp_manager::get_a_var (ptr<location> l)
{
  hostinfo *h = lookup_host (l->addr);
  return h->a_var;
}

float
stp_manager::get_avg_lat ()
{
  return a_lat;
}

float
stp_manager::get_avg_var ()
{
  return a_var;
}

long
stp_manager::doRPC (ptr<location> l,
		    rpc_program prog, int procno, 
		    ptr<void> in, void *out, aclnt_cb cb,
		    long _fake_seqno = 0)
{
  reset_idle_timer ();
  if (left + cwind < seqno) {
    RPC_delay_args *args = New RPC_delay_args (l, prog, procno,
					       in, out, cb);
    args->fake_seqno = fake_seqno;
    enqueue_rpc (args);
    fake_seqno--;
    return args->fake_seqno;
  } else {

    ref<aclnt> c = aclnt::alloc (dgram_xprt, prog, 
				 (sockaddr *)&(l->saddr));
    rpc_state *C = New rpc_state (l, cb, getusec (), seqno, prog.progno);
   
    C->b = rpccb_chord::alloc (c, 
			       wrap (this, &stp_manager::doRPCcb, c, C),
			       wrap (this, &stp_manager::timeout, C),
			       in,
			       out,
			       procno, 
			       (sockaddr *)&(l->saddr));
    
    sent_elm *s = New sent_elm (seqno);
    sent_Q.insert_tail (s);
    
    long sec, nsec;
    hostinfo *h = lookup_host (l->addr);
    setup_rexmit_timer (h, &sec, &nsec);

    C->b->send (sec, nsec);
    seqno++;
    nsent++;

    if (_fake_seqno) {
      C->rexmit_ID = _fake_seqno;
    } else {
      C->rexmit_ID = C->seqno;
    }

    user_rexmit_table.insert (C);
    return seqno;
  }
}

void
stp_manager::timeout (rpc_state *C)
{
#if 0
  u_int64_t now = getusec ();
  warn << "stp_manager::timeout\n"; 
  warn << "\t**now " << now << "\n";
  warn << "\tID " << C->ID << "\n";
  warn << "\ts " << C->s << "\n";
  warn << "\tprogno " << C->progno << "\n";
  warn << "\tseqno " << C->seqno << "\n";
  warn << "\trexmits " << C->rexmits << "\n";
#endif

#ifdef __J__
  warnx << gettime() << " TIMEOUT " << 1 + C->seqno << " cwind " << (int)cwind << " ssthresh " << (int)ssthresh << "\n";
#endif

  C->s = getusec ();
  rpc_done (-1);
}

void
stp_manager::doRPCcb (ref<aclnt> c, rpc_state *C, clnt_stat err)
{
  if (err) {
    nrpcfailed++;
    warn << "RPC failure: " << err << " destined for " << C->ID << "\n";
  } else {
    if (C->s > 0) {
      u_int64_t now = getusec ();
      // prevent overflow, caused by time reversal
      if (now >= C->s) {
	u_int64_t lat = now - C->s;
	update_latency (C->loc, lat, (C->progno == DHASH_PROGRAM));
      } else {
	warn << "*** Ignoring timewarp: C->s " << C->s << " > now " << now << "\n";
	warnx << " " << now       << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
	warnx << " " << getusec() << "\n";
      }
    }
  }

  (C->cb) (err);

  rpc_done (C->seqno);
  user_rexmit_table.remove (C);
  delete C;

}

void
stp_manager::rpc_done (long acked_seqno)
{
  // XXX acked_seqno
  if (acked_seqno < 0) {
    update_cwind (-1);
    return;
  }

#ifdef __JJ__
  warnx << gettime() << " rpc_done " << 1 + acked_seqno << "\n";
#endif

  // XXX this is all just debugging stuff... 
  //     put i suspect it has a reasonable
  //     performance impact.
  // --josh
  if (acked_seqno > 0) {
    acked_seq.push_back (acked_seqno);
    acked_time.push_back ((getusec () - st)/1000000.0);
    if (acked_seq.size () > 10000) acked_seq.pop_front ();
    if (acked_time.size () > 10000) acked_time.pop_front ();
  }

  // Ideally sent_Q should be a lru ordered hash table.
  // this might clean up the clode slight and provide
  // a slight performance boost
  // --josh
  sent_elm *s = sent_Q.first;
  while (s) {
    if (s->seqno == acked_seqno) {
      sent_Q.remove (s);
      delete s;
      break;
    }
    s = sent_Q.next (s);
  }
  
  update_cwind (acked_seqno);

  while (Q.first && (left + cwind >= seqno) ) {
    //int qsize = (num_qed > 100) ? 100 :  num_qed;
    //    int next = (int)(qsize*((float)random()/(float)RAND_MAX));
    RPC_delay_args *args = Q.first;
    //    for (int i = 0; (args) && (i < next); i++)
    //    args = Q.next (args);
    assert (args);
    Q.remove (args);

    doRPC (args->l, args->prog,
	   args->procno, 
	   args->in,
	   args->out,
	   args->cb,
	   args->fake_seqno);
    delete args;
    num_qed--;
  }

}

void
stp_manager::reset_idle_timer ()
{
  if (idle_timer) timecb_remove (idle_timer);
  idle_timer = delaycb (5, 0, wrap (this, &stp_manager::idle));
}

void
stp_manager::idle () 
{
  cwind = 1.0;
  cwind_ewma = 1.0;
  ssthresh = 6;
  idle_timer = NULL;
}

void
stp_manager::rexmit (long seqno)
{
  rpc_state *C = user_rexmit_table[seqno];
  if (!C) {
    warn << seqno << "retransmit ID  not present for retransmission\n";
    return;
  }
  C->b->user_rexmit ();
}
void
stp_manager::update_cwind (int seq) 
{
  if (seq >= 0) {
    if (cwind < ssthresh) 
      cwind += 1.0; //slow start
    else
      cwind += 1.0/cwind; //AI
    if (seq == left) {
      if (sent_Q.first) 
	left = sent_Q.first->seqno;
{'fraction_non_alphanumeric': 0.17813579784355354, 'fraction_numerical': 0.012239290620706882, 'mean_word_length': 2.6402485225034096, 'pattern_counts': {'":': 0, '<': 429, '<?xml version=': 0, '>': 193, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4348220', 'n_tokens_mistral': 9806, 'n_tokens_neox': 9300, 'n_words': 2507}