1
2
3
4
5
6
7
8
9
10
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"
23#include "smc_rx.h"
24
25
26
27
28static void smc_rx_wake_up(struct sock *sk)
29{
30 struct socket_wq *wq;
31
32
33
34 rcu_read_lock();
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);
43 rcu_read_unlock();
44}
45
46
47
48
49
50
51
52
53static int smc_rx_update_consumer(struct smc_sock *smc,
54 union smc_host_cursor cons, size_t len)
55{
56 struct smc_connection *conn = &smc->conn;
57 struct sock *sk = &smc->sk;
58 bool force = false;
59 int diff, rc = 0;
60
61 smc_curs_add(conn->rmb_desc->len, &cons, len);
62
63
64 if (conn->urg_state == SMC_URG_VALID || conn->urg_rx_skip_pend) {
65 diff = smc_curs_comp(conn->rmb_desc->len, &cons,
66 &conn->urg_curs);
67 if (sock_flag(sk, SOCK_URGINLINE)) {
68 if (diff == 0) {
69 force = true;
70 rc = 1;
71 conn->urg_state = SMC_URG_READ;
72 }
73 } else {
74 if (diff == 1) {
75
76 force = true;
77 smc_curs_add(conn->rmb_desc->len, &cons, 1);
78 conn->urg_rx_skip_pend = false;
79 } else if (diff < -1)
80
81 conn->urg_state = SMC_URG_READ;
82 }
83 }
84
85 smc_curs_copy(&conn->local_tx_ctrl.cons, &cons, conn);
86
87
88
89 smc_tx_consumer_update(conn, force);
90
91 return rc;
92}
93
94static void smc_rx_update_cons(struct smc_sock *smc, size_t len)
95{
96 struct smc_connection *conn = &smc->conn;
97 union smc_host_cursor cons;
98
99 smc_curs_copy(&cons, &conn->local_tx_ctrl.cons, conn);
100 smc_rx_update_consumer(smc, cons, len);
101}
102
103struct smc_spd_priv {
104 struct smc_sock *smc;
105 size_t len;
106};
107
108static void smc_rx_pipe_buf_release(struct pipe_inode_info *pipe,
109 struct pipe_buffer *buf)
110{
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;
115
116 if (sk->sk_state == SMC_CLOSED ||
117 sk->sk_state == SMC_PEERFINCLOSEWAIT ||
118 sk->sk_state == SMC_APPFINCLOSEWAIT)
119 goto out;
120 conn = &smc->conn;
121 lock_sock(sk);
122 smc_rx_update_cons(smc, priv->len);
123 release_sock(sk);
124 if (atomic_sub_and_test(priv->len, &conn->splice_pending))
125 smc_rx_wake_up(sk);
126out:
127 kfree(priv);
128 put_page(buf->page);
129 sock_put(sk);
130}
131
132static int smc_rx_pipe_buf_nosteal(struct pipe_inode_info *pipe,
133 struct pipe_buffer *buf)
134{
135 return 1;
136}
137
138static const struct pipe_buf_operations smc_pipe_ops = {
139 .confirm = generic_pipe_buf_confirm,
140 .release = smc_rx_pipe_buf_release,
141 .steal = smc_rx_pipe_buf_nosteal,
142 .get = generic_pipe_buf_get
143};
144
145static void smc_rx_spd_release(struct splice_pipe_desc *spd,
146 unsigned int i)
147{
148 put_page(spd->pages[i]);
149}
150
151static int smc_rx_splice(struct pipe_inode_info *pipe, char *src, size_t len,
152 struct smc_sock *smc)
153{
154 struct splice_pipe_desc spd;
155 struct partial_page partial;
156 struct smc_spd_priv *priv;
157 int bytes;
158
159 priv = kzalloc(sizeof(*priv), GFP_KERNEL);
160 if (!priv)
161 return -ENOMEM;
162 priv->len = len;
163 priv->smc = smc;
164 partial.offset = src - (char *)smc->conn.rmb_desc->cpu_addr;
165 partial.len = len;
166 partial.private = (unsigned long)priv;
167
168 spd.nr_pages_max = 1;
169 spd.nr_pages = 1;
170 spd.pages = &smc->conn.rmb_desc->pages;
171 spd.partial = &partial;
172 spd.ops = &smc_pipe_ops;
173 spd.spd_release = smc_rx_spd_release;
174
175 bytes = splice_to_pipe(pipe, &spd);
176 if (bytes > 0) {
177 sock_hold(&smc->sk);
178 get_page(smc->conn.rmb_desc->pages);
179 atomic_add(bytes, &smc->conn.splice_pending);
180 }
181
182 return bytes;
183}
184
185static int smc_rx_data_available_and_no_splice_pend(struct smc_connection *conn)
186{
187 return atomic_read(&conn->bytes_to_rcv) &&
188 !atomic_read(&conn->splice_pending);
189}
190
191
192
193
194
195
196
197
198
199int smc_rx_wait(struct smc_sock *smc, long *timeo,
200 int (*fcrit)(struct smc_connection *conn))
201{
202 DEFINE_WAIT_FUNC(wait, woken_wake_function);
203 struct smc_connection *conn = &smc->conn;
204 struct sock *sk = &smc->sk;
205 int rc;
206
207 if (fcrit(conn))
208 return 1;
209 sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk);
210 add_wait_queue(sk_sleep(sk), &wait);
211 rc = sk_wait_event(sk, timeo,
212 sk->sk_err ||
213 sk->sk_shutdown & RCV_SHUTDOWN ||
214 fcrit(conn) ||
215 smc_cdc_rxed_any_close_or_senddone(conn),
216 &wait);
217 remove_wait_queue(sk_sleep(sk), &wait);
218 sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk);
219 return rc;
220}
221
222static int smc_rx_recv_urg(struct smc_sock *smc, struct msghdr *msg, int len,
223 int flags)
224{
225 struct smc_connection *conn = &smc->conn;
226 union smc_host_cursor cons;
227 struct sock *sk = &smc->sk;
228 int rc = 0;
229
230 if (sock_flag(sk, SOCK_URGINLINE) ||
231 !(conn->urg_state == SMC_URG_VALID) ||
232 conn->urg_state == SMC_URG_READ)
233 return -EINVAL;
234
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;
239 if (len > 0) {
240 if (!(flags & MSG_TRUNC))
241 rc = memcpy_to_msg(msg, &conn->urg_rx_byte, 1);
242 len = 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
248
249
250 if (!(flags & MSG_PEEK))
251 smc_rx_update_consumer(smc, cons, 0);
252 } else {
253 msg->msg_flags |= MSG_TRUNC;
254 }
255
256 return rc ? -EFAULT : len;
257 }
258
259 if (sk->sk_state == SMC_CLOSED || sk->sk_shutdown & RCV_SHUTDOWN)
260 return 0;
261
262 return -EAGAIN;
263}
264
265
266
267
268
269
270
271
272int smc_rx_recvmsg(struct smc_sock *smc, struct msghdr *msg,
273 struct pipe_inode_info *pipe, size_t len, int flags)
274{
275 size_t copylen, read_done = 0, read_remaining = len;
276 size_t chunk_len, chunk_off, chunk_len_sum;
277 struct smc_connection *conn = &smc->conn;
278 int (*func)(struct smc_connection *conn);
279 union smc_host_cursor cons;
280 int readable, chunk;
281 char *rcvbuf_base;
282 struct sock *sk;
283 int splbytes;
284 long timeo;
285 int target;
286 int rc;
287
288 if (unlikely(flags & MSG_ERRQUEUE))
289 return -EINVAL;
290
291 sk = &smc->sk;
292 if (sk->sk_state == SMC_LISTEN)
293 return -ENOTCONN;
294 if (flags & MSG_OOB)
295 return smc_rx_recv_urg(smc, msg, len, flags);
296 timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
297 target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
298
299
300 rcvbuf_base = conn->rx_off + conn->rmb_desc->cpu_addr;
301
302 do {
303 if (read_done >= target || (pipe && read_done))
304 break;
305
306 if (atomic_read(&conn->bytes_to_rcv))
307 goto copy;
308 else if (conn->urg_state == SMC_URG_VALID)
309
310 smc_rx_update_cons(smc, 0);
311
312 if (sk->sk_shutdown & RCV_SHUTDOWN ||
313 smc_cdc_rxed_any_close_or_senddone(conn) ||
314 conn->local_tx_ctrl.conn_state_flags.peer_conn_abort)
315 break;
316
317 if (read_done) {
318 if (sk->sk_err ||
319 sk->sk_state == SMC_CLOSED ||
320 !timeo ||
321 signal_pending(current))
322 break;
323 } else {
324 if (sk->sk_err) {
325 read_done = sock_error(sk);
326 break;
327 }
328 if (sk->sk_state == SMC_CLOSED) {
329 if (!sock_flag(sk, SOCK_DONE)) {
330
331
332
333 read_done = -ENOTCONN;
334 break;
335 }
336 break;
337 }
338 if (signal_pending(current)) {
339 read_done = sock_intr_errno(timeo);
340 break;
341 }
342 if (!timeo)
343 return -EAGAIN;
344 }
345
346 if (!smc_rx_data_available(conn)) {
347 smc_rx_wait(smc, &timeo, smc_rx_data_available);
348 continue;
349 }
350
351copy:
352
353
354 readable = atomic_read(&conn->bytes_to_rcv);
355 splbytes = atomic_read(&conn->splice_pending);
356 if (!readable || (msg && splbytes)) {
357 if (splbytes)
358 func = smc_rx_data_available_and_no_splice_pend;
359 else
360 func = smc_rx_data_available;
361 smc_rx_wait(smc, &timeo, func);
362 continue;
363 }
364
365 smc_curs_copy(&cons, &conn->local_tx_ctrl.cons, conn);
366
367 if (splbytes)
368 smc_curs_add(conn->rmb_desc->len, &cons, splbytes);
369 if (conn->urg_state == SMC_URG_VALID &&
370 sock_flag(&smc->sk, SOCK_URGINLINE) &&
371 readable > 1)
372 readable--;
373
374 copylen = min_t(size_t, read_remaining, readable);
375
376
377 chunk_len = min_t(size_t, copylen, conn->rmb_desc->len -
378 cons.count);
379 chunk_len_sum = chunk_len;
380 chunk_off = cons.count;
381 smc_rmb_sync_sg_for_cpu(conn);
382 for (chunk = 0; chunk < 2; chunk++) {
383 if (!(flags & MSG_TRUNC)) {
384 if (msg) {
385 rc = memcpy_to_msg(msg, rcvbuf_base +
386 chunk_off,
387 chunk_len);
388 } else {
389 rc = smc_rx_splice(pipe, rcvbuf_base +
390 chunk_off, chunk_len,
391 smc);
392 }
393 if (rc < 0) {
394 if (!read_done)
395 read_done = -EFAULT;
396 smc_rmb_sync_sg_for_device(conn);
397 goto out;
398 }
399 }
400 read_remaining -= chunk_len;
401 read_done += chunk_len;
402
403 if (chunk_len_sum == copylen)
404 break;
405
406 chunk_len = copylen - chunk_len;
407 chunk_len_sum += chunk_len;
408 chunk_off = 0;
409 }
410 smc_rmb_sync_sg_for_device(conn);
411
412
413 if (!(flags & MSG_PEEK)) {
414
415 smp_mb__before_atomic();
416 atomic_sub(copylen, &conn->bytes_to_rcv);
417
418 smp_mb__after_atomic();
419 if (msg && smc_rx_update_consumer(smc, cons, copylen))
420 goto out;
421 }
422 } while (read_remaining);
423out:
424 return read_done;
425}
426
427
428void smc_rx_init(struct smc_sock *smc)
429{
430 smc->sk.sk_data_ready = smc_rx_wake_up;
431 atomic_set(&smc->conn.splice_pending, 0);
432 smc->conn.urg_state = SMC_URG_READ;
433}
434