enhanced sendfile for multiple os variants

lib/hio.c

@@ -468,6 +468,11 @@ static HIO_INLINE void unlink_wq (hio_t* hio, hio_wq_t* q)
 		hio_deltmrjob(hio, q->tmridx);
 		HIO_ASSERT(hio, q->tmridx == HIO_TMRIDX_INVALID);
 	}
+	/* whatever is still unwritten in this request stops being queued. for a
+	 * request that ran to completion q->len is already 0 and this is a no-op;
+	 * for one dropped on a timeout or a device kill it is the remainder. */
+	HIO_ASSERT(hio, q->dev->wq_len >= (hio_oow_t)q->len);
+	q->dev->wq_len -= q->len;
 	HIO_WQ_UNLINK (q);
 }
 
@@ -688,6 +693,7 @@ static HIO_INLINE void handle_event (hio_t* hio, hio_dev_t* dev, int events, int
 					q->off += ulen; /* advance the offset to the remaining data */
 				}
 				q->len -= ulen; /* data remining in the buffer */
+				dev->wq_len -= ulen; /* and stops counting against the queue */
 
 				if (q->len <= 0)
 				{
@@ -1622,7 +1628,7 @@ static void on_write_timeout (hio_t* hio, const hio_ntime_t* now, hio_tmrjob_t*
 	x = dev->dev_evcb->on_write(dev, -1, q->ctx, &q->dstaddr);
 
 	HIO_ASSERT(hio, q->tmridx == HIO_TMRIDX_INVALID);
-	HIO_WQ_UNLINK(q);
+	unlink_wq(hio, q); /* not a bare HIO_WQ_UNLINK - wq_len has to follow */
 	hio_freemem(hio, q);
 
 	if (x <= -1)
@@ -1738,6 +1744,7 @@ static HIO_INLINE int __enqueue_pending_write (hio_dev_t* dev, hio_iolen_t olen,
 	}
 
 	HIO_WQ_ENQ (&dev->wq, q);
+	dev->wq_len += urem;
 	if (!(dev->dev_cap & HIO_DEV_CAP_OUT_WATCHED))
 	{
 		/* if output is not being watched, arrange to do so */
@@ -1812,6 +1819,7 @@ static HIO_INLINE int __enqueue_pending_sendfile (hio_dev_t* dev, hio_iolen_t ol
 	}
 
 	HIO_WQ_ENQ (&dev->wq, q);
+	dev->wq_len += urem;
 	if (!(dev->dev_cap & HIO_DEV_CAP_OUT_WATCHED))
 	{
 		/* if output is not being watched, arrange to do so */
@@ -1841,6 +1849,14 @@ static HIO_INLINE int __dev_write (hio_dev_t* dev, const void* data, hio_iolen_t
 		return -1;
 	}
 
+	if (dev->wq_lim > 0 && dev->wq_len >= dev->wq_lim)
+	{
+		/* refuse the whole request rather than write part of it and queue the
+		 * rest, which would leave a stream with a half-delivered message. */
+		hio_seterrbfmt(hio, HIO_EBUFFULL, "write queue full - %zu bytes queued against a limit of %zu", (hio_oow_t)dev->wq_len, (hio_oow_t)dev->wq_lim);
+		return -1;
+	}
+
 	uptr = data;
 	urem = len;
 
@@ -1935,6 +1951,14 @@ static HIO_INLINE int __dev_writev (hio_dev_t* dev, hio_iovec_t* iov, hio_iolen_
 		return -1;
 	}
 
+	if (dev->wq_lim > 0 && dev->wq_len >= dev->wq_lim)
+	{
+		/* refuse the whole request rather than write part of it and queue the
+		 * rest, which would leave a stream with a half-delivered message. */
+		hio_seterrbfmt(hio, HIO_EBUFFULL, "write queue full - %zu bytes queued against a limit of %zu", (hio_oow_t)dev->wq_len, (hio_oow_t)dev->wq_lim);
+		return -1;
+	}
+
 	len = 0;
 	for (i = 0; i < iovcnt; i++) len += iov[i].iov_len;
 	urem = len;
@@ -2040,6 +2064,14 @@ static int __dev_sendfile (hio_dev_t* dev, hio_syshnd_t in_fd, hio_foff_t foff,
 		return -1;
 	}
 
+	if (dev->wq_lim > 0 && dev->wq_len >= dev->wq_lim)
+	{
+		/* refuse the whole request rather than write part of it and queue the
+		 * rest, which would leave a stream with a half-delivered message. */
+		hio_seterrbfmt(hio, HIO_EBUFFULL, "write queue full - %zu bytes queued against a limit of %zu", (hio_oow_t)dev->wq_len, (hio_oow_t)dev->wq_lim);
+		return -1;
+	}
+
 	if (HIO_UNLIKELY(!dev->dev_mth->sendfile))
 	{
 		hio_seterrbfmt(hio, HIO_ENOCAPA, "unable to senfile over unsupported device");
@@ -2120,6 +2152,21 @@ int hio_dev_write (hio_dev_t* dev, const void* data, hio_iolen_t len, void* wrct
 	return __dev_write(dev, data, len, HIO_NULL, wrctx, dstaddr);
 }
 
+hio_oow_t hio_dev_getwqsize (hio_dev_t* dev)
+{
+	return dev->wq_len;
+}
+
+void hio_dev_setwqlimit (hio_dev_t* dev, hio_oow_t limit)
+{
+	dev->wq_lim = limit;
+}
+
+hio_oow_t hio_dev_getwqlimit (hio_dev_t* dev)
+{
+	return dev->wq_lim;
+}
+
 int hio_dev_writev (hio_dev_t* dev, hio_iovec_t* iov, hio_iolen_t iovcnt, void* wrctx, const hio_devaddr_t* dstaddr)
 {
 	return __dev_writev(dev, iov, iovcnt, HIO_NULL, wrctx, dstaddr);

lib/hio.h

@@ -390,6 +390,8 @@ struct hio_wq_t
 	hio_tmridx_t    rtmridx; \
 	int             dev_extra_events; /* events the transport needs watched regardless of the device's own i/o state. see hio_dev_watch() */ \
 	hio_wq_t        wq; \
+	hio_oow_t       wq_len; /* number of bytes sitting in the write queue, not yet handed to the transport */ \
+	hio_oow_t       wq_lim; /* soft cap on wq_len. 0 means no cap. see hio_dev_setwqlimit() */ \
 	hio_oow_t       cw_count; \
 	hio_dev_t*      dev_prev; \
 	hio_dev_t*      dev_next
@@ -1130,6 +1132,43 @@ HIO_EXPORT int hio_dev_timedwrite (
  * The hio_dev_timedwritev() function is hio_dev_writev() with the deadline
  * described for hio_dev_timedwrite().
  */
+/**
+ * The hio_dev_getwqsize() function returns the number of bytes accepted by
+ * hio_dev_write() and friends that the transport has not taken yet. It counts
+ * bytes, not requests, so a caller applying backpressure can reason about
+ * memory rather than about how many times it happened to call write.
+ */
+HIO_EXPORT hio_oow_t hio_dev_getwqsize (
+	hio_dev_t*            dev
+);
+
+/**
+ * The hio_dev_setwqlimit() function sets a soft cap on the queued byte count
+ * of a single device. Once hio_dev_getwqsize() reaches the cap, every further
+ * write request on that device is refused outright with #HIO_EBUFFULL until the
+ * queue drains below it.
+ *
+ * The cap is checked before anything is written, never in the middle of a
+ * request, so a stream never ends up with a partial message on the wire and the
+ * rest rejected. That is also why it is soft: one oversized request is accepted
+ * whole and may push the queue past the cap.
+ *
+ * A limit of 0, the default, means no cap - matching the behaviour of every
+ * release before this function existed.
+ */
+HIO_EXPORT void hio_dev_setwqlimit (
+	hio_dev_t*            dev,
+	hio_oow_t             limit
+);
+
+/**
+ * The hio_dev_getwqlimit() function returns the cap set by
+ * hio_dev_setwqlimit(), or 0 if the device is uncapped.
+ */
+HIO_EXPORT hio_oow_t hio_dev_getwqlimit (
+	hio_dev_t*            dev
+);
+
 HIO_EXPORT int hio_dev_timedwritev (
 	hio_dev_t*            dev,
 	hio_iovec_t*          iov,

lib/http-pxy.c

@@ -40,7 +40,12 @@
 #define PXY_ALLOW_UNLIMITED_REQ_CONTENT_LENGTH
 
 #define PXY_PEER_CONNECT_TMOUT (5)
-#define PXY_PENDING_IO_THRESHOLD (5)
+/* backpressure is applied on queued bytes, not on the number of outstanding
+ * write requests. counting requests treats a 1MB body chunk and a 1-byte one
+ * as the same amount of pressure, which is exactly wrong for a proxy sitting
+ * between a fast upstream and a slow client. peer_pending_writes is still
+ * kept, but only to know when every write has completed. */
+#define PXY_PENDING_BYTES_THRESHOLD (256 * 1024)
 
 #define PXY_OVER_READ_FROM_CLIENT (1 << 0)
 #define PXY_OVER_READ_FROM_PEER   (1 << 1)
@@ -66,6 +71,8 @@ struct pxy_t
 	hio_becs_t* peer_buf;
 
 	unsigned int over: 4; /* must be large enough to accomodate PXY_OVER_ALL */
+	unsigned int client_read_suspended: 1; /* reading from the client is off because the peer's queue is deep */
+	unsigned int peer_read_suspended: 1;   /* reading from the peer is off because the client's queue is deep */
 	unsigned int peer_connected: 1;
 	unsigned int peer_wr_ended: 1; /* the client side finished before we connected */
 
@@ -110,10 +117,12 @@ static int pxy_write_to_peer (pxy_t* pxy, const void* data, hio_iolen_t dlen)
 			return -1;
 		}
 
-		if (pxy->peer_pending_writes > PXY_PENDING_IO_THRESHOLD)
+		if (!pxy->client_read_suspended &&
+		    hio_dev_getwqsize((hio_dev_t*)pxy->peer) > PXY_PENDING_BYTES_THRESHOLD)
 		{
 			/* suspend input watching */
 			if (pxy->task_csck && hio_dev_sck_read(pxy->task_csck, 0) <= -1) return -1;
+			pxy->client_read_suspended = 1;
 		}
 	}
 	return 0;
@@ -329,8 +338,10 @@ static int pxy_peer_on_write (hio_dev_sck_t* sck, hio_iolen_t wrlen, void* wrctx
 		HIO_ASSERT(hio, pxy->peer_pending_writes > 0);
 
 		pxy->peer_pending_writes--;
-		if (pxy->peer_pending_writes == PXY_PENDING_IO_THRESHOLD)
+		if (pxy->client_read_suspended &&
+		    hio_dev_getwqsize((hio_dev_t*)sck) <= PXY_PENDING_BYTES_THRESHOLD)
 		{
+			pxy->client_read_suspended = 0;
 			if (!(pxy->over & PXY_OVER_READ_FROM_CLIENT) &&
 			    hio_dev_sck_read(pxy->task_csck, 1) <= -1) goto oops;
 		}
@@ -406,9 +417,11 @@ static int peer_htrd_push_content (hio_htrd_t* htrd, hio_htre_t* req, const hio_
 	HIO_ASSERT(pxy->htts->hio, htrd == pxy->peer_htrd);
 
 	n = hio_svc_htts_task_addresbody((hio_svc_htts_task_t*)pxy, data, dlen);
-	if (pxy->task_res_pending_writes > PXY_PENDING_IO_THRESHOLD)
+	if (!pxy->peer_read_suspended && pxy->task_csck &&
+	    hio_dev_getwqsize((hio_dev_t*)pxy->task_csck) > PXY_PENDING_BYTES_THRESHOLD)
 	{
 		if (hio_dev_sck_read(pxy->peer, 0) <= -1) n = -1;
+		else pxy->peer_read_suspended = 1;
 	}
 
 	return n;
@@ -554,9 +567,11 @@ static int pxy_client_on_write (hio_dev_sck_t* sck, hio_iolen_t wrlen, void* wrc
 	}
 	else if (wrlen > 0)
 	{
-		if (pxy->peer && pxy->task_res_pending_writes == PXY_PENDING_IO_THRESHOLD)
+		if (pxy->peer && pxy->peer_read_suspended &&
+		    hio_dev_getwqsize((hio_dev_t*)sck) <= PXY_PENDING_BYTES_THRESHOLD)
 		{
 			/* enable input watching */
+			pxy->peer_read_suspended = 0;
 			if (!(pxy->over & PXY_OVER_READ_FROM_PEER) &&
 			    hio_dev_sck_read(pxy->peer, 1) <= -1) n = -1;
 		}

lib/sck.c

@@ -61,6 +61,36 @@
 #	include <sys/sendfile.h>
 #endif
 
+/* sendfile() is not one function. the three live flavours disagree on the
+ * operand order, on how the offset is passed, and on where the transferred
+ * byte count comes back:
+ *
+ *   linux, solaris      ssize_t sendfile(int out, int in, off_t* off, size_t n);
+ *                       socket first, offset by pointer and updated in place,
+ *                       count is the return value.
+ *
+ *   freebsd, dragonfly  int sendfile(int in, int out, off_t off, size_t n,
+ *                                    struct sf_hdtr* hdtr, off_t* sbytes, int flags);
+ *                       file first, offset by value, count through sbytes,
+ *                       return value is only 0 or -1.
+ *
+ *   darwin              int sendfile(int in, int out, off_t off, off_t* len,
+ *                                    struct sf_hdtr* hdtr, int flags);
+ *                       file first, len is in-out.
+ *
+ * the choice is made from the platform rather than from configure because
+ * AC_CHECK_FUNCS only answers whether the symbol exists, not which of these
+ * it is. this mirrors what libuv and nginx do for the same call. */
+#if defined(HAVE_SENDFILE)
+#	if defined(__FreeBSD__) || defined(__DragonFly__)
+#		define USE_SENDFILE_BSD
+#	elif defined(__APPLE__) && defined(__MACH__)
+#		define USE_SENDFILE_DARWIN
+#	else
+#		define USE_SENDFILE_LINUX
+#	endif
+#endif
+
 #if defined(HAVE_SYS_IOCTL_H)
 #	include <sys/ioctl.h>
 #endif
@@ -507,9 +537,8 @@ static int dev_sck_kill (hio_dev_t* dev, int force)
 {
 	hio_t* hio = dev->hio;
 	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
-	int hnd = rdev->hnd;
 
-	HIO_DEBUG2(hio, "SCK(%p) - being killed [%d]\n", rdev, hnd);
+	HIO_DEBUG2(hio, "SCK(%p) - being killed [%d]\n", rdev, rdev->hnd);
 #if 0
 	if (IS_STREAM(rdev))
 	{
@@ -1180,7 +1209,9 @@ static int dev_sck_sendfile_stream (hio_dev_t* dev, hio_syshnd_t in_fd, hio_foff
 	else
 	{
 #endif
+#if defined(USE_SENDFILE_LINUX)
 		ssize_t x;
+#endif
 
 		if (*len <= 0)
 		{
@@ -1198,8 +1229,10 @@ static int dev_sck_sendfile_stream (hio_dev_t* dev, hio_syshnd_t in_fd, hio_foff
 			return 1;
 		}
 
-#if defined(HAVE_SENDFILE)
+#if defined(USE_SENDFILE_LINUX)
-/* TODO: cater for other systems */
+		/* the offset is passed by pointer and updated in place, but the caller
+		 * keeps its own cursor in the write queue entry, so the update is of no
+		 * use here and foff is a local copy on purpose. */
 		x = sendfile(rdev->hnd, in_fd, &foff, *len);
 		if (x <= -1)
 		{
@@ -1210,6 +1243,57 @@ static int dev_sck_sendfile_stream (hio_dev_t* dev, hio_syshnd_t in_fd, hio_foff
 		}
 		*len = x;
 		if (x == 0) return 0; /* treat it like EWOULDBLOCK? */
+
+#elif defined(USE_SENDFILE_BSD)
+		{
+			off_t sbytes = 0;
+			int rc;
+
+			/* note the reversed operands - the file is the first argument here
+			 * and the socket the second. */
+			rc = sendfile(in_fd, rdev->hnd, (off_t)foff, (size_t)*len, HIO_NULL, &sbytes, 0);
+			if (rc <= -1)
+			{
+				if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR)
+				{
+					/* a partial transfer is reported as a failure with sbytes
+					 * set. saying 'nothing written' here would make the caller
+					 * resend bytes that already went out. */
+					if (sbytes > 0) goto bsd_sent;
+					return 0;
+				}
+				hio_seterrwithsyserr(hio, 0, errno);
+				return -1;
+			}
+
+		bsd_sent:
+			*len = (hio_iolen_t)sbytes;
+			if (sbytes <= 0) return 0; /* treat it like EWOULDBLOCK */
+		}
+
+#elif defined(USE_SENDFILE_DARWIN)
+		{
+			off_t nsent = (off_t)*len;
+			int rc;
+
+			/* nsent is in-out and is set even when the call reports failure. */
+			rc = sendfile(in_fd, rdev->hnd, (off_t)foff, &nsent, HIO_NULL, 0);
+			if (rc <= -1)
+			{
+				if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR)
+				{
+					if (nsent > 0) goto darwin_sent;
+					return 0;
+				}
+				hio_seterrwithsyserr(hio, 0, errno);
+				return -1;
+			}
+
+		darwin_sent:
+			*len = (hio_iolen_t)nsent;
+			if (nsent <= 0) return 0; /* treat it like EWOULDBLOCK */
+		}
+
 #else
 		hio_seterrnum(hio, HIO_ENOIMPL);
 		return -1;

t/Makefile.am

@@ -70,7 +70,7 @@ t_007_CFLAGS = $(CFLAGS_COMMON)
 t_007_LDFLAGS = $(LDFLAGS_COMMON)
 t_007_LDADD = $(LIBADD_COMMON)
 
-t_008_SOURCES = t-008.c tap.h
+t_008_SOURCES = t-008.c t-008-wq.inc tap.h
 t_008_CPPFLAGS = $(CPPFLAGS_COMMON)
 t_008_CFLAGS = $(CFLAGS_COMMON)
 t_008_LDFLAGS = $(LDFLAGS_COMMON)

t/Makefile.in

@@ -663,7 +663,7 @@ t_007_CPPFLAGS = $(CPPFLAGS_COMMON)
 t_007_CFLAGS = $(CFLAGS_COMMON)
 t_007_LDFLAGS = $(LDFLAGS_COMMON)
 t_007_LDADD = $(LIBADD_COMMON)
-t_008_SOURCES = t-008.c tap.h
+t_008_SOURCES = t-008.c t-008-wq.inc tap.h
 t_008_CPPFLAGS = $(CPPFLAGS_COMMON)
 t_008_CFLAGS = $(CFLAGS_COMMON)
 t_008_LDFLAGS = $(LDFLAGS_COMMON)

t/httpecho.c

@@ -28,6 +28,10 @@ struct conn_xtn_t
 	int       answered;
 };
 
+/* large enough to push a proxy's queued bytes past its backpressure
+ * threshold on loopback, where the client drains fast. */
+#define BIGBODY_LEN (8 * 1024 * 1024)
+
 static hio_t* g_hio = HIO_NULL;
 
 static void on_sigint (int sig)
@@ -58,6 +62,29 @@ static int answer (hio_dev_sck_t* sck, conn_xtn_t* cx)
 		reason = "Not Found";
 	}
 
+	/* a body big enough to back up a proxy's write queue. the pattern is
+	 * position-dependent so a truncated or shuffled relay is visible in the
+	 * checksum rather than only in the length. */
+	if (hio_find_bchars_in_bchars(reqline, i, "/big", 4, 0))
+	{
+		hio_oow_t sent = 0;
+		snprintf (head, HIO_COUNTOF(head),
+		          "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: %d\r\nConnection: close\r\n\r\n",
+		          (int)BIGBODY_LEN);
+		if (hio_dev_sck_write(sck, head, hio_count_bcstr(head), HIO_NULL, HIO_NULL) <= -1) return -1;
+
+		while (sent < BIGBODY_LEN)
+		{
+			hio_bch_t chunk[4096];
+			hio_oow_t k, want = BIGBODY_LEN - sent;
+			if (want > HIO_COUNTOF(chunk)) want = HIO_COUNTOF(chunk);
+			for (k = 0; k < want; k++) chunk[k] = (hio_bch_t)('a' + ((sent + k) % 26));
+			if (hio_dev_sck_write(sck, chunk, want, HIO_NULL, HIO_NULL) <= -1) return -1;
+			sent += want;
+		}
+		return 0;
+	}
+
 	n = snprintf(body, HIO_COUNTOF(body), "httpecho\r\n%s\r\n", reqline);
 	snprintf (head, HIO_COUNTOF(head),
 	          "HTTP/1.1 %d %s\r\nContent-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n\r\n",

t/s-002.sh

@@ -92,6 +92,26 @@ test_pxy()
 	# and the upstream's status must come back rather than being invented
 	local hc404=$(curl -s -m 10 -w '%{http_code}' -o /dev/null "http://${SRVADDR}/pxy/missing")
 	tap_ensure "$hc404" "404" "$msg - the upstream status is propagated"
+
+	# an 8MB body from a loopback upstream against a deliberately slow reader.
+	# both numbers matter. the rate limit makes the client the bottleneck,
+	# and the size has to clear the kernel socket buffer - loopback wmem
+	# tops out at 4MB here - before anything queues in user space at all.
+	# at 2MB unlimited the kernel absorbed the whole response and the
+	# suspend/resume path was never entered. with these, a suspension that
+	# never lifts shows up as a short body or a timeout.
+	local big=$(curl -s -m 60 --limit-rate 4M "http://${SRVADDR}/pxy/big" | wc -c | tr -d ' ')
+	tap_ensure "$big" "8388608" "$msg - an 8MB upstream body relays complete under backpressure"
+
+	# 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)
+	if [ -n "$want" ]; then
+		tap_ensure "$sum" "$want" "$msg - the relayed bytes are identical to the upstream's"
+	else
+		tap_skip "$msg - perl unavailable for the reference checksum"
+	fi
 }
 
 test_fcgi()

t/t-008.c

@@ -703,6 +703,11 @@ static void quiet_logging (hio_t* hio)
 	hio_setoption (hio, HIO_LOG_MASK, &mask);
 }
 
+/* the write-queue byte accounting and the soft cap. kept in its own file
+ * purely to stop this one from growing past the point of being readable -
+ * it uses the same stub device and observers as everything above. */
+#include "t-008-wq.inc"
+
 int main (void)
 {
 	hio_errinf_t errinf;
@@ -733,6 +738,10 @@ int main (void)
 	test_zero_length_closes_output ();
 	test_queued_zero_length_closes_output ();
 	test_pending_writes_dropped_on_kill ();
+	test_wq_size_accounting ();
+	test_wq_size_released_on_kill ();
+	test_wq_limit ();
+	test_wq_limit_zero_is_unlimited ();
 
 	hio_close (g_hio);
 	return exit_status();