+/*
+ * TODO
+ * - close idle connections
+ * - -k kill mode ?
+ */
+
/*
* Newsfeeds file entries should look like this:
* host.name.of.site[/exclude,exclude,...]\
static void statemc_setstate(StateMachineState newsms, int periods,
const char *forlog, const char *why);
static void check_master_queue(void);
+static void queue_check_input_done(void);
static void postfork(const char *what);
static void postfork_inputfile(InputFile *ipf);
, counts[art_Unsolicited] x
typedef enum {
-#define RC_INDEX(x) RCI_##x,
+#define RC_INDEX(x) RC_##x,
RESULT_COUNTS(RC_INDEX, RC_INDEX)
RCI_max
} ResultCountIndex;
struct Conn {
ISNODE(Conn);
- int fd, max_queue, stream;
+ int fd, max_queue, stream, quitting;
ArticleList queue; /* not yet told peer, or CHECK said send it */
ArticleList sent; /* offered/transmitted - in xmit or waiting reply */
struct iovec xmit[CONNIOVS];
/*========== making new connections ==========*/
+static void conn_dispose(Conn *conn) {
+ if (!conn) return;
+ perhaps_close(&conn->fd);
+ free(conn);
+ until_connect= reconnect_delay_periods;
+}
+
static int connecting_sockets[2]= {-1,-1};
static pid_t connecting_child;
return 0;
x:
- if (conn) {
- perhaps_close(&conn->fd);
- free(conn);
- }
+ conn_dispose(conn);
connect_attempt_discard();
}
conn->fd, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
free(m);
- close(conn->fd);
LIST_REMOVE(conns,conn);
- free(conn);
-
- until_connect= reconnect_delay_periods;
+ conn_dispose(conn);
check_master_queue();
}
art->state == art_Unchecked ? art_Unsolicited :
art->state == art_Wanted ? art_Wanted :
(abort(),-1);
- art->ipf->counts[art->state][RCI_sent]++;
+ art->ipf->counts[art->state][RC_sent]++;
LIST_ADDTAIL(conn->sent, art);
} else {
}
static void update_nocheck(int accepted) {
- accept_proportion *= accept_decay;
- accept_proportion += accepted;
+ accept_proportion *= nocheck_decay;
+ accept_proportion += accepted * (1.0 - nocheck_decay);
int new_nocheck= accept_proportion >= nocheck_thresh;
if (new_nocheck && !nocheck_reported) {
notice("entering nocheck mode for the first time");
static void article_done(Conn *conn, Article *art, int whichcount) {
art->ipf->counts[art->state][whichcount]++;
- if (whichcount == RCI_accepted) update_nocheck(1);
- else if (whichcount == RCI_unwanted) update_nocheck(0);
+ if (whichcount == RC_accepted) update_nocheck(1);
+ else if (whichcount == RC_unwanted) update_nocheck(0);
InputFile *ipf= art->ipf;
while (art->blanklen) {
}
assert(ev == OOP_RD_OK);
- char *sani= sanitise(data, recsz);
+ char *sani= sanitise(data);
char *ep;
unsigned long code= strtoul(data, &ep, 10);
}
if (conn->quitting) {
- if (code!=205) {
- connfail(conn, "peer gave failure response to QUIT: %s", sani);
- return OOP_CONTINUE;
+ if (code!=205 && code!=503) {
+ connfail(conn, "peer gave unexpected response to QUIT: %s", sani);
+ } else {
+ notice("#%d idle connection closed\n");
+ assert(!conn->queue.count);
+ assert(!conn->sent.count);
+ assert(!conn->xmitu);
+ LIST_REMOVE(conns,conn);
+ conn_dispose(conn);
}
- conn close ok;
- return;
+ return OOP_CONTINUE;
}
Article *art;
if (art) ; else return OOP_CONTINUE /* reply_check has failed the conn */
#define ARTICLE_DEALTWITH(streaming,musthavesent,how) \
- code_streaming= (streaming) \
+ code_streaming= (streaming); \
GET_ARTICLE(musthavesent); \
article_done(conn, art, RC_##how); break;
case 335: /* IHAVE says send it */
GET_ARTICLE(-1);
assert(art->state == art_Unchecked);
- art->ipf->counts[art->state].accepted++;
+ art->ipf->counts[art->state][RC_accepted]++;
art->state= art_Wanted;
- LIST_ADDTAIL(conn->queue);
+ LIST_ADDTAIL(conn->queue, art);
break;
case 431: /* CHECK or TAKETHIS says try later */