linux - net/kcm/kcmsock.c

1953 lines · 42.6 KiB

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