1	// SPDX-License-Identifier: GPL-2.0-only
2	/*
3	* Kernel Connection Multiplexor
4	*
5	* Copyright (c) 2016 Tom Herbert <tom@herbertland.com>
6	*/
7
8	#include <linux/bpf.h>
9	#include <linux/errno.h>
10	#include <linux/errqueue.h>
11	#include <linux/file.h>
12	#include <linux/filter.h>
13	#include <linux/in.h>
14	#include <linux/kernel.h>
15	#include <linux/module.h>
16	#include <linux/net.h>
17	#include <linux/netdevice.h>
18	#include <linux/poll.h>
19	#include <linux/rculist.h>
20	#include <linux/skbuff.h>
21	#include <linux/socket.h>
22	#include <linux/uaccess.h>
23	#include <linux/workqueue.h>
24	#include <linux/syscalls.h>
25	#include <linux/sched/signal.h>
26
27	#include <net/kcm.h>
28	#include <net/netns/generic.h>
29	#include <net/sock.h>
30	#include <uapi/linux/kcm.h>
31	#include <trace/events/sock.h>
32
33	unsigned int kcm_net_id;
34
35	static struct kmem_cache *kcm_psockp __read_mostly;
36	static struct kmem_cache *kcm_muxp __read_mostly;
37	static struct workqueue_struct *kcm_wq;
38
39	static inline struct kcm_sock *kcm_sk(const struct sock *sk)
40	{
41	return (struct kcm_sock *)sk;
42	}
43
44	static inline struct kcm_tx_msg *kcm_tx_msg(struct sk_buff *skb)
45	{
46	return (struct kcm_tx_msg *)skb->cb;
47	}
48
49	static void report_csk_error(struct sock *csk, int err)
50	{
51	csk->sk_err = EPIPE;
52	sk_error_report(sk: csk);
53	}
54
55	static void kcm_abort_tx_psock(struct kcm_psock *psock, int err,
56	bool wakeup_kcm)
57	{
58	struct sock *csk = psock->sk;
59	struct kcm_mux *mux = psock->mux;
60
61	/* Unrecoverable error in transmit */
62
63	spin_lock_bh(lock: &mux->lock);
64
65	if (psock->tx_stopped) {
66	spin_unlock_bh(lock: &mux->lock);
67	return;
68	}
69
70	psock->tx_stopped = 1;
71	KCM_STATS_INCR(psock->stats.tx_aborts);
72
73	if (!psock->tx_kcm) {
74	/* Take off psocks_avail list */
75	list_del(entry: &psock->psock_avail_list);
76	} else if (wakeup_kcm) {
77	/* In this case psock is being aborted while outside of
78	* write_msgs and psock is reserved. Schedule tx_work
79	* to handle the failure there. Need to commit tx_stopped
80	* before queuing work.
81	*/
82	smp_mb();
83
84	queue_work(wq: kcm_wq, work: &psock->tx_kcm->tx_work);
85	}
86
87	spin_unlock_bh(lock: &mux->lock);
88
89	/* Report error on lower socket */
90	report_csk_error(csk, err);
91	}
92
93	/* RX mux lock held. */
94	static void kcm_update_rx_mux_stats(struct kcm_mux *mux,
95	struct kcm_psock *psock)
96	{
97	STRP_STATS_ADD(mux->stats.rx_bytes,
98	psock->strp.stats.bytes -
99	psock->saved_rx_bytes);
100	mux->stats.rx_msgs +=
101	psock->strp.stats.msgs - psock->saved_rx_msgs;
102	psock->saved_rx_msgs = psock->strp.stats.msgs;
103	psock->saved_rx_bytes = psock->strp.stats.bytes;
104	}
105
106	static void kcm_update_tx_mux_stats(struct kcm_mux *mux,
107	struct kcm_psock *psock)
108	{
109	KCM_STATS_ADD(mux->stats.tx_bytes,
110	psock->stats.tx_bytes - psock->saved_tx_bytes);
111	mux->stats.tx_msgs +=
112	psock->stats.tx_msgs - psock->saved_tx_msgs;
113	psock->saved_tx_msgs = psock->stats.tx_msgs;
114	psock->saved_tx_bytes = psock->stats.tx_bytes;
115	}
116
117	static int kcm_queue_rcv_skb(struct sock *sk, struct sk_buff *skb);
118
119	/* KCM is ready to receive messages on its queue-- either the KCM is new or
120	* has become unblocked after being blocked on full socket buffer. Queue any
121	* pending ready messages on a psock. RX mux lock held.
122	*/
123	static void kcm_rcv_ready(struct kcm_sock *kcm)
124	{
125	struct kcm_mux *mux = kcm->mux;
126	struct kcm_psock *psock;
127	struct sk_buff *skb;
128
129	if (unlikely(kcm->rx_wait || kcm->rx_psock || kcm->rx_disabled))
130	return;
131
132	while (unlikely((skb = __skb_dequeue(&mux->rx_hold_queue)))) {
133	if (kcm_queue_rcv_skb(sk: &kcm->sk, skb)) {
134	/* Assuming buffer limit has been reached */
135	skb_queue_head(list: &mux->rx_hold_queue, newsk: skb);
136	WARN_ON(!sk_rmem_alloc_get(&kcm->sk));
137	return;
138	}
139	}
140
141	while (!list_empty(head: &mux->psocks_ready)) {
142	psock = list_first_entry(&mux->psocks_ready, struct kcm_psock,
143	psock_ready_list);
144
145	if (kcm_queue_rcv_skb(sk: &kcm->sk, skb: psock->ready_rx_msg)) {
146	/* Assuming buffer limit has been reached */
147	WARN_ON(!sk_rmem_alloc_get(&kcm->sk));
148	return;
149	}
150
151	/* Consumed the ready message on the psock. Schedule rx_work to
152	* get more messages.
153	*/
154	list_del(entry: &psock->psock_ready_list);
155	psock->ready_rx_msg = NULL;
156	/* Commit clearing of ready_rx_msg for queuing work */
157	smp_mb();
158
159	strp_unpause(strp: &psock->strp);
160	strp_check_rcv(strp: &psock->strp);
161	}
162
163	/* Buffer limit is okay now, add to ready list */
164	list_add_tail(new: &kcm->wait_rx_list,
165	head: &kcm->mux->kcm_rx_waiters);
166	/* paired with lockless reads in kcm_rfree() */
167	WRITE_ONCE(kcm->rx_wait, true);
168	}
169
170	static void kcm_rfree(struct sk_buff *skb)
171	{
172	struct sock *sk = skb->sk;
173	struct kcm_sock *kcm = kcm_sk(sk);
174	struct kcm_mux *mux = kcm->mux;
175	unsigned int len = skb->truesize;
176
177	sk_mem_uncharge(sk, size: len);
178	atomic_sub(i: len, v: &sk->sk_rmem_alloc);
179
180	/* For reading rx_wait and rx_psock without holding lock */
181	smp_mb__after_atomic();
182
183	if (!READ_ONCE(kcm->rx_wait) && !READ_ONCE(kcm->rx_psock) &&
184	sk_rmem_alloc_get(sk) < sk->sk_rcvlowat) {
185	spin_lock_bh(lock: &mux->rx_lock);
186	kcm_rcv_ready(kcm);
187	spin_unlock_bh(lock: &mux->rx_lock);
188	}
189	}
190
191	static int kcm_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
192	{
193	struct sk_buff_head *list = &sk->sk_receive_queue;
194
195	if (atomic_read(v: &sk->sk_rmem_alloc) >= sk->sk_rcvbuf)
196	return -ENOMEM;
197
198	if (!sk_rmem_schedule(sk, skb, size: skb->truesize))
199	return -ENOBUFS;
200
201	skb->dev = NULL;
202
203	skb_orphan(skb);
204	skb->sk = sk;
205	skb->destructor = kcm_rfree;
206	atomic_add(i: skb->truesize, v: &sk->sk_rmem_alloc);
207	sk_mem_charge(sk, size: skb->truesize);
208
209	skb_queue_tail(list, newsk: skb);
210
211	if (!sock_flag(sk, flag: SOCK_DEAD))
212	sk->sk_data_ready(sk);
213
214	return 0;
215	}
216
217	/* Requeue received messages for a kcm socket to other kcm sockets. This is
218	* called with a kcm socket is receive disabled.
219	* RX mux lock held.
220	*/
221	static void requeue_rx_msgs(struct kcm_mux *mux, struct sk_buff_head *head)
222	{
223	struct sk_buff *skb;
224	struct kcm_sock *kcm;
225
226	while ((skb = skb_dequeue(list: head))) {
227	/* Reset destructor to avoid calling kcm_rcv_ready */
228	skb->destructor = sock_rfree;
229	skb_orphan(skb);
230	try_again:
231	if (list_empty(head: &mux->kcm_rx_waiters)) {
232	skb_queue_tail(list: &mux->rx_hold_queue, newsk: skb);
233	continue;
234	}
235
236	kcm = list_first_entry(&mux->kcm_rx_waiters,
237	struct kcm_sock, wait_rx_list);
238
239	if (kcm_queue_rcv_skb(sk: &kcm->sk, skb)) {
240	/* Should mean socket buffer full */
241	list_del(entry: &kcm->wait_rx_list);
242	/* paired with lockless reads in kcm_rfree() */
243	WRITE_ONCE(kcm->rx_wait, false);
244
245	/* Commit rx_wait to read in kcm_free */
246	smp_wmb();
247
248	goto try_again;
249	}
250	}
251	}
252
253	/* Lower sock lock held */
254	static struct kcm_sock *reserve_rx_kcm(struct kcm_psock *psock,
255	struct sk_buff *head)
256	{
257	struct kcm_mux *mux = psock->mux;
258	struct kcm_sock *kcm;
259
260	WARN_ON(psock->ready_rx_msg);
261
262	if (psock->rx_kcm)
263	return psock->rx_kcm;
264
265	spin_lock_bh(lock: &mux->rx_lock);
266
267	if (psock->rx_kcm) {
268	spin_unlock_bh(lock: &mux->rx_lock);
269	return psock->rx_kcm;
270	}
271
272	kcm_update_rx_mux_stats(mux, psock);
273
274	if (list_empty(head: &mux->kcm_rx_waiters)) {
275	psock->ready_rx_msg = head;
276	strp_pause(strp: &psock->strp);
277	list_add_tail(new: &psock->psock_ready_list,
278	head: &mux->psocks_ready);
279	spin_unlock_bh(lock: &mux->rx_lock);
280	return NULL;
281	}
282
283	kcm = list_first_entry(&mux->kcm_rx_waiters,
284	struct kcm_sock, wait_rx_list);
285	list_del(entry: &kcm->wait_rx_list);
286	/* paired with lockless reads in kcm_rfree() */
287	WRITE_ONCE(kcm->rx_wait, false);
288
289	psock->rx_kcm = kcm;
290	/* paired with lockless reads in kcm_rfree() */
291	WRITE_ONCE(kcm->rx_psock, psock);
292
293	spin_unlock_bh(lock: &mux->rx_lock);
294
295	return kcm;
296	}
297
298	static void kcm_done(struct kcm_sock *kcm);
299
300	static void kcm_done_work(struct work_struct *w)
301	{
302	kcm_done(container_of(w, struct kcm_sock, done_work));
303	}
304
305	/* Lower sock held */
306	static void unreserve_rx_kcm(struct kcm_psock *psock,
307	bool rcv_ready)
308	{
309	struct kcm_sock *kcm = psock->rx_kcm;
310	struct kcm_mux *mux = psock->mux;
311
312	if (!kcm)
313	return;
314
315	spin_lock_bh(lock: &mux->rx_lock);
316
317	psock->rx_kcm = NULL;
318	/* paired with lockless reads in kcm_rfree() */
319	WRITE_ONCE(kcm->rx_psock, NULL);
320
321	/* Commit kcm->rx_psock before sk_rmem_alloc_get to sync with
322	* kcm_rfree
323	*/
324	smp_mb();
325
326	if (unlikely(kcm->done)) {
327	spin_unlock_bh(lock: &mux->rx_lock);
328
329	/* Need to run kcm_done in a task since we need to qcquire
330	* callback locks which may already be held here.
331	*/
332	INIT_WORK(&kcm->done_work, kcm_done_work);
333	schedule_work(work: &kcm->done_work);
334	return;
335	}
336
337	if (unlikely(kcm->rx_disabled)) {
338	requeue_rx_msgs(mux, head: &kcm->sk.sk_receive_queue);
339	} else if (rcv_ready || unlikely(!sk_rmem_alloc_get(&kcm->sk))) {
340	/* Check for degenerative race with rx_wait that all
341	* data was dequeued (accounted for in kcm_rfree).
342	*/
343	kcm_rcv_ready(kcm);
344	}
345	spin_unlock_bh(lock: &mux->rx_lock);
346	}
347
348	/* Lower sock lock held */
349	static void psock_data_ready(struct sock *sk)
350	{
351	struct kcm_psock *psock;
352
353	trace_sk_data_ready(sk);
354
355	read_lock_bh(&sk->sk_callback_lock);
356
357	psock = (struct kcm_psock *)sk->sk_user_data;
358	if (likely(psock))
359	strp_data_ready(strp: &psock->strp);
360
361	read_unlock_bh(&sk->sk_callback_lock);
362	}
363
364	/* Called with lower sock held */
365	static void kcm_rcv_strparser(struct strparser *strp, struct sk_buff *skb)
366	{
367	struct kcm_psock *psock = container_of(strp, struct kcm_psock, strp);
368	struct kcm_sock *kcm;
369
370	try_queue:
371	kcm = reserve_rx_kcm(psock, head: skb);
372	if (!kcm) {
373	/* Unable to reserve a KCM, message is held in psock and strp
374	* is paused.
375	*/
376	return;
377	}
378
379	if (kcm_queue_rcv_skb(sk: &kcm->sk, skb)) {
380	/* Should mean socket buffer full */
381	unreserve_rx_kcm(psock, rcv_ready: false);
382	goto try_queue;
383	}
384	}
385
386	static int kcm_parse_func_strparser(struct strparser *strp, struct sk_buff *skb)
387	{
388	struct kcm_psock *psock = container_of(strp, struct kcm_psock, strp);
389	struct bpf_prog *prog = psock->bpf_prog;
390	int res;
391
392	res = bpf_prog_run_pin_on_cpu(prog, ctx: skb);
393	return res;
394	}
395
396	static int kcm_read_sock_done(struct strparser *strp, int err)
397	{
398	struct kcm_psock *psock = container_of(strp, struct kcm_psock, strp);
399
400	unreserve_rx_kcm(psock, rcv_ready: true);
401
402	return err;
403	}
404
405	static void psock_state_change(struct sock *sk)
406	{
407	/* TCP only does a EPOLLIN for a half close. Do a EPOLLHUP here
408	* since application will normally not poll with EPOLLIN
409	* on the TCP sockets.
410	*/
411
412	report_csk_error(csk: sk, EPIPE);
413	}
414
415	static void psock_write_space(struct sock *sk)
416	{
417	struct kcm_psock *psock;
418	struct kcm_mux *mux;
419	struct kcm_sock *kcm;
420
421	read_lock_bh(&sk->sk_callback_lock);
422
423	psock = (struct kcm_psock *)sk->sk_user_data;
424	if (unlikely(!psock))
425	goto out;
426	mux = psock->mux;
427
428	spin_lock_bh(lock: &mux->lock);
429
430	/* Check if the socket is reserved so someone is waiting for sending. */
431	kcm = psock->tx_kcm;
432	if (kcm && !unlikely(kcm->tx_stopped))
433	queue_work(wq: kcm_wq, work: &kcm->tx_work);
434
435	spin_unlock_bh(lock: &mux->lock);
436	out:
437	read_unlock_bh(&sk->sk_callback_lock);
438	}
439
440	static void unreserve_psock(struct kcm_sock *kcm);
441
442	/* kcm sock is locked. */
443	static struct kcm_psock *reserve_psock(struct kcm_sock *kcm)
444	{
445	struct kcm_mux *mux = kcm->mux;
446	struct kcm_psock *psock;
447
448	psock = kcm->tx_psock;
449
450	smp_rmb(); /* Must read tx_psock before tx_wait */
451
452	if (psock) {
453	WARN_ON(kcm->tx_wait);
454	if (unlikely(psock->tx_stopped))
455	unreserve_psock(kcm);
456	else
457	return kcm->tx_psock;
458	}
459
460	spin_lock_bh(lock: &mux->lock);
461
462	/* Check again under lock to see if psock was reserved for this
463	* psock via psock_unreserve.
464	*/
465	psock = kcm->tx_psock;
466	if (unlikely(psock)) {
467	WARN_ON(kcm->tx_wait);
468	spin_unlock_bh(lock: &mux->lock);
469	return kcm->tx_psock;
470	}
471
472	if (!list_empty(head: &mux->psocks_avail)) {
473	psock = list_first_entry(&mux->psocks_avail,
474	struct kcm_psock,
475	psock_avail_list);
476	list_del(entry: &psock->psock_avail_list);
477	if (kcm->tx_wait) {
478	list_del(entry: &kcm->wait_psock_list);
479	kcm->tx_wait = false;
480	}
481	kcm->tx_psock = psock;
482	psock->tx_kcm = kcm;
483	KCM_STATS_INCR(psock->stats.reserved);
484	} else if (!kcm->tx_wait) {
485	list_add_tail(new: &kcm->wait_psock_list,
486	head: &mux->kcm_tx_waiters);
487	kcm->tx_wait = true;
488	}
489
490	spin_unlock_bh(lock: &mux->lock);
491
492	return psock;
493	}
494
495	/* mux lock held */
496	static void psock_now_avail(struct kcm_psock *psock)
497	{
498	struct kcm_mux *mux = psock->mux;
499	struct kcm_sock *kcm;
500
501	if (list_empty(head: &mux->kcm_tx_waiters)) {
502	list_add_tail(new: &psock->psock_avail_list,
503	head: &mux->psocks_avail);
504	} else {
505	kcm = list_first_entry(&mux->kcm_tx_waiters,
506	struct kcm_sock,
507	wait_psock_list);
508	list_del(entry: &kcm->wait_psock_list);
509	kcm->tx_wait = false;
510	psock->tx_kcm = kcm;
511
512	/* Commit before changing tx_psock since that is read in
513	* reserve_psock before queuing work.
514	*/
515	smp_mb();
516
517	kcm->tx_psock = psock;
518	KCM_STATS_INCR(psock->stats.reserved);
519	queue_work(wq: kcm_wq, work: &kcm->tx_work);
520	}
521	}
522
523	/* kcm sock is locked. */
524	static void unreserve_psock(struct kcm_sock *kcm)
525	{
526	struct kcm_psock *psock;
527	struct kcm_mux *mux = kcm->mux;
528
529	spin_lock_bh(lock: &mux->lock);
530
531	psock = kcm->tx_psock;
532
533	if (WARN_ON(!psock)) {
534	spin_unlock_bh(lock: &mux->lock);
535	return;
536	}
537
538	smp_rmb(); /* Read tx_psock before tx_wait */
539
540	kcm_update_tx_mux_stats(mux, psock);
541
542	WARN_ON(kcm->tx_wait);
543
544	kcm->tx_psock = NULL;
545	psock->tx_kcm = NULL;
546	KCM_STATS_INCR(psock->stats.unreserved);
547
548	if (unlikely(psock->tx_stopped)) {
549	if (psock->done) {
550	/* Deferred free */
551	list_del(entry: &psock->psock_list);
552	mux->psocks_cnt--;
553	sock_put(sk: psock->sk);
554	fput(psock->sk->sk_socket->file);
555	kmem_cache_free(s: kcm_psockp, objp: psock);
556	}
557
558	/* Don't put back on available list */
559
560	spin_unlock_bh(lock: &mux->lock);
561
562	return;
563	}
564
565	psock_now_avail(psock);
566
567	spin_unlock_bh(lock: &mux->lock);
568	}
569
570	static void kcm_report_tx_retry(struct kcm_sock *kcm)
571	{
572	struct kcm_mux *mux = kcm->mux;
573
574	spin_lock_bh(lock: &mux->lock);
575	KCM_STATS_INCR(mux->stats.tx_retries);
576	spin_unlock_bh(lock: &mux->lock);
577	}
578
579	/* Write any messages ready on the kcm socket. Called with kcm sock lock
580	* held. Return bytes actually sent or error.
581	*/
582	static int kcm_write_msgs(struct kcm_sock *kcm)
583	{
584	unsigned int total_sent = 0;
585	struct sock *sk = &kcm->sk;
586	struct kcm_psock *psock;
587	struct sk_buff *head;
588	int ret = 0;
589
590	kcm->tx_wait_more = false;
591	psock = kcm->tx_psock;
592	if (unlikely(psock && psock->tx_stopped)) {
593	/* A reserved psock was aborted asynchronously. Unreserve
594	* it and we'll retry the message.
595	*/
596	unreserve_psock(kcm);
597	kcm_report_tx_retry(kcm);
598	if (skb_queue_empty(list: &sk->sk_write_queue))
599	return 0;
600
601	kcm_tx_msg(skb: skb_peek(list_: &sk->sk_write_queue))->started_tx = false;
602	}
603
604	retry:
605	while ((head = skb_peek(list_: &sk->sk_write_queue))) {
606	struct msghdr msg = {
607	.msg_flags = MSG_DONTWAIT | MSG_SPLICE_PAGES,
608	};
609	struct kcm_tx_msg *txm = kcm_tx_msg(skb: head);
610	struct sk_buff *skb;
611	unsigned int msize;
612	int i;
613
614	if (!txm->started_tx) {
615	psock = reserve_psock(kcm);
616	if (!psock)
617	goto out;
618	skb = head;
619	txm->frag_offset = 0;
620	txm->sent = 0;
621	txm->started_tx = true;
622	} else {
623	if (WARN_ON(!psock)) {
624	ret = -EINVAL;
625	goto out;
626	}
627	skb = txm->frag_skb;
628	}
629
630	if (WARN_ON(!skb_shinfo(skb)->nr_frags) ||
631	WARN_ON_ONCE(!skb_frag_page(&skb_shinfo(skb)->frags[0]))) {
632	ret = -EINVAL;
633	goto out;
634	}
635
636	msize = 0;
637	for (i = 0; i < skb_shinfo(skb)->nr_frags; i++)
638	msize += skb_frag_size(frag: &skb_shinfo(skb)->frags[i]);
639
640	iov_iter_bvec(i: &msg.msg_iter, ITER_SOURCE,
641	bvec: (const struct bio_vec *)skb_shinfo(skb)->frags,
642	skb_shinfo(skb)->nr_frags, count: msize);
643	iov_iter_advance(i: &msg.msg_iter, bytes: txm->frag_offset);
644
645	do {
646	ret = sock_sendmsg(sock: psock->sk->sk_socket, msg: &msg);
647	if (ret <= 0) {
648	if (ret == -EAGAIN) {
649	/* Save state to try again when there's
650	* write space on the socket
651	*/
652	txm->frag_skb = skb;
653	ret = 0;
654	goto out;
655	}
656
657	/* Hard failure in sending message, abort this
658	* psock since it has lost framing
659	* synchronization and retry sending the
660	* message from the beginning.
661	*/
662	kcm_abort_tx_psock(psock, err: ret ? -ret : EPIPE,
663	wakeup_kcm: true);
664	unreserve_psock(kcm);
665	psock = NULL;
666
667	txm->started_tx = false;
668	kcm_report_tx_retry(kcm);
669	ret = 0;
670	goto retry;
671	}
672
673	txm->sent += ret;
674	txm->frag_offset += ret;
675	KCM_STATS_ADD(psock->stats.tx_bytes, ret);
676	} while (msg.msg_iter.count > 0);
677
678	if (skb == head) {
679	if (skb_has_frag_list(skb)) {
680	txm->frag_skb = skb_shinfo(skb)->frag_list;
681	txm->frag_offset = 0;
682	continue;
683	}
684	} else if (skb->next) {
685	txm->frag_skb = skb->next;
686	txm->frag_offset = 0;
687	continue;
688	}
689
690	/* Successfully sent the whole packet, account for it. */
691	sk->sk_wmem_queued -= txm->sent;
692	total_sent += txm->sent;
693	skb_dequeue(list: &sk->sk_write_queue);
694	kfree_skb(skb: head);
695	KCM_STATS_INCR(psock->stats.tx_msgs);
696	}
697	out:
698	if (!head) {
699	/* Done with all queued messages. */
700	WARN_ON(!skb_queue_empty(&sk->sk_write_queue));
701	if (psock)
702	unreserve_psock(kcm);
703	}
704
705	/* Check if write space is available */
706	sk->sk_write_space(sk);
707
708	return total_sent ? : ret;
709	}
710
711	static void kcm_tx_work(struct work_struct *w)
712	{
713	struct kcm_sock *kcm = container_of(w, struct kcm_sock, tx_work);
714	struct sock *sk = &kcm->sk;
715	int err;
716
717	lock_sock(sk);
718
719	/* Primarily for SOCK_DGRAM sockets, also handle asynchronous tx
720	* aborts
721	*/
722	err = kcm_write_msgs(kcm);
723	if (err < 0) {
724	/* Hard failure in write, report error on KCM socket */
725	pr_warn("KCM: Hard failure on kcm_write_msgs %d\n", err);
726	report_csk_error(csk: &kcm->sk, err: -err);
727	goto out;
728	}
729
730	/* Primarily for SOCK_SEQPACKET sockets */
731	if (likely(sk->sk_socket) &&
732	test_bit(SOCK_NOSPACE, &sk->sk_socket->flags)) {
733	clear_bit(SOCK_NOSPACE, addr: &sk->sk_socket->flags);
734	sk->sk_write_space(sk);
735	}
736
737	out:
738	release_sock(sk);
739	}
740
741	static void kcm_push(struct kcm_sock *kcm)
742	{
743	if (kcm->tx_wait_more)
744	kcm_write_msgs(kcm);
745	}
746
747	static int kcm_sendmsg(struct socket *sock, struct msghdr *msg, size_t len)
748	{
749	struct sock *sk = sock->sk;
750	struct kcm_sock *kcm = kcm_sk(sk);
751	struct sk_buff *skb = NULL, *head = NULL;
752	size_t copy, copied = 0;
753	long timeo = sock_sndtimeo(sk, noblock: msg->msg_flags & MSG_DONTWAIT);
754	int eor = (sock->type == SOCK_DGRAM) ?
755	!(msg->msg_flags & MSG_MORE) : !!(msg->msg_flags & MSG_EOR);
756	int err = -EPIPE;
757
758	lock_sock(sk);
759
760	/* Per tcp_sendmsg this should be in poll */
761	sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
762
763	if (sk->sk_err)
764	goto out_error;
765
766	if (kcm->seq_skb) {
767	/* Previously opened message */
768	head = kcm->seq_skb;
769	skb = kcm_tx_msg(skb: head)->last_skb;
770	goto start;
771	}
772
773	/* Call the sk_stream functions to manage the sndbuf mem. */
774	if (!sk_stream_memory_free(sk)) {
775	kcm_push(kcm);
776	set_bit(SOCK_NOSPACE, addr: &sk->sk_socket->flags);
777	err = sk_stream_wait_memory(sk, timeo_p: &timeo);
778	if (err)
779	goto out_error;
780	}
781
782	if (msg_data_left(msg)) {
783	/* New message, alloc head skb */
784	head = alloc_skb(size: 0, priority: sk->sk_allocation);
785	while (!head) {
786	kcm_push(kcm);
787	err = sk_stream_wait_memory(sk, timeo_p: &timeo);
788	if (err)
789	goto out_error;
790
791	head = alloc_skb(size: 0, priority: sk->sk_allocation);
792	}
793
794	skb = head;
795
796	/* Set ip_summed to CHECKSUM_UNNECESSARY to avoid calling
797	* csum_and_copy_from_iter from skb_do_copy_data_nocache.
798	*/
799	skb->ip_summed = CHECKSUM_UNNECESSARY;
800	}
801
802	start:
803	while (msg_data_left(msg)) {
804	bool merge = true;
805	int i = skb_shinfo(skb)->nr_frags;
806	struct page_frag *pfrag = sk_page_frag(sk);
807
808	if (!sk_page_frag_refill(sk, pfrag))
809	goto wait_for_memory;
810
811	if (!skb_can_coalesce(skb, i, page: pfrag->page,
812	off: pfrag->offset)) {
813	if (i == MAX_SKB_FRAGS) {
814	struct sk_buff *tskb;
815
816	tskb = alloc_skb(size: 0, priority: sk->sk_allocation);
817	if (!tskb)
818	goto wait_for_memory;
819
820	if (head == skb)
821	skb_shinfo(head)->frag_list = tskb;
822	else
823	skb->next = tskb;
824
825	skb = tskb;
826	skb->ip_summed = CHECKSUM_UNNECESSARY;
827	continue;
828	}
829	merge = false;
830	}
831
832	if (msg->msg_flags & MSG_SPLICE_PAGES) {
833	copy = msg_data_left(msg);
834	if (!sk_wmem_schedule(sk, size: copy))
835	goto wait_for_memory;
836
837	err = skb_splice_from_iter(skb, iter: &msg->msg_iter, maxsize: copy,
838	gfp: sk->sk_allocation);
839	if (err < 0) {
840	if (err == -EMSGSIZE)
841	goto wait_for_memory;
842	goto out_error;
843	}
844
845	copy = err;
846	skb_shinfo(skb)->flags |= SKBFL_SHARED_FRAG;
847	sk_wmem_queued_add(sk, val: copy);
848	sk_mem_charge(sk, size: copy);
849
850	if (head != skb)
851	head->truesize += copy;
852	} else {
853	copy = min_t(int, msg_data_left(msg),
854	pfrag->size - pfrag->offset);
855	if (!sk_wmem_schedule(sk, size: copy))
856	goto wait_for_memory;
857
858	err = skb_copy_to_page_nocache(sk, from: &msg->msg_iter, skb,
859	page: pfrag->page,
860	off: pfrag->offset,
861	copy);
862	if (err)
863	goto out_error;
864
865	/* Update the skb. */
866	if (merge) {
867	skb_frag_size_add(
868	frag: &skb_shinfo(skb)->frags[i - 1], delta: copy);
869	} else {
870	skb_fill_page_desc(skb, i, page: pfrag->page,
871	off: pfrag->offset, size: copy);
872	get_page(page: pfrag->page);
873	}
874
875	pfrag->offset += copy;
876	}
877
878	copied += copy;
879	if (head != skb) {
880	head->len += copy;
881	head->data_len += copy;
882	}
883
884	continue;
885
886	wait_for_memory:
887	kcm_push(kcm);
888	err = sk_stream_wait_memory(sk, timeo_p: &timeo);
889	if (err)
890	goto out_error;
891	}
892
893	if (eor) {
894	bool not_busy = skb_queue_empty(list: &sk->sk_write_queue);
895
896	if (head) {
897	/* Message complete, queue it on send buffer */
898	__skb_queue_tail(list: &sk->sk_write_queue, newsk: head);
899	kcm->seq_skb = NULL;
900	KCM_STATS_INCR(kcm->stats.tx_msgs);
901	}
902
903	if (msg->msg_flags & MSG_BATCH) {
904	kcm->tx_wait_more = true;
905	} else if (kcm->tx_wait_more || not_busy) {
906	err = kcm_write_msgs(kcm);
907	if (err < 0) {
908	/* We got a hard error in write_msgs but have
909	* already queued this message. Report an error
910	* in the socket, but don't affect return value
911	* from sendmsg
912	*/
913	pr_warn("KCM: Hard failure on kcm_write_msgs\n");
914	report_csk_error(csk: &kcm->sk, err: -err);
915	}
916	}
917	} else {
918	/* Message not complete, save state */
919	partial_message:
920	if (head) {
921	kcm->seq_skb = head;
922	kcm_tx_msg(skb: head)->last_skb = skb;
923	}
924	}
925
926	KCM_STATS_ADD(kcm->stats.tx_bytes, copied);
927
928	release_sock(sk);
929	return copied;
930
931	out_error:
932	kcm_push(kcm);
933
934	if (sock->type == SOCK_SEQPACKET) {
935	/* Wrote some bytes before encountering an
936	* error, return partial success.
937	*/
938	if (copied)
939	goto partial_message;
940	if (head != kcm->seq_skb)
941	kfree_skb(skb: head);
942	} else {
943	kfree_skb(skb: head);
944	kcm->seq_skb = NULL;
945	}
946
947	err = sk_stream_error(sk, flags: msg->msg_flags, err);
948
949	/* make sure we wake any epoll edge trigger waiter */
950	if (unlikely(skb_queue_len(&sk->sk_write_queue) == 0 && err == -EAGAIN))
951	sk->sk_write_space(sk);
952
953	release_sock(sk);
954	return err;
955	}
956
957	static void kcm_splice_eof(struct socket *sock)
958	{
959	struct sock *sk = sock->sk;
960	struct kcm_sock *kcm = kcm_sk(sk);
961
962	if (skb_queue_empty_lockless(list: &sk->sk_write_queue))
963	return;
964
965	lock_sock(sk);
966	kcm_write_msgs(kcm);
967	release_sock(sk);
968	}
969
970	static int kcm_recvmsg(struct socket *sock, struct msghdr *msg,
971	size_t len, int flags)
972	{
973	struct sock *sk = sock->sk;
974	struct kcm_sock *kcm = kcm_sk(sk);
975	int err = 0;
976	struct strp_msg *stm;
977	int copied = 0;
978	struct sk_buff *skb;
979
980	skb = skb_recv_datagram(sk, flags, err: &err);
981	if (!skb)
982	goto out;
983
984	/* Okay, have a message on the receive queue */
985
986	stm = strp_msg(skb);
987
988	if (len > stm->full_len)
989	len = stm->full_len;
990
991	err = skb_copy_datagram_msg(from: skb, offset: stm->offset, msg, size: len);
992	if (err < 0)
993	goto out;
994
995	copied = len;
996	if (likely(!(flags & MSG_PEEK))) {
997	KCM_STATS_ADD(kcm->stats.rx_bytes, copied);
998	if (copied < stm->full_len) {
999	if (sock->type == SOCK_DGRAM) {
1000	/* Truncated message */
1001	msg->msg_flags |= MSG_TRUNC;
1002	goto msg_finished;
1003	}
1004	stm->offset += copied;
1005	stm->full_len -= copied;
1006	} else {
1007	msg_finished:
1008	/* Finished with message */
1009	msg->msg_flags |= MSG_EOR;
1010	KCM_STATS_INCR(kcm->stats.rx_msgs);
1011	}
1012	}
1013
1014	out:
1015	skb_free_datagram(sk, skb);
1016	return copied ? : err;
1017	}
1018
1019	static ssize_t kcm_splice_read(struct socket *sock, loff_t *ppos,
1020	struct pipe_inode_info *pipe, size_t len,
1021	unsigned int flags)
1022	{
1023	struct sock *sk = sock->sk;
1024	struct kcm_sock *kcm = kcm_sk(sk);
1025	struct strp_msg *stm;
1026	int err = 0;
1027	ssize_t copied;
1028	struct sk_buff *skb;
1029
1030	/* Only support splice for SOCKSEQPACKET */
1031
1032	skb = skb_recv_datagram(sk, flags, err: &err);
1033	if (!skb)
1034	goto err_out;
1035
1036	/* Okay, have a message on the receive queue */
1037
1038	stm = strp_msg(skb);
1039
1040	if (len > stm->full_len)
1041	len = stm->full_len;
1042
1043	copied = skb_splice_bits(skb, sk, offset: stm->offset, pipe, len, flags);
1044	if (copied < 0) {
1045	err = copied;
1046	goto err_out;
1047	}
1048
1049	KCM_STATS_ADD(kcm->stats.rx_bytes, copied);
1050
1051	stm->offset += copied;
1052	stm->full_len -= copied;
1053
1054	/* We have no way to return MSG_EOR. If all the bytes have been
1055	* read we still leave the message in the receive socket buffer.
1056	* A subsequent recvmsg needs to be done to return MSG_EOR and
1057	* finish reading the message.
1058	*/
1059
1060	skb_free_datagram(sk, skb);
1061	return copied;
1062
1063	err_out:
1064	skb_free_datagram(sk, skb);
1065	return err;
1066	}
1067
1068	/* kcm sock lock held */
1069	static void kcm_recv_disable(struct kcm_sock *kcm)
1070	{
1071	struct kcm_mux *mux = kcm->mux;
1072
1073	if (kcm->rx_disabled)
1074	return;
1075
1076	spin_lock_bh(lock: &mux->rx_lock);
1077
1078	kcm->rx_disabled = 1;
1079
1080	/* If a psock is reserved we'll do cleanup in unreserve */
1081	if (!kcm->rx_psock) {
1082	if (kcm->rx_wait) {
1083	list_del(entry: &kcm->wait_rx_list);
1084	/* paired with lockless reads in kcm_rfree() */
1085	WRITE_ONCE(kcm->rx_wait, false);
1086	}
1087
1088	requeue_rx_msgs(mux, head: &kcm->sk.sk_receive_queue);
1089	}
1090
1091	spin_unlock_bh(lock: &mux->rx_lock);
1092	}
1093
1094	/* kcm sock lock held */
1095	static void kcm_recv_enable(struct kcm_sock *kcm)
1096	{
1097	struct kcm_mux *mux = kcm->mux;
1098
1099	if (!kcm->rx_disabled)
1100	return;
1101
1102	spin_lock_bh(lock: &mux->rx_lock);
1103
1104	kcm->rx_disabled = 0;
1105	kcm_rcv_ready(kcm);
1106
1107	spin_unlock_bh(lock: &mux->rx_lock);
1108	}
1109
1110	static int kcm_setsockopt(struct socket *sock, int level, int optname,
1111	sockptr_t optval, unsigned int optlen)
1112	{
1113	struct kcm_sock *kcm = kcm_sk(sk: sock->sk);
1114	int val, valbool;
1115	int err = 0;
1116
1117	if (level != SOL_KCM)
1118	return -ENOPROTOOPT;
1119
1120	if (optlen < sizeof(int))
1121	return -EINVAL;
1122
1123	if (copy_from_sockptr(dst: &val, src: optval, size: sizeof(int)))
1124	return -EFAULT;
1125
1126	valbool = val ? 1 : 0;
1127
1128	switch (optname) {
1129	case KCM_RECV_DISABLE:
1130	lock_sock(sk: &kcm->sk);
1131	if (valbool)
1132	kcm_recv_disable(kcm);
1133	else
1134	kcm_recv_enable(kcm);
1135	release_sock(sk: &kcm->sk);
1136	break;
1137	default:
1138	err = -ENOPROTOOPT;
1139	}
1140
1141	return err;
1142	}
1143
1144	static int kcm_getsockopt(struct socket *sock, int level, int optname,
1145	char __user *optval, int __user *optlen)
1146	{
1147	struct kcm_sock *kcm = kcm_sk(sk: sock->sk);
1148	int val, len;
1149
1150	if (level != SOL_KCM)
1151	return -ENOPROTOOPT;
1152
1153	if (get_user(len, optlen))
1154	return -EFAULT;
1155
1156	if (len < 0)
1157	return -EINVAL;
1158
1159	len = min_t(unsigned int, len, sizeof(int));
1160
1161	switch (optname) {
1162	case KCM_RECV_DISABLE:
1163	val = kcm->rx_disabled;
1164	break;
1165	default:
1166	return -ENOPROTOOPT;
1167	}
1168
1169	if (put_user(len, optlen))
1170	return -EFAULT;
1171	if (copy_to_user(to: optval, from: &val, n: len))
1172	return -EFAULT;
1173	return 0;
1174	}
1175
1176	static void init_kcm_sock(struct kcm_sock *kcm, struct kcm_mux *mux)
1177	{
1178	struct kcm_sock *tkcm;
1179	struct list_head *head;
1180	int index = 0;
1181
1182	/* For SOCK_SEQPACKET sock type, datagram_poll checks the sk_state, so
1183	* we set sk_state, otherwise epoll_wait always returns right away with
1184	* EPOLLHUP
1185	*/
1186	kcm->sk.sk_state = TCP_ESTABLISHED;
1187
1188	/* Add to mux's kcm sockets list */
1189	kcm->mux = mux;
1190	spin_lock_bh(lock: &mux->lock);
1191
1192	head = &mux->kcm_socks;
1193	list_for_each_entry(tkcm, &mux->kcm_socks, kcm_sock_list) {
1194	if (tkcm->index != index)
1195	break;
1196	head = &tkcm->kcm_sock_list;
1197	index++;
1198	}
1199
1200	list_add(new: &kcm->kcm_sock_list, head);
1201	kcm->index = index;
1202
1203	mux->kcm_socks_cnt++;
1204	spin_unlock_bh(lock: &mux->lock);
1205
1206	INIT_WORK(&kcm->tx_work, kcm_tx_work);
1207
1208	spin_lock_bh(lock: &mux->rx_lock);
1209	kcm_rcv_ready(kcm);
1210	spin_unlock_bh(lock: &mux->rx_lock);
1211	}
1212
1213	static int kcm_attach(struct socket *sock, struct socket *csock,
1214	struct bpf_prog *prog)
1215	{
1216	struct kcm_sock *kcm = kcm_sk(sk: sock->sk);
1217	struct kcm_mux *mux = kcm->mux;
1218	struct sock *csk;
1219	struct kcm_psock *psock = NULL, *tpsock;
1220	struct list_head *head;
1221	int index = 0;
1222	static const struct strp_callbacks cb = {
1223	.rcv_msg = kcm_rcv_strparser,
1224	.parse_msg = kcm_parse_func_strparser,
1225	.read_sock_done = kcm_read_sock_done,
1226	};
1227	int err = 0;
1228
1229	csk = csock->sk;
1230	if (!csk)
1231	return -EINVAL;
1232
1233	lock_sock(sk: csk);
1234
1235	/* Only allow TCP sockets to be attached for now */
1236	if ((csk->sk_family != AF_INET && csk->sk_family != AF_INET6) ||
1237	csk->sk_protocol != IPPROTO_TCP) {
1238	err = -EOPNOTSUPP;
1239	goto out;
1240	}
1241
1242	/* Don't allow listeners or closed sockets */
1243	if (csk->sk_state == TCP_LISTEN || csk->sk_state == TCP_CLOSE) {
1244	err = -EOPNOTSUPP;
1245	goto out;
1246	}
1247
1248	psock = kmem_cache_zalloc(k: kcm_psockp, GFP_KERNEL);
1249	if (!psock) {
1250	err = -ENOMEM;
1251	goto out;
1252	}
1253
1254	psock->mux = mux;
1255	psock->sk = csk;
1256	psock->bpf_prog = prog;
1257
1258	write_lock_bh(&csk->sk_callback_lock);
1259
1260	/* Check if sk_user_data is already by KCM or someone else.
1261	* Must be done under lock to prevent race conditions.
1262	*/
1263	if (csk->sk_user_data) {
1264	write_unlock_bh(&csk->sk_callback_lock);
1265	kmem_cache_free(s: kcm_psockp, objp: psock);
1266	err = -EALREADY;
1267	goto out;
1268	}
1269
1270	err = strp_init(strp: &psock->strp, sk: csk, cb: &cb);
1271	if (err) {
1272	write_unlock_bh(&csk->sk_callback_lock);
1273	kmem_cache_free(s: kcm_psockp, objp: psock);
1274	goto out;
1275	}
1276
1277	psock->save_data_ready = csk->sk_data_ready;
1278	psock->save_write_space = csk->sk_write_space;
1279	psock->save_state_change = csk->sk_state_change;
1280	csk->sk_user_data = psock;
1281	csk->sk_data_ready = psock_data_ready;
1282	csk->sk_write_space = psock_write_space;
1283	csk->sk_state_change = psock_state_change;
1284
1285	write_unlock_bh(&csk->sk_callback_lock);
1286
1287	sock_hold(sk: csk);
1288
1289	/* Finished initialization, now add the psock to the MUX. */
1290	spin_lock_bh(lock: &mux->lock);
1291	head = &mux->psocks;
1292	list_for_each_entry(tpsock, &mux->psocks, psock_list) {
1293	if (tpsock->index != index)
1294	break;
1295	head = &tpsock->psock_list;
1296	index++;
1297	}
1298
1299	list_add(new: &psock->psock_list, head);
1300	psock->index = index;
1301
1302	KCM_STATS_INCR(mux->stats.psock_attach);
1303	mux->psocks_cnt++;
1304	psock_now_avail(psock);
1305	spin_unlock_bh(lock: &mux->lock);
1306
1307	/* Schedule RX work in case there are already bytes queued */
1308	strp_check_rcv(strp: &psock->strp);
1309
1310	out:
1311	release_sock(sk: csk);
1312
1313	return err;
1314	}
1315
1316	static int kcm_attach_ioctl(struct socket *sock, struct kcm_attach *info)
1317	{
1318	struct socket *csock;
1319	struct bpf_prog *prog;
1320	int err;
1321
1322	csock = sockfd_lookup(fd: info->fd, err: &err);
1323	if (!csock)
1324	return -ENOENT;
1325
1326	prog = bpf_prog_get_type(ufd: info->bpf_fd, type: BPF_PROG_TYPE_SOCKET_FILTER);
1327	if (IS_ERR(ptr: prog)) {
1328	err = PTR_ERR(ptr: prog);
1329	goto out;
1330	}
1331
1332	err = kcm_attach(sock, csock, prog);
1333	if (err) {
1334	bpf_prog_put(prog);
1335	goto out;
1336	}
1337
1338	/* Keep reference on file also */
1339
1340	return 0;
1341	out:
1342	sockfd_put(csock);
1343	return err;
1344	}
1345
1346	static void kcm_unattach(struct kcm_psock *psock)
1347	{
1348	struct sock *csk = psock->sk;
1349	struct kcm_mux *mux = psock->mux;
1350
1351	lock_sock(sk: csk);
1352
1353	/* Stop getting callbacks from TCP socket. After this there should
1354	* be no way to reserve a kcm for this psock.
1355	*/
1356	write_lock_bh(&csk->sk_callback_lock);
1357	csk->sk_user_data = NULL;
1358	csk->sk_data_ready = psock->save_data_ready;
1359	csk->sk_write_space = psock->save_write_space;
1360	csk->sk_state_change = psock->save_state_change;
1361	strp_stop(strp: &psock->strp);
1362
1363	if (WARN_ON(psock->rx_kcm)) {
1364	write_unlock_bh(&csk->sk_callback_lock);
1365	release_sock(sk: csk);
1366	return;
1367	}
1368
1369	spin_lock_bh(lock: &mux->rx_lock);
1370
1371	/* Stop receiver activities. After this point psock should not be
1372	* able to get onto ready list either through callbacks or work.
1373	*/
1374	if (psock->ready_rx_msg) {
1375	list_del(entry: &psock->psock_ready_list);
1376	kfree_skb(skb: psock->ready_rx_msg);
1377	psock->ready_rx_msg = NULL;
1378	KCM_STATS_INCR(mux->stats.rx_ready_drops);
1379	}
1380
1381	spin_unlock_bh(lock: &mux->rx_lock);
1382
1383	write_unlock_bh(&csk->sk_callback_lock);
1384
1385	/* Call strp_done without sock lock */
1386	release_sock(sk: csk);
1387	strp_done(strp: &psock->strp);
1388	lock_sock(sk: csk);
1389
1390	bpf_prog_put(prog: psock->bpf_prog);
1391
1392	spin_lock_bh(lock: &mux->lock);
1393
1394	aggregate_psock_stats(stats: &psock->stats, agg_stats: &mux->aggregate_psock_stats);
1395	save_strp_stats(strp: &psock->strp, agg_stats: &mux->aggregate_strp_stats);
1396
1397	KCM_STATS_INCR(mux->stats.psock_unattach);
1398
1399	if (psock->tx_kcm) {
1400	/* psock was reserved. Just mark it finished and we will clean
1401	* up in the kcm paths, we need kcm lock which can not be
1402	* acquired here.
1403	*/
1404	KCM_STATS_INCR(mux->stats.psock_unattach_rsvd);
1405	spin_unlock_bh(lock: &mux->lock);
1406
1407	/* We are unattaching a socket that is reserved. Abort the
1408	* socket since we may be out of sync in sending on it. We need
1409	* to do this without the mux lock.
1410	*/
1411	kcm_abort_tx_psock(psock, EPIPE, wakeup_kcm: false);
1412
1413	spin_lock_bh(lock: &mux->lock);
1414	if (!psock->tx_kcm) {
1415	/* psock now unreserved in window mux was unlocked */
1416	goto no_reserved;
1417	}
1418	psock->done = 1;
1419
1420	/* Commit done before queuing work to process it */
1421	smp_mb();
1422
1423	/* Queue tx work to make sure psock->done is handled */
1424	queue_work(wq: kcm_wq, work: &psock->tx_kcm->tx_work);
1425	spin_unlock_bh(lock: &mux->lock);
1426	} else {
1427	no_reserved:
1428	if (!psock->tx_stopped)
1429	list_del(entry: &psock->psock_avail_list);
1430	list_del(entry: &psock->psock_list);
1431	mux->psocks_cnt--;
1432	spin_unlock_bh(lock: &mux->lock);
1433
1434	sock_put(sk: csk);
1435	fput(csk->sk_socket->file);
1436	kmem_cache_free(s: kcm_psockp, objp: psock);
1437	}
1438
1439	release_sock(sk: csk);
1440	}
1441
1442	static int kcm_unattach_ioctl(struct socket *sock, struct kcm_unattach *info)
1443	{
1444	struct kcm_sock *kcm = kcm_sk(sk: sock->sk);
1445	struct kcm_mux *mux = kcm->mux;
1446	struct kcm_psock *psock;
1447	struct socket *csock;
1448	struct sock *csk;
1449	int err;
1450
1451	csock = sockfd_lookup(fd: info->fd, err: &err);
1452	if (!csock)
1453	return -ENOENT;
1454
1455	csk = csock->sk;
1456	if (!csk) {
1457	err = -EINVAL;
1458	goto out;
1459	}
1460
1461	err = -ENOENT;
1462
1463	spin_lock_bh(lock: &mux->lock);
1464
1465	list_for_each_entry(psock, &mux->psocks, psock_list) {
1466	if (psock->sk != csk)
1467	continue;
1468
1469	/* Found the matching psock */
1470
1471	if (psock->unattaching || WARN_ON(psock->done)) {
1472	err = -EALREADY;
1473	break;
1474	}
1475
1476	psock->unattaching = 1;
1477
1478	spin_unlock_bh(lock: &mux->lock);
1479
1480	/* Lower socket lock should already be held */
1481	kcm_unattach(psock);
1482
1483	err = 0;
1484	goto out;
1485	}
1486
1487	spin_unlock_bh(lock: &mux->lock);
1488
1489	out:
1490	sockfd_put(csock);
1491	return err;
1492	}
1493
1494	static struct proto kcm_proto = {
1495	.name = "KCM",
1496	.owner = THIS_MODULE,
1497	.obj_size = sizeof(struct kcm_sock),
1498	};
1499
1500	/* Clone a kcm socket. */
1501	static struct file *kcm_clone(struct socket *osock)
1502	{
1503	struct socket *newsock;
1504	struct sock *newsk;
1505
1506	newsock = sock_alloc();
1507	if (!newsock)
1508	return ERR_PTR(error: -ENFILE);
1509
1510	newsock->type = osock->type;
1511	newsock->ops = osock->ops;
1512
1513	__module_get(module: newsock->ops->owner);
1514
1515	newsk = sk_alloc(net: sock_net(sk: osock->sk), PF_KCM, GFP_KERNEL,
1516	prot: &kcm_proto, kern: false);
1517	if (!newsk) {
1518	sock_release(sock: newsock);
1519	return ERR_PTR(error: -ENOMEM);
1520	}
1521	sock_init_data(sock: newsock, sk: newsk);
1522	init_kcm_sock(kcm: kcm_sk(sk: newsk), mux: kcm_sk(sk: osock->sk)->mux);
1523
1524	return sock_alloc_file(sock: newsock, flags: 0, dname: osock->sk->sk_prot_creator->name);
1525	}
1526
1527	static int kcm_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
1528	{
1529	int err;
1530
1531	switch (cmd) {
1532	case SIOCKCMATTACH: {
1533	struct kcm_attach info;
1534
1535	if (copy_from_user(to: &info, from: (void __user *)arg, n: sizeof(info)))
1536	return -EFAULT;
1537
1538	err = kcm_attach_ioctl(sock, info: &info);
1539
1540	break;
1541	}
1542	case SIOCKCMUNATTACH: {
1543	struct kcm_unattach info;
1544
1545	if (copy_from_user(to: &info, from: (void __user *)arg, n: sizeof(info)))
1546	return -EFAULT;
1547
1548	err = kcm_unattach_ioctl(sock, info: &info);
1549
1550	break;
1551	}
1552	case SIOCKCMCLONE: {
1553	struct kcm_clone info;
1554	struct file *file;
1555
1556	info.fd = get_unused_fd_flags(flags: 0);
1557	if (unlikely(info.fd < 0))
1558	return info.fd;
1559
1560	file = kcm_clone(osock: sock);
1561	if (IS_ERR(ptr: file)) {
1562	put_unused_fd(fd: info.fd);
1563	return PTR_ERR(ptr: file);
1564	}
1565	if (copy_to_user(to: (void __user *)arg, from: &info,
1566	n: sizeof(info))) {
1567	put_unused_fd(fd: info.fd);
1568	fput(file);
1569	return -EFAULT;
1570	}
1571	fd_install(fd: info.fd, file);
1572	err = 0;
1573	break;
1574	}
1575	default:
1576	err = -ENOIOCTLCMD;
1577	break;
1578	}
1579
1580	return err;
1581	}
1582
1583	static void free_mux(struct rcu_head *rcu)
1584	{
1585	struct kcm_mux *mux = container_of(rcu,
1586	struct kcm_mux, rcu);
1587
1588	kmem_cache_free(s: kcm_muxp, objp: mux);
1589	}
1590
1591	static void release_mux(struct kcm_mux *mux)
1592	{
1593	struct kcm_net *knet = mux->knet;
1594	struct kcm_psock *psock, *tmp_psock;
1595
1596	/* Release psocks */
1597	list_for_each_entry_safe(psock, tmp_psock,
1598	&mux->psocks, psock_list) {
1599	if (!WARN_ON(psock->unattaching))
1600	kcm_unattach(psock);
1601	}
1602
1603	if (WARN_ON(mux->psocks_cnt))
1604	return;
1605
1606	__skb_queue_purge(list: &mux->rx_hold_queue);
1607
1608	mutex_lock(&knet->mutex);
1609	aggregate_mux_stats(stats: &mux->stats, agg_stats: &knet->aggregate_mux_stats);
1610	aggregate_psock_stats(stats: &mux->aggregate_psock_stats,
1611	agg_stats: &knet->aggregate_psock_stats);
1612	aggregate_strp_stats(stats: &mux->aggregate_strp_stats,
1613	agg_stats: &knet->aggregate_strp_stats);
1614	list_del_rcu(entry: &mux->kcm_mux_list);
1615	knet->count--;
1616	mutex_unlock(lock: &knet->mutex);
1617
1618	call_rcu(head: &mux->rcu, func: free_mux);
1619	}
1620
1621	static void kcm_done(struct kcm_sock *kcm)
1622	{
1623	struct kcm_mux *mux = kcm->mux;
1624	struct sock *sk = &kcm->sk;
1625	int socks_cnt;
1626
1627	spin_lock_bh(lock: &mux->rx_lock);
1628	if (kcm->rx_psock) {
1629	/* Cleanup in unreserve_rx_kcm */
1630	WARN_ON(kcm->done);
1631	kcm->rx_disabled = 1;
1632	kcm->done = 1;
1633	spin_unlock_bh(lock: &mux->rx_lock);
1634	return;
1635	}
1636
1637	if (kcm->rx_wait) {
1638	list_del(entry: &kcm->wait_rx_list);
1639	/* paired with lockless reads in kcm_rfree() */
1640	WRITE_ONCE(kcm->rx_wait, false);
1641	}
1642	/* Move any pending receive messages to other kcm sockets */
1643	requeue_rx_msgs(mux, head: &sk->sk_receive_queue);
1644
1645	spin_unlock_bh(lock: &mux->rx_lock);
1646
1647	if (WARN_ON(sk_rmem_alloc_get(sk)))
1648	return;
1649
1650	/* Detach from MUX */
1651	spin_lock_bh(lock: &mux->lock);
1652
1653	list_del(entry: &kcm->kcm_sock_list);
1654	mux->kcm_socks_cnt--;
1655	socks_cnt = mux->kcm_socks_cnt;
1656
1657	spin_unlock_bh(lock: &mux->lock);
1658
1659	if (!socks_cnt) {
1660	/* We are done with the mux now. */
1661	release_mux(mux);
1662	}
1663
1664	WARN_ON(kcm->rx_wait);
1665
1666	sock_put(sk: &kcm->sk);
1667	}
1668
1669	/* Called by kcm_release to close a KCM socket.
1670	* If this is the last KCM socket on the MUX, destroy the MUX.
1671	*/
1672	static int kcm_release(struct socket *sock)
1673	{
1674	struct sock *sk = sock->sk;
1675	struct kcm_sock *kcm;
1676	struct kcm_mux *mux;
1677	struct kcm_psock *psock;
1678
1679	if (!sk)
1680	return 0;
1681
1682	kcm = kcm_sk(sk);
1683	mux = kcm->mux;
1684
1685	lock_sock(sk);
1686	sock_orphan(sk);
1687	kfree_skb(skb: kcm->seq_skb);
1688
1689	/* Purge queue under lock to avoid race condition with tx_work trying
1690	* to act when queue is nonempty. If tx_work runs after this point
1691	* it will just return.
1692	*/
1693	__skb_queue_purge(list: &sk->sk_write_queue);
1694
1695	/* Set tx_stopped. This is checked when psock is bound to a kcm and we
1696	* get a writespace callback. This prevents further work being queued
1697	* from the callback (unbinding the psock occurs after canceling work.
1698	*/
1699	kcm->tx_stopped = 1;
1700
1701	release_sock(sk);
1702
1703	spin_lock_bh(lock: &mux->lock);
1704	if (kcm->tx_wait) {
1705	/* Take of tx_wait list, after this point there should be no way
1706	* that a psock will be assigned to this kcm.
1707	*/
1708	list_del(entry: &kcm->wait_psock_list);
1709	kcm->tx_wait = false;
1710	}
1711	spin_unlock_bh(lock: &mux->lock);
1712
1713	/* Cancel work. After this point there should be no outside references
1714	* to the kcm socket.
1715	*/
1716	cancel_work_sync(work: &kcm->tx_work);
1717
1718	lock_sock(sk);
1719	psock = kcm->tx_psock;
1720	if (psock) {
1721	/* A psock was reserved, so we need to kill it since it
1722	* may already have some bytes queued from a message. We
1723	* need to do this after removing kcm from tx_wait list.
1724	*/
1725	kcm_abort_tx_psock(psock, EPIPE, wakeup_kcm: false);
1726	unreserve_psock(kcm);
1727	}
1728	release_sock(sk);
1729
1730	WARN_ON(kcm->tx_wait);
1731	WARN_ON(kcm->tx_psock);
1732
1733	sock->sk = NULL;
1734
1735	kcm_done(kcm);
1736
1737	return 0;
1738	}
1739
1740	static const struct proto_ops kcm_dgram_ops = {
1741	.family = PF_KCM,
1742	.owner = THIS_MODULE,
1743	.release = kcm_release,
1744	.bind = sock_no_bind,
1745	.connect = sock_no_connect,
1746	.socketpair = sock_no_socketpair,
1747	.accept = sock_no_accept,
1748	.getname = sock_no_getname,
1749	.poll = datagram_poll,
1750	.ioctl = kcm_ioctl,
1751	.listen = sock_no_listen,
1752	.shutdown = sock_no_shutdown,
1753	.setsockopt = kcm_setsockopt,
1754	.getsockopt = kcm_getsockopt,
1755	.sendmsg = kcm_sendmsg,
1756	.recvmsg = kcm_recvmsg,
1757	.mmap = sock_no_mmap,
1758	.splice_eof = kcm_splice_eof,
1759	};
1760
1761	static const struct proto_ops kcm_seqpacket_ops = {
1762	.family = PF_KCM,
1763	.owner = THIS_MODULE,
1764	.release = kcm_release,
1765	.bind = sock_no_bind,
1766	.connect = sock_no_connect,
1767	.socketpair = sock_no_socketpair,
1768	.accept = sock_no_accept,
1769	.getname = sock_no_getname,
1770	.poll = datagram_poll,
1771	.ioctl = kcm_ioctl,
1772	.listen = sock_no_listen,
1773	.shutdown = sock_no_shutdown,
1774	.setsockopt = kcm_setsockopt,
1775	.getsockopt = kcm_getsockopt,
1776	.sendmsg = kcm_sendmsg,
1777	.recvmsg = kcm_recvmsg,
1778	.mmap = sock_no_mmap,
1779	.splice_eof = kcm_splice_eof,
1780	.splice_read = kcm_splice_read,
1781	};
1782
1783	/* Create proto operation for kcm sockets */
1784	static int kcm_create(struct net *net, struct socket *sock,
1785	int protocol, int kern)
1786	{
1787	struct kcm_net *knet = net_generic(net, id: kcm_net_id);
1788	struct sock *sk;
1789	struct kcm_mux *mux;
1790
1791	switch (sock->type) {
1792	case SOCK_DGRAM:
1793	sock->ops = &kcm_dgram_ops;
1794	break;
1795	case SOCK_SEQPACKET:
1796	sock->ops = &kcm_seqpacket_ops;
1797	break;
1798	default:
1799	return -ESOCKTNOSUPPORT;
1800	}
1801
1802	if (protocol != KCMPROTO_CONNECTED)
1803	return -EPROTONOSUPPORT;
1804
1805	sk = sk_alloc(net, PF_KCM, GFP_KERNEL, prot: &kcm_proto, kern);
1806	if (!sk)
1807	return -ENOMEM;
1808
1809	/* Allocate a kcm mux, shared between KCM sockets */
1810	mux = kmem_cache_zalloc(k: kcm_muxp, GFP_KERNEL);
1811	if (!mux) {
1812	sk_free(sk);
1813	return -ENOMEM;
1814	}
1815
1816	spin_lock_init(&mux->lock);
1817	spin_lock_init(&mux->rx_lock);
1818	INIT_LIST_HEAD(list: &mux->kcm_socks);
1819	INIT_LIST_HEAD(list: &mux->kcm_rx_waiters);
1820	INIT_LIST_HEAD(list: &mux->kcm_tx_waiters);
1821
1822	INIT_LIST_HEAD(list: &mux->psocks);
1823	INIT_LIST_HEAD(list: &mux->psocks_ready);
1824	INIT_LIST_HEAD(list: &mux->psocks_avail);
1825
1826	mux->knet = knet;
1827
1828	/* Add new MUX to list */
1829	mutex_lock(&knet->mutex);
1830	list_add_rcu(new: &mux->kcm_mux_list, head: &knet->mux_list);
1831	knet->count++;
1832	mutex_unlock(lock: &knet->mutex);
1833
1834	skb_queue_head_init(list: &mux->rx_hold_queue);
1835
1836	/* Init KCM socket */
1837	sock_init_data(sock, sk);
1838	init_kcm_sock(kcm: kcm_sk(sk), mux);
1839
1840	return 0;
1841	}
1842
1843	static const struct net_proto_family kcm_family_ops = {
1844	.family = PF_KCM,
1845	.create = kcm_create,
1846	.owner = THIS_MODULE,
1847	};
1848
1849	static __net_init int kcm_init_net(struct net *net)
1850	{
1851	struct kcm_net *knet = net_generic(net, id: kcm_net_id);
1852
1853	INIT_LIST_HEAD_RCU(list: &knet->mux_list);
1854	mutex_init(&knet->mutex);
1855
1856	return 0;
1857	}
1858
1859	static __net_exit void kcm_exit_net(struct net *net)
1860	{
1861	struct kcm_net *knet = net_generic(net, id: kcm_net_id);
1862
1863	/* All KCM sockets should be closed at this point, which should mean
1864	* that all multiplexors and psocks have been destroyed.
1865	*/
1866	WARN_ON(!list_empty(&knet->mux_list));
1867
1868	mutex_destroy(lock: &knet->mutex);
1869	}
1870
1871	static struct pernet_operations kcm_net_ops = {
1872	.init = kcm_init_net,
1873	.exit = kcm_exit_net,
1874	.id = &kcm_net_id,
1875	.size = sizeof(struct kcm_net),
1876	};
1877
1878	static int __init kcm_init(void)
1879	{
1880	int err = -ENOMEM;
1881
1882	kcm_muxp = KMEM_CACHE(kcm_mux, SLAB_HWCACHE_ALIGN);
1883	if (!kcm_muxp)
1884	goto fail;
1885
1886	kcm_psockp = KMEM_CACHE(kcm_psock, SLAB_HWCACHE_ALIGN);
1887	if (!kcm_psockp)
1888	goto fail;
1889
1890	kcm_wq = create_singlethread_workqueue("kkcmd");
1891	if (!kcm_wq)
1892	goto fail;
1893
1894	err = proto_register(prot: &kcm_proto, alloc_slab: 1);
1895	if (err)
1896	goto fail;
1897
1898	err = register_pernet_device(&kcm_net_ops);
1899	if (err)
1900	goto net_ops_fail;
1901
1902	err = sock_register(fam: &kcm_family_ops);
1903	if (err)
1904	goto sock_register_fail;
1905
1906	err = kcm_proc_init();
1907	if (err)
1908	goto proc_init_fail;
1909
1910	return 0;
1911
1912	proc_init_fail:
1913	sock_unregister(PF_KCM);
1914
1915	sock_register_fail:
1916	unregister_pernet_device(&kcm_net_ops);
1917
1918	net_ops_fail:
1919	proto_unregister(prot: &kcm_proto);
1920
1921	fail:
1922	kmem_cache_destroy(s: kcm_muxp);
1923	kmem_cache_destroy(s: kcm_psockp);
1924
1925	if (kcm_wq)
1926	destroy_workqueue(wq: kcm_wq);
1927
1928	return err;
1929	}
1930
1931	static void __exit kcm_exit(void)
1932	{
1933	kcm_proc_exit();
1934	sock_unregister(PF_KCM);
1935	unregister_pernet_device(&kcm_net_ops);
1936	proto_unregister(prot: &kcm_proto);
1937	destroy_workqueue(wq: kcm_wq);
1938
1939	kmem_cache_destroy(s: kcm_muxp);
1940	kmem_cache_destroy(s: kcm_psockp);
1941	}
1942
1943	module_init(kcm_init);
1944	module_exit(kcm_exit);
1945
1946	MODULE_LICENSE("GPL");
1947	MODULE_DESCRIPTION("KCM (Kernel Connection Multiplexor) sockets");
1948	MODULE_ALIAS_NETPROTO(PF_KCM);
1949