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