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 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().
29 static void smc_rx_data_ready(struct sock *sk)
33 /* derived from sock_def_readable() */
34 /* called already in smc_listen_work() */
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);
47 /* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
49 * @timeo pointer to max seconds to wait, pointer to value 0 for no timeout
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).
54 static int smc_rx_wait_data(struct smc_sock *smc, long *timeo)
56 DEFINE_WAIT_FUNC(wait, woken_wake_function);
57 struct smc_connection *conn = &smc->conn;
58 struct sock *sk = &smc->sk;
61 if (atomic_read(&conn->bytes_to_rcv))
63 sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
64 add_wait_queue(sk_sleep(sk), &wait);
65 rc = sk_wait_event(sk, timeo,
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),
72 remove_wait_queue(sk_sleep(sk), &wait);
73 sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
77 /* rcvbuf consumer: main API called by socket layer.
78 * called under sk lock.
80 int smc_rx_recvmsg(struct smc_sock *smc, struct msghdr *msg, size_t len,
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;
91 int target; /* Read at least these many bytes */
94 if (unlikely(flags & MSG_ERRQUEUE))
95 return -EINVAL; /* future work for sk.sk_family == AF_SMC */
97 return -EINVAL; /* future work */
100 if (sk->sk_state == SMC_LISTEN)
102 timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
103 target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
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;
109 do { /* while (read_remaining) */
110 if (read_done >= target)
113 if (atomic_read(&conn->bytes_to_rcv))
118 sk->sk_state == SMC_CLOSED ||
119 (sk->sk_shutdown & RCV_SHUTDOWN) ||
121 signal_pending(current) ||
122 smc_cdc_rxed_any_close_or_senddone(conn) ||
123 conn->local_tx_ctrl.conn_state_flags.
127 if (sock_flag(sk, SOCK_DONE))
130 read_done = sock_error(sk);
133 if (sk->sk_shutdown & RCV_SHUTDOWN ||
134 smc_cdc_rxed_any_close_or_senddone(conn) ||
135 conn->local_tx_ctrl.conn_state_flags.
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.
143 read_done = -ENOTCONN;
150 if (signal_pending(current)) {
151 read_done = sock_intr_errno(timeo);
156 if (!atomic_read(&conn->bytes_to_rcv)) {
157 smc_rx_wait_data(smc, &timeo);
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),
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,
184 smc_rmb_sync_sg_for_device(conn);
188 read_remaining -= chunk_len;
189 read_done += chunk_len;
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 */
198 smc_rmb_sync_sg_for_device(conn);
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),
211 /* send consumer cursor update if required */
212 /* similar to advertising new TCP rcv_wnd if required */
213 smc_tx_consumer_update(conn);
215 } while (read_remaining);
220 /* Initialize receive properties on connection establishment. NB: not __init! */
221 void smc_rx_init(struct smc_sock *smc)
223 smc->sk.sk_data_ready = smc_rx_data_ready;