blob: 5cc64039caf71d2378d8f7609ab77331e733c19d [file] [log] [blame]
Andy Grover5c115592009-02-24 15:30:27 +00001/*
2 * Copyright (c) 2006 Oracle. 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>
Paul Gortmakerd9b93842011-09-18 13:21:27 -040034#include <linux/moduleparam.h>
Tejun Heo5a0e3ad2010-03-24 17:04:11 +090035#include <linux/gfp.h>
Andy Grover5c115592009-02-24 15:30:27 +000036#include <net/sock.h>
37#include <linux/in.h>
38#include <linux/list.h>
Manuel Zerpiescb0a6052011-06-16 02:09:57 +000039#include <linux/ratelimit.h>
Paul Gortmakerbc3b2d72011-07-15 11:47:34 -040040#include <linux/export.h>
Santosh Shilimkar4bebdd72015-09-10 11:57:14 -070041#include <linux/sizes.h>
Andy Grover5c115592009-02-24 15:30:27 +000042
43#include "rds.h"
Andy Grover5c115592009-02-24 15:30:27 +000044
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 */
Santosh Shilimkar4bebdd72015-09-10 11:57:14 -070055static int send_batch_count = SZ_1K;
Andy Grover5c115592009-02-24 15:30:27 +000056module_param(send_batch_count, int, 0444);
57MODULE_PARM_DESC(send_batch_count, " batch factor when working the send queue");
58
stephen hemmingerff51bf82010-10-19 08:08:33 +000059static void rds_send_remove_from_sock(struct list_head *messages, int status);
60
Andy Grover5c115592009-02-24 15:30:27 +000061/*
Zach Brown0f4b1c72010-06-04 14:41:41 -070062 * Reset the send state. Callers must ensure that this doesn't race with
63 * rds_send_xmit().
Andy Grover5c115592009-02-24 15:30:27 +000064 */
Sowmini Varadhand769ef82016-06-13 09:44:41 -070065void rds_send_path_reset(struct rds_conn_path *cp)
Andy Grover5c115592009-02-24 15:30:27 +000066{
67 struct rds_message *rm, *tmp;
68 unsigned long flags;
69
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070070 if (cp->cp_xmit_rm) {
71 rm = cp->cp_xmit_rm;
72 cp->cp_xmit_rm = NULL;
Andy Grover5c115592009-02-24 15:30:27 +000073 /* 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 */
Chris Mason7e3f2952010-05-11 15:11:11 -070077 rds_message_unmapped(rm);
Chris Mason7e3f2952010-05-11 15:11:11 -070078 rds_message_put(rm);
Andy Grover5c115592009-02-24 15:30:27 +000079 }
Chris Mason7e3f2952010-05-11 15:11:11 -070080
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070081 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;
Andy Grover5c115592009-02-24 15:30:27 +000087
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070088 cp->cp_conn->c_map_queued = 0;
Andy Grover5c115592009-02-24 15:30:27 +000089
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070090 cp->cp_unacked_packets = rds_sysctl_max_unacked_packets;
91 cp->cp_unacked_bytes = rds_sysctl_max_unacked_bytes;
Andy Grover5c115592009-02-24 15:30:27 +000092
93 /* Mark messages as retransmissions, and move them to the send q */
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070094 spin_lock_irqsave(&cp->cp_lock, flags);
95 list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
Andy Grover5c115592009-02-24 15:30:27 +000096 set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
97 set_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags);
98 }
Sowmini Varadhan4e9b5512016-06-13 09:44:30 -070099 list_splice_init(&cp->cp_retrans, &cp->cp_send_queue);
100 spin_unlock_irqrestore(&cp->cp_lock, flags);
101}
Sowmini Varadhand769ef82016-06-13 09:44:41 -0700102EXPORT_SYMBOL_GPL(rds_send_path_reset);
Andy Grover5c115592009-02-24 15:30:27 +0000103
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700104static int acquire_in_xmit(struct rds_conn_path *cp)
Zach Brown0f4b1c72010-06-04 14:41:41 -0700105{
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700106 return test_and_set_bit(RDS_IN_XMIT, &cp->cp_flags) == 0;
Zach Brown0f4b1c72010-06-04 14:41:41 -0700107}
108
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700109static void release_in_xmit(struct rds_conn_path *cp)
Zach Brown0f4b1c72010-06-04 14:41:41 -0700110{
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700111 clear_bit(RDS_IN_XMIT, &cp->cp_flags);
Peter Zijlstra4e857c52014-03-17 18:06:10 +0100112 smp_mb__after_atomic();
Zach Brown0f4b1c72010-06-04 14:41:41 -0700113 /*
114 * We don't use wait_on_bit()/wake_up_bit() because our waking is in a
115 * hot path and finding waiters is very rare. We don't want to walk
116 * the system-wide hashed waitqueue buckets in the fast path only to
117 * almost never find waiters.
118 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700119 if (waitqueue_active(&cp->cp_waitq))
120 wake_up_all(&cp->cp_waitq);
Zach Brown0f4b1c72010-06-04 14:41:41 -0700121}
122
Andy Grover5c115592009-02-24 15:30:27 +0000123/*
Lucas De Marchi25985ed2011-03-30 22:57:33 -0300124 * We're making the conscious trade-off here to only send one message
Andy Grover5c115592009-02-24 15:30:27 +0000125 * down the connection at a time.
126 * Pro:
127 * - tx queueing is a simple fifo list
128 * - reassembly is optional and easily done by transports per conn
129 * - no per flow rx lookup at all, straight to the socket
130 * - less per-frag memory and wire overhead
131 * Con:
132 * - queued acks can be delayed behind large messages
133 * Depends:
134 * - small message latency is higher behind queued large messages
135 * - large message latency isn't starved by intervening small sends
136 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700137int rds_send_xmit(struct rds_conn_path *cp)
Andy Grover5c115592009-02-24 15:30:27 +0000138{
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700139 struct rds_connection *conn = cp->cp_conn;
Andy Grover5c115592009-02-24 15:30:27 +0000140 struct rds_message *rm;
141 unsigned long flags;
142 unsigned int tmp;
Andy Grover5c115592009-02-24 15:30:27 +0000143 struct scatterlist *sg;
144 int ret = 0;
Andy Grover5c115592009-02-24 15:30:27 +0000145 LIST_HEAD(to_be_dropped);
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400146 int batch_count;
147 unsigned long send_gen = 0;
Andy Grover5c115592009-02-24 15:30:27 +0000148
Andy Groverfcc54502010-03-29 17:08:49 -0700149restart:
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400150 batch_count = 0;
Andy Grover049ee3f2010-03-23 17:39:07 -0700151
Andy Grover5c115592009-02-24 15:30:27 +0000152 /*
153 * sendmsg calls here after having queued its message on the send
154 * queue. We only have one task feeding the connection at a time. If
155 * another thread is already feeding the queue then we back off. This
156 * avoids blocking the caller and trading per-connection data between
157 * caches per message.
Andy Grover5c115592009-02-24 15:30:27 +0000158 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700159 if (!acquire_in_xmit(cp)) {
Andy Grover049ee3f2010-03-23 17:39:07 -0700160 rds_stats_inc(s_send_lock_contention);
Andy Grover5c115592009-02-24 15:30:27 +0000161 ret = -ENOMEM;
162 goto out;
163 }
Zach Brown0f4b1c72010-06-04 14:41:41 -0700164
165 /*
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400166 * we record the send generation after doing the xmit acquire.
167 * if someone else manages to jump in and do some work, we'll use
168 * this to avoid a goto restart farther down.
169 *
170 * The acquire_in_xmit() check above ensures that only one
171 * caller can increment c_send_gen at any time.
172 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700173 cp->cp_send_gen++;
174 send_gen = cp->cp_send_gen;
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400175
176 /*
Zach Brown0f4b1c72010-06-04 14:41:41 -0700177 * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT,
178 * we do the opposite to avoid races.
179 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700180 if (!rds_conn_path_up(cp)) {
181 release_in_xmit(cp);
Zach Brown0f4b1c72010-06-04 14:41:41 -0700182 ret = 0;
183 goto out;
184 }
Andy Grover5c115592009-02-24 15:30:27 +0000185
Sowmini Varadhan226f7a72016-06-30 16:11:10 -0700186 if (conn->c_trans->xmit_path_prepare)
187 conn->c_trans->xmit_path_prepare(cp);
Andy Grover5c115592009-02-24 15:30:27 +0000188
189 /*
190 * spin trying to push headers and data down the connection until
Andy Grover5b2366b2010-02-03 19:36:44 -0800191 * the connection doesn't make forward progress.
Andy Grover5c115592009-02-24 15:30:27 +0000192 */
Andy Groverfcc54502010-03-29 17:08:49 -0700193 while (1) {
Andy Grover5c115592009-02-24 15:30:27 +0000194
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700195 rm = cp->cp_xmit_rm;
Andy Grover5c115592009-02-24 15:30:27 +0000196
Andy Grover5b2366b2010-02-03 19:36:44 -0800197 /*
198 * If between sending messages, we can send a pending congestion
199 * map update.
Andy Grover5c115592009-02-24 15:30:27 +0000200 */
Andy Grover8690bfa2010-01-12 11:56:44 -0800201 if (!rm && test_and_clear_bit(0, &conn->c_map_queued)) {
Andy Grover77dd5502010-03-22 15:22:04 -0700202 rm = rds_cong_update_alloc(conn);
203 if (IS_ERR(rm)) {
204 ret = PTR_ERR(rm);
205 break;
Andy Grover5b2366b2010-02-03 19:36:44 -0800206 }
Andy Grover77dd5502010-03-22 15:22:04 -0700207 rm->data.op_active = 1;
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700208 rm->m_inc.i_conn_path = cp;
209 rm->m_inc.i_conn = cp->cp_conn;
Andy Grover77dd5502010-03-22 15:22:04 -0700210
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700211 cp->cp_xmit_rm = rm;
Andy Grover5c115592009-02-24 15:30:27 +0000212 }
213
214 /*
Andy Grover5b2366b2010-02-03 19:36:44 -0800215 * If not already working on one, grab the next message.
Andy Grover5c115592009-02-24 15:30:27 +0000216 *
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700217 * cp_xmit_rm holds a ref while we're sending this message down
Andy Grover5c115592009-02-24 15:30:27 +0000218 * the connction. We can use this ref while holding the
219 * send_sem.. rds_send_reset() is serialized with it.
220 */
Andy Grover8690bfa2010-01-12 11:56:44 -0800221 if (!rm) {
Andy Grover5c115592009-02-24 15:30:27 +0000222 unsigned int len;
223
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400224 batch_count++;
225
226 /* we want to process as big a batch as we can, but
227 * we also want to avoid softlockups. If we've been
228 * through a lot of messages, lets back off and see
229 * if anyone else jumps in
230 */
Santosh Shilimkar4bebdd72015-09-10 11:57:14 -0700231 if (batch_count >= send_batch_count)
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400232 goto over_batch;
233
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700234 spin_lock_irqsave(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000235
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700236 if (!list_empty(&cp->cp_send_queue)) {
237 rm = list_entry(cp->cp_send_queue.next,
Andy Grover5c115592009-02-24 15:30:27 +0000238 struct rds_message,
239 m_conn_item);
240 rds_message_addref(rm);
241
242 /*
243 * Move the message from the send queue to the retransmit
244 * list right away.
245 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700246 list_move_tail(&rm->m_conn_item,
247 &cp->cp_retrans);
Andy Grover5c115592009-02-24 15:30:27 +0000248 }
249
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700250 spin_unlock_irqrestore(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000251
Andy Groverfcc54502010-03-29 17:08:49 -0700252 if (!rm)
Andy Grover5c115592009-02-24 15:30:27 +0000253 break;
Andy Grover5c115592009-02-24 15:30:27 +0000254
255 /* Unfortunately, the way Infiniband deals with
256 * RDMA to a bad MR key is by moving the entire
257 * queue pair to error state. We cold possibly
258 * recover from that, but right now we drop the
259 * connection.
260 * Therefore, we never retransmit messages with RDMA ops.
261 */
Sowmini Varadhan905dd412016-11-16 13:29:49 -0800262 if (test_bit(RDS_MSG_FLUSH, &rm->m_flags) ||
263 (rm->rdma.op_active &&
264 test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags))) {
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700265 spin_lock_irqsave(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000266 if (test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags))
267 list_move(&rm->m_conn_item, &to_be_dropped);
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700268 spin_unlock_irqrestore(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000269 continue;
270 }
271
272 /* Require an ACK every once in a while */
273 len = ntohl(rm->m_inc.i_hdr.h_len);
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700274 if (cp->cp_unacked_packets == 0 ||
275 cp->cp_unacked_bytes < len) {
Andy Grover5c115592009-02-24 15:30:27 +0000276 __set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
277
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700278 cp->cp_unacked_packets =
279 rds_sysctl_max_unacked_packets;
280 cp->cp_unacked_bytes =
281 rds_sysctl_max_unacked_bytes;
Andy Grover5c115592009-02-24 15:30:27 +0000282 rds_stats_inc(s_send_ack_required);
283 } else {
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700284 cp->cp_unacked_bytes -= len;
285 cp->cp_unacked_packets--;
Andy Grover5c115592009-02-24 15:30:27 +0000286 }
287
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700288 cp->cp_xmit_rm = rm;
Andy Grover5c115592009-02-24 15:30:27 +0000289 }
290
Andy Grover2c3a5f92010-03-01 16:10:40 -0800291 /* The transport either sends the whole rdma or none of it */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700292 if (rm->rdma.op_active && !cp->cp_xmit_rdma_sent) {
Andy Groverff3d7d32010-03-01 14:03:09 -0800293 rm->m_final_op = &rm->rdma;
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700294 /* The transport owns the mapped memory for now.
295 * You can't unmap it while it's on the send queue
296 */
297 set_bit(RDS_MSG_MAPPED, &rm->m_flags);
Andy Grover2c3a5f92010-03-01 16:10:40 -0800298 ret = conn->c_trans->xmit_rdma(conn, &rm->rdma);
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700299 if (ret) {
300 clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
301 wake_up_interruptible(&rm->m_flush_wait);
Andy Grover2c3a5f92010-03-01 16:10:40 -0800302 break;
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700303 }
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700304 cp->cp_xmit_rdma_sent = 1;
Andy Grover2c3a5f92010-03-01 16:10:40 -0800305
Andy Grover2c3a5f92010-03-01 16:10:40 -0800306 }
307
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700308 if (rm->atomic.op_active && !cp->cp_xmit_atomic_sent) {
Andy Groverff3d7d32010-03-01 14:03:09 -0800309 rm->m_final_op = &rm->atomic;
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700310 /* The transport owns the mapped memory for now.
311 * You can't unmap it while it's on the send queue
312 */
313 set_bit(RDS_MSG_MAPPED, &rm->m_flags);
Andy Groverff3d7d32010-03-01 14:03:09 -0800314 ret = conn->c_trans->xmit_atomic(conn, &rm->atomic);
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700315 if (ret) {
316 clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
317 wake_up_interruptible(&rm->m_flush_wait);
Andy Grover15133f62010-01-12 14:33:38 -0800318 break;
santosh.shilimkar@oracle.com4f731132015-08-22 15:45:29 -0700319 }
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700320 cp->cp_xmit_atomic_sent = 1;
Andy Groverff3d7d32010-03-01 14:03:09 -0800321
Andy Grover15133f62010-01-12 14:33:38 -0800322 }
323
Andy Grover2c3a5f92010-03-01 16:10:40 -0800324 /*
325 * A number of cases require an RDS header to be sent
326 * even if there is no data.
327 * We permit 0-byte sends; rds-ping depends on this.
328 * However, if there are exclusively attached silent ops,
329 * we skip the hdr/data send, to enable silent operation.
330 */
331 if (rm->data.op_nents == 0) {
332 int ops_present;
333 int all_ops_are_silent = 1;
Andy Grover241eef32010-01-19 21:25:26 -0800334
Andy Grover2c3a5f92010-03-01 16:10:40 -0800335 ops_present = (rm->atomic.op_active || rm->rdma.op_active);
336 if (rm->atomic.op_active && !rm->atomic.op_silent)
337 all_ops_are_silent = 0;
338 if (rm->rdma.op_active && !rm->rdma.op_silent)
339 all_ops_are_silent = 0;
Andy Grover241eef32010-01-19 21:25:26 -0800340
Andy Grover2c3a5f92010-03-01 16:10:40 -0800341 if (ops_present && all_ops_are_silent
342 && !rm->m_rdma_cookie)
343 rm->data.op_active = 0;
Andy Grover5c115592009-02-24 15:30:27 +0000344 }
345
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700346 if (rm->data.op_active && !cp->cp_xmit_data_sent) {
Andy Groverff3d7d32010-03-01 14:03:09 -0800347 rm->m_final_op = &rm->data;
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700348
Andy Grover5c115592009-02-24 15:30:27 +0000349 ret = conn->c_trans->xmit(conn, rm,
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700350 cp->cp_xmit_hdr_off,
351 cp->cp_xmit_sg,
352 cp->cp_xmit_data_off);
Andy Grover5c115592009-02-24 15:30:27 +0000353 if (ret <= 0)
354 break;
355
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700356 if (cp->cp_xmit_hdr_off < sizeof(struct rds_header)) {
Andy Grover5c115592009-02-24 15:30:27 +0000357 tmp = min_t(int, ret,
358 sizeof(struct rds_header) -
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700359 cp->cp_xmit_hdr_off);
360 cp->cp_xmit_hdr_off += tmp;
Andy Grover5c115592009-02-24 15:30:27 +0000361 ret -= tmp;
362 }
363
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700364 sg = &rm->data.op_sg[cp->cp_xmit_sg];
Andy Grover5c115592009-02-24 15:30:27 +0000365 while (ret) {
366 tmp = min_t(int, ret, sg->length -
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700367 cp->cp_xmit_data_off);
368 cp->cp_xmit_data_off += tmp;
Andy Grover5c115592009-02-24 15:30:27 +0000369 ret -= tmp;
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700370 if (cp->cp_xmit_data_off == sg->length) {
371 cp->cp_xmit_data_off = 0;
Andy Grover5c115592009-02-24 15:30:27 +0000372 sg++;
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700373 cp->cp_xmit_sg++;
374 BUG_ON(ret != 0 && cp->cp_xmit_sg ==
375 rm->data.op_nents);
Andy Grover5c115592009-02-24 15:30:27 +0000376 }
377 }
Andy Grover5b2366b2010-02-03 19:36:44 -0800378
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700379 if (cp->cp_xmit_hdr_off == sizeof(struct rds_header) &&
380 (cp->cp_xmit_sg == rm->data.op_nents))
381 cp->cp_xmit_data_sent = 1;
Andy Grover5b2366b2010-02-03 19:36:44 -0800382 }
383
384 /*
385 * A rm will only take multiple times through this loop
386 * if there is a data op. Thus, if the data is sent (or there was
387 * none), then we're done with the rm.
388 */
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700389 if (!rm->data.op_active || cp->cp_xmit_data_sent) {
390 cp->cp_xmit_rm = NULL;
391 cp->cp_xmit_sg = 0;
392 cp->cp_xmit_hdr_off = 0;
393 cp->cp_xmit_data_off = 0;
394 cp->cp_xmit_rdma_sent = 0;
395 cp->cp_xmit_atomic_sent = 0;
396 cp->cp_xmit_data_sent = 0;
Andy Grover5b2366b2010-02-03 19:36:44 -0800397
398 rds_message_put(rm);
Andy Grover5c115592009-02-24 15:30:27 +0000399 }
400 }
401
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400402over_batch:
Sowmini Varadhan226f7a72016-06-30 16:11:10 -0700403 if (conn->c_trans->xmit_path_complete)
404 conn->c_trans->xmit_path_complete(cp);
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700405 release_in_xmit(cp);
Andy Grover5c115592009-02-24 15:30:27 +0000406
Andy Grover2ad80992010-03-23 17:48:04 -0700407 /* Nuke any messages we decided not to retransmit. */
408 if (!list_empty(&to_be_dropped)) {
409 /* irqs on here, so we can put(), unlike above */
410 list_for_each_entry(rm, &to_be_dropped, m_conn_item)
411 rds_message_put(rm);
412 rds_send_remove_from_sock(&to_be_dropped, RDS_RDMA_DROPPED);
413 }
414
Andy Groverfcc54502010-03-29 17:08:49 -0700415 /*
Zach Brown0f4b1c72010-06-04 14:41:41 -0700416 * Other senders can queue a message after we last test the send queue
417 * but before we clear RDS_IN_XMIT. In that case they'd back off and
418 * not try and send their newly queued message. We need to check the
419 * send queue after having cleared RDS_IN_XMIT so that their message
420 * doesn't get stuck on the send queue.
Andy Groverfcc54502010-03-29 17:08:49 -0700421 *
422 * If the transport cannot continue (i.e ret != 0), then it must
423 * call us when more room is available, such as from the tx
424 * completion handler.
Sowmini Varadhan443be0e2015-04-08 12:33:47 -0400425 *
426 * We have an extra generation check here so that if someone manages
427 * to jump in after our release_in_xmit, we'll see that they have done
428 * some work and we will skip our goto
Andy Groverfcc54502010-03-29 17:08:49 -0700429 */
430 if (ret == 0) {
Chris Mason9e29db02010-04-15 16:38:14 -0400431 smp_mb();
santosh.shilimkar@oracle.com0c484242015-08-22 15:45:27 -0700432 if ((test_bit(0, &conn->c_map_queued) ||
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700433 !list_empty(&cp->cp_send_queue)) &&
434 send_gen == cp->cp_send_gen) {
Andy Grover049ee3f2010-03-23 17:39:07 -0700435 rds_stats_inc(s_send_lock_queue_raced);
Santosh Shilimkar4bebdd72015-09-10 11:57:14 -0700436 if (batch_count < send_batch_count)
437 goto restart;
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -0700438 queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
Andy Grover5c115592009-02-24 15:30:27 +0000439 }
Andy Grover5c115592009-02-24 15:30:27 +0000440 }
441out:
442 return ret;
443}
Santosh Shilimkar0c28c042015-09-06 02:18:51 -0400444EXPORT_SYMBOL_GPL(rds_send_xmit);
Andy Grover5c115592009-02-24 15:30:27 +0000445
446static void rds_send_sndbuf_remove(struct rds_sock *rs, struct rds_message *rm)
447{
448 u32 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
449
450 assert_spin_locked(&rs->rs_lock);
451
452 BUG_ON(rs->rs_snd_bytes < len);
453 rs->rs_snd_bytes -= len;
454
455 if (rs->rs_snd_bytes == 0)
456 rds_stats_inc(s_send_queue_empty);
457}
458
459static inline int rds_send_is_acked(struct rds_message *rm, u64 ack,
460 is_acked_func is_acked)
461{
462 if (is_acked)
463 return is_acked(rm, ack);
464 return be64_to_cpu(rm->m_inc.i_hdr.h_sequence) <= ack;
465}
466
467/*
Andy Grover5c115592009-02-24 15:30:27 +0000468 * This is pretty similar to what happens below in the ACK
469 * handling code - except that we call here as soon as we get
470 * the IB send completion on the RDMA op and the accompanying
471 * message.
472 */
473void rds_rdma_send_complete(struct rds_message *rm, int status)
474{
475 struct rds_sock *rs = NULL;
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800476 struct rm_rdma_op *ro;
Andy Grover5c115592009-02-24 15:30:27 +0000477 struct rds_notifier *notifier;
Andy Grover9de08642010-03-29 16:50:54 -0700478 unsigned long flags;
Santosh Shilimkar941f8d52016-02-18 20:06:47 -0800479 unsigned int notify = 0;
Andy Grover5c115592009-02-24 15:30:27 +0000480
Andy Grover9de08642010-03-29 16:50:54 -0700481 spin_lock_irqsave(&rm->m_rs_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000482
Santosh Shilimkar941f8d52016-02-18 20:06:47 -0800483 notify = rm->rdma.op_notify | rm->data.op_notify;
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800484 ro = &rm->rdma;
Joe Perchesf64f9e72009-11-29 16:55:45 -0800485 if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags) &&
Santosh Shilimkar941f8d52016-02-18 20:06:47 -0800486 ro->op_active && notify && ro->op_notifier) {
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800487 notifier = ro->op_notifier;
Andy Grover5c115592009-02-24 15:30:27 +0000488 rs = rm->m_rs;
489 sock_hold(rds_rs_to_sk(rs));
490
491 notifier->n_status = status;
492 spin_lock(&rs->rs_lock);
493 list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
494 spin_unlock(&rs->rs_lock);
495
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800496 ro->op_notifier = NULL;
Andy Grover5c115592009-02-24 15:30:27 +0000497 }
498
Andy Grover9de08642010-03-29 16:50:54 -0700499 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000500
501 if (rs) {
502 rds_wake_sk_sleep(rs);
503 sock_put(rds_rs_to_sk(rs));
504 }
505}
Andy Grover616b7572009-08-21 12:28:32 +0000506EXPORT_SYMBOL_GPL(rds_rdma_send_complete);
Andy Grover5c115592009-02-24 15:30:27 +0000507
508/*
Andy Grover15133f62010-01-12 14:33:38 -0800509 * Just like above, except looks at atomic op
510 */
511void rds_atomic_send_complete(struct rds_message *rm, int status)
512{
513 struct rds_sock *rs = NULL;
514 struct rm_atomic_op *ao;
515 struct rds_notifier *notifier;
Andy Grovercf4b7382010-03-29 16:50:54 -0700516 unsigned long flags;
Andy Grover15133f62010-01-12 14:33:38 -0800517
Andy Grovercf4b7382010-03-29 16:50:54 -0700518 spin_lock_irqsave(&rm->m_rs_lock, flags);
Andy Grover15133f62010-01-12 14:33:38 -0800519
520 ao = &rm->atomic;
521 if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags)
522 && ao->op_active && ao->op_notify && ao->op_notifier) {
523 notifier = ao->op_notifier;
524 rs = rm->m_rs;
525 sock_hold(rds_rs_to_sk(rs));
526
527 notifier->n_status = status;
528 spin_lock(&rs->rs_lock);
529 list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
530 spin_unlock(&rs->rs_lock);
531
532 ao->op_notifier = NULL;
533 }
534
Andy Grovercf4b7382010-03-29 16:50:54 -0700535 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
Andy Grover15133f62010-01-12 14:33:38 -0800536
537 if (rs) {
538 rds_wake_sk_sleep(rs);
539 sock_put(rds_rs_to_sk(rs));
540 }
541}
542EXPORT_SYMBOL_GPL(rds_atomic_send_complete);
543
544/*
Andy Grover5c115592009-02-24 15:30:27 +0000545 * This is the same as rds_rdma_send_complete except we
546 * don't do any locking - we have all the ingredients (message,
547 * socket, socket lock) and can just move the notifier.
548 */
549static inline void
Andy Grover940786e2010-02-19 18:04:58 -0800550__rds_send_complete(struct rds_sock *rs, struct rds_message *rm, int status)
Andy Grover5c115592009-02-24 15:30:27 +0000551{
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800552 struct rm_rdma_op *ro;
Andy Grover940786e2010-02-19 18:04:58 -0800553 struct rm_atomic_op *ao;
Andy Grover5c115592009-02-24 15:30:27 +0000554
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800555 ro = &rm->rdma;
556 if (ro->op_active && ro->op_notify && ro->op_notifier) {
557 ro->op_notifier->n_status = status;
558 list_add_tail(&ro->op_notifier->n_list, &rs->rs_notify_queue);
559 ro->op_notifier = NULL;
Andy Grover5c115592009-02-24 15:30:27 +0000560 }
561
Andy Grover940786e2010-02-19 18:04:58 -0800562 ao = &rm->atomic;
563 if (ao->op_active && ao->op_notify && ao->op_notifier) {
564 ao->op_notifier->n_status = status;
565 list_add_tail(&ao->op_notifier->n_list, &rs->rs_notify_queue);
566 ao->op_notifier = NULL;
567 }
568
Andy Grover5c115592009-02-24 15:30:27 +0000569 /* No need to wake the app - caller does this */
570}
571
572/*
Andy Grover5c115592009-02-24 15:30:27 +0000573 * This removes messages from the socket's list if they're on it. The list
574 * argument must be private to the caller, we must be able to modify it
575 * without locks. The messages must have a reference held for their
576 * position on the list. This function will drop that reference after
577 * removing the messages from the 'messages' list regardless of if it found
578 * the messages on the socket list or not.
579 */
stephen hemmingerff51bf82010-10-19 08:08:33 +0000580static void rds_send_remove_from_sock(struct list_head *messages, int status)
Andy Grover5c115592009-02-24 15:30:27 +0000581{
Andy Grover561c7df2010-03-11 13:50:06 +0000582 unsigned long flags;
Andy Grover5c115592009-02-24 15:30:27 +0000583 struct rds_sock *rs = NULL;
584 struct rds_message *rm;
585
Andy Grover5c115592009-02-24 15:30:27 +0000586 while (!list_empty(messages)) {
Andy Grover561c7df2010-03-11 13:50:06 +0000587 int was_on_sock = 0;
588
Andy Grover5c115592009-02-24 15:30:27 +0000589 rm = list_entry(messages->next, struct rds_message,
590 m_conn_item);
591 list_del_init(&rm->m_conn_item);
592
593 /*
594 * If we see this flag cleared then we're *sure* that someone
595 * else beat us to removing it from the sock. If we race
596 * with their flag update we'll get the lock and then really
597 * see that the flag has been cleared.
598 *
599 * The message spinlock makes sure nobody clears rm->m_rs
600 * while we're messing with it. It does not prevent the
601 * message from being removed from the socket, though.
602 */
Andy Grover561c7df2010-03-11 13:50:06 +0000603 spin_lock_irqsave(&rm->m_rs_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000604 if (!test_bit(RDS_MSG_ON_SOCK, &rm->m_flags))
605 goto unlock_and_drop;
606
607 if (rs != rm->m_rs) {
608 if (rs) {
Andy Grover5c115592009-02-24 15:30:27 +0000609 rds_wake_sk_sleep(rs);
610 sock_put(rds_rs_to_sk(rs));
611 }
612 rs = rm->m_rs;
Herton R. Krzesinski593cbb32014-10-01 18:49:54 -0300613 if (rs)
614 sock_hold(rds_rs_to_sk(rs));
Andy Grover5c115592009-02-24 15:30:27 +0000615 }
Herton R. Krzesinski593cbb32014-10-01 18:49:54 -0300616 if (!rs)
617 goto unlock_and_drop;
Tina Yang048c15e2010-03-11 13:50:00 +0000618 spin_lock(&rs->rs_lock);
Andy Grover5c115592009-02-24 15:30:27 +0000619
620 if (test_and_clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags)) {
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800621 struct rm_rdma_op *ro = &rm->rdma;
Andy Grover5c115592009-02-24 15:30:27 +0000622 struct rds_notifier *notifier;
623
624 list_del_init(&rm->m_sock_item);
625 rds_send_sndbuf_remove(rs, rm);
626
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800627 if (ro->op_active && ro->op_notifier &&
628 (ro->op_notify || (ro->op_recverr && status))) {
629 notifier = ro->op_notifier;
Andy Grover5c115592009-02-24 15:30:27 +0000630 list_add_tail(&notifier->n_list,
631 &rs->rs_notify_queue);
632 if (!notifier->n_status)
633 notifier->n_status = status;
Andy Groverf8b3aaf2010-03-01 14:11:53 -0800634 rm->rdma.op_notifier = NULL;
Andy Grover5c115592009-02-24 15:30:27 +0000635 }
Andy Grover561c7df2010-03-11 13:50:06 +0000636 was_on_sock = 1;
Andy Grover5c115592009-02-24 15:30:27 +0000637 rm->m_rs = NULL;
638 }
Tina Yang048c15e2010-03-11 13:50:00 +0000639 spin_unlock(&rs->rs_lock);
Andy Grover5c115592009-02-24 15:30:27 +0000640
641unlock_and_drop:
Andy Grover561c7df2010-03-11 13:50:06 +0000642 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000643 rds_message_put(rm);
Andy Grover561c7df2010-03-11 13:50:06 +0000644 if (was_on_sock)
645 rds_message_put(rm);
Andy Grover5c115592009-02-24 15:30:27 +0000646 }
647
648 if (rs) {
Andy Grover5c115592009-02-24 15:30:27 +0000649 rds_wake_sk_sleep(rs);
650 sock_put(rds_rs_to_sk(rs));
651 }
Andy Grover5c115592009-02-24 15:30:27 +0000652}
653
654/*
655 * Transports call here when they've determined that the receiver queued
656 * messages up to, and including, the given sequence number. Messages are
657 * moved to the retrans queue when rds_send_xmit picks them off the send
658 * queue. This means that in the TCP case, the message may not have been
659 * assigned the m_ack_seq yet - but that's fine as long as tcp_is_acked
660 * checks the RDS_MSG_HAS_ACK_SEQ bit.
Andy Grover5c115592009-02-24 15:30:27 +0000661 */
Sowmini Varadhan5c3d2742016-06-13 09:44:31 -0700662void rds_send_path_drop_acked(struct rds_conn_path *cp, u64 ack,
663 is_acked_func is_acked)
Andy Grover5c115592009-02-24 15:30:27 +0000664{
665 struct rds_message *rm, *tmp;
666 unsigned long flags;
667 LIST_HEAD(list);
668
Sowmini Varadhan5c3d2742016-06-13 09:44:31 -0700669 spin_lock_irqsave(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000670
Sowmini Varadhan5c3d2742016-06-13 09:44:31 -0700671 list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
Andy Grover5c115592009-02-24 15:30:27 +0000672 if (!rds_send_is_acked(rm, ack, is_acked))
673 break;
674
675 list_move(&rm->m_conn_item, &list);
676 clear_bit(RDS_MSG_ON_CONN, &rm->m_flags);
677 }
678
679 /* order flag updates with spin locks */
680 if (!list_empty(&list))
Peter Zijlstra4e857c52014-03-17 18:06:10 +0100681 smp_mb__after_atomic();
Andy Grover5c115592009-02-24 15:30:27 +0000682
Sowmini Varadhan5c3d2742016-06-13 09:44:31 -0700683 spin_unlock_irqrestore(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000684
685 /* now remove the messages from the sock list as needed */
686 rds_send_remove_from_sock(&list, RDS_RDMA_SUCCESS);
687}
Sowmini Varadhan5c3d2742016-06-13 09:44:31 -0700688EXPORT_SYMBOL_GPL(rds_send_path_drop_acked);
689
690void rds_send_drop_acked(struct rds_connection *conn, u64 ack,
691 is_acked_func is_acked)
692{
693 WARN_ON(conn->c_trans->t_mp_capable);
694 rds_send_path_drop_acked(&conn->c_path[0], ack, is_acked);
695}
Andy Grover616b7572009-08-21 12:28:32 +0000696EXPORT_SYMBOL_GPL(rds_send_drop_acked);
Andy Grover5c115592009-02-24 15:30:27 +0000697
698void rds_send_drop_to(struct rds_sock *rs, struct sockaddr_in *dest)
699{
700 struct rds_message *rm, *tmp;
701 struct rds_connection *conn;
Sowmini Varadhan01ff34e2016-06-13 09:44:35 -0700702 struct rds_conn_path *cp;
Andy Grover7c82eaf2010-02-19 18:01:41 -0800703 unsigned long flags;
Andy Grover5c115592009-02-24 15:30:27 +0000704 LIST_HEAD(list);
Andy Grover5c115592009-02-24 15:30:27 +0000705
706 /* get all the messages we're dropping under the rs lock */
707 spin_lock_irqsave(&rs->rs_lock, flags);
708
709 list_for_each_entry_safe(rm, tmp, &rs->rs_send_queue, m_sock_item) {
710 if (dest && (dest->sin_addr.s_addr != rm->m_daddr ||
711 dest->sin_port != rm->m_inc.i_hdr.h_dport))
712 continue;
713
Andy Grover5c115592009-02-24 15:30:27 +0000714 list_move(&rm->m_sock_item, &list);
715 rds_send_sndbuf_remove(rs, rm);
716 clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
Andy Grover5c115592009-02-24 15:30:27 +0000717 }
718
719 /* order flag updates with the rs lock */
Peter Zijlstra4e857c52014-03-17 18:06:10 +0100720 smp_mb__after_atomic();
Andy Grover5c115592009-02-24 15:30:27 +0000721
722 spin_unlock_irqrestore(&rs->rs_lock, flags);
723
Andy Grover7c82eaf2010-02-19 18:01:41 -0800724 if (list_empty(&list))
725 return;
Andy Grover5c115592009-02-24 15:30:27 +0000726
Andy Grover7c82eaf2010-02-19 18:01:41 -0800727 /* Remove the messages from the conn */
Andy Grover5c115592009-02-24 15:30:27 +0000728 list_for_each_entry(rm, &list, m_sock_item) {
Andy Grover7c82eaf2010-02-19 18:01:41 -0800729
730 conn = rm->m_inc.i_conn;
Sowmini Varadhan01ff34e2016-06-13 09:44:35 -0700731 if (conn->c_trans->t_mp_capable)
732 cp = rm->m_inc.i_conn_path;
733 else
734 cp = &conn->c_path[0];
Andy Grover7c82eaf2010-02-19 18:01:41 -0800735
Sowmini Varadhan01ff34e2016-06-13 09:44:35 -0700736 spin_lock_irqsave(&cp->cp_lock, flags);
Andy Grover7c82eaf2010-02-19 18:01:41 -0800737 /*
738 * Maybe someone else beat us to removing rm from the conn.
739 * If we race with their flag update we'll get the lock and
740 * then really see that the flag has been cleared.
741 */
742 if (!test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags)) {
Sowmini Varadhan01ff34e2016-06-13 09:44:35 -0700743 spin_unlock_irqrestore(&cp->cp_lock, flags);
Herton R. Krzesinski593cbb32014-10-01 18:49:54 -0300744 spin_lock_irqsave(&rm->m_rs_lock, flags);
745 rm->m_rs = NULL;
746 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
Andy Grover7c82eaf2010-02-19 18:01:41 -0800747 continue;
748 }
Andy Grover9de08642010-03-29 16:50:54 -0700749 list_del_init(&rm->m_conn_item);
Sowmini Varadhan01ff34e2016-06-13 09:44:35 -0700750 spin_unlock_irqrestore(&cp->cp_lock, flags);
Andy Grover7c82eaf2010-02-19 18:01:41 -0800751
752 /*
753 * Couldn't grab m_rs_lock in top loop (lock ordering),
754 * but we can now.
755 */
Andy Grover9de08642010-03-29 16:50:54 -0700756 spin_lock_irqsave(&rm->m_rs_lock, flags);
Andy Grover7c82eaf2010-02-19 18:01:41 -0800757
Tina Yang550a8002010-03-11 13:50:03 +0000758 spin_lock(&rs->rs_lock);
Andy Grover940786e2010-02-19 18:04:58 -0800759 __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
Tina Yang550a8002010-03-11 13:50:03 +0000760 spin_unlock(&rs->rs_lock);
Andy Grover7c82eaf2010-02-19 18:01:41 -0800761
Andy Grover5c115592009-02-24 15:30:27 +0000762 rm->m_rs = NULL;
Andy Grover9de08642010-03-29 16:50:54 -0700763 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +0000764
Andy Grover7c82eaf2010-02-19 18:01:41 -0800765 rds_message_put(rm);
Andy Grover5c115592009-02-24 15:30:27 +0000766 }
767
Andy Grover7c82eaf2010-02-19 18:01:41 -0800768 rds_wake_sk_sleep(rs);
Tina Yang550a8002010-03-11 13:50:03 +0000769
Andy Grover5c115592009-02-24 15:30:27 +0000770 while (!list_empty(&list)) {
771 rm = list_entry(list.next, struct rds_message, m_sock_item);
772 list_del_init(&rm->m_sock_item);
Andy Grover5c115592009-02-24 15:30:27 +0000773 rds_message_wait(rm);
santosh.shilimkar@oracle.comdfcec252015-08-22 15:45:33 -0700774
775 /* just in case the code above skipped this message
776 * because RDS_MSG_ON_CONN wasn't set, run it again here
777 * taking m_rs_lock is the only thing that keeps us
778 * from racing with ack processing.
779 */
780 spin_lock_irqsave(&rm->m_rs_lock, flags);
781
782 spin_lock(&rs->rs_lock);
783 __rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
784 spin_unlock(&rs->rs_lock);
785
786 rm->m_rs = NULL;
787 spin_unlock_irqrestore(&rm->m_rs_lock, flags);
788
Andy Grover5c115592009-02-24 15:30:27 +0000789 rds_message_put(rm);
790 }
791}
792
793/*
794 * we only want this to fire once so we use the callers 'queued'. It's
795 * possible that another thread can race with us and remove the
796 * message from the flow with RDS_CANCEL_SENT_TO.
797 */
798static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
Sowmini Varadhan780a6d92016-06-13 09:44:33 -0700799 struct rds_conn_path *cp,
Andy Grover5c115592009-02-24 15:30:27 +0000800 struct rds_message *rm, __be16 sport,
801 __be16 dport, int *queued)
802{
803 unsigned long flags;
804 u32 len;
805
806 if (*queued)
807 goto out;
808
809 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);
810
811 /* this is the only place which holds both the socket's rs_lock
812 * and the connection's c_lock */
813 spin_lock_irqsave(&rs->rs_lock, flags);
814
815 /*
816 * If there is a little space in sndbuf, we don't queue anything,
817 * and userspace gets -EAGAIN. But poll() indicates there's send
818 * room. This can lead to bad behavior (spinning) if snd_bytes isn't
819 * freed up by incoming acks. So we check the *old* value of
820 * rs_snd_bytes here to allow the last msg to exceed the buffer,
821 * and poll() now knows no more data can be sent.
822 */
823 if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
824 rs->rs_snd_bytes += len;
825
826 /* let recv side know we are close to send space exhaustion.
827 * This is probably not the optimal way to do it, as this
828 * means we set the flag on *all* messages as soon as our
829 * throughput hits a certain threshold.
830 */
831 if (rs->rs_snd_bytes >= rds_sk_sndbuf(rs) / 2)
832 __set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
833
834 list_add_tail(&rm->m_sock_item, &rs->rs_send_queue);
835 set_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
836 rds_message_addref(rm);
837 rm->m_rs = rs;
838
839 /* The code ordering is a little weird, but we're
840 trying to minimize the time we hold c_lock */
841 rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
842 rm->m_inc.i_conn = conn;
Sowmini Varadhan780a6d92016-06-13 09:44:33 -0700843 rm->m_inc.i_conn_path = cp;
Andy Grover5c115592009-02-24 15:30:27 +0000844 rds_message_addref(rm);
845
Sowmini Varadhan780a6d92016-06-13 09:44:33 -0700846 spin_lock(&cp->cp_lock);
847 rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
848 list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
Andy Grover5c115592009-02-24 15:30:27 +0000849 set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
Sowmini Varadhan780a6d92016-06-13 09:44:33 -0700850 spin_unlock(&cp->cp_lock);
Andy Grover5c115592009-02-24 15:30:27 +0000851
852 rdsdebug("queued msg %p len %d, rs %p bytes %d seq %llu\n",
853 rm, len, rs, rs->rs_snd_bytes,
854 (unsigned long long)be64_to_cpu(rm->m_inc.i_hdr.h_sequence));
855
856 *queued = 1;
857 }
858
859 spin_unlock_irqrestore(&rs->rs_lock, flags);
860out:
861 return *queued;
862}
863
Andy Groverfc445082010-01-12 12:56:06 -0800864/*
865 * rds_message is getting to be quite complicated, and we'd like to allocate
866 * it all in one go. This figures out how big it needs to be up front.
867 */
868static int rds_rm_size(struct msghdr *msg, int data_len)
869{
Andy Groverff87e972010-01-12 14:13:15 -0800870 struct cmsghdr *cmsg;
Andy Groverfc445082010-01-12 12:56:06 -0800871 int size = 0;
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700872 int cmsg_groups = 0;
Andy Groverff87e972010-01-12 14:13:15 -0800873 int retval;
Andy Groverfc445082010-01-12 12:56:06 -0800874
Gu Zhengf95b4142014-12-11 11:22:04 +0800875 for_each_cmsghdr(cmsg, msg) {
Andy Groverff87e972010-01-12 14:13:15 -0800876 if (!CMSG_OK(msg, cmsg))
877 return -EINVAL;
878
879 if (cmsg->cmsg_level != SOL_RDS)
880 continue;
881
882 switch (cmsg->cmsg_type) {
883 case RDS_CMSG_RDMA_ARGS:
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700884 cmsg_groups |= 1;
Andy Groverff87e972010-01-12 14:13:15 -0800885 retval = rds_rdma_extra_size(CMSG_DATA(cmsg));
886 if (retval < 0)
887 return retval;
888 size += retval;
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700889
Andy Groverff87e972010-01-12 14:13:15 -0800890 break;
891
892 case RDS_CMSG_RDMA_DEST:
893 case RDS_CMSG_RDMA_MAP:
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700894 cmsg_groups |= 2;
Andy Groverff87e972010-01-12 14:13:15 -0800895 /* these are valid but do no add any size */
896 break;
897
Andy Grover15133f62010-01-12 14:33:38 -0800898 case RDS_CMSG_ATOMIC_CSWP:
899 case RDS_CMSG_ATOMIC_FADD:
Andy Grover20c72bd2010-08-25 05:51:28 -0700900 case RDS_CMSG_MASKED_ATOMIC_CSWP:
901 case RDS_CMSG_MASKED_ATOMIC_FADD:
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700902 cmsg_groups |= 1;
Andy Grover15133f62010-01-12 14:33:38 -0800903 size += sizeof(struct scatterlist);
904 break;
905
Andy Groverff87e972010-01-12 14:13:15 -0800906 default:
907 return -EINVAL;
908 }
909
910 }
911
912 size += ceil(data_len, PAGE_SIZE) * sizeof(struct scatterlist);
Andy Groverfc445082010-01-12 12:56:06 -0800913
Andy Groveraa0a4ef2010-04-13 12:00:35 -0700914 /* Ensure (DEST, MAP) are never used with (ARGS, ATOMIC) */
915 if (cmsg_groups == 3)
916 return -EINVAL;
917
Andy Groverfc445082010-01-12 12:56:06 -0800918 return size;
919}
920
Andy Grover5c115592009-02-24 15:30:27 +0000921static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
922 struct msghdr *msg, int *allocated_mr)
923{
924 struct cmsghdr *cmsg;
925 int ret = 0;
926
Gu Zhengf95b4142014-12-11 11:22:04 +0800927 for_each_cmsghdr(cmsg, msg) {
Andy Grover5c115592009-02-24 15:30:27 +0000928 if (!CMSG_OK(msg, cmsg))
929 return -EINVAL;
930
931 if (cmsg->cmsg_level != SOL_RDS)
932 continue;
933
934 /* As a side effect, RDMA_DEST and RDMA_MAP will set
Andy Grover15133f62010-01-12 14:33:38 -0800935 * rm->rdma.m_rdma_cookie and rm->rdma.m_rdma_mr.
Andy Grover5c115592009-02-24 15:30:27 +0000936 */
937 switch (cmsg->cmsg_type) {
938 case RDS_CMSG_RDMA_ARGS:
939 ret = rds_cmsg_rdma_args(rs, rm, cmsg);
940 break;
941
942 case RDS_CMSG_RDMA_DEST:
943 ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
944 break;
945
946 case RDS_CMSG_RDMA_MAP:
947 ret = rds_cmsg_rdma_map(rs, rm, cmsg);
948 if (!ret)
949 *allocated_mr = 1;
Santosh Shilimkar584a8272016-07-04 17:04:37 -0700950 else if (ret == -ENODEV)
951 /* Accommodate the get_mr() case which can fail
952 * if connection isn't established yet.
953 */
954 ret = -EAGAIN;
Andy Grover5c115592009-02-24 15:30:27 +0000955 break;
Andy Grover15133f62010-01-12 14:33:38 -0800956 case RDS_CMSG_ATOMIC_CSWP:
957 case RDS_CMSG_ATOMIC_FADD:
Andy Grover20c72bd2010-08-25 05:51:28 -0700958 case RDS_CMSG_MASKED_ATOMIC_CSWP:
959 case RDS_CMSG_MASKED_ATOMIC_FADD:
Andy Grover15133f62010-01-12 14:33:38 -0800960 ret = rds_cmsg_atomic(rs, rm, cmsg);
961 break;
Andy Grover5c115592009-02-24 15:30:27 +0000962
963 default:
964 return -EINVAL;
965 }
966
967 if (ret)
968 break;
969 }
970
971 return ret;
972}
973
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -0700974static void rds_send_ping(struct rds_connection *conn);
975
976static int rds_send_mprds_hash(struct rds_sock *rs, struct rds_connection *conn)
977{
978 int hash;
979
980 if (conn->c_npaths == 0)
981 hash = RDS_MPATH_HASH(rs, RDS_MPATH_WORKERS);
982 else
983 hash = RDS_MPATH_HASH(rs, conn->c_npaths);
984 if (conn->c_npaths == 0 && hash != 0) {
985 rds_send_ping(conn);
986
987 if (conn->c_npaths == 0) {
988 wait_event_interruptible(conn->c_hs_waitq,
989 (conn->c_npaths != 0));
990 }
991 if (conn->c_npaths == 1)
992 hash = 0;
993 }
994 return hash;
995}
996
Avinash Repakaf9fb69a2016-02-29 15:30:57 -0800997static int rds_rdma_bytes(struct msghdr *msg, size_t *rdma_bytes)
998{
999 struct rds_rdma_args *args;
1000 struct cmsghdr *cmsg;
1001
1002 for_each_cmsghdr(cmsg, msg) {
1003 if (!CMSG_OK(msg, cmsg))
1004 return -EINVAL;
1005
1006 if (cmsg->cmsg_level != SOL_RDS)
1007 continue;
1008
1009 if (cmsg->cmsg_type == RDS_CMSG_RDMA_ARGS) {
1010 args = CMSG_DATA(cmsg);
1011 *rdma_bytes += args->remote_vec.bytes;
1012 }
1013 }
1014 return 0;
1015}
1016
Ying Xue1b784142015-03-02 15:37:48 +08001017int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
Andy Grover5c115592009-02-24 15:30:27 +00001018{
1019 struct sock *sk = sock->sk;
1020 struct rds_sock *rs = rds_sk_to_rs(sk);
Steffen Hurrle342dfc32014-01-17 22:53:15 +01001021 DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
Andy Grover5c115592009-02-24 15:30:27 +00001022 __be32 daddr;
1023 __be16 dport;
1024 struct rds_message *rm = NULL;
1025 struct rds_connection *conn;
1026 int ret = 0;
1027 int queued = 0, allocated_mr = 0;
1028 int nonblock = msg->msg_flags & MSG_DONTWAIT;
Andy Grover1123fd72010-03-11 13:49:56 +00001029 long timeo = sock_sndtimeo(sk, nonblock);
Sowmini Varadhan780a6d92016-06-13 09:44:33 -07001030 struct rds_conn_path *cpath;
Avinash Repakaf9fb69a2016-02-29 15:30:57 -08001031 size_t total_payload_len = payload_len, rdma_payload_len = 0;
Andy Grover5c115592009-02-24 15:30:27 +00001032
1033 /* Mirror Linux UDP mirror of BSD error message compatibility */
1034 /* XXX: Perhaps MSG_MORE someday */
1035 if (msg->msg_flags & ~(MSG_DONTWAIT | MSG_CMSG_COMPAT)) {
Andy Grover5c115592009-02-24 15:30:27 +00001036 ret = -EOPNOTSUPP;
1037 goto out;
1038 }
1039
1040 if (msg->msg_namelen) {
1041 /* XXX fail non-unicast destination IPs? */
1042 if (msg->msg_namelen < sizeof(*usin) || usin->sin_family != AF_INET) {
1043 ret = -EINVAL;
1044 goto out;
1045 }
1046 daddr = usin->sin_addr.s_addr;
1047 dport = usin->sin_port;
1048 } else {
1049 /* We only care about consistency with ->connect() */
1050 lock_sock(sk);
1051 daddr = rs->rs_conn_addr;
1052 dport = rs->rs_conn_port;
1053 release_sock(sk);
1054 }
1055
Quentin Casasnovas8c7188b2015-11-24 17:13:21 -05001056 lock_sock(sk);
Andy Grover5c115592009-02-24 15:30:27 +00001057 if (daddr == 0 || rs->rs_bound_addr == 0) {
Quentin Casasnovas8c7188b2015-11-24 17:13:21 -05001058 release_sock(sk);
Andy Grover5c115592009-02-24 15:30:27 +00001059 ret = -ENOTCONN; /* XXX not a great errno */
1060 goto out;
1061 }
Quentin Casasnovas8c7188b2015-11-24 17:13:21 -05001062 release_sock(sk);
Andy Grover5c115592009-02-24 15:30:27 +00001063
Avinash Repakaf9fb69a2016-02-29 15:30:57 -08001064 ret = rds_rdma_bytes(msg, &rdma_payload_len);
1065 if (ret)
1066 goto out;
1067
1068 total_payload_len += rdma_payload_len;
1069 if (max_t(size_t, payload_len, rdma_payload_len) > RDS_MAX_MSG_SIZE) {
1070 ret = -EMSGSIZE;
1071 goto out;
1072 }
1073
Mukesh Kacker06e89412015-08-22 15:45:34 -07001074 if (payload_len > rds_sk_sndbuf(rs)) {
1075 ret = -EMSGSIZE;
1076 goto out;
1077 }
1078
Andy Groverfc445082010-01-12 12:56:06 -08001079 /* size of rm including all sgs */
1080 ret = rds_rm_size(msg, payload_len);
1081 if (ret < 0)
1082 goto out;
1083
1084 rm = rds_message_alloc(ret, GFP_KERNEL);
1085 if (!rm) {
1086 ret = -ENOMEM;
Andy Grover5c115592009-02-24 15:30:27 +00001087 goto out;
1088 }
1089
Andy Grover372cd7d2010-02-03 19:40:32 -08001090 /* Attach data to the rm */
1091 if (payload_len) {
1092 rm->data.op_sg = rds_message_alloc_sgs(rm, ceil(payload_len, PAGE_SIZE));
Andy Groverd139ff02010-10-28 15:40:59 +00001093 if (!rm->data.op_sg) {
1094 ret = -ENOMEM;
1095 goto out;
1096 }
Al Viroc0371da2014-11-24 10:42:55 -05001097 ret = rds_message_copy_from_user(rm, &msg->msg_iter);
Andy Grover372cd7d2010-02-03 19:40:32 -08001098 if (ret)
1099 goto out;
1100 }
1101 rm->data.op_active = 1;
Andy Groverfc445082010-01-12 12:56:06 -08001102
Andy Grover5c115592009-02-24 15:30:27 +00001103 rm->m_daddr = daddr;
1104
Andy Grover5c115592009-02-24 15:30:27 +00001105 /* rds_conn_create has a spinlock that runs with IRQ off.
1106 * Caching the conn in the socket helps a lot. */
1107 if (rs->rs_conn && rs->rs_conn->c_faddr == daddr)
1108 conn = rs->rs_conn;
1109 else {
Sowmini Varadhand5a8ac22015-08-05 01:43:25 -04001110 conn = rds_conn_create_outgoing(sock_net(sock->sk),
1111 rs->rs_bound_addr, daddr,
Andy Grover5c115592009-02-24 15:30:27 +00001112 rs->rs_transport,
1113 sock->sk->sk_allocation);
1114 if (IS_ERR(conn)) {
1115 ret = PTR_ERR(conn);
1116 goto out;
1117 }
1118 rs->rs_conn = conn;
1119 }
1120
Andy Grover49f69692009-04-09 14:09:41 +00001121 /* Parse any control messages the user may have included. */
1122 ret = rds_cmsg_send(rs, rm, msg, &allocated_mr);
Santosh Shilimkar584a8272016-07-04 17:04:37 -07001123 if (ret) {
1124 /* Trigger connection so that its ready for the next retry */
1125 if (ret == -EAGAIN)
1126 rds_conn_connect_if_down(conn);
Andy Grover49f69692009-04-09 14:09:41 +00001127 goto out;
Santosh Shilimkar584a8272016-07-04 17:04:37 -07001128 }
Andy Grover49f69692009-04-09 14:09:41 +00001129
Andy Grover2c3a5f92010-03-01 16:10:40 -08001130 if (rm->rdma.op_active && !conn->c_trans->xmit_rdma) {
Manuel Zerpiescb0a6052011-06-16 02:09:57 +00001131 printk_ratelimited(KERN_NOTICE "rdma_op %p conn xmit_rdma %p\n",
Andy Groverf8b3aaf2010-03-01 14:11:53 -08001132 &rm->rdma, conn->c_trans->xmit_rdma);
Andy Grover15133f62010-01-12 14:33:38 -08001133 ret = -EOPNOTSUPP;
1134 goto out;
1135 }
1136
1137 if (rm->atomic.op_active && !conn->c_trans->xmit_atomic) {
Manuel Zerpiescb0a6052011-06-16 02:09:57 +00001138 printk_ratelimited(KERN_NOTICE "atomic_op %p conn xmit_atomic %p\n",
Andy Grover15133f62010-01-12 14:33:38 -08001139 &rm->atomic, conn->c_trans->xmit_atomic);
Andy Grover5c115592009-02-24 15:30:27 +00001140 ret = -EOPNOTSUPP;
1141 goto out;
1142 }
1143
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001144 if (conn->c_trans->t_mp_capable)
1145 cpath = &conn->c_path[rds_send_mprds_hash(rs, conn)];
1146 else
1147 cpath = &conn->c_path[0];
Sowmini Varadhan3c0a5902016-06-13 09:44:37 -07001148
1149 rds_conn_path_connect_if_down(cpath);
Andy Grover5c115592009-02-24 15:30:27 +00001150
1151 ret = rds_cong_wait(conn->c_fcong, dport, nonblock, rs);
Andy Groverb98ba522010-03-11 13:50:04 +00001152 if (ret) {
1153 rs->rs_seen_congestion = 1;
Andy Grover5c115592009-02-24 15:30:27 +00001154 goto out;
Andy Groverb98ba522010-03-11 13:50:04 +00001155 }
Sowmini Varadhan780a6d92016-06-13 09:44:33 -07001156 while (!rds_send_queue_rm(rs, conn, cpath, rm, rs->rs_bound_port,
Andy Grover5c115592009-02-24 15:30:27 +00001157 dport, &queued)) {
1158 rds_stats_inc(s_send_queue_full);
Mukesh Kacker06e89412015-08-22 15:45:34 -07001159
Andy Grover5c115592009-02-24 15:30:27 +00001160 if (nonblock) {
1161 ret = -EAGAIN;
1162 goto out;
1163 }
1164
Eric Dumazetaa395142010-04-20 13:03:51 +00001165 timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
Sowmini Varadhan780a6d92016-06-13 09:44:33 -07001166 rds_send_queue_rm(rs, conn, cpath, rm,
Andy Grover5c115592009-02-24 15:30:27 +00001167 rs->rs_bound_port,
1168 dport,
1169 &queued),
1170 timeo);
1171 rdsdebug("sendmsg woke queued %d timeo %ld\n", queued, timeo);
1172 if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
1173 continue;
1174
1175 ret = timeo;
1176 if (ret == 0)
1177 ret = -ETIMEDOUT;
1178 goto out;
1179 }
1180
1181 /*
1182 * By now we've committed to the send. We reuse rds_send_worker()
1183 * to retry sends in the rds thread if the transport asks us to.
1184 */
1185 rds_stats_inc(s_send_queued);
1186
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -07001187 ret = rds_send_xmit(cpath);
Santosh Shilimkardb6526d2015-09-11 15:44:29 -07001188 if (ret == -ENOMEM || ret == -EAGAIN)
Sowmini Varadhan1f9ecd72016-06-13 09:44:34 -07001189 queue_delayed_work(rds_wq, &cpath->cp_send_w, 1);
Andy Grover5c115592009-02-24 15:30:27 +00001190
1191 rds_message_put(rm);
1192 return payload_len;
1193
1194out:
1195 /* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
1196 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
1197 * or in any other way, we need to destroy the MR again */
1198 if (allocated_mr)
1199 rds_rdma_unuse(rs, rds_rdma_cookie_key(rm->m_rdma_cookie), 1);
1200
1201 if (rm)
1202 rds_message_put(rm);
1203 return ret;
1204}
1205
1206/*
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001207 * send out a probe. Can be shared by rds_send_ping,
1208 * rds_send_pong, rds_send_hb.
1209 * rds_send_hb should use h_flags
1210 * RDS_FLAG_HB_PING|RDS_FLAG_ACK_REQUIRED
1211 * or
1212 * RDS_FLAG_HB_PONG|RDS_FLAG_ACK_REQUIRED
Andy Grover5c115592009-02-24 15:30:27 +00001213 */
Santosh Shilimkarbb789762016-12-04 16:41:29 -08001214static int
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001215rds_send_probe(struct rds_conn_path *cp, __be16 sport,
1216 __be16 dport, u8 h_flags)
Andy Grover5c115592009-02-24 15:30:27 +00001217{
1218 struct rds_message *rm;
1219 unsigned long flags;
1220 int ret = 0;
1221
1222 rm = rds_message_alloc(0, GFP_ATOMIC);
Andy Grover8690bfa2010-01-12 11:56:44 -08001223 if (!rm) {
Andy Grover5c115592009-02-24 15:30:27 +00001224 ret = -ENOMEM;
1225 goto out;
1226 }
1227
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001228 rm->m_daddr = cp->cp_conn->c_faddr;
Andy Groveracfcd4d2010-03-31 18:56:25 -07001229 rm->data.op_active = 1;
Andy Grover5c115592009-02-24 15:30:27 +00001230
Sowmini Varadhan3c0a5902016-06-13 09:44:37 -07001231 rds_conn_path_connect_if_down(cp);
Andy Grover5c115592009-02-24 15:30:27 +00001232
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001233 ret = rds_cong_wait(cp->cp_conn->c_fcong, dport, 1, NULL);
Andy Grover5c115592009-02-24 15:30:27 +00001234 if (ret)
1235 goto out;
1236
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001237 spin_lock_irqsave(&cp->cp_lock, flags);
1238 list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
Andy Grover5c115592009-02-24 15:30:27 +00001239 set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
1240 rds_message_addref(rm);
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001241 rm->m_inc.i_conn = cp->cp_conn;
1242 rm->m_inc.i_conn_path = cp;
Andy Grover5c115592009-02-24 15:30:27 +00001243
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001244 rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport,
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001245 cp->cp_next_tx_seq);
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001246 rm->m_inc.i_hdr.h_flags |= h_flags;
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001247 cp->cp_next_tx_seq++;
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001248
1249 if (RDS_HS_PROBE(sport, dport) && cp->cp_conn->c_trans->t_mp_capable) {
1250 u16 npaths = RDS_MPATH_WORKERS;
1251
1252 rds_message_add_extension(&rm->m_inc.i_hdr,
1253 RDS_EXTHDR_NPATHS, &npaths,
1254 sizeof(npaths));
Sowmini Varadhan905dd412016-11-16 13:29:49 -08001255 rds_message_add_extension(&rm->m_inc.i_hdr,
1256 RDS_EXTHDR_GEN_NUM,
1257 &cp->cp_conn->c_my_gen_num,
1258 sizeof(u32));
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001259 }
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001260 spin_unlock_irqrestore(&cp->cp_lock, flags);
Andy Grover5c115592009-02-24 15:30:27 +00001261
1262 rds_stats_inc(s_send_queued);
1263 rds_stats_inc(s_send_pong);
1264
santosh.shilimkar@oracle.com7b4b0002015-10-16 22:13:21 -04001265 /* schedule the send work on rds_wq */
Sowmini Varadhan45997e92016-06-13 09:44:36 -07001266 queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
Andy Groveracfcd4d2010-03-31 18:56:25 -07001267
Andy Grover5c115592009-02-24 15:30:27 +00001268 rds_message_put(rm);
1269 return 0;
1270
1271out:
1272 if (rm)
1273 rds_message_put(rm);
1274 return ret;
1275}
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001276
1277int
1278rds_send_pong(struct rds_conn_path *cp, __be16 dport)
1279{
1280 return rds_send_probe(cp, 0, dport, 0);
1281}
1282
Santosh Shilimkarbb789762016-12-04 16:41:29 -08001283static void
Sowmini Varadhan5916e2c2016-07-14 03:51:03 -07001284rds_send_ping(struct rds_connection *conn)
1285{
1286 unsigned long flags;
1287 struct rds_conn_path *cp = &conn->c_path[0];
1288
1289 spin_lock_irqsave(&cp->cp_lock, flags);
1290 if (conn->c_ping_triggered) {
1291 spin_unlock_irqrestore(&cp->cp_lock, flags);
1292 return;
1293 }
1294 conn->c_ping_triggered = 1;
1295 spin_unlock_irqrestore(&cp->cp_lock, flags);
1296 rds_send_probe(&conn->c_path[0], RDS_FLAG_PROBE_PORT, 0, 0);
1297}