static const char *sitename, *remote_host;
static const char *feedfile, *path_run, *path_cli, *path_cli_dir;
static int quiet_multiple=0;
-static int become_daemon=1, try_filemon=1;
+static int interactive=0, try_filemon=1;
static int try_stream=1;
static int port=119;
static const char *inndconffile;
static int max_connections=10;
static int max_queue_per_conn=200;
static int target_max_feedfile_size=100000;
-static int period_seconds=60;
+static int period_seconds=30;
static int filepoll_seconds=5;
static int max_queue_per_ipf=-1;
/* all these are initialised to seconds, and converted to periods in main */
static int reconnect_delay_periods=1000;
static int flushfail_retry_periods=1000;
-static int backlog_retry_minperiods=50;
+static int backlog_retry_minperiods=100;
static int backlog_spontrescan_periods=300;
static int spontaneous_flush_periods=100000;
static int max_separated_periods=2000;
static int need_activity_periods=1000;
+static int lowvol_thresh=3;
+static int lowvol_periods=1000;
static double max_bad_data_ratio= 1; /* conv'd from percentage by main */
static int max_bad_data_initial= 30;
ISNODE(Conn);
int fd; /* may be 0, meaning closed (during construction/destruction) */
oop_read *rd; /* likewise */
- int max_queue, stream, quitting;
+ int oopwriting; /* since on_fd is not idempotent */
+ int max_queue, stream;
+ const char *quitting;
int since_activity; /* periods */
ArticleList waiting; /* not yet told peer */
ArticleList priority; /* peer says send it now */
static char *path_lock, *path_flushing, *path_defer, *path_dump;
static char *globpat_backlog;
static pid_t self_pid;
+static int *lowvol_perperiod;
+static int lowvol_circptr;
+static int lowvol_total; /* does not include current period */
/* statemc_init initialises */
static StateMachineState sms;
static void logcore(int sysloglevel, const char *fmt, ...) PRINTF(2,3);
static void logcore(int sysloglevel, const char *fmt, ...) {
VA;
- if (become_daemon) {
+ if (interactive < 2) {
vsyslog(sysloglevel,fmt,al);
} else {
if (self_pid) fprintf(stderr,"[%lu] ",(unsigned long)self_pid);
/*========== management of connections ==========*/
+static void reconnect_blocking_event(void) {
+ until_connect= reconnect_delay_periods;
+}
+
static void conn_closefd(Conn *conn, const char *msgprefix) {
int r= close_perhaps(&conn->fd);
if (r) info("C%d %serror closing socket: %s",
conn->fd, msgprefix, strerror(errno));
}
+static int conn_busy(Conn *conn) {
+ return
+ conn->waiting.count ||
+ conn->priority.count ||
+ conn->sent.count ||
+ conn->xmitu;
+}
+
static void conn_dispose(Conn *conn) {
if (!conn) return;
if (conn->rd) {
}
conn_closefd(conn,"");
free(conn);
- until_connect= reconnect_delay_periods;
}
static void *conn_exception(oop_source *lp, int fd,
for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
xmit_free(d);
+ LIST_REMOVE(conns,conn);
+
char *m= xvasprintf(fmt,al);
- warn("C%d connection failed requeueing " RCI_TRIPLE_FMT_BASE ": %s",
- conn->fd, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
+ warn("C%d (now %d) connection failed requeueing " RCI_TRIPLE_FMT_BASE ": %s",
+ conn->fd, conns.count, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
free(m);
- LIST_REMOVE(conns,conn);
+ reconnect_blocking_event();
conn_dispose(conn);
check_assign_articles();
}
va_end(al);
}
+static void conn_idle_close(Conn *conn, const char *why) {
+ static const char quitcmd[]= "QUIT\r\n";
+ int todo= sizeof(quitcmd)-1;
+ const char *p= quitcmd;
+ for (;;) {
+ int r= write(conn->fd, p, todo);
+ if (r<0) {
+ if (isewouldblock(errno))
+ connfail(conn, "blocked writing QUIT to idle connection");
+ else
+ connfail(conn, "failed to write QUIT to idle connection: %s",
+ strerror(errno));
+ break;
+ }
+ assert(r<=todo);
+ todo -= r;
+ if (!todo) {
+ conn->quitting= why;
+ conn->since_activity= 0;
+ debug("C%d is idle (%s), quitting", conn->fd, why);
+ break;
+ }
+ }
+}
+
+/*
+ * For our last connection, we also shut it down if we have had
+ * less than K in the last L
+ */
static void check_idle_conns(void) {
Conn *conn;
+
+ int volthisperiod= lowvol_perperiod[lowvol_circptr];
+ lowvol_circptr++;
+ lowvol_circptr %= lowvol_periods;
+ lowvol_total += volthisperiod;
+ lowvol_total -= lowvol_perperiod[lowvol_circptr];
+ lowvol_perperiod[lowvol_circptr]= 0;
+
FOR_CONN(conn)
conn->since_activity++;
+
search_again:
FOR_CONN(conn) {
if (conn->since_activity <= need_activity_periods) continue;
/* We need to shut this down */
if (conn->quitting)
- connfail(conn,"timed out waiting for response to QUIT");
+ connfail(conn,"timed out waiting for response to QUIT (%s)",
+ conn->quitting);
else if (conn->sent.count)
connfail(conn,"timed out waiting for responses");
else if (conn->waiting.count || conn->priority.count)
else if (conn->xmitu)
connfail(conn,"peer has been sending responses"
" before receiving our commands!");
- else {
- static const char quitcmd[]= "QUIT\r\n";
- int todo= sizeof(quitcmd)-1;
- const char *p= quitcmd;
- for (;;) {
- int r= write(conn->fd, p, todo);
- if (r<0) {
- if (isewouldblock(errno))
- connfail(conn, "blocked writing QUIT to idle connection");
- else
- connfail(conn, "failed to write QUIT to idle connection: %s",
- strerror(errno));
- break;
- }
- assert(r<=todo);
- todo -= r;
- if (!todo) {
- conn->quitting= 1;
- conn->since_activity= 0;
- debug("C%d is idle, quitting", conn->fd);
- break;
- }
- }
- }
+ else
+ conn_idle_close(conn, "no activity");
+
goto search_again;
}
+
+ conn= LIST_HEAD(conns);
+ if (!volthisperiod &&
+ conns.count==1 &&
+ lowvol_total < lowvol_thresh &&
+ !conn_busy(conn))
+ conn_idle_close(conn, "low volume");
}
/*---------- making new connections ----------*/
&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);
+ notice("C%d (now %d) connected %s",
+ conn->fd, conns.count, conn->stream ? "streaming" : "plain");
connect_attempt_discard();
check_assign_articles();
x:
conn_dispose(conn);
connect_attempt_discard();
+ reconnect_blocking_event();
return OOP_CONTINUE;
}
assert(!connecting_fdpass_sock);
info("starting connection attempt");
+ int ok_until_connect= until_connect;
+ reconnect_blocking_event();
int socks[2];
int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
connecting_fdpass_sock= socks[0];
xsetnonblock(connecting_fdpass_sock, 1);
on_fd_read_except(connecting_fdpass_sock, connchild_event);
+
+ if (!conns.count)
+ until_connect= ok_until_connect;
}
/*---------- assigning articles to conns, and transmitting ----------*/
Article *art= dequeue(0);
if (!art) break;
LIST_ADDTAIL(use->waiting, art);
+ lowvol_perperiod[lowvol_circptr]++;
spare--;
}
conn_maybe_write(use);
} else if (allow_connect_start()) {
- until_connect= reconnect_delay_periods;
connect_start();
break;
} else {
conn_make_some_xmits(conn);
if (!conn->xmitu) {
loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+ conn->oopwriting= 0;
return;
}
void *rp= conn_write_some_xmits(conn);
if (rp==OOP_CONTINUE) {
- loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
+ if (!conn->oopwriting) {
+ loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
+ conn->oopwriting= 1;
+ }
return;
} else if (rp==OOP_HALT) {
return;
}
static void autodefer_input_file(InputFile *ipf) {
+ static const char *const abandon= "stuck";
ipf->autodefer= 0;
autodefer_input_file_articles(ipf);
if (has_article_in(&walk->waiting, ipf) ||
has_article_in(&walk->priority, ipf) ||
has_article_in(&walk->sent, ipf))
- walk->quitting= -1;
+ walk->quitting= abandon;
}
while (ipf->inprogress) {
FOR_CONN(walk)
- if (walk->quitting < 0) goto found;
+ if (walk->quitting == abandon) goto found;
abort(); /* where are they ?? */
found:
assert(rs > 0);
int done;
- for (done=0; rs && done<conn->xmitu; done++) {
+ for (done=0; rs; ) {
+ assert(done<conn->xmitu);
struct iovec *vp= &conn->xmit[done];
XmitDetails *dp= &conn->xmitd[done];
- if (rs > vp->iov_len) {
+ if (rs >= vp->iov_len) {
rs -= vp->iov_len;
- xmit_free(dp);
+ xmit_free(dp); /* vp->iov_len -= vp->iov_len, etc. */
+ done++;
} else {
vp->iov_base= (char*)vp->iov_base + rs;
vp->iov_len -= rs;
+ break; /* rs -= rs */
}
}
int newu= conn->xmitu - done;
return OOP_CONTINUE;
}
- int conn_busy=
- conn->waiting.count ||
- conn->priority.count ||
- conn->sent.count ||
- conn->xmitu;
+ int busy= conn_busy(conn);
if (conn->quitting) {
- if (code!=205 && code!=503) {
- connfail(conn, "peer gave unexpected response to QUIT: %s", sani);
+ if (code!=205 && code!=400) {
+ connfail(conn, "peer gave unexpected response to QUIT (%s): %s",
+ conn->quitting, sani);
} else {
- notice("C%d idle connection closed by us", conn->fd);
- assert(!conn_busy);
LIST_REMOVE(conns,conn);
+ notice("C%d (now %d) idle connection closed (%s)",
+ conn->fd, conns.count, conn->quitting);
+ assert(!busy);
conn_dispose(conn);
}
return OOP_CONTINUE;
default: PEERBADMSG("peer sent unexpected message");
case 400:
- if (conn_busy)
+ if (busy)
PEERBADMSG("peer timed us out or stopped accepting articles");
- notice("C%d idle connection closed by peer", conn->fd);
LIST_REMOVE(conns,conn);
+ notice("C%d (now %d) idle connection closed by peer",
+ conns.count, conn->fd);
conn_dispose(conn);
return OOP_CONTINUE;
main_input_file= open_input_file(feedfile);
if (!main_input_file)
- die("flush succeeded but feedfile %s does not exist!", feedfile);
+ die("flush succeeded but feedfile %s does not exist!"
+ " (this probably means feedfile does not correspond"
+ " to site %s in newsfeeds)", feedfile, sitename);
if (flushing_input_file) {
- SMS(SEPARATED, max_separated_periods, "recovery flush complete");
+ SMS(SEPARATED, max_separated_periods, "flush complete");
} else {
close_defer();
- SMS(NORMAL, spontaneous_flush_periods, "flush complete");
+ SMS(NORMAL, spontaneous_flush_periods, "recovery flush complete");
}
return OOP_CONTINUE;
debug("PERIOD"
" sms=%s[%d] conns=%d until_connect=%d"
" input_files main:%s flushing:%s backlog:%s[%d]"
- " children connecting=%ld inndcomm=%ld"
+ " children connecting=%ld inndcomm=%ld lowvol_total=%d"
,
sms_names[sms], until_flush, conns.count, until_connect,
dipf_main, dipf_flushing, dipf_backlog, until_backlog_nextscan,
- (long)connecting_child, (long)inndcomm_child
+ (long)connecting_child, (long)inndcomm_child, lowvol_total
);
free(dipf_main);
DUMPV("%d", , cli_master);
fprintf(f,"\n");
+ fprintf(f,"lowvol");
+ DUMPV("%d", , lowvol_circptr);
+ DUMPV("%d", , lowvol_total);
+ fprintf(f,":");
+ for (i=0; i<lowvol_periods; i++) {
+ fprintf(f," ");
+ if (i==lowvol_circptr) fprintf(f,"*");
+ fprintf(f,"%d",lowvol_perperiod[i]);
+ }
+ fprintf(f,"\n");
+
fprintf(f,"filemon ");
filemon_method_dump_info(f);
fprintf(f,"C%d",conn->fd);
DUMPV("%p",conn->,rd); DUMPV("%d",conn->,max_queue);
- DUMPV("%d",conn->,stream); DUMPV("%d",conn->,quitting);
+ DUMPV("%d",conn->,stream); DUMPV("\"%s\"",conn->,quitting);
DUMPV("%d",conn->,since_activity);
fprintf(f,"\n");
fprintf(stderr, "bad usage: %s\n"
"say --help for help, or read the manpage\n",
m);
- if (become_daemon)
+ if (interactive < 2)
syslog(LOG_CRIT,"innduct: invoked with bad usage: %s",m);
exit(8);
}
static const Option innduct_options[]= {
{'f',"feedfile", "F", &feedfile, op_string },
{'q',"quiet-multiple", 0, &quiet_multiple, op_setint, 1 },
-{0,"no-daemon", 0, &become_daemon, op_setint, 0 },
+{0,"no-daemon", 0, &interactive, op_setint, 1 },
+{0,"interactive", 0, &interactive, op_setint, 2 },
{0,"no-streaming", 0, &try_stream, op_setint, 0 },
{0,"no-filemon", 0, &try_filemon, op_setint, 0 },
{'C',"inndconf", "F", &inndconffile, op_string },
{0,"max-flush-interval", "PERIOD", &spontaneous_flush_periods,op_seconds },
{0,"flush-finish-timeout", "PERIOD", &max_separated_periods, op_seconds },
{0,"idle-timeout", "PERIOD", &need_activity_periods, op_seconds },
+{0,"low-volume-thresh", "PERIOD", &lowvol_thresh, op_integer },
+{0,"low-volume-window", "PERIOD", &lowvol_periods, op_seconds },
{0,"max-bad-input-data-ratio","PERCENT", &max_bad_data_ratio, op_double },
{0,"max-bad-input-data-init", "PERCENT", &max_bad_data_initial, op_integer },
convert_to_periods_rndup(&spontaneous_flush_periods);
convert_to_periods_rndup(&max_separated_periods);
convert_to_periods_rndup(&need_activity_periods);
+ convert_to_periods_rndup(&lowvol_periods);
if (max_bad_data_ratio < 0 || max_bad_data_ratio > 100)
badusage("bad input data ratio must be between 0..100");
if (strchr(feedfile, c))
badusage("feed filename may not contain metacharacter %c",c);
+ int i;
+ lowvol_perperiod= xcalloc(sizeof(*lowvol_perperiod), lowvol_periods);
+ for (i=0; i<lowvol_periods; i++) {
+ lowvol_perperiod[i]= lowvol_thresh;
+ lowvol_total += lowvol_thresh;
+ }
+ lowvol_total -= lowvol_thresh;
+
/* set things up */
path_lock= xasprintf("%s_lock", feedfile);
LIST_INIT(conns);
- if (become_daemon) {
+ if (interactive < 1) {
int i;
for (i=3; i<255; i++)
/* do this now before we open syslog, etc. */
close(i);
+ }
+
+ if (interactive < 2)
openlog("innduct",LOG_NDELAY|LOG_PID,LOG_NEWS);
+ if (interactive < 1) {
int null= open("/dev/null",O_RDWR);
if (null<0) sysfatal("failed to open /dev/null");
dup2(null,0);
r= SMsetup(SM_PREOPEN, &val); if (!r) warn("SMsetup SM_PREOPEN failed");
r= SMinit(); if (!r) die("storage manager initialisation (SMinit) failed");
- if (!become_daemon)
+ if (interactive >= 2)
cli_stdio();
cli_init();
~ian / inn-innduct.git / blobdiff