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 SDP_MODPARAM_INT(rcvbuf_initial_size, 32 * 1024, 35 "Receive buffer initial size in bytes."); 36 SDP_MODPARAM_SINT(rcvbuf_scale, 0x8, 37 "Receive buffer size scale factor."); 38 39 /* Like tcp_fin - called when SDP_MID_DISCONNECT is received */ 40 static void 41 sdp_handle_disconn(struct sdp_sock *ssk) 42 { 43 44 sdp_dbg(ssk->socket, "%s\n", __func__); 45 46 SDP_WLOCK_ASSERT(ssk); 47 if (TCPS_HAVERCVDFIN(ssk->state) == 0) 48 socantrcvmore(ssk->socket); 49 50 switch (ssk->state) { 51 case TCPS_SYN_RECEIVED: 52 case TCPS_ESTABLISHED: 53 ssk->state = TCPS_CLOSE_WAIT; 54 break; 55 56 case TCPS_FIN_WAIT_1: 57 /* Received a reply FIN - start Infiniband tear down */ 58 sdp_dbg(ssk->socket, 59 "%s: Starting Infiniband tear down sending DREQ\n", 60 __func__); 61 62 sdp_cancel_dreq_wait_timeout(ssk); 63 ssk->qp_active = 0; 64 if (ssk->id) { 65 struct rdma_cm_id *id; 66 67 id = ssk->id; 68 SDP_WUNLOCK(ssk); 69 rdma_disconnect(id); 70 SDP_WLOCK(ssk); 71 } else { 72 sdp_warn(ssk->socket, 73 "%s: ssk->id is NULL\n", __func__); 74 return; 75 } 76 break; 77 case TCPS_TIME_WAIT: 78 /* This is a mutual close situation and we've got the DREQ from 79 the peer before the SDP_MID_DISCONNECT */ 80 break; 81 case TCPS_CLOSED: 82 /* FIN arrived after IB teardown started - do nothing */ 83 sdp_dbg(ssk->socket, "%s: fin in state %s\n", 84 __func__, sdp_state_str(ssk->state)); 85 return; 86 default: 87 sdp_warn(ssk->socket, 88 "%s: FIN in unexpected state. state=%d\n", 89 __func__, ssk->state); 90 break; 91 } 92 } 93 94 static int 95 sdp_post_recv(struct sdp_sock *ssk) 96 { 97 struct sdp_buf *rx_req; 98 int i, rc; 99 u64 addr; 100 struct ib_device *dev; 101 struct ib_recv_wr rx_wr = { NULL }; 102 struct ib_sge ibsge[SDP_MAX_RECV_SGES]; 103 struct ib_sge *sge = ibsge; 104 struct ib_recv_wr *bad_wr; 105 struct mbuf *mb, *m; 106 struct sdp_bsdh *h; 107 int id = ring_head(ssk->rx_ring); 108 109 /* Now, allocate and repost recv */ 110 sdp_prf(ssk->socket, mb, "Posting mb"); 111 mb = m_getm2(NULL, ssk->recv_bytes, M_NOWAIT, MT_DATA, M_PKTHDR); 112 if (mb == NULL) { 113 /* Retry so we can't stall out with no memory. */ 114 if (!rx_ring_posted(ssk)) 115 queue_work(rx_comp_wq, &ssk->rx_comp_work); 116 return -1; 117 } 118 for (m = mb; m != NULL; m = m->m_next) { 119 m->m_len = M_SIZE(m); 120 mb->m_pkthdr.len += m->m_len; 121 } 122 h = mtod(mb, struct sdp_bsdh *); 123 rx_req = ssk->rx_ring.buffer + (id & (SDP_RX_SIZE - 1)); 124 rx_req->mb = mb; 125 dev = ssk->ib_device; 126 for (i = 0; mb != NULL; i++, mb = mb->m_next, sge++) { 127 addr = ib_dma_map_single(dev, mb->m_data, mb->m_len, 128 DMA_TO_DEVICE); 129 /* TODO: proper error handling */ 130 BUG_ON(ib_dma_mapping_error(dev, addr)); 131 BUG_ON(i >= SDP_MAX_RECV_SGES); 132 rx_req->mapping[i] = addr; 133 sge->addr = addr; 134 sge->length = mb->m_len; 135 sge->lkey = ssk->sdp_dev->mr->lkey; 136 } 137 138 rx_wr.next = NULL; 139 rx_wr.wr_id = id | SDP_OP_RECV; 140 rx_wr.sg_list = ibsge; 141 rx_wr.num_sge = i; 142 rc = ib_post_recv(ssk->qp, &rx_wr, &bad_wr); 143 if (unlikely(rc)) { 144 sdp_warn(ssk->socket, "ib_post_recv failed. status %d\n", rc); 145 146 sdp_cleanup_sdp_buf(ssk, rx_req, DMA_FROM_DEVICE); 147 m_freem(mb); 148 149 sdp_notify(ssk, ECONNRESET); 150 151 return -1; 152 } 153 154 atomic_inc(&ssk->rx_ring.head); 155 SDPSTATS_COUNTER_INC(post_recv); 156 157 return 0; 158 } 159 160 static inline int 161 sdp_post_recvs_needed(struct sdp_sock *ssk) 162 { 163 unsigned long bytes_in_process; 164 unsigned long max_bytes; 165 int buffer_size; 166 int posted; 167 168 if (!ssk->qp_active || !ssk->socket) 169 return 0; 170 171 posted = rx_ring_posted(ssk); 172 if (posted >= SDP_RX_SIZE) 173 return 0; 174 if (posted < SDP_MIN_TX_CREDITS) 175 return 1; 176 177 buffer_size = ssk->recv_bytes; 178 max_bytes = max(ssk->socket->so_rcv.sb_hiwat, 179 (1 + SDP_MIN_TX_CREDITS) * buffer_size); 180 max_bytes *= rcvbuf_scale; 181 /* 182 * Compute bytes in the receive queue and socket buffer. 183 */ 184 bytes_in_process = (posted - SDP_MIN_TX_CREDITS) * buffer_size; 185 bytes_in_process += sbused(&ssk->socket->so_rcv); 186 187 return bytes_in_process < max_bytes; 188 } 189 190 static inline void 191 sdp_post_recvs(struct sdp_sock *ssk) 192 { 193 194 while (sdp_post_recvs_needed(ssk)) 195 if (sdp_post_recv(ssk)) 196 return; 197 } 198 199 static inline struct mbuf * 200 sdp_sock_queue_rcv_mb(struct socket *sk, struct mbuf *mb) 201 { 202 struct sdp_sock *ssk = sdp_sk(sk); 203 struct sdp_bsdh *h; 204 205 h = mtod(mb, struct sdp_bsdh *); 206 207 #ifdef SDP_ZCOPY 208 SDP_SKB_CB(mb)->seq = rcv_nxt(ssk); 209 if (h->mid == SDP_MID_SRCAVAIL) { 210 struct sdp_srcah *srcah = (struct sdp_srcah *)(h+1); 211 struct rx_srcavail_state *rx_sa; 212 213 ssk->srcavail_cancel_mseq = 0; 214 215 ssk->rx_sa = rx_sa = RX_SRCAVAIL_STATE(mb) = kzalloc( 216 sizeof(struct rx_srcavail_state), M_NOWAIT); 217 218 rx_sa->mseq = ntohl(h->mseq); 219 rx_sa->used = 0; 220 rx_sa->len = mb_len = ntohl(srcah->len); 221 rx_sa->rkey = ntohl(srcah->rkey); 222 rx_sa->vaddr = be64_to_cpu(srcah->vaddr); 223 rx_sa->flags = 0; 224 225 if (ssk->tx_sa) { 226 sdp_dbg_data(ssk->socket, "got RX SrcAvail while waiting " 227 "for TX SrcAvail. waking up TX SrcAvail" 228 "to be aborted\n"); 229 wake_up(sk->sk_sleep); 230 } 231 232 atomic_add(mb->len, &ssk->rcv_nxt); 233 sdp_dbg_data(sk, "queueing SrcAvail. mb_len = %d vaddr = %lld\n", 234 mb_len, rx_sa->vaddr); 235 } else 236 #endif 237 { 238 atomic_add(mb->m_pkthdr.len, &ssk->rcv_nxt); 239 } 240 241 m_adj(mb, SDP_HEAD_SIZE); 242 SOCKBUF_LOCK(&sk->so_rcv); 243 if (unlikely(h->flags & SDP_OOB_PRES)) 244 sdp_urg(ssk, mb); 245 sbappend_locked(&sk->so_rcv, mb, 0); 246 sorwakeup_locked(sk); 247 return mb; 248 } 249 250 static int 251 sdp_get_recv_bytes(struct sdp_sock *ssk, u32 new_size) 252 { 253 254 return MIN(new_size, SDP_MAX_PACKET); 255 } 256 257 int 258 sdp_init_buffers(struct sdp_sock *ssk, u32 new_size) 259 { 260 261 ssk->recv_bytes = sdp_get_recv_bytes(ssk, new_size); 262 sdp_post_recvs(ssk); 263 264 return 0; 265 } 266 267 int 268 sdp_resize_buffers(struct sdp_sock *ssk, u32 new_size) 269 { 270 u32 curr_size = ssk->recv_bytes; 271 u32 max_size = SDP_MAX_PACKET; 272 273 if (new_size > curr_size && new_size <= max_size) { 274 ssk->recv_bytes = sdp_get_recv_bytes(ssk, new_size); 275 return 0; 276 } 277 return -1; 278 } 279 280 static void 281 sdp_handle_resize_request(struct sdp_sock *ssk, struct sdp_chrecvbuf *buf) 282 { 283 if (sdp_resize_buffers(ssk, ntohl(buf->size)) == 0) 284 ssk->recv_request_head = ring_head(ssk->rx_ring) + 1; 285 else 286 ssk->recv_request_head = ring_tail(ssk->rx_ring); 287 ssk->recv_request = 1; 288 } 289 290 static void 291 sdp_handle_resize_ack(struct sdp_sock *ssk, struct sdp_chrecvbuf *buf) 292 { 293 u32 new_size = ntohl(buf->size); 294 295 if (new_size > ssk->xmit_size_goal) 296 ssk->xmit_size_goal = new_size; 297 } 298 299 static struct mbuf * 300 sdp_recv_completion(struct sdp_sock *ssk, int id) 301 { 302 struct sdp_buf *rx_req; 303 struct ib_device *dev; 304 struct mbuf *mb; 305 306 if (unlikely(id != ring_tail(ssk->rx_ring))) { 307 printk(KERN_WARNING "Bogus recv completion id %d tail %d\n", 308 id, ring_tail(ssk->rx_ring)); 309 return NULL; 310 } 311 312 dev = ssk->ib_device; 313 rx_req = &ssk->rx_ring.buffer[id & (SDP_RX_SIZE - 1)]; 314 mb = rx_req->mb; 315 sdp_cleanup_sdp_buf(ssk, rx_req, DMA_FROM_DEVICE); 316 317 atomic_inc(&ssk->rx_ring.tail); 318 atomic_dec(&ssk->remote_credits); 319 return mb; 320 } 321 322 static void 323 sdp_process_rx_ctl_mb(struct sdp_sock *ssk, struct mbuf *mb) 324 { 325 struct sdp_bsdh *h; 326 struct socket *sk; 327 328 SDP_WLOCK_ASSERT(ssk); 329 330 sk = ssk->socket; 331 h = mtod(mb, struct sdp_bsdh *); 332 switch (h->mid) { 333 case SDP_MID_DATA: 334 case SDP_MID_SRCAVAIL: 335 sdp_dbg(sk, "DATA after socket rcv was shutdown\n"); 336 337 /* got data in RCV_SHUTDOWN */ 338 if (ssk->state == TCPS_FIN_WAIT_1) { 339 sdp_dbg(sk, "RX data when state = FIN_WAIT1\n"); 340 sdp_notify(ssk, ECONNRESET); 341 } 342 343 break; 344 #ifdef SDP_ZCOPY 345 case SDP_MID_RDMARDCOMPL: 346 break; 347 case SDP_MID_SENDSM: 348 sdp_handle_sendsm(ssk, ntohl(h->mseq_ack)); 349 break; 350 case SDP_MID_SRCAVAIL_CANCEL: 351 sdp_dbg_data(sk, "Handling SrcAvailCancel\n"); 352 sdp_prf(sk, NULL, "Handling SrcAvailCancel"); 353 if (ssk->rx_sa) { 354 ssk->srcavail_cancel_mseq = ntohl(h->mseq); 355 ssk->rx_sa->flags |= RX_SA_ABORTED; 356 ssk->rx_sa = NULL; /* TODO: change it into SDP_MID_DATA and get 357 the dirty logic from recvmsg */ 358 } else { 359 sdp_dbg(sk, "Got SrcAvailCancel - " 360 "but no SrcAvail in process\n"); 361 } 362 break; 363 case SDP_MID_SINKAVAIL: 364 sdp_dbg_data(sk, "Got SinkAvail - not supported: ignored\n"); 365 sdp_prf(sk, NULL, "Got SinkAvail - not supported: ignored"); 366 /* FALLTHROUGH */ 367 #endif 368 case SDP_MID_ABORT: 369 sdp_dbg_data(sk, "Handling ABORT\n"); 370 sdp_prf(sk, NULL, "Handling ABORT"); 371 sdp_notify(ssk, ECONNRESET); 372 break; 373 case SDP_MID_DISCONN: 374 sdp_dbg_data(sk, "Handling DISCONN\n"); 375 sdp_prf(sk, NULL, "Handling DISCONN"); 376 sdp_handle_disconn(ssk); 377 break; 378 case SDP_MID_CHRCVBUF: 379 sdp_dbg_data(sk, "Handling RX CHRCVBUF\n"); 380 sdp_handle_resize_request(ssk, (struct sdp_chrecvbuf *)(h+1)); 381 break; 382 case SDP_MID_CHRCVBUF_ACK: 383 sdp_dbg_data(sk, "Handling RX CHRCVBUF_ACK\n"); 384 sdp_handle_resize_ack(ssk, (struct sdp_chrecvbuf *)(h+1)); 385 break; 386 default: 387 /* TODO: Handle other messages */ 388 sdp_warn(sk, "SDP: FIXME MID %d\n", h->mid); 389 break; 390 } 391 m_freem(mb); 392 } 393 394 static int 395 sdp_process_rx_mb(struct sdp_sock *ssk, struct mbuf *mb) 396 { 397 struct socket *sk; 398 struct sdp_bsdh *h; 399 unsigned long mseq_ack; 400 int credits_before; 401 402 h = mtod(mb, struct sdp_bsdh *); 403 sk = ssk->socket; 404 /* 405 * If another thread is in so_pcbfree this may be partially torn 406 * down but no further synchronization is required as the destroying 407 * thread will wait for receive to shutdown before discarding the 408 * socket. 409 */ 410 if (sk == NULL) { 411 m_freem(mb); 412 return 0; 413 } 414 415 SDPSTATS_HIST_LINEAR(credits_before_update, tx_credits(ssk)); 416 417 mseq_ack = ntohl(h->mseq_ack); 418 credits_before = tx_credits(ssk); 419 atomic_set(&ssk->tx_ring.credits, mseq_ack - ring_head(ssk->tx_ring) + 420 1 + ntohs(h->bufs)); 421 if (mseq_ack >= ssk->nagle_last_unacked) 422 ssk->nagle_last_unacked = 0; 423 424 sdp_prf1(ssk->socket, mb, "RX %s +%d c:%d->%d mseq:%d ack:%d\n", 425 mid2str(h->mid), ntohs(h->bufs), credits_before, 426 tx_credits(ssk), ntohl(h->mseq), ntohl(h->mseq_ack)); 427 428 if (unlikely(h->mid == SDP_MID_DATA && 429 mb->m_pkthdr.len == SDP_HEAD_SIZE)) { 430 /* Credit update is valid even after RCV_SHUTDOWN */ 431 m_freem(mb); 432 return 0; 433 } 434 435 if ((h->mid != SDP_MID_DATA && h->mid != SDP_MID_SRCAVAIL) || 436 TCPS_HAVERCVDFIN(ssk->state)) { 437 sdp_prf(sk, NULL, "Control mb - queing to control queue"); 438 #ifdef SDP_ZCOPY 439 if (h->mid == SDP_MID_SRCAVAIL_CANCEL) { 440 sdp_dbg_data(sk, "Got SrcAvailCancel. " 441 "seq: 0x%d seq_ack: 0x%d\n", 442 ntohl(h->mseq), ntohl(h->mseq_ack)); 443 ssk->srcavail_cancel_mseq = ntohl(h->mseq); 444 } 445 446 447 if (h->mid == SDP_MID_RDMARDCOMPL) { 448 struct sdp_rrch *rrch = (struct sdp_rrch *)(h+1); 449 sdp_dbg_data(sk, "RdmaRdCompl message arrived\n"); 450 sdp_handle_rdma_read_compl(ssk, ntohl(h->mseq_ack), 451 ntohl(rrch->len)); 452 } 453 #endif 454 if (mbufq_enqueue(&ssk->rxctlq, mb) != 0) 455 m_freem(mb); 456 return (0); 457 } 458 459 sdp_prf1(sk, NULL, "queueing %s mb\n", mid2str(h->mid)); 460 mb = sdp_sock_queue_rcv_mb(sk, mb); 461 462 463 return 0; 464 } 465 466 /* called only from irq */ 467 static struct mbuf * 468 sdp_process_rx_wc(struct sdp_sock *ssk, struct ib_wc *wc) 469 { 470 struct mbuf *mb; 471 struct sdp_bsdh *h; 472 struct socket *sk = ssk->socket; 473 int mseq; 474 475 mb = sdp_recv_completion(ssk, wc->wr_id); 476 if (unlikely(!mb)) 477 return NULL; 478 479 if (unlikely(wc->status)) { 480 if (ssk->qp_active && sk) { 481 sdp_dbg(sk, "Recv completion with error. " 482 "Status %d, vendor: %d\n", 483 wc->status, wc->vendor_err); 484 sdp_abort(sk); 485 ssk->qp_active = 0; 486 } 487 m_freem(mb); 488 return NULL; 489 } 490 491 sdp_dbg_data(sk, "Recv completion. ID %d Length %d\n", 492 (int)wc->wr_id, wc->byte_len); 493 if (unlikely(wc->byte_len < sizeof(struct sdp_bsdh))) { 494 sdp_warn(sk, "SDP BUG! byte_len %d < %zd\n", 495 wc->byte_len, sizeof(struct sdp_bsdh)); 496 m_freem(mb); 497 return NULL; 498 } 499 /* Use m_adj to trim the tail of data we didn't use. */ 500 m_adj(mb, -(mb->m_pkthdr.len - wc->byte_len)); 501 h = mtod(mb, struct sdp_bsdh *); 502 503 SDP_DUMP_PACKET(ssk->socket, "RX", mb, h); 504 505 ssk->rx_packets++; 506 ssk->rx_bytes += mb->m_pkthdr.len; 507 508 mseq = ntohl(h->mseq); 509 atomic_set(&ssk->mseq_ack, mseq); 510 if (mseq != (int)wc->wr_id) 511 sdp_warn(sk, "SDP BUG! mseq %d != wrid %d\n", 512 mseq, (int)wc->wr_id); 513 514 return mb; 515 } 516 517 /* Wakeup writers if we now have credits. */ 518 static void 519 sdp_bzcopy_write_space(struct sdp_sock *ssk) 520 { 521 struct socket *sk = ssk->socket; 522 523 if (tx_credits(ssk) >= ssk->min_bufs && sk) 524 sowwakeup(sk); 525 } 526 527 /* only from interrupt. */ 528 static int 529 sdp_poll_rx_cq(struct sdp_sock *ssk) 530 { 531 struct ib_cq *cq = ssk->rx_ring.cq; 532 struct ib_wc ibwc[SDP_NUM_WC]; 533 int n, i; 534 int wc_processed = 0; 535 struct mbuf *mb; 536 537 do { 538 n = ib_poll_cq(cq, SDP_NUM_WC, ibwc); 539 for (i = 0; i < n; ++i) { 540 struct ib_wc *wc = &ibwc[i]; 541 542 BUG_ON(!(wc->wr_id & SDP_OP_RECV)); 543 mb = sdp_process_rx_wc(ssk, wc); 544 if (!mb) 545 continue; 546 547 sdp_process_rx_mb(ssk, mb); 548 wc_processed++; 549 } 550 } while (n == SDP_NUM_WC); 551 552 if (wc_processed) 553 sdp_bzcopy_write_space(ssk); 554 555 return wc_processed; 556 } 557 558 static void 559 sdp_rx_comp_work(struct work_struct *work) 560 { 561 struct sdp_sock *ssk = container_of(work, struct sdp_sock, 562 rx_comp_work); 563 564 sdp_prf(ssk->socket, NULL, "%s", __func__); 565 566 SDP_WLOCK(ssk); 567 if (unlikely(!ssk->qp)) { 568 sdp_prf(ssk->socket, NULL, "qp was destroyed"); 569 goto out; 570 } 571 if (unlikely(!ssk->rx_ring.cq)) { 572 sdp_prf(ssk->socket, NULL, "rx_ring.cq is NULL"); 573 goto out; 574 } 575 576 if (unlikely(!ssk->poll_cq)) { 577 struct rdma_cm_id *id = ssk->id; 578 if (id && id->qp) 579 rdma_notify(id, IB_EVENT_COMM_EST); 580 goto out; 581 } 582 583 sdp_do_posts(ssk); 584 out: 585 SDP_WUNLOCK(ssk); 586 } 587 588 void 589 sdp_do_posts(struct sdp_sock *ssk) 590 { 591 struct socket *sk = ssk->socket; 592 int xmit_poll_force; 593 struct mbuf *mb; 594 595 SDP_WLOCK_ASSERT(ssk); 596 if (!ssk->qp_active) { 597 sdp_dbg(sk, "QP is deactivated\n"); 598 return; 599 } 600 601 while ((mb = mbufq_dequeue(&ssk->rxctlq)) != NULL) 602 sdp_process_rx_ctl_mb(ssk, mb); 603 604 if (ssk->state == TCPS_TIME_WAIT) 605 return; 606 607 if (!ssk->rx_ring.cq || !ssk->tx_ring.cq) 608 return; 609 610 sdp_post_recvs(ssk); 611 612 if (tx_ring_posted(ssk)) 613 sdp_xmit_poll(ssk, 1); 614 615 sdp_post_sends(ssk, M_NOWAIT); 616 617 xmit_poll_force = tx_credits(ssk) < SDP_MIN_TX_CREDITS; 618 619 if (credit_update_needed(ssk) || xmit_poll_force) { 620 /* if has pending tx because run out of tx_credits - xmit it */ 621 sdp_prf(sk, NULL, "Processing to free pending sends"); 622 sdp_xmit_poll(ssk, xmit_poll_force); 623 sdp_prf(sk, NULL, "Sending credit update"); 624 sdp_post_sends(ssk, M_NOWAIT); 625 } 626 627 } 628 629 int 630 sdp_process_rx(struct sdp_sock *ssk) 631 { 632 int wc_processed = 0; 633 int credits_before; 634 635 if (!rx_ring_trylock(&ssk->rx_ring)) { 636 sdp_dbg(ssk->socket, "ring destroyed. not polling it\n"); 637 return 0; 638 } 639 640 credits_before = tx_credits(ssk); 641 642 wc_processed = sdp_poll_rx_cq(ssk); 643 sdp_prf(ssk->socket, NULL, "processed %d", wc_processed); 644 645 if (wc_processed) { 646 sdp_prf(ssk->socket, NULL, "credits: %d -> %d", 647 credits_before, tx_credits(ssk)); 648 queue_work(rx_comp_wq, &ssk->rx_comp_work); 649 } 650 sdp_arm_rx_cq(ssk); 651 652 rx_ring_unlock(&ssk->rx_ring); 653 654 return (wc_processed); 655 } 656 657 static void 658 sdp_rx_irq(struct ib_cq *cq, void *cq_context) 659 { 660 struct sdp_sock *ssk; 661 662 ssk = cq_context; 663 KASSERT(cq == ssk->rx_ring.cq, 664 ("%s: mismatched cq on %p", __func__, ssk)); 665 666 SDPSTATS_COUNTER_INC(rx_int_count); 667 668 sdp_prf(sk, NULL, "rx irq"); 669 670 sdp_process_rx(ssk); 671 } 672 673 static 674 void sdp_rx_ring_purge(struct sdp_sock *ssk) 675 { 676 while (rx_ring_posted(ssk) > 0) { 677 struct mbuf *mb; 678 mb = sdp_recv_completion(ssk, ring_tail(ssk->rx_ring)); 679 if (!mb) 680 break; 681 m_freem(mb); 682 } 683 } 684 685 void 686 sdp_rx_ring_init(struct sdp_sock *ssk) 687 { 688 ssk->rx_ring.buffer = NULL; 689 ssk->rx_ring.destroyed = 0; 690 rw_init(&ssk->rx_ring.destroyed_lock, "sdp rx lock"); 691 } 692 693 static void 694 sdp_rx_cq_event_handler(struct ib_event *event, void *data) 695 { 696 } 697 698 int 699 sdp_rx_ring_create(struct sdp_sock *ssk, struct ib_device *device) 700 { 701 struct ib_cq *rx_cq; 702 int rc = 0; 703 704 sdp_dbg(ssk->socket, "rx ring created"); 705 INIT_WORK(&ssk->rx_comp_work, sdp_rx_comp_work); 706 atomic_set(&ssk->rx_ring.head, 1); 707 atomic_set(&ssk->rx_ring.tail, 1); 708 709 ssk->rx_ring.buffer = malloc(sizeof(*ssk->rx_ring.buffer) * SDP_RX_SIZE, 710 M_SDP, M_WAITOK); 711 712 rx_cq = ib_create_cq(device, sdp_rx_irq, sdp_rx_cq_event_handler, 713 ssk, SDP_RX_SIZE, 0); 714 if (IS_ERR(rx_cq)) { 715 rc = PTR_ERR(rx_cq); 716 sdp_warn(ssk->socket, "Unable to allocate RX CQ: %d.\n", rc); 717 goto err_cq; 718 } 719 720 sdp_sk(ssk->socket)->rx_ring.cq = rx_cq; 721 sdp_arm_rx_cq(ssk); 722 723 return 0; 724 725 err_cq: 726 free(ssk->rx_ring.buffer, M_SDP); 727 ssk->rx_ring.buffer = NULL; 728 return rc; 729 } 730 731 void 732 sdp_rx_ring_destroy(struct sdp_sock *ssk) 733 { 734 735 cancel_work_sync(&ssk->rx_comp_work); 736 rx_ring_destroy_lock(&ssk->rx_ring); 737 738 if (ssk->rx_ring.buffer) { 739 sdp_rx_ring_purge(ssk); 740 free(ssk->rx_ring.buffer, M_SDP); 741 ssk->rx_ring.buffer = NULL; 742 } 743 744 if (ssk->rx_ring.cq) { 745 if (ib_destroy_cq(ssk->rx_ring.cq)) { 746 sdp_warn(ssk->socket, "destroy cq(%p) failed\n", 747 ssk->rx_ring.cq); 748 } else { 749 ssk->rx_ring.cq = NULL; 750 } 751 } 752 753 WARN_ON(ring_head(ssk->rx_ring) != ring_tail(ssk->rx_ring)); 754 } 755