libwebsockets/lib/system/smd/smd.c

/*
 * lws System Message Distribution
 *
 * Copyright (C) 2019 - 2020 Andy Green <andy@warmcat.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

#include "private-lib-core.h"
#include <assert.h>

void *
lws_smd_msg_alloc(struct lws_context *ctx, lws_smd_class_t _class, size_t len)
{
	lws_smd_msg_t *msg;

	/* only allow it if someone wants to consume this class of event */

	if (!(ctx->smd._class_filter & _class)) {
		lwsl_info("%s: rejecting class 0x%x as no participant wants it\n",
			  __func__, (unsigned int)_class);
		return NULL;
	}

	assert(len <= LWS_SMD_MAX_PAYLOAD);


	/*
	 * If SS configured, over-allocate LWS_SMD_SS_RX_HEADER_LEN behind
	 * payload, ie,  msg_t (gap LWS_SMD_SS_RX_HEADER_LEN) payload
	 */
	msg = lws_malloc(sizeof(*msg) + LWS_SMD_SS_RX_HEADER_LEN_EFF + len,
			 __func__);
	if (!msg)
		return NULL;

	memset(msg, 0, sizeof(*msg));
	msg->timestamp = lws_now_usecs();
	msg->length = (uint16_t)len;
	msg->_class = _class;

	return ((uint8_t *)&msg[1]) + LWS_SMD_SS_RX_HEADER_LEN_EFF;
}

void
lws_smd_msg_free(void **ppay)
{
	lws_smd_msg_t *msg = (lws_smd_msg_t *)(((uint8_t *)*ppay) -
				LWS_SMD_SS_RX_HEADER_LEN_EFF - sizeof(*msg));

	/* if SS configured, actual alloc is LWS_SMD_SS_RX_HEADER_LEN behind */
	lws_free(msg);
	*ppay = NULL;
}

/*
 * Figure out what to set the initial refcount for the message to
 */

static int
_lws_smd_msg_assess_peers_interested(lws_smd_t *smd, lws_smd_msg_t *msg,
				     struct lws_smd_peer *exc)
{
	struct lws_context *ctx = lws_container_of(smd, struct lws_context, smd);
	int interested = 0;

	lws_start_foreach_dll(struct lws_dll2 *, p, ctx->smd.owner_peers.head) {
		lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

		/*
		 * In order to optimize the tail managment into a refcount,
		 * we have to account exactly for when peers arrived and
		 * departed (including deferring the logical peer destruction
		 * until no message pending he may have contributed to the
		 * refcount of)
		 */

		if (pr != exc &&
		    pr->timestamp_joined <= msg->timestamp &&
		    (!pr->timestamp_left || /* if zombie, only contribute to
					     * refcount if msg from before we
					     * left */
		     pr->timestamp_left >= msg->timestamp) &&
		    (msg->_class & pr->_class_filter))
			/*
			 * This peer wants to consume it
			 */
			interested++;

	} lws_end_foreach_dll(p);

	return interested;
}

static int
_lws_smd_class_mask_union(lws_smd_t *smd)
{
	uint32_t mask = 0;

	lws_start_foreach_dll_safe(struct lws_dll2 *, p, p1,
				   smd->owner_peers.head) {
		lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

		/* may destroy pr if zombie */
		mask |= pr->_class_filter;

	} lws_end_foreach_dll_safe(p, p1);

	smd->_class_filter = mask;

	return 0;
}

int
_lws_smd_msg_send(struct lws_context *ctx, void *pay, struct lws_smd_peer *exc)
{
	lws_smd_msg_t *msg = (lws_smd_msg_t *)(((uint8_t *)pay) -
				LWS_SMD_SS_RX_HEADER_LEN_EFF - sizeof(*msg));

	if (ctx->smd.owner_messages.count >= LWS_SMD_MAX_QUEUE_DEPTH) {
		lwsl_warn("%s: rejecting message on queue depth %d\n",
				__func__, (int)ctx->smd.owner_messages.count);
		/* reject the message due to max queue depth reached */
		return 1;
	}

	if (!ctx->smd.delivering)
		lws_mutex_lock(ctx->smd.lock_peers); /* +++++++++++++++ peers */

	msg->refcount = (uint16_t)_lws_smd_msg_assess_peers_interested(&ctx->smd, msg, exc);
	if (!msg->refcount) {
		/* possible, condsidering exc and no other participants */
		lws_free(msg);
		if (!ctx->smd.delivering)
			lws_mutex_unlock(ctx->smd.lock_peers); /* ------------- peers */

		return 0;
	}

	msg->exc = exc;

	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++++++++++ messages */
	lws_dll2_add_tail(&msg->list, &ctx->smd.owner_messages);
	lws_mutex_unlock(ctx->smd.lock_messages); /* --------------- messages */

	/*
	 * Any peer with no active tail needs to check our class to see if we
	 * should become his tail
	 */

	lws_start_foreach_dll(struct lws_dll2 *, p, ctx->smd.owner_peers.head) {
		lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

		if (pr != exc &&
		    !pr->tail && (pr->_class_filter & msg->_class))
			pr->tail = msg;

	} lws_end_foreach_dll(p);

	if (!ctx->smd.delivering)
		lws_mutex_unlock(ctx->smd.lock_peers); /* ------------- peers */

	/* we may be happening from another thread context */
	lws_cancel_service(ctx);

	return 0;
}

int
lws_smd_msg_send(struct lws_context *ctx, void *pay)
{
	return _lws_smd_msg_send(ctx, pay, NULL);
}

int
lws_smd_msg_printf(struct lws_context *ctx, lws_smd_class_t _class,
		   const char *format, ...)
{
	lws_smd_msg_t *msg;
	va_list ap;
	void *p;
	int n;

	if (!(ctx->smd._class_filter & _class))
		/*
		 * There's nobody interested in messages of this class atm.
		 * Don't bother generating it, and act like all is well.
		 */
		return 0;

	va_start(ap, format);
	n = vsnprintf(NULL, 0, format, ap);
	va_end(ap);
	if (n > LWS_SMD_MAX_PAYLOAD)
		/* too large to send */
		return 1;

	p = lws_smd_msg_alloc(ctx, _class, (size_t)n + 2);
	if (!p)
		return 1;
	msg = (lws_smd_msg_t *)(((uint8_t *)p) - LWS_SMD_SS_RX_HEADER_LEN_EFF -
								sizeof(*msg));
	msg->length = (uint16_t)n;
	va_start(ap, format);
	vsnprintf((char *)p, (unsigned int)n + 2, format, ap);
	va_end(ap);

	/*
	 * locks taken and released in here
	 */

	if (lws_smd_msg_send(ctx, p)) {
		lws_smd_msg_free(&p);
		return 1;
	}

	return 0;
}

#if defined(LWS_WITH_SECURE_STREAMS)
int
lws_smd_ss_msg_printf(const char *tag, uint8_t *buf, size_t *len,
		      lws_smd_class_t _class, const char *format, ...)
{
	char *content = (char *)buf + LWS_SMD_SS_RX_HEADER_LEN;
	va_list ap;
	int n;

	if (*len < LWS_SMD_SS_RX_HEADER_LEN)
		return 1;

	lws_ser_wu64be(buf, _class);
	lws_ser_wu64be(buf + 8, 0); /* valgrind notices uninitialized if left */

	va_start(ap, format);
	n = vsnprintf(content, (*len) - LWS_SMD_SS_RX_HEADER_LEN, format, ap);
	va_end(ap);

	if (n > LWS_SMD_MAX_PAYLOAD ||
	    (unsigned int)n > (*len) - LWS_SMD_SS_RX_HEADER_LEN)
		/* too large to send */
		return 1;

	*len = LWS_SMD_SS_RX_HEADER_LEN + (unsigned int)n;

	lwsl_notice("%s: %s send cl 0x%x, len %u\n", __func__, tag, _class,
			(unsigned int)n);

	return 0;
}

/*
 * This is a helper that user rx handler for LWS_SMD_STREAMTYPENAME SS can
 * call through to with the payload it received from the proxy.  It will then
 * forward the recieved SMD message to all local (same-context) participants
 * that are interested in that class (except ones with callback skip_cb, so
 * we don't loop).
 */

static int
_lws_smd_ss_rx_forward(struct lws_context *ctx, const char *tag,
		       struct lws_smd_peer *pr, const uint8_t *buf, size_t len)
{
	lws_smd_class_t _class;
	lws_smd_msg_t *msg;
	void *p;

	if (len < LWS_SMD_SS_RX_HEADER_LEN_EFF)
		return 1;

	if (len >= LWS_SMD_MAX_PAYLOAD + LWS_SMD_SS_RX_HEADER_LEN_EFF)
		return 1;

	_class = (lws_smd_class_t)lws_ser_ru64be(buf);

	/* only locally forward messages that we care about in this process */

	if (!(ctx->smd._class_filter & _class))
		/*
		 * There's nobody interested in messages of this class atm.
		 * Don't bother generating it, and act like all is well.
		 */
		return 0;

	p = lws_smd_msg_alloc(ctx, _class, len);
	if (!p)
		return 1;

	msg = (lws_smd_msg_t *)(((uint8_t *)p) - LWS_SMD_SS_RX_HEADER_LEN_EFF -
								sizeof(*msg));
	msg->length = (uint16_t)(len - LWS_SMD_SS_RX_HEADER_LEN_EFF);
	/* adopt the original source timestamp, not time we forwarded it */
	msg->timestamp = (lws_usec_t)lws_ser_ru64be(buf + 8);

	/* copy the message payload in */
	memcpy(p, buf + LWS_SMD_SS_RX_HEADER_LEN_EFF, msg->length);

	/*
	 * locks taken and released in here
	 */

	if (_lws_smd_msg_send(ctx, p, pr)) {
		/* we couldn't send it after all that... */
		lws_smd_msg_free(&p);

		return 1;
	}

	lwsl_notice("%s: %s send cl 0x%x, len %u, ts %llu\n", __func__,
		    tag, _class, msg->length,
		    (unsigned long long)msg->timestamp);

	return 0;
}

int
lws_smd_ss_rx_forward(void *ss_user, const uint8_t *buf, size_t len)
{
	struct lws_ss_handle *h = (struct lws_ss_handle *)
					(((char *)ss_user) - sizeof(*h));
	struct lws_context *ctx = lws_ss_get_context(h);

	return _lws_smd_ss_rx_forward(ctx, lws_ss_tag(h), h->u.smd.smd_peer, buf, len);
}

#if defined(LWS_WITH_SECURE_STREAMS_PROXY_API)
int
lws_smd_sspc_rx_forward(void *ss_user, const uint8_t *buf, size_t len)
{
	struct lws_sspc_handle *h = (struct lws_sspc_handle *)
					(((char *)ss_user) - sizeof(*h));
	struct lws_context *ctx = lws_sspc_get_context(h);

	return _lws_smd_ss_rx_forward(ctx, lws_sspc_tag(h), NULL, buf, len);
}
#endif

#endif

static void
_lws_smd_peer_finalize_destroy(lws_smd_peer_t *pr)
{
	lws_dll2_remove(&pr->list);
	lws_free(pr);
}

/*
 * Peers that deregister may need to hang around as zombies, so they account
 * for refcounts on messages they already contributed to.  Because older
 * messages may be in flight over UDS links, we have to stick around and make
 * sure all cases have their refcount handled correctly.
 */

static void
_lws_smd_peer_zombify(lws_smd_peer_t *pr)
{
	lws_smd_t *smd = lws_container_of(pr->list.owner, lws_smd_t,
					  owner_peers);

	/* update the class mask union to reflect this peer no longer active */
	_lws_smd_class_mask_union(smd);

	pr->timestamp_left = lws_now_usecs();
}

static lws_smd_msg_t *
_lws_smd_msg_next_matching_filter(lws_smd_peer_t *pr)
{
	lws_smd_msg_t *msg;

	lws_dll2_t *tail = &pr->tail->list;
	lws_smd_class_t filter = pr->_class_filter;

	do {
		tail = tail->next;
		if (!tail)
			return NULL;

		msg = lws_container_of(tail, lws_smd_msg_t, list);
		if (msg->exc != pr && (msg->_class & filter))
			return msg;
	} while (1);

	return NULL;
}

/*
 * Note: May destroy zombie peers when it sees grace period has expired.
 *
 * Delivers only one message to the peer and advances the tail, or sets to NULL
 * if no more filtered queued messages.  Returns nonzero if tail non-NULL.
 *
 * For Proxied SS, only asks for writeable and does not advance or change the
 * tail.
 *
 * This is done so if multiple messages queued, we don't get a situation where
 * one participant gets them all spammed, then the next etc.  Instead they are
 * delivered round-robin.
 *
 * Requires peer lock, may take message lock
 */

static int
_lws_smd_msg_deliver_peer(struct lws_context *ctx, lws_smd_peer_t *pr)
{
	lws_smd_msg_t *msg;

	if (!pr->tail)
		return 0;

	msg = lws_container_of(pr->tail, lws_smd_msg_t, list);

	/*
	 * Check if zombie peer and the message predates our leaving
	 */

	if (pr->timestamp_left &&
	    msg->timestamp > pr->timestamp_left) {
		/*
		 * We do not need to modify message refcount, if it was
		 * generated after we became a zombie, and so we
		 * definitely did not contribute to its refcount...
		 *
		 * ...have we waited out the grace period?
		 */

		if (lws_now_usecs() - pr->timestamp_left >
			   LWS_SMD_INFLIGHT_GRACE_SECS * LWS_US_PER_SEC)
			/*
			 * ... ok, it's time for the zombie to abandon
			 * its attachment to the Earth and rejoin the
			 * cosmic mandela
			 */
			_lws_smd_peer_finalize_destroy(pr);

		/* ... either way, nothing further to do for this guy */

		return 0;
	}

	if (!pr->timestamp_left) {

		/*
		 * Peer is not a zombie... deliver the tail
		 */
#if 0
		if (pr->type == LSMDT_SECURE_STREAMS_PROXIED) {
#if defined(LWS_WITH_SECURE_STREAMS)
			if (pr->ss_handle)
				lws_ss_request_tx(pr->ss_handle);
#endif
			return 0;
		}
#endif

		lwsl_info("%s: deliver cl 0x%x, len %d, refc %d, to peer %p\n",
			   __func__, (unsigned int)msg->_class, (int)msg->length,
			   (int)msg->refcount, pr);

		pr->cb(pr->opaque, msg->_class, msg->timestamp,
		       ((uint8_t *)&msg[1]) + LWS_SMD_SS_RX_HEADER_LEN_EFF,
		       (size_t)msg->length);
	}

	assert(msg->refcount);

	/*
	 * If there is one, move forward to the next queued
	 * message that meets our filters
	 */
	pr->tail = _lws_smd_msg_next_matching_filter(pr);

	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++ messages */
	if (!--msg->refcount) {
		/*
		 * We have fully delivered the message now, it
		 * can be unlinked and destroyed
		 */
		lwsl_info("%s: destroy msg %p\n", __func__, msg);
		lws_dll2_remove(&msg->list);
		lws_free(msg);
	}
	lws_mutex_unlock(ctx->smd.lock_messages); /* messages ------- */

	/*
	 * Wait out the grace period even if no live messages
	 * for a zombie peer... there may be some in flight
	 */

	return !!pr->tail;
}

/*
 * Called when the event loop could deliver messages synchronously, eg, on
 * entry to idle
 */

int
lws_smd_msg_distribute(struct lws_context *ctx)
{
	char more;

	/* commonly, no messages and nothing to do... */

	if (!ctx->smd.owner_messages.count)
		return 0;

	ctx->smd.delivering = 1;

	do {
		more = 0;
		lws_mutex_lock(ctx->smd.lock_peers); /* +++++++++++++++ peers */

		lws_start_foreach_dll_safe(struct lws_dll2 *, p, p1,
					   ctx->smd.owner_peers.head) {
			lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

			/* may destroy pr if zombie, hence _safe iterator */
			more |= (char)!!_lws_smd_msg_deliver_peer(ctx, pr);

		} lws_end_foreach_dll_safe(p, p1);

		lws_mutex_unlock(ctx->smd.lock_peers); /* ------------- peers */
	} while (more);

	ctx->smd.delivering = 0;

	return 0;
}

struct lws_smd_peer *
lws_smd_register(struct lws_context *ctx, void *opaque, int flags,
		 lws_smd_class_t _class_filter, lws_smd_notification_cb_t cb)
{
	lws_smd_peer_t *pr = lws_zalloc(sizeof(*pr), __func__);

	if (!pr)
		return NULL;

	pr->cb = cb;
	pr->opaque = opaque;
	pr->_class_filter = _class_filter;
	pr->timestamp_joined = lws_now_usecs();

	/*
	 * Figure out the type of peer from the situation...
	 */

#if 0
#if defined(LWS_WITH_SECURE_STREAMS)
	if (!ctx->smd.listen_vh) {
		/*
		 * The guy who is regsitering is actually a SS proxy link
		 * between a client and SMD
		 */
	} else
#endif
#endif
		pr->type = LSMDT_SAME_PROCESS;

	if (!ctx->smd.delivering)
		lws_mutex_lock(ctx->smd.lock_peers); /* +++++++++++++++ peers */
	lws_dll2_add_tail(&pr->list, &ctx->smd.owner_peers);

	/* update the global class mask union to account for new peer mask */
	_lws_smd_class_mask_union(&ctx->smd);
	if (!ctx->smd.delivering)
		lws_mutex_unlock(ctx->smd.lock_peers); /* ------------- peers */

	lwsl_debug("%s: registered\n", __func__);

	return pr;
}

void
lws_smd_unregister(struct lws_smd_peer *pr)
{
	lws_smd_t *smd = lws_container_of(pr->list.owner, lws_smd_t, owner_peers);

	lws_mutex_lock(smd->lock_peers); /* +++++++++++++++++++++++++++ peers */
	_lws_smd_peer_zombify(pr);
	lws_mutex_unlock(smd->lock_peers); /* ------------------------- peers */
}

int
lws_smd_message_pending(struct lws_context *ctx)
{
	int ret = 1;

	/*
	 * First cheaply check the common case no messages pending, so there's
	 * definitely nothing for this tsi or anything else
	 */

	if (!ctx->smd.owner_messages.count)
		return 0;

	/*
	 * Walk the peer list
	 */

	lws_mutex_lock(ctx->smd.lock_peers); /* +++++++++++++++++++++++ peers */
	lws_start_foreach_dll(struct lws_dll2 *, p, ctx->smd.owner_peers.head) {
		lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

		if (pr->tail && pr->type == LSMDT_SAME_PROCESS)
			goto bail;

	} lws_end_foreach_dll(p);

	/*
	 * There's no message pending that we need to handle
	 */

	ret = 0;

bail:
	lws_mutex_unlock(ctx->smd.lock_peers); /* --------------------- peers */

	return ret;
}

int
_lws_smd_destroy(struct lws_context *ctx)
{
	/* stop any message creation */

	ctx->smd._class_filter = 0;

	/*
	 * Walk the message list, destroying them
	 */

	lws_start_foreach_dll_safe(struct lws_dll2 *, p, p1,
				   ctx->smd.owner_messages.head) {
		lws_smd_msg_t *msg = lws_container_of(p, lws_smd_msg_t, list);

		lws_free(msg);

	} lws_end_foreach_dll_safe(p, p1);

	lws_mutex_destroy(ctx->smd.lock_messages);

	/*
	 * Walk the peer list, destroying them
	 */

	lws_start_foreach_dll_safe(struct lws_dll2 *, p, p1,
				   ctx->smd.owner_peers.head) {
		lws_smd_peer_t *pr = lws_container_of(p, lws_smd_peer_t, list);

		_lws_smd_peer_finalize_destroy(pr);

	} lws_end_foreach_dll_safe(p, p1);

	lws_mutex_destroy(ctx->smd.lock_peers);

	return 0;
}