send.c 34.3 KB
Newer Older
A
Andy Grover 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
/*
 * Copyright (c) 2006 Oracle.  All rights reserved.
 *
 * This software is available to you under a choice of one of two
 * licenses.  You may choose to be licensed under the terms of the GNU
 * General Public License (GPL) Version 2, available from the file
 * COPYING in the main directory of this source tree, or the
 * OpenIB.org BSD license below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 */
#include <linux/kernel.h>
34
#include <linux/moduleparam.h>
35
#include <linux/gfp.h>
A
Andy Grover 已提交
36 37 38
#include <net/sock.h>
#include <linux/in.h>
#include <linux/list.h>
39
#include <linux/ratelimit.h>
40
#include <linux/export.h>
41
#include <linux/sizes.h>
A
Andy Grover 已提交
42 43 44 45 46 47 48 49 50 51 52 53 54

#include "rds.h"

/* When transmitting messages in rds_send_xmit, we need to emerge from
 * time to time and briefly release the CPU. Otherwise the softlock watchdog
 * will kick our shin.
 * Also, it seems fairer to not let one busy connection stall all the
 * others.
 *
 * send_batch_count is the number of times we'll loop in send_xmit. Setting
 * it to 0 will restore the old behavior (where we looped until we had
 * drained the queue).
 */
55
static int send_batch_count = SZ_1K;
A
Andy Grover 已提交
56 57 58
module_param(send_batch_count, int, 0444);
MODULE_PARM_DESC(send_batch_count, " batch factor when working the send queue");

59 60
static void rds_send_remove_from_sock(struct list_head *messages, int status);

A
Andy Grover 已提交
61
/*
62 63
 * Reset the send state.  Callers must ensure that this doesn't race with
 * rds_send_xmit().
A
Andy Grover 已提交
64
 */
65
void rds_send_path_reset(struct rds_conn_path *cp)
A
Andy Grover 已提交
66 67 68 69
{
	struct rds_message *rm, *tmp;
	unsigned long flags;

70 71 72
	if (cp->cp_xmit_rm) {
		rm = cp->cp_xmit_rm;
		cp->cp_xmit_rm = NULL;
A
Andy Grover 已提交
73 74 75 76
		/* Tell the user the RDMA op is no longer mapped by the
		 * transport. This isn't entirely true (it's flushed out
		 * independently) but as the connection is down, there's
		 * no ongoing RDMA to/from that memory */
77 78
		rds_message_unmapped(rm);
		rds_message_put(rm);
A
Andy Grover 已提交
79
	}
80

81 82 83 84 85 86
	cp->cp_xmit_sg = 0;
	cp->cp_xmit_hdr_off = 0;
	cp->cp_xmit_data_off = 0;
	cp->cp_xmit_atomic_sent = 0;
	cp->cp_xmit_rdma_sent = 0;
	cp->cp_xmit_data_sent = 0;
A
Andy Grover 已提交
87

88
	cp->cp_conn->c_map_queued = 0;
A
Andy Grover 已提交
89

90 91
	cp->cp_unacked_packets = rds_sysctl_max_unacked_packets;
	cp->cp_unacked_bytes = rds_sysctl_max_unacked_bytes;
A
Andy Grover 已提交
92 93

	/* Mark messages as retransmissions, and move them to the send q */
94 95
	spin_lock_irqsave(&cp->cp_lock, flags);
	list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
A
Andy Grover 已提交
96 97 98
		set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);
		set_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags);
	}
99 100 101
	list_splice_init(&cp->cp_retrans, &cp->cp_send_queue);
	spin_unlock_irqrestore(&cp->cp_lock, flags);
}
102
EXPORT_SYMBOL_GPL(rds_send_path_reset);
A
Andy Grover 已提交
103

104
static int acquire_in_xmit(struct rds_conn_path *cp)
105
{
106
	return test_and_set_bit(RDS_IN_XMIT, &cp->cp_flags) == 0;
107 108
}

109
static void release_in_xmit(struct rds_conn_path *cp)
110
{
111
	clear_bit(RDS_IN_XMIT, &cp->cp_flags);
112
	smp_mb__after_atomic();
113 114 115 116 117 118
	/*
	 * We don't use wait_on_bit()/wake_up_bit() because our waking is in a
	 * hot path and finding waiters is very rare.  We don't want to walk
	 * the system-wide hashed waitqueue buckets in the fast path only to
	 * almost never find waiters.
	 */
119 120
	if (waitqueue_active(&cp->cp_waitq))
		wake_up_all(&cp->cp_waitq);
121 122
}

A
Andy Grover 已提交
123
/*
L
Lucas De Marchi 已提交
124
 * We're making the conscious trade-off here to only send one message
A
Andy Grover 已提交
125 126 127 128 129 130 131 132 133 134 135 136
 * down the connection at a time.
 *   Pro:
 *      - tx queueing is a simple fifo list
 *   	- reassembly is optional and easily done by transports per conn
 *      - no per flow rx lookup at all, straight to the socket
 *   	- less per-frag memory and wire overhead
 *   Con:
 *      - queued acks can be delayed behind large messages
 *   Depends:
 *      - small message latency is higher behind queued large messages
 *      - large message latency isn't starved by intervening small sends
 */
137
int rds_send_xmit(struct rds_conn_path *cp)
A
Andy Grover 已提交
138
{
139
	struct rds_connection *conn = cp->cp_conn;
A
Andy Grover 已提交
140 141 142 143 144 145
	struct rds_message *rm;
	unsigned long flags;
	unsigned int tmp;
	struct scatterlist *sg;
	int ret = 0;
	LIST_HEAD(to_be_dropped);
146 147
	int batch_count;
	unsigned long send_gen = 0;
A
Andy Grover 已提交
148

149
restart:
150
	batch_count = 0;
151

A
Andy Grover 已提交
152 153 154 155 156 157 158
	/*
	 * sendmsg calls here after having queued its message on the send
	 * queue.  We only have one task feeding the connection at a time.  If
	 * another thread is already feeding the queue then we back off.  This
	 * avoids blocking the caller and trading per-connection data between
	 * caches per message.
	 */
159
	if (!acquire_in_xmit(cp)) {
160
		rds_stats_inc(s_send_lock_contention);
A
Andy Grover 已提交
161 162 163
		ret = -ENOMEM;
		goto out;
	}
164

165 166 167 168 169 170 171 172
	/*
	 * we record the send generation after doing the xmit acquire.
	 * if someone else manages to jump in and do some work, we'll use
	 * this to avoid a goto restart farther down.
	 *
	 * The acquire_in_xmit() check above ensures that only one
	 * caller can increment c_send_gen at any time.
	 */
173 174
	cp->cp_send_gen++;
	send_gen = cp->cp_send_gen;
175

176 177 178 179
	/*
	 * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT,
	 * we do the opposite to avoid races.
	 */
180 181
	if (!rds_conn_path_up(cp)) {
		release_in_xmit(cp);
182 183 184
		ret = 0;
		goto out;
	}
A
Andy Grover 已提交
185

186 187
	if (conn->c_trans->xmit_path_prepare)
		conn->c_trans->xmit_path_prepare(cp);
A
Andy Grover 已提交
188 189 190

	/*
	 * spin trying to push headers and data down the connection until
A
Andy Grover 已提交
191
	 * the connection doesn't make forward progress.
A
Andy Grover 已提交
192
	 */
193
	while (1) {
A
Andy Grover 已提交
194

195
		rm = cp->cp_xmit_rm;
A
Andy Grover 已提交
196

A
Andy Grover 已提交
197 198 199
		/*
		 * If between sending messages, we can send a pending congestion
		 * map update.
A
Andy Grover 已提交
200
		 */
201
		if (!rm && test_and_clear_bit(0, &conn->c_map_queued)) {
202 203 204 205
			rm = rds_cong_update_alloc(conn);
			if (IS_ERR(rm)) {
				ret = PTR_ERR(rm);
				break;
A
Andy Grover 已提交
206
			}
207
			rm->data.op_active = 1;
208 209
			rm->m_inc.i_conn_path = cp;
			rm->m_inc.i_conn = cp->cp_conn;
210

211
			cp->cp_xmit_rm = rm;
A
Andy Grover 已提交
212 213 214
		}

		/*
A
Andy Grover 已提交
215
		 * If not already working on one, grab the next message.
A
Andy Grover 已提交
216
		 *
217
		 * cp_xmit_rm holds a ref while we're sending this message down
A
Andy Grover 已提交
218 219 220
		 * the connction.  We can use this ref while holding the
		 * send_sem.. rds_send_reset() is serialized with it.
		 */
221
		if (!rm) {
A
Andy Grover 已提交
222 223
			unsigned int len;

224 225 226 227 228 229 230
			batch_count++;

			/* we want to process as big a batch as we can, but
			 * we also want to avoid softlockups.  If we've been
			 * through a lot of messages, lets back off and see
			 * if anyone else jumps in
			 */
231
			if (batch_count >= send_batch_count)
232 233
				goto over_batch;

234
			spin_lock_irqsave(&cp->cp_lock, flags);
A
Andy Grover 已提交
235

236 237
			if (!list_empty(&cp->cp_send_queue)) {
				rm = list_entry(cp->cp_send_queue.next,
A
Andy Grover 已提交
238 239 240 241 242 243 244 245
						struct rds_message,
						m_conn_item);
				rds_message_addref(rm);

				/*
				 * Move the message from the send queue to the retransmit
				 * list right away.
				 */
246 247
				list_move_tail(&rm->m_conn_item,
					       &cp->cp_retrans);
A
Andy Grover 已提交
248 249
			}

250
			spin_unlock_irqrestore(&cp->cp_lock, flags);
A
Andy Grover 已提交
251

252
			if (!rm)
A
Andy Grover 已提交
253 254 255 256 257 258 259 260 261
				break;

			/* Unfortunately, the way Infiniband deals with
			 * RDMA to a bad MR key is by moving the entire
			 * queue pair to error state. We cold possibly
			 * recover from that, but right now we drop the
			 * connection.
			 * Therefore, we never retransmit messages with RDMA ops.
			 */
262 263 264
			if (test_bit(RDS_MSG_FLUSH, &rm->m_flags) ||
			    (rm->rdma.op_active &&
			    test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags))) {
265
				spin_lock_irqsave(&cp->cp_lock, flags);
A
Andy Grover 已提交
266 267
				if (test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags))
					list_move(&rm->m_conn_item, &to_be_dropped);
268
				spin_unlock_irqrestore(&cp->cp_lock, flags);
A
Andy Grover 已提交
269 270 271 272 273
				continue;
			}

			/* Require an ACK every once in a while */
			len = ntohl(rm->m_inc.i_hdr.h_len);
274 275
			if (cp->cp_unacked_packets == 0 ||
			    cp->cp_unacked_bytes < len) {
A
Andy Grover 已提交
276 277
				__set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);

278 279 280 281
				cp->cp_unacked_packets =
					rds_sysctl_max_unacked_packets;
				cp->cp_unacked_bytes =
					rds_sysctl_max_unacked_bytes;
A
Andy Grover 已提交
282 283
				rds_stats_inc(s_send_ack_required);
			} else {
284 285
				cp->cp_unacked_bytes -= len;
				cp->cp_unacked_packets--;
A
Andy Grover 已提交
286 287
			}

288
			cp->cp_xmit_rm = rm;
A
Andy Grover 已提交
289 290
		}

291
		/* The transport either sends the whole rdma or none of it */
292
		if (rm->rdma.op_active && !cp->cp_xmit_rdma_sent) {
293
			rm->m_final_op = &rm->rdma;
294 295 296 297
			/* The transport owns the mapped memory for now.
			 * You can't unmap it while it's on the send queue
			 */
			set_bit(RDS_MSG_MAPPED, &rm->m_flags);
298
			ret = conn->c_trans->xmit_rdma(conn, &rm->rdma);
299 300 301
			if (ret) {
				clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
				wake_up_interruptible(&rm->m_flush_wait);
A
Andy Grover 已提交
302
				break;
303
			}
304
			cp->cp_xmit_rdma_sent = 1;
305

A
Andy Grover 已提交
306 307
		}

308
		if (rm->atomic.op_active && !cp->cp_xmit_atomic_sent) {
309
			rm->m_final_op = &rm->atomic;
310 311 312 313
			/* The transport owns the mapped memory for now.
			 * You can't unmap it while it's on the send queue
			 */
			set_bit(RDS_MSG_MAPPED, &rm->m_flags);
314
			ret = conn->c_trans->xmit_atomic(conn, &rm->atomic);
315 316 317
			if (ret) {
				clear_bit(RDS_MSG_MAPPED, &rm->m_flags);
				wake_up_interruptible(&rm->m_flush_wait);
A
Andy Grover 已提交
318
				break;
319
			}
320
			cp->cp_xmit_atomic_sent = 1;
321

A
Andy Grover 已提交
322 323
		}

324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345
		/*
		 * A number of cases require an RDS header to be sent
		 * even if there is no data.
		 * We permit 0-byte sends; rds-ping depends on this.
		 * However, if there are exclusively attached silent ops,
		 * we skip the hdr/data send, to enable silent operation.
		 */
		if (rm->data.op_nents == 0) {
			int ops_present;
			int all_ops_are_silent = 1;

			ops_present = (rm->atomic.op_active || rm->rdma.op_active);
			if (rm->atomic.op_active && !rm->atomic.op_silent)
				all_ops_are_silent = 0;
			if (rm->rdma.op_active && !rm->rdma.op_silent)
				all_ops_are_silent = 0;

			if (ops_present && all_ops_are_silent
			    && !rm->m_rdma_cookie)
				rm->data.op_active = 0;
		}

346
		if (rm->data.op_active && !cp->cp_xmit_data_sent) {
347
			rm->m_final_op = &rm->data;
348

A
Andy Grover 已提交
349
			ret = conn->c_trans->xmit(conn, rm,
350 351 352
						  cp->cp_xmit_hdr_off,
						  cp->cp_xmit_sg,
						  cp->cp_xmit_data_off);
A
Andy Grover 已提交
353 354 355
			if (ret <= 0)
				break;

356
			if (cp->cp_xmit_hdr_off < sizeof(struct rds_header)) {
A
Andy Grover 已提交
357 358
				tmp = min_t(int, ret,
					    sizeof(struct rds_header) -
359 360
					    cp->cp_xmit_hdr_off);
				cp->cp_xmit_hdr_off += tmp;
A
Andy Grover 已提交
361 362 363
				ret -= tmp;
			}

364
			sg = &rm->data.op_sg[cp->cp_xmit_sg];
A
Andy Grover 已提交
365 366
			while (ret) {
				tmp = min_t(int, ret, sg->length -
367 368
						      cp->cp_xmit_data_off);
				cp->cp_xmit_data_off += tmp;
A
Andy Grover 已提交
369
				ret -= tmp;
370 371
				if (cp->cp_xmit_data_off == sg->length) {
					cp->cp_xmit_data_off = 0;
A
Andy Grover 已提交
372
					sg++;
373 374 375
					cp->cp_xmit_sg++;
					BUG_ON(ret != 0 && cp->cp_xmit_sg ==
					       rm->data.op_nents);
A
Andy Grover 已提交
376 377
				}
			}
A
Andy Grover 已提交
378

379 380 381
			if (cp->cp_xmit_hdr_off == sizeof(struct rds_header) &&
			    (cp->cp_xmit_sg == rm->data.op_nents))
				cp->cp_xmit_data_sent = 1;
A
Andy Grover 已提交
382 383 384 385 386 387 388
		}

		/*
		 * A rm will only take multiple times through this loop
		 * if there is a data op. Thus, if the data is sent (or there was
		 * none), then we're done with the rm.
		 */
389 390 391 392 393 394 395 396
		if (!rm->data.op_active || cp->cp_xmit_data_sent) {
			cp->cp_xmit_rm = NULL;
			cp->cp_xmit_sg = 0;
			cp->cp_xmit_hdr_off = 0;
			cp->cp_xmit_data_off = 0;
			cp->cp_xmit_rdma_sent = 0;
			cp->cp_xmit_atomic_sent = 0;
			cp->cp_xmit_data_sent = 0;
A
Andy Grover 已提交
397 398

			rds_message_put(rm);
A
Andy Grover 已提交
399 400 401
		}
	}

402
over_batch:
403 404
	if (conn->c_trans->xmit_path_complete)
		conn->c_trans->xmit_path_complete(cp);
405
	release_in_xmit(cp);
A
Andy Grover 已提交
406

407 408 409 410 411 412 413 414
	/* Nuke any messages we decided not to retransmit. */
	if (!list_empty(&to_be_dropped)) {
		/* irqs on here, so we can put(), unlike above */
		list_for_each_entry(rm, &to_be_dropped, m_conn_item)
			rds_message_put(rm);
		rds_send_remove_from_sock(&to_be_dropped, RDS_RDMA_DROPPED);
	}

415
	/*
416 417 418 419 420
	 * Other senders can queue a message after we last test the send queue
	 * but before we clear RDS_IN_XMIT.  In that case they'd back off and
	 * not try and send their newly queued message.  We need to check the
	 * send queue after having cleared RDS_IN_XMIT so that their message
	 * doesn't get stuck on the send queue.
421 422 423 424
	 *
	 * If the transport cannot continue (i.e ret != 0), then it must
	 * call us when more room is available, such as from the tx
	 * completion handler.
425 426 427 428
	 *
	 * We have an extra generation check here so that if someone manages
	 * to jump in after our release_in_xmit, we'll see that they have done
	 * some work and we will skip our goto
429 430
	 */
	if (ret == 0) {
431
		smp_mb();
432
		if ((test_bit(0, &conn->c_map_queued) ||
433 434
		     !list_empty(&cp->cp_send_queue)) &&
		    send_gen == cp->cp_send_gen) {
435
			rds_stats_inc(s_send_lock_queue_raced);
436 437
			if (batch_count < send_batch_count)
				goto restart;
438
			queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
A
Andy Grover 已提交
439 440 441 442 443
		}
	}
out:
	return ret;
}
444
EXPORT_SYMBOL_GPL(rds_send_xmit);
A
Andy Grover 已提交
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475

static void rds_send_sndbuf_remove(struct rds_sock *rs, struct rds_message *rm)
{
	u32 len = be32_to_cpu(rm->m_inc.i_hdr.h_len);

	assert_spin_locked(&rs->rs_lock);

	BUG_ON(rs->rs_snd_bytes < len);
	rs->rs_snd_bytes -= len;

	if (rs->rs_snd_bytes == 0)
		rds_stats_inc(s_send_queue_empty);
}

static inline int rds_send_is_acked(struct rds_message *rm, u64 ack,
				    is_acked_func is_acked)
{
	if (is_acked)
		return is_acked(rm, ack);
	return be64_to_cpu(rm->m_inc.i_hdr.h_sequence) <= ack;
}

/*
 * This is pretty similar to what happens below in the ACK
 * handling code - except that we call here as soon as we get
 * the IB send completion on the RDMA op and the accompanying
 * message.
 */
void rds_rdma_send_complete(struct rds_message *rm, int status)
{
	struct rds_sock *rs = NULL;
A
Andy Grover 已提交
476
	struct rm_rdma_op *ro;
A
Andy Grover 已提交
477
	struct rds_notifier *notifier;
478
	unsigned long flags;
479
	unsigned int notify = 0;
A
Andy Grover 已提交
480

481
	spin_lock_irqsave(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
482

483
	notify =  rm->rdma.op_notify | rm->data.op_notify;
A
Andy Grover 已提交
484
	ro = &rm->rdma;
485
	if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags) &&
486
	    ro->op_active && notify && ro->op_notifier) {
A
Andy Grover 已提交
487
		notifier = ro->op_notifier;
A
Andy Grover 已提交
488 489 490 491 492 493 494 495
		rs = rm->m_rs;
		sock_hold(rds_rs_to_sk(rs));

		notifier->n_status = status;
		spin_lock(&rs->rs_lock);
		list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
		spin_unlock(&rs->rs_lock);

A
Andy Grover 已提交
496
		ro->op_notifier = NULL;
A
Andy Grover 已提交
497 498
	}

499
	spin_unlock_irqrestore(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
500 501 502 503 504 505

	if (rs) {
		rds_wake_sk_sleep(rs);
		sock_put(rds_rs_to_sk(rs));
	}
}
A
Andy Grover 已提交
506
EXPORT_SYMBOL_GPL(rds_rdma_send_complete);
A
Andy Grover 已提交
507

A
Andy Grover 已提交
508 509 510 511 512 513 514 515
/*
 * Just like above, except looks at atomic op
 */
void rds_atomic_send_complete(struct rds_message *rm, int status)
{
	struct rds_sock *rs = NULL;
	struct rm_atomic_op *ao;
	struct rds_notifier *notifier;
516
	unsigned long flags;
A
Andy Grover 已提交
517

518
	spin_lock_irqsave(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534

	ao = &rm->atomic;
	if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags)
	    && ao->op_active && ao->op_notify && ao->op_notifier) {
		notifier = ao->op_notifier;
		rs = rm->m_rs;
		sock_hold(rds_rs_to_sk(rs));

		notifier->n_status = status;
		spin_lock(&rs->rs_lock);
		list_add_tail(&notifier->n_list, &rs->rs_notify_queue);
		spin_unlock(&rs->rs_lock);

		ao->op_notifier = NULL;
	}

535
	spin_unlock_irqrestore(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
536 537 538 539 540 541 542 543

	if (rs) {
		rds_wake_sk_sleep(rs);
		sock_put(rds_rs_to_sk(rs));
	}
}
EXPORT_SYMBOL_GPL(rds_atomic_send_complete);

A
Andy Grover 已提交
544 545 546 547 548 549
/*
 * This is the same as rds_rdma_send_complete except we
 * don't do any locking - we have all the ingredients (message,
 * socket, socket lock) and can just move the notifier.
 */
static inline void
550
__rds_send_complete(struct rds_sock *rs, struct rds_message *rm, int status)
A
Andy Grover 已提交
551
{
A
Andy Grover 已提交
552
	struct rm_rdma_op *ro;
553
	struct rm_atomic_op *ao;
A
Andy Grover 已提交
554

A
Andy Grover 已提交
555 556 557 558 559
	ro = &rm->rdma;
	if (ro->op_active && ro->op_notify && ro->op_notifier) {
		ro->op_notifier->n_status = status;
		list_add_tail(&ro->op_notifier->n_list, &rs->rs_notify_queue);
		ro->op_notifier = NULL;
A
Andy Grover 已提交
560 561
	}

562 563 564 565 566 567 568
	ao = &rm->atomic;
	if (ao->op_active && ao->op_notify && ao->op_notifier) {
		ao->op_notifier->n_status = status;
		list_add_tail(&ao->op_notifier->n_list, &rs->rs_notify_queue);
		ao->op_notifier = NULL;
	}

A
Andy Grover 已提交
569 570 571 572 573 574 575 576 577 578 579
	/* No need to wake the app - caller does this */
}

/*
 * This removes messages from the socket's list if they're on it.  The list
 * argument must be private to the caller, we must be able to modify it
 * without locks.  The messages must have a reference held for their
 * position on the list.  This function will drop that reference after
 * removing the messages from the 'messages' list regardless of if it found
 * the messages on the socket list or not.
 */
580
static void rds_send_remove_from_sock(struct list_head *messages, int status)
A
Andy Grover 已提交
581
{
582
	unsigned long flags;
A
Andy Grover 已提交
583 584 585 586
	struct rds_sock *rs = NULL;
	struct rds_message *rm;

	while (!list_empty(messages)) {
587 588
		int was_on_sock = 0;

A
Andy Grover 已提交
589 590 591 592 593 594 595 596 597 598 599 600 601 602
		rm = list_entry(messages->next, struct rds_message,
				m_conn_item);
		list_del_init(&rm->m_conn_item);

		/*
		 * If we see this flag cleared then we're *sure* that someone
		 * else beat us to removing it from the sock.  If we race
		 * with their flag update we'll get the lock and then really
		 * see that the flag has been cleared.
		 *
		 * The message spinlock makes sure nobody clears rm->m_rs
		 * while we're messing with it. It does not prevent the
		 * message from being removed from the socket, though.
		 */
603
		spin_lock_irqsave(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
604 605 606 607 608 609 610 611 612
		if (!test_bit(RDS_MSG_ON_SOCK, &rm->m_flags))
			goto unlock_and_drop;

		if (rs != rm->m_rs) {
			if (rs) {
				rds_wake_sk_sleep(rs);
				sock_put(rds_rs_to_sk(rs));
			}
			rs = rm->m_rs;
613 614
			if (rs)
				sock_hold(rds_rs_to_sk(rs));
A
Andy Grover 已提交
615
		}
616 617
		if (!rs)
			goto unlock_and_drop;
T
Tina Yang 已提交
618
		spin_lock(&rs->rs_lock);
A
Andy Grover 已提交
619 620

		if (test_and_clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags)) {
A
Andy Grover 已提交
621
			struct rm_rdma_op *ro = &rm->rdma;
A
Andy Grover 已提交
622 623 624 625 626
			struct rds_notifier *notifier;

			list_del_init(&rm->m_sock_item);
			rds_send_sndbuf_remove(rs, rm);

A
Andy Grover 已提交
627 628 629
			if (ro->op_active && ro->op_notifier &&
			       (ro->op_notify || (ro->op_recverr && status))) {
				notifier = ro->op_notifier;
A
Andy Grover 已提交
630 631 632 633
				list_add_tail(&notifier->n_list,
						&rs->rs_notify_queue);
				if (!notifier->n_status)
					notifier->n_status = status;
A
Andy Grover 已提交
634
				rm->rdma.op_notifier = NULL;
A
Andy Grover 已提交
635
			}
636
			was_on_sock = 1;
A
Andy Grover 已提交
637 638
			rm->m_rs = NULL;
		}
T
Tina Yang 已提交
639
		spin_unlock(&rs->rs_lock);
A
Andy Grover 已提交
640 641

unlock_and_drop:
642
		spin_unlock_irqrestore(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
643
		rds_message_put(rm);
644 645
		if (was_on_sock)
			rds_message_put(rm);
A
Andy Grover 已提交
646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661
	}

	if (rs) {
		rds_wake_sk_sleep(rs);
		sock_put(rds_rs_to_sk(rs));
	}
}

/*
 * Transports call here when they've determined that the receiver queued
 * messages up to, and including, the given sequence number.  Messages are
 * moved to the retrans queue when rds_send_xmit picks them off the send
 * queue. This means that in the TCP case, the message may not have been
 * assigned the m_ack_seq yet - but that's fine as long as tcp_is_acked
 * checks the RDS_MSG_HAS_ACK_SEQ bit.
 */
662 663
void rds_send_path_drop_acked(struct rds_conn_path *cp, u64 ack,
			      is_acked_func is_acked)
A
Andy Grover 已提交
664 665 666 667 668
{
	struct rds_message *rm, *tmp;
	unsigned long flags;
	LIST_HEAD(list);

669
	spin_lock_irqsave(&cp->cp_lock, flags);
A
Andy Grover 已提交
670

671
	list_for_each_entry_safe(rm, tmp, &cp->cp_retrans, m_conn_item) {
A
Andy Grover 已提交
672 673 674 675 676 677 678 679 680
		if (!rds_send_is_acked(rm, ack, is_acked))
			break;

		list_move(&rm->m_conn_item, &list);
		clear_bit(RDS_MSG_ON_CONN, &rm->m_flags);
	}

	/* order flag updates with spin locks */
	if (!list_empty(&list))
681
		smp_mb__after_atomic();
A
Andy Grover 已提交
682

683
	spin_unlock_irqrestore(&cp->cp_lock, flags);
A
Andy Grover 已提交
684 685 686 687

	/* now remove the messages from the sock list as needed */
	rds_send_remove_from_sock(&list, RDS_RDMA_SUCCESS);
}
688 689 690 691 692 693 694 695
EXPORT_SYMBOL_GPL(rds_send_path_drop_acked);

void rds_send_drop_acked(struct rds_connection *conn, u64 ack,
			 is_acked_func is_acked)
{
	WARN_ON(conn->c_trans->t_mp_capable);
	rds_send_path_drop_acked(&conn->c_path[0], ack, is_acked);
}
A
Andy Grover 已提交
696
EXPORT_SYMBOL_GPL(rds_send_drop_acked);
A
Andy Grover 已提交
697 698 699 700 701

void rds_send_drop_to(struct rds_sock *rs, struct sockaddr_in *dest)
{
	struct rds_message *rm, *tmp;
	struct rds_connection *conn;
702
	struct rds_conn_path *cp;
703
	unsigned long flags;
A
Andy Grover 已提交
704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719
	LIST_HEAD(list);

	/* get all the messages we're dropping under the rs lock */
	spin_lock_irqsave(&rs->rs_lock, flags);

	list_for_each_entry_safe(rm, tmp, &rs->rs_send_queue, m_sock_item) {
		if (dest && (dest->sin_addr.s_addr != rm->m_daddr ||
			     dest->sin_port != rm->m_inc.i_hdr.h_dport))
			continue;

		list_move(&rm->m_sock_item, &list);
		rds_send_sndbuf_remove(rs, rm);
		clear_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
	}

	/* order flag updates with the rs lock */
720
	smp_mb__after_atomic();
A
Andy Grover 已提交
721 722 723

	spin_unlock_irqrestore(&rs->rs_lock, flags);

724 725
	if (list_empty(&list))
		return;
A
Andy Grover 已提交
726

727
	/* Remove the messages from the conn */
A
Andy Grover 已提交
728
	list_for_each_entry(rm, &list, m_sock_item) {
729 730

		conn = rm->m_inc.i_conn;
731 732 733 734
		if (conn->c_trans->t_mp_capable)
			cp = rm->m_inc.i_conn_path;
		else
			cp = &conn->c_path[0];
A
Andy Grover 已提交
735

736
		spin_lock_irqsave(&cp->cp_lock, flags);
A
Andy Grover 已提交
737
		/*
738 739 740
		 * Maybe someone else beat us to removing rm from the conn.
		 * If we race with their flag update we'll get the lock and
		 * then really see that the flag has been cleared.
A
Andy Grover 已提交
741
		 */
742
		if (!test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags)) {
743
			spin_unlock_irqrestore(&cp->cp_lock, flags);
744 745 746
			spin_lock_irqsave(&rm->m_rs_lock, flags);
			rm->m_rs = NULL;
			spin_unlock_irqrestore(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
747 748
			continue;
		}
749
		list_del_init(&rm->m_conn_item);
750
		spin_unlock_irqrestore(&cp->cp_lock, flags);
A
Andy Grover 已提交
751

752 753 754 755
		/*
		 * Couldn't grab m_rs_lock in top loop (lock ordering),
		 * but we can now.
		 */
756
		spin_lock_irqsave(&rm->m_rs_lock, flags);
A
Andy Grover 已提交
757

758
		spin_lock(&rs->rs_lock);
759
		__rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
760 761 762
		spin_unlock(&rs->rs_lock);

		rm->m_rs = NULL;
763
		spin_unlock_irqrestore(&rm->m_rs_lock, flags);
764 765 766

		rds_message_put(rm);
	}
A
Andy Grover 已提交
767

768
	rds_wake_sk_sleep(rs);
769

A
Andy Grover 已提交
770 771 772 773
	while (!list_empty(&list)) {
		rm = list_entry(list.next, struct rds_message, m_sock_item);
		list_del_init(&rm->m_sock_item);
		rds_message_wait(rm);
774 775 776 777 778 779 780 781 782 783 784 785 786 787 788

		/* just in case the code above skipped this message
		 * because RDS_MSG_ON_CONN wasn't set, run it again here
		 * taking m_rs_lock is the only thing that keeps us
		 * from racing with ack processing.
		 */
		spin_lock_irqsave(&rm->m_rs_lock, flags);

		spin_lock(&rs->rs_lock);
		__rds_send_complete(rs, rm, RDS_RDMA_CANCELED);
		spin_unlock(&rs->rs_lock);

		rm->m_rs = NULL;
		spin_unlock_irqrestore(&rm->m_rs_lock, flags);

A
Andy Grover 已提交
789 790 791 792 793 794 795 796 797 798
		rds_message_put(rm);
	}
}

/*
 * we only want this to fire once so we use the callers 'queued'.  It's
 * possible that another thread can race with us and remove the
 * message from the flow with RDS_CANCEL_SENT_TO.
 */
static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
799
			     struct rds_conn_path *cp,
A
Andy Grover 已提交
800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842
			     struct rds_message *rm, __be16 sport,
			     __be16 dport, int *queued)
{
	unsigned long flags;
	u32 len;

	if (*queued)
		goto out;

	len = be32_to_cpu(rm->m_inc.i_hdr.h_len);

	/* this is the only place which holds both the socket's rs_lock
	 * and the connection's c_lock */
	spin_lock_irqsave(&rs->rs_lock, flags);

	/*
	 * If there is a little space in sndbuf, we don't queue anything,
	 * and userspace gets -EAGAIN. But poll() indicates there's send
	 * room. This can lead to bad behavior (spinning) if snd_bytes isn't
	 * freed up by incoming acks. So we check the *old* value of
	 * rs_snd_bytes here to allow the last msg to exceed the buffer,
	 * and poll() now knows no more data can be sent.
	 */
	if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
		rs->rs_snd_bytes += len;

		/* let recv side know we are close to send space exhaustion.
		 * This is probably not the optimal way to do it, as this
		 * means we set the flag on *all* messages as soon as our
		 * throughput hits a certain threshold.
		 */
		if (rs->rs_snd_bytes >= rds_sk_sndbuf(rs) / 2)
			__set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);

		list_add_tail(&rm->m_sock_item, &rs->rs_send_queue);
		set_bit(RDS_MSG_ON_SOCK, &rm->m_flags);
		rds_message_addref(rm);
		rm->m_rs = rs;

		/* The code ordering is a little weird, but we're
		   trying to minimize the time we hold c_lock */
		rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
		rm->m_inc.i_conn = conn;
843
		rm->m_inc.i_conn_path = cp;
A
Andy Grover 已提交
844 845
		rds_message_addref(rm);

846 847 848
		spin_lock(&cp->cp_lock);
		rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
		list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
A
Andy Grover 已提交
849
		set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
850
		spin_unlock(&cp->cp_lock);
A
Andy Grover 已提交
851 852 853 854 855 856 857 858 859 860 861 862 863

		rdsdebug("queued msg %p len %d, rs %p bytes %d seq %llu\n",
			 rm, len, rs, rs->rs_snd_bytes,
			 (unsigned long long)be64_to_cpu(rm->m_inc.i_hdr.h_sequence));

		*queued = 1;
	}

	spin_unlock_irqrestore(&rs->rs_lock, flags);
out:
	return *queued;
}

864 865 866 867 868 869
/*
 * rds_message is getting to be quite complicated, and we'd like to allocate
 * it all in one go. This figures out how big it needs to be up front.
 */
static int rds_rm_size(struct msghdr *msg, int data_len)
{
870
	struct cmsghdr *cmsg;
871
	int size = 0;
872
	int cmsg_groups = 0;
873 874
	int retval;

875
	for_each_cmsghdr(cmsg, msg) {
876 877 878 879 880 881 882 883
		if (!CMSG_OK(msg, cmsg))
			return -EINVAL;

		if (cmsg->cmsg_level != SOL_RDS)
			continue;

		switch (cmsg->cmsg_type) {
		case RDS_CMSG_RDMA_ARGS:
884
			cmsg_groups |= 1;
885 886 887 888
			retval = rds_rdma_extra_size(CMSG_DATA(cmsg));
			if (retval < 0)
				return retval;
			size += retval;
889

890 891 892 893
			break;

		case RDS_CMSG_RDMA_DEST:
		case RDS_CMSG_RDMA_MAP:
894
			cmsg_groups |= 2;
895 896 897
			/* these are valid but do no add any size */
			break;

A
Andy Grover 已提交
898 899
		case RDS_CMSG_ATOMIC_CSWP:
		case RDS_CMSG_ATOMIC_FADD:
900 901
		case RDS_CMSG_MASKED_ATOMIC_CSWP:
		case RDS_CMSG_MASKED_ATOMIC_FADD:
902
			cmsg_groups |= 1;
A
Andy Grover 已提交
903 904 905
			size += sizeof(struct scatterlist);
			break;

906 907 908 909 910
		default:
			return -EINVAL;
		}

	}
911

912
	size += ceil(data_len, PAGE_SIZE) * sizeof(struct scatterlist);
913

914 915 916 917
	/* Ensure (DEST, MAP) are never used with (ARGS, ATOMIC) */
	if (cmsg_groups == 3)
		return -EINVAL;

918 919 920
	return size;
}

A
Andy Grover 已提交
921 922 923 924 925 926
static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
			 struct msghdr *msg, int *allocated_mr)
{
	struct cmsghdr *cmsg;
	int ret = 0;

927
	for_each_cmsghdr(cmsg, msg) {
A
Andy Grover 已提交
928 929 930 931 932 933 934
		if (!CMSG_OK(msg, cmsg))
			return -EINVAL;

		if (cmsg->cmsg_level != SOL_RDS)
			continue;

		/* As a side effect, RDMA_DEST and RDMA_MAP will set
A
Andy Grover 已提交
935
		 * rm->rdma.m_rdma_cookie and rm->rdma.m_rdma_mr.
A
Andy Grover 已提交
936 937 938 939 940 941 942 943 944 945 946 947 948 949
		 */
		switch (cmsg->cmsg_type) {
		case RDS_CMSG_RDMA_ARGS:
			ret = rds_cmsg_rdma_args(rs, rm, cmsg);
			break;

		case RDS_CMSG_RDMA_DEST:
			ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
			break;

		case RDS_CMSG_RDMA_MAP:
			ret = rds_cmsg_rdma_map(rs, rm, cmsg);
			if (!ret)
				*allocated_mr = 1;
950 951 952 953 954
			else if (ret == -ENODEV)
				/* Accommodate the get_mr() case which can fail
				 * if connection isn't established yet.
				 */
				ret = -EAGAIN;
A
Andy Grover 已提交
955
			break;
A
Andy Grover 已提交
956 957
		case RDS_CMSG_ATOMIC_CSWP:
		case RDS_CMSG_ATOMIC_FADD:
958 959
		case RDS_CMSG_MASKED_ATOMIC_CSWP:
		case RDS_CMSG_MASKED_ATOMIC_FADD:
A
Andy Grover 已提交
960 961
			ret = rds_cmsg_atomic(rs, rm, cmsg);
			break;
A
Andy Grover 已提交
962 963 964 965 966 967 968 969 970 971 972 973

		default:
			return -EINVAL;
		}

		if (ret)
			break;
	}

	return ret;
}

974 975 976 977 978 979 980 981 982
static int rds_send_mprds_hash(struct rds_sock *rs, struct rds_connection *conn)
{
	int hash;

	if (conn->c_npaths == 0)
		hash = RDS_MPATH_HASH(rs, RDS_MPATH_WORKERS);
	else
		hash = RDS_MPATH_HASH(rs, conn->c_npaths);
	if (conn->c_npaths == 0 && hash != 0) {
983
		rds_send_ping(conn, 0);
984 985 986 987 988 989 990 991 992 993 994

		if (conn->c_npaths == 0) {
			wait_event_interruptible(conn->c_hs_waitq,
						 (conn->c_npaths != 0));
		}
		if (conn->c_npaths == 1)
			hash = 0;
	}
	return hash;
}

995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014
static int rds_rdma_bytes(struct msghdr *msg, size_t *rdma_bytes)
{
	struct rds_rdma_args *args;
	struct cmsghdr *cmsg;

	for_each_cmsghdr(cmsg, msg) {
		if (!CMSG_OK(msg, cmsg))
			return -EINVAL;

		if (cmsg->cmsg_level != SOL_RDS)
			continue;

		if (cmsg->cmsg_type == RDS_CMSG_RDMA_ARGS) {
			args = CMSG_DATA(cmsg);
			*rdma_bytes += args->remote_vec.bytes;
		}
	}
	return 0;
}

1015
int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
A
Andy Grover 已提交
1016 1017 1018
{
	struct sock *sk = sock->sk;
	struct rds_sock *rs = rds_sk_to_rs(sk);
1019
	DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
A
Andy Grover 已提交
1020 1021 1022 1023 1024 1025 1026
	__be32 daddr;
	__be16 dport;
	struct rds_message *rm = NULL;
	struct rds_connection *conn;
	int ret = 0;
	int queued = 0, allocated_mr = 0;
	int nonblock = msg->msg_flags & MSG_DONTWAIT;
1027
	long timeo = sock_sndtimeo(sk, nonblock);
1028
	struct rds_conn_path *cpath;
1029
	size_t total_payload_len = payload_len, rdma_payload_len = 0;
A
Andy Grover 已提交
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053

	/* Mirror Linux UDP mirror of BSD error message compatibility */
	/* XXX: Perhaps MSG_MORE someday */
	if (msg->msg_flags & ~(MSG_DONTWAIT | MSG_CMSG_COMPAT)) {
		ret = -EOPNOTSUPP;
		goto out;
	}

	if (msg->msg_namelen) {
		/* XXX fail non-unicast destination IPs? */
		if (msg->msg_namelen < sizeof(*usin) || usin->sin_family != AF_INET) {
			ret = -EINVAL;
			goto out;
		}
		daddr = usin->sin_addr.s_addr;
		dport = usin->sin_port;
	} else {
		/* We only care about consistency with ->connect() */
		lock_sock(sk);
		daddr = rs->rs_conn_addr;
		dport = rs->rs_conn_port;
		release_sock(sk);
	}

1054
	lock_sock(sk);
A
Andy Grover 已提交
1055
	if (daddr == 0 || rs->rs_bound_addr == 0) {
1056
		release_sock(sk);
A
Andy Grover 已提交
1057 1058 1059
		ret = -ENOTCONN; /* XXX not a great errno */
		goto out;
	}
1060
	release_sock(sk);
A
Andy Grover 已提交
1061

1062 1063 1064 1065 1066 1067 1068 1069 1070 1071
	ret = rds_rdma_bytes(msg, &rdma_payload_len);
	if (ret)
		goto out;

	total_payload_len += rdma_payload_len;
	if (max_t(size_t, payload_len, rdma_payload_len) > RDS_MAX_MSG_SIZE) {
		ret = -EMSGSIZE;
		goto out;
	}

1072 1073 1074 1075 1076
	if (payload_len > rds_sk_sndbuf(rs)) {
		ret = -EMSGSIZE;
		goto out;
	}

1077 1078 1079 1080 1081 1082 1083 1084
	/* size of rm including all sgs */
	ret = rds_rm_size(msg, payload_len);
	if (ret < 0)
		goto out;

	rm = rds_message_alloc(ret, GFP_KERNEL);
	if (!rm) {
		ret = -ENOMEM;
A
Andy Grover 已提交
1085 1086 1087
		goto out;
	}

1088 1089 1090
	/* Attach data to the rm */
	if (payload_len) {
		rm->data.op_sg = rds_message_alloc_sgs(rm, ceil(payload_len, PAGE_SIZE));
1091 1092 1093 1094
		if (!rm->data.op_sg) {
			ret = -ENOMEM;
			goto out;
		}
A
Al Viro 已提交
1095
		ret = rds_message_copy_from_user(rm, &msg->msg_iter);
1096 1097 1098 1099
		if (ret)
			goto out;
	}
	rm->data.op_active = 1;
1100

A
Andy Grover 已提交
1101 1102 1103 1104 1105 1106 1107
	rm->m_daddr = daddr;

	/* rds_conn_create has a spinlock that runs with IRQ off.
	 * Caching the conn in the socket helps a lot. */
	if (rs->rs_conn && rs->rs_conn->c_faddr == daddr)
		conn = rs->rs_conn;
	else {
1108 1109
		conn = rds_conn_create_outgoing(sock_net(sock->sk),
						rs->rs_bound_addr, daddr,
A
Andy Grover 已提交
1110 1111 1112 1113 1114 1115 1116 1117 1118
					rs->rs_transport,
					sock->sk->sk_allocation);
		if (IS_ERR(conn)) {
			ret = PTR_ERR(conn);
			goto out;
		}
		rs->rs_conn = conn;
	}

1119 1120
	/* Parse any control messages the user may have included. */
	ret = rds_cmsg_send(rs, rm, msg, &allocated_mr);
1121 1122 1123 1124
	if (ret) {
		/* Trigger connection so that its ready for the next retry */
		if (ret ==  -EAGAIN)
			rds_conn_connect_if_down(conn);
1125
		goto out;
1126
	}
1127

1128
	if (rm->rdma.op_active && !conn->c_trans->xmit_rdma) {
1129
		printk_ratelimited(KERN_NOTICE "rdma_op %p conn xmit_rdma %p\n",
A
Andy Grover 已提交
1130
			       &rm->rdma, conn->c_trans->xmit_rdma);
A
Andy Grover 已提交
1131 1132 1133 1134 1135
		ret = -EOPNOTSUPP;
		goto out;
	}

	if (rm->atomic.op_active && !conn->c_trans->xmit_atomic) {
1136
		printk_ratelimited(KERN_NOTICE "atomic_op %p conn xmit_atomic %p\n",
A
Andy Grover 已提交
1137
			       &rm->atomic, conn->c_trans->xmit_atomic);
A
Andy Grover 已提交
1138 1139 1140 1141
		ret = -EOPNOTSUPP;
		goto out;
	}

1142 1143 1144 1145
	if (conn->c_trans->t_mp_capable)
		cpath = &conn->c_path[rds_send_mprds_hash(rs, conn)];
	else
		cpath = &conn->c_path[0];
1146 1147

	rds_conn_path_connect_if_down(cpath);
A
Andy Grover 已提交
1148 1149

	ret = rds_cong_wait(conn->c_fcong, dport, nonblock, rs);
1150 1151
	if (ret) {
		rs->rs_seen_congestion = 1;
A
Andy Grover 已提交
1152
		goto out;
1153
	}
1154
	while (!rds_send_queue_rm(rs, conn, cpath, rm, rs->rs_bound_port,
A
Andy Grover 已提交
1155 1156
				  dport, &queued)) {
		rds_stats_inc(s_send_queue_full);
1157

A
Andy Grover 已提交
1158 1159 1160 1161 1162
		if (nonblock) {
			ret = -EAGAIN;
			goto out;
		}

E
Eric Dumazet 已提交
1163
		timeo = wait_event_interruptible_timeout(*sk_sleep(sk),
1164
					rds_send_queue_rm(rs, conn, cpath, rm,
A
Andy Grover 已提交
1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
							  rs->rs_bound_port,
							  dport,
							  &queued),
					timeo);
		rdsdebug("sendmsg woke queued %d timeo %ld\n", queued, timeo);
		if (timeo > 0 || timeo == MAX_SCHEDULE_TIMEOUT)
			continue;

		ret = timeo;
		if (ret == 0)
			ret = -ETIMEDOUT;
		goto out;
	}

	/*
	 * By now we've committed to the send.  We reuse rds_send_worker()
	 * to retry sends in the rds thread if the transport asks us to.
	 */
	rds_stats_inc(s_send_queued);

1185
	ret = rds_send_xmit(cpath);
1186
	if (ret == -ENOMEM || ret == -EAGAIN)
1187
		queue_delayed_work(rds_wq, &cpath->cp_send_w, 1);
A
Andy Grover 已提交
1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204

	rds_message_put(rm);
	return payload_len;

out:
	/* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
	 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
	 * or in any other way, we need to destroy the MR again */
	if (allocated_mr)
		rds_rdma_unuse(rs, rds_rdma_cookie_key(rm->m_rdma_cookie), 1);

	if (rm)
		rds_message_put(rm);
	return ret;
}

/*
1205 1206 1207 1208 1209 1210
 * send out a probe. Can be shared by rds_send_ping,
 * rds_send_pong, rds_send_hb.
 * rds_send_hb should use h_flags
 *   RDS_FLAG_HB_PING|RDS_FLAG_ACK_REQUIRED
 * or
 *   RDS_FLAG_HB_PONG|RDS_FLAG_ACK_REQUIRED
A
Andy Grover 已提交
1211
 */
1212
static int
1213 1214
rds_send_probe(struct rds_conn_path *cp, __be16 sport,
	       __be16 dport, u8 h_flags)
A
Andy Grover 已提交
1215 1216 1217 1218 1219 1220
{
	struct rds_message *rm;
	unsigned long flags;
	int ret = 0;

	rm = rds_message_alloc(0, GFP_ATOMIC);
1221
	if (!rm) {
A
Andy Grover 已提交
1222 1223 1224 1225
		ret = -ENOMEM;
		goto out;
	}

1226
	rm->m_daddr = cp->cp_conn->c_faddr;
A
Andy Grover 已提交
1227
	rm->data.op_active = 1;
A
Andy Grover 已提交
1228

1229
	rds_conn_path_connect_if_down(cp);
A
Andy Grover 已提交
1230

1231
	ret = rds_cong_wait(cp->cp_conn->c_fcong, dport, 1, NULL);
A
Andy Grover 已提交
1232 1233 1234
	if (ret)
		goto out;

1235 1236
	spin_lock_irqsave(&cp->cp_lock, flags);
	list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
A
Andy Grover 已提交
1237 1238
	set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
	rds_message_addref(rm);
1239 1240
	rm->m_inc.i_conn = cp->cp_conn;
	rm->m_inc.i_conn_path = cp;
A
Andy Grover 已提交
1241

1242
	rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport,
1243
				    cp->cp_next_tx_seq);
1244
	rm->m_inc.i_hdr.h_flags |= h_flags;
1245
	cp->cp_next_tx_seq++;
1246

1247 1248 1249 1250
	if (RDS_HS_PROBE(be16_to_cpu(sport), be16_to_cpu(dport)) &&
	    cp->cp_conn->c_trans->t_mp_capable) {
		u16 npaths = cpu_to_be16(RDS_MPATH_WORKERS);
		u32 my_gen_num = cpu_to_be32(cp->cp_conn->c_my_gen_num);
1251 1252 1253 1254

		rds_message_add_extension(&rm->m_inc.i_hdr,
					  RDS_EXTHDR_NPATHS, &npaths,
					  sizeof(npaths));
1255 1256
		rds_message_add_extension(&rm->m_inc.i_hdr,
					  RDS_EXTHDR_GEN_NUM,
1257
					  &my_gen_num,
1258
					  sizeof(u32));
1259
	}
1260
	spin_unlock_irqrestore(&cp->cp_lock, flags);
A
Andy Grover 已提交
1261 1262 1263 1264

	rds_stats_inc(s_send_queued);
	rds_stats_inc(s_send_pong);

1265
	/* schedule the send work on rds_wq */
1266
	queue_delayed_work(rds_wq, &cp->cp_send_w, 1);
A
Andy Grover 已提交
1267

A
Andy Grover 已提交
1268 1269 1270 1271 1272 1273 1274 1275
	rds_message_put(rm);
	return 0;

out:
	if (rm)
		rds_message_put(rm);
	return ret;
}
1276 1277 1278 1279 1280 1281 1282

int
rds_send_pong(struct rds_conn_path *cp, __be16 dport)
{
	return rds_send_probe(cp, 0, dport, 0);
}

1283 1284
void
rds_send_ping(struct rds_connection *conn, int cp_index)
1285 1286
{
	unsigned long flags;
1287
	struct rds_conn_path *cp = &conn->c_path[cp_index];
1288 1289 1290 1291 1292 1293 1294 1295

	spin_lock_irqsave(&cp->cp_lock, flags);
	if (conn->c_ping_triggered) {
		spin_unlock_irqrestore(&cp->cp_lock, flags);
		return;
	}
	conn->c_ping_triggered = 1;
	spin_unlock_irqrestore(&cp->cp_lock, flags);
1296
	rds_send_probe(cp, cpu_to_be16(RDS_FLAG_PROBE_PORT), 0, 0);
1297
}
1298
EXPORT_SYMBOL_GPL(rds_send_ping);