+ control_master= socket(PF_UNIX, SOCK_STREAM, 0);
+ if (control_master<0) NOCONTROL("create new control socket");
+
+ sa.un.sun_family= AF_UNIX;
+ int sl= strlen(sa.un.sun_path) + offsetof(struct sockaddr_un, sun_path);
+ r= bind(control_master, &sa.sa, sl);
+ if (r) NOCONTROL("bind to real socket path %s", sa.un.sun_path);
+
+ r= listen(control_master, 5);
+ if (r) NOCONTROL("listen");
+
+ xsetnonblock(control_master, 1);
+
+ loop->on_fd(loop, control_master, OOP_READ, control_master_readable, 0);
+ info("control socket ok, real path %s", sa.un.sun_path);
+
+ return;
+
+ nocontrol:
+ free(real);
+ xclose_perhaps(&control_master, "control master",0);
+ return;
+}
+
+/*========== management of connections ==========*/
+
+static void conn_closefd(Conn *conn, const char *msgprefix) {
+ int r= close_perhaps(&conn->fd);
+ if (r) info("C%d %serror closing socket: %s",
+ conn->fd, msgprefix, strerror(errno));
+}
+
+static void conn_dispose(Conn *conn) {
+ if (!conn) return;
+ if (conn->rd) {
+ oop_rd_cancel(conn->rd);
+ oop_rd_delete_kill(conn->rd);
+ conn->rd= 0;
+ }
+ if (conn->fd) {
+ loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+ loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
+ }
+ conn_closefd(conn,"");
+ free(conn);
+ until_connect= reconnect_delay_periods;
+}
+
+static void *conn_exception(oop_source *lp, int fd,
+ oop_event ev, void *conn_v) {
+ Conn *conn= conn_v;
+ unsigned char ch;
+ assert(fd == conn->fd);
+ assert(ev == OOP_EXCEPTION);
+ int r= read(conn->fd, &ch, 1);
+ if (r<0) connfail(conn,"read failed: %s",strerror(errno));
+ else connfail(conn,"exceptional condition on socket (peer sent urgent"
+ " data? read(,&ch,1)=%d,ch='\\x%02x')",r,ch);
+ return OOP_CONTINUE;
+}
+
+static void vconnfail(Conn *conn, const char *fmt, va_list al) {
+ int requeue[art_MaxState];
+ memset(requeue,0,sizeof(requeue));
+
+ Article *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;
+ LIST_ADDTAIL(queue,art);
+ }
+
+ int i;
+ XmitDetails *d;
+ for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
+ xmit_free(d);
+
+ char *m= xvasprintf(fmt,al);
+ warn("C%d connection failed (requeueing " RCI_TRIPLE_FMT_BASE "): %s",
+ conn->fd, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
+ free(m);
+
+ LIST_REMOVE(conns,conn);
+ conn_dispose(conn);
+ check_assign_articles();
+}
+
+static void connfail(Conn *conn, const char *fmt, ...) {
+ va_list al;
+ va_start(al,fmt);
+ vconnfail(conn,fmt,al);
+ va_end(al);
+}
+
+static void check_idle_conns(void) {
+ Conn *conn;
+ for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn))
+ conn->since_activity++;
+ search_again:
+ for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn)) {
+ if (conn->since_activity <= need_activity_periods) continue;
+
+ /* We need to shut this down */
+ if (conn->quitting)
+ connfail(conn,"timed out waiting for response to QUIT");
+ else if (conn->sent.count)
+ connfail(conn,"timed out waiting for responses");
+ else if (conn->waiting.count || conn->priority.count)
+ connfail(conn,"BUG IN INNDUCT conn has queue but nothing sent");
+ else if (conn->xmitu)
+ connfail(conn,"peer has been sending responses"
+ " before receiving our commands!");
+ else {
+ static const char quitcmd[]= "QUIT\r\n";
+ int todo= sizeof(quitcmd)-1;
+ const char *p= quitcmd;
+ for (;;) {
+ int r= write(conn->fd, p, todo);
+ if (r<0) {
+ if (isewouldblock(errno))
+ connfail(conn, "blocked writing QUIT to idle connection");
+ else
+ connfail(conn, "failed to write QUIT to idle connection: %s",
+ strerror(errno));
+ break;
+ }
+ assert(r<=todo);
+ todo -= r;
+ if (!todo) {
+ conn->quitting= 1;
+ conn->since_activity= 0;
+ debug("C%d is idle, quitting", conn->fd);
+ break;
+ }
+ }
+ }
+ goto search_again;
+ }
+}
+
+/*---------- making new connections ----------*/
+
+static pid_t connecting_child;
+static int connecting_fdpass_sock;
+
+static void connect_attempt_discard(void) {
+ if (connecting_child) {
+ int status= xwaitpid(&connecting_child, "connect");
+ if (!(WIFEXITED(status) ||
+ (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)))
+ report_child_status("connect", status);
+ }
+ if (connecting_fdpass_sock) {
+ cancel_fd_read_except(connecting_fdpass_sock);
+ xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
+ }
+}
+
+#define PREP_DECL_MSG_CMSG(msg) \
+ char msgbyte= 0; \
+ struct iovec msgiov; \
+ msgiov.iov_base= &msgbyte; \
+ msgiov.iov_len= 1; \
+ struct msghdr msg; \
+ memset(&msg,0,sizeof(msg)); \
+ char msg##cbuf[CMSG_SPACE(sizeof(int))]; \
+ msg.msg_iov= &msgiov; \
+ msg.msg_iovlen= 1; \
+ msg.msg_control= msg##cbuf; \
+ msg.msg_controllen= sizeof(msg##cbuf);
+
+static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
+ Conn *conn= 0;
+
+ assert(fd == connecting_fdpass_sock);
+
+ PREP_DECL_MSG_CMSG(msg);
+
+ ssize_t rs= recvmsg(fd, &msg, 0);
+ if (rs<0) {
+ if (isewouldblock(errno)) return OOP_CONTINUE;
+ syswarn("failed to read socket from connecting child");
+ goto x;
+ }
+
+ conn= xmalloc(sizeof(*conn));
+ memset(conn,0,sizeof(*conn));
+ LIST_INIT(conn->waiting);
+ LIST_INIT(conn->priority);
+ LIST_INIT(conn->sent);
+
+ struct cmsghdr *h= 0;
+ if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
+ if (!h) {
+ int status= xwaitpid(&connecting_child, "connect child (broken)");
+
+ if (WIFEXITED(status)) {
+ if (WEXITSTATUS(status) != 0 &&
+ WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
+ WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
+ /* child already reported the problem */;
+ else {
+ if (e == OOP_EXCEPTION)
+ warn("connect: connection child exited code %d but"
+ " unexpected exception on fdpass socket",
+ WEXITSTATUS(status));
+ else
+ warn("connect: connection child exited code %d but"
+ " no cmsg (rs=%d)",
+ WEXITSTATUS(status), (int)rs);
+ }
+ } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
+ warn("connect: connection attempt timed out");
+ } else {
+ report_child_status("connect", status);
+ }
+ goto x;
+ }
+
+#define CHK(field, val) \
+ if (h->cmsg_##field != val) { \
+ die("connect: child sent cmsg with cmsg_" #field "=%d, expected %d", \
+ h->cmsg_##field, val); \
+ goto x; \
+ }
+ CHK(level, SOL_SOCKET);
+ CHK(type, SCM_RIGHTS);
+ CHK(len, CMSG_LEN(sizeof(conn->fd)));
+#undef CHK
+
+ if (CMSG_NXTHDR(&msg,h)) die("connect: child sent many cmsgs");
+
+ memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
+
+ int status;
+ pid_t got= waitpid(connecting_child, &status, 0);
+ if (got==-1) sysdie("connect: real wait for child");
+ assert(got == connecting_child);
+ connecting_child= 0;
+
+ if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
+ int es= WEXITSTATUS(status);
+ switch (es) {
+ case CONNCHILD_ESTATUS_STREAM: conn->stream= 1; break;
+ case CONNCHILD_ESTATUS_NOSTREAM: conn->stream= 0; break;
+ default:
+ fatal("connect: child gave unexpected exit status %d", es);
+ }
+
+ /* Phew! */
+ conn->max_queue= conn->stream ? max_queue_per_conn : 1;
+
+ loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
+ conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
+ if (!conn->fd) die("oop_rd_new_fd conn failed (fd=%d)",conn->fd);
+ int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_STRLEN,
+ &peer_rd_ok, conn,
+ &peer_rd_err, conn);
+ if (r) sysdie("oop_rd_read for peer (fd=%d)",conn->fd);
+
+ notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
+ LIST_ADDHEAD(conns, conn);
+
+ connect_attempt_discard();
+ check_assign_articles();
+ return OOP_CONTINUE;
+
+ x:
+ conn_dispose(conn);
+ connect_attempt_discard();
+ return OOP_CONTINUE;
+}
+
+static int allow_connect_start(void) {
+ return conns.count < max_connections
+ && !connecting_child
+ && !until_connect;
+}
+
+static void connect_start(void) {
+ assert(!connecting_child);
+ assert(!connecting_fdpass_sock);
+
+ info("starting connection attempt");
+
+ int socks[2];
+ int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
+ if (r) { syswarn("connect: cannot create socketpair for child"); return; }
+
+ connecting_child= xfork("connection");
+
+ if (!connecting_child) {
+ FILE *cn_from, *cn_to;
+ char buf[NNTP_STRLEN+100];
+ int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
+
+ xclose(socks[0], "(in child) parent's connection fdpass socket",0);
+
+ alarm(connection_setup_timeout);
+ if (NNTPconnect((char*)remote_host, port, &cn_from, &cn_to, buf) < 0) {
+ int l= strlen(buf);
+ int stripped=0;
+ while (l>0) {
+ unsigned char c= buf[l-1];
+ if (!isspace(c)) break;
+ if (c=='\n' || c=='\r') stripped=1;
+ --l;
+ }
+ if (!buf[0]) {
+ sysfatal("connect: connection attempt failed");
+ } else {
+ buf[l]= 0;
+ fatal("connect: %s: %s", stripped ? "rejected" : "failed",
+ sanitise(buf));
+ }
+ }
+ if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
+ sysfatal("connect: authentication failed");
+ if (try_stream) {
+ if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
+ fflush(cn_to))
+ sysfatal("connect: could not send MODE STREAM");
+ buf[sizeof(buf)-1]= 0;
+ if (!fgets(buf, sizeof(buf)-1, cn_from)) {
+ if (ferror(cn_from))
+ sysfatal("connect: could not read response to MODE STREAM");
+ else
+ fatal("connect: connection close in response to MODE STREAM");
+ }
+ int l= strlen(buf);
+ assert(l>=1);
+ if (buf[l-1]!='\n')
+ fatal("connect: response to MODE STREAM is too long: %.100s...",
+ sanitise(buf));
+ l--; if (l>0 && buf[l-1]=='\r') l--;
+ buf[l]= 0;
+ char *ep;
+ int rcode= strtoul(buf,&ep,10);
+ if (ep != &buf[3])
+ fatal("connect: bad response to MODE STREAM: %.50s", sanitise(buf));
+
+ switch (rcode) {
+ case 203:
+ exitstatus= CONNCHILD_ESTATUS_STREAM;
+ break;
+ case 480:
+ case 500:
+ break;
+ default:
+ warn("connect: unexpected response to MODE STREAM: %.50s",
+ sanitise(buf));
+ exitstatus= 2;
+ break;
+ }
+ }
+ int fd= fileno(cn_from);
+
+ PREP_DECL_MSG_CMSG(msg);
+ struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
+ cmsg->cmsg_level= SOL_SOCKET;
+ cmsg->cmsg_type= SCM_RIGHTS;
+ cmsg->cmsg_len= CMSG_LEN(sizeof(fd));
+ memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
+
+ msg.msg_controllen= cmsg->cmsg_len;
+ r= sendmsg(socks[1], &msg, 0);
+ if (r<0) sysdie("sendmsg failed for new connection");
+ if (r!=1) die("sendmsg for new connection gave wrong result %d",r);
+
+ _exit(exitstatus);
+ }
+
+ xclose(socks[1], "connecting fdpass child's socket",0);
+ connecting_fdpass_sock= socks[0];
+ xsetnonblock(connecting_fdpass_sock, 1);
+ on_fd_read_except(connecting_fdpass_sock, connchild_event);
+}
+
+/*---------- assigning articles to conns, and transmitting ----------*/
+
+static void check_assign_articles(void) {
+ for (;;) {
+ if (!queue.count)
+ break;
+
+ Conn *walk, *use=0;
+ int spare=0, inqueue=0;
+
+ /* 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)) {
+ if (walk->quitting) continue;
+ inqueue= walk->sent.count + walk->priority.count
+ + walk->waiting.count;
+ spare= walk->max_queue - inqueue;
+ assert(inqueue <= max_queue_per_conn);
+ assert(spare >= 0);
+ if (inqueue==0) /*idle*/ { if (!use) use= walk; }
+ else if (spare>0) /*working*/ { use= walk; break; }
+ }
+ if (use) {
+ if (!inqueue) use->since_activity= 0; /* reset idle counter */
+ while (spare>0) {
+ Article *art= LIST_REMHEAD(queue);
+ if (!art) break;
+ LIST_ADDTAIL(use->waiting, art);
+ spare--;
+ }
+ conn_maybe_write(use);
+ } else if (allow_connect_start()) {
+ until_connect= reconnect_delay_periods;
+ connect_start();
+ break;
+ } else {
+ break;
+ }
+ }
+}
+
+static void *conn_writeable(oop_source *l, int fd, oop_event ev, void *u) {
+ conn_maybe_write(u);
+ return OOP_CONTINUE;
+}
+
+static void conn_maybe_write(Conn *conn) {
+ for (;;) {
+ conn_make_some_xmits(conn);
+ if (!conn->xmitu) {
+ loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+ return;
+ }
+
+ void *rp= conn_write_some_xmits(conn);
+ if (rp==OOP_CONTINUE) {
+ loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
+ return;
+ } else if (rp==OOP_HALT) {
+ return;
+ } else if (!rp) {
+ /* transmitted everything */
+ } else {
+ abort();
+ }
+ }
+}
+
+/*========== article transmission ==========*/