static void check_assign_articles(void);
static void queue_check_input_done(void);
+static void check_reading_pause_resume(InputFile *ipf);
static void statemc_check_flushing_done(void);
static void statemc_check_backlog_done(void);
static void inputfile_reading_start(InputFile *ipf);
static void inputfile_reading_stop(InputFile *ipf);
+static void inputfile_reading_pause(InputFile *ipf);
+static void inputfile_reading_resume(InputFile *ipf);
+ /* pause and resume are idempotent, and no-op if not done _reading_start */
static void filemon_start(InputFile *ipf);
static void filemon_stop(InputFile *ipf);
static int target_max_feedfile_size=100000;
static int period_seconds=60;
static int filepoll_seconds=5;
+static int max_queue_per_ipf=-1;
static int connection_setup_timeout=200;
static int inndcomm_flush_timeout=100;
oop_read *rd; /* non-0: reading; 0: constructing, or had EOF */
off_t offset;
- int skippinglong;
+ int skippinglong, paused;
ArticleList queue;
long inprogress; /* includes queue.count and also articles in conns */
long autodefer; /* -1 means not doing autodefer */
int counts[art_MaxState][RCI_max];
- int readcount_ok, readcount_blank, readcount_err;
+ int readcount_ok, readcount_blank, readcount_err, count_nooffer_missing;
char path[];
};
requeue[art->state]++;
if (art->state==art_Unsolicited) art->state= art_Unchecked;
LIST_ADDTAIL(art->ipf->queue,art);
+ check_reading_pause_resume(art->ipf);
}
int i;
static Article *dequeue_from(int peek, InputFile *ipf) {
if (!ipf) return 0;
if (peek) return LIST_HEAD(ipf->queue);
- else return LIST_REMHEAD(ipf->queue);
+
+ Article *art= LIST_REMHEAD(ipf->queue);
+ if (!art) return 0;
+ check_reading_pause_resume(ipf);
+ return art;
}
static Article *dequeue(int peek) {
}
}
-/*---------- expiry and deferral ----------*/
+/*---------- expiry, flow control and deferral ----------*/
+
+static void check_reading_pause_resume(InputFile *ipf) {
+ if (ipf->queue.count >= max_queue_per_ipf)
+ inputfile_reading_pause(ipf);
+ else
+ inputfile_reading_resume(ipf);
+}
static void article_defer(Article *art /* not on a queue */, int whichcount) {
open_defer();
LIST_REMOVE(art->ipf->queue, art);
art->missing= 1;
- art->ipf->counts[art_Unchecked][RC_missing]++;
+ art->ipf->count_nooffer_missing++;
article_done(art,-1);
return 1;
}
int exp= article_check_expired(art);
if (!exp) break;
}
+ check_reading_pause_resume(ipf);
}
static void article_autodefer(InputFile *ipf, Article *art) {
return 0;
}
-static void autodefer_input_file(InputFile *ipf) {
- ipf->autodefer= 0;
-
+static void autodefer_input_file_articles(InputFile *ipf) {
Article *art;
while ((art= LIST_REMHEAD(ipf->queue)))
article_autodefer(ipf, art);
+}
+
+static void autodefer_input_file(InputFile *ipf) {
+ ipf->autodefer= 0;
+
+ autodefer_input_file_articles(ipf);
if (ipf->inprogress) {
Conn *walk;
found:
connfail(walk, "connection is stuck or crawling,"
" and we need to finish flush");
+ autodefer_input_file_articles(ipf);
}
}
+
+ check_reading_pause_resume(ipf);
}
/*========== article transmission ==========*/
}
static void article_done(Article *art, int whichcount) {
- if (!art->missing) art->ipf->counts[art->state][whichcount]++;
+ if (whichcount>=0 && !art->missing)
+ art->ipf->counts[art->state][whichcount]++;
if (whichcount == RC_accepted) update_nocheck(1);
else if (whichcount == RC_unwanted) update_nocheck(0);
ipf->offset >= target_max_feedfile_size)
statemc_start_flush("feed file size");
- check_assign_articles();
+ check_assign_articles(); /* may destroy conn but that's OK */
+ check_reading_pause_resume(ipf);
return OOP_CONTINUE;
}
OOP_RD_SHORTREC_LONG,
};
+static void inputfile_reading_resume(InputFile *ipf) {
+ if (!ipf->rd) return;
+ if (!ipf->paused) return;
+
+ int r= oop_rd_read(ipf->rd, &feedfile_rdstyle, MAX_LINE_FEEDFILE,
+ feedfile_got_article,ipf, feedfile_read_err, ipf);
+ if (r) sysdie("unable start reading feedfile %s",ipf->path);
+
+ ipf->paused= 0;
+}
+
+static void inputfile_reading_pause(InputFile *ipf) {
+ if (!ipf->rd) return;
+ if (ipf->paused) return;
+ oop_rd_cancel(ipf->rd);
+ ipf->paused= 1;
+}
+
static void inputfile_reading_start(InputFile *ipf) {
assert(!ipf->rd);
ipf->readable.on_readable= tailing_on_readable;
ipf->rd= oop_rd_new(loop, &ipf->readable, 0,0);
assert(ipf->rd);
- int r= oop_rd_read(ipf->rd, &feedfile_rdstyle, MAX_LINE_FEEDFILE,
- feedfile_got_article,ipf, feedfile_read_err, ipf);
- if (r) sysdie("unable start reading feedfile %s",ipf->path);
+ ipf->paused= 1;
+ inputfile_reading_resume(ipf);
}
static void inputfile_reading_stop(InputFile *ipf) {
assert(ipf->rd);
- oop_rd_cancel(ipf->rd);
+ inputfile_reading_pause(ipf);
oop_rd_delete(ipf->rd);
ipf->rd= 0;
assert(!ipf->filemon); /* we shouldn't be monitoring it now */
: xasprintf("%s","");
info("%s %s%s read=%d (+bl=%d,+err=%d)%s%s"
- " offered=%d (ch=%d,nc=%d) accepted=%d (ch=%d,nc=%d)"
+ " missing=%d offered=%d (ch=%d,nc=%d) accepted=%d (ch=%d,nc=%d)"
RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_FMT)
,
completed?"completed":"processed", what, spec,
ipf->readcount_ok, ipf->readcount_blank, ipf->readcount_err,
- inprog, autodefer,
+ inprog, autodefer, ipf->count_nooffer_missing,
CNT(Unchecked,sent) + CNT(Unsolicited,sent)
, CNT(Unchecked,sent), CNT(Unsolicited,sent),
CNT(Wanted,accepted) + CNT(Unsolicited,accepted)
const char *slash= strrchr(ipf->path,'/');
const char *path= slash ? slash+1 : ipf->path;
- return xasprintf("%p/%s:queue=%d,ip=%ld,autodef=%ld,off=%ld,fd=%d%s%s",
+ return xasprintf("%p/%s:queue=%d,ip=%ld,autodef=%ld,off=%ld,fd=%d%s%s%s",
ipf, path,
ipf->queue.count, ipf->inprogress, ipf->autodefer,
(long)ipf->offset, ipf->fd,
ipf->rd ? "" : ",!rd",
- ipf->skippinglong ? "*skiplong" : "");
+ ipf->skippinglong ? "*skiplong" : "",
+ ipf->rd && ipf->paused ? "*paused" : "");
}
static void period(void) {
debug("PERIOD"
" sms=%s[%d] conns=%d until_connect=%d"
- " input_files main:%s flushing:%s backlog:%s"
+ " input_files main:%s flushing:%s backlog:%s[%d]"
" children connecting=%ld inndcomm=%ld"
,
sms_names[sms], until_flush, conns.count, until_connect,
- dipf_main, dipf_flushing, dipf_backlog,
+ dipf_main, dipf_flushing, dipf_backlog, until_backlog_nextscan,
(long)connecting_child, (long)inndcomm_child
);
{0,"max-connections", "N", &max_connections, op_integer },
{0,"max-queue-per-conn", "N", &max_queue_per_conn, op_integer },
+{0,"max-queue-per-file", "N", &max_queue_per_ipf, op_integer },
{0,"feedfile-flush-size","BYTES", &target_max_feedfile_size, op_integer },
{0,"period-interval", "TIME", &period_seconds, op_seconds },
feedfile= xasprintf("%s%s",feedfile,sitename);
}
+ if (max_queue_per_ipf<0)
+ max_queue_per_ipf= max_queue_per_conn * 2;
+
const char *feedfile_forbidden= "?*[~#";
int c;
while ((c= *feedfile_forbidden++))