?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Linux 4.8-rc8
[linux/fpc-iii.git] / net / kcm / kcmsock.c
blob411693288648dc3eb27b0b2a0c542be92762ad8a
1 #include <linux/bpf.h>
2 #include <linux/errno.h>
3 #include <linux/errqueue.h>
4 #include <linux/file.h>
5 #include <linux/in.h>
6 #include <linux/kernel.h>
7 #include <linux/module.h>
8 #include <linux/net.h>
9 #include <linux/netdevice.h>
10 #include <linux/poll.h>
11 #include <linux/rculist.h>
12 #include <linux/skbuff.h>
13 #include <linux/socket.h>
14 #include <linux/uaccess.h>
15 #include <linux/workqueue.h>
16 #include <linux/syscalls.h>
17 #include <net/kcm.h>
18 #include <net/netns/generic.h>
19 #include <net/sock.h>
20 #include <net/tcp.h>
21 #include <uapi/linux/kcm.h>
22
23 unsigned int kcm_net_id;
24
25 static struct kmem_cache *kcm_psockp __read_mostly;
26 static struct kmem_cache *kcm_muxp __read_mostly;
27 static struct workqueue_struct *kcm_wq;
28
29 static inline struct kcm_sock *kcm_sk(const struct sock *sk)
30 {
31 return (struct kcm_sock *)sk;
32 }
33
34 static inline struct kcm_tx_msg *kcm_tx_msg(struct sk_buff *skb)
35 {
36 return (struct kcm_tx_msg *)skb->cb;
37 }
38
39 static inline struct kcm_rx_msg *kcm_rx_msg(struct sk_buff *skb)
40 {
41 return (struct kcm_rx_msg *)((void *)skb->cb +
42 offsetof(struct qdisc_skb_cb, data));
43 }
44
45 static void report_csk_error(struct sock *csk, int err)
46 {
47 csk->sk_err = EPIPE;
48 csk->sk_error_report(csk);
49 }
50
51 /* Callback lock held */
52 static void kcm_abort_rx_psock(struct kcm_psock *psock, int err,
53 struct sk_buff *skb)
54 {
55 struct sock *csk = psock->sk;
56
57 /* Unrecoverable error in receive */
58
59 del_timer(&psock->rx_msg_timer);
60
61 if (psock->rx_stopped)
62 return;
63
64 psock->rx_stopped = 1;
65 KCM_STATS_INCR(psock->stats.rx_aborts);
66
67 /* Report an error on the lower socket */
68 report_csk_error(csk, err);
69 }
70
71 static void kcm_abort_tx_psock(struct kcm_psock *psock, int err,
72 bool wakeup_kcm)
73 {
74 struct sock *csk = psock->sk;
75 struct kcm_mux *mux = psock->mux;
76
77 /* Unrecoverable error in transmit */
78
79 spin_lock_bh(&mux->lock);
80
81 if (psock->tx_stopped) {
82 spin_unlock_bh(&mux->lock);
83 return;
84 }
85
86 psock->tx_stopped = 1;
87 KCM_STATS_INCR(psock->stats.tx_aborts);
88
89 if (!psock->tx_kcm) {
90 /* Take off psocks_avail list */
91 list_del(&psock->psock_avail_list);
92 } else if (wakeup_kcm) {
93 /* In this case psock is being aborted while outside of
94 * write_msgs and psock is reserved. Schedule tx_work
95 * to handle the failure there. Need to commit tx_stopped
96 * before queuing work.
97 */
98 smp_mb();
99
100 queue_work(kcm_wq, &psock->tx_kcm->tx_work);
101 }
102
103 spin_unlock_bh(&mux->lock);
104
105 /* Report error on lower socket */
106 report_csk_error(csk, err);
107 }
108
109 /* RX mux lock held. */
110 static void kcm_update_rx_mux_stats(struct kcm_mux *mux,
111 struct kcm_psock *psock)
112 {
113 KCM_STATS_ADD(mux->stats.rx_bytes,
114 psock->stats.rx_bytes - psock->saved_rx_bytes);
115 mux->stats.rx_msgs +=
116 psock->stats.rx_msgs - psock->saved_rx_msgs;
117 psock->saved_rx_msgs = psock->stats.rx_msgs;
118 psock->saved_rx_bytes = psock->stats.rx_bytes;
119 }
120
121 static void kcm_update_tx_mux_stats(struct kcm_mux *mux,
122 struct kcm_psock *psock)
123 {
124 KCM_STATS_ADD(mux->stats.tx_bytes,
125 psock->stats.tx_bytes - psock->saved_tx_bytes);
126 mux->stats.tx_msgs +=
127 psock->stats.tx_msgs - psock->saved_tx_msgs;
128 psock->saved_tx_msgs = psock->stats.tx_msgs;
129 psock->saved_tx_bytes = psock->stats.tx_bytes;
130 }
131
132 static int kcm_queue_rcv_skb(struct sock *sk, struct sk_buff *skb);
133
134 /* KCM is ready to receive messages on its queue-- either the KCM is new or
135 * has become unblocked after being blocked on full socket buffer. Queue any
136 * pending ready messages on a psock. RX mux lock held.
137 */
138 static void kcm_rcv_ready(struct kcm_sock *kcm)
139 {
140 struct kcm_mux *mux = kcm->mux;
141 struct kcm_psock *psock;
142 struct sk_buff *skb;
143
144 if (unlikely(kcm->rx_wait || kcm->rx_psock || kcm->rx_disabled))
145 return;
146
147 while (unlikely((skb = __skb_dequeue(&mux->rx_hold_queue)))) {
148 if (kcm_queue_rcv_skb(&kcm->sk, skb)) {
149 /* Assuming buffer limit has been reached */
150 skb_queue_head(&mux->rx_hold_queue, skb);
151 WARN_ON(!sk_rmem_alloc_get(&kcm->sk));
152 return;
153 }
154 }
155
156 while (!list_empty(&mux->psocks_ready)) {
157 psock = list_first_entry(&mux->psocks_ready, struct kcm_psock,
158 psock_ready_list);
159
160 if (kcm_queue_rcv_skb(&kcm->sk, psock->ready_rx_msg)) {
161 /* Assuming buffer limit has been reached */
162 WARN_ON(!sk_rmem_alloc_get(&kcm->sk));
163 return;
164 }
165
166 /* Consumed the ready message on the psock. Schedule rx_work to
167 * get more messages.
168 */
169 list_del(&psock->psock_ready_list);
170 psock->ready_rx_msg = NULL;
171
172 /* Commit clearing of ready_rx_msg for queuing work */
173 smp_mb();
174
175 queue_work(kcm_wq, &psock->rx_work);
176 }
177
178 /* Buffer limit is okay now, add to ready list */
179 list_add_tail(&kcm->wait_rx_list,
180 &kcm->mux->kcm_rx_waiters);
181 kcm->rx_wait = true;
182 }
183
184 static void kcm_rfree(struct sk_buff *skb)
185 {
186 struct sock *sk = skb->sk;
187 struct kcm_sock *kcm = kcm_sk(sk);
188 struct kcm_mux *mux = kcm->mux;
189 unsigned int len = skb->truesize;
190
191 sk_mem_uncharge(sk, len);
192 atomic_sub(len, &sk->sk_rmem_alloc);
193
194 /* For reading rx_wait and rx_psock without holding lock */
195 smp_mb__after_atomic();
196
197 if (!kcm->rx_wait && !kcm->rx_psock &&
198 sk_rmem_alloc_get(sk) < sk->sk_rcvlowat) {
199 spin_lock_bh(&mux->rx_lock);
200 kcm_rcv_ready(kcm);
201 spin_unlock_bh(&mux->rx_lock);
202 }
203 }
204
205 static int kcm_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
206 {
207 struct sk_buff_head *list = &sk->sk_receive_queue;
208
209 if (atomic_read(&sk->sk_rmem_alloc) >= sk->sk_rcvbuf)
210 return -ENOMEM;
211
212 if (!sk_rmem_schedule(sk, skb, skb->truesize))
213 return -ENOBUFS;
214
215 skb->dev = NULL;
216
217 skb_orphan(skb);
218 skb->sk = sk;
219 skb->destructor = kcm_rfree;
220 atomic_add(skb->truesize, &sk->sk_rmem_alloc);
221 sk_mem_charge(sk, skb->truesize);
222
223 skb_queue_tail(list, skb);
224
225 if (!sock_flag(sk, SOCK_DEAD))
226 sk->sk_data_ready(sk);
227
228 return 0;
229 }
230
231 /* Requeue received messages for a kcm socket to other kcm sockets. This is
232 * called with a kcm socket is receive disabled.
233 * RX mux lock held.
234 */
235 static void requeue_rx_msgs(struct kcm_mux *mux, struct sk_buff_head *head)
236 {
237 struct sk_buff *skb;
238 struct kcm_sock *kcm;
239
240 while ((skb = __skb_dequeue(head))) {
241 /* Reset destructor to avoid calling kcm_rcv_ready */
242 skb->destructor = sock_rfree;
243 skb_orphan(skb);
244 try_again:
245 if (list_empty(&mux->kcm_rx_waiters)) {
246 skb_queue_tail(&mux->rx_hold_queue, skb);
247 continue;
248 }
249
250 kcm = list_first_entry(&mux->kcm_rx_waiters,
251 struct kcm_sock, wait_rx_list);
252
253 if (kcm_queue_rcv_skb(&kcm->sk, skb)) {
254 /* Should mean socket buffer full */
255 list_del(&kcm->wait_rx_list);
256 kcm->rx_wait = false;
257
258 /* Commit rx_wait to read in kcm_free */
259 smp_wmb();
260
261 goto try_again;
262 }
263 }
264 }
265
266 /* Lower sock lock held */
267 static struct kcm_sock *reserve_rx_kcm(struct kcm_psock *psock,
268 struct sk_buff *head)
269 {
270 struct kcm_mux *mux = psock->mux;
271 struct kcm_sock *kcm;
272
273 WARN_ON(psock->ready_rx_msg);
274
275 if (psock->rx_kcm)
276 return psock->rx_kcm;
277
278 spin_lock_bh(&mux->rx_lock);
279
280 if (psock->rx_kcm) {
281 spin_unlock_bh(&mux->rx_lock);
282 return psock->rx_kcm;
283 }
284
285 kcm_update_rx_mux_stats(mux, psock);
286
287 if (list_empty(&mux->kcm_rx_waiters)) {
288 psock->ready_rx_msg = head;
289 list_add_tail(&psock->psock_ready_list,
290 &mux->psocks_ready);
291 spin_unlock_bh(&mux->rx_lock);
292 return NULL;
293 }
294
295 kcm = list_first_entry(&mux->kcm_rx_waiters,
296 struct kcm_sock, wait_rx_list);
297 list_del(&kcm->wait_rx_list);
298 kcm->rx_wait = false;
299
300 psock->rx_kcm = kcm;
301 kcm->rx_psock = psock;
302
303 spin_unlock_bh(&mux->rx_lock);
304
305 return kcm;
306 }
307
308 static void kcm_done(struct kcm_sock *kcm);
309
310 static void kcm_done_work(struct work_struct *w)
311 {
312 kcm_done(container_of(w, struct kcm_sock, done_work));
313 }
314
315 /* Lower sock held */
316 static void unreserve_rx_kcm(struct kcm_psock *psock,
317 bool rcv_ready)
318 {
319 struct kcm_sock *kcm = psock->rx_kcm;
320 struct kcm_mux *mux = psock->mux;
321
322 if (!kcm)
323 return;
324
325 spin_lock_bh(&mux->rx_lock);
326
327 psock->rx_kcm = NULL;
328 kcm->rx_psock = NULL;
329
330 /* Commit kcm->rx_psock before sk_rmem_alloc_get to sync with
331 * kcm_rfree
332 */
333 smp_mb();
334
335 if (unlikely(kcm->done)) {
336 spin_unlock_bh(&mux->rx_lock);
337
338 /* Need to run kcm_done in a task since we need to qcquire
339 * callback locks which may already be held here.
340 */
341 INIT_WORK(&kcm->done_work, kcm_done_work);
342 schedule_work(&kcm->done_work);
343 return;
344 }
345
346 if (unlikely(kcm->rx_disabled)) {
347 requeue_rx_msgs(mux, &kcm->sk.sk_receive_queue);
348 } else if (rcv_ready || unlikely(!sk_rmem_alloc_get(&kcm->sk))) {
349 /* Check for degenerative race with rx_wait that all
350 * data was dequeued (accounted for in kcm_rfree).
351 */
352 kcm_rcv_ready(kcm);
353 }
354 spin_unlock_bh(&mux->rx_lock);
355 }
356
357 static void kcm_start_rx_timer(struct kcm_psock *psock)
358 {
359 if (psock->sk->sk_rcvtimeo)
360 mod_timer(&psock->rx_msg_timer, psock->sk->sk_rcvtimeo);
361 }
362
363 /* Macro to invoke filter function. */
364 #define KCM_RUN_FILTER(prog, ctx) \
365 (*prog->bpf_func)(ctx, prog->insnsi)
366
367 /* Lower socket lock held */
368 static int kcm_tcp_recv(read_descriptor_t *desc, struct sk_buff *orig_skb,
369 unsigned int orig_offset, size_t orig_len)
370 {
371 struct kcm_psock *psock = (struct kcm_psock *)desc->arg.data;
372 struct kcm_rx_msg *rxm;
373 struct kcm_sock *kcm;
374 struct sk_buff *head, *skb;
375 size_t eaten = 0, cand_len;
376 ssize_t extra;
377 int err;
378 bool cloned_orig = false;
379
380 if (psock->ready_rx_msg)
381 return 0;
382
383 head = psock->rx_skb_head;
384 if (head) {
385 /* Message already in progress */
386
387 rxm = kcm_rx_msg(head);
388 if (unlikely(rxm->early_eaten)) {
389 /* Already some number of bytes on the receive sock
390 * data saved in rx_skb_head, just indicate they
391 * are consumed.
392 */
393 eaten = orig_len <= rxm->early_eaten ?
394 orig_len : rxm->early_eaten;
395 rxm->early_eaten -= eaten;
396
397 return eaten;
398 }
399
400 if (unlikely(orig_offset)) {
401 /* Getting data with a non-zero offset when a message is
402 * in progress is not expected. If it does happen, we
403 * need to clone and pull since we can't deal with
404 * offsets in the skbs for a message expect in the head.
405 */
406 orig_skb = skb_clone(orig_skb, GFP_ATOMIC);
407 if (!orig_skb) {
408 KCM_STATS_INCR(psock->stats.rx_mem_fail);
409 desc->error = -ENOMEM;
410 return 0;
411 }
412 if (!pskb_pull(orig_skb, orig_offset)) {
413 KCM_STATS_INCR(psock->stats.rx_mem_fail);
414 kfree_skb(orig_skb);
415 desc->error = -ENOMEM;
416 return 0;
417 }
418 cloned_orig = true;
419 orig_offset = 0;
420 }
421
422 if (!psock->rx_skb_nextp) {
423 /* We are going to append to the frags_list of head.
424 * Need to unshare the frag_list.
425 */
426 err = skb_unclone(head, GFP_ATOMIC);
427 if (err) {
428 KCM_STATS_INCR(psock->stats.rx_mem_fail);
429 desc->error = err;
430 return 0;
431 }
432
433 if (unlikely(skb_shinfo(head)->frag_list)) {
434 /* We can't append to an sk_buff that already
435 * has a frag_list. We create a new head, point
436 * the frag_list of that to the old head, and
437 * then are able to use the old head->next for
438 * appending to the message.
439 */
440 if (WARN_ON(head->next)) {
441 desc->error = -EINVAL;
442 return 0;
443 }
444
445 skb = alloc_skb(0, GFP_ATOMIC);
446 if (!skb) {
447 KCM_STATS_INCR(psock->stats.rx_mem_fail);
448 desc->error = -ENOMEM;
449 return 0;
450 }
451 skb->len = head->len;
452 skb->data_len = head->len;
453 skb->truesize = head->truesize;
454 *kcm_rx_msg(skb) = *kcm_rx_msg(head);
455 psock->rx_skb_nextp = &head->next;
456 skb_shinfo(skb)->frag_list = head;
457 psock->rx_skb_head = skb;
458 head = skb;
459 } else {
460 psock->rx_skb_nextp =
461 &skb_shinfo(head)->frag_list;
462 }
463 }
464 }
465
466 while (eaten < orig_len) {
467 /* Always clone since we will consume something */
468 skb = skb_clone(orig_skb, GFP_ATOMIC);
469 if (!skb) {
470 KCM_STATS_INCR(psock->stats.rx_mem_fail);
471 desc->error = -ENOMEM;
472 break;
473 }
474
475 cand_len = orig_len - eaten;
476
477 head = psock->rx_skb_head;
478 if (!head) {
479 head = skb;
480 psock->rx_skb_head = head;
481 /* Will set rx_skb_nextp on next packet if needed */
482 psock->rx_skb_nextp = NULL;
483 rxm = kcm_rx_msg(head);
484 memset(rxm, 0, sizeof(*rxm));
485 rxm->offset = orig_offset + eaten;
486 } else {
487 /* Unclone since we may be appending to an skb that we
488 * already share a frag_list with.
489 */
490 err = skb_unclone(skb, GFP_ATOMIC);
491 if (err) {
492 KCM_STATS_INCR(psock->stats.rx_mem_fail);
493 desc->error = err;
494 break;
495 }
496
497 rxm = kcm_rx_msg(head);
498 *psock->rx_skb_nextp = skb;
499 psock->rx_skb_nextp = &skb->next;
500 head->data_len += skb->len;
501 head->len += skb->len;
502 head->truesize += skb->truesize;
503 }
504
505 if (!rxm->full_len) {
506 ssize_t len;
507
508 len = KCM_RUN_FILTER(psock->bpf_prog, head);
509
510 if (!len) {
511 /* Need more header to determine length */
512 if (!rxm->accum_len) {
513 /* Start RX timer for new message */
514 kcm_start_rx_timer(psock);
515 }
516 rxm->accum_len += cand_len;
517 eaten += cand_len;
518 KCM_STATS_INCR(psock->stats.rx_need_more_hdr);
519 WARN_ON(eaten != orig_len);
520 break;
521 } else if (len > psock->sk->sk_rcvbuf) {
522 /* Message length exceeds maximum allowed */
523 KCM_STATS_INCR(psock->stats.rx_msg_too_big);
524 desc->error = -EMSGSIZE;
525 psock->rx_skb_head = NULL;
526 kcm_abort_rx_psock(psock, EMSGSIZE, head);
527 break;
528 } else if (len <= (ssize_t)head->len -
529 skb->len - rxm->offset) {
530 /* Length must be into new skb (and also
531 * greater than zero)
532 */
533 KCM_STATS_INCR(psock->stats.rx_bad_hdr_len);
534 desc->error = -EPROTO;
535 psock->rx_skb_head = NULL;
536 kcm_abort_rx_psock(psock, EPROTO, head);
537 break;
538 }
539
540 rxm->full_len = len;
541 }
542
543 extra = (ssize_t)(rxm->accum_len + cand_len) - rxm->full_len;
544
545 if (extra < 0) {
546 /* Message not complete yet. */
547 if (rxm->full_len - rxm->accum_len >
548 tcp_inq(psock->sk)) {
549 /* Don't have the whole messages in the socket
550 * buffer. Set psock->rx_need_bytes to wait for
551 * the rest of the message. Also, set "early
552 * eaten" since we've already buffered the skb
553 * but don't consume yet per tcp_read_sock.
554 */
555
556 if (!rxm->accum_len) {
557 /* Start RX timer for new message */
558 kcm_start_rx_timer(psock);
559 }
560
561 psock->rx_need_bytes = rxm->full_len -
562 rxm->accum_len;
563 rxm->accum_len += cand_len;
564 rxm->early_eaten = cand_len;
565 KCM_STATS_ADD(psock->stats.rx_bytes, cand_len);
566 desc->count = 0; /* Stop reading socket */
567 break;
568 }
569 rxm->accum_len += cand_len;
570 eaten += cand_len;
571 WARN_ON(eaten != orig_len);
572 break;
573 }
574
575 /* Positive extra indicates ore bytes than needed for the
576 * message
577 */
578
579 WARN_ON(extra > cand_len);
580
581 eaten += (cand_len - extra);
582
583 /* Hurray, we have a new message! */
584 del_timer(&psock->rx_msg_timer);
585 psock->rx_skb_head = NULL;
586 KCM_STATS_INCR(psock->stats.rx_msgs);
587
588 try_queue:
589 kcm = reserve_rx_kcm(psock, head);
590 if (!kcm) {
591 /* Unable to reserve a KCM, message is held in psock. */
592 break;
593 }
594
595 if (kcm_queue_rcv_skb(&kcm->sk, head)) {
596 /* Should mean socket buffer full */
597 unreserve_rx_kcm(psock, false);
598 goto try_queue;
599 }
600 }
601
602 if (cloned_orig)
603 kfree_skb(orig_skb);
604
605 KCM_STATS_ADD(psock->stats.rx_bytes, eaten);
606
607 return eaten;
608 }
609
610 /* Called with lock held on lower socket */
611 static int psock_tcp_read_sock(struct kcm_psock *psock)
612 {
613 read_descriptor_t desc;
614
615 desc.arg.data = psock;
616 desc.error = 0;
617 desc.count = 1; /* give more than one skb per call */
618
619 /* sk should be locked here, so okay to do tcp_read_sock */
620 tcp_read_sock(psock->sk, &desc, kcm_tcp_recv);
621
622 unreserve_rx_kcm(psock, true);
623
624 return desc.error;
625 }
626
627 /* Lower sock lock held */
628 static void psock_tcp_data_ready(struct sock *sk)
629 {
630 struct kcm_psock *psock;
631
632 read_lock_bh(&sk->sk_callback_lock);
633
634 psock = (struct kcm_psock *)sk->sk_user_data;
635 if (unlikely(!psock || psock->rx_stopped))
636 goto out;
637
638 if (psock->ready_rx_msg)
639 goto out;
640
641 if (psock->rx_need_bytes) {
642 if (tcp_inq(sk) >= psock->rx_need_bytes)
643 psock->rx_need_bytes = 0;
644 else
645 goto out;
646 }
647
648 if (psock_tcp_read_sock(psock) == -ENOMEM)
649 queue_delayed_work(kcm_wq, &psock->rx_delayed_work, 0);
650
651 out:
652 read_unlock_bh(&sk->sk_callback_lock);
653 }
654
655 static void do_psock_rx_work(struct kcm_psock *psock)
656 {
657 read_descriptor_t rd_desc;
658 struct sock *csk = psock->sk;
659
660 /* We need the read lock to synchronize with psock_tcp_data_ready. We
661 * need the socket lock for calling tcp_read_sock.
662 */
663 lock_sock(csk);
664 read_lock_bh(&csk->sk_callback_lock);
665
666 if (unlikely(csk->sk_user_data != psock))
667 goto out;
668
669 if (unlikely(psock->rx_stopped))
670 goto out;
671
672 if (psock->ready_rx_msg)
673 goto out;
674
675 rd_desc.arg.data = psock;
676
677 if (psock_tcp_read_sock(psock) == -ENOMEM)
678 queue_delayed_work(kcm_wq, &psock->rx_delayed_work, 0);
679
680 out:
681 read_unlock_bh(&csk->sk_callback_lock);
682 release_sock(csk);
683 }
684
685 static void psock_rx_work(struct work_struct *w)
686 {
687 do_psock_rx_work(container_of(w, struct kcm_psock, rx_work));
688 }
689
690 static void psock_rx_delayed_work(struct work_struct *w)
691 {
692 do_psock_rx_work(container_of(w, struct kcm_psock,
693 rx_delayed_work.work));
694 }
695
696 static void psock_tcp_state_change(struct sock *sk)
697 {
698 /* TCP only does a POLLIN for a half close. Do a POLLHUP here
699 * since application will normally not poll with POLLIN
700 * on the TCP sockets.
701 */
702
703 report_csk_error(sk, EPIPE);
704 }
705
706 static void psock_tcp_write_space(struct sock *sk)
707 {
708 struct kcm_psock *psock;
709 struct kcm_mux *mux;
710 struct kcm_sock *kcm;
711
712 read_lock_bh(&sk->sk_callback_lock);
713
714 psock = (struct kcm_psock *)sk->sk_user_data;
715 if (unlikely(!psock))
716 goto out;
717
718 mux = psock->mux;
719
720 spin_lock_bh(&mux->lock);
721
722 /* Check if the socket is reserved so someone is waiting for sending. */
723 kcm = psock->tx_kcm;
724 if (kcm)
725 queue_work(kcm_wq, &kcm->tx_work);
726
727 spin_unlock_bh(&mux->lock);
728 out:
729 read_unlock_bh(&sk->sk_callback_lock);
730 }
731
732 static void unreserve_psock(struct kcm_sock *kcm);
733
734 /* kcm sock is locked. */
735 static struct kcm_psock *reserve_psock(struct kcm_sock *kcm)
736 {
737 struct kcm_mux *mux = kcm->mux;
738 struct kcm_psock *psock;
739
740 psock = kcm->tx_psock;
741
742 smp_rmb(); /* Must read tx_psock before tx_wait */
743
744 if (psock) {
745 WARN_ON(kcm->tx_wait);
746 if (unlikely(psock->tx_stopped))
747 unreserve_psock(kcm);
748 else
749 return kcm->tx_psock;
750 }
751
752 spin_lock_bh(&mux->lock);
753
754 /* Check again under lock to see if psock was reserved for this
755 * psock via psock_unreserve.
756 */
757 psock = kcm->tx_psock;
758 if (unlikely(psock)) {
759 WARN_ON(kcm->tx_wait);
760 spin_unlock_bh(&mux->lock);
761 return kcm->tx_psock;
762 }
763
764 if (!list_empty(&mux->psocks_avail)) {
765 psock = list_first_entry(&mux->psocks_avail,
766 struct kcm_psock,
767 psock_avail_list);
768 list_del(&psock->psock_avail_list);
769 if (kcm->tx_wait) {
770 list_del(&kcm->wait_psock_list);
771 kcm->tx_wait = false;
772 }
773 kcm->tx_psock = psock;
774 psock->tx_kcm = kcm;
775 KCM_STATS_INCR(psock->stats.reserved);
776 } else if (!kcm->tx_wait) {
777 list_add_tail(&kcm->wait_psock_list,
778 &mux->kcm_tx_waiters);
779 kcm->tx_wait = true;
780 }
781
782 spin_unlock_bh(&mux->lock);
783
784 return psock;
785 }
786
787 /* mux lock held */
788 static void psock_now_avail(struct kcm_psock *psock)
789 {
790 struct kcm_mux *mux = psock->mux;
791 struct kcm_sock *kcm;
792
793 if (list_empty(&mux->kcm_tx_waiters)) {
794 list_add_tail(&psock->psock_avail_list,
795 &mux->psocks_avail);
796 } else {
797 kcm = list_first_entry(&mux->kcm_tx_waiters,
798 struct kcm_sock,
799 wait_psock_list);
800 list_del(&kcm->wait_psock_list);
801 kcm->tx_wait = false;
802 psock->tx_kcm = kcm;
803
804 /* Commit before changing tx_psock since that is read in
805 * reserve_psock before queuing work.
806 */
807 smp_mb();
808
809 kcm->tx_psock = psock;
810 KCM_STATS_INCR(psock->stats.reserved);
811 queue_work(kcm_wq, &kcm->tx_work);
812 }
813 }
814
815 /* kcm sock is locked. */
816 static void unreserve_psock(struct kcm_sock *kcm)
817 {
818 struct kcm_psock *psock;
819 struct kcm_mux *mux = kcm->mux;
820
821 spin_lock_bh(&mux->lock);
822
823 psock = kcm->tx_psock;
824
825 if (WARN_ON(!psock)) {
826 spin_unlock_bh(&mux->lock);
827 return;
828 }
829
830 smp_rmb(); /* Read tx_psock before tx_wait */
831
832 kcm_update_tx_mux_stats(mux, psock);
833
834 WARN_ON(kcm->tx_wait);
835
836 kcm->tx_psock = NULL;
837 psock->tx_kcm = NULL;
838 KCM_STATS_INCR(psock->stats.unreserved);
839
840 if (unlikely(psock->tx_stopped)) {
841 if (psock->done) {
842 /* Deferred free */
843 list_del(&psock->psock_list);
844 mux->psocks_cnt--;
845 sock_put(psock->sk);
846 fput(psock->sk->sk_socket->file);
847 kmem_cache_free(kcm_psockp, psock);
848 }
849
850 /* Don't put back on available list */
851
852 spin_unlock_bh(&mux->lock);
853
854 return;
855 }
856
857 psock_now_avail(psock);
858
859 spin_unlock_bh(&mux->lock);
860 }
861
862 static void kcm_report_tx_retry(struct kcm_sock *kcm)
863 {
864 struct kcm_mux *mux = kcm->mux;
865
866 spin_lock_bh(&mux->lock);
867 KCM_STATS_INCR(mux->stats.tx_retries);
868 spin_unlock_bh(&mux->lock);
869 }
870
871 /* Write any messages ready on the kcm socket. Called with kcm sock lock
872 * held. Return bytes actually sent or error.
873 */
874 static int kcm_write_msgs(struct kcm_sock *kcm)
875 {
876 struct sock *sk = &kcm->sk;
877 struct kcm_psock *psock;
878 struct sk_buff *skb, *head;
879 struct kcm_tx_msg *txm;
880 unsigned short fragidx, frag_offset;
881 unsigned int sent, total_sent = 0;
882 int ret = 0;
883
884 kcm->tx_wait_more = false;
885 psock = kcm->tx_psock;
886 if (unlikely(psock && psock->tx_stopped)) {
887 /* A reserved psock was aborted asynchronously. Unreserve
888 * it and we'll retry the message.
889 */
890 unreserve_psock(kcm);
891 kcm_report_tx_retry(kcm);
892 if (skb_queue_empty(&sk->sk_write_queue))
893 return 0;
894
895 kcm_tx_msg(skb_peek(&sk->sk_write_queue))->sent = 0;
896
897 } else if (skb_queue_empty(&sk->sk_write_queue)) {
898 return 0;
899 }
900
901 head = skb_peek(&sk->sk_write_queue);
902 txm = kcm_tx_msg(head);
903
904 if (txm->sent) {
905 /* Send of first skbuff in queue already in progress */
906 if (WARN_ON(!psock)) {
907 ret = -EINVAL;
908 goto out;
909 }
910 sent = txm->sent;
911 frag_offset = txm->frag_offset;
912 fragidx = txm->fragidx;
913 skb = txm->frag_skb;
914
915 goto do_frag;
916 }
917
918 try_again:
919 psock = reserve_psock(kcm);
920 if (!psock)
921 goto out;
922
923 do {
924 skb = head;
925 txm = kcm_tx_msg(head);
926 sent = 0;
927
928 do_frag_list:
929 if (WARN_ON(!skb_shinfo(skb)->nr_frags)) {
930 ret = -EINVAL;
931 goto out;
932 }
933
934 for (fragidx = 0; fragidx < skb_shinfo(skb)->nr_frags;
935 fragidx++) {
936 skb_frag_t *frag;
937
938 frag_offset = 0;
939 do_frag:
940 frag = &skb_shinfo(skb)->frags[fragidx];
941 if (WARN_ON(!frag->size)) {
942 ret = -EINVAL;
943 goto out;
944 }
945
946 ret = kernel_sendpage(psock->sk->sk_socket,
947 frag->page.p,
948 frag->page_offset + frag_offset,
949 frag->size - frag_offset,
950 MSG_DONTWAIT);
951 if (ret <= 0) {
952 if (ret == -EAGAIN) {
953 /* Save state to try again when there's
954 * write space on the socket
955 */
956 txm->sent = sent;
957 txm->frag_offset = frag_offset;
958 txm->fragidx = fragidx;
959 txm->frag_skb = skb;
960
961 ret = 0;
962 goto out;
963 }
964
965 /* Hard failure in sending message, abort this
966 * psock since it has lost framing
967 * synchonization and retry sending the
968 * message from the beginning.
969 */
970 kcm_abort_tx_psock(psock, ret ? -ret : EPIPE,
971 true);
972 unreserve_psock(kcm);
973
974 txm->sent = 0;
975 kcm_report_tx_retry(kcm);
976 ret = 0;
977
978 goto try_again;
979 }
980
981 sent += ret;
982 frag_offset += ret;
983 KCM_STATS_ADD(psock->stats.tx_bytes, ret);
984 if (frag_offset < frag->size) {
985 /* Not finished with this frag */
986 goto do_frag;
987 }
988 }
989
990 if (skb == head) {
991 if (skb_has_frag_list(skb)) {
992 skb = skb_shinfo(skb)->frag_list;
993 goto do_frag_list;
994 }
995 } else if (skb->next) {
996 skb = skb->next;
997 goto do_frag_list;
998 }
999
1000 /* Successfully sent the whole packet, account for it. */
1001 skb_dequeue(&sk->sk_write_queue);
1002 kfree_skb(head);
1003 sk->sk_wmem_queued -= sent;
1004 total_sent += sent;
1005 KCM_STATS_INCR(psock->stats.tx_msgs);
1006 } while ((head = skb_peek(&sk->sk_write_queue)));
1007 out:
1008 if (!head) {
1009 /* Done with all queued messages. */
1010 WARN_ON(!skb_queue_empty(&sk->sk_write_queue));
1011 unreserve_psock(kcm);
1012 }
1013
1014 /* Check if write space is available */
1015 sk->sk_write_space(sk);
1016
1017 return total_sent ? : ret;
1018 }
1019
1020 static void kcm_tx_work(struct work_struct *w)
1021 {
1022 struct kcm_sock *kcm = container_of(w, struct kcm_sock, tx_work);
1023 struct sock *sk = &kcm->sk;
1024 int err;
1025
1026 lock_sock(sk);
1027
1028 /* Primarily for SOCK_DGRAM sockets, also handle asynchronous tx
1029 * aborts
1030 */
1031 err = kcm_write_msgs(kcm);
1032 if (err < 0) {
1033 /* Hard failure in write, report error on KCM socket */
1034 pr_warn("KCM: Hard failure on kcm_write_msgs %d\n", err);
1035 report_csk_error(&kcm->sk, -err);
1036 goto out;
1037 }
1038
1039 /* Primarily for SOCK_SEQPACKET sockets */
1040 if (likely(sk->sk_socket) &&
1041 test_bit(SOCK_NOSPACE, &sk->sk_socket->flags)) {
1042 clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
1043 sk->sk_write_space(sk);
1044 }
1045
1046 out:
1047 release_sock(sk);
1048 }
1049
1050 static void kcm_push(struct kcm_sock *kcm)
1051 {
1052 if (kcm->tx_wait_more)
1053 kcm_write_msgs(kcm);
1054 }
1055
1056 static ssize_t kcm_sendpage(struct socket *sock, struct page *page,
1057 int offset, size_t size, int flags)
1058
1059 {
1060 struct sock *sk = sock->sk;
1061 struct kcm_sock *kcm = kcm_sk(sk);
1062 struct sk_buff *skb = NULL, *head = NULL;
1063 long timeo = sock_sndtimeo(sk, flags & MSG_DONTWAIT);
1064 bool eor;
1065 int err = 0;
1066 int i;
1067
1068 if (flags & MSG_SENDPAGE_NOTLAST)
1069 flags |= MSG_MORE;
1070
1071 /* No MSG_EOR from splice, only look at MSG_MORE */
1072 eor = !(flags & MSG_MORE);
1073
1074 lock_sock(sk);
1075
1076 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
1077
1078 err = -EPIPE;
1079 if (sk->sk_err)
1080 goto out_error;
1081
1082 if (kcm->seq_skb) {
1083 /* Previously opened message */
1084 head = kcm->seq_skb;
1085 skb = kcm_tx_msg(head)->last_skb;
1086 i = skb_shinfo(skb)->nr_frags;
1087
1088 if (skb_can_coalesce(skb, i, page, offset)) {
1089 skb_frag_size_add(&skb_shinfo(skb)->frags[i - 1], size);
1090 skb_shinfo(skb)->tx_flags |= SKBTX_SHARED_FRAG;
1091 goto coalesced;
1092 }
1093
1094 if (i >= MAX_SKB_FRAGS) {
1095 struct sk_buff *tskb;
1096
1097 tskb = alloc_skb(0, sk->sk_allocation);
1098 while (!tskb) {
1099 kcm_push(kcm);
1100 err = sk_stream_wait_memory(sk, &timeo);
1101 if (err)
1102 goto out_error;
1103 }
1104
1105 if (head == skb)
1106 skb_shinfo(head)->frag_list = tskb;
1107 else
1108 skb->next = tskb;
1109
1110 skb = tskb;
1111 skb->ip_summed = CHECKSUM_UNNECESSARY;
1112 i = 0;
1113 }
1114 } else {
1115 /* Call the sk_stream functions to manage the sndbuf mem. */
1116 if (!sk_stream_memory_free(sk)) {
1117 kcm_push(kcm);
1118 set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
1119 err = sk_stream_wait_memory(sk, &timeo);
1120 if (err)
1121 goto out_error;
1122 }
1123
1124 head = alloc_skb(0, sk->sk_allocation);
1125 while (!head) {
1126 kcm_push(kcm);
1127 err = sk_stream_wait_memory(sk, &timeo);
1128 if (err)
1129 goto out_error;
1130 }
1131
1132 skb = head;
1133 i = 0;
1134 }
1135
1136 get_page(page);
1137 skb_fill_page_desc(skb, i, page, offset, size);
1138 skb_shinfo(skb)->tx_flags |= SKBTX_SHARED_FRAG;
1139
1140 coalesced:
1141 skb->len += size;
1142 skb->data_len += size;
1143 skb->truesize += size;
1144 sk->sk_wmem_queued += size;
1145 sk_mem_charge(sk, size);
1146
1147 if (head != skb) {
1148 head->len += size;
1149 head->data_len += size;
1150 head->truesize += size;
1151 }
1152
1153 if (eor) {
1154 bool not_busy = skb_queue_empty(&sk->sk_write_queue);
1155
1156 /* Message complete, queue it on send buffer */
1157 __skb_queue_tail(&sk->sk_write_queue, head);
1158 kcm->seq_skb = NULL;
1159 KCM_STATS_INCR(kcm->stats.tx_msgs);
1160
1161 if (flags & MSG_BATCH) {
1162 kcm->tx_wait_more = true;
1163 } else if (kcm->tx_wait_more || not_busy) {
1164 err = kcm_write_msgs(kcm);
1165 if (err < 0) {
1166 /* We got a hard error in write_msgs but have
1167 * already queued this message. Report an error
1168 * in the socket, but don't affect return value
1169 * from sendmsg
1170 */
1171 pr_warn("KCM: Hard failure on kcm_write_msgs\n");
1172 report_csk_error(&kcm->sk, -err);
1173 }
1174 }
1175 } else {
1176 /* Message not complete, save state */
1177 kcm->seq_skb = head;
1178 kcm_tx_msg(head)->last_skb = skb;
1179 }
1180
1181 KCM_STATS_ADD(kcm->stats.tx_bytes, size);
1182
1183 release_sock(sk);
1184 return size;
1185
1186 out_error:
1187 kcm_push(kcm);
1188
1189 err = sk_stream_error(sk, flags, err);
1190
1191 /* make sure we wake any epoll edge trigger waiter */
1192 if (unlikely(skb_queue_len(&sk->sk_write_queue) == 0 && err == -EAGAIN))
1193 sk->sk_write_space(sk);
1194
1195 release_sock(sk);
1196 return err;
1197 }
1198
1199 static int kcm_sendmsg(struct socket *sock, struct msghdr *msg, size_t len)
1200 {
1201 struct sock *sk = sock->sk;
1202 struct kcm_sock *kcm = kcm_sk(sk);
1203 struct sk_buff *skb = NULL, *head = NULL;
1204 size_t copy, copied = 0;
1205 long timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);
1206 int eor = (sock->type == SOCK_DGRAM) ?
1207 !(msg->msg_flags & MSG_MORE) : !!(msg->msg_flags & MSG_EOR);
1208 int err = -EPIPE;
1209
1210 lock_sock(sk);
1211
1212 /* Per tcp_sendmsg this should be in poll */
1213 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
1214
1215 if (sk->sk_err)
1216 goto out_error;
1217
1218 if (kcm->seq_skb) {
1219 /* Previously opened message */
1220 head = kcm->seq_skb;
1221 skb = kcm_tx_msg(head)->last_skb;
1222 goto start;
1223 }
1224
1225 /* Call the sk_stream functions to manage the sndbuf mem. */
1226 if (!sk_stream_memory_free(sk)) {
1227 kcm_push(kcm);
1228 set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
1229 err = sk_stream_wait_memory(sk, &timeo);
1230 if (err)
1231 goto out_error;
1232 }
1233
1234 /* New message, alloc head skb */
1235 head = alloc_skb(0, sk->sk_allocation);
1236 while (!head) {
1237 kcm_push(kcm);
1238 err = sk_stream_wait_memory(sk, &timeo);
1239 if (err)
1240 goto out_error;
1241
1242 head = alloc_skb(0, sk->sk_allocation);
1243 }
1244
1245 skb = head;
1246
1247 /* Set ip_summed to CHECKSUM_UNNECESSARY to avoid calling
1248 * csum_and_copy_from_iter from skb_do_copy_data_nocache.
1249 */
1250 skb->ip_summed = CHECKSUM_UNNECESSARY;
1251
1252 start:
1253 while (msg_data_left(msg)) {
1254 bool merge = true;
1255 int i = skb_shinfo(skb)->nr_frags;
1256 struct page_frag *pfrag = sk_page_frag(sk);
1257
1258 if (!sk_page_frag_refill(sk, pfrag))
1259 goto wait_for_memory;
1260
1261 if (!skb_can_coalesce(skb, i, pfrag->page,
1262 pfrag->offset)) {
1263 if (i == MAX_SKB_FRAGS) {
1264 struct sk_buff *tskb;
1265
1266 tskb = alloc_skb(0, sk->sk_allocation);
1267 if (!tskb)
1268 goto wait_for_memory;
1269
1270 if (head == skb)
1271 skb_shinfo(head)->frag_list = tskb;
1272 else
1273 skb->next = tskb;
1274
1275 skb = tskb;
1276 skb->ip_summed = CHECKSUM_UNNECESSARY;
1277 continue;
1278 }
1279 merge = false;
1280 }
1281
1282 copy = min_t(int, msg_data_left(msg),
1283 pfrag->size - pfrag->offset);
1284
1285 if (!sk_wmem_schedule(sk, copy))
1286 goto wait_for_memory;
1287
1288 err = skb_copy_to_page_nocache(sk, &msg->msg_iter, skb,
1289 pfrag->page,
1290 pfrag->offset,
1291 copy);
1292 if (err)
1293 goto out_error;
1294
1295 /* Update the skb. */
1296 if (merge) {
1297 skb_frag_size_add(&skb_shinfo(skb)->frags[i - 1], copy);
1298 } else {
1299 skb_fill_page_desc(skb, i, pfrag->page,
1300 pfrag->offset, copy);
1301 get_page(pfrag->page);
1302 }
1303
1304 pfrag->offset += copy;
1305 copied += copy;
1306 if (head != skb) {
1307 head->len += copy;
1308 head->data_len += copy;
1309 }
1310
1311 continue;
1312
1313 wait_for_memory:
1314 kcm_push(kcm);
1315 err = sk_stream_wait_memory(sk, &timeo);
1316 if (err)
1317 goto out_error;
1318 }
1319
1320 if (eor) {
1321 bool not_busy = skb_queue_empty(&sk->sk_write_queue);
1322
1323 /* Message complete, queue it on send buffer */
1324 __skb_queue_tail(&sk->sk_write_queue, head);
1325 kcm->seq_skb = NULL;
1326 KCM_STATS_INCR(kcm->stats.tx_msgs);
1327
1328 if (msg->msg_flags & MSG_BATCH) {
1329 kcm->tx_wait_more = true;
1330 } else if (kcm->tx_wait_more || not_busy) {
1331 err = kcm_write_msgs(kcm);
1332 if (err < 0) {
1333 /* We got a hard error in write_msgs but have
1334 * already queued this message. Report an error
1335 * in the socket, but don't affect return value
1336 * from sendmsg
1337 */
1338 pr_warn("KCM: Hard failure on kcm_write_msgs\n");
1339 report_csk_error(&kcm->sk, -err);
1340 }
1341 }
1342 } else {
1343 /* Message not complete, save state */
1344 partial_message:
1345 kcm->seq_skb = head;
1346 kcm_tx_msg(head)->last_skb = skb;
1347 }
1348
1349 KCM_STATS_ADD(kcm->stats.tx_bytes, copied);
1350
1351 release_sock(sk);
1352 return copied;
1353
1354 out_error:
1355 kcm_push(kcm);
1356
1357 if (copied && sock->type == SOCK_SEQPACKET) {
1358 /* Wrote some bytes before encountering an
1359 * error, return partial success.
1360 */
1361 goto partial_message;
1362 }
1363
1364 if (head != kcm->seq_skb)
1365 kfree_skb(head);
1366
1367 err = sk_stream_error(sk, msg->msg_flags, err);
1368
1369 /* make sure we wake any epoll edge trigger waiter */
1370 if (unlikely(skb_queue_len(&sk->sk_write_queue) == 0 && err == -EAGAIN))
1371 sk->sk_write_space(sk);
1372
1373 release_sock(sk);
1374 return err;
1375 }
1376
1377 static struct sk_buff *kcm_wait_data(struct sock *sk, int flags,
1378 long timeo, int *err)
1379 {
1380 struct sk_buff *skb;
1381
1382 while (!(skb = skb_peek(&sk->sk_receive_queue))) {
1383 if (sk->sk_err) {
1384 *err = sock_error(sk);
1385 return NULL;
1386 }
1387
1388 if (sock_flag(sk, SOCK_DONE))
1389 return NULL;
1390
1391 if ((flags & MSG_DONTWAIT) || !timeo) {
1392 *err = -EAGAIN;
1393 return NULL;
1394 }
1395
1396 sk_wait_data(sk, &timeo, NULL);
1397
1398 /* Handle signals */
1399 if (signal_pending(current)) {
1400 *err = sock_intr_errno(timeo);
1401 return NULL;
1402 }
1403 }
1404
1405 return skb;
1406 }
1407
1408 static int kcm_recvmsg(struct socket *sock, struct msghdr *msg,
1409 size_t len, int flags)
1410 {
1411 struct sock *sk = sock->sk;
1412 struct kcm_sock *kcm = kcm_sk(sk);
1413 int err = 0;
1414 long timeo;
1415 struct kcm_rx_msg *rxm;
1416 int copied = 0;
1417 struct sk_buff *skb;
1418
1419 timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1420
1421 lock_sock(sk);
1422
1423 skb = kcm_wait_data(sk, flags, timeo, &err);
1424 if (!skb)
1425 goto out;
1426
1427 /* Okay, have a message on the receive queue */
1428
1429 rxm = kcm_rx_msg(skb);
1430
1431 if (len > rxm->full_len)
1432 len = rxm->full_len;
1433
1434 err = skb_copy_datagram_msg(skb, rxm->offset, msg, len);
1435 if (err < 0)
1436 goto out;
1437
1438 copied = len;
1439 if (likely(!(flags & MSG_PEEK))) {
1440 KCM_STATS_ADD(kcm->stats.rx_bytes, copied);
1441 if (copied < rxm->full_len) {
1442 if (sock->type == SOCK_DGRAM) {
1443 /* Truncated message */
1444 msg->msg_flags |= MSG_TRUNC;
1445 goto msg_finished;
1446 }
1447 rxm->offset += copied;
1448 rxm->full_len -= copied;
1449 } else {
1450 msg_finished:
1451 /* Finished with message */
1452 msg->msg_flags |= MSG_EOR;
1453 KCM_STATS_INCR(kcm->stats.rx_msgs);
1454 skb_unlink(skb, &sk->sk_receive_queue);
1455 kfree_skb(skb);
1456 }
1457 }
1458
1459 out:
1460 release_sock(sk);
1461
1462 return copied ? : err;
1463 }
1464
1465 static ssize_t kcm_sock_splice(struct sock *sk,
1466 struct pipe_inode_info *pipe,
1467 struct splice_pipe_desc *spd)
1468 {
1469 int ret;
1470
1471 release_sock(sk);
1472 ret = splice_to_pipe(pipe, spd);
1473 lock_sock(sk);
1474
1475 return ret;
1476 }
1477
1478 static ssize_t kcm_splice_read(struct socket *sock, loff_t *ppos,
1479 struct pipe_inode_info *pipe, size_t len,
1480 unsigned int flags)
1481 {
1482 struct sock *sk = sock->sk;
1483 struct kcm_sock *kcm = kcm_sk(sk);
1484 long timeo;
1485 struct kcm_rx_msg *rxm;
1486 int err = 0;
1487 ssize_t copied;
1488 struct sk_buff *skb;
1489
1490 /* Only support splice for SOCKSEQPACKET */
1491
1492 timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1493
1494 lock_sock(sk);
1495
1496 skb = kcm_wait_data(sk, flags, timeo, &err);
1497 if (!skb)
1498 goto err_out;
1499
1500 /* Okay, have a message on the receive queue */
1501
1502 rxm = kcm_rx_msg(skb);
1503
1504 if (len > rxm->full_len)
1505 len = rxm->full_len;
1506
1507 copied = skb_splice_bits(skb, sk, rxm->offset, pipe, len, flags,
1508 kcm_sock_splice);
1509 if (copied < 0) {
1510 err = copied;
1511 goto err_out;
1512 }
1513
1514 KCM_STATS_ADD(kcm->stats.rx_bytes, copied);
1515
1516 rxm->offset += copied;
1517 rxm->full_len -= copied;
1518
1519 /* We have no way to return MSG_EOR. If all the bytes have been
1520 * read we still leave the message in the receive socket buffer.
1521 * A subsequent recvmsg needs to be done to return MSG_EOR and
1522 * finish reading the message.
1523 */
1524
1525 release_sock(sk);
1526
1527 return copied;
1528
1529 err_out:
1530 release_sock(sk);
1531
1532 return err;
1533 }
1534
1535 /* kcm sock lock held */
1536 static void kcm_recv_disable(struct kcm_sock *kcm)
1537 {
1538 struct kcm_mux *mux = kcm->mux;
1539
1540 if (kcm->rx_disabled)
1541 return;
1542
1543 spin_lock_bh(&mux->rx_lock);
1544
1545 kcm->rx_disabled = 1;
1546
1547 /* If a psock is reserved we'll do cleanup in unreserve */
1548 if (!kcm->rx_psock) {
1549 if (kcm->rx_wait) {
1550 list_del(&kcm->wait_rx_list);
1551 kcm->rx_wait = false;
1552 }
1553
1554 requeue_rx_msgs(mux, &kcm->sk.sk_receive_queue);
1555 }
1556
1557 spin_unlock_bh(&mux->rx_lock);
1558 }
1559
1560 /* kcm sock lock held */
1561 static void kcm_recv_enable(struct kcm_sock *kcm)
1562 {
1563 struct kcm_mux *mux = kcm->mux;
1564
1565 if (!kcm->rx_disabled)
1566 return;
1567
1568 spin_lock_bh(&mux->rx_lock);
1569
1570 kcm->rx_disabled = 0;
1571 kcm_rcv_ready(kcm);
1572
1573 spin_unlock_bh(&mux->rx_lock);
1574 }
1575
1576 static int kcm_setsockopt(struct socket *sock, int level, int optname,
1577 char __user *optval, unsigned int optlen)
1578 {
1579 struct kcm_sock *kcm = kcm_sk(sock->sk);
1580 int val, valbool;
1581 int err = 0;
1582
1583 if (level != SOL_KCM)
1584 return -ENOPROTOOPT;
1585
1586 if (optlen < sizeof(int))
1587 return -EINVAL;
1588
1589 if (get_user(val, (int __user *)optval))
1590 return -EINVAL;
1591
1592 valbool = val ? 1 : 0;
1593
1594 switch (optname) {
1595 case KCM_RECV_DISABLE:
1596 lock_sock(&kcm->sk);
1597 if (valbool)
1598 kcm_recv_disable(kcm);
1599 else
1600 kcm_recv_enable(kcm);
1601 release_sock(&kcm->sk);
1602 break;
1603 default:
1604 err = -ENOPROTOOPT;
1605 }
1606
1607 return err;
1608 }
1609
1610 static int kcm_getsockopt(struct socket *sock, int level, int optname,
1611 char __user *optval, int __user *optlen)
1612 {
1613 struct kcm_sock *kcm = kcm_sk(sock->sk);
1614 int val, len;
1615
1616 if (level != SOL_KCM)
1617 return -ENOPROTOOPT;
1618
1619 if (get_user(len, optlen))
1620 return -EFAULT;
1621
1622 len = min_t(unsigned int, len, sizeof(int));
1623 if (len < 0)
1624 return -EINVAL;
1625
1626 switch (optname) {
1627 case KCM_RECV_DISABLE:
1628 val = kcm->rx_disabled;
1629 break;
1630 default:
1631 return -ENOPROTOOPT;
1632 }
1633
1634 if (put_user(len, optlen))
1635 return -EFAULT;
1636 if (copy_to_user(optval, &val, len))
1637 return -EFAULT;
1638 return 0;
1639 }
1640
1641 static void init_kcm_sock(struct kcm_sock *kcm, struct kcm_mux *mux)
1642 {
1643 struct kcm_sock *tkcm;
1644 struct list_head *head;
1645 int index = 0;
1646
1647 /* For SOCK_SEQPACKET sock type, datagram_poll checks the sk_state, so
1648 * we set sk_state, otherwise epoll_wait always returns right away with
1649 * POLLHUP
1650 */
1651 kcm->sk.sk_state = TCP_ESTABLISHED;
1652
1653 /* Add to mux's kcm sockets list */
1654 kcm->mux = mux;
1655 spin_lock_bh(&mux->lock);
1656
1657 head = &mux->kcm_socks;
1658 list_for_each_entry(tkcm, &mux->kcm_socks, kcm_sock_list) {
1659 if (tkcm->index != index)
1660 break;
1661 head = &tkcm->kcm_sock_list;
1662 index++;
1663 }
1664
1665 list_add(&kcm->kcm_sock_list, head);
1666 kcm->index = index;
1667
1668 mux->kcm_socks_cnt++;
1669 spin_unlock_bh(&mux->lock);
1670
1671 INIT_WORK(&kcm->tx_work, kcm_tx_work);
1672
1673 spin_lock_bh(&mux->rx_lock);
1674 kcm_rcv_ready(kcm);
1675 spin_unlock_bh(&mux->rx_lock);
1676 }
1677
1678 static void kcm_rx_msg_timeout(unsigned long arg)
1679 {
1680 struct kcm_psock *psock = (struct kcm_psock *)arg;
1681
1682 /* Message assembly timed out */
1683 KCM_STATS_INCR(psock->stats.rx_msg_timeouts);
1684 kcm_abort_rx_psock(psock, ETIMEDOUT, NULL);
1685 }
1686
1687 static int kcm_attach(struct socket *sock, struct socket *csock,
1688 struct bpf_prog *prog)
1689 {
1690 struct kcm_sock *kcm = kcm_sk(sock->sk);
1691 struct kcm_mux *mux = kcm->mux;
1692 struct sock *csk;
1693 struct kcm_psock *psock = NULL, *tpsock;
1694 struct list_head *head;
1695 int index = 0;
1696
1697 if (csock->ops->family != PF_INET &&
1698 csock->ops->family != PF_INET6)
1699 return -EINVAL;
1700
1701 csk = csock->sk;
1702 if (!csk)
1703 return -EINVAL;
1704
1705 /* Only support TCP for now */
1706 if (csk->sk_protocol != IPPROTO_TCP)
1707 return -EINVAL;
1708
1709 psock = kmem_cache_zalloc(kcm_psockp, GFP_KERNEL);
1710 if (!psock)
1711 return -ENOMEM;
1712
1713 psock->mux = mux;
1714 psock->sk = csk;
1715 psock->bpf_prog = prog;
1716
1717 setup_timer(&psock->rx_msg_timer, kcm_rx_msg_timeout,
1718 (unsigned long)psock);
1719
1720 INIT_WORK(&psock->rx_work, psock_rx_work);
1721 INIT_DELAYED_WORK(&psock->rx_delayed_work, psock_rx_delayed_work);
1722
1723 sock_hold(csk);
1724
1725 write_lock_bh(&csk->sk_callback_lock);
1726 psock->save_data_ready = csk->sk_data_ready;
1727 psock->save_write_space = csk->sk_write_space;
1728 psock->save_state_change = csk->sk_state_change;
1729 csk->sk_user_data = psock;
1730 csk->sk_data_ready = psock_tcp_data_ready;
1731 csk->sk_write_space = psock_tcp_write_space;
1732 csk->sk_state_change = psock_tcp_state_change;
1733 write_unlock_bh(&csk->sk_callback_lock);
1734
1735 /* Finished initialization, now add the psock to the MUX. */
1736 spin_lock_bh(&mux->lock);
1737 head = &mux->psocks;
1738 list_for_each_entry(tpsock, &mux->psocks, psock_list) {
1739 if (tpsock->index != index)
1740 break;
1741 head = &tpsock->psock_list;
1742 index++;
1743 }
1744
1745 list_add(&psock->psock_list, head);
1746 psock->index = index;
1747
1748 KCM_STATS_INCR(mux->stats.psock_attach);
1749 mux->psocks_cnt++;
1750 psock_now_avail(psock);
1751 spin_unlock_bh(&mux->lock);
1752
1753 /* Schedule RX work in case there are already bytes queued */
1754 queue_work(kcm_wq, &psock->rx_work);
1755
1756 return 0;
1757 }
1758
1759 static int kcm_attach_ioctl(struct socket *sock, struct kcm_attach *info)
1760 {
1761 struct socket *csock;
1762 struct bpf_prog *prog;
1763 int err;
1764
1765 csock = sockfd_lookup(info->fd, &err);
1766 if (!csock)
1767 return -ENOENT;
1768
1769 prog = bpf_prog_get_type(info->bpf_fd, BPF_PROG_TYPE_SOCKET_FILTER);
1770 if (IS_ERR(prog)) {
1771 err = PTR_ERR(prog);
1772 goto out;
1773 }
1774
1775 err = kcm_attach(sock, csock, prog);
1776 if (err) {
1777 bpf_prog_put(prog);
1778 goto out;
1779 }
1780
1781 /* Keep reference on file also */
1782
1783 return 0;
1784 out:
1785 fput(csock->file);
1786 return err;
1787 }
1788
1789 static void kcm_unattach(struct kcm_psock *psock)
1790 {
1791 struct sock *csk = psock->sk;
1792 struct kcm_mux *mux = psock->mux;
1793
1794 /* Stop getting callbacks from TCP socket. After this there should
1795 * be no way to reserve a kcm for this psock.
1796 */
1797 write_lock_bh(&csk->sk_callback_lock);
1798 csk->sk_user_data = NULL;
1799 csk->sk_data_ready = psock->save_data_ready;
1800 csk->sk_write_space = psock->save_write_space;
1801 csk->sk_state_change = psock->save_state_change;
1802 psock->rx_stopped = 1;
1803
1804 if (WARN_ON(psock->rx_kcm)) {
1805 write_unlock_bh(&csk->sk_callback_lock);
1806 return;
1807 }
1808
1809 spin_lock_bh(&mux->rx_lock);
1810
1811 /* Stop receiver activities. After this point psock should not be
1812 * able to get onto ready list either through callbacks or work.
1813 */
1814 if (psock->ready_rx_msg) {
1815 list_del(&psock->psock_ready_list);
1816 kfree_skb(psock->ready_rx_msg);
1817 psock->ready_rx_msg = NULL;
1818 KCM_STATS_INCR(mux->stats.rx_ready_drops);
1819 }
1820
1821 spin_unlock_bh(&mux->rx_lock);
1822
1823 write_unlock_bh(&csk->sk_callback_lock);
1824
1825 del_timer_sync(&psock->rx_msg_timer);
1826 cancel_work_sync(&psock->rx_work);
1827 cancel_delayed_work_sync(&psock->rx_delayed_work);
1828
1829 bpf_prog_put(psock->bpf_prog);
1830
1831 kfree_skb(psock->rx_skb_head);
1832 psock->rx_skb_head = NULL;
1833
1834 spin_lock_bh(&mux->lock);
1835
1836 aggregate_psock_stats(&psock->stats, &mux->aggregate_psock_stats);
1837
1838 KCM_STATS_INCR(mux->stats.psock_unattach);
1839
1840 if (psock->tx_kcm) {
1841 /* psock was reserved. Just mark it finished and we will clean
1842 * up in the kcm paths, we need kcm lock which can not be
1843 * acquired here.
1844 */
1845 KCM_STATS_INCR(mux->stats.psock_unattach_rsvd);
1846 spin_unlock_bh(&mux->lock);
1847
1848 /* We are unattaching a socket that is reserved. Abort the
1849 * socket since we may be out of sync in sending on it. We need
1850 * to do this without the mux lock.
1851 */
1852 kcm_abort_tx_psock(psock, EPIPE, false);
1853
1854 spin_lock_bh(&mux->lock);
1855 if (!psock->tx_kcm) {
1856 /* psock now unreserved in window mux was unlocked */
1857 goto no_reserved;
1858 }
1859 psock->done = 1;
1860
1861 /* Commit done before queuing work to process it */
1862 smp_mb();
1863
1864 /* Queue tx work to make sure psock->done is handled */
1865 queue_work(kcm_wq, &psock->tx_kcm->tx_work);
1866 spin_unlock_bh(&mux->lock);
1867 } else {
1868 no_reserved:
1869 if (!psock->tx_stopped)
1870 list_del(&psock->psock_avail_list);
1871 list_del(&psock->psock_list);
1872 mux->psocks_cnt--;
1873 spin_unlock_bh(&mux->lock);
1874
1875 sock_put(csk);
1876 fput(csk->sk_socket->file);
1877 kmem_cache_free(kcm_psockp, psock);
1878 }
1879 }
1880
1881 static int kcm_unattach_ioctl(struct socket *sock, struct kcm_unattach *info)
1882 {
1883 struct kcm_sock *kcm = kcm_sk(sock->sk);
1884 struct kcm_mux *mux = kcm->mux;
1885 struct kcm_psock *psock;
1886 struct socket *csock;
1887 struct sock *csk;
1888 int err;
1889
1890 csock = sockfd_lookup(info->fd, &err);
1891 if (!csock)
1892 return -ENOENT;
1893
1894 csk = csock->sk;
1895 if (!csk) {
1896 err = -EINVAL;
1897 goto out;
1898 }
1899
1900 err = -ENOENT;
1901
1902 spin_lock_bh(&mux->lock);
1903
1904 list_for_each_entry(psock, &mux->psocks, psock_list) {
1905 if (psock->sk != csk)
1906 continue;
1907
1908 /* Found the matching psock */
1909
1910 if (psock->unattaching || WARN_ON(psock->done)) {
1911 err = -EALREADY;
1912 break;
1913 }
1914
1915 psock->unattaching = 1;
1916
1917 spin_unlock_bh(&mux->lock);
1918
1919 kcm_unattach(psock);
1920
1921 err = 0;
1922 goto out;
1923 }
1924
1925 spin_unlock_bh(&mux->lock);
1926
1927 out:
1928 fput(csock->file);
1929 return err;
1930 }
1931
1932 static struct proto kcm_proto = {
1933 .name = "KCM",
1934 .owner = THIS_MODULE,
1935 .obj_size = sizeof(struct kcm_sock),
1936 };
1937
1938 /* Clone a kcm socket. */
1939 static int kcm_clone(struct socket *osock, struct kcm_clone *info,
1940 struct socket **newsockp)
1941 {
1942 struct socket *newsock;
1943 struct sock *newsk;
1944 struct file *newfile;
1945 int err, newfd;
1946
1947 err = -ENFILE;
1948 newsock = sock_alloc();
1949 if (!newsock)
1950 goto out;
1951
1952 newsock->type = osock->type;
1953 newsock->ops = osock->ops;
1954
1955 __module_get(newsock->ops->owner);
1956
1957 newfd = get_unused_fd_flags(0);
1958 if (unlikely(newfd < 0)) {
1959 err = newfd;
1960 goto out_fd_fail;
1961 }
1962
1963 newfile = sock_alloc_file(newsock, 0, osock->sk->sk_prot_creator->name);
1964 if (unlikely(IS_ERR(newfile))) {
1965 err = PTR_ERR(newfile);
1966 goto out_sock_alloc_fail;
1967 }
1968
1969 newsk = sk_alloc(sock_net(osock->sk), PF_KCM, GFP_KERNEL,
1970 &kcm_proto, true);
1971 if (!newsk) {
1972 err = -ENOMEM;
1973 goto out_sk_alloc_fail;
1974 }
1975
1976 sock_init_data(newsock, newsk);
1977 init_kcm_sock(kcm_sk(newsk), kcm_sk(osock->sk)->mux);
1978
1979 fd_install(newfd, newfile);
1980 *newsockp = newsock;
1981 info->fd = newfd;
1982
1983 return 0;
1984
1985 out_sk_alloc_fail:
1986 fput(newfile);
1987 out_sock_alloc_fail:
1988 put_unused_fd(newfd);
1989 out_fd_fail:
1990 sock_release(newsock);
1991 out:
1992 return err;
1993 }
1994
1995 static int kcm_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
1996 {
1997 int err;
1998
1999 switch (cmd) {
2000 case SIOCKCMATTACH: {
2001 struct kcm_attach info;
2002
2003 if (copy_from_user(&info, (void __user *)arg, sizeof(info)))
2004 err = -EFAULT;
2005
2006 err = kcm_attach_ioctl(sock, &info);
2007
2008 break;
2009 }
2010 case SIOCKCMUNATTACH: {
2011 struct kcm_unattach info;
2012
2013 if (copy_from_user(&info, (void __user *)arg, sizeof(info)))
2014 err = -EFAULT;
2015
2016 err = kcm_unattach_ioctl(sock, &info);
2017
2018 break;
2019 }
2020 case SIOCKCMCLONE: {
2021 struct kcm_clone info;
2022 struct socket *newsock = NULL;
2023
2024 if (copy_from_user(&info, (void __user *)arg, sizeof(info)))
2025 err = -EFAULT;
2026
2027 err = kcm_clone(sock, &info, &newsock);
2028
2029 if (!err) {
2030 if (copy_to_user((void __user *)arg, &info,
2031 sizeof(info))) {
2032 err = -EFAULT;
2033 sys_close(info.fd);
2034 }
2035 }
2036
2037 break;
2038 }
2039 default:
2040 err = -ENOIOCTLCMD;
2041 break;
2042 }
2043
2044 return err;
2045 }
2046
2047 static void free_mux(struct rcu_head *rcu)
2048 {
2049 struct kcm_mux *mux = container_of(rcu,
2050 struct kcm_mux, rcu);
2051
2052 kmem_cache_free(kcm_muxp, mux);
2053 }
2054
2055 static void release_mux(struct kcm_mux *mux)
2056 {
2057 struct kcm_net *knet = mux->knet;
2058 struct kcm_psock *psock, *tmp_psock;
2059
2060 /* Release psocks */
2061 list_for_each_entry_safe(psock, tmp_psock,
2062 &mux->psocks, psock_list) {
2063 if (!WARN_ON(psock->unattaching))
2064 kcm_unattach(psock);
2065 }
2066
2067 if (WARN_ON(mux->psocks_cnt))
2068 return;
2069
2070 __skb_queue_purge(&mux->rx_hold_queue);
2071
2072 mutex_lock(&knet->mutex);
2073 aggregate_mux_stats(&mux->stats, &knet->aggregate_mux_stats);
2074 aggregate_psock_stats(&mux->aggregate_psock_stats,
2075 &knet->aggregate_psock_stats);
2076 list_del_rcu(&mux->kcm_mux_list);
2077 knet->count--;
2078 mutex_unlock(&knet->mutex);
2079
2080 call_rcu(&mux->rcu, free_mux);
2081 }
2082
2083 static void kcm_done(struct kcm_sock *kcm)
2084 {
2085 struct kcm_mux *mux = kcm->mux;
2086 struct sock *sk = &kcm->sk;
2087 int socks_cnt;
2088
2089 spin_lock_bh(&mux->rx_lock);
2090 if (kcm->rx_psock) {
2091 /* Cleanup in unreserve_rx_kcm */
2092 WARN_ON(kcm->done);
2093 kcm->rx_disabled = 1;
2094 kcm->done = 1;
2095 spin_unlock_bh(&mux->rx_lock);
2096 return;
2097 }
2098
2099 if (kcm->rx_wait) {
2100 list_del(&kcm->wait_rx_list);
2101 kcm->rx_wait = false;
2102 }
2103 /* Move any pending receive messages to other kcm sockets */
2104 requeue_rx_msgs(mux, &sk->sk_receive_queue);
2105
2106 spin_unlock_bh(&mux->rx_lock);
2107
2108 if (WARN_ON(sk_rmem_alloc_get(sk)))
2109 return;
2110
2111 /* Detach from MUX */
2112 spin_lock_bh(&mux->lock);
2113
2114 list_del(&kcm->kcm_sock_list);
2115 mux->kcm_socks_cnt--;
2116 socks_cnt = mux->kcm_socks_cnt;
2117
2118 spin_unlock_bh(&mux->lock);
2119
2120 if (!socks_cnt) {
2121 /* We are done with the mux now. */
2122 release_mux(mux);
2123 }
2124
2125 WARN_ON(kcm->rx_wait);
2126
2127 sock_put(&kcm->sk);
2128 }
2129
2130 /* Called by kcm_release to close a KCM socket.
2131 * If this is the last KCM socket on the MUX, destroy the MUX.
2132 */
2133 static int kcm_release(struct socket *sock)
2134 {
2135 struct sock *sk = sock->sk;
2136 struct kcm_sock *kcm;
2137 struct kcm_mux *mux;
2138 struct kcm_psock *psock;
2139
2140 if (!sk)
2141 return 0;
2142
2143 kcm = kcm_sk(sk);
2144 mux = kcm->mux;
2145
2146 sock_orphan(sk);
2147 kfree_skb(kcm->seq_skb);
2148
2149 lock_sock(sk);
2150 /* Purge queue under lock to avoid race condition with tx_work trying
2151 * to act when queue is nonempty. If tx_work runs after this point
2152 * it will just return.
2153 */
2154 __skb_queue_purge(&sk->sk_write_queue);
2155 release_sock(sk);
2156
2157 spin_lock_bh(&mux->lock);
2158 if (kcm->tx_wait) {
2159 /* Take of tx_wait list, after this point there should be no way
2160 * that a psock will be assigned to this kcm.
2161 */
2162 list_del(&kcm->wait_psock_list);
2163 kcm->tx_wait = false;
2164 }
2165 spin_unlock_bh(&mux->lock);
2166
2167 /* Cancel work. After this point there should be no outside references
2168 * to the kcm socket.
2169 */
2170 cancel_work_sync(&kcm->tx_work);
2171
2172 lock_sock(sk);
2173 psock = kcm->tx_psock;
2174 if (psock) {
2175 /* A psock was reserved, so we need to kill it since it
2176 * may already have some bytes queued from a message. We
2177 * need to do this after removing kcm from tx_wait list.
2178 */
2179 kcm_abort_tx_psock(psock, EPIPE, false);
2180 unreserve_psock(kcm);
2181 }
2182 release_sock(sk);
2183
2184 WARN_ON(kcm->tx_wait);
2185 WARN_ON(kcm->tx_psock);
2186
2187 sock->sk = NULL;
2188
2189 kcm_done(kcm);
2190
2191 return 0;
2192 }
2193
2194 static const struct proto_ops kcm_dgram_ops = {
2195 .family = PF_KCM,
2196 .owner = THIS_MODULE,
2197 .release = kcm_release,
2198 .bind = sock_no_bind,
2199 .connect = sock_no_connect,
2200 .socketpair = sock_no_socketpair,
2201 .accept = sock_no_accept,
2202 .getname = sock_no_getname,
2203 .poll = datagram_poll,
2204 .ioctl = kcm_ioctl,
2205 .listen = sock_no_listen,
2206 .shutdown = sock_no_shutdown,
2207 .setsockopt = kcm_setsockopt,
2208 .getsockopt = kcm_getsockopt,
2209 .sendmsg = kcm_sendmsg,
2210 .recvmsg = kcm_recvmsg,
2211 .mmap = sock_no_mmap,
2212 .sendpage = kcm_sendpage,
2213 };
2214
2215 static const struct proto_ops kcm_seqpacket_ops = {
2216 .family = PF_KCM,
2217 .owner = THIS_MODULE,
2218 .release = kcm_release,
2219 .bind = sock_no_bind,
2220 .connect = sock_no_connect,
2221 .socketpair = sock_no_socketpair,
2222 .accept = sock_no_accept,
2223 .getname = sock_no_getname,
2224 .poll = datagram_poll,
2225 .ioctl = kcm_ioctl,
2226 .listen = sock_no_listen,
2227 .shutdown = sock_no_shutdown,
2228 .setsockopt = kcm_setsockopt,
2229 .getsockopt = kcm_getsockopt,
2230 .sendmsg = kcm_sendmsg,
2231 .recvmsg = kcm_recvmsg,
2232 .mmap = sock_no_mmap,
2233 .sendpage = kcm_sendpage,
2234 .splice_read = kcm_splice_read,
2235 };
2236
2237 /* Create proto operation for kcm sockets */
2238 static int kcm_create(struct net *net, struct socket *sock,
2239 int protocol, int kern)
2240 {
2241 struct kcm_net *knet = net_generic(net, kcm_net_id);
2242 struct sock *sk;
2243 struct kcm_mux *mux;
2244
2245 switch (sock->type) {
2246 case SOCK_DGRAM:
2247 sock->ops = &kcm_dgram_ops;
2248 break;
2249 case SOCK_SEQPACKET:
2250 sock->ops = &kcm_seqpacket_ops;
2251 break;
2252 default:
2253 return -ESOCKTNOSUPPORT;
2254 }
2255
2256 if (protocol != KCMPROTO_CONNECTED)
2257 return -EPROTONOSUPPORT;
2258
2259 sk = sk_alloc(net, PF_KCM, GFP_KERNEL, &kcm_proto, kern);
2260 if (!sk)
2261 return -ENOMEM;
2262
2263 /* Allocate a kcm mux, shared between KCM sockets */
2264 mux = kmem_cache_zalloc(kcm_muxp, GFP_KERNEL);
2265 if (!mux) {
2266 sk_free(sk);
2267 return -ENOMEM;
2268 }
2269
2270 spin_lock_init(&mux->lock);
2271 spin_lock_init(&mux->rx_lock);
2272 INIT_LIST_HEAD(&mux->kcm_socks);
2273 INIT_LIST_HEAD(&mux->kcm_rx_waiters);
2274 INIT_LIST_HEAD(&mux->kcm_tx_waiters);
2275
2276 INIT_LIST_HEAD(&mux->psocks);
2277 INIT_LIST_HEAD(&mux->psocks_ready);
2278 INIT_LIST_HEAD(&mux->psocks_avail);
2279
2280 mux->knet = knet;
2281
2282 /* Add new MUX to list */
2283 mutex_lock(&knet->mutex);
2284 list_add_rcu(&mux->kcm_mux_list, &knet->mux_list);
2285 knet->count++;
2286 mutex_unlock(&knet->mutex);
2287
2288 skb_queue_head_init(&mux->rx_hold_queue);
2289
2290 /* Init KCM socket */
2291 sock_init_data(sock, sk);
2292 init_kcm_sock(kcm_sk(sk), mux);
2293
2294 return 0;
2295 }
2296
2297 static struct net_proto_family kcm_family_ops = {
2298 .family = PF_KCM,
2299 .create = kcm_create,
2300 .owner = THIS_MODULE,
2301 };
2302
2303 static __net_init int kcm_init_net(struct net *net)
2304 {
2305 struct kcm_net *knet = net_generic(net, kcm_net_id);
2306
2307 INIT_LIST_HEAD_RCU(&knet->mux_list);
2308 mutex_init(&knet->mutex);
2309
2310 return 0;
2311 }
2312
2313 static __net_exit void kcm_exit_net(struct net *net)
2314 {
2315 struct kcm_net *knet = net_generic(net, kcm_net_id);
2316
2317 /* All KCM sockets should be closed at this point, which should mean
2318 * that all multiplexors and psocks have been destroyed.
2319 */
2320 WARN_ON(!list_empty(&knet->mux_list));
2321 }
2322
2323 static struct pernet_operations kcm_net_ops = {
2324 .init = kcm_init_net,
2325 .exit = kcm_exit_net,
2326 .id = &kcm_net_id,
2327 .size = sizeof(struct kcm_net),
2328 };
2329
2330 static int __init kcm_init(void)
2331 {
2332 int err = -ENOMEM;
2333
2334 kcm_muxp = kmem_cache_create("kcm_mux_cache",
2335 sizeof(struct kcm_mux), 0,
2336 SLAB_HWCACHE_ALIGN | SLAB_PANIC, NULL);
2337 if (!kcm_muxp)
2338 goto fail;
2339
2340 kcm_psockp = kmem_cache_create("kcm_psock_cache",
2341 sizeof(struct kcm_psock), 0,
2342 SLAB_HWCACHE_ALIGN | SLAB_PANIC, NULL);
2343 if (!kcm_psockp)
2344 goto fail;
2345
2346 kcm_wq = create_singlethread_workqueue("kkcmd");
2347 if (!kcm_wq)
2348 goto fail;
2349
2350 err = proto_register(&kcm_proto, 1);
2351 if (err)
2352 goto fail;
2353
2354 err = sock_register(&kcm_family_ops);
2355 if (err)
2356 goto sock_register_fail;
2357
2358 err = register_pernet_device(&kcm_net_ops);
2359 if (err)
2360 goto net_ops_fail;
2361
2362 err = kcm_proc_init();
2363 if (err)
2364 goto proc_init_fail;
2365
2366 return 0;
2367
2368 proc_init_fail:
2369 unregister_pernet_device(&kcm_net_ops);
2370
2371 net_ops_fail:
2372 sock_unregister(PF_KCM);
2373
2374 sock_register_fail:
2375 proto_unregister(&kcm_proto);
2376
2377 fail:
2378 kmem_cache_destroy(kcm_muxp);
2379 kmem_cache_destroy(kcm_psockp);
2380
2381 if (kcm_wq)
2382 destroy_workqueue(kcm_wq);
2383
2384 return err;
2385 }
2386
2387 static void __exit kcm_exit(void)
2388 {
2389 kcm_proc_exit();
2390 unregister_pernet_device(&kcm_net_ops);
2391 sock_unregister(PF_KCM);
2392 proto_unregister(&kcm_proto);
2393 destroy_workqueue(kcm_wq);
2394
2395 kmem_cache_destroy(kcm_muxp);
2396 kmem_cache_destroy(kcm_psockp);
2397 }
2398
2399 module_init(kcm_init);
2400 module_exit(kcm_exit);
2401
2402 MODULE_LICENSE("GPL");
2403 MODULE_ALIAS_NETPROTO(PF_KCM);
2404
Linux with added FPC-III support