[linux-2.6-block.git] / net / smc / smc_rx.c

// SPDX-License-Identifier: GPL-2.0
/*
 * Shared Memory Communications over RDMA (SMC-R) and RoCE
 *
 * Manage RMBE
 * copy new RMBE data into user space
 *
 * Copyright IBM Corp. 2016
 *
 * Author(s):  Ursula Braun <ubraun@linux.vnet.ibm.com>
 */

#include <linux/net.h>
#include <linux/rcupdate.h>
#include <linux/sched/signal.h>

#include <net/sock.h>

#include "smc.h"
#include "smc_core.h"
#include "smc_cdc.h"
#include "smc_tx.h" /* smc_tx_consumer_update() */
#include "smc_rx.h"

/* callback implementation for sk.sk_data_ready()
 * to wakeup rcvbuf consumers that blocked with smc_rx_wait_data().
 * indirectly called by smc_cdc_msg_recv_action().
 */
static void smc_rx_data_ready(struct sock *sk)
{
	struct socket_wq *wq;

	/* derived from sock_def_readable() */
	/* called already in smc_listen_work() */
	rcu_read_lock();
	wq = rcu_dereference(sk->sk_wq);
	if (skwq_has_sleeper(wq))
		wake_up_interruptible_sync_poll(&wq->wait, EPOLLIN | EPOLLPRI |
						EPOLLRDNORM | EPOLLRDBAND);
	sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
	if ((sk->sk_shutdown == SHUTDOWN_MASK) ||
	    (sk->sk_state == SMC_CLOSED))
		sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_HUP);
	rcu_read_unlock();
}

/* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
 *   @smc    smc socket
 *   @timeo  pointer to max seconds to wait, pointer to value 0 for no timeout
 * Returns:
 * 1 if at least 1 byte available in rcvbuf or if socket error/shutdown.
 * 0 otherwise (nothing in rcvbuf nor timeout, e.g. interrupted).
 */
static int smc_rx_wait_data(struct smc_sock *smc, long *timeo)
{
	DEFINE_WAIT_FUNC(wait, woken_wake_function);
	struct smc_connection *conn = &smc->conn;
	struct sock *sk = &smc->sk;
	int rc;

	if (atomic_read(&conn->bytes_to_rcv))
		return 1;
	sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
	add_wait_queue(sk_sleep(sk), &wait);
	rc = sk_wait_event(sk, timeo,
			   sk->sk_err ||
			   sk->sk_shutdown & RCV_SHUTDOWN ||
			   atomic_read(&conn->bytes_to_rcv) ||
			   smc_cdc_rxed_any_close_or_senddone(conn),
			   &wait);
	remove_wait_queue(sk_sleep(sk), &wait);
	sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
	return rc;
}

/* rcvbuf consumer: main API called by socket layer.
 * called under sk lock.
 */
int smc_rx_recvmsg(struct smc_sock *smc, struct msghdr *msg, size_t len,
		   int flags)
{
	size_t copylen, read_done = 0, read_remaining = len;
	size_t chunk_len, chunk_off, chunk_len_sum;
	struct smc_connection *conn = &smc->conn;
	union smc_host_cursor cons;
	int readable, chunk;
	char *rcvbuf_base;
	struct sock *sk;
	long timeo;
	int target;		/* Read at least these many bytes */
	int rc;

	if (unlikely(flags & MSG_ERRQUEUE))
		return -EINVAL; /* future work for sk.sk_family == AF_SMC */
	if (flags & MSG_OOB)
		return -EINVAL; /* future work */

	sk = &smc->sk;
	if (sk->sk_state == SMC_LISTEN)
		return -ENOTCONN;
	timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
	target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);

	msg->msg_namelen = 0;
	/* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
	rcvbuf_base = conn->rmb_desc->cpu_addr;

	do { /* while (read_remaining) */
		if (read_done >= target)
			break;

		if (atomic_read(&conn->bytes_to_rcv))
			goto copy;

		if (read_done) {
			if (sk->sk_err ||
			    sk->sk_state == SMC_CLOSED ||
			    sk->sk_shutdown & RCV_SHUTDOWN ||
			    !timeo ||
			    signal_pending(current) ||
			    smc_cdc_rxed_any_close_or_senddone(conn) ||
			    conn->local_tx_ctrl.conn_state_flags.
			    peer_conn_abort)
				break;
		} else {
			if (sk->sk_err) {
				read_done = sock_error(sk);
				break;
			}
			if (sk->sk_shutdown & RCV_SHUTDOWN ||
			    smc_cdc_rxed_any_close_or_senddone(conn) ||
			    conn->local_tx_ctrl.conn_state_flags.
			    peer_conn_abort)
				break;
			if (sk->sk_state == SMC_CLOSED) {
				if (!sock_flag(sk, SOCK_DONE)) {
					/* This occurs when user tries to read
					 * from never connected socket.
					 */
					read_done = -ENOTCONN;
					break;
				}
				break;
			}
			if (signal_pending(current)) {
				read_done = sock_intr_errno(timeo);
				break;
			}
			if (!timeo)
				return -EAGAIN;
		}

		if (!atomic_read(&conn->bytes_to_rcv)) {
			smc_rx_wait_data(smc, &timeo);
			continue;
		}

copy:
		/* initialize variables for 1st iteration of subsequent loop */
		/* could be just 1 byte, even after smc_rx_wait_data above */
		readable = atomic_read(&conn->bytes_to_rcv);
		/* not more than what user space asked for */
		copylen = min_t(size_t, read_remaining, readable);
		smc_curs_write(&cons,
			       smc_curs_read(&conn->local_tx_ctrl.cons, conn),
			       conn);
		/* determine chunks where to read from rcvbuf */
		/* either unwrapped case, or 1st chunk of wrapped case */
		chunk_len = min_t(size_t,
				  copylen, conn->rmbe_size - cons.count);
		chunk_len_sum = chunk_len;
		chunk_off = cons.count;
		smc_rmb_sync_sg_for_cpu(conn);
		for (chunk = 0; chunk < 2; chunk++) {
			if (!(flags & MSG_TRUNC)) {
				rc = memcpy_to_msg(msg, rcvbuf_base + chunk_off,
						   chunk_len);
				if (rc) {
					if (!read_done)
						read_done = -EFAULT;
					smc_rmb_sync_sg_for_device(conn);
					goto out;
				}
			}
			read_remaining -= chunk_len;
			read_done += chunk_len;

			if (chunk_len_sum == copylen)
				break; /* either on 1st or 2nd iteration */
			/* prepare next (== 2nd) iteration */
			chunk_len = copylen - chunk_len; /* remainder */
			chunk_len_sum += chunk_len;
			chunk_off = 0; /* modulo offset in recv ring buffer */
		}
		smc_rmb_sync_sg_for_device(conn);

		/* update cursors */
		if (!(flags & MSG_PEEK)) {
			smc_curs_add(conn->rmbe_size, &cons, copylen);
			/* increased in recv tasklet smc_cdc_msg_rcv() */
			smp_mb__before_atomic();
			atomic_sub(copylen, &conn->bytes_to_rcv);
			/* guarantee 0 <= bytes_to_rcv <= rmbe_size */
			smp_mb__after_atomic();
			smc_curs_write(&conn->local_tx_ctrl.cons,
				       smc_curs_read(&cons, conn),
				       conn);
			/* send consumer cursor update if required */
			/* similar to advertising new TCP rcv_wnd if required */
			smc_tx_consumer_update(conn);
		}
	} while (read_remaining);
out:
	return read_done;
}

/* Initialize receive properties on connection establishment. NB: not __init! */
void smc_rx_init(struct smc_sock *smc)
{
	smc->sk.sk_data_ready = smc_rx_data_ready;
}
Commit	Line	Data
b2441318	1	// SPDX-License-Identifier: GPL-2.0
952310cc UB	2	/*
	3	* Shared Memory Communications over RDMA (SMC-R) and RoCE
	4	*
	5	* Manage RMBE
	6	* copy new RMBE data into user space
	7	*
	8	* Copyright IBM Corp. 2016
	9	*
	10	* Author(s): Ursula Braun <ubraun@linux.vnet.ibm.com>
	11	*/
	12
	13	#include <linux/net.h>
	14	#include <linux/rcupdate.h>
c3edc401 IM	15	#include <linux/sched/signal.h>
c3edc401 IM	16
952310cc UB	17	#include <net/sock.h>
	18
	19	#include "smc.h"
	20	#include "smc_core.h"
	21	#include "smc_cdc.h"
	22	#include "smc_tx.h" /* smc_tx_consumer_update() */
	23	#include "smc_rx.h"
	24
	25	/* callback implementation for sk.sk_data_ready()
	26	* to wakeup rcvbuf consumers that blocked with smc_rx_wait_data().
	27	* indirectly called by smc_cdc_msg_recv_action().
	28	*/
	29	static void smc_rx_data_ready(struct sock *sk)
	30	{
	31	struct socket_wq *wq;
	32
	33	/* derived from sock_def_readable() */
	34	/* called already in smc_listen_work() */
	35	rcu_read_lock();
	36	wq = rcu_dereference(sk->sk_wq);
	37	if (skwq_has_sleeper(wq))
a9a08845 LT	38	wake_up_interruptible_sync_poll(&wq->wait, EPOLLIN \| EPOLLPRI \|
a9a08845 LT	39	EPOLLRDNORM \| EPOLLRDBAND);
90e9517e	40	sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
952310cc UB	41	if ((sk->sk_shutdown == SHUTDOWN_MASK) \|\|
	42	(sk->sk_state == SMC_CLOSED))
	43	sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_HUP);
952310cc UB	44	rcu_read_unlock();
	45	}
	46
	47	/* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
	48	* @smc smc socket
	49	* @timeo pointer to max seconds to wait, pointer to value 0 for no timeout
	50	* Returns:
	51	* 1 if at least 1 byte available in rcvbuf or if socket error/shutdown.
	52	* 0 otherwise (nothing in rcvbuf nor timeout, e.g. interrupted).
	53	*/
	54	static int smc_rx_wait_data(struct smc_sock smc, long timeo)
	55	{
	56	DEFINE_WAIT_FUNC(wait, woken_wake_function);
	57	struct smc_connection *conn = &smc->conn;
	58	struct sock *sk = &smc->sk;
	59	int rc;
	60
	61	if (atomic_read(&conn->bytes_to_rcv))
	62	return 1;
	63	sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
	64	add_wait_queue(sk_sleep(sk), &wait);
	65	rc = sk_wait_event(sk, timeo,
	66	sk->sk_err \|\|
	67	sk->sk_shutdown & RCV_SHUTDOWN \|\|
952310cc UB	68	atomic_read(&conn->bytes_to_rcv) \|\|
	69	smc_cdc_rxed_any_close_or_senddone(conn),
	70	&wait);
	71	remove_wait_queue(sk_sleep(sk), &wait);
	72	sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
	73	return rc;
	74	}
	75
	76	/* rcvbuf consumer: main API called by socket layer.
	77	* called under sk lock.
	78	*/
	79	int smc_rx_recvmsg(struct smc_sock smc, struct msghdr msg, size_t len,
	80	int flags)
	81	{
	82	size_t copylen, read_done = 0, read_remaining = len;
	83	size_t chunk_len, chunk_off, chunk_len_sum;
	84	struct smc_connection *conn = &smc->conn;
	85	union smc_host_cursor cons;
	86	int readable, chunk;
	87	char *rcvbuf_base;
	88	struct sock *sk;
	89	long timeo;
	90	int target; /* Read at least these many bytes */
	91	int rc;
	92
	93	if (unlikely(flags & MSG_ERRQUEUE))
	94	return -EINVAL; /* future work for sk.sk_family == AF_SMC */
	95	if (flags & MSG_OOB)
	96	return -EINVAL; /* future work */
	97
	98	sk = &smc->sk;
	99	if (sk->sk_state == SMC_LISTEN)
	100	return -ENOTCONN;
	101	timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
	102	target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
	103
	104	msg->msg_namelen = 0;
	105	/* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
	106	rcvbuf_base = conn->rmb_desc->cpu_addr;
	107
	108	do { /* while (read_remaining) */
	109	if (read_done >= target)
	110	break;
	111
	112	if (atomic_read(&conn->bytes_to_rcv))
	113	goto copy;
	114
	115	if (read_done) {
	116	if (sk->sk_err \|\|
	117	sk->sk_state == SMC_CLOSED \|\|
71c125c3	118	sk->sk_shutdown & RCV_SHUTDOWN \|\|
952310cc UB	119	!timeo \|\|
	120	signal_pending(current) \|\|
	121	smc_cdc_rxed_any_close_or_senddone(conn) \|\|
	122	conn->local_tx_ctrl.conn_state_flags.
	123	peer_conn_abort)
	124	break;
	125	} else {
952310cc UB	126	if (sk->sk_err) {
	127	read_done = sock_error(sk);
	128	break;
	129	}
	130	if (sk->sk_shutdown & RCV_SHUTDOWN \|\|
	131	smc_cdc_rxed_any_close_or_senddone(conn) \|\|
	132	conn->local_tx_ctrl.conn_state_flags.
	133	peer_conn_abort)
	134	break;
	135	if (sk->sk_state == SMC_CLOSED) {
	136	if (!sock_flag(sk, SOCK_DONE)) {
	137	/* This occurs when user tries to read
	138	* from never connected socket.
	139	*/
	140	read_done = -ENOTCONN;
	141	break;
	142	}
	143	break;
	144	}
	145	if (signal_pending(current)) {
	146	read_done = sock_intr_errno(timeo);
	147	break;
	148	}
846e344e HW	149	if (!timeo)
846e344e HW	150	return -EAGAIN;
952310cc UB	151	}
	152
	153	if (!atomic_read(&conn->bytes_to_rcv)) {
	154	smc_rx_wait_data(smc, &timeo);
	155	continue;
	156	}
	157
	158	copy:
	159	/* initialize variables for 1st iteration of subsequent loop */
	160	/* could be just 1 byte, even after smc_rx_wait_data above */
	161	readable = atomic_read(&conn->bytes_to_rcv);
	162	/* not more than what user space asked for */
	163	copylen = min_t(size_t, read_remaining, readable);
	164	smc_curs_write(&cons,
	165	smc_curs_read(&conn->local_tx_ctrl.cons, conn),
	166	conn);
	167	/* determine chunks where to read from rcvbuf */
	168	/* either unwrapped case, or 1st chunk of wrapped case */
	169	chunk_len = min_t(size_t,
	170	copylen, conn->rmbe_size - cons.count);
	171	chunk_len_sum = chunk_len;
	172	chunk_off = cons.count;
10428dd8	173	smc_rmb_sync_sg_for_cpu(conn);
952310cc UB	174	for (chunk = 0; chunk < 2; chunk++) {
	175	if (!(flags & MSG_TRUNC)) {
	176	rc = memcpy_to_msg(msg, rcvbuf_base + chunk_off,
	177	chunk_len);
	178	if (rc) {
	179	if (!read_done)
	180	read_done = -EFAULT;
10428dd8	181	smc_rmb_sync_sg_for_device(conn);
952310cc UB	182	goto out;
	183	}
	184	}
	185	read_remaining -= chunk_len;
	186	read_done += chunk_len;
	187
	188	if (chunk_len_sum == copylen)
	189	break; /* either on 1st or 2nd iteration */
	190	/* prepare next (== 2nd) iteration */
	191	chunk_len = copylen - chunk_len; /* remainder */
	192	chunk_len_sum += chunk_len;
	193	chunk_off = 0; /* modulo offset in recv ring buffer */
	194	}
10428dd8	195	smc_rmb_sync_sg_for_device(conn);
952310cc UB	196
	197	/* update cursors */
	198	if (!(flags & MSG_PEEK)) {
	199	smc_curs_add(conn->rmbe_size, &cons, copylen);
	200	/* increased in recv tasklet smc_cdc_msg_rcv() */
	201	smp_mb__before_atomic();
	202	atomic_sub(copylen, &conn->bytes_to_rcv);
	203	/* guarantee 0 <= bytes_to_rcv <= rmbe_size */
	204	smp_mb__after_atomic();
	205	smc_curs_write(&conn->local_tx_ctrl.cons,
	206	smc_curs_read(&cons, conn),
	207	conn);
	208	/* send consumer cursor update if required */
	209	/* similar to advertising new TCP rcv_wnd if required */
	210	smc_tx_consumer_update(conn);
	211	}
	212	} while (read_remaining);
	213	out:
	214	return read_done;
	215	}
	216
	217	/* Initialize receive properties on connection establishment. NB: not __init! */
	218	void smc_rx_init(struct smc_sock *smc)
	219	{
	220	smc->sk.sk_data_ready = smc_rx_data_ready;
	221	}