lws_wsmsg: buflists for ws msg reassembly

New apis allow very cheap management of multiple incoming fragmented
messages into a single queue that doesn't interrupt any ongoing
message on the upstream queue.

Author: lws-team

.gitignore

@@ -64,3 +64,4 @@ doc
 /destdir/
 /bb1/
 /bb3/
+/bin/

include/libwebsockets/lws-misc.h

@@ -198,6 +198,64 @@ LWS_VISIBLE LWS_EXTERN void *
 lws_buflist_get_frag_start_or_NULL(struct lws_buflist **head);
 
 
+struct lws_wsmsg_info;
+
+typedef void (*lws_wsmsg_transfer_cb)(struct lws_wsmsg_info *info);
+
+typedef struct lws_wsmsg_info {
+	struct lws_buflist	**head_upstream;	/* the upstream buflist */
+       	struct lws_buflist	**private_heads;	/* the private reassembly heads */
+	int			private_source_idx;	/* which index to use in private_heads */
+	lws_wsmsg_transfer_cb	optional_cb;		/* optional transfer callback */
+	void			*opaque;		/* optional opaque pointer */
+       	const uint8_t		*buf;			/* array to add */
+       	size_t			len;			/* length of bytes in array */
+       	unsigned int		ss_flags;		/* SS flags for SOM / EOM */
+} lws_wsmsg_info_t;
+
+/**
+ * lws_wsmsg_append() - append to buflist via private buflists
+ *
+ * \param info: the info struct, filled in before calling
+ *
+ * This api allows you to pass incoming fragments through a private buflist
+ * until the message it contains is reassembled and complete.  At that point, if
+ * the upstream buflist is not blocked by being in the middle of a message itself,
+ * the fragments are added to the end of an upstream buflist and the private
+ * buflist left empty.
+ *
+ * As an optimization, if the upstream buflist is not blocked waiting for an EOM,
+ * the private buflist is empty, and the incoming data represents a complete
+ * message, then the incoming message is added directly to the upstream buflist.
+ *
+ * This method allows potentially many async connections with fragmented messages
+ * to contribute to a single buflist containing only complete messages.  It's
+ * not possible to add a partial message (without the EOM flag) to the upstream
+ * buflist; when all parts of it have arrived it will be added.
+ *
+ * An array of private buflists is supported so that messages from many different
+ * connections can be reassembled before moving to the upstream buflist.
+ *
+ * Optionally, instead of transfer to another buflist when the message is complete,
+ * if the optional_cb is provided, this is called instead with the info struct, so
+ * arbitrary operations can be performed by the user.
+ *
+ * The info->opaque pointer is not used by lws, it's there to facilitate passing
+ * related user parameters in the callback case.
+ */
+LWS_VISIBLE LWS_EXTERN int
+lws_wsmsg_append(lws_wsmsg_info_t *info);
+
+/*
+ * lws_wsmsg_destroy() - free all allocations in private buflists
+ *
+ * \param private_heads: the private buflists
+ * \param count_private_heads: the number of private buflists
+ */
+LWS_VISIBLE LWS_EXTERN void
+lws_wsmsg_destroy(struct lws_buflist *private_heads[], size_t count_private_heads);
+
+
 
 /*
  * Optional helpers for closely-managed stream flow control.  These are useful

lib/core/buflist.c

@@ -338,3 +338,127 @@ lws_flow_req(lws_flow_t *flow)
 	       flow->state != LWSDLOFLOW_STATE_READ ? LWS_SRET_OK :
 					              LWS_SRET_WANT_INPUT;
 }
+
+
+static void
+lws_wsmsg_transfer(lws_wsmsg_info_t *info)
+{
+	struct lws_buflist *bl = info->private_heads[info->private_source_idx],
+			   *ubl = *info->head_upstream;
+
+	/*
+	 * If we arrived at a complete message, and the upstream is
+	 * not blocked awaiting EOM, transfer the segments to the
+	 * upstream, emptying the private buflist
+	 */
+
+	if (!bl) {
+		lwsl_notice("%s: denied: no content to transfer\n", __func__);
+		return;
+	}
+
+	while (bl && bl->next)
+		bl = bl->next;
+
+	if (bl->awaiting_eom) {
+		lwsl_notice("%s: denied: head awaiting EOM\n", __func__);
+		return;
+	}
+
+	if (!*info->head_upstream) {
+		/*
+		 * If the upstream is empty, create it by pointing
+		 * it to the whole private chain, taking ownership
+		 */
+
+		*info->head_upstream = info->private_heads[info->private_source_idx];
+		info->private_heads[info->private_source_idx] = NULL;
+
+		lwsl_notice("%s: transferred: head -> head_upstream\n", __func__);
+
+		return;
+	}
+
+
+	/* find the end of the existing upstream */
+
+	while (ubl && ubl->next)
+		ubl = ubl->next;
+
+	if (ubl->awaiting_eom) {
+		lwsl_notice("%s: denied: no content to transfer\n", __func__);
+		return;
+	}
+
+	/*
+	 * Add the private buflist on to the end of
+	 * the upstream buflist, taking ownership
+	 */
+
+	ubl->next					= info->private_heads[info->private_source_idx];
+	info->private_heads[info->private_source_idx]	= NULL; /* now it transferred upstream, private owns nothing */
+}
+
+int
+lws_wsmsg_append(lws_wsmsg_info_t *info)
+{
+	struct lws_buflist *bl;
+
+	/*
+	 * if there's nothing already stored, the new message is complete,
+	 * and the upstream is either empty, or is not blocked awaiting EOM,
+	 * then just apply the message directly to the upstream.
+	 */
+
+	if (!info->private_heads[info->private_source_idx] &&
+	    (info->ss_flags == (LWSSS_FLAG_SOM | LWSSS_FLAG_EOM)) &&
+	    (!(*info->head_upstream) || !(*info->head_upstream)->awaiting_eom)) {
+
+		lwsl_notice("%s: directly applying upstream\n", __func__);
+
+		if (lws_buflist_append_segment(info->head_upstream, info->buf, info->len) < 0)
+			return -1;
+
+		/*
+		 * Let's tag the tail buflist we just added,
+		 * with extra information useful for debugging
+		 */
+
+		bl = *info->head_upstream;
+	} else {
+		/*
+		 * Otherwise, apply the message to the private buflist first
+		 */
+
+		lwsl_notice("%s: applying via private buflist\n", __func__);
+
+		if (lws_buflist_append_segment(&info->private_heads[info->private_source_idx],
+					       info->buf, info->len) < 0)
+			return -1;
+
+		bl = info->private_heads[info->private_source_idx];
+	}
+
+	while (bl && bl->next)
+		bl = bl->next;
+
+	if (!bl)
+		return 0;
+
+	bl->awaiting_eom	= !(info->ss_flags & LWSSS_FLAG_EOM);
+	bl->src_channel		= (unsigned char)info->private_source_idx;
+
+	lws_wsmsg_transfer(info);
+
+	return 0;
+}
+
+void
+lws_wsmsg_destroy(struct lws_buflist *private_heads[], size_t count_private_heads)
+{
+	size_t m = 0;
+
+	while (m < count_private_heads)
+		lws_buflist_destroy_all_segments(&private_heads[m++]);
+}
+

lib/core/private-lib-core.h

@@ -436,9 +436,11 @@ typedef struct lws_ss_sinks {
 #endif
 
 typedef struct lws_buflist {
-	struct lws_buflist *next;
-	size_t len;
-	size_t pos;
+	struct lws_buflist			*next;
+	size_t					len;
+	size_t					pos;
+	unsigned char				awaiting_eom;
+	unsigned char				src_channel;
 } lws_buflist_t;