1 // SPDX-License-Identifier: GPL-2.0
3 * Shared Memory Communications over RDMA (SMC-R) and RoCE
6 * copy new RMBE data into user space
8 * Copyright IBM Corp. 2016
10 * Author(s): Ursula Braun <ubraun@linux.vnet.ibm.com>
13 #include <linux/net.h>
14 #include <linux/rcupdate.h>
15 #include <linux/sched/signal.h>
22 #include "smc_tx.h" /* smc_tx_consumer_update() */
25 /* callback implementation to wakeup consumers blocked with smc_rx_wait().
26 * indirectly called by smc_cdc_msg_recv_action().
28 static void smc_rx_wake_up(struct sock
*sk
)
32 /* derived from sock_def_readable() */
33 /* called already in smc_listen_work() */
35 wq
= rcu_dereference(sk
->sk_wq
);
36 if (skwq_has_sleeper(wq
))
37 wake_up_interruptible_sync_poll(&wq
->wait
, EPOLLIN
| EPOLLPRI
|
38 EPOLLRDNORM
| EPOLLRDBAND
);
39 sk_wake_async(sk
, SOCK_WAKE_WAITD
, POLL_IN
);
40 if ((sk
->sk_shutdown
== SHUTDOWN_MASK
) ||
41 (sk
->sk_state
== SMC_CLOSED
))
42 sk_wake_async(sk
, SOCK_WAKE_WAITD
, POLL_HUP
);
46 /* Update consumer cursor
47 * @conn connection to update
48 * @cons consumer cursor
49 * @len number of Bytes consumed
51 * 1 if we should end our receive, 0 otherwise
53 static int smc_rx_update_consumer(struct smc_sock
*smc
,
54 union smc_host_cursor cons
, size_t len
)
56 struct smc_connection
*conn
= &smc
->conn
;
57 struct sock
*sk
= &smc
->sk
;
61 smc_curs_add(conn
->rmb_desc
->len
, &cons
, len
);
63 /* did we process urgent data? */
64 if (conn
->urg_state
== SMC_URG_VALID
|| conn
->urg_rx_skip_pend
) {
65 diff
= smc_curs_comp(conn
->rmb_desc
->len
, &cons
,
67 if (sock_flag(sk
, SOCK_URGINLINE
)) {
71 conn
->urg_state
= SMC_URG_READ
;
75 /* skip urgent byte */
77 smc_curs_add(conn
->rmb_desc
->len
, &cons
, 1);
78 conn
->urg_rx_skip_pend
= false;
80 /* we read past urgent byte */
81 conn
->urg_state
= SMC_URG_READ
;
85 smc_curs_copy(&conn
->local_tx_ctrl
.cons
, &cons
, conn
);
87 /* send consumer cursor update if required */
88 /* similar to advertising new TCP rcv_wnd if required */
89 smc_tx_consumer_update(conn
, force
);
94 static void smc_rx_update_cons(struct smc_sock
*smc
, size_t len
)
96 struct smc_connection
*conn
= &smc
->conn
;
97 union smc_host_cursor cons
;
99 smc_curs_copy(&cons
, &conn
->local_tx_ctrl
.cons
, conn
);
100 smc_rx_update_consumer(smc
, cons
, len
);
103 struct smc_spd_priv
{
104 struct smc_sock
*smc
;
108 static void smc_rx_pipe_buf_release(struct pipe_inode_info
*pipe
,
109 struct pipe_buffer
*buf
)
111 struct smc_spd_priv
*priv
= (struct smc_spd_priv
*)buf
->private;
112 struct smc_sock
*smc
= priv
->smc
;
113 struct smc_connection
*conn
;
114 struct sock
*sk
= &smc
->sk
;
116 if (sk
->sk_state
== SMC_CLOSED
||
117 sk
->sk_state
== SMC_PEERFINCLOSEWAIT
||
118 sk
->sk_state
== SMC_APPFINCLOSEWAIT
)
122 smc_rx_update_cons(smc
, priv
->len
);
124 if (atomic_sub_and_test(priv
->len
, &conn
->splice_pending
))
132 static int smc_rx_pipe_buf_nosteal(struct pipe_inode_info
*pipe
,
133 struct pipe_buffer
*buf
)
138 static const struct pipe_buf_operations smc_pipe_ops
= {
140 .confirm
= generic_pipe_buf_confirm
,
141 .release
= smc_rx_pipe_buf_release
,
142 .steal
= smc_rx_pipe_buf_nosteal
,
143 .get
= generic_pipe_buf_get
146 static void smc_rx_spd_release(struct splice_pipe_desc
*spd
,
149 put_page(spd
->pages
[i
]);
152 static int smc_rx_splice(struct pipe_inode_info
*pipe
, char *src
, size_t len
,
153 struct smc_sock
*smc
)
155 struct splice_pipe_desc spd
;
156 struct partial_page partial
;
157 struct smc_spd_priv
*priv
;
160 priv
= kzalloc(sizeof(*priv
), GFP_KERNEL
);
165 partial
.offset
= src
- (char *)smc
->conn
.rmb_desc
->cpu_addr
;
167 partial
.private = (unsigned long)priv
;
169 spd
.nr_pages_max
= 1;
171 spd
.pages
= &smc
->conn
.rmb_desc
->pages
;
172 spd
.partial
= &partial
;
173 spd
.ops
= &smc_pipe_ops
;
174 spd
.spd_release
= smc_rx_spd_release
;
176 bytes
= splice_to_pipe(pipe
, &spd
);
179 get_page(smc
->conn
.rmb_desc
->pages
);
180 atomic_add(bytes
, &smc
->conn
.splice_pending
);
186 static int smc_rx_data_available_and_no_splice_pend(struct smc_connection
*conn
)
188 return atomic_read(&conn
->bytes_to_rcv
) &&
189 !atomic_read(&conn
->splice_pending
);
192 /* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
194 * @timeo pointer to max seconds to wait, pointer to value 0 for no timeout
195 * @fcrit add'l criterion to evaluate as function pointer
197 * 1 if at least 1 byte available in rcvbuf or if socket error/shutdown.
198 * 0 otherwise (nothing in rcvbuf nor timeout, e.g. interrupted).
200 int smc_rx_wait(struct smc_sock
*smc
, long *timeo
,
201 int (*fcrit
)(struct smc_connection
*conn
))
203 DEFINE_WAIT_FUNC(wait
, woken_wake_function
);
204 struct smc_connection
*conn
= &smc
->conn
;
205 struct sock
*sk
= &smc
->sk
;
210 sk_set_bit(SOCKWQ_ASYNC_WAITDATA
, sk
);
211 add_wait_queue(sk_sleep(sk
), &wait
);
212 rc
= sk_wait_event(sk
, timeo
,
214 sk
->sk_shutdown
& RCV_SHUTDOWN
||
217 remove_wait_queue(sk_sleep(sk
), &wait
);
218 sk_clear_bit(SOCKWQ_ASYNC_WAITDATA
, sk
);
222 static int smc_rx_recv_urg(struct smc_sock
*smc
, struct msghdr
*msg
, int len
,
225 struct smc_connection
*conn
= &smc
->conn
;
226 union smc_host_cursor cons
;
227 struct sock
*sk
= &smc
->sk
;
230 if (sock_flag(sk
, SOCK_URGINLINE
) ||
231 !(conn
->urg_state
== SMC_URG_VALID
) ||
232 conn
->urg_state
== SMC_URG_READ
)
235 if (conn
->urg_state
== SMC_URG_VALID
) {
236 if (!(flags
& MSG_PEEK
))
237 smc
->conn
.urg_state
= SMC_URG_READ
;
238 msg
->msg_flags
|= MSG_OOB
;
240 if (!(flags
& MSG_TRUNC
))
241 rc
= memcpy_to_msg(msg
, &conn
->urg_rx_byte
, 1);
243 smc_curs_copy(&cons
, &conn
->local_tx_ctrl
.cons
, conn
);
244 if (smc_curs_diff(conn
->rmb_desc
->len
, &cons
,
245 &conn
->urg_curs
) > 1)
246 conn
->urg_rx_skip_pend
= true;
247 /* Urgent Byte was already accounted for, but trigger
248 * skipping the urgent byte in non-inline case
250 if (!(flags
& MSG_PEEK
))
251 smc_rx_update_consumer(smc
, cons
, 0);
253 msg
->msg_flags
|= MSG_TRUNC
;
256 return rc
? -EFAULT
: len
;
259 if (sk
->sk_state
== SMC_CLOSED
|| sk
->sk_shutdown
& RCV_SHUTDOWN
)
265 static bool smc_rx_recvmsg_data_available(struct smc_sock
*smc
)
267 struct smc_connection
*conn
= &smc
->conn
;
269 if (smc_rx_data_available(conn
))
271 else if (conn
->urg_state
== SMC_URG_VALID
)
272 /* we received a single urgent Byte - skip */
273 smc_rx_update_cons(smc
, 0);
277 /* smc_rx_recvmsg - receive data from RMBE
278 * @msg: copy data to receive buffer
279 * @pipe: copy data to pipe if set - indicates splice() call
281 * rcvbuf consumer: main API called by socket layer.
282 * Called under sk lock.
284 int smc_rx_recvmsg(struct smc_sock
*smc
, struct msghdr
*msg
,
285 struct pipe_inode_info
*pipe
, size_t len
, int flags
)
287 size_t copylen
, read_done
= 0, read_remaining
= len
;
288 size_t chunk_len
, chunk_off
, chunk_len_sum
;
289 struct smc_connection
*conn
= &smc
->conn
;
290 int (*func
)(struct smc_connection
*conn
);
291 union smc_host_cursor cons
;
297 int target
; /* Read at least these many bytes */
300 if (unlikely(flags
& MSG_ERRQUEUE
))
301 return -EINVAL
; /* future work for sk.sk_family == AF_SMC */
304 if (sk
->sk_state
== SMC_LISTEN
)
307 return smc_rx_recv_urg(smc
, msg
, len
, flags
);
308 timeo
= sock_rcvtimeo(sk
, flags
& MSG_DONTWAIT
);
309 target
= sock_rcvlowat(sk
, flags
& MSG_WAITALL
, len
);
311 /* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
312 rcvbuf_base
= conn
->rx_off
+ conn
->rmb_desc
->cpu_addr
;
314 do { /* while (read_remaining) */
315 if (read_done
>= target
|| (pipe
&& read_done
))
318 if (smc_rx_recvmsg_data_available(smc
))
321 if (sk
->sk_shutdown
& RCV_SHUTDOWN
||
322 conn
->local_tx_ctrl
.conn_state_flags
.peer_conn_abort
) {
323 /* smc_cdc_msg_recv_action() could have run after
324 * above smc_rx_recvmsg_data_available()
326 if (smc_rx_recvmsg_data_available(smc
))
333 sk
->sk_state
== SMC_CLOSED
||
335 signal_pending(current
))
339 read_done
= sock_error(sk
);
342 if (sk
->sk_state
== SMC_CLOSED
) {
343 if (!sock_flag(sk
, SOCK_DONE
)) {
344 /* This occurs when user tries to read
345 * from never connected socket.
347 read_done
= -ENOTCONN
;
352 if (signal_pending(current
)) {
353 read_done
= sock_intr_errno(timeo
);
360 if (!smc_rx_data_available(conn
)) {
361 smc_rx_wait(smc
, &timeo
, smc_rx_data_available
);
366 /* initialize variables for 1st iteration of subsequent loop */
367 /* could be just 1 byte, even after waiting on data above */
368 readable
= atomic_read(&conn
->bytes_to_rcv
);
369 splbytes
= atomic_read(&conn
->splice_pending
);
370 if (!readable
|| (msg
&& splbytes
)) {
372 func
= smc_rx_data_available_and_no_splice_pend
;
374 func
= smc_rx_data_available
;
375 smc_rx_wait(smc
, &timeo
, func
);
379 smc_curs_copy(&cons
, &conn
->local_tx_ctrl
.cons
, conn
);
380 /* subsequent splice() calls pick up where previous left */
382 smc_curs_add(conn
->rmb_desc
->len
, &cons
, splbytes
);
383 if (conn
->urg_state
== SMC_URG_VALID
&&
384 sock_flag(&smc
->sk
, SOCK_URGINLINE
) &&
386 readable
--; /* always stop at urgent Byte */
387 /* not more than what user space asked for */
388 copylen
= min_t(size_t, read_remaining
, readable
);
389 /* determine chunks where to read from rcvbuf */
390 /* either unwrapped case, or 1st chunk of wrapped case */
391 chunk_len
= min_t(size_t, copylen
, conn
->rmb_desc
->len
-
393 chunk_len_sum
= chunk_len
;
394 chunk_off
= cons
.count
;
395 smc_rmb_sync_sg_for_cpu(conn
);
396 for (chunk
= 0; chunk
< 2; chunk
++) {
397 if (!(flags
& MSG_TRUNC
)) {
399 rc
= memcpy_to_msg(msg
, rcvbuf_base
+
403 rc
= smc_rx_splice(pipe
, rcvbuf_base
+
404 chunk_off
, chunk_len
,
410 smc_rmb_sync_sg_for_device(conn
);
414 read_remaining
-= chunk_len
;
415 read_done
+= chunk_len
;
417 if (chunk_len_sum
== copylen
)
418 break; /* either on 1st or 2nd iteration */
419 /* prepare next (== 2nd) iteration */
420 chunk_len
= copylen
- chunk_len
; /* remainder */
421 chunk_len_sum
+= chunk_len
;
422 chunk_off
= 0; /* modulo offset in recv ring buffer */
424 smc_rmb_sync_sg_for_device(conn
);
427 if (!(flags
& MSG_PEEK
)) {
428 /* increased in recv tasklet smc_cdc_msg_rcv() */
429 smp_mb__before_atomic();
430 atomic_sub(copylen
, &conn
->bytes_to_rcv
);
431 /* guarantee 0 <= bytes_to_rcv <= rmb_desc->len */
432 smp_mb__after_atomic();
433 if (msg
&& smc_rx_update_consumer(smc
, cons
, copylen
))
436 } while (read_remaining
);
441 /* Initialize receive properties on connection establishment. NB: not __init! */
442 void smc_rx_init(struct smc_sock
*smc
)
444 smc
->sk
.sk_data_ready
= smc_rx_wake_up
;
445 atomic_set(&smc
->conn
.splice_pending
, 0);
446 smc
->conn
.urg_state
= SMC_URG_READ
;