]>
Commit | Line | Data |
---|---|---|
952310cc UB |
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 | } |