added association id for sctp.

enhanced the SEQPACKET support for sctp

Makefile.in

@@ -165,8 +165,8 @@ am__DIST_COMMON = $(srcdir)/Dockerfile.in $(srcdir)/Makefile.in \
 	$(top_srcdir)/ac/install-sh $(top_srcdir)/ac/ltmain.sh \
 	$(top_srcdir)/ac/missing $(top_srcdir)/ac/tap-driver.sh \
 	$(top_srcdir)/pkgs/hio.spec.in ac/ar-lib ac/compile \
-	ac/config.guess ac/config.sub ac/install-sh ac/ltmain.sh \
-	ac/missing
+	ac/config.guess ac/config.sub ac/depcomp ac/install-sh \
+	ac/ltmain.sh ac/missing
 DISTFILES = $(DIST_COMMON) $(DIST_SOURCES) $(TEXINFOS) $(EXTRA_DIST)
 distdir = $(PACKAGE)-$(VERSION)
 top_distdir = $(distdir)
@@ -294,6 +294,7 @@ PTHREAD_CXX = @PTHREAD_CXX@
 PTHREAD_LIBS = @PTHREAD_LIBS@
 QUADMATH_LIBS = @QUADMATH_LIBS@
 RANLIB = @RANLIB@
+SCTP_ENABLED = @SCTP_ENABLED@
 SED = @SED@
 SENDFILE_LIBS = @SENDFILE_LIBS@
 SET_MAKE = @SET_MAKE@

bin/Makefile.in

@@ -328,6 +328,7 @@ PTHREAD_CXX = @PTHREAD_CXX@
 PTHREAD_LIBS = @PTHREAD_LIBS@
 QUADMATH_LIBS = @QUADMATH_LIBS@
 RANLIB = @RANLIB@
+SCTP_ENABLED = @SCTP_ENABLED@
 SED = @SED@
 SENDFILE_LIBS = @SENDFILE_LIBS@
 SET_MAKE = @SET_MAKE@

configure

@@ -689,6 +689,7 @@ ax_pthread_config
 CPP
 SENDFILE_LIBS
 SOCKET_LIBS
+SCTP_ENABLED
 LIBM
 MACOSX_FALSE
 MACOSX_TRUE
@@ -830,6 +831,7 @@ with_gnu_ld
 with_sysroot
 enable_libtool_lock
 enable_largefile
+enable_sctp
 enable_all_static
 with_all_static_libs
 enable_quadmath
@@ -1496,6 +1498,8 @@ Optional Features:
                           provide on AIX, [default=aix].
   --disable-libtool-lock  avoid locking (might break parallel builds)
   --disable-largefile     omit support for large files
+  --enable-sctp           build with sctp support if the system provides it
+                          (default. yes)
   --enable-all-static     build the full static binaries(default. no)
   --enable-quadmath       attempt to support 128-bit floating point number if
                           available(default. no)
@@ -1802,6 +1806,60 @@ printf "%s\n" "$ac_res" >&6; }
 
 } # ac_fn_c_check_func
 
+# ac_fn_check_decl LINENO SYMBOL VAR INCLUDES EXTRA-OPTIONS FLAG-VAR
+# ------------------------------------------------------------------
+# Tests whether SYMBOL is declared in INCLUDES, setting cache variable VAR
+# accordingly. Pass EXTRA-OPTIONS to the compiler, using FLAG-VAR.
+ac_fn_check_decl ()
+{
+  as_lineno=${as_lineno-"$1"} as_lineno_stack=as_lineno_stack=$as_lineno_stack
+  as_decl_name=`echo $2|sed 's/ *(.*//'`
+  { printf "%s\n" "$as_me:${as_lineno-$LINENO}: checking whether $as_decl_name is declared" >&5
+printf %s "checking whether $as_decl_name is declared... " >&6; }
+if eval test \${$3+y}
+then :
+  printf %s "(cached) " >&6
+else case e in #(
+  e) as_decl_use=`echo $2|sed -e 's/(/((/' -e 's/)/) 0&/' -e 's/,/) 0& (/g'`
+  eval ac_save_FLAGS=\$$6
+  as_fn_append $6 " $5"
+  cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h.  */
+$4
+int
+main (void)
+{
+#ifndef $as_decl_name
+#ifdef __cplusplus
+  (void) $as_decl_use;
+#else
+  (void) $as_decl_name;
+#endif
+#endif
+
+  ;
+  return 0;
+}
+_ACEOF
+if ac_fn_c_try_compile "$LINENO"
+then :
+  eval "$3=yes"
+else case e in #(
+  e) eval "$3=no" ;;
+esac
+fi
+rm -f core conftest.err conftest.$ac_objext conftest.beam conftest.$ac_ext
+  eval $6=\$ac_save_FLAGS
+ ;;
+esac
+fi
+eval ac_res=\$$3
+	       { printf "%s\n" "$as_me:${as_lineno-$LINENO}: result: $ac_res" >&5
+printf "%s\n" "$ac_res" >&6; }
+  eval $as_lineno_stack; ${as_lineno_stack:+:} unset as_lineno
+
+} # ac_fn_check_decl
+
 # ac_fn_c_try_cpp LINENO
 # ----------------------
 # Try to preprocess conftest.$ac_ext, and return whether this succeeded.
@@ -16117,13 +16175,131 @@ then :
   printf "%s\n" "#define HAVE_NETINET_IN_H 1" >>confdefs.h
 
 fi
-ac_fn_c_check_header_compile "$LINENO" "netinet/sctp.h" "ac_cv_header_netinet_sctp_h" "$ac_includes_default"
+
+
+# Check whether --enable-sctp was given.
+if test ${enable_sctp+y}
+then :
+  enableval=$enable_sctp; enable_sctp_is=$enableval
+else case e in #(
+  e) enable_sctp_is=yes
+ ;;
+esac
+fi
+
+SCTP_ENABLED="no"
+if test "x$enable_sctp_is" = "xyes"
+then
+	ac_fn_c_check_header_compile "$LINENO" "netinet/sctp.h" "ac_cv_header_netinet_sctp_h" "$ac_includes_default"
 if test "x$ac_cv_header_netinet_sctp_h" = xyes
 then :
   printf "%s\n" "#define HAVE_NETINET_SCTP_H 1" >>confdefs.h
 
 fi
 
+	if test "x$ac_cv_header_netinet_sctp_h" = "xyes"
+	then
+				{ printf "%s\n" "$as_me:${as_lineno-$LINENO}: checking for $CC options needed to detect all undeclared functions" >&5
+printf %s "checking for $CC options needed to detect all undeclared functions... " >&6; }
+if test ${ac_cv_c_undeclared_builtin_options+y}
+then :
+  printf %s "(cached) " >&6
+else case e in #(
+  e) ac_save_CFLAGS=$CFLAGS
+   ac_cv_c_undeclared_builtin_options='cannot detect'
+   for ac_arg in '' -fno-builtin; do
+     CFLAGS="$ac_save_CFLAGS $ac_arg"
+     # This test program should *not* compile successfully.
+     cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h.  */
+
+int
+main (void)
+{
+(void) strchr;
+  ;
+  return 0;
+}
+_ACEOF
+if ac_fn_c_try_compile "$LINENO"
+then :
+
+else case e in #(
+  e) # This test program should compile successfully.
+        # No library function is consistently available on
+        # freestanding implementations, so test against a dummy
+        # declaration.  Include always-available headers on the
+        # off chance that they somehow elicit warnings.
+        cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h.  */
+#include <float.h>
+#include <limits.h>
+#include <stdarg.h>
+#include <stddef.h>
+extern void ac_decl (int, char *);
+
+int
+main (void)
+{
+(void) ac_decl (0, (char *) 0);
+  (void) ac_decl;
+
+  ;
+  return 0;
+}
+_ACEOF
+if ac_fn_c_try_compile "$LINENO"
+then :
+  if test x"$ac_arg" = x
+then :
+  ac_cv_c_undeclared_builtin_options='none needed'
+else case e in #(
+  e) ac_cv_c_undeclared_builtin_options=$ac_arg ;;
+esac
+fi
+          break
+fi
+rm -f core conftest.err conftest.$ac_objext conftest.beam conftest.$ac_ext ;;
+esac
+fi
+rm -f core conftest.err conftest.$ac_objext conftest.beam conftest.$ac_ext
+    done
+    CFLAGS=$ac_save_CFLAGS
+   ;;
+esac
+fi
+{ printf "%s\n" "$as_me:${as_lineno-$LINENO}: result: $ac_cv_c_undeclared_builtin_options" >&5
+printf "%s\n" "$ac_cv_c_undeclared_builtin_options" >&6; }
+  case $ac_cv_c_undeclared_builtin_options in #(
+  'cannot detect') :
+    { { printf "%s\n" "$as_me:${as_lineno-$LINENO}: error: in '$ac_pwd':" >&5
+printf "%s\n" "$as_me: error: in '$ac_pwd':" >&2;}
+as_fn_error $? "cannot make $CC report undeclared builtins
+See 'config.log' for more details" "$LINENO" 5; } ;; #(
+  'none needed') :
+    ac_c_undeclared_builtin_options='' ;; #(
+  *) :
+    ac_c_undeclared_builtin_options=$ac_cv_c_undeclared_builtin_options ;;
+esac
+
+ac_fn_check_decl "$LINENO" "IPPROTO_SCTP" "ac_cv_have_decl_IPPROTO_SCTP" "
+			#include <sys/types.h>
+			#include <sys/socket.h>
+			#include <netinet/in.h>
+
+" "$ac_c_undeclared_builtin_options" "CFLAGS"
+if test "x$ac_cv_have_decl_IPPROTO_SCTP" = xyes
+then :
+
+
+printf "%s\n" "#define HIO_ENABLE_SCTP 1" >>confdefs.h
+
+			SCTP_ENABLED="yes"
+
+fi
+	fi
+fi
+
 ac_fn_c_check_header_compile "$LINENO" "net/if.h" "ac_cv_header_net_if_h" "
 	#include <sys/types.h>
 	#include <sys/socket.h>
@@ -20787,7 +20963,9 @@ else case e in #(
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
-		}
+			unsigned int ppid;
+			int assoc;
+		};
 		struct sockaddr_in_x {
 			struct sockaddr_in a;
 			struct sockaddr_extra_t u;
@@ -20834,6 +21012,8 @@ else case e in #(
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in6_x {
 			struct sockaddr_in6 a;
@@ -21267,6 +21447,8 @@ else case e in #(
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in_x {
 			struct sockaddr_in a;
@@ -21315,6 +21497,8 @@ else case e in #(
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in6_x {
 			struct sockaddr_in6 a;
@@ -24719,6 +24903,7 @@ echo "  Math library: ${LIBM}"
 echo "  Socket library: ${SOCKET_LIBS}"
 echo "  Sendfile library: ${SENDFILE_LIBS}"
 echo "  SSL library: ${SSL_LIBS}"
+echo "  SCTP support: ${SCTP_ENABLED}"
 echo "-------------------------------------------------------------------------"
 
 

configure.ac

@@ -116,7 +116,34 @@ AC_CHECK_HEADERS([time.h sys/time.h utime.h spawn.h execinfo.h ucontext.h])
 AC_CHECK_HEADERS([sys/resource.h sys/wait.h sys/syscall.h sys/ioctl.h])
 AC_CHECK_HEADERS([sys/sendfile.h sys/epoll.h sys/event.h sys/poll.h sys/select.h])
 AC_CHECK_HEADERS([sys/sysctl.h sys/socket.h sys/sockio.h sys/un.h])
-AC_CHECK_HEADERS([ifaddrs.h tiuser.h linux/netfilter_ipv4.h netinet/in.h netinet/sctp.h])
+AC_CHECK_HEADERS([ifaddrs.h tiuser.h linux/netfilter_ipv4.h netinet/in.h])
+
+dnl sctp support. it used to be switched on purely by whether netinet/sctp.h
+dnl happened to be installed, which meant two machines could build libraries
+dnl with different capabilities and nothing said so. make it a decision.
+AC_ARG_ENABLE([sctp],
+	[AS_HELP_STRING([--enable-sctp],[build with sctp support if the system provides it (default. yes)])],
+	enable_sctp_is=$enableval,
+	enable_sctp_is=yes
+)
+SCTP_ENABLED="no"
+if test "x$enable_sctp_is" = "xyes"
+then
+	AC_CHECK_HEADERS([netinet/sctp.h])
+	if test "x$ac_cv_header_netinet_sctp_h" = "xyes"
+	then
+		dnl the header alone is not enough - IPPROTO_SCTP has to be defined too
+		AC_CHECK_DECL([IPPROTO_SCTP], [
+			AC_DEFINE([HIO_ENABLE_SCTP], [1], [enable sctp support])
+			SCTP_ENABLED="yes"
+		], [], [[
+			#include <sys/types.h>
+			#include <sys/socket.h>
+			#include <netinet/in.h>
+		]])
+	fi
+fi
+AC_SUBST(SCTP_ENABLED)
 AC_CHECK_HEADERS([net/if.h net/if_dl.h netinet/if_ether.h netpacket/packet.h net/bpf.h], [], [], [
 	#include <sys/types.h>
 	#include <sys/socket.h>])
@@ -497,7 +524,9 @@ then
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
-		}
+			unsigned int ppid;
+			int assoc;
+		};
 		struct sockaddr_in_x {
 			struct sockaddr_in a;
 			struct sockaddr_extra_t u;
@@ -509,6 +538,8 @@ then
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in6_x {
 			struct sockaddr_in6 a;
@@ -575,6 +606,8 @@ else
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in_x {
 			struct sockaddr_in a;
@@ -588,6 +621,8 @@ else
 		struct sockaddr_extra_t
 		{
 			unsigned short chan;
+			unsigned int ppid;
+			int assoc;
 		};
 		struct sockaddr_in6_x {
 			struct sockaddr_in6 a;
@@ -911,5 +946,6 @@ echo "  Math library: ${LIBM}"
 echo "  Socket library: ${SOCKET_LIBS}"
 echo "  Sendfile library: ${SENDFILE_LIBS}"
 echo "  SSL library: ${SSL_LIBS}"
+echo "  SCTP support: ${SCTP_ENABLED}"
 echo "-------------------------------------------------------------------------"
 ]

lib/Makefile.in

@@ -373,6 +373,7 @@ PTHREAD_CXX = @PTHREAD_CXX@
 PTHREAD_LIBS = @PTHREAD_LIBS@
 QUADMATH_LIBS = @QUADMATH_LIBS@
 RANLIB = @RANLIB@
+SCTP_ENABLED = @SCTP_ENABLED@
 SED = @SED@
 SENDFILE_LIBS = @SENDFILE_LIBS@
 SET_MAKE = @SET_MAKE@

lib/dhcp-svr.c

@@ -29,7 +29,6 @@
 struct hio_svc_dhcs_t
 {
 	HIO_SVC_HEADER;
-
 	int stopping;
 	hio_dev_sck_t* sck;
 };
@@ -79,9 +78,9 @@ static hio_dev_sck_t* open_socket (hio_t* hio, hio_svc_dhcs_cfg_t* cfg)
 
 /*
 	#if defined(IPV6_RECVPKTINFO)
-		hio_dev_sck_setsockopt (dhcs->sck, IPPROTO_IPV6, IPV6_RECVPKTINFO, &v);
+		hio_dev_sck_setsockopt(dhcs->sck, IPPROTO_IPV6, IPV6_RECVPKTINFO, &v);
 	#elif defined(IPV6_PKTINFO)
-		hio_dev_sck_setsockopt (dhcs->sck, IPPROTO_IPV6, IPV6_PKTINFO, &v);
+		hio_dev_sck_setsockopt(dhcs->sck, IPPROTO_IPV6, IPV6_PKTINFO, &v);
 	#else
 	//#	error no ipv6 pktinfo
 	#endif
@@ -104,7 +103,7 @@ static hio_dev_sck_t* open_socket (hio_t* hio, hio_svc_dhcs_cfg_t* cfg)
 	return sck;
 
 oops:
-	if (sck) hio_dev_sck_kill (sck);
+	if (sck) hio_dev_sck_kill(sck);
 	return HIO_NULL;
 }
 
@@ -137,13 +136,13 @@ hio_svc_dhcs_t* hio_svc_dhcs_start (hio_t* hio, const hio_skad_t* local_binds, h
 /* TODO: remember this in dhcs... */
 	}
 
-	HIO_SVCL_APPEND_SVC (&hio->actsvc, (hio_svc_t*)dhcs);
+	HIO_SVCL_APPEND_SVC(&hio->actsvc, (hio_svc_t*)dhcs);
 	return dhcs;
 
 oops:
 	if (dhcs)
 	{
-		if (sck) hio_dev_sck_kill (sck);
+		if (sck) hio_dev_sck_kill(sck);
 /*TODO:
 		for (i = 0; i < local_nbinds; i++)
 		{
@@ -158,11 +157,11 @@ void hio_svc_dhcs_stop (hio_svc_dhcs_t* dhcs)
 {
 	hio_t* hio = dhcs->hio;
 
-	HIO_DEBUG1 (hio, "FCGIC - STOPPING SERVICE %p\n", dhcs);
+	HIO_DEBUG1(hio, "FCGIC - STOPPING SERVICE %p\n", dhcs);
 	dhcs->stopping = 1;
 
-	HIO_SVCL_UNLINK_SVC (dhcs);
+	HIO_SVCL_UNLINK_SVC(dhcs);
 	hio_freemem(hio, dhcs);
 
-	HIO_DEBUG1 (hio, "FCGIC - STOPPED SERVICE %p\n", dhcs);
+	HIO_DEBUG1(hio, "FCGIC - STOPPED SERVICE %p\n", dhcs);
 }

lib/hio-cfg.h.in

@@ -591,6 +591,9 @@
 /* build the mariadb device */
 #undef HIO_ENABLE_MARIADB
 
+/* enable sctp support */
+#undef HIO_ENABLE_SCTP
+
 /* Use the wide-character type as the default character type */
 #undef HIO_ENABLE_WIDE_CHAR
 

lib/hio-sck.h

@@ -251,6 +251,23 @@ typedef void (*hio_dev_sck_on_connect_t) (
 	hio_dev_sck_t* dev
 );
 
+/**
+ * Called for an SCTP notification - an association coming up or going down, a
+ * path changing state, a send failing. These arrive interleaved with data on
+ * the same socket and are flagged MSG_NOTIFICATION; they are not payload, so
+ * they are never handed to on_read().
+ *
+ * 'data' points at a union sctp_notification of 'dlen' octets. It belongs to
+ * the loop's read buffer and is only valid for the duration of this call.
+ *
+ * Leave it unset to have notifications discarded.
+ */
+typedef void (*hio_dev_sck_on_notification_t) (
+	hio_dev_sck_t*       dev,
+	const void*          data,
+	hio_iolen_t          dlen
+);
+
 typedef void (*hio_dev_sck_on_raw_accept_t) (
 	hio_dev_sck_t* dev,
 	hio_syshnd_t   syshnd,
@@ -321,6 +338,14 @@ struct hio_dev_sck_make_t
 	hio_dev_sck_on_connect_t on_connect;
 	hio_dev_sck_on_disconnect_t on_disconnect;
 	hio_dev_sck_on_raw_accept_t on_raw_accept; /* optional */
+	hio_dev_sck_on_notification_t on_notification; /* optional. sctp only */
+
+	/** number of outbound streams to ask for on an SCTP association.
+	 *  0 leaves it to the system default, which is what every release before
+	 *  this field did. only meaningful for the SCTP types. */
+	hio_uint16_t sctp_ostreams;
+	/** maximum number of inbound streams to accept. 0 for the default. */
+	hio_uint16_t sctp_instreams;
 };
 
 enum hio_dev_sck_bind_option_t
@@ -414,6 +439,7 @@ struct hio_dev_sck_t
 	hio_dev_sck_on_connect_t on_connect;
 	hio_dev_sck_on_disconnect_t on_disconnect;
 	hio_dev_sck_on_raw_accept_t on_raw_accept;
+	hio_dev_sck_on_notification_t on_notification;
 
 	/* timer job index for handling
 	 *  - connect() timeout for a connecting socket.
@@ -429,6 +455,10 @@ struct hio_dev_sck_t
 	void* ssl;
 
 	hio_syshnd_t side_chan; /* side-channel for HIO_DEV_SCK_QX */
+
+	/* set while a message too large for the read buffer is being thrown away.
+	 * see the MSG_EOR handling in the sctp seqpacket read method. */
+	int sctp_discarding;
 };
 
 enum hio_dev_sck_shutdown_how_t

lib/hio-skad.h

@@ -222,7 +222,40 @@ HIO_EXPORT void hio_skad_set_scope_id (
 	int               scope_id
 );
 
+/**
+ * The SCTP payload protocol identifier carried alongside the address. SCTP
+ * passes it through unchanged; it means whatever the two ends agree it means.
+ * Rides in the same extra area as the channel - see hio_skad_get_chan().
+ */
 /* for sctp */
+/**
+ * The SCTP association identifier carried alongside the address. It is the
+ * stable name for an association - the addresses of one can change under
+ * multi-homing, this does not - so a reply is best addressed by it rather than
+ * by the address it happened to arrive from.
+ *
+ * On a one-to-many socket, setting it on the destination directs the message
+ * to that association; a value of 0 falls back to addressing by address.
+ */
+HIO_EXPORT hio_int32_t hio_skad_get_assoc (
+	const hio_skad_t* skad
+);
+
+HIO_EXPORT void hio_skad_set_assoc (
+	hio_skad_t* skad,
+	hio_int32_t assoc
+);
+
+HIO_EXPORT hio_uint32_t hio_skad_get_ppid (
+	const hio_skad_t* skad
+);
+
+/* for sctp */
+HIO_EXPORT void hio_skad_set_ppid (
+	hio_skad_t*  skad,
+	hio_uint32_t ppid
+);
+
 HIO_EXPORT hio_uint16_t hio_skad_get_chan (
 	const hio_skad_t* skad
 );

lib/hio.c

@@ -32,6 +32,7 @@ static void clear_unneeded_cfmbs (hio_t* hio);
 static int schedule_kill_zombie_job (hio_dev_t* dev);
 static int kill_and_free_device (hio_dev_t* dev, int force);
 static void free_dead_devices (hio_t* hio);
+static void unlink_cwq (hio_t* hio, hio_cwq_t* cwq);
 
 static void on_read_timeout (hio_t* hio, const hio_ntime_t* now, hio_tmrjob_t* job);
 static void on_write_timeout (hio_t* hio, const hio_ntime_t* now, hio_tmrjob_t* job);
@@ -135,6 +136,11 @@ int hio_init (hio_t* hio, hio_mmgr_t* mmgr, hio_cmgr_t* cmgr, hio_bitmask_t feat
 
 	hio->tmr.capa = tmrcapa;
 
+	/* the shared read buffer. resizable later through HIO_READ_BUFFER_SIZE. */
+	hio->bigbuf.ptr = hio_allocmem(hio, HIO_DFL_READ_BUFFER_SIZE);
+	if (HIO_UNLIKELY(!hio->bigbuf.ptr)) goto oops;
+	hio->bigbuf.capa = HIO_DFL_READ_BUFFER_SIZE;
+
 	HIO_CFMBL_INIT(&hio->cfmb);
 	HIO_DEVL_INIT(&hio->actdev);
 	HIO_DEVL_INIT(&hio->hltdev);
@@ -147,6 +153,13 @@ int hio_init (hio_t* hio, hio_mmgr_t* mmgr, hio_cmgr_t* cmgr, hio_bitmask_t feat
 	return 0;
 
 oops:
+	if (hio->bigbuf.ptr)
+	{
+		hio_freemem(hio, hio->bigbuf.ptr);
+		hio->bigbuf.ptr = HIO_NULL;
+		hio->bigbuf.capa = 0;
+	}
+
 	if (hio->tmr.jobs) hio_freemem(hio, hio->tmr.jobs);
 
 	if (sys_inited) hio_sys_fini(hio);
@@ -165,6 +178,13 @@ void hio_fini (hio_t* hio)
 
 	hio->_fini_in_progress = 1;
 
+	if (hio->bigbuf.ptr)
+	{
+		hio_freemem(hio, hio->bigbuf.ptr);
+		hio->bigbuf.ptr = HIO_NULL;
+		hio->bigbuf.capa = 0;
+	}
+
 	/* clean up free cwq list */
 	for (i = 0; i < HIO_COUNTOF(hio->cwqfl); i++)
 	{
@@ -176,12 +196,14 @@ void hio_fini (hio_t* hio)
 		}
 	}
 
-	/* clean up unfired cwq entries - calling fire_cwq_handlers() might not be good here. */
+	/* clean up unfired cwq entries - calling fire_cwq_handlers() might not be good here.
+	 * unlink_cwq() rather than a bare HIO_CWQ_UNLINK: the devices are killed further
+	 * down and hio_dev_kill() consults cwq_head, which must not be left dangling. */
 	while (!HIO_CWQ_IS_EMPTY(&hio->cwq))
 	{
 		hio_cwq_t* cwq;
 		cwq = HIO_CWQ_HEAD(&hio->cwq);
-		HIO_CWQ_UNLINK (cwq);
+		unlink_cwq(hio, cwq);
 		hio_freemem(hio, cwq);
 	}
 
@@ -396,6 +418,49 @@ int hio_setoption (hio_t* hio, hio_option_t id, const void* value)
 			hio->option.log_writer = (hio_log_writer_t)value;
 			break;
 
+		case HIO_READ_BUFFER_SIZE:
+		{
+			hio_oow_t newcapa = *(const hio_oow_t*)value;
+			hio_uint8_t* p;
+
+			if (newcapa < HIO_MIN_READ_BUFFER_SIZE) goto einval;
+			if (newcapa == hio->bigbuf.capa) break; /* nothing to do */
+
+			if (newcapa < hio->bigbuf.capa)
+			{
+				/* shrinking. a device already running may need more than the
+				 * new size, and it has no way to object after the fact. */
+				hio_dev_t* d;
+				for (d = HIO_DEVL_FIRST_DEV(&hio->actdev); !HIO_DEVL_IS_NIL_DEV(&hio->actdev, d); d = d->dev_next)
+				{
+					if (d->dev_rdmin > newcapa)
+					{
+						hio_seterrbfmt(hio, HIO_EPERM, "a running device requires a read buffer of at least %zu octets", d->dev_rdmin);
+						return -1;
+					}
+				}
+			}
+
+			/* on_read() is handed a pointer into this buffer, so it must not
+			 * move while a callback might be holding one. mux_depth is
+			 * non-zero for exactly the window in which that is possible. */
+			if (hio->mux_depth > 0)
+			{
+				hio_seterrbfmt(hio, HIO_EBUSY, "unable to resize the read buffer while dispatching events");
+				return -1;
+			}
+
+			/* the contents are scratch, so there is nothing to carry over.
+			 * allocate first all the same - a failure must leave the existing
+			 * buffer in place rather than leave the loop without one. */
+			p = hio_allocmem(hio, newcapa);
+			if (HIO_UNLIKELY(!p)) return -1;
+			hio_freemem(hio, hio->bigbuf.ptr);
+			hio->bigbuf.ptr = p;
+			hio->bigbuf.capa = newcapa;
+			break;
+		}
+
 		default:
 			goto einval;
 	}
@@ -444,6 +509,10 @@ int hio_getoption (hio_t* hio, hio_option_t id, void* value)
 		case HIO_LOG_WRITER:
 			*(hio_log_writer_t*)value = hio->option.log_writer;
 			return 0;
+
+		case HIO_READ_BUFFER_SIZE:
+			*(hio_oow_t*)value = hio->bigbuf.capa;
+			return 0;
 	};
 
 	hio_seterrnum(hio, HIO_EINVAL);
@@ -473,31 +542,30 @@ static HIO_INLINE void unlink_wq (hio_t* hio, hio_wq_t* q)
 	 * 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);
+	HIO_WQ_UNLINK(q);
 }
 
-static void fire_cwq_handlers (hio_t* hio)
+/* take an entry off both queues it belongs to and off its device's count.
+ * an entry is only ever removed from the head of its device's queue - both
+ * firing paths consume in order - so a singly-linked device queue suffices. */
+static void unlink_cwq (hio_t* hio, hio_cwq_t* cwq)
 {
-	/* execute callbacks for completed write operations */
-	while (!HIO_CWQ_IS_EMPTY(&hio->cwq))
-	{
-		hio_cwq_t* cwq;
-		hio_oow_t cwqfl_index;
-		hio_dev_t* dev_to_halt;
+	hio_dev_t* dev = cwq->dev;
 
-		cwq = HIO_CWQ_HEAD(&hio->cwq);
-		if (cwq->dev->dev_evcb->on_write(cwq->dev, cwq->olen, cwq->ctx, &cwq->dstaddr) <= -1)
-		{
-			dev_to_halt = cwq->dev;
-		}
-		else
-		{
-			dev_to_halt = HIO_NULL;
-		}
-		cwq->dev->cw_count--;
-		HIO_CWQ_UNLINK (cwq);
+	HIO_ASSERT(hio, dev->cwq_head == cwq);
+	dev->cwq_head = cwq->d_next;
+	if (!dev->cwq_head) dev->cwq_tail = HIO_NULL;
 
-		cwqfl_index = HIO_ALIGN_POW2(cwq->dstaddr.len, HIO_CWQFL_ALIGN) / HIO_CWQFL_SIZE;
+	HIO_ASSERT(hio, dev->cw_count > 0);
+	dev->cw_count--;
+
+	HIO_CWQ_UNLINK(cwq);
+}
+
+/* hand the entry back to the size-bucketed free list, or release it */
+static HIO_INLINE void recycle_cwq (hio_t* hio, hio_cwq_t* cwq)
+{
+	hio_oow_t cwqfl_index = HIO_ALIGN_POW2(cwq->dstaddr.len, HIO_CWQFL_ALIGN) / HIO_CWQFL_SIZE;
 	if (cwqfl_index < HIO_COUNTOF(hio->cwqfl))
 	{
 		/* reuse the cwq object if dstaddr is 0 in size. chain it to the free list */
@@ -509,7 +577,34 @@ static void fire_cwq_handlers (hio_t* hio)
 		/* TODO: more reuse of objects of different size? */
 		hio_freemem(hio, cwq);
 	}
+}
 
+/* fire one entry and dispose of it. returns the device to halt, or null. */
+static HIO_INLINE hio_dev_t* fire_one_cwq (hio_t* hio, hio_cwq_t* cwq)
+{
+	hio_dev_t* dev_to_halt;
+
+	if (cwq->dev->dev_evcb->on_write(cwq->dev, cwq->olen, cwq->ctx, &cwq->dstaddr) <= -1)
+	{
+		dev_to_halt = cwq->dev;
+	}
+	else
+	{
+		dev_to_halt = HIO_NULL;
+	}
+
+	unlink_cwq(hio, cwq);
+	recycle_cwq(hio, cwq);
+	return dev_to_halt;
+}
+
+static void fire_cwq_handlers (hio_t* hio)
+{
+	/* execute callbacks for completed write operations, in the order they
+	 * completed, across every device */
+	while (!HIO_CWQ_IS_EMPTY(&hio->cwq))
+	{
+		hio_dev_t* dev_to_halt = fire_one_cwq(hio, HIO_CWQ_HEAD(&hio->cwq));
 		if (dev_to_halt)
 		{
 			HIO_DEBUG2(hio, "DEV(%p) - halting a device for on_write error upon write completion[1] - %js\n", dev_to_halt, hio_geterrmsg(hio));
@@ -520,53 +615,20 @@ static void fire_cwq_handlers (hio_t* hio)
 
 static void fire_cwq_handlers_for_dev (hio_t* hio, hio_dev_t* dev, int for_kill)
 {
-	hio_cwq_t* cwq, * next;
-
 	HIO_ASSERT(hio, dev->cw_count > 0);  /* Ensure to check dev->cw_count before calling this function */
 
-	cwq = HIO_CWQ_HEAD(&hio->cwq);
-	while (cwq != &hio->cwq)
+	/* this used to walk the loop-wide queue looking for entries belonging to
+	 * this device, which made it cost the length of everything queued rather
+	 * than the length of this device's own share. */
+	while (dev->cwq_head)
 	{
-		next = HIO_CWQ_NEXT(cwq);
-		if (cwq->dev == dev) /* TODO: THIS LOOP TOO INEFFICIENT??? MAINTAIN PER-DEVICE LIST OF CWQ? */
-		{
-			hio_dev_t* dev_to_halt;
-			hio_oow_t cwqfl_index;
-
-			if (cwq->dev->dev_evcb->on_write(cwq->dev, cwq->olen, cwq->ctx, &cwq->dstaddr) <= -1)
-			{
-				dev_to_halt = cwq->dev;
-			}
-			else
-			{
-				dev_to_halt = HIO_NULL;
-			}
-
-			cwq->dev->cw_count--;
-			HIO_CWQ_UNLINK (cwq);
-
-			cwqfl_index = HIO_ALIGN_POW2(cwq->dstaddr.len, HIO_CWQFL_ALIGN) / HIO_CWQFL_SIZE;
-			if (cwqfl_index < HIO_COUNTOF(hio->cwqfl))
-			{
-				/* reuse the cwq object if dstaddr is 0 in size. chain it to the free list */
-				cwq->q_next = hio->cwqfl[cwqfl_index];
-				hio->cwqfl[cwqfl_index] = cwq;
-			}
-			else
-			{
-				/* TODO: more reuse of objects of different size? */
-				hio_freemem(hio, cwq);
-			}
-
+		hio_dev_t* dev_to_halt = fire_one_cwq(hio, dev->cwq_head);
 		if (!for_kill && dev_to_halt)
 		{
 			HIO_DEBUG2(hio, "DEV(%p) - halting a device for on_write error upon write completion[2] - %js\n", dev_to_halt, hio_geterrmsg(hio));
 			hio_dev_halt(dev_to_halt);
 		}
 	}
-		cwq = next;
-	}
-
 }
 
 static HIO_INLINE void clear_read_pending (hio_dev_t* dev)
@@ -765,8 +827,8 @@ static HIO_INLINE void handle_event (hio_t* hio, hio_dev_t* dev, int events, int
 		 * if the on_read calllback returns 0. */
 		while (1)
 		{
-			len = HIO_COUNTOF(hio->bigbuf);
-			x = dev->dev_mth->read(dev, hio->bigbuf, &len, &srcaddr);
+			len = hio->bigbuf.capa;
+			x = dev->dev_mth->read(dev, hio->bigbuf.ptr, &len, &srcaddr);
 			if (x <= -1)
 			{
 				HIO_DEBUG2(hio, "DEV(%p) - halting a device for read failure - %js\n", dev, hio_geterrmsg(hio));
@@ -807,18 +869,16 @@ static HIO_INLINE void handle_event (hio_t* hio, hio_dev_t* dev, int events, int
 				 * is started from within on_read() callback, and the input data is available
 				 * in the next iteration of this loop, the on_read() callback is triggered
 				 * before the on_write() callbacks scheduled before that on_read() callback. */
-			#if 0
+				/* only this device's completions, so the ordering the comment
+				 * above describes is the ordering that actually happens. firing
+				 * every device's here was a workaround for the old cost of
+				 * finding one device's entries. */
 				if (dev->cw_count > 0)
 				{
-					fire_cwq_handlers_for_dev (hio, dev);
-					/* it will still invoke the on_read() callbak below even if
+					fire_cwq_handlers_for_dev(hio, dev, 0);
+					/* it will still invoke the on_read() callback below even if
 					 * the device gets halted inside fire_cwq_handlers_for_dev() */
 				}
-			#else
-				/* currently fire_cwq_handlers_for_dev() scans the entire cwq list.
-				 * i might as well triggger handlers for all devices */
-				fire_cwq_handlers (hio);
-			#endif
 
 				if (len <= 0 && (dev->dev_cap & HIO_DEV_CAP_STREAM))
 				{
@@ -828,7 +888,7 @@ static HIO_INLINE void handle_event (hio_t* hio, hio_dev_t* dev, int events, int
 					dev->dev_cap |= HIO_DEV_CAP_RENEW_REQUIRED;
 
 					/* call the on_read callback to report EOF */
-					if (dev->dev_evcb->on_read(dev, hio->bigbuf, len, &srcaddr) <= -1 ||
+					if (dev->dev_evcb->on_read(dev, hio->bigbuf.ptr, len, &srcaddr) <= -1 ||
 					    (dev->dev_cap & HIO_DEV_CAP_OUT_CLOSED))
 					{
 						/* 1. input ended and its reporting failed or
@@ -847,11 +907,9 @@ static HIO_INLINE void handle_event (hio_t* hio, hio_dev_t* dev, int events, int
 				else
 				{
 					int y;
-		/* TODO: for a stream device, merge received data if bigbuf isn't full and fire the on_read callback
-		 *        when x == 0 or <= -1. you can  */
 
 					/* data available */
-					y = dev->dev_evcb->on_read(dev, hio->bigbuf, len, &srcaddr);
+					y = dev->dev_evcb->on_read(dev, hio->bigbuf.ptr, len, &srcaddr);
 					if (y <= -1)
 					{
 						HIO_DEBUG2(hio, "DEV(%p) - halting a non-stream device for on_read failure while output is closed - %js\n", dev, hio_geterrmsg(hio));
@@ -1133,6 +1191,21 @@ hio_dev_t* hio_dev_make (hio_t* hio, hio_oow_t dev_size, hio_dev_mth_t* dev_mth,
 	HIO_ASSERT(hio, dev->dev_prev == HIO_NULL);
 	HIO_ASSERT(hio, dev->dev_next == HIO_NULL);
 
+	/* make() has settled dev_rdmin by now, so this is the first point at
+	 * which the device's appetite for buffer space is known.
+	 *
+	 * a device carrying whole messages gets one attempt at each: whatever
+	 * does not fit is discarded and nothing says so. refuse the device rather
+	 * than let it lose data quietly. a stream device leaves dev_rdmin at 0 -
+	 * a short read simply leaves the remainder for the next one. */
+	if (dev->dev_rdmin > hio->bigbuf.capa)
+	{
+		hio_seterrbfmt(hio, HIO_ENOCAPA,
+			"read buffer of %zu octets too small for this device - %zu required",
+			hio->bigbuf.capa, dev->dev_rdmin);
+		goto oops_after_make;
+	}
+
 	/* set some internal capability bits according to the capabilities
 	 * removed by the device making callback for convenience sake. */
 	dev->dev_cap &= HIO_DEV_CAP_ALL_MASK; /* keep valid capability bits only. drop all internal-use bits */
@@ -1321,7 +1394,7 @@ void hio_dev_kill (hio_dev_t* dev)
 	clear_read_pending (dev);
 
 	/* clear completed write event queues */
-	if (dev->cw_count > 0) fire_cwq_handlers_for_dev (hio, dev, 1);
+	if (dev->cw_count > 0) fire_cwq_handlers_for_dev(hio, dev, 1);
 
 	/* clear pending write requests - won't fire on_write for pending write requests */
 	while (!HIO_WQ_IS_EMPTY(&dev->wq))
@@ -1676,7 +1749,12 @@ static HIO_INLINE int __enqueue_completed_write (hio_dev_t* dev, hio_iolen_t len
 
 	cwq->olen = len;
 
-	HIO_CWQ_ENQ (&dev->hio->cwq, cwq);
+	cwq->d_next = HIO_NULL;
+	if (dev->cwq_tail) dev->cwq_tail->d_next = cwq;
+	else dev->cwq_head = cwq;
+	dev->cwq_tail = cwq;
+
+	HIO_CWQ_ENQ(&dev->hio->cwq, cwq);
 	dev->cw_count++; /* increment the number of complete write operations */
 	return 0;
 }
@@ -1743,7 +1821,7 @@ static HIO_INLINE int __enqueue_pending_write (hio_dev_t* dev, hio_iolen_t olen,
 		}
 	}
 
-	HIO_WQ_ENQ (&dev->wq, q);
+	HIO_WQ_ENQ(&dev->wq, q);
 	dev->wq_len += urem;
 	if (!(dev->dev_cap & HIO_DEV_CAP_OUT_WATCHED))
 	{
@@ -1818,7 +1896,7 @@ static HIO_INLINE int __enqueue_pending_sendfile (hio_dev_t* dev, hio_iolen_t ol
 		}
 	}
 
-	HIO_WQ_ENQ (&dev->wq, q);
+	HIO_WQ_ENQ(&dev->wq, q);
 	dev->wq_len += urem;
 	if (!(dev->dev_cap & HIO_DEV_CAP_OUT_WATCHED))
 	{

lib/hio.h

@@ -134,6 +134,23 @@ enum hio_feature_t
 };
 typedef enum hio_feature_t hio_feature_t;
 
+/** default size of the shared read buffer, in octets. large enough to hold
+ *  any udp datagram, which is why it was this figure before it became
+ *  configurable. */
+#define HIO_DFL_READ_BUFFER_SIZE (65536)
+
+/** smallest read buffer that is safe for any stream device. one maximum-size
+ *  tls record must fit, or a single SSL_read() stops draining the record and
+ *  the surplus becomes unreachable - see the note in sck.c. */
+#define HIO_MIN_READ_BUFFER_SIZE (16384 + 2048)
+
+/** smallest read buffer that is safe for a device carrying datagrams of
+ *  arbitrary size. such a message is delivered once: a short buffer takes
+ *  what fits and the kernel discards the rest, with nothing reported. a
+ *  device declares its own requirement in dev_rdmin - this is what the
+ *  network datagram socket types use. */
+#define HIO_DGRAM_READ_BUFFER_SIZE (65535)
+
 enum hio_option_t
 {
 	HIO_TRAIT,
@@ -156,7 +173,25 @@ enum hio_option_t
 #endif
 
 	/* user-defined log writer */
-	HIO_LOG_WRITER
+	HIO_LOG_WRITER,
+
+	/** hio_oow_t. size of the shared read buffer, in octets.
+	 *
+	 * bigger means fewer read calls for the same traffic - a bulk transfer
+	 * saturates the buffer, so the count falls in proportion. smaller means
+	 * less memory per loop, which matters when a process runs one loop per
+	 * thread.
+	 *
+	 * it cannot be set below #HIO_MIN_READ_BUFFER_SIZE, nor below the largest
+	 * dev_rdmin among the devices already running, and a device whose
+	 * dev_rdmin exceeds it refuses to start. a datagram socket asks for
+	 * #HIO_DGRAM_READ_BUFFER_SIZE that way, because a truncated datagram is
+	 * lost silently rather than reported.
+	 *
+	 * setting it is refused with #HIO_EBUSY while the loop is dispatching
+	 * events, since on_read() is handed a pointer into the buffer and moving
+	 * it under a callback would leave that pointer dangling. */
+	HIO_READ_BUFFER_SIZE
 };
 typedef enum hio_option_t hio_option_t;
 
@@ -326,6 +361,14 @@ struct hio_cwq_t
 	hio_cwq_t*    q_next;
 	hio_cwq_t*    q_prev;
 
+	/* every entry sits on two queues at once. q_next/q_prev thread the
+	 * loop-wide queue, which is what fixes the order callbacks run in.
+	 * d_next threads the owning device's own queue, which is what makes
+	 * finding one device's entries cheap instead of a walk of everything.
+	 * both are filled in the same order, so the head of one is the head of
+	 * the other for any given device. */
+	hio_cwq_t*    d_next;
+
 	hio_iolen_t   olen;
 	void*         ctx;
 	hio_dev_t*    dev;
@@ -387,9 +430,12 @@ struct hio_wq_t
 	hio_dev_mth_t*  dev_mth; \
 	hio_dev_evcb_t* dev_evcb; \
 	hio_ntime_t     rtmout; \
+	hio_oow_t       dev_rdmin; /* smallest read buffer this device can work with. 0 means it does not care. set by the make() method */ \
 	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_cwq_t*      cwq_head; /* this device's completed-write queue, oldest first */ \
+	hio_cwq_t*      cwq_tail; \
 	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; \
@@ -810,7 +856,14 @@ struct hio_t
 		} xbuf; /* buffer to support sprintf */
 	} sprintf;
 
-	hio_uint8_t bigbuf[65535]; /* TODO: make this dynamic depending on devices added. device may indicate a buffer size required??? */
+	/* the one read buffer every device reads through. the pointer handed to
+	 * on_read() points into this, and is only valid for the duration of that
+	 * call - see HIO_READ_BUFFER_SIZE for how the size is chosen. */
+	struct
+	{
+		hio_uint8_t* ptr;
+		hio_oow_t capa;
+	} bigbuf;
 
 	hio_cfmb_t cfmb; /* list head of cfmbs */
 	hio_dev_t actdev; /* list head of active devices */

lib/sck.c

@@ -42,7 +42,12 @@
 #	include <netinet/if_ether.h>
 #endif
 
-#if defined(HAVE_NETINET_SCTP_H)
+/* HIO_ENABLE_SCTP is settled by configure (--enable-sctp, on by default where
+ * the system provides it) and reported in its summary. it used to be derived
+ * here from nothing but the presence of the header, so the feature came and
+ * went with whatever happened to be installed and no one was told - which is
+ * how the SCTP_INITMSG call below stayed wrong for so long. */
+#if defined(HIO_ENABLE_SCTP) && defined(HAVE_NETINET_SCTP_H)
 #	include <netinet/sctp.h>
 #	if defined(IPPROTO_SCTP)
 #		define ENABLE_SCTP
@@ -134,9 +139,12 @@
 
 /* see the note at SSL_set_read_ahead() in do_ssl(). the core reads through
  * hio->bigbuf, and that buffer being at least one record wide is what keeps
- * SSL_pending() at zero. this fires the day the TODO on bigbuf in hio.h gets
- * acted on and the buffer shrinks below a record. */
-HIO_STATIC_ASSERT(HIO_COUNTOF(((hio_t*)0)->bigbuf) >= HIO_SSL_MAX_READ_RECORD);
+ * SSL_pending() at zero.
+ *
+ * the buffer is sized at run time now, so what is checked here is that the
+ * floor the core refuses to go below is itself wide enough. the runtime
+ * enforcement comes from that floor - see HIO_READ_BUFFER_SIZE. */
+HIO_STATIC_ASSERT(HIO_MIN_READ_BUFFER_SIZE >= HIO_SSL_MAX_READ_RECORD);
 #endif
 
 /* ========================================================================= */
@@ -314,13 +322,13 @@ static struct sck_type_map_t sck_type_map[] =
 	/* HIO_DEV_SCK_SCTP6 */
 	{ AF_INET6,  SOCK_STREAM,     IPPROTO_SCTP,      1, 1, HIO_DEV_CAP_STREAM },
 
-	/* HIO_DEV_SCK_SCTP4_SP - not implemented */
-	/*{ AF_INET,   SOCK_SEQPACKET,  IPPROTO_SCTP,      1, 1, 0 },*/
-	{ -1,        0,               0,                 0, 0, 0 },
+	/* HIO_DEV_SCK_SCTP4_SP - one-to-many. listen() is called but hio never
+	 * accepts: associations are not devices in this model, they are told apart
+	 * by the source address on each message. */
+	{ AF_INET,   SOCK_SEQPACKET,  IPPROTO_SCTP,      1, 1, 0 },
 
-	/* HIO_DEV_SCK_SCTP6_SP - not implemented */
-	/*{ AF_INET6,  SOCK_SEQPACKET,  IPPROTO_SCTP,      1, 1, 0 },*/
-	{ -1,        0,               0,                 0, 0, 0 },
+	/* HIO_DEV_SCK_SCTP6_SP */
+	{ AF_INET6,  SOCK_SEQPACKET,  IPPROTO_SCTP,      1, 1, 0 },
 #else
 	{ -1,        0,               0,                 0, 0, 0 },
 	{ -1,        0,               0,                 0, 0, 0 },
@@ -367,6 +375,24 @@ static struct sck_type_map_t sck_type_map[] =
 	{ __AF_BPF, 0, 0, 0, 0, 0 } /* not implemented yet */
 };
 
+/* how much read buffer one device of this type needs in a single read.
+ *
+ * a stream is content to be read in pieces and asks for nothing - a short read
+ * leaves the remainder for the next one. a datagram socket is not: whatever
+ * does not fit in one recvfrom() is discarded with nothing reported, so it asks
+ * for enough to hold the largest datagram it could ever see. the qx channel is
+ * message-oriented too, but its message is a fixed struct, so it asks for only
+ * that much.
+ *
+ * both the listening/connecting path and the accepted path go through here, so
+ * the two cannot drift apart. */
+static hio_oow_t sck_type_rdmin (hio_dev_sck_type_t type)
+{
+	if (sck_type_map[type].domain == HIO_AF_QX) return HIO_SIZEOF(hio_dev_sck_qxmsg_t);
+	if (!(sck_type_map[type].extra_dev_cap & HIO_DEV_CAP_STREAM)) return HIO_DGRAM_READ_BUFFER_SIZE;
+	return 0;
+}
+
 /* ======================================================================== */
 
 static void connect_timedout (hio_t* hio, const hio_ntime_t* now, hio_tmrjob_t* job)
@@ -465,14 +491,48 @@ static int dev_sck_make (hio_dev_t* dev, void* ctx)
 		if (hnd == HIO_SYSHND_INVALID) goto oops;
 
 	#if defined(ENABLE_SCTP)
-		if (sck_type_map[arg->type].type == SOCK_SEQPACKET &&
-		    sck_type_map[arg->type].proto == IPPROTO_SCTP)
+		if (sck_type_map[arg->type].proto == IPPROTO_SCTP)
 		{
-			struct sctp_event_subscribe sctp_ev_s;
+			struct sctp_event_subscribe ev;
 
-			HIO_MEMSET(&sctp_ev_s, 0, HIO_SIZEOF(sctp_ev_s));
-			sctp_ev_s.sctp_data_io_event = 1;
-			setsockopt (hnd, IPPROTO_SCTP, SCTP_EVENTS, &sctp_ev_s, HIO_SIZEOF(sctp_ev_s));
+			/* [NOTE] this used to hand a struct sctp_initmsg to SCTP_EVENTS,
+			 * which takes a struct sctp_event_subscribe. eight zero octets
+			 * read as "subscribe to nothing", so it returned success and did
+			 * nothing at all - and the SCTP_INITMSG that was meant to request
+			 * streams never happened. */
+			if (arg->sctp_ostreams > 0 || arg->sctp_instreams > 0)
+			{
+				struct sctp_initmsg im;
+				HIO_MEMSET(&im, 0, HIO_SIZEOF(im));
+				im.sinit_num_ostreams = arg->sctp_ostreams;
+				im.sinit_max_instreams = arg->sctp_instreams;
+				if (setsockopt(hnd, IPPROTO_SCTP, SCTP_INITMSG, &im, HIO_SIZEOF(im)) <= -1)
+				{
+					hio_seterrwithsyserr(hio, 0, errno);
+					goto oops;
+				}
+			}
+
+			/* sctp_data_io_event is what delivers the per-message ancillary
+			 * data - the stream number among it - so it is required for any
+			 * of the stream handling to work.
+			 *
+			 * the rest are notifications: association up/down, a path
+			 * changing state, a send that failed. they arrive interleaved
+			 * with data on this same socket, flagged MSG_NOTIFICATION, and
+			 * are routed to on_notification() rather than on_read(). */
+			HIO_MEMSET(&ev, 0, HIO_SIZEOF(ev));
+			ev.sctp_data_io_event = 1;
+			ev.sctp_association_event = 1;
+			ev.sctp_address_event = 1;
+			ev.sctp_send_failure_event = 1;
+			ev.sctp_peer_error_event = 1;
+			ev.sctp_shutdown_event = 1;
+			if (setsockopt(hnd, IPPROTO_SCTP, SCTP_EVENTS, &ev, HIO_SIZEOF(ev)) <= -1)
+			{
+				hio_seterrwithsyserr(hio, 0, errno);
+				goto oops;
+			}
 		}
 	#endif
 	}
@@ -480,10 +540,13 @@ static int dev_sck_make (hio_dev_t* dev, void* ctx)
 	rdev->hnd = hnd;
 	rdev->side_chan = side_chan;
 	rdev->dev_cap = HIO_DEV_CAP_IN | HIO_DEV_CAP_OUT | sck_type_map[arg->type].extra_dev_cap;
+
+	rdev->dev_rdmin = sck_type_rdmin(arg->type);
 	rdev->on_write = arg->on_write;
 	rdev->on_read = arg->on_read;
 	rdev->on_connect = arg->on_connect;
 	rdev->on_disconnect = arg->on_disconnect;
+	rdev->on_notification = arg->on_notification;
 	rdev->on_raw_accept = arg->on_raw_accept;
 	rdev->type = arg->type;
 
@@ -497,11 +560,19 @@ oops:
 	return -1;
 }
 
+/* what make_accepted_client_connection() hands to dev_sck_make_client() */
+typedef struct sck_make_client_ctx_t sck_make_client_ctx_t;
+struct sck_make_client_ctx_t
+{
+	hio_syshnd_t hnd;
+	hio_dev_sck_type_t type;
+};
+
 static int dev_sck_make_client (hio_dev_t* dev, void* ctx)
 {
 	hio_t* hio = dev->hio;
 	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
-	hio_syshnd_t* clisckhnd = (hio_syshnd_t*)ctx;
+	sck_make_client_ctx_t* mc = (sck_make_client_ctx_t*)ctx;
 
 	/* create a socket device that is made of a socket connection
 	 * on a listening socket.
@@ -509,10 +580,17 @@ static int dev_sck_make_client (hio_dev_t* dev, void* ctx)
 	 * most of the initialization is done by the listening socket device
 	 * after a client socket has been created. */
 
-	rdev->hnd = *clisckhnd;
+	rdev->hnd = mc->hnd;
 	rdev->tmrjob_index = HIO_TMRIDX_INVALID;
 	rdev->side_chan = HIO_SYSHND_INVALID;
 
+	/* the type is settled again by the caller, but dev_rdmin has to be known
+	 * before this method returns: hio_dev_make() checks it against the loop's
+	 * read buffer the moment make() is done, and there is no second chance to
+	 * refuse the device after that. */
+	rdev->type = mc->type;
+	rdev->dev_rdmin = sck_type_rdmin(mc->type);
+
 	if (hio_makesyshndasync(hio, rdev->hnd) <= -1 ||
 	    hio_makesyshndcloexec(hio, rdev->hnd) <= -1) goto oops;
 
@@ -537,6 +615,10 @@ 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;
+	/* remembered for the trace at the end, by which point rdev->hnd has been
+	 * closed and reset. HIO_UNUSED because that trace is the only reader and
+	 * it compiles away in a release build. */
+	hio_syshnd_t hnd HIO_UNUSED = rdev->hnd;
 
 	HIO_DEBUG2(hio, "SCK(%p) - being killed [%d]\n", rdev, rdev->hnd);
 #if 0
@@ -547,7 +629,7 @@ static int dev_sck_kill (hio_dev_t* dev, int force)
 			/* for HIO_DEV_SCK_CONNECTING, HIO_DEV_SCK_CONNECTING_SSL, and HIO_DEV_SCK_ACCEPTING_SSL
 			 * on_disconnect() is called without corresponding on_connect().
 			 * it is the same if connect or accept has not been called. */
-			if (rdev->on_disconnect) rdev->on_disconnect (rdev);
+			if (rdev->on_disconnect) rdev->on_disconnect(rdev);
 		/*}*/
 	}
 	else
@@ -759,14 +841,22 @@ static int recvmsg_sctp(
 }
 
 static int sendmsg_sctp(
-	int s, const hio_iovec_t* iov, hio_oow_t iovcnt, struct sockaddr* dstaddr, hio_scklen_t dstaddrlen,
-	hio_uint32_t ppid, hio_uint32_t flags, hio_uint16_t stream_no, hio_uint32_t ttl, hio_uint32_t context)
+	int s, const hio_iovec_t* iov, hio_oow_t iovcnt,struct sockaddr* dstaddr, hio_scklen_t dstaddrlen,
+	hio_uint32_t ppid, hio_uint32_t flags, hio_uint16_t stream_no, hio_uint32_t ttl, hio_uint32_t context,
+	hio_int32_t assoc_id)
 {
 	struct sctp_sndrcvinfo* sinfo;
 	struct msghdr msg;
 	struct cmsghdr* cmsg;
 	hio_uint8_t cmsg_buf[CMSG_SPACE(HIO_SIZEOF(*sinfo))];
 
+	/* both are handed to the kernel, so neither may carry anything
+	 * uninitialised. recvmsg_sctp() above already zeroes its msghdr and this
+	 * one did not - struct msghdr has padding members on some ABIs. the
+	 * control buffer is fully written below, so zeroing it is belt and braces. */
+	HIO_MEMSET(&msg, 0, HIO_SIZEOF(msg));
+	HIO_MEMSET(cmsg_buf, 0, HIO_SIZEOF(cmsg_buf));
+
 	msg.msg_name = dstaddr;
 	msg.msg_namelen = dstaddrlen;
 	msg.msg_iov = (struct iovec*)iov;
@@ -789,10 +879,42 @@ static int sendmsg_sctp(
 	sinfo->sinfo_stream = stream_no;
 	sinfo->sinfo_timetolive = ttl;
 	sinfo->sinfo_context = context;
+	/* on a one-to-many socket a non-zero association id names the association
+	 * directly, which survives a peer changing address; zero falls back to
+	 * addressing by msg_name. it is ignored on a one-to-one socket. */
+	sinfo->sinfo_assoc_id = (sctp_assoc_t)assoc_id;
 
 	return sendmsg(s, &msg, 0);
 }
 
+/* the core passes no destination for a connected socket - hio_dev_sck_write()
+ * with a null dstaddr - so these must not be dereferenced blindly. an absent
+ * destination means "the peer we are associated with", stream 0. */
+static HIO_INLINE struct sockaddr* dstaddr_sockaddr (const hio_devaddr_t* dstaddr)
+{
+	return (dstaddr && dstaddr->ptr)? (struct sockaddr*)dstaddr->ptr: HIO_NULL;
+}
+
+static HIO_INLINE hio_scklen_t dstaddr_len (const hio_devaddr_t* dstaddr)
+{
+	return (dstaddr && dstaddr->ptr)? (hio_scklen_t)dstaddr->len: 0;
+}
+
+static HIO_INLINE hio_uint16_t dstaddr_chan (const hio_devaddr_t* dstaddr)
+{
+	return (dstaddr && dstaddr->ptr)? hio_skad_get_chan((const hio_skad_t*)dstaddr->ptr): 0;
+}
+
+static HIO_INLINE hio_uint32_t dstaddr_ppid (const hio_devaddr_t* dstaddr)
+{
+	return (dstaddr && dstaddr->ptr)? hio_skad_get_ppid((const hio_skad_t*)dstaddr->ptr): 0;
+}
+
+static HIO_INLINE hio_int32_t dstaddr_assoc (const hio_devaddr_t* dstaddr)
+{
+	return (dstaddr && dstaddr->ptr)? hio_skad_get_assoc((const hio_skad_t*)dstaddr->ptr): 0;
+}
+
 static int dev_sck_read_sctp_sp (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_devaddr_t* srcaddr)
 {
 /* NOTE: sctp support is far away from complete */
@@ -819,12 +941,47 @@ static int dev_sck_read_sctp_sp (hio_dev_t* dev, void* buf, hio_iolen_t* len, hi
 		return -1;
 	}
 
-/*
-if (msg_flags & MSG_EOR) end of message??
-else push to buffer???
-*/
-	hio_skad_set_chan (&rdev->remoteaddr, sri.sinfo_stream);
-/* TODO: how to store sri.sinfo_ppid or sri.sinfo_context? */
+	if (msg_flags & MSG_NOTIFICATION)
+	{
+		/* an association or path event, not payload. handing this to
+		 * on_read() would splice an event record into the data stream. */
+		if (rdev->on_notification) rdev->on_notification(rdev, buf, x);
+		return 0; /* nothing readable for the caller this time round */
+	}
+
+	/* SOCK_SEQPACKET keeps message boundaries. MSG_EOR says this read reached
+	 * the end of one; without it, what we hold is the front of a message too
+	 * large for the read buffer, and the rest is still queued.
+	 *
+	 * such a fragment must not go to on_read(): it would arrive looking like a
+	 * whole message and the remainder would arrive as another, quietly turning
+	 * one message into two. reassembling here is not possible either - one
+	 * device carries every association, so partial messages from different
+	 * associations interleave and the buffer would have to be keyed by
+	 * association. that belongs with the peel-off model, where an association
+	 * is a device of its own and reassembly is per-device.
+	 *
+	 * so the message is dropped whole and the loss is reported. one oversized
+	 * message from one peer costs that message and nothing else - the socket
+	 * keeps serving everyone else. the remedy is a larger read buffer, which
+	 * HIO_READ_BUFFER_SIZE now makes possible. */
+	if (rdev->sctp_discarding || !(msg_flags & MSG_EOR))
+	{
+		if (!rdev->sctp_discarding)
+		{
+			rdev->sctp_discarding = 1;
+			HIO_INFO2 (hio, "SCK(%p) - discarding an sctp message too large for the read buffer of %zu octets\n", rdev, hio->bigbuf.capa);
+		}
+		/* keep swallowing fragments until the one that ends the message */
+		if (msg_flags & MSG_EOR) rdev->sctp_discarding = 0;
+		return 0;
+	}
+
+	hio_skad_set_chan(&rdev->remoteaddr, sri.sinfo_stream);
+	hio_skad_set_ppid(&rdev->remoteaddr, sri.sinfo_ppid);
+	/* the stable name for the association this arrived on. a reply addressed
+	 * by it reaches the same association even if the peer's address changed. */
+	hio_skad_set_assoc(&rdev->remoteaddr, (hio_int32_t)sri.sinfo_assoc_id);
 	srcaddr->ptr = &rdev->remoteaddr;
 	srcaddr->len = srcaddrlen;
 
@@ -1114,12 +1271,13 @@ static int dev_sck_write_sctp_sp (hio_dev_t* dev, const void* data, hio_iolen_t*
 	iov.iov_ptr = (void*)data;
 	iov.iov_len = *len;
 	x = sendmsg_sctp(rdev->hnd,
-		&iov, 1, dstaddr->ptr, dstaddr->len,
-		0, /* ppid - opaque */
+		&iov, 1, dstaddr_sockaddr(dstaddr), dstaddr_len(dstaddr),
+		dstaddr_ppid(dstaddr), /* ppid - opaque, carried on the address */
 		0, /* flags (e.g. SCTP_UNORDERED, SCTP_EOF, SCT_ABORT, ...) */
-		hio_skad_get_chan(dstaddr->ptr), /* stream number */
+		dstaddr_chan(dstaddr), /* stream number */
 		0, /* ttl */
-		0 /* context*/);
+		0, /* context*/
+		dstaddr_assoc(dstaddr));
 	if (x <= -1)
 	{
 		if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN) return 0;  /* no data can be written */
@@ -1132,6 +1290,119 @@ static int dev_sck_write_sctp_sp (hio_dev_t* dev, const void* data, hio_iolen_t*
 	return 1;
 }
 
+/* the one-to-one sctp types carry HIO_DEV_CAP_STREAM, so the core treats them
+ * like a tcp socket: a zero-length write is the writing-end shutdown, and a
+ * zero-length read is EOF. the seqpacket methods above answer neither of those
+ * conventions - a zero-length sendmsg() puts an empty message on the wire
+ * rather than closing anything - so the stream variants need their own pair.
+ *
+ * what they add over the plain stream methods is the ancillary data: the
+ * stream number and ppid, which is the whole point of using sctp. */
+static int dev_sck_read_sctp_stream (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_devaddr_t* srcaddr)
+{
+	hio_t* hio = dev->hio;
+	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
+	hio_scklen_t srcaddrlen;
+	ssize_t x;
+	int msg_flags = 0;
+	struct sctp_sndrcvinfo sri;
+
+	HIO_MEMSET(&sri, 0, HIO_SIZEOF(sri));
+	srcaddrlen = HIO_SIZEOF(rdev->remoteaddr);
+
+	x = recvmsg_sctp(rdev->hnd, buf, *len, (struct sockaddr*)&rdev->remoteaddr, &srcaddrlen, &sri, &msg_flags);
+	if (x <= -1)
+	{
+		int eno = errno;
+		if (eno == EINPROGRESS || eno == EWOULDBLOCK || eno == EAGAIN) return 0;
+		if (eno == EINTR) return 0;
+		hio_seterrwithsyserr(hio, 0, eno);
+		return -1;
+	}
+
+	if (msg_flags & MSG_NOTIFICATION)
+	{
+		/* an association or path event, not payload */
+		if (rdev->on_notification) rdev->on_notification(rdev, buf, x);
+		return 0;
+	}
+
+	/* x of 0 falls through as a zero length, which is what the core reads as
+	 * EOF on a stream device */
+	hio_skad_set_chan (&rdev->remoteaddr, sri.sinfo_stream);
+	hio_skad_set_ppid (&rdev->remoteaddr, sri.sinfo_ppid);
+	hio_skad_set_assoc (&rdev->remoteaddr, (hio_int32_t)sri.sinfo_assoc_id);
+
+	*len = x;
+	return 1;
+}
+
+static int dev_sck_write_sctp_stream (hio_dev_t* dev, const void* data, hio_iolen_t* len, const hio_devaddr_t* dstaddr)
+{
+	hio_t* hio = dev->hio;
+	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
+	ssize_t x;
+	hio_iovec_t iov;
+
+	if (*len <= 0)
+	{
+		/* the writing-end shutdown indicator, as for any stream device */
+		if (shutdown(rdev->hnd, SHUT_WR) <= -1)
+		{
+			hio_seterrwithsyserr(hio, 0, errno);
+			return -1;
+		}
+		return 1; /* must be non-zero, or the core queues the request */
+	}
+
+	iov.iov_ptr = (void*)data;
+	iov.iov_len = *len;
+	x = sendmsg_sctp(rdev->hnd,
+		&iov, 1, dstaddr_sockaddr(dstaddr), dstaddr_len(dstaddr),
+		dstaddr_ppid(dstaddr), 0, dstaddr_chan(dstaddr), 0, 0, dstaddr_assoc(dstaddr));
+	if (x <= -1)
+	{
+		if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN) return 0;
+		if (errno == EINTR) return 0;
+		hio_seterrwithsyserr(hio, 0, errno);
+		return -1;
+	}
+
+	*len = x;
+	return 1;
+}
+
+static int dev_sck_writev_sctp_stream (hio_dev_t* dev, const hio_iovec_t* iov, hio_iolen_t* iovcnt, const hio_devaddr_t* dstaddr)
+{
+	hio_t* hio = dev->hio;
+	hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
+	ssize_t x;
+
+	if (*iovcnt <= 0)
+	{
+		if (shutdown(rdev->hnd, SHUT_WR) <= -1)
+		{
+			hio_seterrwithsyserr(hio, 0, errno);
+			return -1;
+		}
+		return 1;
+	}
+
+	x = sendmsg_sctp(rdev->hnd,
+		iov, *iovcnt, dstaddr_sockaddr(dstaddr), dstaddr_len(dstaddr),
+		dstaddr_ppid(dstaddr), 0, dstaddr_chan(dstaddr), 0, 0, dstaddr_assoc(dstaddr));
+	if (x <= -1)
+	{
+		if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN) return 0;
+		if (errno == EINTR) return 0;
+		hio_seterrwithsyserr(hio, 0, errno);
+		return -1;
+	}
+
+	*iovcnt = x;
+	return 1;
+}
+
 static int dev_sck_writev_sctp_sp (hio_dev_t* dev, const hio_iovec_t* iov, hio_iolen_t* iovcnt, const hio_devaddr_t* dstaddr)
 {
 	hio_t* hio = dev->hio;
@@ -1139,12 +1410,13 @@ static int dev_sck_writev_sctp_sp (hio_dev_t* dev, const hio_iovec_t* iov, hio_i
 	ssize_t x;
 
 	x = sendmsg_sctp(rdev->hnd,
-		iov, *iovcnt, dstaddr->ptr, dstaddr->len,
-		0, /* ppid - opaque */
+		iov, *iovcnt, dstaddr_sockaddr(dstaddr), dstaddr_len(dstaddr),
+		dstaddr_ppid(dstaddr), /* ppid - opaque, carried on the address */
 		0, /* flags (e.g. SCTP_UNORDERED, SCTP_EOF, SCT_ABORT, ...) */
-		hio_skad_get_chan(dstaddr->ptr), /* stream number */
+		dstaddr_chan(dstaddr), /* stream number */
 		0, /* ttl */
-		0 /* context*/);
+		0, /* context*/
+		dstaddr_assoc(dstaddr));
 	if (x <= -1)
 	{
 		if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN) return 0;  /* no data can be written */
@@ -1430,7 +1702,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 				{
 					if (!(bnd->options & HIO_DEV_SCK_BIND_IGNERR))
 					{
-						hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to set IPV6_V6ONLY");
+						hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to set IPV6_V6ONLY");
 						return -1;
 					}
 				}
@@ -1443,7 +1715,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 				if (setsockopt(rdev->hnd, SOL_SOCKET, SO_BROADCAST, &v, HIO_SIZEOF(v)) <= -1)
 				{
 					/* not affected by HIO_DEV_SCK_BIND_IGNERR */
-					hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to set SO_BROADCAST");
+					hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to set SO_BROADCAST");
 					return -1;
 				}
 			}
@@ -1456,7 +1728,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 				{
 					if (!(bnd->options & HIO_DEV_SCK_BIND_IGNERR))
 					{
-						hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to set SO_REUSEADDR");
+						hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to set SO_REUSEADDR");
 						return -1;
 					}
 				}
@@ -1476,7 +1748,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 				{
 					if (!(bnd->options & HIO_DEV_SCK_BIND_IGNERR))
 					{
-						hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to set SO_REUSEPORT");
+						hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to set SO_REUSEPORT");
 						return -1;
 					}
 				}
@@ -1494,7 +1766,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 				int v = 1;
 				if (setsockopt(rdev->hnd, SOL_IP, IP_TRANSPARENT, &v, HIO_SIZEOF(v)) <= -1)
 				{
-					hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to set IP_TRANSPARENT");
+					hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to set IP_TRANSPARENT");
 					return -1;
 				}
 			/* ignore it if not available
@@ -1639,13 +1911,13 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
 		#endif
 			/* the socket is already non-blocking */
 /*{
-int flags = fcntl (rdev->hnd, F_GETFL);
-fcntl (rdev->hnd, F_SETFL, flags & ~O_NONBLOCK);
+int flags = fcntl(rdev->hnd, F_GETFL);
+fcntl(rdev->hnd, F_SETFL, flags & ~O_NONBLOCK);
 }*/
 			x = connect(rdev->hnd, sa, sl);
 /*{
-int flags = fcntl (rdev->hnd, F_GETFL);
-fcntl (rdev->hnd, F_SETFL, flags | O_NONBLOCK);
+int flags = fcntl(rdev->hnd, F_GETFL);
+fcntl(rdev->hnd, F_SETFL, flags | O_NONBLOCK);
 }*/
 			if (x <= -1)
 			{
@@ -1794,6 +2066,43 @@ static hio_dev_mth_t dev_mth_sck_stateless =
 };
 
 
+#if defined(ENABLE_SCTP)
+/* one-to-one sctp. a stream device like tcp, but reads and writes carry the
+ * per-message ancillary data, so the stream number and ppid are reachable.
+ * no sendfile: it would bypass sendmsg() and lose that data. */
+static hio_dev_mth_t dev_mth_sck_sctp_stream =
+{
+	dev_sck_make,
+	dev_sck_kill,
+	HIO_NULL,
+	dev_sck_getsyshnd,
+	HIO_NULL,
+	dev_sck_ioctl,
+
+	dev_sck_read_sctp_stream,
+	dev_sck_write_sctp_stream,
+	dev_sck_writev_sctp_stream,
+	HIO_NULL,          /* sendfile */
+	HIO_NULL           /* readpending */
+};
+
+static hio_dev_mth_t dev_mth_clisck_sctp_stream =
+{
+	dev_sck_make_client,
+	dev_sck_kill,
+	HIO_NULL,
+	dev_sck_getsyshnd,
+	HIO_NULL,
+	dev_sck_ioctl,
+
+	dev_sck_read_sctp_stream,
+	dev_sck_write_sctp_stream,
+	dev_sck_writev_sctp_stream,
+	HIO_NULL,          /* sendfile */
+	HIO_NULL           /* readpending */
+};
+#endif
+
 static hio_dev_mth_t dev_mth_sck_stream =
 {
 	dev_sck_make,
@@ -1921,7 +2230,7 @@ static int harvest_outgoing_connection (hio_dev_sck_t* rdev)
 	len = HIO_SIZEOF(errcode);
 	if (getsockopt(rdev->hnd, SOL_SOCKET, SO_ERROR, (char*)&errcode, &len) <= -1)
 	{
-		hio_seterrbfmtwithsyserr (hio, 0, errno, "unable to get SO_ERROR");
+		hio_seterrbfmtwithsyserr(hio, 0, errno, "unable to get SO_ERROR");
 		return -1;
 	}
 	else if (errcode == 0)
@@ -1929,11 +2238,13 @@ static int harvest_outgoing_connection (hio_dev_sck_t* rdev)
 		hio_skad_t localaddr;
 		hio_scklen_t addrlen;
 
+		HIO_MEMSET(&localaddr, 0, HIO_SIZEOF(localaddr)); /* see hio_dev_sck_getsockaddr() */
+
 		/* connected */
 
 		if (rdev->tmrjob_index != HIO_TMRIDX_INVALID)
 		{
-			hio_deltmrjob (hio, rdev->tmrjob_index);
+			hio_deltmrjob(hio, rdev->tmrjob_index);
 			HIO_ASSERT(hio, rdev->tmrjob_index == HIO_TMRIDX_INVALID);
 		}
 
@@ -1984,7 +2295,7 @@ static int harvest_outgoing_connection (hio_dev_sck_t* rdev)
 		ssl_connected:
 	#endif
 			HIO_DEV_SCK_SET_PROGRESS(rdev, HIO_DEV_SCK_CONNECTED);
-			if (rdev->on_connect) rdev->on_connect (rdev);
+			if (rdev->on_connect) rdev->on_connect(rdev);
 	#if defined(USE_SSL)
 		}
 	#endif
@@ -2005,6 +2316,7 @@ static int harvest_outgoing_connection (hio_dev_sck_t* rdev)
 
 static int make_accepted_client_connection (hio_dev_sck_t* rdev, hio_syshnd_t clisck, hio_skad_t* remoteaddr, hio_dev_sck_type_t clisck_type)
 {
+	sck_make_client_ctx_t mc;
 	hio_t* hio = rdev->hio;
 	hio_dev_sck_t* clidev;
 	hio_scklen_t addrlen;
@@ -2029,12 +2341,15 @@ static int make_accepted_client_connection (hio_dev_sck_t* rdev, hio_syshnd_t cl
 	 *   device capability. currently, stream or non-stream is supported.
 	 */
 #if defined(ENABLE_SCTP)
-	dev_mth = (sck_type_map[clisck_type].extra_dev_cap & HIO_DEV_CAP_STREAM)? &dev_mth_clisck_stream:
+	dev_mth = (sck_type_map[clisck_type].extra_dev_cap & HIO_DEV_CAP_STREAM)?
+	              ((sck_type_map[clisck_type].proto == IPPROTO_SCTP)? &dev_mth_clisck_sctp_stream: &dev_mth_clisck_stream):
 	          (sck_type_map[clisck_type].proto == IPPROTO_SCTP)? &dev_mth_clisck_sctp_sp: &dev_mth_clisck_stateless;
 #else
 	dev_mth = (sck_type_map[clisck_type].extra_dev_cap & HIO_DEV_CAP_STREAM)? &dev_mth_clisck_stream: &dev_mth_clisck_stateless;
 #endif
-	clidev = (hio_dev_sck_t*)hio_dev_make(hio, rdev->dev_size, dev_mth, rdev->dev_evcb, &clisck);
+	mc.hnd = clisck;
+	mc.type = clisck_type;
+	clidev = (hio_dev_sck_t*)hio_dev_make(hio, rdev->dev_size, dev_mth, rdev->dev_evcb, &mc);
 	if (HIO_UNLIKELY(!clidev))
 	{
 		/* [NOTE] 'clisck' is closed by callback methods called by hio_dev_make() upon failure */
@@ -2273,12 +2588,12 @@ static int dev_evcb_sck_ready_stream (hio_dev_t* dev, int events)
 
 				if (rdev->tmrjob_index != HIO_TMRIDX_INVALID)
 				{
-					hio_deltmrjob (rdev->hio, rdev->tmrjob_index);
+					hio_deltmrjob(rdev->hio, rdev->tmrjob_index);
 					rdev->tmrjob_index = HIO_TMRIDX_INVALID;
 				}
 
 				HIO_DEV_SCK_SET_PROGRESS(rdev, HIO_DEV_SCK_CONNECTED);
-				if (rdev->on_connect) rdev->on_connect (rdev);
+				if (rdev->on_connect) rdev->on_connect(rdev);
 				return 0;
 			}
 			else
@@ -2345,12 +2660,12 @@ static int dev_evcb_sck_ready_stream (hio_dev_t* dev, int events)
 
 				if (rdev->tmrjob_index != HIO_TMRIDX_INVALID)
 				{
-					hio_deltmrjob (rdev->hio, rdev->tmrjob_index);
+					hio_deltmrjob(rdev->hio, rdev->tmrjob_index);
 					rdev->tmrjob_index = HIO_TMRIDX_INVALID;
 				}
 
 				HIO_DEV_SCK_SET_PROGRESS(rdev, HIO_DEV_SCK_ACCEPTED);
-				if (rdev->on_connect) rdev->on_connect (rdev);
+				if (rdev->on_connect) rdev->on_connect(rdev);
 
 				return 0;
 			}
@@ -2617,7 +2932,12 @@ hio_dev_sck_t* hio_dev_sck_make (hio_t* hio, hio_oow_t xtnsize, const hio_dev_sc
 	{
 		rdev = (hio_dev_sck_t*)hio_dev_make(
 			hio, HIO_SIZEOF(hio_dev_sck_t) + xtnsize,
-			&dev_mth_sck_stream, &dev_sck_event_callbacks_stream, (void*)info);
+		#if defined(ENABLE_SCTP)
+			(sck_type_map[info->type].proto == IPPROTO_SCTP)? &dev_mth_sck_sctp_stream: &dev_mth_sck_stream,
+		#else
+			&dev_mth_sck_stream,
+		#endif
+			&dev_sck_event_callbacks_stream, (void*)info);
 	}
 #if defined(ENABLE_SCTP)
 	else if (sck_type_map[info->type].proto == IPPROTO_SCTP)
@@ -2699,9 +3019,13 @@ int hio_dev_sck_getsockopt (hio_dev_sck_t* dev, int level, int optname, void* op
 int hio_dev_sck_getsockaddr (hio_dev_sck_t* dev, hio_skad_t* skad)
 {
 	hio_scklen_t addrlen = HIO_SIZEOF(*skad);
+	/* the system fills the sockaddr and nothing else. the extra area past it -
+	 * chan, ppid, assoc - would keep whatever the caller's memory happened to
+	 * hold, and those are read back later and passed to the kernel. */
+	HIO_MEMSET(skad, 0, HIO_SIZEOF(*skad));
 	if (dev->type == HIO_DEV_SCK_QX)
 	{
-		hio_skad_init_for_qx (skad);
+		hio_skad_init_for_qx(skad);
 	}
 	else if (getsockname(dev->hnd, (struct sockaddr*)skad, &addrlen) <= -1)
 	{
@@ -2714,9 +3038,10 @@ int hio_dev_sck_getsockaddr (hio_dev_sck_t* dev, hio_skad_t* skad)
 int hio_dev_sck_getpeeraddr (hio_dev_sck_t* dev, hio_skad_t* skad)
 {
 	hio_scklen_t addrlen = HIO_SIZEOF(*skad);
+	HIO_MEMSET(skad, 0, HIO_SIZEOF(*skad)); /* see hio_dev_sck_getsockaddr() */
 	if (dev->type == HIO_DEV_SCK_QX)
 	{
-		hio_skad_init_for_qx (skad);
+		hio_skad_init_for_qx(skad);
 	}
 	else if (getpeername(dev->hnd, (struct sockaddr*)skad, &addrlen) <= -1)
 	{

lib/skad.c

@@ -45,6 +45,8 @@
 struct sockaddr_extra_t
 {
 	hio_uint16_t chan;
+	hio_uint32_t ppid;
+	hio_int32_t assoc;
 };
 typedef struct sockaddr_extra_t sockaddr_extra_t;
 
@@ -57,6 +59,12 @@ typedef struct sockaddr_extra_t sockaddr_extra_t;
  * extra fields:
  *   chan - used as a stream number for SCTP PACKETSEQ sockets.
  *          use hio_skad_get_chan() and hio_skad_set_chan() for safe access.
+ *   ppid - the SCTP payload protocol identifier, opaque to SCTP itself and
+ *          passed through unchanged. use hio_skad_get_ppid()/set_ppid().
+ *   assoc - the SCTP association identifier (sctp_assoc_t, an int32 on the
+ *          systems that have it). the stable name for an association: its
+ *          addresses can change under multi-homing, this does not. use
+ *          hio_skad_get_assoc()/set_assoc().
  */
 struct sockaddr_in_x
 {
@@ -1475,6 +1483,68 @@ void hio_skad_set_scope_id (hio_skad_t* _skad, int scope_id)
 #endif
 }
 
+hio_int32_t hio_skad_get_assoc (const hio_skad_t* _skad)
+{
+	const hio_skad_alt_t* skad = (const hio_skad_alt_t*)_skad;
+
+	switch (skad->sa.sa_family)
+	{
+	#if defined(AF_INET) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN > 0)
+		case AF_INET: return skad->in4.x.assoc;
+	#endif
+	#if defined(AF_INET6) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN6 > 0)
+		case AF_INET6: return skad->in6.x.assoc;
+	#endif
+	}
+	return 0;
+}
+
+void hio_skad_set_assoc (hio_skad_t* _skad, hio_int32_t assoc)
+{
+	hio_skad_alt_t* skad = (hio_skad_alt_t*)_skad;
+
+	switch (skad->sa.sa_family)
+	{
+	#if defined(AF_INET) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN > 0)
+		case AF_INET: skad->in4.x.assoc = assoc; break;
+	#endif
+	#if defined(AF_INET6) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN6 > 0)
+		case AF_INET6: skad->in6.x.assoc = assoc; break;
+	#endif
+	}
+}
+
+hio_uint32_t hio_skad_get_ppid (const hio_skad_t* _skad)
+{
+	const hio_skad_alt_t* skad = (const hio_skad_alt_t*)_skad;
+
+	switch (skad->sa.sa_family)
+	{
+	#if defined(AF_INET) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN > 0)
+		case AF_INET: return skad->in4.x.ppid;
+	#endif
+	#if defined(AF_INET6) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN6 > 0)
+		case AF_INET6: return skad->in6.x.ppid;
+	#endif
+	}
+	return 0;
+}
+
+void hio_skad_set_ppid (hio_skad_t* _skad, hio_uint32_t ppid)
+{
+	hio_skad_alt_t* skad = (hio_skad_alt_t*)_skad;
+
+	switch (skad->sa.sa_family)
+	{
+	#if defined(AF_INET) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN > 0)
+		case AF_INET: skad->in4.x.ppid = ppid; break;
+	#endif
+	#if defined(AF_INET6) && (HIO_SIZEOF_STRUCT_SOCKADDR_IN6 > 0)
+		case AF_INET6: skad->in6.x.ppid = ppid; break;
+	#endif
+	}
+}
+
 hio_uint16_t hio_skad_get_chan (const hio_skad_t* _skad)
 {
 	const hio_skad_alt_t* skad = (const hio_skad_alt_t*)_skad;

t/Makefile.am

@@ -18,7 +18,7 @@ check_SCRIPTS = s-001.sh s-002.sh
 EXTRA_DIST = $(check_SCRIPTS) tap.inc t-cgi.sh
 
 ## programs run as tests
-TESTPROGS = t-001 t-002 t-003 t-004 t-005 t-006 t-007 t-008 t-009 t-010 t-011
+TESTPROGS = t-001 t-002 t-003 t-004 t-005 t-006 t-007 t-008 t-009 t-010 t-011 t-012
 
 ## helpers the test scripts drive. built with the tests, but not tests
 ## themselves - they are servers and would never exit.
@@ -99,6 +99,14 @@ t_011_CFLAGS = $(CFLAGS_COMMON)
 t_011_LDFLAGS = $(LDFLAGS_COMMON)
 t_011_LDADD = $(LIBADD_COMMON)
 
+## t-012 drives an sctp association. it skips itself where the build has no
+## sctp support, which it discovers the same way a caller would.
+t_012_SOURCES = t-012.c tap.h
+t_012_CPPFLAGS = $(CPPFLAGS_COMMON)
+t_012_CFLAGS = $(CFLAGS_COMMON)
+t_012_LDFLAGS = $(LDFLAGS_COMMON)
+t_012_LDADD = $(LIBADD_COMMON)
+
 ## a minimal FastCGI responder, driven by s-002.sh
 fcgis_SOURCES = fcgis.c
 fcgis_CPPFLAGS = $(CPPFLAGS_COMMON)

t/Makefile.in

@@ -108,7 +108,8 @@ CONFIG_CLEAN_FILES =
 CONFIG_CLEAN_VPATH_FILES =
 am__EXEEXT_1 = t-001$(EXEEXT) t-002$(EXEEXT) t-003$(EXEEXT) \
 	t-004$(EXEEXT) t-005$(EXEEXT) t-006$(EXEEXT) t-007$(EXEEXT) \
-	t-008$(EXEEXT) t-009$(EXEEXT) t-010$(EXEEXT) t-011$(EXEEXT)
+	t-008$(EXEEXT) t-009$(EXEEXT) t-010$(EXEEXT) t-011$(EXEEXT) \
+	t-012$(EXEEXT)
 am__EXEEXT_2 = fcgis$(EXEEXT) httpecho$(EXEEXT) httssvr$(EXEEXT) \
 	slowclient$(EXEEXT)
 am_fcgis_OBJECTS = fcgis-fcgis.$(OBJEXT)
@@ -207,6 +208,12 @@ t_011_DEPENDENCIES = $(am__DEPENDENCIES_2)
 t_011_LINK = $(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) \
 	$(LIBTOOLFLAGS) --mode=link $(CCLD) $(t_011_CFLAGS) $(CFLAGS) \
 	$(t_011_LDFLAGS) $(LDFLAGS) -o $@
+am_t_012_OBJECTS = t_012-t-012.$(OBJEXT)
+t_012_OBJECTS = $(am_t_012_OBJECTS)
+t_012_DEPENDENCIES = $(am__DEPENDENCIES_2)
+t_012_LINK = $(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) \
+	$(LIBTOOLFLAGS) --mode=link $(CCLD) $(t_012_CFLAGS) $(CFLAGS) \
+	$(t_012_LDFLAGS) $(LDFLAGS) -o $@
 AM_V_P = $(am__v_P_@AM_V@)
 am__v_P_ = $(am__v_P_@AM_DEFAULT_V@)
 am__v_P_0 = false
@@ -231,7 +238,7 @@ am__depfiles_remade = ./$(DEPDIR)/fcgis-fcgis.Po \
 	./$(DEPDIR)/t_005-t-005.Po ./$(DEPDIR)/t_006-t-006.Po \
 	./$(DEPDIR)/t_007-t-007.Po ./$(DEPDIR)/t_008-t-008.Po \
 	./$(DEPDIR)/t_009-t-009.Po ./$(DEPDIR)/t_010-t-010.Po \
-	./$(DEPDIR)/t_011-t-011.Po
+	./$(DEPDIR)/t_011-t-011.Po ./$(DEPDIR)/t_012-t-012.Po
 am__mv = mv -f
 COMPILE = $(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(AM_CPPFLAGS) \
 	$(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS)
@@ -255,12 +262,14 @@ SOURCES = $(fcgis_SOURCES) $(httpecho_SOURCES) $(httssvr_SOURCES) \
 	$(slowclient_SOURCES) $(t_001_SOURCES) $(t_002_SOURCES) \
 	$(t_003_SOURCES) $(t_004_SOURCES) $(t_005_SOURCES) \
 	$(t_006_SOURCES) $(t_007_SOURCES) $(t_008_SOURCES) \
-	$(t_009_SOURCES) $(t_010_SOURCES) $(t_011_SOURCES)
+	$(t_009_SOURCES) $(t_010_SOURCES) $(t_011_SOURCES) \
+	$(t_012_SOURCES)
 DIST_SOURCES = $(fcgis_SOURCES) $(httpecho_SOURCES) $(httssvr_SOURCES) \
 	$(slowclient_SOURCES) $(t_001_SOURCES) $(t_002_SOURCES) \
 	$(t_003_SOURCES) $(t_004_SOURCES) $(t_005_SOURCES) \
 	$(t_006_SOURCES) $(t_007_SOURCES) $(t_008_SOURCES) \
-	$(t_009_SOURCES) $(t_010_SOURCES) $(t_011_SOURCES)
+	$(t_009_SOURCES) $(t_010_SOURCES) $(t_011_SOURCES) \
+	$(t_012_SOURCES)
 am__can_run_installinfo = \
   case $$AM_UPDATE_INFO_DIR in \
     n|no|NO) false;; \
@@ -572,6 +581,7 @@ PTHREAD_CXX = @PTHREAD_CXX@
 PTHREAD_LIBS = @PTHREAD_LIBS@
 QUADMATH_LIBS = @QUADMATH_LIBS@
 RANLIB = @RANLIB@
+SCTP_ENABLED = @SCTP_ENABLED@
 SED = @SED@
 SENDFILE_LIBS = @SENDFILE_LIBS@
 SET_MAKE = @SET_MAKE@
@@ -650,7 +660,7 @@ LDFLAGS_COMMON = -L$(abs_builddir)/../lib  -L$(libdir)
 LIBADD_COMMON = ../lib/libhio.la $(LIBM)
 check_SCRIPTS = s-001.sh s-002.sh
 EXTRA_DIST = $(check_SCRIPTS) tap.inc t-cgi.sh
-TESTPROGS = t-001 t-002 t-003 t-004 t-005 t-006 t-007 t-008 t-009 t-010 t-011
+TESTPROGS = t-001 t-002 t-003 t-004 t-005 t-006 t-007 t-008 t-009 t-010 t-011 t-012
 HELPERPROGS = fcgis httpecho httssvr slowclient
 t_001_SOURCES = t-001.c tap.h
 t_001_CPPFLAGS = $(CPPFLAGS_COMMON)
@@ -707,6 +717,11 @@ t_011_CPPFLAGS = $(CPPFLAGS_COMMON)
 t_011_CFLAGS = $(CFLAGS_COMMON)
 t_011_LDFLAGS = $(LDFLAGS_COMMON)
 t_011_LDADD = $(LIBADD_COMMON)
+t_012_SOURCES = t-012.c tap.h
+t_012_CPPFLAGS = $(CPPFLAGS_COMMON)
+t_012_CFLAGS = $(CFLAGS_COMMON)
+t_012_LDFLAGS = $(LDFLAGS_COMMON)
+t_012_LDADD = $(LIBADD_COMMON)
 fcgis_SOURCES = fcgis.c
 fcgis_CPPFLAGS = $(CPPFLAGS_COMMON)
 fcgis_CFLAGS = $(CFLAGS_COMMON)
@@ -830,6 +845,10 @@ t-011$(EXEEXT): $(t_011_OBJECTS) $(t_011_DEPENDENCIES) $(EXTRA_t_011_DEPENDENCIE
 	@rm -f t-011$(EXEEXT)
 	$(AM_V_CCLD)$(t_011_LINK) $(t_011_OBJECTS) $(t_011_LDADD) $(LIBS)
 
+t-012$(EXEEXT): $(t_012_OBJECTS) $(t_012_DEPENDENCIES) $(EXTRA_t_012_DEPENDENCIES) 
+	@rm -f t-012$(EXEEXT)
+	$(AM_V_CCLD)$(t_012_LINK) $(t_012_OBJECTS) $(t_012_LDADD) $(LIBS)
+
 mostlyclean-compile:
 	-rm -f *.$(OBJEXT)
 
@@ -851,6 +870,7 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/t_009-t-009.Po@am__quote@ # am--include-marker
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/t_010-t-010.Po@am__quote@ # am--include-marker
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/t_011-t-011.Po@am__quote@ # am--include-marker
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/t_012-t-012.Po@am__quote@ # am--include-marker
 
 $(am__depfiles_remade):
 	@$(MKDIR_P) $(@D)
@@ -1092,6 +1112,20 @@ t_011-t-011.obj: t-011.c
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
 @am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(t_011_CPPFLAGS) $(CPPFLAGS) $(t_011_CFLAGS) $(CFLAGS) -c -o t_011-t-011.obj `if test -f 't-011.c'; then $(CYGPATH_W) 't-011.c'; else $(CYGPATH_W) '$(srcdir)/t-011.c'; fi`
 
+t_012-t-012.o: t-012.c
+@am__fastdepCC_TRUE@	$(AM_V_CC)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(t_012_CPPFLAGS) $(CPPFLAGS) $(t_012_CFLAGS) $(CFLAGS) -MT t_012-t-012.o -MD -MP -MF $(DEPDIR)/t_012-t-012.Tpo -c -o t_012-t-012.o `test -f 't-012.c' || echo '$(srcdir)/'`t-012.c
+@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/t_012-t-012.Tpo $(DEPDIR)/t_012-t-012.Po
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	$(AM_V_CC)source='t-012.c' object='t_012-t-012.o' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(t_012_CPPFLAGS) $(CPPFLAGS) $(t_012_CFLAGS) $(CFLAGS) -c -o t_012-t-012.o `test -f 't-012.c' || echo '$(srcdir)/'`t-012.c
+
+t_012-t-012.obj: t-012.c
+@am__fastdepCC_TRUE@	$(AM_V_CC)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(t_012_CPPFLAGS) $(CPPFLAGS) $(t_012_CFLAGS) $(CFLAGS) -MT t_012-t-012.obj -MD -MP -MF $(DEPDIR)/t_012-t-012.Tpo -c -o t_012-t-012.obj `if test -f 't-012.c'; then $(CYGPATH_W) 't-012.c'; else $(CYGPATH_W) '$(srcdir)/t-012.c'; fi`
+@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/t_012-t-012.Tpo $(DEPDIR)/t_012-t-012.Po
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	$(AM_V_CC)source='t-012.c' object='t_012-t-012.obj' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(t_012_CPPFLAGS) $(CPPFLAGS) $(t_012_CFLAGS) $(CFLAGS) -c -o t_012-t-012.obj `if test -f 't-012.c'; then $(CYGPATH_W) 't-012.c'; else $(CYGPATH_W) '$(srcdir)/t-012.c'; fi`
+
 mostlyclean-libtool:
 	-rm -f *.lo
 
@@ -1393,6 +1427,13 @@ t-011.log: t-011$(EXEEXT)
 	--log-file $$b.log --trs-file $$b.trs \
 	$(am__common_driver_flags) $(AM_LOG_DRIVER_FLAGS) $(LOG_DRIVER_FLAGS) -- $(LOG_COMPILE) \
 	"$$tst" $(AM_TESTS_FD_REDIRECT)
+t-012.log: t-012$(EXEEXT)
+	@p='t-012$(EXEEXT)'; \
+	b='t-012'; \
+	$(am__check_pre) $(LOG_DRIVER) --test-name "$$f" \
+	--log-file $$b.log --trs-file $$b.trs \
+	$(am__common_driver_flags) $(AM_LOG_DRIVER_FLAGS) $(LOG_DRIVER_FLAGS) -- $(LOG_COMPILE) \
+	"$$tst" $(AM_TESTS_FD_REDIRECT)
 .sh.log:
 	@p='$<'; \
 	$(am__set_b); \
@@ -1502,6 +1543,7 @@ distclean: distclean-am
 	-rm -f ./$(DEPDIR)/t_009-t-009.Po
 	-rm -f ./$(DEPDIR)/t_010-t-010.Po
 	-rm -f ./$(DEPDIR)/t_011-t-011.Po
+	-rm -f ./$(DEPDIR)/t_012-t-012.Po
 	-rm -f Makefile
 distclean-am: clean-am distclean-compile distclean-generic \
 	distclean-tags
@@ -1562,6 +1604,7 @@ maintainer-clean: maintainer-clean-am
 	-rm -f ./$(DEPDIR)/t_009-t-009.Po
 	-rm -f ./$(DEPDIR)/t_010-t-010.Po
 	-rm -f ./$(DEPDIR)/t_011-t-011.Po
+	-rm -f ./$(DEPDIR)/t_012-t-012.Po
 	-rm -f Makefile
 maintainer-clean-am: distclean-am maintainer-clean-generic
 

t/t-008-wq.inc

@@ -133,3 +133,159 @@ static void test_wq_limit_zero_is_unlimited (void)
 	hio_dev_kill ((hio_dev_t*)dev);
 	close (peerfd);
 }
+
+/* ------------------------------------------------------------------ */
+/* completion firing inside the greedy read loop.
+ *
+ * when on_read asks for another read, the loop fires the completions queued
+ * so far before calling on_read again - otherwise a write started from inside
+ * on_read would be reported after a later on_read, which is the reordering
+ * the comment in handle_event() warns about. the firing is per device, so an
+ * unrelated connection's callbacks are not dragged into this device's pass. */
+
+static void test_completion_fires_within_read_loop (void)
+{
+	tdev_t* dev;
+	int peerfd;
+	hio_uint8_t chunk[30];
+
+	obs_reset ();
+	g_wr_chunk = 0;
+	g_wr_calls_left = 100;   /* every write completes at once */
+	g_rd_chunk = 10;         /* so 30 octets take three read iterations */
+	g_write_on_read = 1;
+
+	dev = make_tdev(&peerfd);
+	if (!dev) { skip ("device creation failed", 3); return; }
+
+	HIO_MEMSET (chunk, 'x', HIO_SIZEOF(chunk));
+	if (write(peerfd, chunk, HIO_SIZEOF(chunk)) != (ssize_t)HIO_SIZEOF(chunk))
+	{
+		skip ("peer write failed", 3);
+		hio_dev_kill ((hio_dev_t*)dev);
+		close (peerfd);
+		return;
+	}
+
+	pump ();
+
+	/* iteration 1 reads and writes; iteration 2 fires that write's completion
+	 * before reading again. hence R, then W and R alternating. */
+	OK (g_seq_n >= 3, "the greedy read loop ran more than once");
+	OK (g_seq[0] == 'R', "no completion is fired before the first read");
+	OK (strncmp(g_seq, "RWRWR", 5) == 0,
+	    "each iteration's write completion is reported before the next on_read");
+
+	g_write_on_read = 0;
+	hio_dev_kill ((hio_dev_t*)dev);
+	close (peerfd);
+}
+
+/* ------------------------------------------------------------------ */
+/* the shared read buffer's configurable size.
+ *
+ * every device reads through one buffer owned by the loop. its size decides
+ * how much a single read can take, so it is the ceiling on read size for a
+ * bulk transfer - and it is memory held per loop, which matters when a
+ * process runs one loop per thread. */
+
+static void test_read_buffer_size_option (void)
+{
+	hio_oow_t capa = 0, want;
+	int rc;
+
+	obs_reset ();
+
+	OK (hio_getoption(g_hio, HIO_READ_BUFFER_SIZE, &capa) == 0 && capa == HIO_DFL_READ_BUFFER_SIZE,
+	    "the read buffer starts at HIO_DFL_READ_BUFFER_SIZE");
+
+	want = HIO_MIN_READ_BUFFER_SIZE - 1;
+	rc = hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &want);
+	OK (rc <= -1 && hio_geterrnum(g_hio) == HIO_EINVAL,
+	    "a size below HIO_MIN_READ_BUFFER_SIZE is refused with HIO_EINVAL");
+
+	OK (hio_getoption(g_hio, HIO_READ_BUFFER_SIZE, &capa) == 0 && capa == HIO_DFL_READ_BUFFER_SIZE,
+	    "a refused set leaves the existing buffer alone");
+
+	want = HIO_DFL_READ_BUFFER_SIZE * 4;
+	OK (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &want) == 0 &&
+	    hio_getoption(g_hio, HIO_READ_BUFFER_SIZE, &capa) == 0 && capa == want,
+	    "a larger size is accepted and reads back");
+
+	want = HIO_DFL_READ_BUFFER_SIZE;
+	OK (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &want) == 0, "and it can be set back");
+}
+
+static void test_read_buffer_size_is_honoured (void)
+{
+	/* the size has to reach the read path, not merely be stored. the stub
+	 * device records the length the core offers it. */
+	tdev_t* dev;
+	int peerfd;
+	hio_oow_t want, restore = HIO_DFL_READ_BUFFER_SIZE;
+	hio_uint8_t buf[4096];
+
+	obs_reset ();
+	g_wr_calls_left = 100;
+
+	want = HIO_MIN_READ_BUFFER_SIZE;   /* the smallest the core allows */
+	if (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &want) <= -1) { skip ("cannot set the buffer size", 2); return; }
+
+	dev = make_tdev(&peerfd);
+	if (!dev) { skip ("device creation failed", 2); hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &restore); return; }
+
+	HIO_MEMSET (buf, 'z', HIO_SIZEOF(buf));
+	if (write(peerfd, buf, HIO_SIZEOF(buf)) != (ssize_t)HIO_SIZEOF(buf))
+	{
+		skip ("peer write failed", 2);
+		hio_dev_kill ((hio_dev_t*)dev);
+		close (peerfd);
+		hio_setoption (g_hio, HIO_READ_BUFFER_SIZE, &restore);
+		return;
+	}
+
+	pump ();
+
+	OK (g_rd_calls > 0, "the device was read from");
+	OK (g_rd_offered == (hio_iolen_t)HIO_MIN_READ_BUFFER_SIZE,
+	    "the core offers the read method exactly the configured buffer size");
+
+	hio_dev_kill ((hio_dev_t*)dev);
+	close (peerfd);
+	hio_setoption (g_hio, HIO_READ_BUFFER_SIZE, &restore);
+}
+
+static void test_read_buffer_resize_refused_in_callback (void)
+{
+	/* on_read receives a pointer into the buffer. resizing it there would
+	 * leave that pointer dangling, so the attempt has to fail. */
+	tdev_t* dev;
+	int peerfd;
+	hio_uint8_t buf[64];
+
+	obs_reset ();
+	g_wr_calls_left = 100;
+	g_setopt_from_on_read = 1;
+
+	dev = make_tdev(&peerfd);
+	if (!dev) { skip ("device creation failed", 2); return; }
+
+	HIO_MEMSET (buf, 'q', HIO_SIZEOF(buf));
+	if (write(peerfd, buf, HIO_SIZEOF(buf)) != (ssize_t)HIO_SIZEOF(buf))
+	{
+		skip ("peer write failed", 2);
+		hio_dev_kill ((hio_dev_t*)dev);
+		close (peerfd);
+		return;
+	}
+
+	pump ();
+
+	OK (g_setopt_rc <= -1 && g_setopt_errnum == HIO_EBUSY,
+	    "resizing the read buffer from inside on_read fails with HIO_EBUSY");
+	OK (hio_getoption(g_hio, HIO_READ_BUFFER_SIZE, &(hio_oow_t){0}) == 0,
+	    "and the option is still readable afterwards");
+
+	hio_dev_kill ((hio_dev_t*)dev);
+	close (peerfd);
+}

t/t-008.c

@@ -34,12 +34,41 @@ static hio_iolen_t g_cw_len[MAX_CW];
 static void*       g_cw_ctx[MAX_CW];
 static int         g_cw_n = 0;
 
+/* the greedy-read interleaving cases below. g_rd_chunk caps how much one read
+ * delivers; g_write_on_read makes on_read issue a write and ask for another
+ * read, which is the only shape that reaches the completion firing inside the
+ * read loop. g_seq records 'R' per on_read and 'W' per on_write. */
+static hio_iolen_t g_rd_chunk = 0;
+static hio_iolen_t g_rd_offered = 0;   /* the buffer size the core last offered a read */
+static int         g_rd_calls = 0;
+static int         g_setopt_from_on_read = 0;
+static int         g_setopt_rc = 0;
+static int         g_setopt_errnum = 0;
+static int         g_write_on_read = 0;
+static hio_bch_t   g_seq[64];
+static int         g_seq_n = 0;
+
+static void seq_put (hio_bch_t c)
+{
+	if (g_seq_n < (int)HIO_COUNTOF(g_seq) - 1) g_seq[g_seq_n++] = c;
+	g_seq[g_seq_n] = '\0';
+}
+
 static hio_uint8_t g_pattern[PATLEN];
 
 static void obs_reset (void)
 {
 	g_wr_calls = 0;
 	g_cw_n = 0;
+	g_rd_chunk = 0;
+	g_rd_offered = 0;
+	g_rd_calls = 0;
+	g_setopt_from_on_read = 0;
+	g_setopt_rc = 0;
+	g_setopt_errnum = 0;
+	g_write_on_read = 0;
+	g_seq_n = 0;
+	g_seq[0] = '\0';
 }
 
 /* ------------------------------------------------------------------ */
@@ -73,7 +102,11 @@ static hio_syshnd_t tdev_getsyshnd (hio_dev_t* dev)
 
 static int tdev_read (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_devaddr_t* srcaddr)
 {
-	ssize_t n = recv(((tdev_t*)dev)->fd, buf, *len, 0);
+	ssize_t n;
+	g_rd_offered = *len;   /* what the core is willing to take in one go */
+	g_rd_calls++;
+	if (g_rd_chunk > 0 && *len > g_rd_chunk) *len = g_rd_chunk; /* force several iterations */
+	n = recv(((tdev_t*)dev)->fd, buf, *len, 0);
 	if (n <= -1)
 	{
 		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return 0;
@@ -206,11 +239,29 @@ static hio_dev_mth_t tdev_mth =
 
 static int tdev_on_read (hio_dev_t* dev, const void* data, hio_iolen_t len, const hio_devaddr_t* srcaddr)
 {
-	return 0;
+	if (g_setopt_from_on_read)
+	{
+		/* the data pointer we were just handed points into the buffer being
+		 * resized, so this must be refused rather than honoured */
+		hio_oow_t want = HIO_DFL_READ_BUFFER_SIZE * 2;
+		g_setopt_from_on_read = 0;
+		g_setopt_rc = hio_setoption(dev->hio, HIO_READ_BUFFER_SIZE, &want);
+		g_setopt_errnum = hio_geterrnum(dev->hio);
+	}
+
+	if (!g_write_on_read) return 0;
+
+	seq_put ('R');
+	if (len <= 0) return 0;
+	/* completes straight away, so a completion is queued for the next
+	 * iteration of the read loop to fire */
+	if (hio_dev_write(dev, g_pattern, 1, HIO_NULL, HIO_NULL) <= -1) return -1;
+	return 1; /* be greedy - keep reading in this same pass */
 }
 
 static int tdev_on_write (hio_dev_t* dev, hio_iolen_t wrlen, void* wrctx, const hio_devaddr_t* dstaddr)
 {
+	if (g_write_on_read) seq_put ('W');
 	if (g_cw_n < MAX_CW)
 	{
 		g_cw_len[g_cw_n] = wrlen;
@@ -742,6 +793,10 @@ int main (void)
 	test_wq_size_released_on_kill ();
 	test_wq_limit ();
 	test_wq_limit_zero_is_unlimited ();
+	test_completion_fires_within_read_loop ();
+	test_read_buffer_size_option ();
+	test_read_buffer_size_is_honoured ();
+	test_read_buffer_resize_refused_in_callback ();
 
 	hio_close (g_hio);
 	return exit_status();

t/t-011.c

@@ -11,6 +11,10 @@
  * is. the race is made deterministic here: the peer is accepted and written
  * to before the loop is ever run, so the first wakeup is guaranteed to carry
  * both bits.
+ *
+ * the second group of cases covers what a socket device asks of the loop's
+ * shared read buffer, which is the other place the socket layer and the core
+ * have to agree on something.
  */
 
 #include <hio-sck.h>
@@ -30,6 +34,8 @@
 static hio_t* g_hio = HIO_NULL;
 static int g_connected;     /* on_connect saw HIO_DEV_SCK_CONNECTED */
 static int g_disconnected;
+static int g_accepted;      /* on_connect saw an ACCEPTED device */
+static hio_dev_sck_t* g_qx; /* cleared by on_disconnect - the device is freed with it */
 static int g_read_len;      /* octets delivered to on_read */
 static hio_bch_t g_read_buf[128];
 static int g_done;
@@ -67,6 +73,7 @@ static int cli_on_write (hio_dev_sck_t* sck, hio_iolen_t wrlen, void* wrctx, con
 static void cli_on_connect (hio_dev_sck_t* sck)
 {
 	if (sck->state & HIO_DEV_SCK_CONNECTED) g_connected = 1;
+	if (sck->state & HIO_DEV_SCK_ACCEPTED) g_accepted = 1;
 }
 
 static void cli_on_disconnect (hio_dev_sck_t* sck)
@@ -74,6 +81,7 @@ static void cli_on_disconnect (hio_dev_sck_t* sck)
 	/* reached when the device is killed - which is what the defect did */
 	g_disconnected = 1;
 	g_done = 1;
+	if (sck == g_qx) g_qx = HIO_NULL;
 }
 
 /* a plain listener, so nothing about the peer depends on the code under test */
@@ -176,6 +184,165 @@ static void test_connect_with_data_already_waiting (void)
 	close (lfd);
 }
 
+/* the read buffer is shared by every device on the loop, so a device that can
+ * only work with a whole message at a time has to say so - what will not fit
+ * in one read is discarded by the kernel with nothing reported. a stream
+ * device has no such need and asks for nothing. */
+static void test_device_read_buffer_requirement (void)
+{
+	hio_dev_sck_make_t mi;
+	hio_dev_sck_t* d;
+	hio_oow_t small = HIO_MIN_READ_BUFFER_SIZE, dfl = HIO_DFL_READ_BUFFER_SIZE;
+
+	HIO_MEMSET (&mi, 0, HIO_SIZEOF(mi));
+	mi.on_read = cli_on_read;
+	mi.on_write = cli_on_write;
+	mi.on_connect = cli_on_connect;
+	mi.on_disconnect = cli_on_disconnect;
+
+	mi.type = HIO_DEV_SCK_TCP4;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (d && d->dev_rdmin == 0, "a stream socket asks for no particular read buffer size");
+	if (d) hio_dev_sck_kill (d);
+
+	mi.type = HIO_DEV_SCK_UDP4;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (d && d->dev_rdmin == HIO_DGRAM_READ_BUFFER_SIZE,
+	    "a datagram socket asks for enough to hold the largest datagram");
+	if (d) hio_dev_sck_kill (d);
+
+	/* the qx channel is message-oriented too, but its message is a fixed
+	 * struct - it must not be made to demand a datagram's worth */
+	mi.type = HIO_DEV_SCK_QX;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (d && d->dev_rdmin > 0 && d->dev_rdmin < HIO_MIN_READ_BUFFER_SIZE,
+	    "the qx channel asks only for its own message size");
+	if (d) hio_dev_sck_kill (d);
+
+	if (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &small) <= -1) { skip ("cannot shrink the read buffer", 3); return; }
+
+	mi.type = HIO_DEV_SCK_UDP4;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (!d && hio_geterrnum(g_hio) == HIO_ENOCAPA,
+	    "a datagram socket refuses to start when the read buffer is too small");
+	if (d) hio_dev_sck_kill (d);
+
+	/* ...while the types that do not need it are unaffected, which is what
+	 * makes the option usable at all - an http service uses a qx channel */
+	mi.type = HIO_DEV_SCK_TCP4;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (d != HIO_NULL, "a stream socket still starts on the smaller buffer");
+	if (d) hio_dev_sck_kill (d);
+
+	mi.type = HIO_DEV_SCK_QX;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	OK (d != HIO_NULL, "and so does the qx channel");
+	if (d) hio_dev_sck_kill (d);
+
+	hio_setoption (g_hio, HIO_READ_BUFFER_SIZE, &dfl);
+
+	/* and a live device's requirement cannot be pulled out from under it */
+	mi.type = HIO_DEV_SCK_UDP4;
+	d = hio_dev_sck_make(g_hio, 0, &mi);
+	if (d)
+	{
+		OK (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &small) <= -1 && hio_geterrnum(g_hio) == HIO_EPERM,
+		    "shrinking below a running device's requirement is refused");
+		hio_dev_sck_kill (d);
+	}
+	else skip ("cannot make a udp device", 1);
+
+	hio_setoption (g_hio, HIO_READ_BUFFER_SIZE, &dfl);
+}
+
+/* the accepted-client path takes its socket type from elsewhere - for the qx
+ * channel, from a message another thread sent - so it cannot rely on a local
+ * listener of that type having already vouched for the read buffer. this hands
+ * the qx channel a datagram socket while the buffer is too small for one.
+ *
+ * it is the only way to reach that path today, since every listenable type in
+ * sck_type_map happens to be a stream type. that is what makes the check worth
+ * having rather than obvious. */
+static int handover_via_qx (hio_dev_sck_t* qx, hio_dev_sck_type_t type)
+{
+	hio_dev_sck_qxmsg_t msg;
+	int fd;
+
+	fd = socket(AF_INET, SOCK_DGRAM, 0);
+	if (fd <= -1) return -1;
+
+	HIO_MEMSET (&msg, 0, HIO_SIZEOF(msg));
+	msg.cmd = HIO_DEV_SCK_QXMSG_NEWCONN;
+	msg.scktype = type;
+	msg.syshnd = fd;
+	if (hio_bcstrtoskad(g_hio, "127.0.0.1:1", &msg.remoteaddr) <= -1) { close(fd); return -1; }
+
+	/* the qx device reads from its own handle; the side channel is the end a
+	 * producer writes to */
+	if (write(qx->side_chan, &msg, HIO_SIZEOF(msg)) != (ssize_t)HIO_SIZEOF(msg)) { close(fd); return -1; }
+	return 0;
+}
+
+/* returns 1 if a device was accepted, 0 if not, -1 if the setup failed */
+static int try_qx_handover (hio_oow_t bufsz, hio_dev_sck_type_t type)
+{
+	hio_dev_sck_make_t mi;
+	hio_dev_sck_t* qx;
+	int i, r;
+
+	g_qx = HIO_NULL;
+
+	if (hio_setoption(g_hio, HIO_READ_BUFFER_SIZE, &bufsz) <= -1) return -1;
+
+	HIO_MEMSET (&mi, 0, HIO_SIZEOF(mi));
+	mi.type = HIO_DEV_SCK_QX;
+	mi.on_read = cli_on_read;
+	mi.on_write = cli_on_write;
+	mi.on_connect = cli_on_connect;
+	mi.on_disconnect = cli_on_disconnect;
+
+	/* a fresh channel each time - a refused handover halts the one that
+	 * carried it, so it cannot be reused for the next attempt */
+	qx = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!qx) return -1;
+	g_qx = qx;
+
+	g_accepted = 0;
+	if (handover_via_qx(qx, type) <= -1) r = -1;
+	else
+	{
+		for (i = 0; i < 20 && !g_accepted; i++) hio_exec(g_hio);
+		r = g_accepted;
+	}
+
+	/* a refused handover halts this channel, and the loop above will already
+	 * have reaped it - so the pointer cannot be dereferenced to find out.
+	 * on_disconnect clears g_qx, which is the only safe way to ask. */
+	if (g_qx) { hio_dev_sck_kill (g_qx); g_qx = HIO_NULL; }
+	hio_exec (g_hio);
+	return r;
+}
+
+static void test_accepted_device_read_buffer_requirement (void)
+{
+	hio_oow_t dfl = HIO_DFL_READ_BUFFER_SIZE;
+	int refused, allowed;
+
+	refused = try_qx_handover(HIO_MIN_READ_BUFFER_SIZE, HIO_DEV_SCK_UDP4);
+	allowed = try_qx_handover(HIO_DFL_READ_BUFFER_SIZE, HIO_DEV_SCK_UDP4);
+
+	if (refused <= -1 || allowed <= -1) skip ("qx handover setup failed", 2);
+	else
+	{
+		OK (refused == 0, "a datagram socket handed over the qx channel is refused when the buffer is too small");
+		/* the control: the same handover with room for it must succeed, or the
+		 * case above proves nothing about the buffer check */
+		OK (allowed == 1, "and the same handover succeeds once the buffer is large enough");
+	}
+
+	hio_setoption (g_hio, HIO_READ_BUFFER_SIZE, &dfl);
+}
+
 int main (void)
 {
 	hio_errinf_t errinf;
@@ -191,6 +358,8 @@ int main (void)
 	quiet_logging (g_hio);
 
 	test_connect_with_data_already_waiting ();
+	test_device_read_buffer_requirement ();
+	test_accepted_device_read_buffer_requirement ();
 
 	hio_close (g_hio);
 	return exit_status();

t/t-012.c

@@ -0,0 +1,473 @@
+/*
+ * sctp transport tests.
+ *
+ * these cover the one-to-one types, HIO_DEV_SCK_SCTP4 and its v6 twin. they
+ * are stream devices like tcp, but their reads and writes go through
+ * sendmsg()/recvmsg() so that the per-message ancillary data is reachable -
+ * which is the only reason to prefer sctp here in the first place.
+ *
+ * what is checked: an association forms and carries data; a stream number and
+ * a ppid set on the destination arrive at the far end; notifications are
+ * delivered to on_notification() and never spliced into the data stream; and a
+ * zero-length write still means "close the writing end" rather than "send an
+ * empty message".
+ *
+ * the second group covers the one-to-many (SOCK_SEQPACKET) types. those are
+ * not accept-based: one device carries every association, and a message is
+ * attributed by the source address and the association id it arrives with.
+ *
+ * the whole file skips where the system has no sctp - the library reports that
+ * as HIO_ENOIMPL from hio_dev_sck_make(), which is also worth asserting, since
+ * it is how a caller is meant to find out.
+ */
+
+#include <hio-sck.h>
+#include <hio-prv.h>
+#include "tap.h"
+
+#include <string.h>
+
+/* deliberately beyond what a system gives an association by default - linux
+ * negotiates ten outbound streams unless asked otherwise. writing on stream
+ * 40 therefore only works if SCTP_INITMSG really was set, which is the point:
+ * with the old call, which handed a struct sctp_initmsg to SCTP_EVENTS, the
+ * request silently did nothing and a stream this high is rejected. */
+#define OSTREAMS 64
+#define TEST_STREAM 40
+#define TEST_PPID 0xC0FFEEu
+#define PAYLOAD "sctp-payload"
+
+static hio_t* g_hio = HIO_NULL;
+static hio_dev_sck_t* g_srv = HIO_NULL;
+static hio_dev_sck_t* g_acc = HIO_NULL;
+static hio_dev_sck_t* g_cli = HIO_NULL;
+
+static int g_connected, g_accepted, g_timeout;
+static int g_got_data;              /* payload arrived */
+static int g_notifications;         /* on_notification calls */
+static int g_eof;                   /* the accepted side saw a zero-length read */
+static int g_reads;
+static hio_bch_t g_rbuf[256];
+static int g_rlen;
+static hio_uint16_t g_rchan;        /* stream number the receiver saw */
+static hio_uint32_t g_rppid;
+static hio_tmridx_t g_deadline = HIO_TMRIDX_INVALID;
+
+static void quiet_logging (hio_t* hio)
+{
+	hio_bitmask_t mask = HIO_LOG_ERROR | HIO_LOG_FATAL | HIO_LOG_ALL_TYPES;
+	hio_setoption (hio, HIO_LOG_MASK, &mask);
+}
+
+static void on_deadline (hio_t* hio, const hio_ntime_t* now, hio_tmrjob_t* job)
+{
+	g_timeout = 1;
+}
+
+/* ------------------------------------------------------------------ */
+
+static int srv_on_read (hio_dev_sck_t* sck, const void* data, hio_iolen_t dlen, const hio_skad_t* srcaddr)
+{
+	if (dlen <= 0)
+	{
+		/* the client shut its writing end - EOF, as for any stream device */
+		g_eof = 1;
+		return 0;
+	}
+
+	g_reads++;
+	g_got_data = 1;
+	if (dlen > (hio_iolen_t)HIO_SIZEOF(g_rbuf) - 1) dlen = HIO_SIZEOF(g_rbuf) - 1;
+	HIO_MEMCPY (g_rbuf, data, dlen);
+	g_rbuf[dlen] = '\0';
+	g_rlen = (int)dlen;
+
+	/* the ancillary data rides on the source address */
+	g_rchan = hio_skad_get_chan(srcaddr? srcaddr: &sck->remoteaddr);
+	g_rppid = hio_skad_get_ppid(srcaddr? srcaddr: &sck->remoteaddr);
+	return 0;
+}
+
+static int on_write (hio_dev_sck_t* sck, hio_iolen_t wrlen, void* wrctx, const hio_skad_t* dstaddr)
+{
+	return 0;
+}
+
+static void on_notification (hio_dev_sck_t* sck, const void* data, hio_iolen_t dlen)
+{
+	/* an association or path event. it must never reach on_read(). */
+	g_notifications++;
+}
+
+static void srv_on_connect (hio_dev_sck_t* sck)
+{
+	if (sck->state & HIO_DEV_SCK_ACCEPTED)
+	{
+		g_accepted = 1;
+		g_acc = sck;
+	}
+}
+
+static void srv_on_disconnect (hio_dev_sck_t* sck)
+{
+	if (sck == g_acc) g_acc = HIO_NULL;
+	else if (sck == g_srv) g_srv = HIO_NULL;
+}
+
+static int cli_on_read (hio_dev_sck_t* sck, const void* data, hio_iolen_t dlen, const hio_skad_t* srcaddr)
+{
+	return 0;
+}
+
+static void cli_on_connect (hio_dev_sck_t* sck)
+{
+	if (sck->state & HIO_DEV_SCK_CONNECTED) g_connected = 1;
+}
+
+static void cli_on_disconnect (hio_dev_sck_t* sck)
+{
+	if (sck == g_cli) g_cli = HIO_NULL;
+}
+
+/* ------------------------------------------------------------------ */
+
+static void fill_make (hio_dev_sck_make_t* mi, int server)
+{
+	HIO_MEMSET (mi, 0, HIO_SIZEOF(*mi));
+	mi->type = HIO_DEV_SCK_SCTP4;
+	mi->on_write = on_write;
+	mi->on_read = server? srv_on_read: cli_on_read;
+	mi->on_connect = server? srv_on_connect: cli_on_connect;
+	mi->on_disconnect = server? srv_on_disconnect: cli_on_disconnect;
+	mi->on_notification = on_notification;
+	/* ask for more streams than the system default. the negotiated count is
+	 * min(our sinit_num_ostreams, the peer's sinit_max_instreams), so both
+	 * ends have to ask. */
+	mi->sctp_ostreams = OSTREAMS;
+	mi->sctp_instreams = OSTREAMS;
+}
+
+/* run the loop until the flag is set, or five seconds pass. the timeout is
+ * cleared on entry: a phase that legitimately waits must not be starved by an
+ * earlier phase having already used the budget up. */
+static void run_until (int* flag)
+{
+	hio_tmrjob_t j;
+
+	g_timeout = 0;
+	HIO_MEMSET (&j, 0, HIO_SIZEOF(j));
+	hio_gettime (g_hio, &j.when);
+	j.when.sec += 5;
+	j.handler = on_deadline;
+	j.idxptr = &g_deadline;
+	g_deadline = hio_instmrjob(g_hio, &j);
+
+	while (!*flag && !g_timeout)
+	{
+		if (hio_exec(g_hio) <= -1) break;
+	}
+
+	if (g_deadline != HIO_TMRIDX_INVALID) { hio_deltmrjob (g_hio, g_deadline); g_deadline = HIO_TMRIDX_INVALID; }
+}
+
+static void teardown (void)
+{
+	if (g_cli) hio_dev_sck_halt (g_cli);
+	if (g_acc) hio_dev_sck_halt (g_acc);
+	if (g_srv) hio_dev_sck_halt (g_srv);
+	hio_exec (g_hio);
+	hio_exec (g_hio);
+	g_cli = g_acc = g_srv = HIO_NULL;
+}
+
+/* ------------------------------------------------------------------ */
+
+static void test_association_and_ancillary (void)
+{
+	hio_dev_sck_make_t mi;
+	hio_dev_sck_bind_t bi;
+	hio_dev_sck_listen_t li;
+	hio_dev_sck_connect_t ci;
+	hio_skad_t peer;
+
+	g_connected = g_accepted = g_timeout = g_got_data = 0;
+	g_notifications = g_eof = g_reads = g_rlen = 0;
+	g_rchan = 0; g_rppid = 0;
+
+	fill_make (&mi, 1);
+	g_srv = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!g_srv) { skip ("cannot make an sctp server device", 8); return; }
+
+	HIO_MEMSET (&bi, 0, HIO_SIZEOF(bi));
+	if (hio_bcstrtoskad(g_hio, "127.0.0.1:0", &bi.localaddr) <= -1) { skip ("bad address", 8); teardown(); return; }
+	bi.options = HIO_DEV_SCK_BIND_REUSEADDR;
+	OK (hio_dev_sck_bind(g_srv, &bi) == 0, "an sctp socket binds");
+
+	HIO_MEMSET (&li, 0, HIO_SIZEOF(li));
+	li.backlogs = 4;
+	HIO_INIT_NTIME (&li.accept_tmout, -1, 0);
+	OK (hio_dev_sck_listen(g_srv, &li) == 0, "and listens");
+
+	if (hio_dev_sck_getsockaddr(g_srv, &peer) <= -1) { skip ("cannot read the bound address", 6); teardown(); return; }
+
+	fill_make (&mi, 0);
+	g_cli = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!g_cli) { skip ("cannot make an sctp client device", 6); teardown(); return; }
+
+	HIO_MEMSET (&ci, 0, HIO_SIZEOF(ci));
+	ci.remoteaddr = peer;
+	HIO_INIT_NTIME (&ci.connect_tmout, 5, 0);
+	if (hio_dev_sck_connect(g_cli, &ci) <= -1) { skip ("connect failed to start", 6); teardown(); return; }
+
+	run_until (&g_accepted);
+	OK (g_connected, "the client reaches the CONNECTED state");
+	OK (g_accepted, "the server is handed an ACCEPTED association");
+
+	if (!g_accepted) { skip ("no association", 4); teardown(); return; }
+
+	/* write on a specific stream with a specific ppid. both ride on the
+	 * destination address rather than on a separate argument, which is why
+	 * hio_skad_t carries an extra area at all. */
+	{
+		hio_skad_t dst = peer;
+		hio_skad_set_chan (&dst, TEST_STREAM);
+		hio_skad_set_ppid (&dst, TEST_PPID);
+		/* [NOTE] a failure here is a failure, not a missing prerequisite -
+		 * skipping would hide exactly what this case exists to catch. writing
+		 * on a stream above the system default is rejected outright unless
+		 * SCTP_INITMSG asked for the streams. */
+		if (hio_dev_sck_write(g_cli, PAYLOAD, HIO_SIZEOF(PAYLOAD) - 1, HIO_NULL, &dst) <= -1)
+		{
+			FAIL ("a write on a requested stream is accepted");
+			FAIL ("the stream number set on the destination arrives at the far end");
+			FAIL ("and so does the ppid");
+			FAIL ("a zero-length write closes the writing end rather than sending an empty message");
+			teardown ();
+			return;
+		}
+	}
+
+	run_until (&g_got_data);
+	OK (g_got_data && g_rlen == (int)HIO_SIZEOF(PAYLOAD) - 1 &&
+	    HIO_MEMCMP(g_rbuf, PAYLOAD, HIO_SIZEOF(PAYLOAD) - 1) == 0,
+	    "the association carries data intact");
+	OK (g_rchan == TEST_STREAM, "the stream number set on the destination arrives at the far end");
+	OK (g_rppid == TEST_PPID, "and so does the ppid");
+
+	/* a zero-length write is the shutdown indicator for a stream device. the
+	 * seqpacket write method would have put an empty message on the wire
+	 * instead, which the peer would see as data rather than as EOF. */
+	if (hio_dev_sck_write(g_cli, HIO_NULL, 0, HIO_NULL, HIO_NULL) <= -1) FAIL ("the zero-length write is accepted");
+	else
+	{
+		run_until (&g_eof);
+		OK (g_eof, "a zero-length write closes the writing end rather than sending an empty message");
+	}
+
+	teardown ();
+}
+
+static void test_notifications_are_not_data (void)
+{
+	/* the events subscribed at socket setup mean an association coming up
+	 * produces a notification. it must be reported through on_notification
+	 * and must not appear as payload - the reads seen must be exactly the
+	 * ones the test sent. */
+	OK (g_notifications > 0, "association setup produces at least one notification");
+	OK (g_reads == 1, "and no notification is delivered to on_read as data");
+}
+
+/* ------------------------------------------------------------------ */
+/* one-to-many */
+
+static hio_dev_sck_t* g_sp_srv = HIO_NULL;
+static hio_dev_sck_t* g_sp_cli = HIO_NULL;
+static int g_sp_got, g_sp_reads;
+static hio_int32_t g_sp_assoc;
+static hio_uint16_t g_sp_chan;
+static int g_sp_len;
+static hio_bch_t g_sp_buf[512];
+
+static int sp_srv_on_read (hio_dev_sck_t* sck, const void* data, hio_iolen_t dlen, const hio_skad_t* srcaddr)
+{
+	if (dlen <= 0) return 0;
+	g_sp_reads++;
+	if (dlen > (hio_iolen_t)HIO_SIZEOF(g_sp_buf) - 1) dlen = HIO_SIZEOF(g_sp_buf) - 1;
+	HIO_MEMCPY (g_sp_buf, data, dlen);
+	g_sp_buf[dlen] = '\0';
+	g_sp_len = (int)dlen;
+	g_sp_chan = hio_skad_get_chan(srcaddr);
+	g_sp_assoc = hio_skad_get_assoc(srcaddr);
+	g_sp_got = 1;
+	return 0;
+}
+
+static int sp_cli_on_read (hio_dev_sck_t* sck, const void* data, hio_iolen_t dlen, const hio_skad_t* srcaddr)
+{
+	return 0;
+}
+
+static void sp_on_connect (hio_dev_sck_t* sck) { }
+static void sp_on_disconnect (hio_dev_sck_t* sck)
+{
+	if (sck == g_sp_srv) g_sp_srv = HIO_NULL;
+	else if (sck == g_sp_cli) g_sp_cli = HIO_NULL;
+}
+
+static void sp_fill_make (hio_dev_sck_make_t* mi, int server)
+{
+	HIO_MEMSET (mi, 0, HIO_SIZEOF(*mi));
+	mi->type = HIO_DEV_SCK_SCTP4_SP;
+	mi->on_write = on_write;
+	mi->on_read = server? sp_srv_on_read: sp_cli_on_read;
+	mi->on_connect = sp_on_connect;
+	mi->on_disconnect = sp_on_disconnect;
+	mi->on_notification = on_notification;
+	mi->sctp_ostreams = OSTREAMS;
+	mi->sctp_instreams = OSTREAMS;
+}
+
+static void sp_teardown (void)
+{
+	if (g_sp_cli) hio_dev_sck_halt (g_sp_cli);
+	if (g_sp_srv) hio_dev_sck_halt (g_sp_srv);
+	hio_exec (g_hio);
+	hio_exec (g_hio);
+	g_sp_cli = g_sp_srv = HIO_NULL;
+}
+
+static void test_one_to_many (void)
+{
+	hio_dev_sck_make_t mi;
+	hio_dev_sck_bind_t bi;
+	hio_dev_sck_listen_t li;
+	hio_skad_t peer, dst;
+	hio_oow_t bigsz;
+	hio_uint8_t* big;
+
+	g_sp_got = g_sp_reads = g_sp_len = 0;
+	g_sp_assoc = 0; g_sp_chan = 0;
+	g_notifications = 0;
+
+	sp_fill_make (&mi, 1);
+	g_sp_srv = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!g_sp_srv) { skip ("cannot make a one-to-many device", 6); return; }
+
+	/* a message-oriented device asks for a read buffer wide enough for the
+	 * largest datagram, since a short read would truncate */
+	OK (g_sp_srv->dev_rdmin == HIO_DGRAM_READ_BUFFER_SIZE,
+	    "a one-to-many socket requires a whole-message read buffer");
+
+	HIO_MEMSET (&bi, 0, HIO_SIZEOF(bi));
+	if (hio_bcstrtoskad(g_hio, "127.0.0.1:0", &bi.localaddr) <= -1) { skip ("bad address", 5); sp_teardown(); return; }
+	bi.options = HIO_DEV_SCK_BIND_REUSEADDR;
+	if (hio_dev_sck_bind(g_sp_srv, &bi) <= -1) { skip ("bind failed", 5); sp_teardown(); return; }
+
+	/* listen() is called, but nothing is ever accepted - associations are not
+	 * devices in this model */
+	HIO_MEMSET (&li, 0, HIO_SIZEOF(li));
+	li.backlogs = 4;
+	HIO_INIT_NTIME (&li.accept_tmout, -1, 0);
+	OK (hio_dev_sck_listen(g_sp_srv, &li) == 0, "and it listens without accepting");
+
+	if (hio_dev_sck_getsockaddr(g_sp_srv, &peer) <= -1) { skip ("cannot read the bound address", 4); sp_teardown(); return; }
+
+	sp_fill_make (&mi, 0);
+	g_sp_cli = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!g_sp_cli) { skip ("cannot make a one-to-many client", 4); sp_teardown(); return; }
+
+	/* no connect: sending to an address forms the association implicitly */
+	dst = peer;
+	hio_skad_set_chan (&dst, TEST_STREAM);
+	if (hio_dev_sck_write(g_sp_cli, PAYLOAD, HIO_SIZEOF(PAYLOAD) - 1, HIO_NULL, &dst) <= -1)
+	{
+		FAIL ("a write forms the association implicitly");
+		FAIL ("the stream number arrives with the message");
+		FAIL ("the association id arrives with the message");
+		FAIL ("an oversized message is dropped whole rather than split");
+		sp_teardown ();
+		return;
+	}
+
+	run_until (&g_sp_got);
+	OK (g_sp_got && g_sp_len == (int)HIO_SIZEOF(PAYLOAD) - 1 &&
+	    HIO_MEMCMP(g_sp_buf, PAYLOAD, HIO_SIZEOF(PAYLOAD) - 1) == 0,
+	    "a write forms the association implicitly and the message arrives");
+	OK (g_sp_chan == TEST_STREAM, "the stream number arrives with the message");
+	OK (g_sp_assoc != 0, "the association id arrives with the message");
+
+	/* now a message larger than the read buffer. without the MSG_EOR check
+	 * the receiver would see it as two separate messages; with it, the whole
+	 * message is dropped and nothing bogus is delivered. */
+	bigsz = HIO_DGRAM_READ_BUFFER_SIZE * 2;
+	big = (hio_uint8_t*)hio_allocmem(g_hio, bigsz);
+	if (!big) skip ("out of memory", 1);
+	else
+	{
+		int reads_before = g_sp_reads;
+		HIO_MEMSET (big, 'Z', bigsz);
+		g_sp_got = 0;
+		if (hio_dev_sck_write(g_sp_cli, big, bigsz, HIO_NULL, &dst) <= -1)
+		{
+			/* the system may refuse a message this large outright, which is
+			 * also a correct outcome - nothing bogus reaches on_read */
+			OK (g_sp_reads == reads_before, "an oversized message is dropped whole rather than split");
+		}
+		else
+		{
+			run_until (&g_sp_got);
+			OK (g_sp_reads == reads_before,
+			    "an oversized message is dropped whole rather than split");
+		}
+		hio_freemem (g_hio, big);
+	}
+
+	sp_teardown ();
+}
+
+/* ------------------------------------------------------------------ */
+
+int main (void)
+{
+	hio_errinf_t errinf;
+	hio_dev_sck_make_t mi;
+	hio_dev_sck_t* probe;
+
+	g_hio = hio_open(HIO_NULL, 0, HIO_NULL, HIO_FEATURE_ALL, 16, &errinf);
+	if (!g_hio)
+	{
+		no_plan ();
+		bail_out ("unable to open hio");
+		return -1;
+	}
+	quiet_logging (g_hio);
+
+	/* a library built without sctp reports HIO_ENOIMPL here, which is how a
+	 * caller is meant to discover it. skip_all has to come before no_plan. */
+	fill_make (&mi, 0);
+	probe = hio_dev_sck_make(g_hio, 0, &mi);
+	if (!probe)
+	{
+		if (hio_geterrnum(g_hio) == HIO_ENOIMPL)
+		{
+			skip_all ("this build of hio has no sctp support");
+			hio_close (g_hio);
+			return exit_status();
+		}
+		no_plan ();
+		bail_out ("sctp device creation failed for an unexpected reason");
+		hio_close (g_hio);
+		return -1;
+	}
+	hio_dev_sck_kill (probe);
+	hio_exec (g_hio);
+
+	no_plan ();
+
+	test_association_and_ancillary ();
+	test_notifications_are_not_data ();
+	test_one_to_many ();
+
+	hio_close (g_hio);
+	return exit_status();
+}