1// SPDX-License-Identifier: GPL-2.0
  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>
 15#include <linux/sched/signal.h>
 16
 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 */
 29static 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, EPOLLIN | EPOLLPRI |
 39						EPOLLRDNORM | EPOLLRDBAND);
 40	sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
 41	if ((sk->sk_shutdown == SHUTDOWN_MASK) ||
 42	    (sk->sk_state == SMC_CLOSED))
 43		sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_HUP);
 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 */
 54static 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			   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 */
 79int 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 ||
118			    sk->sk_shutdown & RCV_SHUTDOWN ||
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 {
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			}
149			if (!timeo)
150				return -EAGAIN;
151		}
152
153		if (!atomic_read(&conn->bytes_to_rcv)) {
154			smc_rx_wait_data(smc, &timeo);
155			continue;
156		}
157
158copy:
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;
173		smc_rmb_sync_sg_for_cpu(conn);
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;
181					smc_rmb_sync_sg_for_device(conn);
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		}
195		smc_rmb_sync_sg_for_device(conn);
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);
213out:
214	return read_done;
215}
216
217/* Initialize receive properties on connection establishment. NB: not __init! */
218void smc_rx_init(struct smc_sock *smc)
219{
220	smc->sk.sk_data_ready = smc_rx_data_ready;
221}