/*
 * As its sole author, I explicitly place this program in the public domain.
 * It may be used by anyone in any way for any purpose, though I would
 * appreciate credit where it's due.
 *		der Mouse, mouse@rodents.montreal.qc.ca, 1998-08-05
 */

#include <poll.h>
#include <time.h>
#include <stdio.h>
#include <errno.h>
#include <netdb.h>
#include <paths.h>
#include <stdlib.h>
#include <signal.h>
#include <unistd.h>
#include <strings.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

extern const char *__progname;

#include "oq.h"
#include "pollloop.h"

/* max # bytes we're willing to buffer */
#define MAXBUFFERED 10240

static const char *arg1 = 0;
static const char *arg2 = 0;

static int nowait_in = 0;
static int nowait_net = 0;
static int hold_in = 0;
static int hold_net = 0;
static int time_in = 0;
static int time_net = 0;
static int reuse_a = 0;
static int reuse_p = 0;
static int keepalive = 0;
static int nodelay = 0;
static int retry = 0;
static int reblock = 0;
static int lockstep = 0;
static const char *srcspec = 0;
static const char *cmd = 0;
static int server = 0; /* 1 for -server1, 2 for -server */
static int protos = 0;
#define PROTO_IPV4 1
#define PROTO_IPV6 2

enum lock_state {
  LS_IDLE = 1,
  LS_LEN1,
  LS_DATA,
  LS_XCPT
  } ;

typedef enum lock_state LOCK_STATE;

static int net;
static OQ oq_net;
static OQ oq_out;
static int goteof_in;
static int goteof_net;
static struct timeval last_in;
static struct timeval last_net;
static struct timeval now_;
static int now_valid;
static int delay_net;
static struct timeval delay_net_time;
static int delay_out;
static struct timeval delay_out_time;
static int block_in;
static LOCK_STATE lock_state;
static int lock_len;
static int lock_eof;
static int until_ping;

#ifdef NO_INET_ATON
static int inet_aton(const char *cp, struct in_addr *inp)
{
 unsigned long int a;

 a = inet_addr(cp);
 if (a == (unsigned long int)-1UL) return(0);
 inp->s_addr = a;
 return(1);
}
#endif

static void handleargs(int ac, char **av)
{
 int skip;
 int errs;

 skip = 0;
 errs = 0;
 for (ac--,av++;ac;ac--,av++)
  { if (skip > 0)
     { skip --;
       continue;
     }
    if (**av != '-')
     { if (! arg1)
	{ arg1 = *av;
	}
       else if (! arg2)
	{ arg2 = *av;
	}
       else
	{ fprintf(stderr,"%s: extra argument `%s'\n",__progname,*av);
	  errs ++;
	}
       continue;
     }
    if (0)
     {
needarg:;
       fprintf(stderr,"%s: %s needs a following argument\n",__progname,*av);
       errs ++;
       continue;
     }
#define WANTARG() do { if (++skip >= ac) goto needarg; } while (0)
    if (!strcmp(*av,"-stdin"))
     { nowait_in = 1;
       continue;
     }
    if (!strcmp(*av,"-net"))
     { nowait_net = 1;
       continue;
     }
    if (!strcmp(*av,"-holdstdin"))
     { hold_in = 1;
       continue;
     }
    if (!strcmp(*av,"-holdnet"))
     { hold_net = 1;
       continue;
     }
    if (!strcmp(*av,"-timestdin"))
     { WANTARG();
       time_in = atoi(av[skip]);
       continue;
     }
    if (!strcmp(*av,"-timenet"))
     { WANTARG();
       time_net = atoi(av[skip]);
       continue;
     }
    if (!strcmp(*av,"-ra"))
     { reuse_a = 1;
       continue;
     }
    if (!strcmp(*av,"-rp"))
     {
#ifdef SO_REUSEPORT
       reuse_p = 1;
#else
       fprintf(stderr,"%s: -rp not supported in this build (no SO_REUSEPORT)\n",__progname);
       errs ++;
#endif
       continue;
     }
    if (!strcmp(*av,"-ka"))
     { keepalive = 1;
       continue;
     }
    if (!strcmp(*av,"-nd"))
     { nodelay = 1;
       continue;
     }
    if (!strcmp(*av,"-4"))
     { protos |= PROTO_IPV4;
       continue;
     }
    if (!strcmp(*av,"-6"))
     { protos |= PROTO_IPV6;
       continue;
     }
    if (!strcmp(*av,"-any"))
     { protos = 0;
       continue;
     }
    if (!strcmp(*av,"-retry"))
     { retry = 1;
       continue;
     }
    if (!strcmp(*av,"-reblock"))
     { reblock = 1;
       continue;
     }
    if (!strcmp(*av,"-src"))
     { WANTARG();
       srcspec = av[skip];
       continue;
     }
    if (!strcmp(*av,"-cmd"))
     { WANTARG();
       cmd = av[skip];
       continue;
     }
    if (!strcmp(*av,"-server"))
     { WANTARG();
       cmd = av[skip];
       server = 2;
       continue;
     }
    if (!strcmp(*av,"-server1"))
     { WANTARG();
       cmd = av[skip];
       server = 1;
       continue;
     }
    if (!strcmp(*av,"-lockstep"))
     { WANTARG();
       lockstep = atoi(av[skip]);
       if (lockstep < 8)
	{ fprintf(stderr,"%s: -lockstep value must be at least 8\n",__progname);
	  errs ++;
	}
       continue;
     }
#undef WANTARG
    fprintf(stderr,"%s: unrecognized option `%s'\n",__progname,*av);
    errs ++;
  }
 if (errs) exit(1);
}

static __inline__ int tvcmp(struct timeval a, struct timeval b)
{
 if (a.tv_sec < b.tv_sec) return(-1);
 if (a.tv_sec > b.tv_sec) return(1);
 if (a.tv_usec < b.tv_usec) return(-1);
 if (a.tv_usec > b.tv_usec) return(1);
 return(0);
}

static __inline__ int tvsub_ms_clip(struct timeval a, struct timeval b)
{
 if (a.tv_usec >= b.tv_usec)
  { if (a.tv_sec < b.tv_sec) return(0);
    return( ((a.tv_sec - b.tv_sec) * 1000) +
	    ((a.tv_usec - b.tv_usec) / 1000) );
  }
 else
  { if (a.tv_sec <= b.tv_sec) return(0);
    return( ((a.tv_sec - 1 - b.tv_sec) * 1000) +
	    ((a.tv_usec + 1000000 - b.tv_usec) / 1000) );
  }
}

static void set_nonblocking(int fd)
{
 static int on = 1;

 ioctl(fd,FIONBIO,&on);
}

static void run_cmd(void)
{
 const char *shell;

 if (reblock || lockstep)
  { pid_t kid;
    int ipipe[2];
    int opipe[2];
    if ( (pipe(&ipipe[0]) < 0) ||
	 (pipe(&opipe[0]) < 0) )
     { fprintf(stderr,"%s: pipe: %s\n",__progname,strerror(errno));
       exit(1);
     }
    kid = fork();
    if (kid == 0)
     { close(ipipe[1]);
       close(opipe[0]);
       if (ipipe[0] != 0)
	{ dup2(ipipe[0],0);
	  close(ipipe[0]);
	}
       if (opipe[1] != 1)
	{ dup2(opipe[1],1);
	  close(opipe[1]);
	}
       close(net);
     }
    else
     { close(ipipe[0]);
       close(opipe[1]);
       if (opipe[0] != 0)
	{ dup2(opipe[0],0);
	  close(opipe[0]);
	}
       if (ipipe[1] != 1)
	{ dup2(ipipe[1],1);
	  close(ipipe[1]);
	}
       return;
     }
  }
 else
  { if (net != 0)
     { dup2(net,0);
       close(net);
     }
    dup2(0,1);
  }
 shell = getenv("SHELL");
 if (shell == 0) shell = _PATH_BSHELL;
 execl(shell,shell,"-c",cmd,(char *)0);
 fprintf(stderr,"%s: can't exec %s for command: %s\n",__progname,shell,strerror(errno));
 exit(1);
}

static int proto_not_listed(int pf)
{
 if (protos == 0) return(0);
 switch (pf)
  { case PF_INET:
       if (protos & PROTO_IPV4) return(0);
       break;
    case PF_INET6:
       if (protos & PROTO_IPV6) return(0);
       break;
  }
 return(1);
}

static void setup_connect(const char *host, const char *port)
{
 struct addrinfo hints;
 struct addrinfo *ai0;
 struct addrinfo *ai;
 int e;
 int se;
 int s;
 char *ssslash;
 const char *ssh;
 const char *ssp;
 struct addrinfo *ssai0;
 struct addrinfo *ssai;
 char hnf[NI_MAXHOST];
 char pnf[NI_MAXSERV];
 char hnt[NI_MAXHOST];
 char pnt[NI_MAXSERV];

 if (srcspec == 0)
  { ssh = 0;
    ssp = 0;
  }
 else
  { char *f;
    int l;
    f = 0;
    ssslash = index(srcspec,'/');
    if (ssslash)
     { l = ssslash - srcspec;
       if (l == 0)
	{ ssh = 0;
	}
       else
	{ f = malloc(l+1);
	  bcopy(srcspec,f,l);
	  f[l] = '\0';
	  ssh = f;
	}
       ssp = ssslash + 1;
       if (! *ssp) ssp = 0;
     }
    else
     { ssh = srcspec;
       ssp = 0;
     }
  }
 if (ssh || ssp)
  { hints.ai_flags = AI_PASSIVE;
    hints.ai_family = PF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_protocol = 0;
    hints.ai_addrlen = 0;
    hints.ai_canonname = 0;
    hints.ai_addr = 0;
    hints.ai_next = 0;
    e = getaddrinfo(ssh,ssp,&hints,&ssai0);
    if (e)
     { fprintf(stderr,"%s: %s/%s: %s\n",__progname,ssh?:"",ssp?:"",gai_strerror(e));
       exit(1);
     }
    if (! ssai0)
     { fprintf(stderr,"%s: %s/%s: successful lookup but no addresses?\n",__progname,ssh?:"",ssp?:"");
       exit(1);
     }
  }
 else
  { ssai0 = 0;
  }
 hints.ai_flags = 0;
 hints.ai_family = PF_UNSPEC;
 hints.ai_socktype = SOCK_STREAM;
 hints.ai_protocol = 0;
 hints.ai_addrlen = 0;
 hints.ai_canonname = 0;
 hints.ai_addr = 0;
 hints.ai_next = 0;
 e = getaddrinfo(host,port,&hints,&ai0);
 if (e)
  { fprintf(stderr,"%s: %s/%s: %s\n",__progname,host,port,gai_strerror(e));
    exit(1);
  }
 if (! ai0)
  { fprintf(stderr,"%s: %s: successful lookup but no addresses?\n",__progname,strerror(se));
    exit(1);
  }
 se = 0;
 do
  { for (ai=ai0;ai;ai=ai->ai_next)
     { if (ssai0)
	{ if (proto_not_listed(ai->ai_family)) continue;
	  for (ssai=ssai0;ssai;ssai=ssai->ai_next)
	   { if (ssai->ai_family != ai->ai_family) continue;
	     s = socket(ai->ai_family,ai->ai_socktype,ai->ai_protocol);
	     if (s < 0)
	      { if (se == 0) se = errno;
		continue;
	      }
	     se = -1;
	     if (bind(s,ssai->ai_addr,ssai->ai_addrlen) < 0)
	      { e = errno;
		if (getnameinfo(ssai->ai_addr,ssai->ai_addrlen,&hnf[0],NI_MAXHOST,&pnf[0],NI_MAXSERV,NI_NUMERICHOST|NI_NUMERICSERV))
		 { fprintf(stderr,"%s: %s/%s: bind failed [%s], can't get numeric hostname info [%s]\n",__progname,ssh?:"",ssp?:"",strerror(e),strerror(errno));
		 }
		else
		 { fprintf(stderr,"%s: bind to %s/%s: %s\n",__progname,&hnf[0],&pnf[0],strerror(e));
		 }
		close(s);
		continue;
	      }
	     if (connect(s,ai->ai_addr,ai->ai_addrlen) < 0)
	      { char *fs;
		char *ts;
		if (! retry)
		 { e = errno;
		   if (getnameinfo(ssai->ai_addr,ssai->ai_addrlen,&hnf[0],NI_MAXHOST,&pnf[0],NI_MAXSERV,NI_NUMERICHOST|NI_NUMERICSERV))
		    { asprintf(&fs,"%s/%s[%s]",ssh?:"",ssp?:"",strerror(errno));
		    }
		   else
		    { asprintf(&fs,"%s/%s",&hnf[0],&pnf[0]);
		    }
		   if (getnameinfo(ai->ai_addr,ai->ai_addrlen,&hnt[0],NI_MAXHOST,&pnt[0],NI_MAXSERV,NI_NUMERICHOST|NI_NUMERICSERV))
		    { asprintf(&ts,"%s%s%s[%s]",host,port?"/":"",port?:"",strerror(errno));
		    }
		   else
		    { asprintf(&ts,"%s/%s",&hnt[0],&pnt[0]);
		    }
		   fprintf(stderr,"%s: connect %s -> %s: %s\n",__progname,fs,ts,strerror(e));
		   free(fs);
		   free(ts);
		 }
		close(s);
		continue;
	      }
	     freeaddrinfo(ssai0);
	     freeaddrinfo(ai0);
	     net = s;
	     return;
	   }
	}
       else
	{ if (proto_not_listed(ai->ai_family)) continue;
	  s = socket(ai->ai_family,ai->ai_socktype,ai->ai_protocol);
	  if (s < 0)
	   { if (se == 0) se = errno;
	     continue;
	   }
	  se = -1;
	  if (connect(s,ai->ai_addr,ai->ai_addrlen) < 0)
	   { char hnbuf[NI_MAXHOST];
	     char pnbuf[NI_MAXSERV];
	     if (! retry)
	      { e = errno;
		if (getnameinfo(ai->ai_addr,ai->ai_addrlen,&hnbuf[0],NI_MAXHOST,&pnbuf[0],NI_MAXSERV,NI_NUMERICHOST|NI_NUMERICSERV))
		 { fprintf(stderr,"%s: %s%s%s: connect failed [%s], can't get numeric hostname info [%s]\n",__progname,host,port?"/":"",port?:"",strerror(e),strerror(errno));
		 }
		else
		 { if (ai0->ai_next)
		    { fprintf(stderr,"%s: connect %s [%s/%s]: %s\n",__progname,host,&hnbuf[0],&pnbuf[0],strerror(e));
		    }
		   else
		    { fprintf(stderr,"%s: connect %s/%s: %s\n",__progname,&hnbuf[0],&pnbuf[0],strerror(e));
		    }
		 }
	      }
	     close(s);
	     continue;
	   }
	  freeaddrinfo(ai0);
	  net = s;
	  return;
	}
     }
  } while (retry && (sleep(5),1));
 if (se > 0)
  { fprintf(stderr,"%s: socket: %s\n",__progname,strerror(se));
  }
 else if (se == 0)
  { fprintf(stderr,"%s: source and destination incompatible\n",__progname);
  }
 exit(1);
}

static void setup_accept(const char *port)
{
 typedef struct acc ACC;
 struct acc {
   ACC *link;
   char *txt;
   int fd;
   int px;
   } ;
 struct addrinfo hints;
 struct addrinfo *ai0;
 struct addrinfo *ai;
 struct sockaddr_storage from;
 socklen_t fromlen;
 const char *hostpart;
 ACC *alist;
 ACC *a;
 int nacc;
 struct pollfd *pfds;
 int s;
 int se;
 int i;
 int haderr;
 pid_t pid;
 pid_t pid2;
 char hn[NI_MAXHOST];
 char pn[NI_MAXSERV];

 hints.ai_flags = AI_PASSIVE;
 hints.ai_family = PF_UNSPEC;
 hints.ai_socktype = SOCK_STREAM;
 hints.ai_protocol = 0;
 hints.ai_addrlen = 0;
 hints.ai_canonname = 0;
 hints.ai_addr = 0;
 hints.ai_next = 0;
 if (srcspec)
  { char *slash;
    slash = index(srcspec,'/');
    if (slash)
     { char *t;
       i = slash - srcspec;
       t = malloc(i+1);
       bcopy(srcspec,t,i);
       t[i] = '\0';
       if (getaddrinfo(t,port,&hints,&ai0))
	{ fprintf(stderr,"%s: %s/%s: %s\n",__progname,t,port,gai_strerror(errno));
	  exit(1);
	}
       hostpart = t;
     }
    else
     { if (getaddrinfo(srcspec,port,&hints,&ai0))
	{ fprintf(stderr,"%s: %s/%s: %s\n",__progname,srcspec,port,gai_strerror(errno));
	  exit(1);
	}
       hostpart = srcspec;
     }
  }
 else
  { if (getaddrinfo(0,port,&hints,&ai0))
     { fprintf(stderr,"%s: %s: %s\n",__progname,port,gai_strerror(errno));
       exit(1);
     }
    hostpart = 0;
  }
 if (! ai0)
  { fprintf(stderr,"%s: %s: successful lookup but no addresses?\n",__progname,port);
    exit(1);
  }
 haderr = 0;
 alist = 0;
 nacc = 0;
 se = 0;
 for (ai=ai0;ai;ai=ai->ai_next)
  { if (proto_not_listed(ai->ai_family)) continue;
    s = socket(ai->ai_family,ai->ai_socktype,ai->ai_protocol);
    if (s < 0)
     { if (se == 0) se = errno;
       continue;
     }
    se = -1;
    if (getnameinfo(ai->ai_addr,ai->ai_addrlen,&hn[0],NI_MAXHOST,&pn[0],NI_MAXSERV,NI_NUMERICHOST|NI_NUMERICSERV))
     { fprintf(stderr,"%s: can't get printable form: %s\n",__progname,strerror(errno));
       exit(1);
     }
    if (reuse_a)
     { i = 1;
       if (setsockopt(s,SOL_SOCKET,SO_REUSEADDR,&i,sizeof(i)) < 0)
	{ fprintf(stderr,"%s: setsockopt SO_REUSEADDR: %s\n",__progname,strerror(errno));
	  exit(1);
	}
     }
    if (reuse_p)
     {
#ifdef SO_REUSEPORT
       i = 1;
       if (setsockopt(s,SOL_SOCKET,SO_REUSEPORT,&i,sizeof(i)) < 0)
	{ fprintf(stderr,"%s: setsockopt SO_REUSEPORT: %s\n",__progname,strerror(errno));
	  exit(1);
	}
#else
       abort();
#endif
     }
    if (bind(s,ai->ai_addr,ai->ai_addrlen) < 0)
     { fprintf(stderr,"%s: bind: %s\n",__progname,strerror(errno));
       close(s);
       haderr = 1;
       continue;
     }
    listen(s,10);
    a = malloc(sizeof(ACC));
    a->link = alist;
    alist = a;
    a->fd = s;
    if (hostpart)
     { asprintf(&a->txt,"%s/%s",&hn[0],&pn[0]);
     }
    else
     { asprintf(&a->txt,"/%s",&pn[0]);
     }
    nacc ++;
  }
 if (nacc < 1)
  { if (! haderr)
     { fprintf(stderr,"%s: no addresses to listen on!\n",__progname);
     }
    exit(1);
  }
 else
  { if (haderr)
     { fprintf(stderr,"%s: note: listening on other %s\n",__progname,alist->link?"addresses":"address");
     }
  }
 pfds = malloc(nacc*sizeof(struct pollfd));
 while (1)
  { for (i=0,a=alist;a;i++,a=a->link)
     { a->px = i;
       if (i >= nacc) abort();
       pfds[i].fd = a->fd;
       pfds[i].events = POLLIN | POLLRDNORM;
     }
    if (i != nacc) abort();
    i = poll(pfds,nacc,INFTIM);
    if (i < 0)
     { if (errno == EINTR) continue;
       fprintf(stderr,"%s: poll: %s\n",__progname,strerror(errno));
       exit(1);
     }
    for (a=alist;a;a=a->link)
     { if (pfds[a->px].revents & (POLLIN|POLLRDNORM))
	{ fromlen = sizeof(from);
	  i = accept(a->fd,(struct sockaddr *)&from,&fromlen);
	  if (i < 0)
	   { fprintf(stderr,"%s: accept %s: %s\n",__progname,a->txt,strerror(errno));
	     exit(1);
	   }
	  switch (server)
	   { case 0:
		pid = 0;
		break;
	     case 1:
		pid = fork();
		break;
	     case 2:
		pid = fork();
		if (pid == 0)
		 { pid2 = fork();
		   if (pid2 == 0) break;
		   exit(0);
		 }
		break;
	   }
	  if (pid == 0)
	   { for (a=alist;a;a=a->link) close(a->fd);
	     free(pfds);
	     net = i;
	     return;
	   }
	  close(i);
	  while (1)
	   { pid2 = wait(0);
	     if (pid2 == pid) break;
	     if (pid2 < 0)
	      { if (errno == EINTR) continue;
		if (errno == ECHILD) break;
		fprintf(stderr,"%s: wait: %s\n",__progname,strerror(errno));
		exit(1);
	      }
	   }
	}
     }
  }
}

static struct timeval now(void)
{
 if (! now_valid)
  { gettimeofday(&now_,0);
    now_valid = 1;
  }
 return(now_);
}

static void got_eof_in(void)
{
 if (shutdown(0,SHUT_RD) < 0) close(0);
 goteof_in = 1;
}

static void got_eof_net(void)
{
 shutdown(net,SHUT_RD);
 goteof_net = 1;
}

static void put_eof_out(void)
{
 if (shutdown(1,SHUT_WR) < 0) close(1);
}

static void put_eof_net(void)
{
 shutdown(net,SHUT_WR);
}

static void maybe_eof_net(void)
{
 if (goteof_in && !hold_in && oq_empty(&oq_net)) put_eof_net();
}

static void maybe_eof_out(void)
{
 if (goteof_net && !hold_net && oq_empty(&oq_out)) put_eof_out();
}

static int timeout_delay_net(int id __attribute__((__unused__)))
{
 int n;

 if (! delay_net) return(BLOCK_NIL);
 n = tvsub_ms_clip(delay_net_time,now());
 if (n < 1)
  { delay_net = 0;
    return(BLOCK_LOOP);
  }
 return(n);
}

static int timeout_delay_out(int id __attribute__((__unused__)))
{
 int n;

 if (! delay_out) return(BLOCK_NIL);
 n = tvsub_ms_clip(delay_out_time,now());
 if (n < 1)
  { delay_out = 0;
    return(BLOCK_LOOP);
  }
 return(n);
}

/*
 * Lockstep protocol:
 *
 * Sender writes data as <length><data>, where <length> is a byte
 *  count, two bytes in big endian, and <data> is <length> bytes of
 *  data.  <length> must be 1 through 0xffff.  A <length> value of 0 is
 *  an exception.  In this case, one byte of data follows which
 *  indicates what kind of exception it is.  Exception kinds are:
 *
 *	0x00	Ping.
 *	0x01	Pong.
 *	0x02	EOF.
 *
 * A ping exception is sent when the sender wants to interlock with the
 *  receiver.  Upon receiving a ping, the other end generates a pong.
 *  An EOF exception indicates the end of the data stream; closing the
 *  data stream without sending an EOF exception is an error.
 *
 * In order to support the ping/pong protocol, we can't shutdown() the
 *  connection before we die.  This is why the EOF exception exists.
 */
#define LSX_PING 0x00
#define LSX_PONG 0x01
#define LSX_EOF  0x02
static const unsigned char xcpt_ping[]
 = { 0, 0, LSX_PING };
static const unsigned char xcpt_pong[]
 = { 0, 0, LSX_PONG };
static const unsigned char xcpt_eof[]
 = { 0, 0, LSX_EOF };

static int rtest_lockstep_in(void)
{
 return(!goteof_in && !block_in && (oq_qlen(&oq_net) < MAXBUFFERED));
}

static void rd_lockstep_in(void)
{
 unsigned char rbuf[8192];
 int w;
 int r;
 unsigned char len[2];

 if (until_ping < 8)
  { oq_queue_point(&oq_net,&xcpt_ping,sizeof(xcpt_ping));
    block_in = 1;
    return;
  }
 w = until_ping - 8;
 if (w > sizeof(rbuf)) w = sizeof(rbuf);
 r = read(0,&rbuf[0],w);
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: stdin read error: %s\n",__progname,strerror(errno));
    r = 0;
  }
 if (r == 0)
  { got_eof_in();
    oq_queue_point(&oq_net,&xcpt_eof,sizeof(xcpt_eof));
    hold_in = 1;
    return;
  }
 if (r > 65535) abort();
 len[0] = r >> 8;
 len[1] = r & 0xff;
 oq_queue_copy(&oq_net,&len[0],2);
 oq_queue_copy(&oq_net,&rbuf[0],r);
 until_ping -= r + 2;
}

static int rtest_lockstep_net(void)
{
 return((!goteof_net || block_in) && (oq_qlen(&oq_out) < MAXBUFFERED));
}

static void rd_lockstep_net(void)
{
 unsigned char rbuf[8192];
 int r;
 int o;
 int n;

 if (goteof_net && !block_in) abort();
 r = read(net,&rbuf[0],sizeof(rbuf));
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: net read error: %s\n",__progname,strerror(errno));
    r = 0;
  }
 if (r == 0)
  { fprintf(stderr,"%s: unexpected net EOF\n",__progname);
    exit(1);
  }
 o = 0;
 while (o < r)
  { switch (lock_state)
     { case LS_IDLE:
	  lock_len = rbuf[o++];
	  lock_state = LS_LEN1;
	  break;
       case LS_LEN1:
	  lock_len = (lock_len << 8) | rbuf[o++];
	  lock_state = (lock_len == 0) ? LS_XCPT : LS_DATA;
	  break;
       case LS_DATA:
	  if (lock_eof)
	   { fprintf(stderr,"%s: lockstep protocol error, data after EOF\n",__progname);
	     exit(1);
	   }
	  n = r - o;
	  if (n > lock_len) n = lock_len;
	  oq_queue_copy(&oq_out,&rbuf[o],n);
	  o += n;
	  lock_len -= n;
	  if (lock_len < 1) lock_state = LS_IDLE;
	  break;
       case LS_XCPT:
	  switch (rbuf[o++])
	   { case LSX_PING:
		oq_queue_point(&oq_net,&xcpt_pong,sizeof(xcpt_pong));
		lock_state = LS_IDLE;
		break;
	     case LSX_PONG:
		block_in = 0;
		until_ping = lockstep;
		lock_state = LS_IDLE;
		break;
	     case LSX_EOF:
		lock_state = LS_IDLE;
		lock_eof = 1;
		goteof_net = 1;
		maybe_eof_out();
		break;
	     default:
		fprintf(stderr,"%s: lockstep protocol error, exception 0x%02x\n",__progname,rbuf[o-1]);
		exit(1);
		break;
	   }
	  break;
     }
  }
}

static int wtest_out(void)
{
 return(oq_nonempty(&oq_out) && (!reblock || !delay_out));
}

static void wr_out(void)
{
 int n;

 if (reblock)
  { if (random() & 1)
     { delay_out = 1;
       delay_out_time = now();
       delay_out_time.tv_usec += random() & 0x7ffff;
       if (delay_out_time.tv_usec >= 1000000)
	{ delay_out_time.tv_usec -= 1000000;
	  delay_out_time.tv_sec ++;
	}
       return;
     }
  }
 n = oq_qlen(&oq_out);
 if (reblock && (random() & 3)) n = (random() % n) + 1;
 n = oq_writev(&oq_out,1,n);
 if (n < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: output write: %s\n",__progname,strerror(errno));
    got_eof_net();
    oq_flush(&oq_out);
    put_eof_out();
    return;
  }
 if (oq_dropdata(&oq_out,n)) maybe_eof_out();
}

static int wtest_net(void)
{
 return(oq_nonempty(&oq_net) && (!reblock || !delay_net));
}

static void wr_net(void)
{
 int n;

 if (reblock)
  { if (random() & 1)
     { delay_net = 1;
       delay_net_time = now();
       delay_net_time.tv_usec += n = random() & 0x7ffff;
       if (delay_net_time.tv_usec >= 1000000)
	{ delay_net_time.tv_usec -= 1000000;
	  delay_net_time.tv_sec ++;
	}
       return;
     }
  }
 n = oq_qlen(&oq_net);
 if (reblock && (random() & 3)) n = (random() % n) + 1;
 n = oq_writev(&oq_net,net,n);
 if (n < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: net write: %s\n",__progname,strerror(errno));
    got_eof_in();
    oq_flush(&oq_net);
    put_eof_net();
    return;
  }
 if (oq_dropdata(&oq_net,n)) maybe_eof_net();
}

static int rtest_copy_in(void)
{
 return(!goteof_in && (oq_qlen(&oq_net) < MAXBUFFERED));
}

static void rd_copy_in(void)
{
 char rbuf[8192];
 int r;

 if (goteof_in) abort();
 r = read(0,&rbuf[0],sizeof(rbuf));
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: stdin read error: %s\n",__progname,strerror(errno));
    r = 0;
  }
 if (r == 0)
  { got_eof_in();
    maybe_eof_net();
    return;
  }
 oq_queue_copy(&oq_net,&rbuf[0],r);
}

static int rtest_copy_net(void)
{
 return(!goteof_net && (oq_qlen(&oq_out) < MAXBUFFERED));
}

static void rd_copy_net(void)
{
 char rbuf[8192];
 int r;

 if (goteof_net) abort();
 r = read(net,&rbuf[0],sizeof(rbuf));
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: net read error: %s\n",__progname,strerror(errno));
    r = 0;
  }
 if (r == 0)
  { got_eof_net();
    maybe_eof_out();
    return;
  }
 oq_queue_copy(&oq_out,&rbuf[0],r);
}

static int timeout_in(int id)
{
 struct timeval t;

 t = (struct timeval){ .tv_sec = last_in.tv_sec + time_in,
		       .tv_usec = last_in.tv_usec };
 if (tvcmp(now(),t) <= 0)
  { got_eof_in();
    remove_block_id(id);
    return(BLOCK_LOOP);
  }
 return(tvsub_ms_clip(t,now()));
}

static int timeout_net(int id)
{
 struct timeval t;

 t = (struct timeval){ .tv_sec = last_net.tv_sec + time_net,
		       .tv_usec = last_net.tv_usec };
 if (tvcmp(now(),t) <= 0)
  { got_eof_net();
    remove_block_id(id);
    return(BLOCK_LOOP);
  }
 return(tvsub_ms_clip(t,now()));
}

int main(int, char **);
int main(int ac, char **av)
{
 signal(SIGPIPE,SIG_IGN);
 handleargs(ac,av);
 if (! arg1)
  { fprintf(stderr,"Usage: %s port [for accept mode]\n",__progname);
    fprintf(stderr,"   or: %s host port [for connect mode]\n",__progname);
    fprintf(stderr,"flags: -stdin, -net, -holdstdin, -holdnet, -ra, -rp, -ka, -nd\n");
    fprintf(stderr,"       -4, -6, -any, -retry, -reblock,\n");
    fprintf(stderr,"       -src [addr][:[port]], -timestdin sec, -timenet sec\n");
    fprintf(stderr,"       -cmd/-server/-server1 command\n");
    fprintf(stderr,"       -lockstep nbytes\n");
    exit(1);
  }
 if (arg2)
  { if (server != 0)
     { fprintf(stderr,"%s: -server or -server1 work only when accepting\n",__progname);
       exit(1);
     }
    setup_connect(arg1,arg2);
  }
 else
  { setup_accept(arg1);
  }
 if (keepalive)
  { int v;
    v = 1;
    if (setsockopt(net,SOL_SOCKET,SO_KEEPALIVE,&v,sizeof(v)) < 0)
     { fprintf(stderr,"%s: setsockopt SO_KEEPALIVE: %s\n",__progname,strerror(errno));
     }
  }
 if (nodelay)
  { int v;
    v = 1;
    if (setsockopt(net,IPPROTO_TCP,TCP_NODELAY,&v,sizeof(v)) < 0)
     { fprintf(stderr,"%s: setsockopt TCP_NODELAY: %s\n",__progname,strerror(errno));
     }
  }
 init_polling();
 if (cmd) run_cmd();
 if (reblock)
  { srandom(time(0));
    add_block_fn(&timeout_delay_net);
    add_block_fn(&timeout_delay_out);
    delay_out = 0;
    delay_net = 0;
  }
 set_nonblocking(1);
 set_nonblocking(net);
 oq_init(&oq_net);
 oq_init(&oq_out);
 if (lockstep)
  { add_poll_fd( 0,
		 &rtest_lockstep_in, &rwtest_never,
		 &rd_lockstep_in, 0 );
    add_poll_fd( 1,
		 &rwtest_never, &wtest_out,
		 0, &wr_out );
    add_poll_fd( net,
		 &rtest_lockstep_net, &wtest_net,
		 &rd_lockstep_net, &wr_net );
    block_in = 0;
    lock_state = LS_IDLE;
    lock_eof = 0;
    until_ping = lockstep;
  }
 else
  { add_poll_fd( 0,
		 &rtest_copy_in, &rwtest_never,
		 &rd_copy_in, 0 );
    add_poll_fd( 1,
		 &rwtest_never, &wtest_out,
		 0, &wr_out );
    add_poll_fd( net,
		 &rtest_copy_net, &wtest_net,
		 &rd_copy_net, &wr_net );
  }
 if (time_in || time_net)
  { gettimeofday(&last_in,0);
    last_net = last_in;
    if (time_in) add_block_fn(&timeout_in);
    if (time_net) add_block_fn(&timeout_net);
  }
 goteof_in = 0;
 goteof_net = 0;
 while (1)
  { if ( (nowait_net || (goteof_net && oq_empty(&oq_out))) &&
	 (nowait_in || (goteof_in && oq_empty(&oq_net))) ) break;
    now_valid = 0;
    pre_poll();
    if (do_poll() < 0)
     { if (errno == EINTR) continue;
       fprintf(stderr,"%s: poll: %s\n",__progname,strerror(errno));
       exit(1);
     }
    post_poll();
  }
 exit(0);
}