[linux-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, POLLIN | POLLPRI |
						POLLRDNORM | POLLRDBAND);
	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 ||
			   sock_flag(sk, SOCK_DONE) ||
			   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 (sock_flag(sk, SOCK_DONE))
				break;
			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))
	38	wake_up_interruptible_sync_poll(&wq->wait, POLLIN \| POLLPRI \|
	39	POLLRDNORM \| POLLRDBAND);
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 \|\|
	68	sock_flag(sk, SOCK_DONE) \|\|
	69	atomic_read(&conn->bytes_to_rcv) \|\|
	70	smc_cdc_rxed_any_close_or_senddone(conn),
	71	&wait);
	72	remove_wait_queue(sk_sleep(sk), &wait);
	73	sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
	74	return rc;
	75	}
	76
	77	/* rcvbuf consumer: main API called by socket layer.
	78	* called under sk lock.
	79	*/
	80	int smc_rx_recvmsg(struct smc_sock smc, struct msghdr msg, size_t len,
	81	int flags)
	82	{
	83	size_t copylen, read_done = 0, read_remaining = len;
	84	size_t chunk_len, chunk_off, chunk_len_sum;
	85	struct smc_connection *conn = &smc->conn;
	86	union smc_host_cursor cons;
	87	int readable, chunk;
	88	char *rcvbuf_base;
	89	struct sock *sk;
	90	long timeo;
	91	int target; /* Read at least these many bytes */
	92	int rc;
	93
	94	if (unlikely(flags & MSG_ERRQUEUE))
	95	return -EINVAL; /* future work for sk.sk_family == AF_SMC */
	96	if (flags & MSG_OOB)
	97	return -EINVAL; /* future work */
	98
	99	sk = &smc->sk;
	100	if (sk->sk_state == SMC_LISTEN)
	101	return -ENOTCONN;
	102	timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
	103	target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
	104
	105	msg->msg_namelen = 0;
	106	/* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
	107	rcvbuf_base = conn->rmb_desc->cpu_addr;
108
109	do { /* while (read_remaining) */
110	if (read_done >= target)
111	break;
112
113	if (atomic_read(&conn->bytes_to_rcv))
114	goto copy;
115
116	if (read_done) {
117	if (sk->sk_err \|\|
118	sk->sk_state == SMC_CLOSED \|\|
119	(sk->sk_shutdown & RCV_SHUTDOWN) \|\|
120	!timeo \|\|
121	signal_pending(current) \|\|
122	smc_cdc_rxed_any_close_or_senddone(conn) \|\|
123	conn->local_tx_ctrl.conn_state_flags.
124	peer_conn_abort)
125	break;
126	} else {
127	if (sock_flag(sk, SOCK_DONE))
128	break;
129	if (sk->sk_err) {
130	read_done = sock_error(sk);
131	break;
132	}
133	if (sk->sk_shutdown & RCV_SHUTDOWN \|\|
134	smc_cdc_rxed_any_close_or_senddone(conn) \|\|
135	conn->local_tx_ctrl.conn_state_flags.
136	peer_conn_abort)
137	break;
138	if (sk->sk_state == SMC_CLOSED) {
139	if (!sock_flag(sk, SOCK_DONE)) {
140	/* This occurs when user tries to read
141	* from never connected socket.
142	*/
143	read_done = -ENOTCONN;
144	break;
145	}
146	break;
147	}
148	if (signal_pending(current)) {
149	read_done = sock_intr_errno(timeo);
150	break;
151	}
846e344e HW	152	if (!timeo)
846e344e HW	153	return -EAGAIN;
952310cc UB	154	}
	155
	156	if (!atomic_read(&conn->bytes_to_rcv)) {
	157	smc_rx_wait_data(smc, &timeo);
	158	continue;
	159	}
	160
	161	copy:
	162	/* initialize variables for 1st iteration of subsequent loop */
	163	/* could be just 1 byte, even after smc_rx_wait_data above */
	164	readable = atomic_read(&conn->bytes_to_rcv);
	165	/* not more than what user space asked for */
	166	copylen = min_t(size_t, read_remaining, readable);
	167	smc_curs_write(&cons,
	168	smc_curs_read(&conn->local_tx_ctrl.cons, conn),
	169	conn);
	170	/* determine chunks where to read from rcvbuf */
	171	/* either unwrapped case, or 1st chunk of wrapped case */
	172	chunk_len = min_t(size_t,
	173	copylen, conn->rmbe_size - cons.count);
	174	chunk_len_sum = chunk_len;
	175	chunk_off = cons.count;
10428dd8	176	smc_rmb_sync_sg_for_cpu(conn);
952310cc UB	177	for (chunk = 0; chunk < 2; chunk++) {
	178	if (!(flags & MSG_TRUNC)) {
	179	rc = memcpy_to_msg(msg, rcvbuf_base + chunk_off,
	180	chunk_len);
	181	if (rc) {
	182	if (!read_done)
	183	read_done = -EFAULT;
10428dd8	184	smc_rmb_sync_sg_for_device(conn);
952310cc UB	185	goto out;
	186	}
	187	}
	188	read_remaining -= chunk_len;
	189	read_done += chunk_len;
	190
	191	if (chunk_len_sum == copylen)
	192	break; /* either on 1st or 2nd iteration */
	193	/* prepare next (== 2nd) iteration */
	194	chunk_len = copylen - chunk_len; /* remainder */
	195	chunk_len_sum += chunk_len;
	196	chunk_off = 0; /* modulo offset in recv ring buffer */
	197	}
10428dd8	198	smc_rmb_sync_sg_for_device(conn);
952310cc UB	199
	200	/* update cursors */
	201	if (!(flags & MSG_PEEK)) {
	202	smc_curs_add(conn->rmbe_size, &cons, copylen);
	203	/* increased in recv tasklet smc_cdc_msg_rcv() */
	204	smp_mb__before_atomic();
	205	atomic_sub(copylen, &conn->bytes_to_rcv);
	206	/* guarantee 0 <= bytes_to_rcv <= rmbe_size */
	207	smp_mb__after_atomic();
	208	smc_curs_write(&conn->local_tx_ctrl.cons,
	209	smc_curs_read(&cons, conn),
	210	conn);
	211	/* send consumer cursor update if required */
	212	/* similar to advertising new TCP rcv_wnd if required */
	213	smc_tx_consumer_update(conn);
	214	}
	215	} while (read_remaining);
	216	out:
	217	return read_done;
	218	}
	219
	220	/* Initialize receive properties on connection establishment. NB: not __init! */
	221	void smc_rx_init(struct smc_sock *smc)
	222	{
	223	smc->sk.sk_data_ready = smc_rx_data_ready;
	224	}