1 /*
2 * Copyright (c) 2009 Mellanox Technologies Ltd. All rights reserved.
3 *
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the
8 * OpenIB.org BSD license below:
9 *
10 * Redistribution and use in source and binary forms, with or
11 * without modification, are permitted provided that the following
12 * conditions are met:
13 *
14 * - Redistributions of source code must retain the above
15 * copyright notice, this list of conditions and the following
16 * disclaimer.
17 *
18 * - Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials
21 * provided with the distribution.
22 *
23 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
27 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
28 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
29 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
30 * SOFTWARE.
31 */
32 #include "sdp.h"
33
34 #define sdp_cnt(var) do { (var)++; } while (0)
35
36 SDP_MODPARAM_SINT(sdp_keepalive_probes_sent, 0,
37 "Total number of keepalive probes sent.");
38
39 static int sdp_process_tx_cq(struct sdp_sock *ssk);
40 static void sdp_poll_tx_timeout(void *data);
41
42 int
sdp_xmit_poll(struct sdp_sock * ssk,int force)43 sdp_xmit_poll(struct sdp_sock *ssk, int force)
44 {
45 int wc_processed = 0;
46
47 SDP_WLOCK_ASSERT(ssk);
48 sdp_prf(ssk->socket, NULL, "%s", __func__);
49
50 /* If we don't have a pending timer, set one up to catch our recent
51 post in case the interface becomes idle */
52 if (!callout_pending(&ssk->tx_ring.timer))
53 callout_reset(&ssk->tx_ring.timer, SDP_TX_POLL_TIMEOUT,
54 sdp_poll_tx_timeout, ssk);
55
56 /* Poll the CQ every SDP_TX_POLL_MODER packets */
57 if (force || (++ssk->tx_ring.poll_cnt & (SDP_TX_POLL_MODER - 1)) == 0)
58 wc_processed = sdp_process_tx_cq(ssk);
59
60 return wc_processed;
61 }
62
63 void
sdp_post_send(struct sdp_sock * ssk,struct mbuf * mb)64 sdp_post_send(struct sdp_sock *ssk, struct mbuf *mb)
65 {
66 struct sdp_buf *tx_req;
67 struct sdp_bsdh *h;
68 unsigned long mseq;
69 struct ib_device *dev;
70 struct ib_send_wr *bad_wr;
71 struct ib_sge ibsge[SDP_MAX_SEND_SGES];
72 struct ib_sge *sge;
73 struct ib_send_wr tx_wr = { NULL };
74 int i, rc;
75 u64 addr;
76
77 SDPSTATS_COUNTER_MID_INC(post_send, h->mid);
78 SDPSTATS_HIST(send_size, mb->len);
79
80 if (!ssk->qp_active) {
81 m_freem(mb);
82 return;
83 }
84
85 mseq = ring_head(ssk->tx_ring);
86 h = mtod(mb, struct sdp_bsdh *);
87 ssk->tx_packets++;
88 ssk->tx_bytes += mb->m_pkthdr.len;
89
90 #ifdef SDP_ZCOPY
91 if (unlikely(h->mid == SDP_MID_SRCAVAIL)) {
92 struct tx_srcavail_state *tx_sa = TX_SRCAVAIL_STATE(mb);
93 if (ssk->tx_sa != tx_sa) {
94 sdp_dbg_data(ssk->socket, "SrcAvail cancelled "
95 "before being sent!\n");
96 WARN_ON(1);
97 m_freem(mb);
98 return;
99 }
100 TX_SRCAVAIL_STATE(mb)->mseq = mseq;
101 }
102 #endif
103
104 if (unlikely(mb->m_flags & M_URG))
105 h->flags = SDP_OOB_PRES | SDP_OOB_PEND;
106 else
107 h->flags = 0;
108
109 mb->m_flags |= M_RDONLY; /* Don't allow compression once sent. */
110 h->bufs = htons(rx_ring_posted(ssk));
111 h->len = htonl(mb->m_pkthdr.len);
112 h->mseq = htonl(mseq);
113 h->mseq_ack = htonl(mseq_ack(ssk));
114
115 sdp_prf1(ssk->socket, mb, "TX: %s bufs: %d mseq:%ld ack:%d",
116 mid2str(h->mid), rx_ring_posted(ssk), mseq,
117 ntohl(h->mseq_ack));
118
119 SDP_DUMP_PACKET(ssk->socket, "TX", mb, h);
120
121 tx_req = &ssk->tx_ring.buffer[mseq & (SDP_TX_SIZE - 1)];
122 tx_req->mb = mb;
123 dev = ssk->ib_device;
124 sge = &ibsge[0];
125 for (i = 0; mb != NULL; i++, mb = mb->m_next, sge++) {
126 addr = ib_dma_map_single(dev, mb->m_data, mb->m_len,
127 DMA_TO_DEVICE);
128 /* TODO: proper error handling */
129 BUG_ON(ib_dma_mapping_error(dev, addr));
130 BUG_ON(i >= SDP_MAX_SEND_SGES);
131 tx_req->mapping[i] = addr;
132 sge->addr = addr;
133 sge->length = mb->m_len;
134 sge->lkey = ssk->sdp_dev->mr->lkey;
135 }
136 tx_wr.next = NULL;
137 tx_wr.wr_id = mseq | SDP_OP_SEND;
138 tx_wr.sg_list = ibsge;
139 tx_wr.num_sge = i;
140 tx_wr.opcode = IB_WR_SEND;
141 tx_wr.send_flags = IB_SEND_SIGNALED;
142 if (unlikely(tx_req->mb->m_flags & M_URG))
143 tx_wr.send_flags |= IB_SEND_SOLICITED;
144
145 rc = ib_post_send(ssk->qp, &tx_wr, &bad_wr);
146 if (unlikely(rc)) {
147 sdp_dbg(ssk->socket,
148 "ib_post_send failed with status %d.\n", rc);
149
150 sdp_cleanup_sdp_buf(ssk, tx_req, DMA_TO_DEVICE);
151
152 sdp_notify(ssk, ECONNRESET);
153 m_freem(tx_req->mb);
154 return;
155 }
156
157 atomic_inc(&ssk->tx_ring.head);
158 atomic_dec(&ssk->tx_ring.credits);
159 atomic_set(&ssk->remote_credits, rx_ring_posted(ssk));
160
161 return;
162 }
163
164 static struct mbuf *
sdp_send_completion(struct sdp_sock * ssk,int mseq)165 sdp_send_completion(struct sdp_sock *ssk, int mseq)
166 {
167 struct ib_device *dev;
168 struct sdp_buf *tx_req;
169 struct mbuf *mb = NULL;
170 struct sdp_tx_ring *tx_ring = &ssk->tx_ring;
171
172 if (unlikely(mseq != ring_tail(*tx_ring))) {
173 printk(KERN_WARNING "Bogus send completion id %d tail %d\n",
174 mseq, ring_tail(*tx_ring));
175 goto out;
176 }
177
178 dev = ssk->ib_device;
179 tx_req = &tx_ring->buffer[mseq & (SDP_TX_SIZE - 1)];
180 mb = tx_req->mb;
181 sdp_cleanup_sdp_buf(ssk, tx_req, DMA_TO_DEVICE);
182
183 #ifdef SDP_ZCOPY
184 /* TODO: AIO and real zcopy code; add their context support here */
185 if (BZCOPY_STATE(mb))
186 BZCOPY_STATE(mb)->busy--;
187 #endif
188
189 atomic_inc(&tx_ring->tail);
190
191 out:
192 return mb;
193 }
194
195 static int
sdp_handle_send_comp(struct sdp_sock * ssk,struct ib_wc * wc)196 sdp_handle_send_comp(struct sdp_sock *ssk, struct ib_wc *wc)
197 {
198 struct mbuf *mb = NULL;
199 struct sdp_bsdh *h;
200
201 if (unlikely(wc->status)) {
202 if (wc->status != IB_WC_WR_FLUSH_ERR) {
203 sdp_prf(ssk->socket, mb, "Send completion with error. "
204 "Status %d", wc->status);
205 sdp_dbg_data(ssk->socket, "Send completion with error. "
206 "Status %d\n", wc->status);
207 sdp_notify(ssk, ECONNRESET);
208 }
209 }
210
211 mb = sdp_send_completion(ssk, wc->wr_id);
212 if (unlikely(!mb))
213 return -1;
214
215 h = mtod(mb, struct sdp_bsdh *);
216 sdp_prf1(ssk->socket, mb, "tx completion. mseq:%d", ntohl(h->mseq));
217 sdp_dbg(ssk->socket, "tx completion. %p %d mseq:%d",
218 mb, mb->m_pkthdr.len, ntohl(h->mseq));
219 m_freem(mb);
220
221 return 0;
222 }
223
224 static inline void
sdp_process_tx_wc(struct sdp_sock * ssk,struct ib_wc * wc)225 sdp_process_tx_wc(struct sdp_sock *ssk, struct ib_wc *wc)
226 {
227
228 if (likely(wc->wr_id & SDP_OP_SEND)) {
229 sdp_handle_send_comp(ssk, wc);
230 return;
231 }
232
233 #ifdef SDP_ZCOPY
234 if (wc->wr_id & SDP_OP_RDMA) {
235 /* TODO: handle failed RDMA read cqe */
236
237 sdp_dbg_data(ssk->socket,
238 "TX comp: RDMA read. status: %d\n", wc->status);
239 sdp_prf1(sk, NULL, "TX comp: RDMA read");
240
241 if (!ssk->tx_ring.rdma_inflight) {
242 sdp_warn(ssk->socket, "ERROR: unexpected RDMA read\n");
243 return;
244 }
245
246 if (!ssk->tx_ring.rdma_inflight->busy) {
247 sdp_warn(ssk->socket,
248 "ERROR: too many RDMA read completions\n");
249 return;
250 }
251
252 /* Only last RDMA read WR is signalled. Order is guaranteed -
253 * therefore if Last RDMA read WR is completed - all other
254 * have, too */
255 ssk->tx_ring.rdma_inflight->busy = 0;
256 sowwakeup(ssk->socket);
257 sdp_dbg_data(ssk->socket, "woke up sleepers\n");
258 return;
259 }
260 #endif
261
262 /* Keepalive probe sent cleanup */
263 sdp_cnt(sdp_keepalive_probes_sent);
264
265 if (likely(!wc->status))
266 return;
267
268 sdp_dbg(ssk->socket, " %s consumes KEEPALIVE status %d\n",
269 __func__, wc->status);
270
271 if (wc->status == IB_WC_WR_FLUSH_ERR)
272 return;
273
274 sdp_notify(ssk, ECONNRESET);
275 }
276
277 static int
sdp_process_tx_cq(struct sdp_sock * ssk)278 sdp_process_tx_cq(struct sdp_sock *ssk)
279 {
280 struct ib_wc ibwc[SDP_NUM_WC];
281 int n, i;
282 int wc_processed = 0;
283
284 SDP_WLOCK_ASSERT(ssk);
285
286 if (!ssk->tx_ring.cq) {
287 sdp_dbg(ssk->socket, "tx irq on destroyed tx_cq\n");
288 return 0;
289 }
290
291 do {
292 n = ib_poll_cq(ssk->tx_ring.cq, SDP_NUM_WC, ibwc);
293 for (i = 0; i < n; ++i) {
294 sdp_process_tx_wc(ssk, ibwc + i);
295 wc_processed++;
296 }
297 } while (n == SDP_NUM_WC);
298
299 if (wc_processed) {
300 sdp_post_sends(ssk, M_NOWAIT);
301 sdp_prf1(sk, NULL, "Waking sendmsg. inflight=%d",
302 (u32) tx_ring_posted(ssk));
303 sowwakeup(ssk->socket);
304 }
305
306 return wc_processed;
307 }
308
309 static void
sdp_poll_tx(struct sdp_sock * ssk)310 sdp_poll_tx(struct sdp_sock *ssk)
311 {
312 struct socket *sk = ssk->socket;
313 u32 inflight, wc_processed;
314
315 sdp_prf1(ssk->socket, NULL, "TX timeout: inflight=%d, head=%d tail=%d",
316 (u32) tx_ring_posted(ssk),
317 ring_head(ssk->tx_ring), ring_tail(ssk->tx_ring));
318
319 if (unlikely(ssk->state == TCPS_CLOSED)) {
320 sdp_warn(sk, "Socket is closed\n");
321 goto out;
322 }
323
324 wc_processed = sdp_process_tx_cq(ssk);
325 if (!wc_processed)
326 SDPSTATS_COUNTER_INC(tx_poll_miss);
327 else
328 SDPSTATS_COUNTER_INC(tx_poll_hit);
329
330 inflight = (u32) tx_ring_posted(ssk);
331 sdp_prf1(ssk->socket, NULL, "finished tx proccessing. inflight = %d",
332 inflight);
333
334 /* If there are still packets in flight and the timer has not already
335 * been scheduled by the Tx routine then schedule it here to guarantee
336 * completion processing of these packets */
337 if (inflight)
338 callout_reset(&ssk->tx_ring.timer, SDP_TX_POLL_TIMEOUT,
339 sdp_poll_tx_timeout, ssk);
340 out:
341 #ifdef SDP_ZCOPY
342 if (ssk->tx_ring.rdma_inflight && ssk->tx_ring.rdma_inflight->busy) {
343 sdp_prf1(sk, NULL, "RDMA is inflight - arming irq");
344 sdp_arm_tx_cq(ssk);
345 }
346 #endif
347 return;
348 }
349
350 static void
sdp_poll_tx_timeout(void * data)351 sdp_poll_tx_timeout(void *data)
352 {
353 struct sdp_sock *ssk = (struct sdp_sock *)data;
354
355 if (!callout_active(&ssk->tx_ring.timer))
356 return;
357 callout_deactivate(&ssk->tx_ring.timer);
358 sdp_poll_tx(ssk);
359 }
360
361 static void
sdp_tx_irq(struct ib_cq * cq,void * cq_context)362 sdp_tx_irq(struct ib_cq *cq, void *cq_context)
363 {
364 struct sdp_sock *ssk;
365
366 ssk = cq_context;
367 sdp_prf1(ssk->socket, NULL, "tx irq");
368 sdp_dbg_data(ssk->socket, "Got tx comp interrupt\n");
369 SDPSTATS_COUNTER_INC(tx_int_count);
370 SDP_WLOCK(ssk);
371 sdp_poll_tx(ssk);
372 SDP_WUNLOCK(ssk);
373 }
374
375 static
sdp_tx_ring_purge(struct sdp_sock * ssk)376 void sdp_tx_ring_purge(struct sdp_sock *ssk)
377 {
378 while (tx_ring_posted(ssk)) {
379 struct mbuf *mb;
380 mb = sdp_send_completion(ssk, ring_tail(ssk->tx_ring));
381 if (!mb)
382 break;
383 m_freem(mb);
384 }
385 }
386
387 void
sdp_post_keepalive(struct sdp_sock * ssk)388 sdp_post_keepalive(struct sdp_sock *ssk)
389 {
390 int rc;
391 struct ib_send_wr wr, *bad_wr;
392
393 sdp_dbg(ssk->socket, "%s\n", __func__);
394
395 memset(&wr, 0, sizeof(wr));
396
397 wr.next = NULL;
398 wr.wr_id = 0;
399 wr.sg_list = NULL;
400 wr.num_sge = 0;
401 wr.opcode = IB_WR_RDMA_WRITE;
402
403 rc = ib_post_send(ssk->qp, &wr, &bad_wr);
404 if (rc) {
405 sdp_dbg(ssk->socket,
406 "ib_post_keepalive failed with status %d.\n", rc);
407 sdp_notify(ssk, ECONNRESET);
408 }
409
410 sdp_cnt(sdp_keepalive_probes_sent);
411 }
412
413 static void
sdp_tx_cq_event_handler(struct ib_event * event,void * data)414 sdp_tx_cq_event_handler(struct ib_event *event, void *data)
415 {
416 }
417
418 int
sdp_tx_ring_create(struct sdp_sock * ssk,struct ib_device * device)419 sdp_tx_ring_create(struct sdp_sock *ssk, struct ib_device *device)
420 {
421 struct ib_cq *tx_cq;
422 int rc = 0;
423
424 sdp_dbg(ssk->socket, "tx ring create\n");
425 callout_init_rw(&ssk->tx_ring.timer, &ssk->lock, 0);
426 callout_init_rw(&ssk->nagle_timer, &ssk->lock, 0);
427 atomic_set(&ssk->tx_ring.head, 1);
428 atomic_set(&ssk->tx_ring.tail, 1);
429
430 ssk->tx_ring.buffer = kzalloc(
431 sizeof *ssk->tx_ring.buffer * SDP_TX_SIZE, GFP_KERNEL);
432 if (!ssk->tx_ring.buffer) {
433 rc = -ENOMEM;
434 sdp_warn(ssk->socket, "Can't allocate TX Ring size %zd.\n",
435 sizeof(*ssk->tx_ring.buffer) * SDP_TX_SIZE);
436
437 goto out;
438 }
439
440 tx_cq = ib_create_cq(device, sdp_tx_irq, sdp_tx_cq_event_handler,
441 ssk, SDP_TX_SIZE, IB_CQ_VECTOR_LEAST_ATTACHED);
442
443 if (IS_ERR(tx_cq)) {
444 rc = PTR_ERR(tx_cq);
445 sdp_warn(ssk->socket, "Unable to allocate TX CQ: %d.\n", rc);
446 goto err_cq;
447 }
448 ssk->tx_ring.cq = tx_cq;
449 ssk->tx_ring.poll_cnt = 0;
450 sdp_arm_tx_cq(ssk);
451
452 return 0;
453
454 err_cq:
455 kfree(ssk->tx_ring.buffer);
456 ssk->tx_ring.buffer = NULL;
457 out:
458 return rc;
459 }
460
461 void
sdp_tx_ring_destroy(struct sdp_sock * ssk)462 sdp_tx_ring_destroy(struct sdp_sock *ssk)
463 {
464
465 sdp_dbg(ssk->socket, "tx ring destroy\n");
466 SDP_WLOCK(ssk);
467 callout_stop(&ssk->tx_ring.timer);
468 callout_stop(&ssk->nagle_timer);
469 SDP_WUNLOCK(ssk);
470 callout_drain(&ssk->tx_ring.timer);
471 callout_drain(&ssk->nagle_timer);
472
473 if (ssk->tx_ring.buffer) {
474 sdp_tx_ring_purge(ssk);
475
476 kfree(ssk->tx_ring.buffer);
477 ssk->tx_ring.buffer = NULL;
478 }
479
480 if (ssk->tx_ring.cq) {
481 if (ib_destroy_cq(ssk->tx_ring.cq)) {
482 sdp_warn(ssk->socket, "destroy cq(%p) failed\n",
483 ssk->tx_ring.cq);
484 } else {
485 ssk->tx_ring.cq = NULL;
486 }
487 }
488
489 WARN_ON(ring_head(ssk->tx_ring) != ring_tail(ssk->tx_ring));
490 }
491