conn.c

   1 /*
   2  *  innduct
   3  *  tailing reliable realtime streaming feeder for inn
   4  *  conn.c - connection establishment and teardown
   5  *
   6  *  Copyright Ian Jackson <ijackson@chiark.greenend.org.uk>
   7  *  and contributors; see LICENCE.txt.
   8  *  SPDX-License-Identifier: GPL-3.0-or-later
   9  */
  10
  11 #include "innduct.h"
  12
  13 /*========== management of connections ==========*/
  14
  15 static void reconnect_blocking_event(void) {
  16   until_connect= reconnect_delay_periods;
  17 }
  18
  19 void conn_closefd(Conn *conn, const char *msgprefix) {
  20   int r= close_perhaps(&conn->fd);
  21   if (r) info("C%d %serror closing socket: %s",
  22               conn->fd, msgprefix, strerror(errno));
  23 }
  24
  25 int conn_busy(Conn *conn) {
  26   return
  27     conn->waiting.count ||
  28     conn->priority.count ||
  29     conn->sent.count ||
  30     conn->xmitu;
  31 }
  32
  33 void conn_dispose(Conn *conn) {
  34   if (!conn) return;
  35   if (conn->rd) {
  36     oop_rd_cancel(conn->rd);
  37     oop_rd_delete_kill(conn->rd);
  38     conn->rd= 0;
  39   }
  40   if (conn->fd) {
  41     loop->cancel_fd(loop, conn->fd, OOP_WRITE);
  42     loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
  43   }
  44   conn_closefd(conn,"");
  45   free(conn);
  46 }
  47
  48 static void *conn_exception(oop_source *lp, int fd,
  49                             oop_event ev, void *conn_v) {
  50   Conn *conn= conn_v;
  51   unsigned char ch;
  52   assert(fd == conn->fd);
  53   assert(ev == OOP_EXCEPTION);
  54   int r= read(conn->fd, &ch, 1);
  55   if (r<0) connfail(conn,"read failed: %s",strerror(errno));
  56   else connfail(conn,"exceptional condition on socket (peer sent urgent"
  57                 " data? read(,&ch,1)=%d,ch='\\x%02x')",r,ch);
  58   return OOP_CONTINUE;
  59 }
  60
  61 void vconnfail(Conn *conn, const char *fmt, va_list al) {
  62   int requeue[art_MaxState];
  63   memset(requeue,0,sizeof(requeue));
  64
  65   Article *art;
  66
  67   while ((art= LIST_REMHEAD(conn->priority)))
  68     LIST_ADDTAIL(art->ipf->queue, art);
  69
  70   while ((art= LIST_REMHEAD(conn->waiting)))
  71     LIST_ADDTAIL(art->ipf->queue, art);
  72
  73   while ((art= LIST_REMHEAD(conn->sent))) {
  74     requeue[art->state]++;
  75     if (art->state==art_Unsolicited) art->state= art_Unchecked;
  76     LIST_ADDTAIL(art->ipf->queue,art);
  77     check_reading_pause_resume(art->ipf);
  78   }
  79
  80   int i;
  81   XmitDetails *d;
  82   for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
  83     xmit_free(d);
  84
  85   LIST_REMOVE(conns,conn);
  86
  87   char *m= mvasprintf(fmt,al);
  88   warn("C%d (now %d) connection failed "
  89        "(requeueing " RCI_TRIPLE_FMT_BASE "): %s",
  90        conn->fd, conns.count, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
  91   free(m);
  92
  93   reconnect_blocking_event();
  94   conn_dispose(conn);
  95   check_assign_articles();
  96 }
  97
  98 void connfail(Conn *conn, const char *fmt, ...) {
  99   va_list al;
 100   va_start(al,fmt);
 101   vconnfail(conn,fmt,al);
 102   va_end(al);
 103 }
 104
 105 static void conn_idle_close(Conn *conn, const char *why) {
 106   static const char quitcmd[]= "QUIT\r\n";
 107   int todo= sizeof(quitcmd)-1;
 108   const char *p= quitcmd;
 109   for (;;) {
 110     int r= write(conn->fd, p, todo);
 111     if (r<0) {
 112       if (isewouldblock(errno))
 113         connfail(conn, "blocked writing QUIT to idle connection");
 114       else
 115         connfail(conn, "failed to write QUIT to idle connection: %s",
 116                  strerror(errno));
 117       break;
 118     }
 119     assert(r<=todo);
 120     todo -= r;
 121     if (!todo) {
 122       conn->quitting= why;
 123       conn->since_activity= 0;
 124       dbg("C%d is idle (%s), quitting", conn->fd, why);
 125       break;
 126     }
 127   }
 128 }
 129
 130 /*
 131  * For our last connection, we also shut it down if we have had
 132  * less than K in the last L
 133  */
 134 void check_idle_conns(void) {
 135   Conn *conn;
 136
 137   int volthisperiod= lowvol_perperiod[lowvol_circptr];
 138   lowvol_circptr++;
 139   lowvol_circptr %= lowvol_periods;
 140   lowvol_total += volthisperiod;
 141   lowvol_total -= lowvol_perperiod[lowvol_circptr];
 142   lowvol_perperiod[lowvol_circptr]= 0;
 143
 144   FOR_CONN(conn)
 145     conn->since_activity++;
 146
 147  search_again:
 148   FOR_CONN(conn) {
 149     if (conn->since_activity <= need_activity_periods) continue;
 150
 151     /* We need to shut this down */
 152     if (conn->quitting)
 153       connfail(conn,"timed out waiting for response to QUIT (%s)",
 154                conn->quitting);
 155     else if (conn->sent.count)
 156       connfail(conn,"timed out waiting for responses");
 157     else if (conn->waiting.count || conn->priority.count)
 158       connfail(conn,"BUG IN INNDUCT conn has queue but nothing sent");
 159     else if (conn->xmitu)
 160       connfail(conn,"peer has been sending responses"
 161                " before receiving our commands!");
 162     else
 163       conn_idle_close(conn, "no activity");
 164
 165     goto search_again;
 166   }
 167
 168   conn= LIST_HEAD(conns);
 169   if (!volthisperiod &&
 170       conns.count==1 &&
 171       lowvol_total < lowvol_thresh &&
 172       !conn_busy(conn))
 173     conn_idle_close(conn, "low volume");
 174 }
 175
 176 /*---------- reporting numbers of connections ----------*/
 177
 178 static int conns_max_reported, conns_idle_reported;
 179
 180 void notice_conns_more(const char *new_kind) {
 181   if (conns.count > conns_max_reported) {
 182     notice("up to %d connection(s) (%s)", conns.count, new_kind);
 183     conns_max_reported= conns.count;
 184   }
 185 }
 186
 187 void notice_conns_fewer(void) {
 188   if (!conns.count && !conns_idle_reported) {
 189     notice("low volume, using intermittent connection");
 190     conns_idle_reported= 1;
 191   }
 192 }
 193
 194 void notice_conns_stats(void) {
 195   notice("currently %d connection(s)", conns.count);
 196   conns_max_reported= conns.count;
 197   conns_idle_reported= 0;
 198 }
 199
 200 /*---------- making new connections ----------*/
 201
 202 pid_t connecting_child;
 203 int connecting_fdpass_sock;
 204
 205 static void connect_attempt_discard(void) {
 206   if (connecting_child) {
 207     int status= xwaitpid(&connecting_child, "connect");
 208     if (!(WIFEXITED(status) ||
 209           (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)))
 210       report_child_status("connect", status);
 211   }
 212   if (connecting_fdpass_sock) {
 213     cancel_fd_read_except(connecting_fdpass_sock);
 214     xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
 215   }
 216 }
 217
 218 #define PREP_DECL_MSG_CMSG(msg)                 \
 219   char msgbyte= 0;                              \
 220   struct iovec msgiov;                          \
 221   msgiov.iov_base= &msgbyte;                    \
 222   msgiov.iov_len= 1;                            \
 223   struct msghdr msg;                            \
 224   memset(&msg,0,sizeof(msg));                   \
 225   char msg##cbuf[CMSG_SPACE(sizeof(int))];      \
 226   msg.msg_iov= &msgiov;                         \
 227   msg.msg_iovlen= 1;                            \
 228   msg.msg_control= msg##cbuf;                   \
 229   msg.msg_controllen= sizeof(msg##cbuf);
 230
 231 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
 232   Conn *conn= 0;
 233
 234   assert(fd == connecting_fdpass_sock);
 235
 236   PREP_DECL_MSG_CMSG(msg);
 237
 238   ssize_t rs= recvmsg(fd, &msg, 0);
 239   if (rs<0) {
 240     if (isewouldblock(errno)) return OOP_CONTINUE;
 241     syswarn("failed to read socket from connecting child");
 242     goto x;
 243   }
 244
 245   NEW(conn);
 246   LIST_INIT(conn->waiting);
 247   LIST_INIT(conn->priority);
 248   LIST_INIT(conn->sent);
 249
 250   struct cmsghdr *h= 0;
 251   if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
 252   if (!h) {
 253     int status= xwaitpid(&connecting_child, "connect child (broken)");
 254
 255     if (WIFEXITED(status)) {
 256       if (WEXITSTATUS(status) != 0 &&
 257           WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
 258           WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
 259         /* child already reported the problem */;
 260       else {
 261         if (e == OOP_EXCEPTION)
 262           warn("connect: connection child exited code %d but"
 263                " unexpected exception on fdpass socket",
 264                WEXITSTATUS(status));
 265         else
 266           warn("connect: connection child exited code %d but"
 267                " no cmsg (rs=%d)",
 268                WEXITSTATUS(status), (int)rs);
 269       }
 270     } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
 271       warn("connect: connection attempt timed out");
 272     } else {
 273       report_child_status("connect", status);
 274     }
 275     goto x;
 276   }
 277
 278 #define CHK(field, val)                                                    \
 279   if (h->cmsg_##field != val) {                                            \
 280     crash("connect: child sent cmsg with cmsg_" #field "=%ld, expected %ld", \
 281           (long)h->cmsg_##field, (long)val);                               \
 282     goto x;                                                                \
 283   }
 284   CHK(level, SOL_SOCKET);
 285   CHK(type,  SCM_RIGHTS);
 286   CHK(len,   CMSG_LEN(sizeof(conn->fd)));
 287 #undef CHK
 288
 289   if (CMSG_NXTHDR(&msg,h)) crash("connect: child sent many cmsgs");
 290
 291   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
 292
 293   int status;
 294   pid_t got= waitpid(connecting_child, &status, 0);
 295   if (got==-1) syscrash("connect: real wait for child");
 296   assert(got == connecting_child);
 297   connecting_child= 0;
 298
 299   if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
 300   int es= WEXITSTATUS(status);
 301   switch (es) {
 302   case CONNCHILD_ESTATUS_STREAM:    conn->stream= 1;   break;
 303   case CONNCHILD_ESTATUS_NOSTREAM:  conn->stream= 0;   break;
 304   default:
 305     die("connect: child gave unexpected exit status %d", es);
 306   }
 307
 308   /* Phew! */
 309   conn->max_queue= conn->stream ? max_queue_per_conn : 1;
 310
 311   loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
 312   conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
 313   if (!conn->fd) crash("oop_rd_new_fd conn failed (fd=%d)",conn->fd);
 314   int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_MAXLEN_COMMAND+1,
 315                      &peer_rd_ok, conn,
 316                      &peer_rd_err, conn);
 317   if (r) syscrash("oop_rd_read for peer (fd=%d)",conn->fd);
 318
 319   LIST_ADDHEAD(conns, conn);
 320   const char *streamdesc= conn->stream ? "streaming" : "plain";
 321   info("C%d (now %d) connected %s", conn->fd, conns.count, streamdesc);
 322   notice_conns_more(streamdesc);
 323
 324   connect_attempt_discard();
 325   check_assign_articles();
 326   return OOP_CONTINUE;
 327
 328  x:
 329   conn_dispose(conn);
 330   connect_attempt_discard();
 331   reconnect_blocking_event();
 332   return OOP_CONTINUE;
 333 }
 334
 335 int allow_connect_start(void) {
 336   return conns.count < max_connections
 337     && !connecting_child
 338     && !until_connect;
 339 }
 340
 341 void connect_start(void) {
 342   assert(!connecting_child);
 343   assert(!connecting_fdpass_sock);
 344
 345   info("starting connection attempt");
 346   int ok_until_connect= until_connect;
 347   reconnect_blocking_event();
 348
 349   int socks[2];
 350   int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
 351   if (r) { syswarn("connect: cannot create socketpair for child"); return; }
 352
 353   connecting_child= xfork("connection");
 354
 355   if (!connecting_child) {
 356     FILE *cn_from, *cn_to;
 357     char buf[NNTP_MAXLEN_COMMAND+100];
 358     int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
 359
 360     xclose(socks[0], "(in child) parent's connection fdpass socket",0);
 361
 362     alarm(connection_setup_timeout);
 363     buf[sizeof(buf)-1] = 0;
 364     if (NNTPconnect(remote_host, port, &cn_from, &cn_to,
 365                     buf, sizeof(buf)-1) < 0) {
 366       int l= strlen(buf);
 367       int stripped=0;
 368       while (l>0) {
 369         unsigned char c= buf[l-1];
 370         if (!isspace(c)) break;
 371         if (c=='\n' || c=='\r') stripped=1;
 372         --l;
 373       }
 374       if (!buf[0]) {
 375         sysdie("connect: connection attempt failed");
 376       } else {
 377         buf[l]= 0;
 378         die("connect: %s: %s", stripped ? "rejected" : "failed",
 379             sanitise(buf,-1));
 380       }
 381     }
 382     if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
 383       sysdie("connect: authentication failed");
 384     if (try_stream) {
 385       if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
 386           fflush(cn_to))
 387         sysdie("connect: could not send MODE STREAM");
 388       buf[sizeof(buf)-1]= 0;
 389       if (!fgets(buf, sizeof(buf)-1, cn_from)) {
 390         if (ferror(cn_from))
 391           sysdie("connect: could not read response to MODE STREAM");
 392         else
 393           die("connect: connection close in response to MODE STREAM");
 394       }
 395       int l= strlen(buf);
 396       assert(l>=1);
 397       if (buf[l-1]!='\n')
 398         die("connect: response to MODE STREAM is too long: %.100s...",
 399             sanitise(buf,-1));
 400       l--;  if (l>0 && buf[l-1]=='\r') l--;
 401       buf[l]= 0;
 402       char *ep;
 403       int rcode= strtoul(buf,&ep,10);
 404       if (ep != &buf[3])
 405         die("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1));
 406
 407       switch (rcode) {
 408       case 203:
 409         exitstatus= CONNCHILD_ESTATUS_STREAM;
 410         break;
 411       case 480:
 412       case 500:
 413         break;
 414       default:
 415         warn("connect: unexpected response to MODE STREAM: %.50s",
 416              sanitise(buf,-1));
 417         exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
 418         break;
 419       }
 420     }
 421     int fd= fileno(cn_from);
 422
 423     PREP_DECL_MSG_CMSG(msg);
 424     struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
 425     cmsg->cmsg_level= SOL_SOCKET;
 426     cmsg->cmsg_type=  SCM_RIGHTS;
 427     cmsg->cmsg_len=   CMSG_LEN(sizeof(fd));
 428     memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
 429
 430     msg.msg_controllen= cmsg->cmsg_len;
 431     r= sendmsg(socks[1], &msg, 0);
 432     if (r<0) syscrash("sendmsg failed for new connection");
 433     if (r!=1) crash("sendmsg for new connection gave wrong result %d",r);
 434
 435     _exit(exitstatus);
 436   }
 437
 438   xclose(socks[1], "connecting fdpass child's socket",0);
 439   connecting_fdpass_sock= socks[0];
 440   xsetnonblock(connecting_fdpass_sock, 1);
 441   on_fd_read_except(connecting_fdpass_sock, connchild_event);
 442
 443   if (!conns.count)
 444     until_connect= ok_until_connect;
 445 }
 446