Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 255adfc7 authored by Bjorn Andersson's avatar Bjorn Andersson Committed by Chris Lew
Browse files

net: qrtr: Implement outgoing flow control



It is possible to flood a remote with messages and cause memory
exhaustion. To avoid this scenario implement the following flow control
handshake. After sending 10 packets to a remote port, set a control
flag on the message and wait for a confirm rx packet from that remote
port before sending more data.

Change-Id: I81c641d4be6d2014df3f0392d140690f6afac2ac
Signed-off-by: default avatarBjorn Andersson <bjorn.andersson@linaro.org>
Git-Commit: 347753ab1cdac06af321950337a54b03f8203960
Git-Repo: https://github.com/andersson/kernel.git


[clew@codeaurora.org: Add commit text and fix checkpatch warnings.
 Use the control packet instead of header addresses for resume tx and
 fix missing mutex init. Change flow to atomic_t to ensure the compiler
 orders the flow control set and signal correctly.]
Signed-off-by: default avatarChris Lew <clew@codeaurora.org>
parent b93572a7
Loading
Loading
Loading
Loading
+139 −9
Original line number Diff line number Diff line
@@ -15,6 +15,7 @@
#include <linux/netlink.h>
#include <linux/qrtr.h>
#include <linux/termios.h>	/* For TIOCINQ/OUTQ */
#include <linux/wait.h>

#include <net/sock.h>

@@ -120,6 +121,9 @@ static DEFINE_MUTEX(qrtr_port_lock);
 * @ep: endpoint
 * @ref: reference count for node
 * @nid: node id
 * @qrtr_tx_flow: remote port tx flow control list
 * @resume_tx: wait until remote port acks control flag
 * @qrtr_tx_lock: lock for qrtr_tx_flow
 * @rx_queue: receive queue
 * @work: scheduled work struct for recv work
 * @item: list item for broadcast list
@@ -130,11 +134,22 @@ struct qrtr_node {
	struct kref ref;
	unsigned int nid;

	struct radix_tree_root qrtr_tx_flow;
	struct wait_queue_head resume_tx;
	struct mutex qrtr_tx_lock;	/* for qrtr_tx_flow */

	struct sk_buff_head rx_queue;
	struct work_struct work;
	struct list_head item;
};

struct qrtr_tx_flow {
	atomic_t pending;
};

#define QRTR_TX_FLOW_HIGH	10
#define QRTR_TX_FLOW_LOW	5

static int qrtr_local_enqueue(struct qrtr_node *node, struct sk_buff *skb,
			      int type, struct sockaddr_qrtr *from,
			      struct sockaddr_qrtr *to);
@@ -149,9 +164,9 @@ static int qrtr_bcast_enqueue(struct qrtr_node *node, struct sk_buff *skb,
 */
static void __qrtr_node_release(struct kref *kref)
{
	struct qrtr_node *node = container_of(kref, struct qrtr_node, ref);
	struct radix_tree_iter iter;
	void **slot;
	struct qrtr_node *node = container_of(kref, struct qrtr_node, ref);
	void __rcu **slot;

	if (node->nid != QRTR_EP_NID_AUTO) {
		radix_tree_for_each_slot(slot, &qrtr_nodes, &iter, 0) {
@@ -163,6 +178,12 @@ static void __qrtr_node_release(struct kref *kref)
	list_del(&node->item);
	mutex_unlock(&qrtr_node_lock);

	/* Free tx flow counters */
	radix_tree_for_each_slot(slot, &node->qrtr_tx_flow, &iter, 0) {
		radix_tree_iter_delete(&node->qrtr_tx_flow, &iter, slot);
		kfree(*slot);
	}

	skb_queue_purge(&node->rx_queue);
	kfree(node);
}
@@ -183,15 +204,111 @@ static void qrtr_node_release(struct qrtr_node *node)
	kref_put_mutex(&node->ref, __qrtr_node_release, &qrtr_node_lock);
}

/**
 * qrtr_tx_resume() - reset flow control counter
 * @node:	qrtr_node that the QRTR_TYPE_RESUME_TX packet arrived on
 * @skb:	skb for resume tx control packet
 */
static void qrtr_tx_resume(struct qrtr_node *node, struct sk_buff *skb)
{
	struct qrtr_ctrl_pkt *pkt;
	struct qrtr_tx_flow *flow;
	unsigned long key;
	int dest_node;
	int dest_port;

	pkt = (struct qrtr_ctrl_pkt *)skb->data;
	if (le32_to_cpu(pkt->cmd) != QRTR_TYPE_RESUME_TX)
		return;

	dest_node = le32_to_cpu(pkt->client.node);
	dest_port = le32_to_cpu(pkt->client.port);
	key = (u64)dest_node << 32 | dest_port;

	flow = radix_tree_lookup(&node->qrtr_tx_flow, key);
	if (flow)
		atomic_set(&flow->pending, 0);

	wake_up_interruptible_all(&node->resume_tx);
}

/**
 * qrtr_tx_wait() - flow control for outgoing packets
 * @node:	qrtr_node that the packet is to be send to
 * @dest_node:	node id of the destination
 * @dest_port:	port number of the destination
 * @type:	type of message
 *
 * The flow control scheme is based around the low and high "watermarks". When
 * the low watermark is passed the confirm_rx flag is set on the outgoing
 * message, which will trigger the remote to send a control message of the type
 * QRTR_TYPE_RESUME_TX to reset the counter. If the high watermark is hit
 * further transmision should be paused.
 *
 * Return: 1 if confirm_rx should be set, 0 otherwise or errno failure
 */
static int qrtr_tx_wait(struct qrtr_node *node, int dest_node, int dest_port,
			int type)
{
	struct qrtr_tx_flow *flow;
	unsigned long key = (u64)dest_node << 32 | dest_port;
	int confirm_rx = 0;
	int ret;

	/* Never set confirm_rx on non-data packets */
	if (type != QRTR_TYPE_DATA)
		return 0;

	mutex_lock(&node->qrtr_tx_lock);
	flow = radix_tree_lookup(&node->qrtr_tx_flow, key);
	if (!flow) {
		flow = kzalloc(sizeof(*flow), GFP_KERNEL);
		if (!flow)
			return 1;
		else
			radix_tree_insert(&node->qrtr_tx_flow, key, flow);
	}
	mutex_unlock(&node->qrtr_tx_lock);

	for (;;) {
		ret = wait_event_interruptible(node->resume_tx,
					       atomic_read(&flow->pending) <
					       QRTR_TX_FLOW_HIGH || !node->ep);
		if (ret)
			return ret;

		if (!node->ep)
			return -EPIPE;

		mutex_lock(&node->qrtr_tx_lock);
		if (atomic_read(&flow->pending) < QRTR_TX_FLOW_HIGH) {
			atomic_inc(&flow->pending);
			confirm_rx = atomic_read(&flow->pending) == QRTR_TX_FLOW_LOW;
			mutex_unlock(&node->qrtr_tx_lock);
			break;
		}
		mutex_unlock(&node->qrtr_tx_lock);
	}

	return confirm_rx;
}

/* Pass an outgoing packet socket buffer to the endpoint driver. */
static int qrtr_node_enqueue(struct qrtr_node *node, struct sk_buff *skb,
			     int type, struct sockaddr_qrtr *from,
			     struct sockaddr_qrtr *to)
{
	struct qrtr_hdr_v1 *hdr;
	int confirm_rx;
	size_t len = skb->len;
	int rc = -ENODEV;

	confirm_rx = qrtr_tx_wait(node, to->sq_node, to->sq_port, type);
	if (confirm_rx < 0) {
		kfree_skb(skb);
		return confirm_rx;
	}

	hdr = skb_push(skb, sizeof(*hdr));
	hdr->version = cpu_to_le32(QRTR_PROTO_VER_1);
	hdr->type = cpu_to_le32(type);
@@ -201,7 +318,7 @@ static int qrtr_node_enqueue(struct qrtr_node *node, struct sk_buff *skb,
	hdr->dst_port_id = cpu_to_le32(to->sq_port);

	hdr->size = cpu_to_le32(len);
	hdr->confirm_rx = 0;
	hdr->confirm_rx = !!confirm_rx;

	skb_put_padto(skb, ALIGN(len, 4) + sizeof(*hdr));

@@ -321,7 +438,8 @@ int qrtr_endpoint_post(struct qrtr_endpoint *ep, const void *data, size_t len)
	if (len != ALIGN(size, 4) + hdrlen)
		goto err;

	if (cb->dst_port != QRTR_PORT_CTRL && cb->type != QRTR_TYPE_DATA)
	if (cb->dst_port != QRTR_PORT_CTRL && cb->type != QRTR_TYPE_DATA &&
	    cb->type != QRTR_TYPE_RESUME_TX)
		goto err;

	skb_put_data(skb, data + hdrlen, size);
@@ -381,6 +499,10 @@ static void qrtr_node_rx_work(struct work_struct *work)
		cb = (struct qrtr_cb *)skb->cb;
		qrtr_node_assign(node, cb->src_node);

		if (cb->type == QRTR_TYPE_RESUME_TX) {
			qrtr_tx_resume(node, skb);
			consume_skb(skb);
		} else {
			ipc = qrtr_port_lookup(cb->dst_port);
			if (!ipc) {
				kfree_skb(skb);
@@ -392,6 +514,7 @@ static void qrtr_node_rx_work(struct work_struct *work)
			}
		}
	}
}

/**
 * qrtr_endpoint_register() - register a new endpoint
@@ -419,6 +542,10 @@ int qrtr_endpoint_register(struct qrtr_endpoint *ep, unsigned int nid)
	node->nid = QRTR_EP_NID_AUTO;
	node->ep = ep;

	mutex_init(&node->qrtr_tx_lock);
	INIT_RADIX_TREE(&node->qrtr_tx_flow, GFP_KERNEL);
	init_waitqueue_head(&node->resume_tx);

	qrtr_node_assign(node, nid);

	mutex_lock(&qrtr_node_lock);
@@ -453,6 +580,9 @@ void qrtr_endpoint_unregister(struct qrtr_endpoint *ep)
		qrtr_local_enqueue(NULL, skb, QRTR_TYPE_BYE, &src, &dst);
	}

	/* Wake up any transmitters waiting for resume-tx from the node */
	wake_up_interruptible_all(&node->resume_tx);

	qrtr_node_release(node);
	ep->node = NULL;
}