xref: /linux/net/smc/smc_rx.c (revision fd639726bf15fca8ee1a00dce8e0096d0ad9bd18)
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  */
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);
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  */
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 			}
152 			if (!timeo)
153 				return -EAGAIN;
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;
176 		smc_rmb_sync_sg_for_cpu(conn);
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;
184 					smc_rmb_sync_sg_for_device(conn);
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 		}
198 		smc_rmb_sync_sg_for_device(conn);
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 }
225