static void vconnfail(Conn *conn, const char *fmt, va_list al) PRINTF(2,0);
static void connfail(Conn *conn, const char *fmt, ...) PRINTF(2,3);
+static const oop_rd_style peer_rd_style;
+static oop_rd_call peer_rd_err, peer_rd_ok;
+
/*----- configuration options -----*/
static const char *sitename, *feedfile, *pathoutgoing;
static int connection_setup_timeout=200, port=119, try_stream=1;
static int inndcomm_flush_timeout=100;
-static int reconnect_delay_periods, flushfail_retry_periods, open_wait_periods;
+static int reconnect_delay_periods, flushfail_retry_periods;
static int backlog_retry_minperiods, backlog_spontaneous_rescan_periods;
static int spontaneous_flush_periods, need_activity_periods;
static const char *inndconffile;
struct Conn {
ISNODE(Conn);
int fd; /* may be 0, meaning closed (during construction/destruction) */
+ oop_read *rd; /* likewise */
int max_queue, stream, quitting;
int since_activity; /* periods */
ArticleList waiting; /* not yet told peer */
static time_t xtime(void) {
time_t now= time(0);
if (now==-1) sysdie("time(2) failed");
+ return now;
}
static void check_isreg(const struct stat *stab, const char *path,
check_isreg(stab, path, what);
}
-static void setnonblock(int fd, int nonblocking) {
- int r= fcntl(fd, F_GETFL); if (r<0) sysdie("setnonblocking fcntl F_GETFL");
- if (nonblocking) r |= O_NONBLOCK;
- else r &= ~O_NONBLOCK;
- r= fcntl(fd, F_SETFL, r); if (r<0) sysdie("setnonblocking fcntl F_SETFL");
-}
-
static int samefile(const struct stat *a, const struct stat *b) {
assert(S_ISREG(a->st_mode));
assert(S_ISREG(b->st_mode));
static void conn_dispose(Conn *conn) {
if (!conn) return;
+ if (conn->rd) {
+ oop_rd_cancel(conn->rd);
+ oop_rd_delete_kill(conn->rd);
+ conn->rd= 0;
+ }
if (conn->fd) {
loop->cancel_fd(loop, conn->fd, OOP_WRITE);
loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
goto x;
}
-#define CHK(field, val) \
- if (h->cmsg_##field != val) { \
- die("connect: child sent cmsg with cmsg_" #field "=%d, expected %d"); \
- goto x; \
+#define CHK(field, val) \
+ if (h->cmsg_##field != val) { \
+ die("connect: child sent cmsg with cmsg_" #field "=%d, expected %d", \
+ h->cmsg_##field, val); \
+ goto x; \
}
CHK(level, SOL_SOCKET);
CHK(type, SCM_RIGHTS);
if (CMSG_NXTHDR(&msg,h)) die("connect: child sent many cmsgs");
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);
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;
+
+ loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
+ conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
+ if (!conn->fd) sysdie("oop_rd_new_fd (fd=%d)",conn->fd);
+ int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_STRLEN,
+ &peer_rd_ok, conn,
+ &peer_rd_err, conn);
+ if (r) sysdie("oop_rd_read for peer (fd=%d)",conn->fd);
+
notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
LIST_ADDHEAD(conns, conn);
x:
conn_dispose(conn);
connect_attempt_discard();
+ return OOP_CONTINUE;
}
static int allow_connect_start(void) {
assert(l>=1);
if (buf[-1]!='\n')
fatal("connect: response to MODE STREAM is too long: %.100s...",
- remote_host, sanitise(buf));
+ sanitise(buf));
l--; if (l>0 && buf[l-1]=='\r') l--;
buf[l]= 0;
char *ep;
break;
Conn *walk, *use=0;
- int spare, inqueue;
+ int spare=0, inqueue=0;
/* Find a connection to offer this article. We prefer a busy
* connection to an idle one, provided it's not full. We take the
}
static void conn_maybe_write(Conn *conn) {
- void *rp= 0;
for (;;) {
conn_make_some_xmits(conn);
if (!conn->xmitu) {
rs -= vp->iov_len;
xmit_free(dp);
} else {
- vp->iov_base += rs;
+ vp->iov_base= (char*)vp->iov_base + rs;
vp->iov_len -= rs;
}
}
OOP_RD_SHORTREC_FORBID
};
-static void *peer_rd_err(oop_source *lp, oop_read *oread, oop_event ev,
+static void *peer_rd_err(oop_source *lp, oop_read *oread, oop_rd_event ev,
const char *errmsg, int errnoval,
const char *data, size_t recsz, void *conn_v) {
Conn *conn= conn_v;
if (r==-1) {
if (errno==EINTR) continue;
sysdie("failed to blank entry for %s (length %d at offset %lu) in %s",
- art->messageid, art->blanklen, art->offset, ipf->path);
+ art->messageid, art->blanklen,
+ (unsigned long)art->offset, ipf->path);
}
assert(r>=0 && r<=w);
art->blanklen -= w;
queue_check_input_done();
}
-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_rd_event ev,
const char *errmsg, int errnoval,
const char *data, size_t recsz, void *conn_v) {
Conn *conn= conn_v;
if (ipf->skippinglong) {
if (ev==OOP_RD_OK) ipf->skippinglong= 0; /* fine now */
- return;
+ return OOP_CONTINUE;
}
if (ev==OOP_RD_LONG) {
ipf->skippinglong= 1;
const char *xtra= "";
switch (sms) {
- case sm_FLUSHING: sm_FLUSHFAILED:
+ case sm_FLUSHING:
+ case sm_FLUSHFAILED:
if (!main_input_file) xtra= "-ABSENT";
break;
- case sm_SEPARATED: case sm_DROPPING:
+ case sm_SEPARATED:
+ case sm_DROPPING:
xtra= flushing_input_file->rd ? "-1" : "-2";
break;
default:;
int r, i;
struct stat stab;
const char *oldest_path=0;
- time_t oldest_mtime, now;
+ time_t oldest_mtime=0, now;
if (backlog_input_file) return;
" nonzero (error?) return value %d", globpat_backlog, r);
}
- globfree(&gl);
-
if (!oldest_path) {
debug("backlog scan: none");
exit(0);
}
until_backlog_nextscan= backlog_spontaneous_rescan_periods;
- return;
+ goto xfree;
}
now= xtime();
backlog_input_file= open_input_file(oldest_path);
if (!backlog_input_file) {
- warn("backlog file %s vanished as we opened it", backlog_input_file);
+ warn("backlog file %s vanished as we opened it", oldest_path);
+ globfree(&gl);
goto try_again;
}
inputfile_reading_start(backlog_input_file);
until_backlog_nextscan= -1;
- return;
+ goto xfree;
}
until_backlog_nextscan= age_deficiency / period_seconds;
debug("backlog scan: young age=%f deficiency=%ld nextscan=%d oldest=%s",
age, age_deficiency, until_backlog_nextscan, oldest_path);
+
+ xfree:
+ globfree(&gl);
return;
}
failed:
SMS(FLUSHFAILED, flushfail_retry_periods, "flush failed, will retry");
+ return OOP_CONTINUE;
}
static void inndcommfail(const char *what) {
static struct timeval what##_timeout = { interval_sec, interval_usec }; \
static void what##_schedule(void); \
static void *what##_timedout(oop_source *lp, struct timeval tv, void *u) { \
- { body } \
+ body; \
what##_schedule(); \
+ return OOP_CONTINUE; \
} \
static void what##_schedule(void) { \
loop->on_time(loop, what##_timeout, what##_timedout, 0); \
}
-EVERY(filepoll, 5,0, {
+EVERY(filepoll, 5,0, ({
if (main_input_file && main_input_file->readable_callback)
filemon_callback(main_input_file);
-});
+}));
+
+static char *debug_report_ipf(InputFile *ipf) {
+ if (!ipf) return xasprintf("-");
+
+ const char *slash= strrchr(ipf->path,'/');
+ const char *path= slash ? slash+1 : ipf->path;
-#define DEBUGF_IPF(wh) " " #wh "=%p/%s:ip=%ld,off=%ld,fd=%d%s"
-#define DEBUG_IPF(wh) \
- wh##_input_file, debug_ipf_path(wh##_input_file), \
- wh##_input_file->inprogress, (long)wh##_input_file->offset, \
- wh##_input_file->fd, wh##_input_file->rd ? "+" : ""
-static const char *debug_ipf_path(InputFile *ipf) {
- char *slash= strrchr(ipf->path,'/');
- return slash ? slash+1 : ipf->path;
+ return xasprintf("%p/%s:ip=%ld,off=%ld,fd=%d%s",
+ ipf, path,
+ ipf->inprogress, (long)ipf->offset,
+ ipf->fd, ipf->rd ? "+" : "");
}
-EVERY(period, -1,0, {
+EVERY(period, -1,0, ({
+ char *dipf_main= debug_report_ipf(main_input_file);
+ char *dipf_flushing= debug_report_ipf(flushing_input_file);
+ char *dipf_backlog= debug_report_ipf(backlog_input_file);
+
debug("PERIOD"
" sms=%s[%d] conns=%d queue=%d until_connect=%d"
- " input_files" DEBUGF_IPF(main) DEBUGF_IPF(old) DEBUGF_IPF(flushing)
- " children connecting=%ld inndcomm_child"
+ " input_files main:%s old:%s flushing:%s"
+ " children connecting=%ld inndcomm_child=%ld"
,
sms_names[sms], sm_period_counter,
- queue.count, conns.count, until_connect,
- DEBUG_IPF(main), DEBUG_IPF(flushing), DEBUG_IPF(flushing),
+ conns.count, queue.count, until_connect,
+ dipf_main, dipf_flushing, dipf_backlog,
(long)connecting_child, (long)inndcomm_child
);
+ free(dipf_main);
+ free(dipf_flushing);
+ free(dipf_backlog);
+
if (until_connect) until_connect--;
poll_backlog_file();
statemc_period_poll();
check_assign_articles();
check_idle_conns();
-});
+}));
/*========== option parsing ==========*/
static void vbadusage(const char *fmt, va_list al) NORET_PRINTF(1,0);
static void vbadusage(const char *fmt, va_list al) {
char *m= xvasprintf(fmt,al);
- fprintf(stderr, "bad usage: %s\n%s"
+ fprintf(stderr, "bad usage: %s\n"
"say --help for help, or read the manpage\n",
m);
if (become_daemon)
for (o=options; o->shrt || o->lng; o++) {
char shrt[2] = { o->shrt, 0 };
char *optspec= xasprintf("%s%s%s%s%s",
- o->shrt ? "-" : "", o->shrt,
+ o->shrt ? "-" : "", shrt,
o->shrt && o->lng ? "|" : "",
DELIMPERHAPS("--", o->lng));
fprintf(f, " %s%s%s\n", optspec, DELIMPERHAPS(" ", o->formarg));
~ian / innduct.git / commitdiff