Prioritise TAKETHIS transmissions over further CHECKs

diff --git a/backends/innduct.c b/backends/innduct.c
index c2ca557455c4f5927f54f027a98b6b7155f2b5a1..018d6bca9adfdba6d6bf37b2501365d8f9218612 100644 (file)
--- a/backends/innduct.c
+++ b/backends/innduct.c
@@ -394,7 +394,8 @@ static const char *sms_names[]= {
 struct Conn {
   ISNODE(Conn);
   int fd, max_queue, stream, quitting;
 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];
   ArticleList sent; /* offered/transmitted - in xmit or waiting reply */
   struct iovec xmit[CONNIOVS];
   XmitDetails xmitd[CONNIOVS];


@@ -845,7 +846,8 @@ static void check_master_queue(void) {
      * connections, the spare ones will go away eventually.
      */
     for (walk=LIST_HEAD(conns); walk; walk=LIST_NEXT(walk)) {
      * 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);
       spare= walk->max_queue - inqueue;
       assert(inqueue <= max_queue_per_conn);
       assert(spare >= 0);


@@ -855,7 +857,7 @@ static void check_master_queue(void) {
     if (use) {
       while (spare>0) {
        Article *art= LIST_REMHEAD(queue);
     if (use) {
       while (spare>0) {
        Article *art= LIST_REMHEAD(queue);

-       LIST_ADDTAIL(use->queue, art);
+       LIST_ADDTAIL(use->waiting, art);

        spare--;
       }
       conn_maybe_write(use);
        spare--;
       }
       conn_maybe_write(use);


@@ -905,7 +907,8 @@ static void vconnfail(Conn *conn, const char *fmt, va_list al) {
   int requeue[art_MaxState];
 
   Article *art;
   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;
   while ((art= LIST_REMHEAD(conn->sent))) {
     requeue[art->state]++;
     if (art->state==art_Unsolicited) art->state= art_Unchecked;


@@ -1010,7 +1013,8 @@ static void conn_make_some_xmits(Conn *conn) {
     if (conn->xmitu+5 > CONNIOVS)
       break;
 
     if (conn->xmitu+5 > CONNIOVS)
       break;
 

-    Article *art= LIST_REMHEAD(queue);
+    Article *art= LIST_REMHEAD(priority);
+    if (!art) art= LIST_REMHEAD(waiting);

     if (!art) break;
 
     if (art->state >= art_Wanted || (conn->stream && nocheck)) {
     if (!art) break;
 
     if (art->state >= art_Wanted || (conn->stream && nocheck)) {


@@ -1206,7 +1210,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");
       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);
       assert(!conn->sent.count);
       assert(!conn->xmitu);
       LIST_REMOVE(conns,conn);


@@ -1252,7 +1257,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;
     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 */
     break;
 
   case 431: /* CHECK or TAKETHIS says try later */