net/smc/smc_rx.c

   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 }