projects/linux·blob master · refs

1589 lines · 41.7 KiB

1/*
2 * Copyright (c) 2006, 2018 Oracle and/or its affiliates. All rights reserved.
3 *
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the
8 * OpenIB.org BSD license below:
9 *
10 * Redistribution and use in source and binary forms, with or
11 * without modification, are permitted provided that the following
12 * conditions are met:
13 *
14 * - Redistributions of source code must retain the above
15 * copyright notice, this list of conditions and the following
16 * disclaimer.
17 *
18 * - Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials
21 * provided with the distribution.
22 *
23 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
27 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
28 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
29 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
30 * SOFTWARE.
31 *
32 */
33#include <linux/kernel.h>
34#include <linux/moduleparam.h>
35#include <linux/gfp.h>
36#include <net/sock.h>
37#include <linux/in.h>
38#include <linux/list.h>
39#include <linux/ratelimit.h>
40#include <linux/export.h>
41#include <linux/sizes.h>
42
43#include "rds.h"
44
45/* When transmitting messages in rds_send_xmit, we need to emerge from
46 * time to time and briefly release the CPU. Otherwise the softlock watchdog
47 * will kick our shin.
48 * Also, it seems fairer to not let one busy connection stall all the
49 * others.
50 *
51 * send_batch_count is the number of times we'll loop in send_xmit. Setting
52 * it to 0 will restore the old behavior (where we looped until we had
53 * drained the queue).
54 */
55static int send_batch_count = SZ_1K;
56module_param(send_batch_count, int, 0444);
57MODULE_PARM_DESC(send_batch_count, " batch factor when working the send queue");
58
59static void rds_send_remove_from_sock(struct list_head *messages, int status);
60
61/*
62 * Reset the send state. Callers must ensure that this doesn't race with
63 * rds_send_xmit().
64 */
65void rds_send_path_reset(struct rds_conn_path *cp)
66{
67 struct rds_message *rm, *tmp;
68 unsigned long flags;
69
70 if (cp->cp_xmit_rm) {
71 rm = cp->cp_xmit_rm;
72 cp->cp_xmit_rm = NULL;
73 /* Tell the user the RDMA op is no longer mapped by the
74 * transport. This isn't entirely true (it's flushed out
75 * independently) but as the connection is down, there's
76 * no ongoing RDMA to/from that memory */
77 rds_message_unmapped(rm);
78 rds_message_put(rm);
79 }
80
81 cp->cp_xmit_sg = 0;
82 cp->cp_xmit_hdr_off = 0;
83 cp->cp_xmit_data_off = 0;
84 cp->cp_xmit_atomic_sent = 0;
85 cp->cp_xmit_rdma_sent = 0;
86 cp->cp_xmit_data_sent = 0;
87
88 cp->cp_conn->c_map_queued = 0;
89
90 cp->cp_unacked_packets = rds_sysctl_max_unacked_packets;
91 cp->cp_unacked_bytes = rds_sysctl_max_unacked_bytes;
92
93 /* Mark messages as retransmissions, and move them to the send q */
94 spin_lock_irqsave(&cp->cp_lock, flags);
95 list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
96 set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
97 set_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags);
98 }
99 list_splice_init(&cp->cp_retrans, &cp->cp_send_queue);
100 spin_unlock_irqrestore(&cp->cp_lock, flags);
101}
102EXPORT_SYMBOL_GPL(rds_send_path_reset);
103
104static int acquire_in_xmit(struct rds_conn_path *cp)
105{
106 return test_and_set_bit_lock(RDS_IN_XMIT, &cp->cp_flags) == 0;
107}
108
109static void release_in_xmit(struct rds_conn_path *cp)
110{
111 clear_bit_unlock(RDS_IN_XMIT, &cp->cp_flags);
112 /*
113 * We don't use wait_on_bit()/wake_up_bit() because our waking is in a
114 * hot path and finding waiters is very rare. We don't want to walk
115 * the system-wide hashed waitqueue buckets in the fast path only to
116 * almost never find waiters.
117 *
118 * wq_has_sleeper() supplies the full barrier that orders the wait
119 * queue read after the bit clear; clear_bit_unlock() alone is only
120 * a release and would let this check read a stale empty queue,
121 * losing the wake-up.
122 */
123 if (wq_has_sleeper(&cp->cp_waitq))
124 wake_up_all(&cp->cp_waitq);
125}
126
127/*
128 * Helper function for multipath fanout to ensure lane 0 transmits queued
129 * messages before other lanes to prevent out-of-order delivery.
130 *
131 * Returns true if lane 0 still has messages or false otherwise
132 */
133static bool rds_mprds_cp0_catchup(struct rds_connection *conn)
134{
135 struct rds_conn_path *cp0 = conn->c_path;
136 struct rds_message *rm0;
137 unsigned long flags;
138 bool ret = false;
139
140 spin_lock_irqsave(&cp0->cp_lock, flags);
141
142 /* the oldest / first message in the retransmit queue
143 * has to be at or beyond c_cp0_mprds_catchup_tx_seq
144 */
145 if (!list_empty(&cp0->cp_retrans)) {
146 rm0 = list_entry(cp0->cp_retrans.next, struct rds_message,
147 m_conn_item);
148 if (be64_to_cpu(rm0->m_inc.i_hdr.h_sequence) <
149 conn->c_cp0_mprds_catchup_tx_seq) {
150 /* the retransmit queue of cp_index#0 has not
151 * quite caught up yet
152 */
153 ret = true;
154 goto unlock;
155 }
156 }
157
158 /* the oldest / first message of the send queue
159 * has to be at or beyond c_cp0_mprds_catchup_tx_seq
160 */
161 rm0 = cp0->cp_xmit_rm;
162 if (!rm0 && !list_empty(&cp0->cp_send_queue))
163 rm0 = list_entry(cp0->cp_send_queue.next, struct rds_message,
164 m_conn_item);
165 if (rm0 && be64_to_cpu(rm0->m_inc.i_hdr.h_sequence) <
166 conn->c_cp0_mprds_catchup_tx_seq) {
167 /* the send queue of cp_index#0 has not quite
168 * caught up yet
169 */
170 ret = true;
171 }
172
173unlock:
174 spin_unlock_irqrestore(&cp0->cp_lock, flags);
175 return ret;
176}
177
178/*
179 * We're making the conscious trade-off here to only send one message
180 * down the connection at a time.
181 * Pro:
182 * - tx queueing is a simple fifo list
183 * - reassembly is optional and easily done by transports per conn
184 * - no per flow rx lookup at all, straight to the socket
185 * - less per-frag memory and wire overhead
186 * Con:
187 * - queued acks can be delayed behind large messages
188 * Depends:
189 * - small message latency is higher behind queued large messages
190 * - large message latency isn't starved by intervening small sends
191 */
192int rds_send_xmit(struct rds_conn_path *cp)
193{
194 struct rds_connection *conn = cp->cp_conn;
195 struct rds_message *rm;
196 unsigned long flags;
197 unsigned int tmp;
198 struct scatterlist *sg;
199 int ret = 0;
200 LIST_HEAD(to_be_dropped);
201 int batch_count;
202 unsigned long send_gen = 0;
203 int same_rm = 0;
204
205restart:
206 batch_count = 0;
207
208 /* The drop processing after over_batch relies on
209 * rds_send_remove_from_sock() emptying to_be_dropped entry by
210 * entry; warn if that post-condition ever stops holding, and
211 * re-initialize the list head.
212 */
213 WARN_ON_ONCE(!list_empty(&to_be_dropped));
214 INIT_LIST_HEAD(&to_be_dropped);
215
216 /*
217 * sendmsg calls here after having queued its message on the send
218 * queue. We only have one task feeding the connection at a time. If
219 * another thread is already feeding the queue then we back off. This
220 * avoids blocking the caller and trading per-connection data between
221 * caches per message.
222 */
223 if (!acquire_in_xmit(cp)) {
224 rds_stats_inc(s_send_lock_contention);
225 ret = -ENOMEM;
226 goto out;
227 }
228
229 if (rds_destroy_pending(cp->cp_conn)) {
230 release_in_xmit(cp);
231 ret = -ENETUNREACH; /* dont requeue send work */
232 goto out;
233 }
234
235 /*
236 * we record the send generation after doing the xmit acquire.
237 * if someone else manages to jump in and do some work, we'll use
238 * this to avoid a goto restart farther down.
239 *
240 * The acquire_in_xmit() check above ensures that only one
241 * caller can increment c_send_gen at any time.
242 */
243 send_gen = READ_ONCE(cp->cp_send_gen) + 1;
244 WRITE_ONCE(cp->cp_send_gen, send_gen);
245
246 /*
247 * rds_conn_shutdown() sets the conn state and then acquires
248 * RDS_IN_XMIT; we take the lock first and then check the state.
249 * Ownership is decided by the atomic RMW on the cp_flags word:
250 * if the teardown won the bit we back off here, and if we won
251 * it the teardown waits until we release it.
252 */
253 if (!rds_conn_path_up(cp)) {
254 release_in_xmit(cp);
255 ret = 0;
256 goto out;
257 }
258
259 if (conn->c_trans->xmit_path_prepare)
260 conn->c_trans->xmit_path_prepare(cp);
261
262 /*
263 * spin trying to push headers and data down the connection until
264 * the connection doesn't make forward progress.
265 */
266 while (1) {
267
268 rm = cp->cp_xmit_rm;
269
270 if (!rm) {
271 same_rm = 0;
272 } else {
273 same_rm++;
274 if (same_rm >= 4096) {
275 rds_stats_inc(s_send_stuck_rm);
276 ret = -EAGAIN;
277 break;
278 }
279 }
280
281 /*
282 * If between sending messages, we can send a pending congestion
283 * map update.
284 */
285 if (!rm && test_and_clear_bit(0, &conn->c_map_queued)) {
286 rm = rds_cong_update_alloc(conn);
287 if (IS_ERR(rm)) {
288 ret = PTR_ERR(rm);
289 break;
290 }
291 rm->data.op_active = 1;
292 rm->m_inc.i_conn_path = cp;
293 rm->m_inc.i_conn = cp->cp_conn;
294
295 cp->cp_xmit_rm = rm;
296 }
297
298 /*
299 * If not already working on one, grab the next message.
300 *
301 * cp_xmit_rm holds a ref while we're sending this message down
302 * the connection. We can use this ref while holding the
303 * send_sem.. rds_send_path_reset() is serialized with it.
304 */
305 if (!rm) {
306 unsigned int len;
307
308 batch_count++;
309
310 /* we want to process as big a batch as we can, but
311 * we also want to avoid softlockups. If we've been
312 * through a lot of messages, lets back off and see
313 * if anyone else jumps in
314 */
315 if (batch_count >= send_batch_count)
316 goto over_batch;
317
318 /* make sure cp_index#0 caught up during fan-out in
319 * order to avoid lane races
320 */
321 if (cp->cp_index > 0 && rds_mprds_cp0_catchup(conn)) {
322 rds_stats_inc(s_mprds_catchup_tx0_retries);
323 goto over_batch;
324 }
325
326 spin_lock_irqsave(&cp->cp_lock, flags);
327
328 if (!list_empty(&cp->cp_send_queue)) {
329 rm = list_entry(cp->cp_send_queue.next,
330 struct rds_message,
331 m_conn_item);
332 rds_message_addref(rm);
333
334 /*
335 * Move the message from the send queue to the retransmit
336 * list right away.
337 */
338 list_move_tail(&rm->m_conn_item,
339 &cp->cp_retrans);
340 }
341
342 spin_unlock_irqrestore(&cp->cp_lock, flags);
343
344 if (!rm)
345 break;
346
347 /* Unfortunately, the way Infiniband deals with
348 * RDMA to a bad MR key is by moving the entire
349 * queue pair to error state. We could possibly
350 * recover from that, but right now we drop the
351 * connection.
352 * Therefore, we never retransmit messages with RDMA ops.
353 */
354 if (test_bit(RDS_MSG_FLUSH, &rm->m_flags) ||
355 (rm->rdma.op_active &&
356 test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags))) {
357 spin_lock_irqsave(&cp->cp_lock, flags);
358 if (test_and_clear_bit(RDS_MSG_ON_CONN,
359 &rm->m_flags)) {
360 /* our ref is put after the batch */
361 list_move(&rm->m_conn_item,
362 &to_be_dropped);
363 spin_unlock_irqrestore(&cp->cp_lock,
364 flags);
365 } else {
366 /* already off the conn list; drop
367 * the ref taken above ourselves
368 */
369 spin_unlock_irqrestore(&cp->cp_lock,
370 flags);
371 rds_message_put(rm);
372 }
373 continue;
374 }
375
376 /* Require an ACK every once in a while */
377 len = ntohl(rm->m_inc.i_hdr.h_len);
378 if (cp->cp_unacked_packets == 0 ||
379 cp->cp_unacked_bytes < len) {
380 set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
381
382 cp->cp_unacked_packets =
383 rds_sysctl_max_unacked_packets;
384 cp->cp_unacked_bytes =
385 rds_sysctl_max_unacked_bytes;
386 rds_stats_inc(s_send_ack_required);
387 } else {
388 cp->cp_unacked_bytes -= len;
389 cp->cp_unacked_packets--;
390 }
391
392 cp->cp_xmit_rm = rm;
393 }
394
395 /* The transport either sends the whole rdma or none of it */
396 if (rm->rdma.op_active && !cp->cp_xmit_rdma_sent) {
397 rm->m_final_op = &rm->rdma;
398 /* The transport owns the mapped memory for now.
399 * You can't unmap it while it's on the send queue
400 */
401 set_bit(RDS_MSG_MAPPED, &rm->m_flags);
402 ret = conn->c_trans->xmit_rdma(conn, &rm->rdma);
403 if (ret) {
404 clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
405 wake_up_interruptible(&rm->m_flush_wait);
406 break;
407 }
408 cp->cp_xmit_rdma_sent = 1;
409
410 }
411
412 if (rm->atomic.op_active && !cp->cp_xmit_atomic_sent) {
413 rm->m_final_op = &rm->atomic;
414 /* The transport owns the mapped memory for now.
415 * You can't unmap it while it's on the send queue
416 */
417 set_bit(RDS_MSG_MAPPED, &rm->m_flags);
418 ret = conn->c_trans->xmit_atomic(conn, &rm->atomic);
419 if (ret) {
420 clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
421 wake_up_interruptible(&rm->m_flush_wait);
422 break;
423 }
424 cp->cp_xmit_atomic_sent = 1;
425
426 }
427
428 /*
429 * A number of cases require an RDS header to be sent
430 * even if there is no data.
431 * We permit 0-byte sends; rds-ping depends on this.
432 * However, if there are exclusively attached silent ops,
433 * we skip the hdr/data send, to enable silent operation.
434 */
435 if (rm->data.op_nents == 0) {
436 int ops_present;
437 int all_ops_are_silent = 1;
438
439 ops_present = (rm->atomic.op_active || rm->rdma.op_active);
440 if (rm->atomic.op_active && !rm->atomic.op_silent)
441 all_ops_are_silent = 0;
442 if (rm->rdma.op_active && !rm->rdma.op_silent)
443 all_ops_are_silent = 0;
444
445 if (ops_present && all_ops_are_silent
446 && !rm->m_rdma_cookie)
447 rm->data.op_active = 0;
448 }
449
450 if (rm->data.op_active && !cp->cp_xmit_data_sent) {
451 rm->m_final_op = &rm->data;
452
453 ret = conn->c_trans->xmit(conn, rm,
454 cp->cp_xmit_hdr_off,
455 cp->cp_xmit_sg,
456 cp->cp_xmit_data_off);
457 if (ret <= 0)
458 break;
459
460 if (cp->cp_xmit_hdr_off < sizeof(struct rds_header)) {
461 tmp = min_t(int, ret,
462 sizeof(struct rds_header) -
463 cp->cp_xmit_hdr_off);
464 cp->cp_xmit_hdr_off += tmp;
465 ret -= tmp;
466 }
467
468 sg = &rm->data.op_sg[cp->cp_xmit_sg];
469 while (ret) {
470 tmp = min_t(int, ret, sg->length -
471 cp->cp_xmit_data_off);
472 cp->cp_xmit_data_off += tmp;
473 ret -= tmp;
474 if (cp->cp_xmit_data_off == sg->length) {
475 cp->cp_xmit_data_off = 0;
476 sg++;
477 cp->cp_xmit_sg++;
478 BUG_ON(ret != 0 && cp->cp_xmit_sg ==
479 rm->data.op_nents);
480 }
481 }
482
483 if (cp->cp_xmit_hdr_off == sizeof(struct rds_header) &&
484 (cp->cp_xmit_sg == rm->data.op_nents))
485 cp->cp_xmit_data_sent = 1;
486 }
487
488 /*
489 * A rm will only take multiple times through this loop
490 * if there is a data op. Thus, if the data is sent (or there was
491 * none), then we're done with the rm.
492 */
493 if (!rm->data.op_active || cp->cp_xmit_data_sent) {
494 cp->cp_xmit_rm = NULL;
495 cp->cp_xmit_sg = 0;
496 cp->cp_xmit_hdr_off = 0;
497 cp->cp_xmit_data_off = 0;
498 cp->cp_xmit_rdma_sent = 0;
499 cp->cp_xmit_atomic_sent = 0;
500 cp->cp_xmit_data_sent = 0;
501
502 rds_message_put(rm);
503 }
504 }
505
506over_batch:
507 if (conn->c_trans->xmit_path_complete)
508 conn->c_trans->xmit_path_complete(cp);
509 release_in_xmit(cp);
510
511 /* Nuke any messages we decided not to retransmit. */
512 if (!list_empty(&to_be_dropped)) {
513 /* irqs on here, so we can put(), unlike above */
514 list_for_each_entry(rm, &to_be_dropped, m_conn_item)
515 rds_message_put(rm);
516 rds_send_remove_from_sock(&to_be_dropped, RDS_RDMA_DROPPED);
517 }
518
519 /*
520 * Other senders can queue a message after we last test the send queue
521 * but before we clear RDS_IN_XMIT. In that case they'd back off and
522 * not try and send their newly queued message. We need to check the
523 * send queue after having cleared RDS_IN_XMIT so that their message
524 * doesn't get stuck on the send queue.
525 *
526 * If the transport cannot continue (i.e ret != 0), then it must
527 * call us when more room is available, such as from the tx
528 * completion handler.
529 *
530 * We have an extra generation check here so that if someone manages
531 * to jump in after our release_in_xmit, we'll see that they have done
532 * some work and we will skip our goto
533 */
534 if (ret == 0) {
535 bool raced;
536
537 smp_mb();
538 raced = send_gen != READ_ONCE(cp->cp_send_gen);
539
540 if ((test_bit(0, &conn->c_map_queued) ||
541 !list_empty(&cp->cp_send_queue)) && !raced) {
542 if (batch_count < send_batch_count)
543 goto restart;
544 rcu_read_lock();
545 if (rds_destroy_pending(cp->cp_conn))
546 ret = -ENETUNREACH;
547 else
548 queue_delayed_work(cp->cp_wq,
549 &cp->cp_send_w, 1);
550 rcu_read_unlock();
551 } else if (raced) {
552 rds_stats_inc(s_send_lock_queue_raced);
553 }
554 }
555out:
556 return ret;
557}
558EXPORT_SYMBOL_GPL(rds_send_xmit);
559
560static void rds_send_sndbuf_remove(struct rds_sock *rs, struct rds_message *rm)
561{
562 u32 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
563
564 assert_spin_locked(&rs->rs_lock);
565
566 BUG_ON(rs->rs_snd_bytes < len);
567 rs->rs_snd_bytes -= len;
568
569 if (rs->rs_snd_bytes == 0)
570 rds_stats_inc(s_send_queue_empty);
571}
572
573static inline int rds_send_is_acked(struct rds_message *rm, u64 ack,
574 is_acked_func is_acked)
575{
576 if (is_acked)
577 return is_acked(rm, ack);
578 return be64_to_cpu(rm->m_inc.i_hdr.h_sequence) <= ack;
579}
580
581/*
582 * This is pretty similar to what happens below in the ACK
583 * handling code - except that we call here as soon as we get
584 * the IB send completion on the RDMA op and the accompanying
585 * message.
586 */
587void rds_rdma_send_complete(struct rds_message *rm, int status)
588{
589 struct rds_sock *rs = NULL;
590 struct rm_rdma_op *ro;
591 struct rds_notifier *notifier;
592 unsigned long flags;
593
594 spin_lock_irqsave(&rm->m_rs_lock, flags);
595
596 ro = &rm->rdma;
597 if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags) &&
598 ro->op_active && ro->op_notify && ro->op_notifier) {
599 notifier = ro->op_notifier;
600 rs = rm->m_rs;
601 sock_hold(rds_rs_to_sk(rs));
602
603 notifier->n_status = status;
604 spin_lock(&rs->rs_lock);
605 list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
606 spin_unlock(&rs->rs_lock);
607
608 ro->op_notifier = NULL;
609 }
610
611 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
612
613 if (rs) {
614 rds_wake_sk_sleep(rs);
615 sock_put(rds_rs_to_sk(rs));
616 }
617}
618EXPORT_SYMBOL_GPL(rds_rdma_send_complete);
619
620/*
621 * Just like above, except looks at atomic op
622 */
623void rds_atomic_send_complete(struct rds_message *rm, int status)
624{
625 struct rds_sock *rs = NULL;
626 struct rm_atomic_op *ao;
627 struct rds_notifier *notifier;
628 unsigned long flags;
629
630 spin_lock_irqsave(&rm->m_rs_lock, flags);
631
632 ao = &rm->atomic;
633 if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags)
634 && ao->op_active && ao->op_notify && ao->op_notifier) {
635 notifier = ao->op_notifier;
636 rs = rm->m_rs;
637 sock_hold(rds_rs_to_sk(rs));
638
639 notifier->n_status = status;
640 spin_lock(&rs->rs_lock);
641 list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
642 spin_unlock(&rs->rs_lock);
643
644 ao->op_notifier = NULL;
645 }
646
647 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
648
649 if (rs) {
650 rds_wake_sk_sleep(rs);
651 sock_put(rds_rs_to_sk(rs));
652 }
653}
654EXPORT_SYMBOL_GPL(rds_atomic_send_complete);
655
656/*
657 * This is the same as rds_rdma_send_complete except we
658 * don't do any locking - we have all the ingredients (message,
659 * socket, socket lock) and can just move the notifier.
660 */
661static inline void
662__rds_send_complete(struct rds_sock *rs, struct rds_message *rm, int status)
663{
664 struct rm_rdma_op *ro;
665 struct rm_atomic_op *ao;
666
667 ro = &rm->rdma;
668 if (ro->op_active && ro->op_notify && ro->op_notifier) {
669 ro->op_notifier->n_status = status;
670 list_add_tail(&ro->op_notifier->n_list, &rs->rs_notify_queue);
671 ro->op_notifier = NULL;
672 }
673
674 ao = &rm->atomic;
675 if (ao->op_active && ao->op_notify && ao->op_notifier) {
676 ao->op_notifier->n_status = status;
677 list_add_tail(&ao->op_notifier->n_list, &rs->rs_notify_queue);
678 ao->op_notifier = NULL;
679 }
680
681 /* No need to wake the app - caller does this */
682}
683
684/*
685 * This removes messages from the socket's list if they're on it. The list
686 * argument must be private to the caller, we must be able to modify it
687 * without locks. The messages must have a reference held for their
688 * position on the list. This function will drop that reference after
689 * removing the messages from the 'messages' list regardless of if it found
690 * the messages on the socket list or not.
691 */
692static void rds_send_remove_from_sock(struct list_head *messages, int status)
693{
694 unsigned long flags;
695 struct rds_sock *rs = NULL;
696 struct rds_message *rm;
697
698 while (!list_empty(messages)) {
699 int was_on_sock = 0;
700
701 rm = list_entry(messages->next, struct rds_message,
702 m_conn_item);
703 list_del_init(&rm->m_conn_item);
704
705 /*
706 * If we see this flag cleared then we're *sure* that someone
707 * else beat us to removing it from the sock. If we race
708 * with their flag update we'll get the lock and then really
709 * see that the flag has been cleared.
710 *
711 * The message spinlock makes sure nobody clears rm->m_rs
712 * while we're messing with it. It does not prevent the
713 * message from being removed from the socket, though.
714 */
715 spin_lock_irqsave(&rm->m_rs_lock, flags);
716 if (!test_bit(RDS_MSG_ON_SOCK, &rm->m_flags))
717 goto unlock_and_drop;
718
719 if (rs != rm->m_rs) {
720 if (rs) {
721 rds_wake_sk_sleep(rs);
722 sock_put(rds_rs_to_sk(rs));
723 }
724 rs = rm->m_rs;
725 if (rs)
726 sock_hold(rds_rs_to_sk(rs));
727 }
728 if (!rs)
729 goto unlock_and_drop;
730 spin_lock(&rs->rs_lock);
731
732 if (test_and_clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags)) {
733 struct rm_rdma_op *ro = &rm->rdma;
734 struct rds_notifier *notifier;
735
736 list_del_init(&rm->m_sock_item);
737 rds_send_sndbuf_remove(rs, rm);
738
739 if (ro->op_active && ro->op_notifier &&
740 (ro->op_notify || (ro->op_recverr && status))) {
741 notifier = ro->op_notifier;
742 list_add_tail(&notifier->n_list,
743 &rs->rs_notify_queue);
744 if (!notifier->n_status)
745 notifier->n_status = status;
746 rm->rdma.op_notifier = NULL;
747 }
748 was_on_sock = 1;
749 }
750 spin_unlock(&rs->rs_lock);
751
752unlock_and_drop:
753 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
754 rds_message_put(rm);
755 if (was_on_sock)
756 rds_message_put(rm);
757 }
758
759 if (rs) {
760 rds_wake_sk_sleep(rs);
761 sock_put(rds_rs_to_sk(rs));
762 }
763}
764
765/*
766 * Transports call here when they've determined that the receiver queued
767 * messages up to, and including, the given sequence number. Messages are
768 * moved to the retrans queue when rds_send_xmit picks them off the send
769 * queue. This means that in the TCP case, the message may not have been
770 * assigned the m_ack_seq yet - but that's fine as long as tcp_is_acked
771 * checks the RDS_MSG_HAS_ACK_SEQ bit.
772 */
773void rds_send_path_drop_acked(struct rds_conn_path *cp, u64 ack,
774 is_acked_func is_acked)
775{
776 struct rds_message *rm, *tmp;
777 unsigned long flags;
778 LIST_HEAD(list);
779
780 spin_lock_irqsave(&cp->cp_lock, flags);
781
782 list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
783 if (!rds_send_is_acked(rm, ack, is_acked))
784 break;
785
786 list_move(&rm->m_conn_item, &list);
787 clear_bit(RDS_MSG_ON_CONN, &rm->m_flags);
788 }
789
790 /* order flag updates with spin locks */
791 if (!list_empty(&list))
792 smp_mb__after_atomic();
793
794 spin_unlock_irqrestore(&cp->cp_lock, flags);
795
796 /* now remove the messages from the sock list as needed */
797 rds_send_remove_from_sock(&list, RDS_RDMA_SUCCESS);
798}
799EXPORT_SYMBOL_GPL(rds_send_path_drop_acked);
800
801void rds_send_drop_acked(struct rds_connection *conn, u64 ack,
802 is_acked_func is_acked)
803{
804 WARN_ON(conn->c_trans->t_mp_capable);
805 rds_send_path_drop_acked(&conn->c_path[0], ack, is_acked);
806}
807EXPORT_SYMBOL_GPL(rds_send_drop_acked);
808
809void rds_send_drop_to(struct rds_sock *rs, struct sockaddr_in6 *dest)
810{
811 struct rds_message *rm, *tmp;
812 struct rds_connection *conn;
813 struct rds_conn_path *cp;
814 unsigned long flags;
815 LIST_HEAD(list);
816
817 /* get all the messages we're dropping under the rs lock */
818 spin_lock_irqsave(&rs->rs_lock, flags);
819
820 list_for_each_entry_safe(rm, tmp, &rs->rs_send_queue, m_sock_item) {
821 if (dest &&
822 (!ipv6_addr_equal(&dest->sin6_addr, &rm->m_daddr) ||
823 dest->sin6_port != rm->m_inc.i_hdr.h_dport))
824 continue;
825
826 list_move(&rm->m_sock_item, &list);
827 rds_send_sndbuf_remove(rs, rm);
828 clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
829 }
830
831 /* order flag updates with the rs lock */
832 smp_mb__after_atomic();
833
834 spin_unlock_irqrestore(&rs->rs_lock, flags);
835
836 if (list_empty(&list))
837 return;
838
839 /* Remove the messages from the conn */
840 list_for_each_entry(rm, &list, m_sock_item) {
841
842 conn = rm->m_inc.i_conn;
843 if (conn->c_trans->t_mp_capable)
844 cp = rm->m_inc.i_conn_path;
845 else
846 cp = &conn->c_path[0];
847
848 spin_lock_irqsave(&cp->cp_lock, flags);
849 /*
850 * Maybe someone else beat us to removing rm from the conn.
851 * If we race with their flag update we'll get the lock and
852 * then really see that the flag has been cleared.
853 */
854 if (!test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags)) {
855 spin_unlock_irqrestore(&cp->cp_lock, flags);
856 continue;
857 }
858 list_del_init(&rm->m_conn_item);
859 spin_unlock_irqrestore(&cp->cp_lock, flags);
860
861 /*
862 * Couldn't grab m_rs_lock in top loop (lock ordering),
863 * but we can now.
864 */
865 spin_lock_irqsave(&rm->m_rs_lock, flags);
866
867 spin_lock(&rs->rs_lock);
868 __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
869 spin_unlock(&rs->rs_lock);
870
871 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
872
873 rds_message_put(rm);
874 }
875
876 rds_wake_sk_sleep(rs);
877
878 while (!list_empty(&list)) {
879 rm = list_entry(list.next, struct rds_message, m_sock_item);
880 list_del_init(&rm->m_sock_item);
881 rds_message_wait(rm);
882
883 /* just in case the code above skipped this message
884 * because RDS_MSG_ON_CONN wasn't set, run it again here
885 * taking m_rs_lock is the only thing that keeps us
886 * from racing with ack processing.
887 */
888 spin_lock_irqsave(&rm->m_rs_lock, flags);
889
890 spin_lock(&rs->rs_lock);
891 __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
892 spin_unlock(&rs->rs_lock);
893
894 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
895
896 rds_message_put(rm);
897 }
898}
899
900/*
901 * we only want this to fire once so we use the callers 'queued'. It's
902 * possible that another thread can race with us and remove the
903 * message from the flow with RDS_CANCEL_SENT_TO.
904 */
905static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
906 struct rds_conn_path *cp,
907 struct rds_message *rm, __be16 sport,
908 __be16 dport, int *queued)
909{
910 unsigned long flags;
911 u32 len;
912
913 if (*queued)
914 goto out;
915
916 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
917
918 /* this is the only place which holds both the socket's rs_lock
919 * and the connection's c_lock */
920 spin_lock_irqsave(&rs->rs_lock, flags);
921
922 /*
923 * If there is a little space in sndbuf, we don't queue anything,
924 * and userspace gets -EAGAIN. But poll() indicates there's send
925 * room. This can lead to bad behavior (spinning) if snd_bytes isn't
926 * freed up by incoming acks. So we check the *old* value of
927 * rs_snd_bytes here to allow the last msg to exceed the buffer,
928 * and poll() now knows no more data can be sent.
929 */
930 if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
931 rs->rs_snd_bytes += len;
932
933 /* let recv side know we are close to send space exhaustion.
934 * This is probably not the optimal way to do it, as this
935 * means we set the flag on *all* messages as soon as our
936 * throughput hits a certain threshold.
937 */
938 if (rs->rs_snd_bytes >= rds_sk_sndbuf(rs) / 2)
939 set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
940
941 list_add_tail(&rm->m_sock_item, &rs->rs_send_queue);
942 set_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
943 rds_message_addref(rm);
944 sock_hold(rds_rs_to_sk(rs));
945 rm->m_rs = rs;
946
947 /* The code ordering is a little weird, but we're
948 trying to minimize the time we hold c_lock */
949 rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
950 rm->m_inc.i_conn = conn;
951 rm->m_inc.i_conn_path = cp;
952 rds_message_addref(rm);
953
954 spin_lock(&cp->cp_lock);
955 rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
956 list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
957 set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
958 spin_unlock(&cp->cp_lock);
959
960 rdsdebug("queued msg %p len %d, rs %p bytes %d seq %llu\n",
961 rm, len, rs, rs->rs_snd_bytes,
962 (unsigned long long)be64_to_cpu(rm->m_inc.i_hdr.h_sequence));
963
964 *queued = 1;
965 }
966
967 spin_unlock_irqrestore(&rs->rs_lock, flags);
968out:
969 return *queued;
970}
971
972/*
973 * rds_message is getting to be quite complicated, and we'd like to allocate
974 * it all in one go. This figures out how big it needs to be up front.
975 */
976static int rds_rm_size(struct msghdr *msg, int num_sgs,
977 struct rds_iov_vector_arr *vct)
978{
979 struct cmsghdr *cmsg;
980 int size = 0;
981 int cmsg_groups = 0;
982 int retval;
983 bool zcopy_cookie = false;
984 struct rds_iov_vector *iov, *tmp_iov;
985
986 if (num_sgs < 0)
987 return -EINVAL;
988
989 for_each_cmsghdr(cmsg, msg) {
990 if (!CMSG_OK(msg, cmsg))
991 return -EINVAL;
992
993 if (cmsg->cmsg_level != SOL_RDS)
994 continue;
995
996 switch (cmsg->cmsg_type) {
997 case RDS_CMSG_RDMA_ARGS:
998 if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_rdma_args)))
999 return -EINVAL;
1000 if (vct->indx >= vct->len) {
1001 vct->len += vct->incr;
1002 tmp_iov = krealloc_array(vct->vec, vct->len,
1003 sizeof(*vct->vec), GFP_KERNEL);
1004 if (!tmp_iov) {
1005 vct->len -= vct->incr;
1006 return -ENOMEM;
1007 }
1008 vct->vec = tmp_iov;
1009 }
1010 iov = &vct->vec[vct->indx];
1011 memset(iov, 0, sizeof(struct rds_iov_vector));
1012 vct->indx++;
1013 cmsg_groups |= 1;
1014 retval = rds_rdma_extra_size(CMSG_DATA(cmsg), iov);
1015 if (retval < 0)
1016 return retval;
1017 size += retval;
1018
1019 break;
1020
1021 case RDS_CMSG_ZCOPY_COOKIE:
1022 zcopy_cookie = true;
1023 fallthrough;
1024
1025 case RDS_CMSG_RDMA_DEST:
1026 case RDS_CMSG_RDMA_MAP:
1027 cmsg_groups |= 2;
1028 /* these are valid but do no add any size */
1029 break;
1030
1031 case RDS_CMSG_ATOMIC_CSWP:
1032 case RDS_CMSG_ATOMIC_FADD:
1033 case RDS_CMSG_MASKED_ATOMIC_CSWP:
1034 case RDS_CMSG_MASKED_ATOMIC_FADD:
1035 cmsg_groups |= 1;
1036 size += sizeof(struct scatterlist);
1037 break;
1038
1039 default:
1040 return -EINVAL;
1041 }
1042
1043 }
1044
1045 if ((msg->msg_flags & MSG_ZEROCOPY) && !zcopy_cookie)
1046 return -EINVAL;
1047
1048 size += num_sgs * sizeof(struct scatterlist);
1049
1050 /* Ensure (DEST, MAP) are never used with (ARGS, ATOMIC) */
1051 if (cmsg_groups == 3)
1052 return -EINVAL;
1053
1054 return size;
1055}
1056
1057static int rds_cmsg_zcopy(struct rds_sock *rs, struct rds_message *rm,
1058 struct cmsghdr *cmsg)
1059{
1060 u32 *cookie;
1061
1062 if (cmsg->cmsg_len < CMSG_LEN(sizeof(*cookie)) ||
1063 !rm->data.op_mmp_znotifier)
1064 return -EINVAL;
1065 cookie = CMSG_DATA(cmsg);
1066 rm->data.op_mmp_znotifier->z_cookie = *cookie;
1067 return 0;
1068}
1069
1070static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
1071 struct msghdr *msg, int *allocated_mr,
1072 struct rds_iov_vector_arr *vct)
1073{
1074 struct cmsghdr *cmsg;
1075 int ret = 0, ind = 0;
1076
1077 for_each_cmsghdr(cmsg, msg) {
1078 if (!CMSG_OK(msg, cmsg))
1079 return -EINVAL;
1080
1081 if (cmsg->cmsg_level != SOL_RDS)
1082 continue;
1083
1084 /* As a side effect, RDMA_DEST and RDMA_MAP will set
1085 * rm->rdma.m_rdma_cookie and rm->rdma.m_rdma_mr.
1086 */
1087 switch (cmsg->cmsg_type) {
1088 case RDS_CMSG_RDMA_ARGS:
1089 if (ind >= vct->indx)
1090 return -ENOMEM;
1091 ret = rds_cmsg_rdma_args(rs, rm, cmsg, &vct->vec[ind]);
1092 ind++;
1093 break;
1094
1095 case RDS_CMSG_RDMA_DEST:
1096 ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
1097 break;
1098
1099 case RDS_CMSG_RDMA_MAP:
1100 ret = rds_cmsg_rdma_map(rs, rm, cmsg);
1101 if (!ret)
1102 *allocated_mr = 1;
1103 else if (ret == -ENODEV)
1104 /* Accommodate the get_mr() case which can fail
1105 * if connection isn't established yet.
1106 */
1107 ret = -EAGAIN;
1108 break;
1109 case RDS_CMSG_ATOMIC_CSWP:
1110 case RDS_CMSG_ATOMIC_FADD:
1111 case RDS_CMSG_MASKED_ATOMIC_CSWP:
1112 case RDS_CMSG_MASKED_ATOMIC_FADD:
1113 ret = rds_cmsg_atomic(rs, rm, cmsg);
1114 break;
1115
1116 case RDS_CMSG_ZCOPY_COOKIE:
1117 ret = rds_cmsg_zcopy(rs, rm, cmsg);
1118 break;
1119
1120 default:
1121 return -EINVAL;
1122 }
1123
1124 if (ret)
1125 break;
1126 }
1127
1128 return ret;
1129}
1130
1131static int rds_rdma_bytes(struct msghdr *msg, size_t *rdma_bytes)
1132{
1133 struct rds_rdma_args *args;
1134 struct cmsghdr *cmsg;
1135
1136 for_each_cmsghdr(cmsg, msg) {
1137 if (!CMSG_OK(msg, cmsg))
1138 return -EINVAL;
1139
1140 if (cmsg->cmsg_level != SOL_RDS)
1141 continue;
1142
1143 if (cmsg->cmsg_type == RDS_CMSG_RDMA_ARGS) {
1144 if (cmsg->cmsg_len <
1145 CMSG_LEN(sizeof(struct rds_rdma_args)))
1146 return -EINVAL;
1147 args = CMSG_DATA(cmsg);
1148 *rdma_bytes += args->remote_vec.bytes;
1149 }
1150 }
1151 return 0;
1152}
1153
1154int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
1155{
1156 struct sock *sk = sock->sk;
1157 struct rds_sock *rs = rds_sk_to_rs(sk);
1158 DECLARE_SOCKADDR(struct sockaddr_in6 *, sin6, msg->msg_name);
1159 DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
1160 __be16 dport;
1161 struct rds_message *rm = NULL;
1162 struct rds_connection *conn;
1163 int ret = 0;
1164 int queued = 0, allocated_mr = 0;
1165 int nonblock = msg->msg_flags & MSG_DONTWAIT;
1166 long timeo = sock_sndtimeo(sk, nonblock);
1167 struct rds_conn_path *cpath;
1168 struct in6_addr daddr;
1169 __u32 scope_id = 0;
1170 size_t rdma_payload_len = 0;
1171 bool zcopy = ((msg->msg_flags & MSG_ZEROCOPY) &&
1172 sock_flag(rds_rs_to_sk(rs), SOCK_ZEROCOPY));
1173 int num_sgs = DIV_ROUND_UP(payload_len, PAGE_SIZE);
1174 int namelen;
1175 struct rds_iov_vector_arr vct;
1176 int ind;
1177
1178 memset(&vct, 0, sizeof(vct));
1179
1180 /* expect 1 RDMA CMSG per rds_sendmsg. can still grow if more needed. */
1181 vct.incr = 1;
1182
1183 /* Mirror Linux UDP mirror of BSD error message compatibility */
1184 /* XXX: Perhaps MSG_MORE someday */
1185 if (msg->msg_flags & ~(MSG_DONTWAIT | MSG_CMSG_COMPAT | MSG_ZEROCOPY)) {
1186 ret = -EOPNOTSUPP;
1187 goto out;
1188 }
1189
1190 namelen = msg->msg_namelen;
1191 if (namelen != 0) {
1192 if (namelen < sizeof(*usin)) {
1193 ret = -EINVAL;
1194 goto out;
1195 }
1196 switch (usin->sin_family) {
1197 case AF_INET:
1198 if (usin->sin_addr.s_addr == htonl(INADDR_ANY) ||
1199 usin->sin_addr.s_addr == htonl(INADDR_BROADCAST) ||
1200 ipv4_is_multicast(usin->sin_addr.s_addr)) {
1201 ret = -EINVAL;
1202 goto out;
1203 }
1204 ipv6_addr_set_v4mapped(usin->sin_addr.s_addr, &daddr);
1205 dport = usin->sin_port;
1206 break;
1207
1208#if IS_ENABLED(CONFIG_IPV6)
1209 case AF_INET6: {
1210 int addr_type;
1211
1212 if (namelen < sizeof(*sin6)) {
1213 ret = -EINVAL;
1214 goto out;
1215 }
1216 addr_type = ipv6_addr_type(&sin6->sin6_addr);
1217 if (!(addr_type & IPV6_ADDR_UNICAST)) {
1218 __be32 addr4;
1219
1220 if (!(addr_type & IPV6_ADDR_MAPPED)) {
1221 ret = -EINVAL;
1222 goto out;
1223 }
1224
1225 /* It is a mapped address. Need to do some
1226 * sanity checks.
1227 */
1228 addr4 = sin6->sin6_addr.s6_addr32[3];
1229 if (addr4 == htonl(INADDR_ANY) ||
1230 addr4 == htonl(INADDR_BROADCAST) ||
1231 ipv4_is_multicast(addr4)) {
1232 ret = -EINVAL;
1233 goto out;
1234 }
1235 }
1236 if (addr_type & IPV6_ADDR_LINKLOCAL) {
1237 if (sin6->sin6_scope_id == 0) {
1238 ret = -EINVAL;
1239 goto out;
1240 }
1241 scope_id = sin6->sin6_scope_id;
1242 }
1243
1244 daddr = sin6->sin6_addr;
1245 dport = sin6->sin6_port;
1246 break;
1247 }
1248#endif
1249
1250 default:
1251 ret = -EINVAL;
1252 goto out;
1253 }
1254 } else {
1255 /* We only care about consistency with ->connect() */
1256 lock_sock(sk);
1257 daddr = rs->rs_conn_addr;
1258 dport = rs->rs_conn_port;
1259 scope_id = rs->rs_bound_scope_id;
1260 release_sock(sk);
1261 }
1262
1263 lock_sock(sk);
1264 if (ipv6_addr_any(&rs->rs_bound_addr) || ipv6_addr_any(&daddr)) {
1265 release_sock(sk);
1266 ret = -ENOTCONN;
1267 goto out;
1268 } else if (namelen != 0) {
1269 /* Cannot send to an IPv4 address using an IPv6 source
1270 * address and cannot send to an IPv6 address using an
1271 * IPv4 source address.
1272 */
1273 if (ipv6_addr_v4mapped(&daddr) ^
1274 ipv6_addr_v4mapped(&rs->rs_bound_addr)) {
1275 release_sock(sk);
1276 ret = -EOPNOTSUPP;
1277 goto out;
1278 }
1279 /* If the socket is already bound to a link local address,
1280 * it can only send to peers on the same link. But allow
1281 * communicating between link local and non-link local address.
1282 */
1283 if (scope_id != rs->rs_bound_scope_id) {
1284 if (!scope_id) {
1285 scope_id = rs->rs_bound_scope_id;
1286 } else if (rs->rs_bound_scope_id) {
1287 release_sock(sk);
1288 ret = -EINVAL;
1289 goto out;
1290 }
1291 }
1292 }
1293 release_sock(sk);
1294
1295 ret = rds_rdma_bytes(msg, &rdma_payload_len);
1296 if (ret)
1297 goto out;
1298
1299 if (max_t(size_t, payload_len, rdma_payload_len) > RDS_MAX_MSG_SIZE) {
1300 ret = -EMSGSIZE;
1301 goto out;
1302 }
1303
1304 if (payload_len > rds_sk_sndbuf(rs)) {
1305 ret = -EMSGSIZE;
1306 goto out;
1307 }
1308
1309 if (zcopy) {
1310 if (rs->rs_transport->t_type != RDS_TRANS_TCP) {
1311 ret = -EOPNOTSUPP;
1312 goto out;
1313 }
1314 num_sgs = iov_iter_npages(&msg->msg_iter, INT_MAX);
1315 }
1316 /* size of rm including all sgs */
1317 ret = rds_rm_size(msg, num_sgs, &vct);
1318 if (ret < 0)
1319 goto out;
1320
1321 rm = rds_message_alloc(ret, GFP_KERNEL);
1322 if (!rm) {
1323 ret = -ENOMEM;
1324 goto out;
1325 }
1326
1327 /* Attach data to the rm */
1328 if (payload_len) {
1329 rm->data.op_sg = rds_message_alloc_sgs(rm, num_sgs);
1330 if (IS_ERR(rm->data.op_sg)) {
1331 ret = PTR_ERR(rm->data.op_sg);
1332 goto out;
1333 }
1334 ret = rds_message_copy_from_user(rm, &msg->msg_iter, zcopy);
1335 if (ret)
1336 goto out;
1337 }
1338 rm->data.op_active = 1;
1339
1340 rm->m_daddr = daddr;
1341
1342 /* rds_conn_create has a spinlock that runs with IRQ off.
1343 * Caching the conn in the socket helps a lot. */
1344 if (rs->rs_conn && ipv6_addr_equal(&rs->rs_conn->c_faddr, &daddr) &&
1345 rs->rs_tos == rs->rs_conn->c_tos) {
1346 conn = rs->rs_conn;
1347 } else {
1348 conn = rds_conn_create_outgoing(sock_net(sock->sk),
1349 &rs->rs_bound_addr, &daddr,
1350 rs->rs_transport, rs->rs_tos,
1351 sock->sk->sk_allocation,
1352 scope_id);
1353 if (IS_ERR(conn)) {
1354 ret = PTR_ERR(conn);
1355 goto out;
1356 }
1357 rs->rs_conn = conn;
1358 }
1359
1360 if (conn->c_trans->t_mp_capable) {
1361 /* Use c_path[0] until we learn that
1362 * the peer supports more (c_npaths > 1)
1363 */
1364 cpath = &conn->c_path[RDS_MPATH_HASH(rs, conn->c_npaths ? : 1)];
1365 } else {
1366 cpath = &conn->c_path[0];
1367 }
1368
1369 /* If we're multipath capable and path 0 is down, queue reconnect
1370 * and send a ping. This initiates the multipath handshake through
1371 * rds_send_probe(), which sends RDS_EXTHDR_NPATHS to the peer,
1372 * starting multipath capability negotiation.
1373 */
1374 if (conn->c_trans->t_mp_capable &&
1375 !rds_conn_path_up(&conn->c_path[0])) {
1376 /* Ensures that only one request is queued. And
1377 * rds_send_ping() ensures that only one ping is
1378 * outstanding.
1379 */
1380 if (!test_and_set_bit(RDS_RECONNECT_PENDING,
1381 &conn->c_path[0].cp_flags))
1382 queue_delayed_work(conn->c_path[0].cp_wq,
1383 &conn->c_path[0].cp_conn_w, 0);
1384 rds_send_ping(conn, 0);
1385 }
1386
1387 rm->m_conn_path = cpath;
1388
1389 /* Parse any control messages the user may have included. */
1390 ret = rds_cmsg_send(rs, rm, msg, &allocated_mr, &vct);
1391 if (ret)
1392 goto out;
1393
1394 if (rm->rdma.op_active && !conn->c_trans->xmit_rdma) {
1395 printk_ratelimited(KERN_NOTICE "rdma_op %p conn xmit_rdma %p\n",
1396 &rm->rdma, conn->c_trans->xmit_rdma);
1397 ret = -EOPNOTSUPP;
1398 goto out;
1399 }
1400
1401 if (rm->atomic.op_active && !conn->c_trans->xmit_atomic) {
1402 printk_ratelimited(KERN_NOTICE "atomic_op %p conn xmit_atomic %p\n",
1403 &rm->atomic, conn->c_trans->xmit_atomic);
1404 ret = -EOPNOTSUPP;
1405 goto out;
1406 }
1407
1408 if (rds_destroy_pending(conn)) {
1409 ret = -EAGAIN;
1410 goto out;
1411 }
1412
1413 if (rds_conn_path_down(cpath))
1414 rds_check_all_paths(conn);
1415
1416 ret = rds_cong_wait(conn->c_fcong, dport, nonblock, rs);
1417 if (ret) {
1418 WRITE_ONCE(rs->rs_seen_congestion, 1);
1419 goto out;
1420 }
1421 while (!rds_send_queue_rm(rs, conn, cpath, rm, rs->rs_bound_port,
1422 dport, &queued)) {
1423 rds_stats_inc(s_send_queue_full);
1424
1425 if (nonblock) {
1426 ret = -EAGAIN;
1427 goto out;
1428 }
1429
1430 timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
1431 rds_send_queue_rm(rs, conn, cpath, rm,
1432 rs->rs_bound_port,
1433 dport,
1434 &queued),
1435 timeo);
1436 rdsdebug("sendmsg woke queued %d timeo %ld\n", queued, timeo);
1437 if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
1438 continue;
1439
1440 ret = timeo;
1441 if (ret == 0)
1442 ret = -ETIMEDOUT;
1443 goto out;
1444 }
1445
1446 /*
1447 * By now we've committed to the send. We reuse rds_send_worker()
1448 * to retry sends in the rds thread if the transport asks us to.
1449 */
1450 rds_stats_inc(s_send_queued);
1451
1452 ret = rds_send_xmit(cpath);
1453 if (ret == -ENOMEM || ret == -EAGAIN) {
1454 ret = 0;
1455 rcu_read_lock();
1456 if (rds_destroy_pending(cpath->cp_conn))
1457 ret = -ENETUNREACH;
1458 else
1459 queue_delayed_work(cpath->cp_wq, &cpath->cp_send_w, 1);
1460 rcu_read_unlock();
1461
1462 if (ret)
1463 goto out;
1464 }
1465
1466 rds_message_put(rm);
1467
1468 for (ind = 0; ind < vct.indx; ind++)
1469 kfree(vct.vec[ind].iov);
1470 kfree(vct.vec);
1471
1472 return payload_len;
1473
1474out:
1475 for (ind = 0; ind < vct.indx; ind++)
1476 kfree(vct.vec[ind].iov);
1477 kfree(vct.vec);
1478
1479 /* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
1480 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
1481 * or in any other way, we need to destroy the MR again */
1482 if (allocated_mr)
1483 rds_rdma_unuse(rs, rds_rdma_cookie_key(rm->m_rdma_cookie), 1);
1484
1485 if (rm)
1486 rds_message_put(rm);
1487 return ret;
1488}
1489
1490/*
1491 * send out a probe. Can be shared by rds_send_ping,
1492 * rds_send_pong, rds_send_hb.
1493 * rds_send_hb should use h_flags
1494 * RDS_FLAG_HB_PING|RDS_FLAG_ACK_REQUIRED
1495 * or
1496 * RDS_FLAG_HB_PONG|RDS_FLAG_ACK_REQUIRED
1497 */
1498static int
1499rds_send_probe(struct rds_conn_path *cp, __be16 sport,
1500 __be16 dport, u8 h_flags)
1501{
1502 struct rds_message *rm;
1503 unsigned long flags;
1504 int ret = 0;
1505
1506 rm = rds_message_alloc(0, GFP_ATOMIC);
1507 if (!rm) {
1508 ret = -ENOMEM;
1509 goto out;
1510 }
1511
1512 rm->m_daddr = cp->cp_conn->c_faddr;
1513 rm->data.op_active = 1;
1514
1515 rds_conn_path_connect_if_down(cp);
1516
1517 ret = rds_cong_wait(cp->cp_conn->c_fcong, dport, 1, NULL);
1518 if (ret)
1519 goto out;
1520
1521 spin_lock_irqsave(&cp->cp_lock, flags);
1522 list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
1523 set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
1524 rds_message_addref(rm);
1525 rm->m_inc.i_conn = cp->cp_conn;
1526 rm->m_inc.i_conn_path = cp;
1527
1528 rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport,
1529 cp->cp_next_tx_seq);
1530 rm->m_inc.i_hdr.h_flags |= h_flags;
1531 cp->cp_next_tx_seq++;
1532
1533 if (RDS_HS_PROBE(be16_to_cpu(sport), be16_to_cpu(dport)) &&
1534 cp->cp_conn->c_trans->t_mp_capable) {
1535 __be16 npaths = cpu_to_be16(RDS_MPATH_WORKERS);
1536 __be32 my_gen_num = cpu_to_be32(cp->cp_conn->c_my_gen_num);
1537 u8 dummy = 0;
1538
1539 rds_message_add_extension(&rm->m_inc.i_hdr,
1540 RDS_EXTHDR_NPATHS, &npaths);
1541 rds_message_add_extension(&rm->m_inc.i_hdr,
1542 RDS_EXTHDR_GEN_NUM,
1543 &my_gen_num);
1544 rds_message_add_extension(&rm->m_inc.i_hdr,
1545 RDS_EXTHDR_SPORT_IDX,
1546 &dummy);
1547 }
1548 spin_unlock_irqrestore(&cp->cp_lock, flags);
1549
1550 rds_stats_inc(s_send_queued);
1551 rds_stats_inc(s_send_pong);
1552
1553 /* schedule the send work on cp_wq */
1554 rcu_read_lock();
1555 if (!rds_destroy_pending(cp->cp_conn))
1556 queue_delayed_work(cp->cp_wq, &cp->cp_send_w, 1);
1557 rcu_read_unlock();
1558
1559 rds_message_put(rm);
1560 return 0;
1561
1562out:
1563 if (rm)
1564 rds_message_put(rm);
1565 return ret;
1566}
1567
1568int
1569rds_send_pong(struct rds_conn_path *cp, __be16 dport)
1570{
1571 return rds_send_probe(cp, 0, dport, 0);
1572}
1573
1574void
1575rds_send_ping(struct rds_connection *conn, int cp_index)
1576{
1577 unsigned long flags;
1578 struct rds_conn_path *cp = &conn->c_path[cp_index];
1579
1580 spin_lock_irqsave(&cp->cp_lock, flags);
1581 if (conn->c_ping_triggered) {
1582 spin_unlock_irqrestore(&cp->cp_lock, flags);
1583 return;
1584 }
1585 conn->c_ping_triggered = 1;
1586 spin_unlock_irqrestore(&cp->cp_lock, flags);
1587 rds_send_probe(cp, cpu_to_be16(RDS_FLAG_PROBE_PORT), 0, 0);
1588}
1589EXPORT_SYMBOL_GPL(rds_send_ping);