wip compile

[innduct.git] / backends / innduct.c

diff --git a/backends/innduct.c b/backends/innduct.c
index 1104ef4a7244ea160d890e85657265781a1e291e..8942f6e560ed4247f45823d812e6dec69e27361f 100644 (file)
--- a/backends/innduct.c
+++ b/backends/innduct.c
@@ -252,7 +252,7 @@ DEFLIST(Article);
 
 /*----- function predeclarations -----*/
 
-static void conn_check_work(Conn *conn);
+static void conn_maybe_write(Conn *conn);
 static void conn_make_some_xmits(Conn *conn);
 static void *conn_write_some_xmits(Conn *conn);
 
@@ -270,6 +270,8 @@ static void queue_check_input_done(void);
 static void postfork(const char *what);
 static void postfork_inputfile(InputFile *ipf);
 
+static void open_defer(void);
+
 /*----- configuration options -----*/
 
 static char *sitename, *feedfile;
@@ -394,7 +396,8 @@ static const char *sms_names[]= {
 struct Conn {
   ISNODE(Conn);
   int fd, max_queue, stream, quitting;
-  ArticleList queue; /* not yet told peer, or CHECK said send it */
+  ArticleList waiting; /* not yet told peer */
+  ArticleList priority; /* peer says send it now */
   ArticleList sent; /* offered/transmitted - in xmit or waiting reply */
   struct iovec xmit[CONNIOVS];
   XmitDetails xmitd[CONNIOVS];
@@ -737,6 +740,12 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
   connect_attempt_discard();
 }
 
+static int allow_connect_start(void) {
+  return conns.count < max_connections
+    && !connecting_child
+    && !until_connect;
+}
+
 static void connect_start(void) {
   assert(!connecting_sockets[0]);
   assert(!connecting_sockets[1]);
@@ -837,8 +846,16 @@ static void check_master_queue(void) {
     
     Conn *walk, *use=0;
     int spare;
+
+    /* Find a connection to offer this article.  We prefer a busy
+     * connection to an idle one, provided it's not full.  We take the
+     * first (oldest) and since that's stable, it will mean we fill up
+     * connections in order.  That way if we have too many
+     * connections, the spare ones will go away eventually.
+     */
     for (walk=LIST_HEAD(conns); walk; walk=LIST_NEXT(walk)) {
-      int inqueue= walk->sent.count + walk->queue.count;
+      int inqueue= walk->sent.count + walk->priority.count
+                + walk->waiting.count;
       spare= walk->max_queue - inqueue;
       assert(inqueue <= max_queue_per_conn);
       assert(spare >= 0);
@@ -848,12 +865,11 @@ static void check_master_queue(void) {
     if (use) {
       while (spare>0) {
        Article *art= LIST_REMHEAD(queue);
-       LIST_ADDTAIL(use->queue, art);
+       LIST_ADDTAIL(use->waiting, art);
        spare--;
       }
-      conn_check_work(use);
-    } else if (conns.count < max_connections &&
-            !connecting_child && !until_connect) {
+      conn_maybe_write(use);
+    } else if (allow_connect_start()) {
       until_connect= reconnect_delay_periods;
       connect_start();
       break;
@@ -864,11 +880,11 @@ static void check_master_queue(void) {
 }
 
 static void *conn_writeable(oop_source *l, int fd, oop_event ev, void *u) {
-  conn_check_work(u);
+  conn_maybe_write(u);
   return OOP_CONTINUE;
 }
 
-static void conn_check_work(Conn *conn)  {
+static void conn_maybe_write(Conn *conn)  {
   void *rp= 0;
   for (;;) {
     conn_make_some_xmits(conn);
@@ -898,7 +914,8 @@ static void vconnfail(Conn *conn, const char *fmt, va_list al) {
   int requeue[art_MaxState];
 
   Article *art;
-  while ((art= LIST_REMHEAD(conn->queue))) LIST_ADDTAIL(queue, art);
+  while ((art= LIST_REMHEAD(conn->priority))) LIST_ADDTAIL(queue, art);
+  while ((art= LIST_REMHEAD(conn->waiting))) LIST_ADDTAIL(queue, art);
   while ((art= LIST_REMHEAD(conn->sent))) {
     requeue[art->state]++;
     if (art->state==art_Unsolicited) art->state= art_Unchecked;
@@ -1003,7 +1020,8 @@ static void conn_make_some_xmits(Conn *conn) {
     if (conn->xmitu+5 > CONNIOVS)
       break;
 
-    Article *art= LIST_REMHEAD(queue);
+    Article *art= LIST_REMHEAD(conn->priority);
+    if (!art) art= LIST_REMHEAD(conn->waiting);
     if (!art) break;
 
     if (art->state >= art_Wanted || (conn->stream && nocheck)) {
@@ -1045,7 +1063,7 @@ static void conn_make_some_xmits(Conn *conn) {
       XMIT_LITERAL("\r\n");
 
       assert(art->state == art_Unchecked);
-      art->ipf->counts[art->state][RCI_sent]++;
+      art->ipf->counts[art->state][RC_sent]++;
       LIST_ADDTAIL(conn->sent, art);
     }
   }
@@ -1199,7 +1217,8 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
       connfail(conn, "peer gave unexpected response to QUIT: %s", sani);
     } else {
       notice("#%d idle connection closed\n");
-      assert(!conn->queue.count);
+      assert(!conn->waiting.count);
+      assert(!conn->priority.count);
       assert(!conn->sent.count);
       assert(!conn->xmitu);
       LIST_REMOVE(conns,conn);
@@ -1245,7 +1264,7 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
     assert(art->state == art_Unchecked);
     art->ipf->counts[art->state][RC_accepted]++;
     art->state= art_Wanted;
-    LIST_ADDTAIL(conn->queue, art);
+    LIST_ADDTAIL(conn->priority, art);
     break;
 
   case 431: /* CHECK or TAKETHIS says try later */
@@ -1261,7 +1280,8 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
 
   }
 
-  check_check_work(conn);
+  conn_maybe_write(sendon);
+  check_master_queue();
   return OOP_CONTINUE;
 }