system/smd/smd.c

/*
 * lws System Message Distribution
 *
 * Copyright (C) 2019 - 2021 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>

/* comment me to remove extra debug and sanity checks */
// #define LWS_SMD_DEBUG


#if defined(LWS_SMD_DEBUG)
#define lwsl_smd lwsl_notice
#else
#define lwsl_smd(_s, ...)
#endif

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;
}

#if defined(LWS_SMD_DEBUG)
static void
lws_smd_dump(lws_smd_t *smd)
{
	int n = 1;

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

		lwsl_info(" msg %d: %p: ref %d, lat %dms, cls: 0x%x, len %u: '%s'\n",
			    n++, msg, msg->refcount,
			    (unsigned int)((lws_now_usecs() - msg->timestamp) / 1000),
			    msg->length, msg->_class,
			    (const char *)&msg[1] + LWS_SMD_SS_RX_HEADER_LEN_EFF);

	} lws_end_foreach_dll_safe(p, p1);

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

		lwsl_info(" peer %d: %p: tail: %p, filt 0x%x\n",
			    n++, pr, pr->tail, pr->_class_filter);
	} lws_end_foreach_dll(p);
}
#endif

static int
_lws_smd_msg_peer_interested_in_msg(lws_smd_peer_t *pr, lws_smd_msg_t *msg)
{
    return !!(msg->_class & pr->_class_filter);
}

/*
 * 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);

		if (pr != exc && _lws_smd_msg_peer_interested_in_msg(pr, msg))
			/*
			 * 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);

		mask |= pr->_class_filter;

	} lws_end_foreach_dll_safe(p, p1);

	smd->_class_filter = mask;

	return 0;
}

/* Call with message lock held */

static void
_lws_smd_msg_destroy(lws_smd_t *smd, lws_smd_msg_t *msg)
{
	/*
	 * We think we gave the message to everyone and can destroy it.
	 * Sanity check that no peer holds a pointer to this guy
	 */

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

		if (xpr->tail == msg) {
			lwsl_err("%s: peer %p has msg %p "
				 "we are about to destroy as tail\n",
				 __func__, xpr, msg);
#if !defined(LWS_PLAT_FREERTOS)
			assert(0);
#endif
		}

	} lws_end_foreach_dll_safe(p, p1);

	/*
	 * 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);
}

/*
 * This is wanting to be threadsafe, limiting the apis we can call
 */

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 >= ctx->smd_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;

	/* let's add him on the queue... */

	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++++++++++ messages */
	lws_dll2_add_tail(&msg->list, &ctx->smd.owner_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 && _lws_smd_msg_peer_interested_in_msg(pr, msg)) {
			pr->tail = msg;
			/* tail message has to actually be of interest to the peer */
			assert(!pr->tail || (pr->tail->_class & pr->_class_filter));
		}

	} lws_end_foreach_dll(p);

#if defined(LWS_SMD_DEBUG)
	lwsl_smd("%s: added %p (refc %u) depth now %d\n", __func__,
		 msg, msg->refcount, ctx->smd.owner_messages.count);
	lws_smd_dump(&ctx->smd);
#endif

	lws_mutex_unlock(ctx->smd.lock_messages); /* --------------- messages */

	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;
}

/*
 * This is wanting to be threadsafe, limiting the apis we can call
 */

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

/*
 * This is wanting to be threadsafe, limiting the apis we can call
 */

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_info("%s: %s send cl 0x%x, len %u\n", __func__, tag, (unsigned int)_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);

	if (_class == LWSSMDCL_METRICS) {

	}

	/* 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_info("%s: %s send cl 0x%x, len %u, ts %llu\n", __func__,
		    tag, (unsigned int)_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

/*
 * Peers that deregister need to adjust the refcount of messages they would
 * have been interested in, but didn't take delivery of yet
 */

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

	lws_mutex_lock(smd->lock_messages); /* +++++++++ messages */

	lws_dll2_remove(&pr->list);

	/*
	 * We take the approach to adjust the refcount of every would-have-been
	 * delivered message we were interested in
	 */

	while (pr->tail) {

		lws_smd_msg_t *m1 = lws_container_of(pr->tail->list.next,
							lws_smd_msg_t, list);

		if (_lws_smd_msg_peer_interested_in_msg(pr, pr->tail)) {
			if (!--pr->tail->refcount)
				_lws_smd_msg_destroy(smd, pr->tail);
		}

		pr->tail = m1;
	}

	lws_free(pr);

	lws_mutex_unlock(smd->lock_messages); /* messages ------- */
}

static lws_smd_msg_t *
_lws_smd_msg_next_matching_filter(lws_smd_peer_t *pr)
{
	lws_dll2_t *tail = &pr->tail->list;
	lws_smd_msg_t *msg;

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

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

	return NULL;
}

/*
 * 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);


	lwsl_smd("%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 the filters of this peer
	 */
	pr->tail = _lws_smd_msg_next_matching_filter(pr);

	/* tail message has to actually be of interest to the peer */
	assert(!pr->tail || (pr->tail->_class & pr->_class_filter));

	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++ messages */
	if (!--msg->refcount)
		_lws_smd_msg_destroy(&ctx->smd, msg);
	lws_mutex_unlock(ctx->smd.lock_messages); /* messages ------- */

	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);

			more = (char)(more | !!_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;

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

	/*
	 * Let's lock the message list before adding this peer... because...
	 */

	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++ messages */

	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);

	/*
	 * Now there's a new peer added, any messages we have stashed will try
	 * to deliver to this guy too, if he's interested in that class.  So we
	 * have to update the message refcounts for queued messages-he's-
	 * interested-in accordingly.
	 */

	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);

		if (_lws_smd_msg_peer_interested_in_msg(pr, msg))
			msg->refcount++;

	} lws_end_foreach_dll_safe(p, p1);

	/* ... ok we are done adding the peer */

	lws_mutex_unlock(ctx->smd.lock_messages); /* messages ------- */

	lwsl_info("%s: peer %p (count %u) registered\n", __func__, pr,
			(unsigned int)ctx->smd.owner_peers.count);

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

	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);

	if (!smd->delivering)
		lws_mutex_lock(smd->lock_peers); /* +++++++++++++++++++ peers */
	lwsl_notice("%s: destroying peer %p\n", __func__, pr);
	_lws_smd_peer_destroy(pr);
	if (!smd->delivering)
		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;

	/*
	 * If there are any messages, check their age and expire ones that
	 * have been hanging around too long
	 */

	lws_mutex_lock(ctx->smd.lock_peers); /* +++++++++++++++++++++++ peers */
	lws_mutex_lock(ctx->smd.lock_messages); /* +++++++++++++++++ messages */

	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);

		if ((lws_now_usecs() - msg->timestamp) > ctx->smd_ttl_us) {
			lwsl_warn("%s: timing out queued message %p\n",
					__func__, msg);

			/*
			 * We're forcibly yanking this guy, we can expect that
			 * there might be peers that point to it as their tail.
			 *
			 * In that case, move their tails on to the next guy
			 * they are interested in, if any.
			 */

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

				if (pr->tail == msg)
					pr->tail = _lws_smd_msg_next_matching_filter(pr);

			} lws_end_foreach_dll_safe(pp, pp1);

			/*
			 * No peer should fall foul of the peer tail checks
			 * when destroying the message now.
			 */

			_lws_smd_msg_destroy(&ctx->smd, msg);
		}
	} lws_end_foreach_dll_safe(p, p1);

	lws_mutex_unlock(ctx->smd.lock_messages); /* --------------- messages */

	/*
	 * Walk the peer list
	 */

	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)
			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_dll2_remove(&msg->list);
		lws_free(msg);

	} lws_end_foreach_dll_safe(p, p1);

	/*
	 * 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);

		pr->tail = NULL; /* we just nuked all the messages, ignore */
		_lws_smd_peer_destroy(pr);

	} lws_end_foreach_dll_safe(p, p1);

	lws_mutex_destroy(ctx->smd.lock_messages);
	lws_mutex_destroy(ctx->smd.lock_peers);

	return 0;
}