kcmsock.c

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