From 01b08fdd532d4353f9c671969b36e1a88d316f36 Mon Sep 17 00:00:00 2001 From: Chris Lew Date: Thu, 7 Jun 2018 11:55:06 -0700 Subject: [PATCH 1/2] net: qrtr: Add socket mode optimization A remote endpoint should not need to know when a client socket is freed if the socket never established commnication with the endpoint. Add a mode to keep track of which endpoints a socket communicates with. There are three modes a socket can be in: INIT - Socket has not sent anything or only local messages, only send client close to local services. SINGLE - Socket has sent messages to a single ept, send event to this single ept. MULTI - Socket has sent messages to multiple epts, broadcast release of this socket. Server state changes should be broadcast throughout the system. Change the ipc state of a port when it sends a NEW SERVER control packet. This ensures the DEL CLIENT control packet is propagated correctly for servers. This change squashes the following commits from msm-4.14: commit 3373eb77cd6d ("net: qrtr: Add socket mode optimization") commit d984a2d6f150 ("net: qrtr: Make servers multi state ports") In addition, fix minor format issues. Change-Id: I416742cff598bf7aa9da2baa2e86a53b18367870 Signed-off-by: Chris Lew --- net/qrtr/qrtr.c | 72 ++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 59 insertions(+), 13 deletions(-) diff --git a/net/qrtr/qrtr.c b/net/qrtr/qrtr.c index 8e6a392ea34c..2d96f1a8388c 100644 --- a/net/qrtr/qrtr.c +++ b/net/qrtr/qrtr.c @@ -22,6 +22,10 @@ #define QRTR_MIN_EPH_SOCKET 0x4000 #define QRTR_MAX_EPH_SOCKET 0x7fff +/* qrtr socket states */ +#define QRTR_STATE_MULTI -2 +#define QRTR_STATE_INIT -1 + /** * struct qrtr_hdr_v1 - (I|R)PCrouter packet header version 1 * @version: protocol version @@ -88,6 +92,8 @@ struct qrtr_sock { struct sock sk; struct sockaddr_qrtr us; struct sockaddr_qrtr peer; + + int state; }; static inline struct qrtr_sock *qrtr_sk(struct sock *sk) @@ -643,29 +649,59 @@ static void qrtr_port_put(struct qrtr_sock *ipc) sock_put(&ipc->sk); } -/* Remove port assignment. */ -static void qrtr_port_remove(struct qrtr_sock *ipc) +static void qrtr_send_del_client(struct qrtr_sock *ipc) { struct qrtr_ctrl_pkt *pkt; - struct sk_buff *skb; - int port = ipc->us.sq_port; struct sockaddr_qrtr to; + struct qrtr_node *node; + struct sk_buff *skbn; + struct sk_buff *skb; + int type = QRTR_TYPE_DEL_CLIENT; + + skb = qrtr_alloc_ctrl_packet(&pkt); + if (!skb) + return; to.sq_family = AF_QIPCRTR; to.sq_node = QRTR_NODE_BCAST; to.sq_port = QRTR_PORT_CTRL; - skb = qrtr_alloc_ctrl_packet(&pkt); - if (skb) { - pkt->cmd = cpu_to_le32(QRTR_TYPE_DEL_CLIENT); - pkt->client.node = cpu_to_le32(ipc->us.sq_node); - pkt->client.port = cpu_to_le32(ipc->us.sq_port); + pkt->cmd = cpu_to_le32(QRTR_TYPE_DEL_CLIENT); + pkt->client.node = cpu_to_le32(ipc->us.sq_node); + pkt->client.port = cpu_to_le32(ipc->us.sq_port); - skb_set_owner_w(skb, &ipc->sk); - qrtr_bcast_enqueue(NULL, skb, QRTR_TYPE_DEL_CLIENT, &ipc->us, - &to); + skb_set_owner_w(skb, &ipc->sk); + + if (ipc->state == QRTR_STATE_MULTI) { + qrtr_bcast_enqueue(NULL, skb, type, &ipc->us, &to); + return; } + if (ipc->state > QRTR_STATE_INIT) { + node = qrtr_node_lookup(ipc->state); + if (!node) + goto exit; + + skbn = skb_clone(skb, GFP_KERNEL); + if (!skbn) { + qrtr_node_release(node); + goto exit; + } + + skb_set_owner_w(skbn, &ipc->sk); + qrtr_node_enqueue(node, skbn, type, &ipc->us, &to); + qrtr_node_release(node); + } +exit: + qrtr_local_enqueue(NULL, skb, type, &ipc->us, &to); +} + +/* Remove port assignment. */ +static void qrtr_port_remove(struct qrtr_sock *ipc) +{ + int port = ipc->us.sq_port; + + qrtr_send_del_client(ipc); if (port == QRTR_PORT_CTRL) port = 0; @@ -928,6 +964,11 @@ static int qrtr_sendmsg(struct socket *sock, struct msghdr *msg, size_t len) release_sock(sk); return -ECONNRESET; } + + if (ipc->state > QRTR_STATE_INIT && ipc->state != node->nid) + ipc->state = QRTR_STATE_MULTI; + else if (ipc->state == QRTR_STATE_INIT) + ipc->state = node->nid; } plen = (len + 3) & ~3; @@ -944,7 +985,8 @@ static int qrtr_sendmsg(struct socket *sock, struct msghdr *msg, size_t len) goto out_node; } - if (ipc->us.sq_port == QRTR_PORT_CTRL) { + if (ipc->us.sq_port == QRTR_PORT_CTRL || + addr->sq_port == QRTR_PORT_CTRL) { if (len < 4) { rc = -EINVAL; kfree_skb(skb); @@ -956,6 +998,9 @@ static int qrtr_sendmsg(struct socket *sock, struct msghdr *msg, size_t len) } type = le32_to_cpu(qrtr_type); + if (addr->sq_port == QRTR_PORT_CTRL && type == QRTR_TYPE_NEW_SERVER) + ipc->state = QRTR_STATE_MULTI; + rc = enqueue_fn(node, skb, type, &ipc->us, addr); if (rc >= 0) rc = len; @@ -1244,6 +1289,7 @@ static int qrtr_create(struct net *net, struct socket *sock, ipc->us.sq_family = AF_QIPCRTR; ipc->us.sq_node = qrtr_local_nid; ipc->us.sq_port = 0; + ipc->state = QRTR_STATE_INIT; return 0; } From 5e6fd7cda2d982131abd7b1ae82c278742f2e1ef Mon Sep 17 00:00:00 2001 From: Chris Lew Date: Tue, 26 Jun 2018 10:39:00 -0700 Subject: [PATCH 2/2] net: qrtr: Change node mutex to a rw_semaphore There is a chance for a remote processor to become unresponsive while QRTR is trying to broadcast a control message. This results in the node lock being held for an extended period of time and prevents node lookup which is needed to send messages. Change the mutex to a rw_semaphore to allow concurrency during node lookups and broadcasts. Change-Id: I2e3fdde22edaac64f164cf08900e9d09b16c380d Signed-off-by: Chris Lew --- net/qrtr/qrtr.c | 41 ++++++++++++++++++++++++++++++++++------- 1 file changed, 34 insertions(+), 7 deletions(-) diff --git a/net/qrtr/qrtr.c b/net/qrtr/qrtr.c index 2d96f1a8388c..bba9972abd75 100644 --- a/net/qrtr/qrtr.c +++ b/net/qrtr/qrtr.c @@ -10,6 +10,7 @@ #include #include #include +#include #include @@ -110,7 +111,7 @@ static DEFINE_SPINLOCK(qrtr_nodes_lock); /* broadcast list */ static LIST_HEAD(qrtr_all_nodes); /* lock for qrtr_all_nodes and node reference */ -static DEFINE_MUTEX(qrtr_node_lock); +static DECLARE_RWSEM(qrtr_node_lock); /* local port allocation management */ static DEFINE_IDR(qrtr_ports); @@ -168,6 +169,32 @@ static int qrtr_bcast_enqueue(struct qrtr_node *node, struct sk_buff *skb, static struct qrtr_sock *qrtr_port_lookup(int port); static void qrtr_port_put(struct qrtr_sock *ipc); +static bool refcount_dec_and_rwsem_lock(refcount_t *r, + struct rw_semaphore *sem) +{ + if (refcount_dec_not_one(r)) + return false; + + down_write(sem); + if (!refcount_dec_and_test(r)) { + up_write(sem); + return false; + } + + return true; +} + +static inline int kref_put_rwsem_lock(struct kref *kref, + void (*release)(struct kref *kref), + struct rw_semaphore *sem) +{ + if (refcount_dec_and_rwsem_lock(&kref->refcount, sem)) { + release(kref); + return 1; + } + return 0; +} + /* Release node resources and free the node. * * Do not call directly, use qrtr_node_release. To be used with @@ -186,7 +213,7 @@ static void __qrtr_node_release(struct kref *kref) spin_unlock_irqrestore(&qrtr_nodes_lock, flags); list_del(&node->item); - mutex_unlock(&qrtr_node_lock); + up_write(&qrtr_node_lock); /* Free tx flow counters */ radix_tree_for_each_slot(slot, &node->qrtr_tx_flow, &iter, 0) { @@ -211,7 +238,7 @@ static void qrtr_node_release(struct qrtr_node *node) { if (!node) return; - kref_put_mutex(&node->ref, __qrtr_node_release, &qrtr_node_lock); + kref_put_rwsem_lock(&node->ref, __qrtr_node_release, &qrtr_node_lock); } /** @@ -583,9 +610,9 @@ int qrtr_endpoint_register(struct qrtr_endpoint *ep, unsigned int nid) qrtr_node_assign(node, nid); - mutex_lock(&qrtr_node_lock); + down_write(&qrtr_node_lock); list_add(&node->item, &qrtr_all_nodes); - mutex_unlock(&qrtr_node_lock); + up_write(&qrtr_node_lock); ep->node = node; return 0; @@ -886,7 +913,7 @@ static int qrtr_bcast_enqueue(struct qrtr_node *node, struct sk_buff *skb, { struct sk_buff *skbn; - mutex_lock(&qrtr_node_lock); + down_read(&qrtr_node_lock); list_for_each_entry(node, &qrtr_all_nodes, item) { if (node->nid == QRTR_EP_NID_AUTO) continue; @@ -896,7 +923,7 @@ static int qrtr_bcast_enqueue(struct qrtr_node *node, struct sk_buff *skb, skb_set_owner_w(skbn, skb->sk); qrtr_node_enqueue(node, skbn, type, from, to); } - mutex_unlock(&qrtr_node_lock); + up_read(&qrtr_node_lock); qrtr_local_enqueue(node, skb, type, from, to);