EliasFleckenstein03
/
plan9front
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

aan: fix bugs, initial timeout, portable network encoding

front
cinap_lenrek 2012-06-27 12:36:15 +02:00
parent 0b003bb7c7
commit 623901ed1f
1 changed files with 138 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

249
sys/src/cmd/aan.c
View File

@ -4,12 +4,10 @@
#include <fcall.h> #include <fcall.h>
#include <thread.h> #include <thread.h>
#define NS(x) ((vlong)x) #define NS(x) ((vlong)x)
#define US(x) (NS(x) * 1000LL) #define US(x) (NS(x) * 1000LL)
#define MS(x) (US(x) * 1000LL) #define MS(x) (US(x) * 1000LL)
#define S(x) (MS(x) * 1000LL) #define S(x) (MS(x) * 1000LL)
#define LOGNAME "aan"
enum { enum {
Synctime = S(8), Synctime = S(8),
@ -17,7 +15,7 @@ enum {
K = 1024, K = 1024,
Bufsize = 8 * K, Bufsize = 8 * K,
Stacksize = 8 * K, Stacksize = 8 * K,
Timer = 0, // Alt channels. Timer = 0, // Alt channels.
Unsent = 1, Unsent = 1,
Maxto = 24 * 3600, // A full day to reconnect. Maxto = 24 * 3600, // A full day to reconnect.
}; };
@ -31,29 +29,28 @@ struct Endpoints {
}; };
typedef struct { typedef struct {
ulong nb; // Number of data bytes in this message uchar nb[4]; // Number of data bytes in this message
ulong msg; // Message number uchar msg[4]; // Message number
ulong acked; // Number of messages acked uchar acked[4]; // Number of messages acked
} Hdr; } Hdr;
typedef struct t_Buf { typedef struct t_Buf {
Hdr hdr; Hdr hdr;
uchar buf[Bufsize]; uchar buf[Bufsize];
} Buf; } Buf;
static char *progname;
static Channel *unsent; static Channel *unsent;
static Channel *unacked; static Channel *unacked;
static Channel *empty; static Channel *empty;
static int netfd; static int netfd;
static int inmsg; static int inmsg;
static char *devdir; static char *devdir;
static int debug; static int debug;
static int done; static int done;
static char *dialstring; static char *dialstring;
static int maxto = Maxto; static int maxto = Maxto;
static char *Logname = LOGNAME; static char *Logname = "aan";
static int client; static int client;
static Alt a[] = { static Alt a[] = {
/* c v op */ /* c v op */
@ -64,7 +61,7 @@ static Alt a[] = {
static void fromnet(void*); static void fromnet(void*);
static void fromclient(void*); static void fromclient(void*);
static void reconnect(void); static void reconnect(int);
static void synchronize(void); static void synchronize(void);
static int sendcommand(ulong, ulong); static int sendcommand(ulong, ulong);
static void showmsg(int, char *, Buf *); static void showmsg(int, char *, Buf *);
@ -72,13 +69,13 @@ static int writen(int, uchar *, int);
static int getport(char *); static int getport(char *);
static void dmessage(int, char *, ...); static void dmessage(int, char *, ...);
static void timerproc(void *); static void timerproc(void *);
static Endpoints *getendpoints(char *); static Endpoints* getendpoints(char *);
static void freeendpoints(Endpoints *); static void freeendpoints(Endpoints *);
static void static void
usage(void) usage(void)
{ {
fprint(2, "Usage: %s [-cd] [-m maxto] dialstring|netdir\n", progname); fprint(2, "Usage: %s [-cd] [-m maxto] dialstring|netdir\n", argv0);
exits("usage"); exits("usage");
} }
@ -86,22 +83,46 @@ static int
catch(void *, char *s) catch(void *, char *s)
{ {
if (!strcmp(s, "alarm")) { if (!strcmp(s, "alarm")) {
syslog(0, Logname, "Timed out while waiting for client on %s, exiting...", syslog(0, Logname, "Timed out while waiting for reconnect, exiting...");
devdir);
threadexitsall(nil); threadexitsall(nil);
} }
return 0; return 0;
} }
static void*
emalloc(int n)
{
ulong pc;
void *v;
pc = getcallerpc(&n);
v = malloc(n);
if(v == nil)
sysfatal("Cannot allocate memory; pc=%lux", pc);
setmalloctag(v, pc);
return v;
}
static char*
estrdup(char *s)
{
char *v;
int n;
n = strlen(s)+1;
v = emalloc(n);
memmove(v, s, n);
return v;
}
void void
threadmain(int argc, char **argv) threadmain(int argc, char **argv)
{ {
int i, failed;
Buf *b;
Channel *timer;
vlong synctime; vlong synctime;
int i, n, failed;
Channel *timer;
Buf *b;
progname = argv[0];
ARGBEGIN { ARGBEGIN {
case 'c': case 'c':
client++; client++;
@ -110,7 +131,7 @@ threadmain(int argc, char **argv)
debug++; debug++;
break; break;
case 'm': case 'm':
maxto = (int)strtol(EARGF(usage()), (char **)nil, 0); maxto = (int)strtol(EARGF(usage()), nil, 0);
break; break;
default: default:
usage(); usage();
@ -138,29 +159,33 @@ threadmain(int argc, char **argv)
atnotify(catch, 1); atnotify(catch, 1);
/*
* Set up initial connection. use short timeout
* of 60 seconds so we wont hang arround for too
* long if there is some general connection problem
* (like NAT).
*/
netfd = -1;
reconnect(60);
unsent = chancreate(sizeof(Buf *), Nbuf); unsent = chancreate(sizeof(Buf *), Nbuf);
unacked = chancreate(sizeof(Buf *), Nbuf); unacked = chancreate(sizeof(Buf *), Nbuf);
empty = chancreate(sizeof(Buf *), Nbuf); empty = chancreate(sizeof(Buf *), Nbuf);
timer = chancreate(sizeof(uchar *), 1); timer = chancreate(sizeof(uchar *), 1);
if(unsent == nil || unacked == nil || empty == nil || timer == nil)
sysfatal("Cannot allocate channels");
for (i = 0; i != Nbuf; i++) { for (i = 0; i < Nbuf; i++)
Buf *b = malloc(sizeof(Buf)); sendp(empty, emalloc(sizeof(Buf)));
sendp(empty, b);
}
netfd = -1;
if (proccreate(fromnet, nil, Stacksize) < 0) if (proccreate(fromnet, nil, Stacksize) < 0)
sysfatal("%s; Cannot start fromnet; %r", progname); sysfatal("Cannot start fromnet; %r");
reconnect(); // Set up the initial connection.
synchronize();
if (proccreate(fromclient, nil, Stacksize) < 0) if (proccreate(fromclient, nil, Stacksize) < 0)
sysfatal("%s; Cannot start fromclient; %r", progname); sysfatal("Cannot start fromclient; %r");
if (proccreate(timerproc, timer, Stacksize) < 0) if (proccreate(timerproc, timer, Stacksize) < 0)
sysfatal("%s; Cannot start timerproc; %r", progname); sysfatal("Cannot start timerproc; %r");
a[Timer].c = timer; a[Timer].c = timer;
a[Unsent].c = unsent; a[Unsent].c = unsent;
@ -169,9 +194,6 @@ threadmain(int argc, char **argv)
synctime = nsec() + Synctime; synctime = nsec() + Synctime;
failed = 0; failed = 0;
while (!done) { while (!done) {
vlong now;
int delta;
if (failed) { if (failed) {
// Wait for the netreader to die. // Wait for the netreader to die.
while (netfd >= 0) { while (netfd >= 0) {
@ -180,20 +202,17 @@ threadmain(int argc, char **argv)
} }
// the reader died; reestablish the world. // the reader died; reestablish the world.
reconnect(); reconnect(maxto);
synchronize(); synchronize();
failed = 0; failed = 0;
} }
now = nsec(); if (nsec() >= synctime) {
delta = (synctime - nsec()) / MS(1);
if (delta <= 0) {
Hdr hdr; Hdr hdr;
hdr.nb = 0; PBIT32(hdr.nb, 0);
hdr.acked = inmsg; PBIT32(hdr.acked, inmsg);
hdr.msg = -1; PBIT32(hdr.msg, -1);
if (writen(netfd, (uchar *)&hdr, sizeof(Hdr)) < 0) { if (writen(netfd, (uchar *)&hdr, sizeof(Hdr)) < 0) {
dmessage(2, "main; writen failed; %r\n"); dmessage(2, "main; writen failed; %r\n");
@ -201,7 +220,6 @@ threadmain(int argc, char **argv)
continue; continue;
} }
synctime = nsec() + Synctime; synctime = nsec() + Synctime;
assert(synctime > now);
} }
switch (alt(a)) { switch (alt(a)) {
@ -211,19 +229,20 @@ threadmain(int argc, char **argv)
case Unsent: case Unsent:
sendp(unacked, b); sendp(unacked, b);
b->hdr.acked = inmsg; PBIT32(b->hdr.acked, inmsg);
if (writen(netfd, (uchar *)&b->hdr, sizeof(Hdr)) < 0) { if (writen(netfd, (uchar *)&b->hdr, sizeof(Hdr)) < 0) {
dmessage(2, "main; writen failed; %r\n"); dmessage(2, "main; writen failed; %r\n");
failed = 1; failed = 1;
} }
if (writen(netfd, b->buf, b->hdr.nb) < 0) { n = GBIT32(b->hdr.nb);
if (writen(netfd, b->buf, n) < 0) {
dmessage(2, "main; writen failed; %r\n"); dmessage(2, "main; writen failed; %r\n");
failed = 1; failed = 1;
} }
if (b->hdr.nb == 0) if (n == 0)
done = 1; done = 1;
break; break;
} }
@ -237,48 +256,46 @@ static void
fromclient(void*) fromclient(void*)
{ {
static int outmsg; static int outmsg;
int n;
Buf *b;
for (;;) { do {
Buf *b; b = recvp(empty);
n = read(0, b->buf, Bufsize);
b = recvp(empty); if (n <= 0) {
if ((int)(b->hdr.nb = read(0, b->buf, Bufsize)) <= 0) { if (n < 0)
if ((int)b->hdr.nb < 0)
dmessage(2, "fromclient; Cannot read 9P message; %r\n"); dmessage(2, "fromclient; Cannot read 9P message; %r\n");
else else
dmessage(2, "fromclient; Client terminated\n"); dmessage(2, "fromclient; Client terminated\n");
b->hdr.nb = 0; n = 0;
} }
b->hdr.msg = outmsg++; PBIT32(b->hdr.nb, n);
PBIT32(b->hdr.msg, outmsg);
showmsg(1, "fromclient", b); showmsg(1, "fromclient", b);
sendp(unsent, b); sendp(unsent, b);
outmsg++;
if (b->hdr.nb == 0) } while(n > 0);
break;
}
} }
static void static void
fromnet(void*) fromnet(void*)
{ {
static int lastacked; static int lastacked;
int n, m, len, acked;
Buf *b; Buf *b;
b = (Buf *)malloc(sizeof(Buf)); b = emalloc(sizeof(Buf));
assert(b);
while (!done) { while (!done) {
int len, acked, i;
while (netfd < 0) { while (netfd < 0) {
dmessage(1, "fromnet; waiting for connection... (inmsg %d)\n", if(done)
inmsg); return;
dmessage(1, "fromnet; waiting for connection... (inmsg %d)\n", inmsg);
sleep(1000); sleep(1000);
} }
// Read the header. // Read the header.
if ((len = readn(netfd, &b->hdr, sizeof(Hdr))) <= 0) { len = readn(netfd, (uchar *)&b->hdr, sizeof(Hdr));
if (len <= 0) {
if (len < 0) if (len < 0)
dmessage(1, "fromnet; (hdr) network failure; %r\n"); dmessage(1, "fromnet; (hdr) network failure; %r\n");
else else
@ -287,18 +304,27 @@ fromnet(void*)
netfd = -1; netfd = -1;
continue; continue;
} }
dmessage(2, "fromnet: Got message, size %d, nb %d, msg %d\n", len, n = GBIT32(b->hdr.nb);
b->hdr.nb, b->hdr.msg); m = GBIT32(b->hdr.msg);
acked = GBIT32(b->hdr.acked);
dmessage(2, "fromnet: Got message, size %d, nb %d, msg %d, acked %d, lastacked %d\n",
len, n, m, acked, lastacked);
if (b->hdr.nb == 0) { if (n == 0) {
if ((long)b->hdr.msg >= 0) { if (m >= 0) {
dmessage(1, "fromnet; network closed\n"); dmessage(1, "fromnet; network closed\n");
break; break;
} }
continue; continue;
} }
if ((len = readn(netfd, b->buf, b->hdr.nb)) <= 0 || len != b->hdr.nb) { if (n > Bufsize) {
dmessage(1, "fromnet; message too big %d > %d\n", n, Bufsize);
break;
}
len = readn(netfd, b->buf, n);
if (len <= 0 || len != n) {
if (len == 0) if (len == 0)
dmessage(1, "fromnet; network closed\n"); dmessage(1, "fromnet; network closed\n");
else else
@ -308,28 +334,25 @@ fromnet(void*)
continue; continue;
} }
if (b->hdr.msg < inmsg) { if (m < inmsg) {
dmessage(1, "fromnet; skipping message %d, currently at %d\n", dmessage(1, "fromnet; skipping message %d, currently at %d\n", m, inmsg);
b->hdr.msg, inmsg);
continue; continue;
} }
// Process the acked list. // Process the acked list.
acked = b->hdr.acked - lastacked; while(lastacked != acked) {
for (i = 0; i != acked; i++) {
Buf *rb; Buf *rb;
rb = recvp(unacked); rb = recvp(unacked);
if (rb->hdr.msg != lastacked + i) { m = GBIT32(rb->hdr.msg);
dmessage(1, "rb %p, msg %d, lastacked %d, i %d\n", if (m != lastacked) {
rb, rb? rb->hdr.msg: -2, lastacked, i); dmessage(1, "fromnet; rb %p, msg %d, lastacked %d\n", rb, m, lastacked);
assert(0); sysfatal("fromnet; bug");
} }
rb->hdr.msg = -1; PBIT32(rb->hdr.msg, -1);
sendp(empty, rb); sendp(empty, rb);
lastacked++;
} }
lastacked = b->hdr.acked;
inmsg++; inmsg++;
showmsg(1, "fromnet", b); showmsg(1, "fromnet", b);
@ -341,13 +364,14 @@ fromnet(void*)
} }
static void static void
reconnect(void) reconnect(int secs)
{ {
char ldir[40]; char ldir[40];
int lcfd, fd; int lcfd, fd;
if (dialstring) { if (dialstring) {
syslog(0, Logname, "dialing %s", dialstring); syslog(0, Logname, "dialing %s", dialstring);
alarm(secs*1000);
while ((fd = dial(dialstring, nil, nil, nil)) < 0) { while ((fd = dial(dialstring, nil, nil, nil)) < 0) {
char err[32]; char err[32];
@ -360,16 +384,16 @@ reconnect(void)
dmessage(1, "reconnect: dialed %s; %s\n", dialstring, err); dmessage(1, "reconnect: dialed %s; %s\n", dialstring, err);
sleep(1000); sleep(1000);
} }
alarm(0);
syslog(0, Logname, "reconnected to %s", dialstring); syslog(0, Logname, "reconnected to %s", dialstring);
} }
else { else {
Endpoints *ep; Endpoints *ep;
syslog(0, Logname, "waiting for connection on %s", devdir); syslog(0, Logname, "waiting for connection on %s", devdir);
alarm(maxto * 1000); alarm(secs*1000);
if ((lcfd = listen(devdir, ldir)) < 0) if ((lcfd = listen(devdir, ldir)) < 0)
sysfatal("reconnect; cannot listen; %r"); sysfatal("reconnect; cannot listen; %r");
if ((fd = accept(lcfd, ldir)) < 0) if ((fd = accept(lcfd, ldir)) < 0)
sysfatal("reconnect; cannot accept; %r"); sysfatal("reconnect; cannot accept; %r");
alarm(0); alarm(0);
@ -381,7 +405,8 @@ reconnect(void)
freeendpoints(ep); freeendpoints(ep);
} }
netfd = fd; // Wakes up the netreader. // Wakes up the netreader.
netfd = fd;
} }
static void static void
@ -389,6 +414,7 @@ synchronize(void)
{ {
Channel *tmp; Channel *tmp;
Buf *b; Buf *b;
int n;
// Ignore network errors here. If we fail during // Ignore network errors here. If we fail during
// synchronization, the next alarm will pick up // synchronization, the next alarm will pick up
@ -396,7 +422,9 @@ synchronize(void)
tmp = chancreate(sizeof(Buf *), Nbuf); tmp = chancreate(sizeof(Buf *), Nbuf);
while ((b = nbrecvp(unacked)) != nil) { while ((b = nbrecvp(unacked)) != nil) {
writen(netfd, (uchar *)b, sizeof(Hdr) + b->hdr.nb); n = GBIT32(b->hdr.nb);
writen(netfd, (uchar *)&b->hdr, sizeof(Hdr));
writen(netfd, b->buf, n);
sendp(tmp, b); sendp(tmp, b);
} }
chanfree(unacked); chanfree(unacked);
@ -406,14 +434,14 @@ synchronize(void)
static void static void
showmsg(int level, char *s, Buf *b) showmsg(int level, char *s, Buf *b)
{ {
int n;
if (b == nil) { if (b == nil) {
dmessage(level, "%s; b == nil\n", s); dmessage(level, "%s; b == nil\n", s);
return; return;
} }
n = GBIT32(b->hdr.nb);
dmessage(level, dmessage(level, "%s; (len %d) %X %X %X %X %X %X %X %X %X (%p)\n", s, n,
"%s; (len %d) %X %X %X %X %X %X %X %X %X (%p)\n", s,
b->hdr.nb,
b->buf[0], b->buf[1], b->buf[2], b->buf[0], b->buf[1], b->buf[2],
b->buf[3], b->buf[4], b->buf[5], b->buf[3], b->buf[4], b->buf[5],
b->buf[6], b->buf[7], b->buf[8], b); b->buf[6], b->buf[7], b->buf[8], b);
@ -483,16 +511,16 @@ getendpoint(char *dir, char *file, char **sysp, char **servp)
serv = strchr(buf, '!'); serv = strchr(buf, '!');
if(serv){ if(serv){
*serv++ = 0; *serv++ = 0;
serv = strdup(serv); serv = estrdup(serv);
} }
sys = strdup(buf); sys = estrdup(buf);
} }
close(fd); close(fd);
} }
if(serv == 0) if(serv == 0)
serv = strdup("unknown"); serv = estrdup("unknown");
if(sys == 0) if(sys == 0)
sys = strdup("unknown"); sys = estrdup("unknown");
*servp = serv; *servp = serv;
*sysp = sys; *sysp = sys;
} }
@ -502,7 +530,7 @@ getendpoints(char *dir)
{ {
Endpoints *ep; Endpoints *ep;
ep = malloc(sizeof(*ep)); ep = emalloc(sizeof(*ep));
getendpoint(dir, "local", &ep->lsys, &ep->lserv); getendpoint(dir, "local", &ep->lsys, &ep->lserv);
getendpoint(dir, "remote", &ep->rsys, &ep->rserv); getendpoint(dir, "remote", &ep->rsys, &ep->rserv);
return ep; return ep;
@ -517,4 +545,3 @@ freeendpoints(Endpoints *ep)
free(ep->rserv); free(ep->rserv);
free(ep); free(ep);
} }