chiark
/
gitweb
/
~ian
/
inn-innduct.git
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (parent:
5995964
)
check fd watches; other bugfixes
author
Ian Jackson
<ian@liberator.(none)>
Tue, 27 Apr 2010 13:04:53 +0000
(14:04 +0100)
committer
Ian Jackson
<ian@liberator.(none)>
Tue, 27 Apr 2010 13:04:53 +0000
(14:04 +0100)
backends/innduct.c
patch
|
blob
|
history
diff --git
a/backends/innduct.c
b/backends/innduct.c
index 1328d372f8d8390cb822f734a2476f044b1c6404..43ae915a1f3646fe58e96c9726b6de405d962d81 100644
(file)
--- a/
backends/innduct.c
+++ b/
backends/innduct.c
@@
-1,8
+1,8
@@
/*
* TODO
/*
* TODO
- * - check all structs initialised
- * - check all fd watches properly undone
* - check all init functions called
* - check all init functions called
+ * - check ipf->inprogress managed properly
+ * - xperhaps_close
* - actually implement badusage
* - options for all options
* - manpage
* - actually implement badusage
* - options for all options
* - manpage
@@
-238,6
+238,7
@@
perl -ne 'print if m/-8\<-/..m/-\>8-/; print "\f" if m/-\^L-/' backends/innduct.
#define LIST_REMHEAD(l) LIST_REMSOMEHOW((l),list_remhead)
#define LIST_REMTAIL(l) LIST_REMSOMEHOW((l),list_remtail)
#define LIST_REMHEAD(l) LIST_REMSOMEHOW((l),list_remhead)
#define LIST_REMTAIL(l) LIST_REMSOMEHOW((l),list_remtail)
+#define LIST_INIT(l) ((l).hd, list_new((struct list*)&(l)))
#define LIST_HEAD(l) ((typeof((l).hd))(list_head((struct list*)&(l))))
#define LIST_NEXT(n) ((typeof(n))list_succ(NODE((n))))
#define LIST_BACK(n) ((typeof(n))list_pred(NODE((n))))
#define LIST_HEAD(l) ((typeof((l).hd))(list_head((struct list*)&(l))))
#define LIST_NEXT(n) ((typeof(n))list_succ(NODE((n))))
#define LIST_BACK(n) ((typeof(n))list_pred(NODE((n))))
@@
-391,7
+392,7
@@
struct InputFile {
oop_readable_call *readable_callback;
void *readable_callback_user;
oop_readable_call *readable_callback;
void *readable_callback_user;
- int fd;
+ int fd;
/* may be 0, meaning closed */
Filemon_Perfile *filemon;
oop_read *rd;
Filemon_Perfile *filemon;
oop_read *rd;
@@
-436,7
+437,8
@@
static const char *sms_names[]= {
struct Conn {
ISNODE(Conn);
struct Conn {
ISNODE(Conn);
- int fd, max_queue, stream, quitting;
+ int fd; /* may be 0, meaning closed (during construction/destruction) */
+ int max_queue, stream, quitting;
int since_activity; /* periods */
ArticleList waiting; /* not yet told peer */
ArticleList priority; /* peer says send it now */
int since_activity; /* periods */
ArticleList waiting; /* not yet told peer */
ArticleList priority; /* peer says send it now */
@@
-679,6
+681,10
@@
static int isewouldblock(int errnoval) {
static void conn_dispose(Conn *conn) {
if (!conn) return;
static void conn_dispose(Conn *conn) {
if (!conn) return;
+ if (conn->fd) {
+ loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+ loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
+ }
perhaps_close(&conn->fd);
free(conn);
until_connect= reconnect_delay_periods;
perhaps_close(&conn->fd);
free(conn);
until_connect= reconnect_delay_periods;
@@
-728,16
+734,18
@@
static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
if (got != -1) {
assert(got==connecting_child);
connecting_child= 0;
if (got != -1) {
assert(got==connecting_child);
connecting_child= 0;
- if (WIFEXITED(status) &&
- (WEXITSTATUS(status) != 0 &&
- WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
- WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)) {
- /* child already reported the problem */
+ if (WIFEXITED(status)) {
+ if (WEXITSTATUS(status) != 0 &&
+ WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
+ WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
+ /* child already reported the problem */;
+ else
+ warn("connect: connection child exited code %d but no cmsg",
+ WEXITSTATUS(status));
} else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
warn("connect: connection attempt timed out");
} else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
warn("connect: connection attempt timed out");
- } else
if (!WIFEXITED(status))
{
+ } else {
report_child_status("connect", status);
report_child_status("connect", status);
- /* that's probably the root cause then */
}
} else {
/* child is still running apparently, report the socket problem */
}
} else {
/* child is still running apparently, report the socket problem */
@@
-745,8
+753,10
@@
static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
syswarn("connect: read from child socket failed");
else if (e == OOP_EXCEPTION)
warn("connect: unexpected exception on child socket");
syswarn("connect: read from child socket failed");
else if (e == OOP_EXCEPTION)
warn("connect: unexpected exception on child socket");
- else
+ else
if (!rs)
warn("connect: unexpected EOF on child socket");
warn("connect: unexpected EOF on child socket");
+ else
+ fatal("connect: unexpected lack of cmsg from child");
}
goto x;
}
}
goto x;
}
@@
-761,9
+771,10
@@
static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
CHK(len, CMSG_LEN(sizeof(conn->fd)));
#undef CHK
CHK(len, CMSG_LEN(sizeof(conn->fd)));
#undef CHK
- if (CMSG_NXTHDR(&msg,h))
{ die("connect: child sent many cmsgs"); goto x; }
+ if (CMSG_NXTHDR(&msg,h))
die("connect: child sent many cmsgs");
memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
+ loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
int status;
pid_t got= waitpid(connecting_child, &status, 0);
int status;
pid_t got= waitpid(connecting_child, &status, 0);
@@
-781,10
+792,14
@@
static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
}
/* Phew! */
}
/* Phew! */
+ LIST_INIT(conn->waiting);
+ LIST_INIT(conn->priority);
+ LIST_INIT(conn->sent);
setnonblock(conn->fd, 1);
conn->max_queue= conn->stream ? max_queue_per_conn : 1;
setnonblock(conn->fd, 1);
conn->max_queue= conn->stream ? max_queue_per_conn : 1;
- LIST_ADDHEAD(conns, conn);
notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
+ LIST_ADDHEAD(conns, conn);
+
connect_attempt_discard();
check_master_queue();
return 0;
connect_attempt_discard();
check_master_queue();
return 0;
@@
-794,6
+809,18
@@
static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
connect_attempt_discard();
}
connect_attempt_discard();
}
+static void conn_exception(oop_source *lp, int fd,
+ oop_event ev, void *conn_v) {
+ Conn *conn= conn_v;
+ char ch;
+ assert(fd == conn->fd);
+ assert(ev == OOP_EXCEPTION);
+ r= read(conn->fd, &ch, 1);
+ if (r<0) connfail(conn,"read failed: %s",strerror(errno));
+ connfail(conn,"exceptional condition on socket (peer sent urgent"
+ " data? read(,,1)=%d)",r);
+}
+
static int allow_connect_start(void) {
return conns.count < max_connections
&& !connecting_child
static int allow_connect_start(void) {
return conns.count < max_connections
&& !connecting_child
@@
-1261,6
+1288,8
@@
static void article_done(Conn *conn, Article *art, int whichcount) {
else if (whichcount == RC_unwanted) update_nocheck(0);
InputFile *ipf= art->ipf;
else if (whichcount == RC_unwanted) update_nocheck(0);
InputFile *ipf= art->ipf;
+ assert(ipf->fd);
+
while (art->blanklen) {
static const char spaces[]=
" "
while (art->blanklen) {
static const char spaces[]=
" "
@@
-1281,11
+1310,10
@@
static void article_done(Conn *conn, Article *art, int whichcount) {
ipf->inprogress--;
assert(ipf->inprogress >= 0);
ipf->inprogress--;
assert(ipf->inprogress >= 0);
+ free(art);
if (!ipf->inprogress && ipf != main_input_file)
queue_check_input_done();
if (!ipf->inprogress && ipf != main_input_file)
queue_check_input_done();
-
- free(art);
}
static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
}
static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
@@
-1389,9
+1417,10
@@
static void feedfile_eof(InputFile *ipf) {
assert(ipf != main_input_file); /* promised by tailing_try_read */
inputfile_tailing_stop(ipf);
assert(ipf != main_input_file); /* promised by tailing_try_read */
inputfile_tailing_stop(ipf);
- assert(ipf->fd >= 0);
+ assert(ipf->fd > 0);
+fixme do not close fd do something else because art_done needs it to blank entries;
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
- ipf->fd=
-1
;
+ ipf->fd=
0
;
if (ipf == flushing_input_file) {
assert(sms==sm_SEPARATED || sms==sm_DROPPING);
if (ipf == flushing_input_file) {
assert(sms==sm_SEPARATED || sms==sm_DROPPING);
@@
-1410,6
+1439,7
@@
static InputFile *open_input_file(const char *path) {
if (errno==ENOENT) return 0;
sysfatal("unable to open input file %s", path);
}
if (errno==ENOENT) return 0;
sysfatal("unable to open input file %s", path);
}
+ assert(fd>0);
InputFile *ipf= xmalloc(sizeof(*ipf) + strlen(path) + 1);
memset(ipf,0,sizeof(*ipf));
InputFile *ipf= xmalloc(sizeof(*ipf) + strlen(path) + 1);
memset(ipf,0,sizeof(*ipf));
@@
-1420,14
+1450,16
@@
static InputFile *open_input_file(const char *path) {
return ipf;
}
return ipf;
}
-static void close_input_file(InputFile *ipf) {
+static void close_input_file(InputFile *ipf) {
/* does not free */
assert(!ipf->readable_callback); /* must have had ->on_cancel */
assert(!ipf->filemon); /* must have had inputfile_tailing_stop */
assert(!ipf->rd); /* must have had inputfile_tailing_stop */
assert(!ipf->inprogress); /* no dangling pointers pointing here */
assert(!ipf->readable_callback); /* must have had ->on_cancel */
assert(!ipf->filemon); /* must have had inputfile_tailing_stop */
assert(!ipf->rd); /* must have had inputfile_tailing_stop */
assert(!ipf->inprogress); /* no dangling pointers pointing here */
- if (ipf->fd
>= 0)
+ if (ipf->fd
) {
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
+ ipf->fd= 0;
+ }
}
}
@@
-1507,12
+1539,12
@@
static void *feedfile_got_article(oop_source *lp, oop_read *rd,
ipf->readcount_ok++;
art= xmalloc(sizeof(*art) - 1 + midlen + 1);
ipf->readcount_ok++;
art= xmalloc(sizeof(*art) - 1 + midlen + 1);
- art->offset= ipf->offset;
- art->blanklen= recsz;
- art->midlen= midlen;
art->state= art_Unchecked;
art->state= art_Unchecked;
+ art->midlen= midlen;
art->ipf= ipf; ipf->inprogress++;
art->token= TextToToken(tokentextbuf);
art->ipf= ipf; ipf->inprogress++;
art->token= TextToToken(tokentextbuf);
+ art->offset= ipf->offset;
+ art->blanklen= recsz;
strcpy(art->messageid, space+1);
LIST_ADDTAIL(queue, art);
strcpy(art->messageid, space+1);
LIST_ADDTAIL(queue, art);
@@
-1574,10
+1606,10
@@
static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer,
errno=EAGAIN;
return -1;
} else if (ipf==flushing_input_file) {
errno=EAGAIN;
return -1;
} else if (ipf==flushing_input_file) {
- assert(ipf->fd
>=0
);
+ assert(ipf->fd);
assert(sms==sm_SEPARATED || sms==sm_DROPPING);
} else if (ipf==backlog_input_file) {
assert(sms==sm_SEPARATED || sms==sm_DROPPING);
} else if (ipf==backlog_input_file) {
- assert(ipf->fd
>=0
);
+ assert(ipf->fd);
} else {
abort();
}
} else {
abort();
}
@@
-1708,7
+1740,7
@@
static const oop_rd_style feedfile_rdstyle= {
};
static void inputfile_tailing_start(InputFile *ipf) {
};
static void inputfile_tailing_start(InputFile *ipf) {
- assert(!ipf->
f
d);
+ assert(!ipf->
r
d);
ipf->readable.on_readable= tailing_on_readable;
ipf->readable.on_cancel= tailing_on_cancel;
ipf->readable.try_read= tailing_try_read;
ipf->readable.on_readable= tailing_on_readable;
ipf->readable.on_cancel= tailing_on_cancel;
ipf->readable.try_read= tailing_try_read;
@@
-1727,7
+1759,7
@@
static void inputfile_tailing_start(InputFile *ipf) {
}
static void inputfile_tailing_stop(InputFile *ipf) {
}
static void inputfile_tailing_stop(InputFile *ipf) {
- assert(ipf->
f
d);
+ assert(ipf->
r
d);
oop_rd_cancel(ipf->rd);
oop_rd_delete(ipf->rd);
ipf->rd= 0;
oop_rd_cancel(ipf->rd);
oop_rd_delete(ipf->rd);
ipf->rd= 0;
@@
-1794,7
+1826,7
@@
static void inputfile_tailing_stop(InputFile *ipf) {
| V | | V
| ============= V V ============
| SEPARATED-1 | | DROPPING-1
| V | | V
| ============= V V ============
| SEPARATED-1 | | DROPPING-1
- | flsh->fd>
=0 | | flsh->fd>=
0
+ | flsh->fd>
0 | | flsh->fd>
0
| [Separated] | | [Dropping]
| main F idle | | main none
| old D tail | | old D tail
| [Separated] | | [Dropping]
| main F idle | | main none
| old D tail | | old D tail
@@
-1804,7
+1836,7
@@
static void inputfile_tailing_stop(InputFile *ipf) {
| V | | V
| =============== | | ===============
| SEPARATED-2 | | DROPPING-2
| V | | V
| =============== | | ===============
| SEPARATED-2 | | DROPPING-2
- | flsh->fd==
-1 | V flsh->fd==-1
+ | flsh->fd==
0 | V flsh->fd==0
| [Finishing] | | [Dropping]
| main F tail | `. main none
| old D closed | `. old D closed
| [Finishing] | | [Dropping]
| main F tail | `. main none
| old D closed | `. old D closed
@@
-1976,7
+2008,7
@@
static void statemc_period_poll(void) {
static int inputfile_is_done(InputFile *ipf) {
if (!ipf) return 0;
if (ipf->inprogress) return 0; /* new article in the meantime */
static int inputfile_is_done(InputFile *ipf) {
if (!ipf) return 0;
if (ipf->inprogress) return 0; /* new article in the meantime */
- if (ipf->fd
>= 0);
return 0; /* not had EOF */
+ if (ipf->fd
)
return 0; /* not had EOF */
return 1;
}
return 1;
}
@@
-2078,7
+2110,7
@@
static void statemc_setstate(StateMachineState newsms, int periods,
if (!main_input_file) xtra= "-ABSENT";
break;
case sm_SEPARATED: case sm_DROPPING:
if (!main_input_file) xtra= "-ABSENT";
break;
case sm_SEPARATED: case sm_DROPPING:
- xtra= flushing_input_file->fd
>= 0
? "-1" : "-2";
+ xtra= flushing_input_file->fd ? "-1" : "-2";
break;
default:;
}
break;
default:;
}
@@
-2388,9
+2420,8
@@
void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */
static void postfork_inputfile(InputFile *ipf) {
if (!ipf) return;
static void postfork_inputfile(InputFile *ipf) {
if (!ipf) return;
- assert(ipf->fd >= 0);
- close(ipf->fd);
- ipf->fd= -1;
+ assert(ipf->fd);
+ perhaps_close(&ipf->fd);
}
static void postfork_stdio(FILE *f) {
}
static void postfork_stdio(FILE *f) {
@@
-2690,6
+2721,9
@@
int main(int argc, char **argv) {
notice("starting");
notice("starting");
+ LIST_INIT(conns);
+ LIST_INIT(queue);
+
if (!filemon_method_init()) {
warn("no file monitoring available, polling");
filepoll_schedule();
if (!filemon_method_init()) {
warn("no file monitoring available, polling");
filepoll_schedule();