1 /*========== management of connections ==========*/
3 static void reconnect_blocking_event(void) {
4 until_connect= reconnect_delay_periods;
7 void conn_closefd(Conn *conn, const char *msgprefix) {
8 int r= close_perhaps(&conn->fd);
9 if (r) info("C%d %serror closing socket: %s",
10 conn->fd, msgprefix, strerror(errno));
13 static int conn_busy(Conn *conn) {
15 conn->waiting.count ||
16 conn->priority.count ||
21 static void conn_dispose(Conn *conn) {
24 oop_rd_cancel(conn->rd);
25 oop_rd_delete_kill(conn->rd);
29 loop->cancel_fd(loop, conn->fd, OOP_WRITE);
30 loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
32 conn_closefd(conn,"");
36 static void *conn_exception(oop_source *lp, int fd,
37 oop_event ev, void *conn_v) {
40 assert(fd == conn->fd);
41 assert(ev == OOP_EXCEPTION);
42 int r= read(conn->fd, &ch, 1);
43 if (r<0) connfail(conn,"read failed: %s",strerror(errno));
44 else connfail(conn,"exceptional condition on socket (peer sent urgent"
45 " data? read(,&ch,1)=%d,ch='\\x%02x')",r,ch);
49 static void vconnfail(Conn *conn, const char *fmt, va_list al) {
50 int requeue[art_MaxState];
51 memset(requeue,0,sizeof(requeue));
55 while ((art= LIST_REMHEAD(conn->priority)))
56 LIST_ADDTAIL(art->ipf->queue, art);
58 while ((art= LIST_REMHEAD(conn->waiting)))
59 LIST_ADDTAIL(art->ipf->queue, art);
61 while ((art= LIST_REMHEAD(conn->sent))) {
62 requeue[art->state]++;
63 if (art->state==art_Unsolicited) art->state= art_Unchecked;
64 LIST_ADDTAIL(art->ipf->queue,art);
65 check_reading_pause_resume(art->ipf);
70 for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
73 LIST_REMOVE(conns,conn);
75 char *m= xvasprintf(fmt,al);
76 warn("C%d (now %d) connection failed requeueing " RCI_TRIPLE_FMT_BASE ": %s",
77 conn->fd, conns.count, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
80 reconnect_blocking_event();
82 check_assign_articles();
85 static void connfail(Conn *conn, const char *fmt, ...) {
88 vconnfail(conn,fmt,al);
92 static void conn_idle_close(Conn *conn, const char *why) {
93 static const char quitcmd[]= "QUIT\r\n";
94 int todo= sizeof(quitcmd)-1;
95 const char *p= quitcmd;
97 int r= write(conn->fd, p, todo);
99 if (isewouldblock(errno))
100 connfail(conn, "blocked writing QUIT to idle connection");
102 connfail(conn, "failed to write QUIT to idle connection: %s",
110 conn->since_activity= 0;
111 dbg("C%d is idle (%s), quitting", conn->fd, why);
118 * For our last connection, we also shut it down if we have had
119 * less than K in the last L
121 static void check_idle_conns(void) {
124 int volthisperiod= lowvol_perperiod[lowvol_circptr];
126 lowvol_circptr %= lowvol_periods;
127 lowvol_total += volthisperiod;
128 lowvol_total -= lowvol_perperiod[lowvol_circptr];
129 lowvol_perperiod[lowvol_circptr]= 0;
132 conn->since_activity++;
136 if (conn->since_activity <= need_activity_periods) continue;
138 /* We need to shut this down */
140 connfail(conn,"timed out waiting for response to QUIT (%s)",
142 else if (conn->sent.count)
143 connfail(conn,"timed out waiting for responses");
144 else if (conn->waiting.count || conn->priority.count)
145 connfail(conn,"BUG IN INNDUCT conn has queue but nothing sent");
146 else if (conn->xmitu)
147 connfail(conn,"peer has been sending responses"
148 " before receiving our commands!");
150 conn_idle_close(conn, "no activity");
155 conn= LIST_HEAD(conns);
156 if (!volthisperiod &&
158 lowvol_total < lowvol_thresh &&
160 conn_idle_close(conn, "low volume");
163 /*---------- making new connections ----------*/
165 static pid_t connecting_child;
166 int connecting_fdpass_sock;
168 static void connect_attempt_discard(void) {
169 if (connecting_child) {
170 int status= xwaitpid(&connecting_child, "connect");
171 if (!(WIFEXITED(status) ||
172 (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)))
173 report_child_status("connect", status);
175 if (connecting_fdpass_sock) {
176 cancel_fd_read_except(connecting_fdpass_sock);
177 xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
181 #define PREP_DECL_MSG_CMSG(msg) \
183 struct iovec msgiov; \
184 msgiov.iov_base= &msgbyte; \
187 memset(&msg,0,sizeof(msg)); \
188 char msg##cbuf[CMSG_SPACE(sizeof(int))]; \
189 msg.msg_iov= &msgiov; \
191 msg.msg_control= msg##cbuf; \
192 msg.msg_controllen= sizeof(msg##cbuf);
194 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
197 assert(fd == connecting_fdpass_sock);
199 PREP_DECL_MSG_CMSG(msg);
201 ssize_t rs= recvmsg(fd, &msg, 0);
203 if (isewouldblock(errno)) return OOP_CONTINUE;
204 syswarn("failed to read socket from connecting child");
209 LIST_INIT(conn->waiting);
210 LIST_INIT(conn->priority);
211 LIST_INIT(conn->sent);
213 struct cmsghdr *h= 0;
214 if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
216 int status= xwaitpid(&connecting_child, "connect child (broken)");
218 if (WIFEXITED(status)) {
219 if (WEXITSTATUS(status) != 0 &&
220 WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
221 WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
222 /* child already reported the problem */;
224 if (e == OOP_EXCEPTION)
225 warn("connect: connection child exited code %d but"
226 " unexpected exception on fdpass socket",
227 WEXITSTATUS(status));
229 warn("connect: connection child exited code %d but"
231 WEXITSTATUS(status), (int)rs);
233 } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
234 warn("connect: connection attempt timed out");
236 report_child_status("connect", status);
241 #define CHK(field, val) \
242 if (h->cmsg_##field != val) { \
243 crash("connect: child sent cmsg with cmsg_" #field "=%d, expected %d", \
244 h->cmsg_##field, val); \
247 CHK(level, SOL_SOCKET);
248 CHK(type, SCM_RIGHTS);
249 CHK(len, CMSG_LEN(sizeof(conn->fd)));
252 if (CMSG_NXTHDR(&msg,h)) crash("connect: child sent many cmsgs");
254 memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
257 pid_t got= waitpid(connecting_child, &status, 0);
258 if (got==-1) syscrash("connect: real wait for child");
259 assert(got == connecting_child);
262 if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
263 int es= WEXITSTATUS(status);
265 case CONNCHILD_ESTATUS_STREAM: conn->stream= 1; break;
266 case CONNCHILD_ESTATUS_NOSTREAM: conn->stream= 0; break;
268 die("connect: child gave unexpected exit status %d", es);
272 conn->max_queue= conn->stream ? max_queue_per_conn : 1;
274 loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
275 conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
276 if (!conn->fd) crash("oop_rd_new_fd conn failed (fd=%d)",conn->fd);
277 int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_STRLEN,
280 if (r) syscrash("oop_rd_read for peer (fd=%d)",conn->fd);
282 LIST_ADDHEAD(conns, conn);
283 notice("C%d (now %d) connected %s",
284 conn->fd, conns.count, conn->stream ? "streaming" : "plain");
286 connect_attempt_discard();
287 check_assign_articles();
292 connect_attempt_discard();
293 reconnect_blocking_event();
297 static int allow_connect_start(void) {
298 return conns.count < max_connections
303 static void connect_start(void) {
304 assert(!connecting_child);
305 assert(!connecting_fdpass_sock);
307 info("starting connection attempt");
308 int ok_until_connect= until_connect;
309 reconnect_blocking_event();
312 int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
313 if (r) { syswarn("connect: cannot create socketpair for child"); return; }
315 connecting_child= xfork("connection");
317 if (!connecting_child) {
318 FILE *cn_from, *cn_to;
319 char buf[NNTP_STRLEN+100];
320 int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
322 xclose(socks[0], "(in child) parent's connection fdpass socket",0);
324 alarm(connection_setup_timeout);
325 if (NNTPconnect((char*)remote_host, port, &cn_from, &cn_to, buf) < 0) {
329 unsigned char c= buf[l-1];
330 if (!isspace(c)) break;
331 if (c=='\n' || c=='\r') stripped=1;
335 sysdie("connect: connection attempt failed");
338 die("connect: %s: %s", stripped ? "rejected" : "failed",
342 if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
343 sysdie("connect: authentication failed");
345 if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
347 sysdie("connect: could not send MODE STREAM");
348 buf[sizeof(buf)-1]= 0;
349 if (!fgets(buf, sizeof(buf)-1, cn_from)) {
351 sysdie("connect: could not read response to MODE STREAM");
353 die("connect: connection close in response to MODE STREAM");
358 die("connect: response to MODE STREAM is too long: %.100s...",
360 l--; if (l>0 && buf[l-1]=='\r') l--;
363 int rcode= strtoul(buf,&ep,10);
365 die("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1));
369 exitstatus= CONNCHILD_ESTATUS_STREAM;
375 warn("connect: unexpected response to MODE STREAM: %.50s",
381 int fd= fileno(cn_from);
383 PREP_DECL_MSG_CMSG(msg);
384 struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
385 cmsg->cmsg_level= SOL_SOCKET;
386 cmsg->cmsg_type= SCM_RIGHTS;
387 cmsg->cmsg_len= CMSG_LEN(sizeof(fd));
388 memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
390 msg.msg_controllen= cmsg->cmsg_len;
391 r= sendmsg(socks[1], &msg, 0);
392 if (r<0) syscrash("sendmsg failed for new connection");
393 if (r!=1) crash("sendmsg for new connection gave wrong result %d",r);
398 xclose(socks[1], "connecting fdpass child's socket",0);
399 connecting_fdpass_sock= socks[0];
400 xsetnonblock(connecting_fdpass_sock, 1);
401 on_fd_read_except(connecting_fdpass_sock, connchild_event);
404 until_connect= ok_until_connect;