|
| 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 <[email protected]> |
| 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 | +} |
0 commit comments