/* * 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>#include <linux/moduleparam.h>#include <linux/gfp.h>#include <net/sock.h>#include <linux/in.h>#include <linux/list.h>#include <linux/ratelimit.h>#include <linux/export.h> #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). */static int send_batch_count = 64;module_param(send_batch_count, int, 0444);MODULE_PARM_DESC(send_batch_count, " batch factor when working the send queue"); static void rds_send_remove_from_sock(struct list_head *messages, int status); /* * Reset the send state.  Callers must ensure that this doesn't race with * rds_send_xmit(). */void rds_send_reset(struct rds_connection *conn){	struct rds_message *rm, *tmp;	unsigned long flags; 	if (conn->c_xmit_rm) {		rm = conn->c_xmit_rm;		conn->c_xmit_rm = NULL;		/* 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 */		rds_message_unmapped(rm);		rds_message_put(rm);	} 	conn->c_xmit_sg = 0;	conn->c_xmit_hdr_off = 0;	conn->c_xmit_data_off = 0;	conn->c_xmit_atomic_sent = 0;	conn->c_xmit_rdma_sent = 0;	conn->c_xmit_data_sent = 0; 	conn->c_map_queued = 0; 	conn->c_unacked_packets = rds_sysctl_max_unacked_packets;	conn->c_unacked_bytes = rds_sysctl_max_unacked_bytes; 	/* Mark messages as retransmissions, and move them to the send q */	spin_lock_irqsave(&conn->c_lock, flags);	list_for_each_entry_safe(rm, tmp, &conn->c_retrans, m_conn_item) {		set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags);		set_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags);	}	list_splice_init(&conn->c_retrans, &conn->c_send_queue);	spin_unlock_irqrestore(&conn->c_lock, flags);} static int acquire_in_xmit(struct rds_connection *conn){	return test_and_set_bit(RDS_IN_XMIT, &conn->c_flags) == 0;} static void release_in_xmit(struct rds_connection *conn){	clear_bit(RDS_IN_XMIT, &conn->c_flags);	smp_mb__after_clear_bit();	/*	 * 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.	 */	if (waitqueue_active(&conn->c_waitq))		wake_up_all(&conn->c_waitq);} /* * We're making the conscious trade-off here to only send one message * 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 */int rds_send_xmit(struct rds_connection *conn){	struct rds_message *rm;	unsigned long flags;	unsigned int tmp;	struct scatterlist *sg;	int ret = 0;	LIST_HEAD(to_be_dropped); restart: 	/*	 * 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.	 */	if (!acquire_in_xmit(conn)) {		rds_stats_inc(s_send_lock_contention);		ret = -ENOMEM;		goto out;	} 	/*	 * rds_conn_shutdown() sets the conn state and then tests RDS_IN_XMIT,	 * we do the opposite to avoid races.	 */	if (!rds_conn_up(conn)) {		release_in_xmit(conn);		ret = 0;		goto out;	} 	if (conn->c_trans->xmit_prepare)		conn->c_trans->xmit_prepare(conn); 	/*	 * spin trying to push headers and data down the connection until	 * the connection doesn't make forward progress.	 */	while (1) { 		rm = conn->c_xmit_rm; 		/*		 * If between sending messages, we can send a pending congestion		 * map update.		 */		if (!rm && test_and_clear_bit(0, &conn->c_map_queued)) {			rm = rds_cong_update_alloc(conn);			if (IS_ERR(rm)) {				ret = PTR_ERR(rm);				break;			}			rm->data.op_active = 1; 			conn->c_xmit_rm = rm;		} 		/*		 * If not already working on one, grab the next message.		 *		 * c_xmit_rm holds a ref while we're sending this message down		 * the connction.  We can use this ref while holding the		 * send_sem.. rds_send_reset() is serialized with it.		 */		if (!rm) {			unsigned int len; 			spin_lock_irqsave(&conn->c_lock, flags); 			if (!list_empty(&conn->c_send_queue)) {				rm = list_entry(conn->c_send_queue.next,						struct rds_message,						m_conn_item);				rds_message_addref(rm); 				/*				 * Move the message from the send queue to the retransmit				 * list right away.				 */				list_move_tail(&rm->m_conn_item, &conn->c_retrans);			} 			spin_unlock_irqrestore(&conn->c_lock, flags); 			if (!rm)				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.			 */			if (rm->rdma.op_active &&			    test_bit(RDS_MSG_RETRANSMITTED, &rm->m_flags)) {				spin_lock_irqsave(&conn->c_lock, flags);				if (test_and_clear_bit(RDS_MSG_ON_CONN, &rm->m_flags))					list_move(&rm->m_conn_item, &to_be_dropped);				spin_unlock_irqrestore(&conn->c_lock, flags);				continue;			} 			/* Require an ACK every once in a while */			len = ntohl(rm->m_inc.i_hdr.h_len);			if (conn->c_unacked_packets == 0 ||			    conn->c_unacked_bytes < len) {				__set_bit(RDS_MSG_ACK_REQUIRED, &rm->m_flags); 				conn->c_unacked_packets = rds_sysctl_max_unacked_packets;				conn->c_unacked_bytes = rds_sysctl_max_unacked_bytes;				rds_stats_inc(s_send_ack_required);			} else {				conn->c_unacked_bytes -= len;				conn->c_unacked_packets--;			} 			conn->c_xmit_rm = rm;		} 		/* The transport either sends the whole rdma or none of it */		if (rm->rdma.op_active && !conn->c_xmit_rdma_sent) {			rm->m_final_op = &rm->rdma;			ret = conn->c_trans->xmit_rdma(conn, &rm->rdma);			if (ret)				break;			conn->c_xmit_rdma_sent = 1; 			/* 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);		} 		if (rm->atomic.op_active && !conn->c_xmit_atomic_sent) {			rm->m_final_op = &rm->atomic;			ret = conn->c_trans->xmit_atomic(conn, &rm->atomic);			if (ret)				break;			conn->c_xmit_atomic_sent = 1; 			/* 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);		} 		/*		 * 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;		} 		if (rm->data.op_active && !conn->c_xmit_data_sent) {			rm->m_final_op = &rm->data;			ret = conn->c_trans->xmit(conn, rm,						  conn->c_xmit_hdr_off,						  conn->c_xmit_sg,						  conn->c_xmit_data_off);			if (ret <= 0)				break; 			if (conn->c_xmit_hdr_off < sizeof(struct rds_header)) {				tmp = min_t(int, ret,					    sizeof(struct rds_header) -					    conn->c_xmit_hdr_off);				conn->c_xmit_hdr_off += tmp;				ret -= tmp;			} 			sg = &rm->data.op_sg[conn->c_xmit_sg];			while (ret) {				tmp = min_t(int, ret, sg->length -						      conn->c_xmit_data_off);				conn->c_xmit_data_off += tmp;				ret -= tmp;				if (conn->c_xmit_data_off == sg->length) {					conn->c_xmit_data_off = 0;					sg++;					conn->c_xmit_sg++;					BUG_ON(ret != 0 &&					       conn->c_xmit_sg == rm->data.op_nents);				}			} 			if (conn->c_xmit_hdr_off == sizeof(struct rds_header) &&			    (conn->c_xmit_sg == rm->data.op_nents))				conn->c_xmit_data_sent = 1;		} 		/*		 * 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.		 */		if (!rm->data.op_active || conn->c_xmit_data_sent) {			conn->c_xmit_rm = NULL;			conn->c_xmit_sg = 0;			conn->c_xmit_hdr_off = 0;			conn->c_xmit_data_off = 0;			conn->c_xmit_rdma_sent = 0;			conn->c_xmit_atomic_sent = 0;			conn->c_xmit_data_sent = 0; 			rds_message_put(rm);		}	} 	if (conn->c_trans->xmit_complete)		conn->c_trans->xmit_complete(conn); 	release_in_xmit(conn); 	/* 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);	} 	/*	 * 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.	 *	 * 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.	 */	if (ret == 0) {		smp_mb();		if (!list_empty(&conn->c_send_queue)) {			rds_stats_inc(s_send_lock_queue_raced);			goto restart;		}	}out:	return ret;} 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;	struct rm_rdma_op *ro;	struct rds_notifier *notifier;	unsigned long flags; 	spin_lock_irqsave(&rm->m_rs_lock, flags); 	ro = &rm->rdma;	if (test_bit(RDS_MSG_ON_SOCK, &rm->m_flags) &&	    ro->op_active && ro->op_notify && ro->op_notifier) {		notifier = ro->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); 		ro->op_notifier = NULL;	} 	spin_unlock_irqrestore(&rm->m_rs_lock, flags); 	if (rs) {		rds_wake_sk_sleep(rs);		sock_put(rds_rs_to_sk(rs));	}}EXPORT_SYMBOL_GPL(rds_rdma_send_complete); /* * 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;	unsigned long flags; 	spin_lock_irqsave(&rm->m_rs_lock, flags); 	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;	} 	spin_unlock_irqrestore(&rm->m_rs_lock, flags); 	if (rs) {		rds_wake_sk_sleep(rs);		sock_put(rds_rs_to_sk(rs));	}}EXPORT_SYMBOL_GPL(rds_atomic_send_complete); /* * 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__rds_send_complete(struct rds_sock *rs, struct rds_message *rm, int status){	struct rm_rdma_op *ro;	struct rm_atomic_op *ao;

Lines 1–500 of 1134. Load more