1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16#include <linux/net.h>
17#include <linux/rcupdate.h>
18#include <linux/workqueue.h>
19#include <linux/sched/signal.h>
20
21#include <net/sock.h>
22
23#include "smc.h"
24#include "smc_wr.h"
25#include "smc_cdc.h"
26#include "smc_tx.h"
27
28#define SMC_TX_WORK_DELAY HZ
29
30
31
32
33
34
35
36static void smc_tx_write_space(struct sock *sk)
37{
38 struct socket *sock = sk->sk_socket;
39 struct smc_sock *smc = smc_sk(sk);
40 struct socket_wq *wq;
41
42
43 if (atomic_read(&smc->conn.sndbuf_space) && sock) {
44 clear_bit(SOCK_NOSPACE, &sock->flags);
45 rcu_read_lock();
46 wq = rcu_dereference(sk->sk_wq);
47 if (skwq_has_sleeper(wq))
48 wake_up_interruptible_poll(&wq->wait,
49 EPOLLOUT | EPOLLWRNORM |
50 EPOLLWRBAND);
51 if (wq && wq->fasync_list && !(sk->sk_shutdown & SEND_SHUTDOWN))
52 sock_wake_async(wq, SOCK_WAKE_SPACE, POLL_OUT);
53 rcu_read_unlock();
54 }
55}
56
57
58
59
60void smc_tx_sndbuf_nonfull(struct smc_sock *smc)
61{
62 if (smc->sk.sk_socket &&
63 test_bit(SOCK_NOSPACE, &smc->sk.sk_socket->flags))
64 smc->sk.sk_write_space(&smc->sk);
65}
66
67
68static int smc_tx_wait_memory(struct smc_sock *smc, int flags)
69{
70 DEFINE_WAIT_FUNC(wait, woken_wake_function);
71 struct smc_connection *conn = &smc->conn;
72 struct sock *sk = &smc->sk;
73 bool noblock;
74 long timeo;
75 int rc = 0;
76
77
78 timeo = sock_sndtimeo(sk, flags & MSG_DONTWAIT);
79 noblock = timeo ? false : true;
80 add_wait_queue(sk_sleep(sk), &wait);
81 while (1) {
82 sk_set_bit(SOCKWQ_ASYNC_NOSPACE, sk);
83 if (sk->sk_err ||
84 (sk->sk_shutdown & SEND_SHUTDOWN) ||
85 conn->local_tx_ctrl.conn_state_flags.peer_done_writing) {
86 rc = -EPIPE;
87 break;
88 }
89 if (smc_cdc_rxed_any_close(conn)) {
90 rc = -ECONNRESET;
91 break;
92 }
93 if (!timeo) {
94 if (noblock)
95 set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
96 rc = -EAGAIN;
97 break;
98 }
99 if (signal_pending(current)) {
100 rc = sock_intr_errno(timeo);
101 break;
102 }
103 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
104 if (atomic_read(&conn->sndbuf_space))
105 break;
106 set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
107 sk_wait_event(sk, &timeo,
108 sk->sk_err ||
109 (sk->sk_shutdown & SEND_SHUTDOWN) ||
110 smc_cdc_rxed_any_close(conn) ||
111 atomic_read(&conn->sndbuf_space),
112 &wait);
113 }
114 remove_wait_queue(sk_sleep(sk), &wait);
115 return rc;
116}
117
118
119
120
121int smc_tx_sendmsg(struct smc_sock *smc, struct msghdr *msg, size_t len)
122{
123 size_t copylen, send_done = 0, send_remaining = len;
124 size_t chunk_len, chunk_off, chunk_len_sum;
125 struct smc_connection *conn = &smc->conn;
126 union smc_host_cursor prep;
127 struct sock *sk = &smc->sk;
128 char *sndbuf_base;
129 int tx_cnt_prep;
130 int writespace;
131 int rc, chunk;
132
133
134 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
135
136 if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN)) {
137 rc = -EPIPE;
138 goto out_err;
139 }
140
141 while (msg_data_left(msg)) {
142 if (sk->sk_state == SMC_INIT)
143 return -ENOTCONN;
144 if (smc->sk.sk_shutdown & SEND_SHUTDOWN ||
145 (smc->sk.sk_err == ECONNABORTED) ||
146 conn->local_tx_ctrl.conn_state_flags.peer_conn_abort)
147 return -EPIPE;
148 if (smc_cdc_rxed_any_close(conn))
149 return send_done ?: -ECONNRESET;
150
151 if (!atomic_read(&conn->sndbuf_space)) {
152 rc = smc_tx_wait_memory(smc, msg->msg_flags);
153 if (rc) {
154 if (send_done)
155 return send_done;
156 goto out_err;
157 }
158 continue;
159 }
160
161
162
163 writespace = atomic_read(&conn->sndbuf_space);
164
165 copylen = min_t(size_t, send_remaining, writespace);
166
167 sndbuf_base = conn->sndbuf_desc->cpu_addr;
168 smc_curs_write(&prep,
169 smc_curs_read(&conn->tx_curs_prep, conn),
170 conn);
171 tx_cnt_prep = prep.count;
172
173
174 chunk_len = min_t(size_t,
175 copylen, conn->sndbuf_size - tx_cnt_prep);
176 chunk_len_sum = chunk_len;
177 chunk_off = tx_cnt_prep;
178 smc_sndbuf_sync_sg_for_cpu(conn);
179 for (chunk = 0; chunk < 2; chunk++) {
180 rc = memcpy_from_msg(sndbuf_base + chunk_off,
181 msg, chunk_len);
182 if (rc) {
183 smc_sndbuf_sync_sg_for_device(conn);
184 if (send_done)
185 return send_done;
186 goto out_err;
187 }
188 send_done += chunk_len;
189 send_remaining -= chunk_len;
190
191 if (chunk_len_sum == copylen)
192 break;
193
194 chunk_len = copylen - chunk_len;
195 chunk_len_sum += chunk_len;
196 chunk_off = 0;
197 }
198 smc_sndbuf_sync_sg_for_device(conn);
199
200 smc_curs_add(conn->sndbuf_size, &prep, copylen);
201 smc_curs_write(&conn->tx_curs_prep,
202 smc_curs_read(&prep, conn),
203 conn);
204
205 smp_mb__before_atomic();
206 atomic_sub(copylen, &conn->sndbuf_space);
207
208 smp_mb__after_atomic();
209
210
211
212 smc_tx_sndbuf_nonempty(conn);
213 }
214
215 return send_done;
216
217out_err:
218 rc = sk_stream_error(sk, msg->msg_flags, rc);
219
220 if (unlikely(rc == -EAGAIN))
221 sk->sk_write_space(sk);
222 return rc;
223}
224
225
226
227
228static int smc_tx_rdma_write(struct smc_connection *conn, int peer_rmbe_offset,
229 int num_sges, struct ib_sge sges[])
230{
231 struct smc_link_group *lgr = conn->lgr;
232 struct ib_send_wr *failed_wr = NULL;
233 struct ib_rdma_wr rdma_wr;
234 struct smc_link *link;
235 int rc;
236
237 memset(&rdma_wr, 0, sizeof(rdma_wr));
238 link = &lgr->lnk[SMC_SINGLE_LINK];
239 rdma_wr.wr.wr_id = smc_wr_tx_get_next_wr_id(link);
240 rdma_wr.wr.sg_list = sges;
241 rdma_wr.wr.num_sge = num_sges;
242 rdma_wr.wr.opcode = IB_WR_RDMA_WRITE;
243 rdma_wr.remote_addr =
244 lgr->rtokens[conn->rtoken_idx][SMC_SINGLE_LINK].dma_addr +
245
246 ((conn->peer_conn_idx - 1) * conn->peer_rmbe_size) +
247
248 peer_rmbe_offset;
249 rdma_wr.rkey = lgr->rtokens[conn->rtoken_idx][SMC_SINGLE_LINK].rkey;
250 rc = ib_post_send(link->roce_qp, &rdma_wr.wr, &failed_wr);
251 if (rc) {
252 conn->local_tx_ctrl.conn_state_flags.peer_conn_abort = 1;
253 smc_lgr_terminate(lgr);
254 }
255 return rc;
256}
257
258
259static inline void smc_tx_advance_cursors(struct smc_connection *conn,
260 union smc_host_cursor *prod,
261 union smc_host_cursor *sent,
262 size_t len)
263{
264 smc_curs_add(conn->peer_rmbe_size, prod, len);
265
266 smp_mb__before_atomic();
267
268 atomic_sub(len, &conn->peer_rmbe_space);
269
270 smp_mb__after_atomic();
271 smc_curs_add(conn->sndbuf_size, sent, len);
272}
273
274
275
276
277static int smc_tx_rdma_writes(struct smc_connection *conn)
278{
279 size_t src_off, src_len, dst_off, dst_len;
280 size_t len, dst_len_sum, src_len_sum, dstchunk, srcchunk;
281 union smc_host_cursor sent, prep, prod, cons;
282 struct ib_sge sges[SMC_IB_MAX_SEND_SGE];
283 struct smc_link_group *lgr = conn->lgr;
284 int to_send, rmbespace;
285 struct smc_link *link;
286 dma_addr_t dma_addr;
287 int num_sges;
288 int rc;
289
290
291 smc_curs_write(&sent, smc_curs_read(&conn->tx_curs_sent, conn), conn);
292 smc_curs_write(&prep, smc_curs_read(&conn->tx_curs_prep, conn), conn);
293
294 to_send = smc_curs_diff(conn->sndbuf_size, &sent, &prep);
295 if (to_send <= 0)
296 return 0;
297
298
299
300 rmbespace = atomic_read(&conn->peer_rmbe_space);
301 if (rmbespace <= 0)
302 return 0;
303 smc_curs_write(&prod,
304 smc_curs_read(&conn->local_tx_ctrl.prod, conn),
305 conn);
306 smc_curs_write(&cons,
307 smc_curs_read(&conn->local_rx_ctrl.cons, conn),
308 conn);
309
310
311 conn->local_tx_ctrl.prod_flags.write_blocked = (to_send >= rmbespace);
312
313 len = min(to_send, rmbespace);
314
315
316 link = &lgr->lnk[SMC_SINGLE_LINK];
317 dst_off = prod.count;
318 if (prod.wrap == cons.wrap) {
319
320
321
322
323
324 dst_len = min_t(size_t,
325 conn->peer_rmbe_size - prod.count, len);
326 } else {
327
328
329
330
331 dst_len = len;
332 }
333 dst_len_sum = dst_len;
334 src_off = sent.count;
335
336 if (sent.count + dst_len <= conn->sndbuf_size) {
337
338 src_len = dst_len;
339 } else {
340
341 src_len = conn->sndbuf_size - sent.count;
342 }
343 src_len_sum = src_len;
344 dma_addr = sg_dma_address(conn->sndbuf_desc->sgt[SMC_SINGLE_LINK].sgl);
345 for (dstchunk = 0; dstchunk < 2; dstchunk++) {
346 num_sges = 0;
347 for (srcchunk = 0; srcchunk < 2; srcchunk++) {
348 sges[srcchunk].addr = dma_addr + src_off;
349 sges[srcchunk].length = src_len;
350 sges[srcchunk].lkey = link->roce_pd->local_dma_lkey;
351 num_sges++;
352 src_off += src_len;
353 if (src_off >= conn->sndbuf_size)
354 src_off -= conn->sndbuf_size;
355
356 if (src_len_sum == dst_len)
357 break;
358
359 src_len = dst_len - src_len;
360 src_len_sum += src_len;
361 }
362 rc = smc_tx_rdma_write(conn, dst_off, num_sges, sges);
363 if (rc)
364 return rc;
365 if (dst_len_sum == len)
366 break;
367
368 dst_off = 0;
369 dst_len = len - dst_len;
370 dst_len_sum += dst_len;
371 src_len = min_t(int,
372 dst_len, conn->sndbuf_size - sent.count);
373 src_len_sum = src_len;
374 }
375
376 smc_tx_advance_cursors(conn, &prod, &sent, len);
377
378 smc_curs_write(&conn->local_tx_ctrl.prod,
379 smc_curs_read(&prod, conn),
380 conn);
381
382 smc_curs_write(&conn->tx_curs_sent,
383 smc_curs_read(&sent, conn),
384 conn);
385
386
387 return 0;
388}
389
390
391
392
393int smc_tx_sndbuf_nonempty(struct smc_connection *conn)
394{
395 struct smc_cdc_tx_pend *pend;
396 struct smc_wr_buf *wr_buf;
397 int rc;
398
399 spin_lock_bh(&conn->send_lock);
400 rc = smc_cdc_get_free_slot(conn, &wr_buf, &pend);
401 if (rc < 0) {
402 if (rc == -EBUSY) {
403 struct smc_sock *smc =
404 container_of(conn, struct smc_sock, conn);
405
406 if (smc->sk.sk_err == ECONNABORTED) {
407 rc = sock_error(&smc->sk);
408 goto out_unlock;
409 }
410 rc = 0;
411 if (conn->alert_token_local)
412 schedule_delayed_work(&conn->tx_work,
413 SMC_TX_WORK_DELAY);
414 }
415 goto out_unlock;
416 }
417
418 rc = smc_tx_rdma_writes(conn);
419 if (rc) {
420 smc_wr_tx_put_slot(&conn->lgr->lnk[SMC_SINGLE_LINK],
421 (struct smc_wr_tx_pend_priv *)pend);
422 goto out_unlock;
423 }
424
425 rc = smc_cdc_msg_send(conn, wr_buf, pend);
426
427out_unlock:
428 spin_unlock_bh(&conn->send_lock);
429 return rc;
430}
431
432
433
434
435static void smc_tx_work(struct work_struct *work)
436{
437 struct smc_connection *conn = container_of(to_delayed_work(work),
438 struct smc_connection,
439 tx_work);
440 struct smc_sock *smc = container_of(conn, struct smc_sock, conn);
441 int rc;
442
443 lock_sock(&smc->sk);
444 if (smc->sk.sk_err ||
445 !conn->alert_token_local ||
446 conn->local_rx_ctrl.conn_state_flags.peer_conn_abort)
447 goto out;
448
449 rc = smc_tx_sndbuf_nonempty(conn);
450 if (!rc && conn->local_rx_ctrl.prod_flags.write_blocked &&
451 !atomic_read(&conn->bytes_to_rcv))
452 conn->local_rx_ctrl.prod_flags.write_blocked = 0;
453
454out:
455 release_sock(&smc->sk);
456}
457
458void smc_tx_consumer_update(struct smc_connection *conn)
459{
460 union smc_host_cursor cfed, cons;
461 int to_confirm;
462
463 smc_curs_write(&cons,
464 smc_curs_read(&conn->local_tx_ctrl.cons, conn),
465 conn);
466 smc_curs_write(&cfed,
467 smc_curs_read(&conn->rx_curs_confirmed, conn),
468 conn);
469 to_confirm = smc_curs_diff(conn->rmbe_size, &cfed, &cons);
470
471 if (conn->local_rx_ctrl.prod_flags.cons_curs_upd_req ||
472 ((to_confirm > conn->rmbe_update_limit) &&
473 ((to_confirm > (conn->rmbe_size / 2)) ||
474 conn->local_rx_ctrl.prod_flags.write_blocked))) {
475 if ((smc_cdc_get_slot_and_msg_send(conn) < 0) &&
476 conn->alert_token_local) {
477 schedule_delayed_work(&conn->tx_work,
478 SMC_TX_WORK_DELAY);
479 return;
480 }
481 smc_curs_write(&conn->rx_curs_confirmed,
482 smc_curs_read(&conn->local_tx_ctrl.cons, conn),
483 conn);
484 conn->local_rx_ctrl.prod_flags.cons_curs_upd_req = 0;
485 }
486 if (conn->local_rx_ctrl.prod_flags.write_blocked &&
487 !atomic_read(&conn->bytes_to_rcv))
488 conn->local_rx_ctrl.prod_flags.write_blocked = 0;
489}
490
491
492
493
494void smc_tx_init(struct smc_sock *smc)
495{
496 smc->sk.sk_write_space = smc_tx_write_space;
497 INIT_DELAYED_WORK(&smc->conn.tx_work, smc_tx_work);
498 spin_lock_init(&smc->conn.send_lock);
499}
500