not using perl in t/s-002.sh. using the yes command instead

lib/hio-sck.h

@@ -457,6 +457,18 @@ struct hio_dev_sck_qxmsg_t
 };
 typedef struct hio_dev_sck_qxmsg_t hio_dev_sck_qxmsg_t;
 
+/* the qx side channel is a socketpair. on most systems it is a datagram pair,
+ * where one write is one message and the reader is handed whole messages. where
+ * a datagram pair cannot carry messages reliably it is a stream pair instead,
+ * and the reader has to recover the message boundaries itself.
+ *
+ * the socket type, the writing side and the reader must agree on which of the
+ * two it is, so all three ask this one macro rather than testing for the system
+ * separately. */
+#if defined(__BEOS__) || defined(__HAIKU__)
+#	define HIO_DEV_SCK_QX_STREAM
+#endif
+
 struct hio_dev_sck_t
 {
 	HIO_DEV_HEADER;
@@ -512,8 +524,16 @@ struct hio_dev_sck_t
 		struct
 		{
 			hio_syshnd_t side_chan; /* side-channel for HIO_DEV_SCK_QX */
+		#if defined(HIO_DEV_SCK_QX_STREAM)
+			/* the message being reassembled, and how much of it has arrived.
+			 * a stream side channel splits and merges writes freely, so a read
+			 * can end in the middle of a message and the remainder has to be
+			 * held until the rest comes. only the stream variant needs it, and
+			 * the hio_skad_t inside the message makes it large, so it is not
+			 * carried on the systems that never look at it. */
 			hio_dev_sck_qxmsg_t qxacc;
 			hio_oow_t qxacc_len;
+		#endif
 		} qx;
 
 		struct

lib/sck.c

@@ -209,7 +209,11 @@ open_socket:
 	if (hio_makesyshndasync(hio, sck) <= -1 ||
 	    hio_makesyshndcloexec(hio, sck) <= -1) goto oops;
 
+/* the label is only jumped to from the branch that asks socket() for the flags
+ * directly, so it exists only where that branch does */
+#if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
 done:
+#endif
 	return sck;
 
 oops:
@@ -221,13 +225,11 @@ oops:
 static hio_syshnd_t open_async_qx (hio_t* hio, hio_syshnd_t* side_chan)
 {
 	int fd[2];
-	int type = SOCK_DGRAM;
+#if defined(HIO_DEV_SCK_QX_STREAM)
-
-#if defined(__BEOS__) || defined(__HAIKU__)
 	/* on haiku os r6beta, SOCK_DGRAM isn't reliable. more than one write() on it causes SIGPIPE */
-	type = SOCK_STREAM;
+	int type = SOCK_STREAM;
 #else
-	type = SOCK_DGRAM;
+	int type = SOCK_DGRAM;
 #endif
 
 #if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
@@ -265,7 +267,11 @@ open_socket:
 		return HIO_SYSHND_INVALID;
 	}
 
+/* the label is only jumped to from the branch that asks socketpair() for the
+ * flags directly, so it exists only where that branch does */
+#if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
 done:
+#endif
 	*side_chan = fd[1]; /* write end of the pipe */
 	return fd[0]; /* read end of the pipe */
 }
@@ -340,7 +346,7 @@ static int dev_sck_read_bpf (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_de
 {
 	hio_t* hio = dev->hio;
 	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
-	bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
+	bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
 	struct bpf_hdr* bh;
 	hio_oow_t caplen;
 
@@ -400,7 +406,7 @@ static int dev_sck_read_bpf (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_de
 static int dev_sck_readpending_bpf (hio_dev_t* dev)
 {
 	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
-	bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
+	bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
 	return st && st->pos < st->len;
 }
 
@@ -928,9 +934,9 @@ static int dev_sck_kill (hio_dev_t* dev, int force)
 	}
 
 #if defined(USE_BPF)
-	if (rdev->type == HIO_DEV_SCK_PACKET && rdev->u.bpf.state)
+	if (sck_type_map[rdev->type].domain == __AF_BPF && rdev->u.bpf.state)
 	{
-		bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
+		bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
 		if (st->buf) hio_freemem(hio, st->buf);
 		hio_freemem(hio, st);
 		rdev->u.bpf.state = HIO_NULL;
@@ -2114,7 +2120,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 			}
 
 		#if defined(USE_BPF)
-			if (rdev->bpf_state)
+			if (sck_type_map[rdev->type].domain == __AF_BPF)
 			{
 				/* a bpf device attaches to an interface by name, and what the
 				 * caller gave is the ifindex hio_skad_init_for_eth() puts in an
@@ -2314,6 +2320,11 @@ fcntl(rdev->hnd, F_SETFL, flags | O_NONBLOCK);
 				/* as i know it's connected already,
 				 * i don't schedule a connection timeout job */
 
+				/* the connection is up already, so there is no connect timeout
+				 * to schedule. the field must still be marked unset: the ssl
+				 * handshake inherits it as its deadline, and a zero there is a
+				 * deadline already in the past. set it to a negative value */
+				HIO_INIT_NTIME(&rdev->tmout, -1, 0);
 				rdev->remoteaddr = conn->remoteaddr;
 			#if defined(USE_SSL)
 				rdev->ssl_ctx = ssl_ctx;
@@ -3165,9 +3176,27 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
 
 	if (rdev->type == HIO_DEV_SCK_QX)
 	{
-	#if defined(__BEOS__) || defined(__HAIKU__)
+	#if defined(HIO_DEV_SCK_QX_STREAM)
-		const hio_uint8_t* p = (const hio_uint8_t*)data;
+		const hio_uint8_t* p;
-		hio_oow_t rem = (hio_oow_t)dlen;
+		hio_oow_t rem;
+	#else
+		hio_dev_sck_qxmsg_t* qxmsg;
+	#endif
+
+		if (dlen <= 0)
+		{
+			/* the writing end of the side channel is gone. a qx device carries
+			 * no stream capability bit, so the loop has nothing to recognise
+			 * end-of-file by and would keep reading a handle that stays
+			 * readable with nothing on it. report it as a read failure and let
+			 * the device be halted. */
+			hio_seterrbfmt(hio, HIO_EINVAL, "qx side channel closed");
+			return -1;
+		}
+
+	#if defined(HIO_DEV_SCK_QX_STREAM)
+		p = (const hio_uint8_t*)data;
+		rem = (hio_oow_t)dlen;
 
 		while (rem > 0)
 		{
@@ -3179,7 +3208,6 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
 			p += take;
 			rem -= take;
 
-
 			if (rdev->u.qx.qxacc_len < HIO_SIZEOF(rdev->u.qx.qxacc)) break; /* wait for more */
 			rdev->u.qx.qxacc_len = 0;
 
@@ -3188,21 +3216,24 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
 				if (make_accepted_client_connection(rdev, rdev->u.qx.qxacc.syshnd, &rdev->u.qx.qxacc.remoteaddr, rdev->u.qx.qxacc.scktype) <= -1)
 				{
 /*printf ("unable to accept new client connection %d\n", qxmsg->syshnd);*/
-					return (rdev->state & HIO_DEV_SCK_LENIENT)? 0: -1;
+					if (!(rdev->state & HIO_DEV_SCK_LENIENT)) return -1;
+					/* one read can carry several messages. the accumulator is
+					 * empty again, so the framing is intact and the messages
+					 * queued behind this one are still deliverable - returning
+					 * here would throw them away along with their handles. */
 				}
 			}
 			else
 			{
 				/* the stream is framed by size alone, so a bad command means
 				 * the two ends disagree about the message layout and nothing
-				 * after this point can be trusted to be a message boundary. */
+				 * after this point can be trusted to be a message boundary.
+				 * there is no resynchronising from that, so the device goes. */
 				hio_seterrbfmt(hio, HIO_EINVAL, "wrong qx command code");
-				return 0;
+				return -1;
 			}
 		}
 	#else
-		hio_dev_sck_qxmsg_t* qxmsg;
-
 		if (dlen != HIO_SIZEOF(*qxmsg))
 		{
 			hio_seterrbfmt(hio, HIO_EINVAL, "wrong qx packet size");
@@ -3981,12 +4012,61 @@ int hio_dev_sck_sendfileok (hio_dev_sck_t* dev)
 
 int hio_dev_sck_writetosidechan (hio_dev_sck_t* dev, const void* dptr, hio_oow_t dlen)
 {
-	if (dev->type == HIO_DEV_SCK_QX && write(dev->u.qx.side_chan, dptr, dlen) <= -1)
+	if (dev->type != HIO_DEV_SCK_QX)
+	{
+		errno = EINVAL;
+		return -1;
+	}
+
+#if defined(HIO_DEV_SCK_QX_STREAM)
+	/* the reader recovers message boundaries by counting octets, so a message
+	 * left half-written would shift every message behind it and there is no
+	 * recovering from that. a channel that has taken nothing yet is reported as
+	 * a failure, the way a datagram channel reports a full socket buffer; once
+	 * part of a message is out the channel is committed and the remainder is
+	 * pushed out here. the reader drains this channel from the event loop, so
+	 * the room it needs does come back. */
+	{
+		const hio_uint8_t* p = (const hio_uint8_t*)dptr;
+		hio_oow_t rem = dlen;
+
+		while (rem > 0)
+		{
+			ssize_t n = write(dev->u.qx.side_chan, p, rem);
+			if (n > 0)
+			{
+				p += n;
+				rem -= n;
+			}
+			else if (n == 0)
+			{
+				/* neither progress nor an error to wait on. it cannot happen on
+				 * a socket, but looping on it would hang the caller. */
+				errno = EIO;
+				return -1;
+			}
+			else if (errno == EINTR ||
+			         ((errno == EAGAIN || errno == EWOULDBLOCK) && rem < dlen))
+			{
+				/* retry: interrupted, or out of room part-way through a message */
+			}
+			else
+			{
+				/* this doesn't set the error information on the main socket. if you may check errno, though */
+				/* TODO: make hio_seterrbfmt() thread safe and set the error information properly. still the caller may be in the thread-unsafe context */
+				return -1;
+			}
+		}
+	}
+#else
+	if (write(dev->u.qx.side_chan, dptr, dlen) <= -1)
 	{
 		/* this doesn't set the error information on the main socket. if you may check errno, though */
 		/* TODO: make hio_seterrbfmt() thread safe and set the error information properly. still the caller may be in the thread-unsafe context */
 		return -1;
 	}
+#endif
+
 	return 0;
 }
 

lib/sys-mux.c

@@ -646,9 +646,9 @@ int hio_sys_ctrlmux (hio_t* hio, hio_sys_mux_cmd_t cmd, hio_dev_t* dev, int dev_
 			if (x >= 0)
 			{
 				if (i_flag == EV_DISABLE && o_flag == EV_DISABLE)
-					dev->dev_cap &= ~HIO_DEV_CAP_WATCH_SUSPENDED;
-				else
 					dev->dev_cap |= HIO_DEV_CAP_WATCH_SUSPENDED;
+				else
+					dev->dev_cap &= ~HIO_DEV_CAP_WATCH_SUSPENDED;
 			}
 			break;
 		}
@@ -876,6 +876,7 @@ int hio_sys_waitmux (hio_t* hio, const hio_ntime_t* tmout, hio_sys_mux_evtcb_t e
 	for (i = 0; i < nentries; i++)
 	{
 		int events = 0;
+		int rdhup = 0;
 		hio_dev_t* dev;
 
 		dev = mux->revs[i].udata;
@@ -887,7 +888,11 @@ int hio_sys_waitmux (hio_t* hio, const hio_ntime_t* tmout, hio_sys_mux_evtcb_t e
 			if (mux->revs[i].ident != dev->dev_mth->getsyshnd(dev)) continue; /* already closed or something mid-loop? */
 
 			if (mux->revs[i].flags & EV_ERROR) events |= HIO_DEV_EVENT_ERR;
-			if (mux->revs[i].flags & EV_EOF) events |= HIO_DEV_EVENT_HUP;
+			if (mux->revs[i].flags & EV_EOF)
+			{
+				if (mux->revs[i].filter == EVFILT_READ) rdhup = 1;
+				else events |= HIO_DEV_EVENT_HUP;
+			}
 			if (mux->revs[i].filter == EVFILT_READ) events |= HIO_DEV_EVENT_IN;
 			else if (mux->revs[i].filter == EVFILT_WRITE) events |= HIO_DEV_EVENT_OUT;
 		#else
@@ -903,13 +908,17 @@ int hio_sys_waitmux (hio_t* hio, const hio_ntime_t* tmout, hio_sys_mux_evtcb_t e
 			{
 				if (mux->revs[j].udata != dev) continue;
 				if (mux->revs[j].flags & EV_ERROR) events |= HIO_DEV_EVENT_ERR;
-				if (mux->revs[j].flags & EV_EOF) events |= HIO_DEV_EVENT_HUP;
+				if (mux->revs[j].flags & EV_EOF)
+				{
+					if (mux->revs[j].filter == EVFILT_READ) rdhup = 1;
+					else events |= HIO_DEV_EVENT_HUP;
+				}
 				if (mux->revs[j].filter == EVFILT_READ) events |= HIO_DEV_EVENT_IN;
 				else if (mux->revs[j].filter == EVFILT_WRITE) events |= HIO_DEV_EVENT_OUT;
 			}
 		#endif
 
-			if (HIO_LIKELY(events)) event_handler(hio, dev, events, 0);
+			if (HIO_LIKELY(events)) event_handler(hio, dev, events, rdhup);
 		}
 		else if (mux->ctrlp[0] != HIO_SYSHND_INVALID)
 		{

t/s-002.sh

@@ -121,7 +121,8 @@ test_pxy()
 	# and the bytes themselves, not just the count. the upstream emits a
 	# repeating a-z pattern keyed to the offset.
 	local sum=$(curl -s -m 60 "http://${SRVADDR}/pxy/big" | cksum | cut -d' ' -f1)
-	local want=$(perl -e 'print map { chr(97 + ($_ % 26)) } 0 .. (8*1024*1024 - 1)' 2>/dev/null | cksum | cut -d' ' -f1)
+	##local want=$(perl -e 'print map { chr(97 + ($_ % 26)) } 0 .. (8*1024*1024 - 1)' 2>/dev/null | cksum | cut -d' ' -f1)
+	local want=$(yes abcdefghijklmnopqrstuvwxyz | tr -d '\n' | head -c 8388608 | cksum | cut -d' ' -f1)
 	if [ -n "$want" ]; then
 		tap_ensure "$sum" "$want" "$msg - the relayed bytes are identical to the upstream's"
 	else