| 602 | } |
| 603 | } |
| 604 | bool Step(NHPTimer::STime tCurrent) override { |
| 605 | bool rv = false; |
| 606 | // only have to process completions, everything is done on completion of something |
| 607 | ibv_wc wcArr[10]; |
| 608 | for (;;) { |
| 609 | int wcCount = CQ->Poll(wcArr, Y_ARRAY_SIZE(wcArr)); |
| 610 | if (wcCount == 0) { |
| 611 | break; |
| 612 | } |
| 613 | rv = true; |
| 614 | for (int z = 0; z < wcCount; ++z) { |
| 615 | ibv_wc& wc = wcArr[z]; |
| 616 | if (wc.opcode & IBV_WC_RECV) { |
| 617 | // received msg |
| 618 | if ((int)wc.qp_num == WelcomeQPN) { |
| 619 | if (wc.status != IBV_WC_SUCCESS) { |
| 620 | Y_ABORT_UNLESS(0, "ud recv op completed with error %d\n", (int)wc.status); |
| 621 | } |
| 622 | Y_ASSERT(wc.opcode == IBV_WC_RECV | IBV_WC_SEND); |
| 623 | ParseWelcomePacket(&wc); |
| 624 | } else { |
| 625 | ParsePacket(&wc, tCurrent); |
| 626 | } |
| 627 | } else { |
| 628 | // send completion |
| 629 | if ((int)wc.qp_num == WelcomeQPN) { |
| 630 | // ok |
| 631 | BP.FreeBuf(wc.wr_id); |
| 632 | } else { |
| 633 | OnComplete(&wc, tCurrent); |
| 634 | } |
| 635 | } |
| 636 | } |
| 637 | } |
| 638 | { |
| 639 | TIntrusivePtr<TIBMemBlock> memBlock; |
| 640 | i64 msgHandle; |
| 641 | TIntrusivePtr<TIBPeer> peer; |
| 642 | while (CopyResults->GetCopyResult(&memBlock, &msgHandle, &peer)) { |
| 643 | if (peer->GetState() != IIBPeer::OK) { |
| 644 | continue; |
| 645 | } |
| 646 | TDeque<TQueuedSend>::iterator z = peer->GetSend(msgHandle); |
| 647 | if (z == peer->SendQueue.end()) { |
| 648 | Y_ABORT_UNLESS(0, "peer %p, copy completed, msg %d not found?\n", peer.Get(), (int)msgHandle); |
| 649 | continue; |
| 650 | } |
| 651 | TQueuedSend& qs = *z; |
| 652 | qs.MemBlock = memBlock; |
| 653 | if (qs.RemoteAddr != 0) { |
| 654 | peer->PostRDMA(qs); |
| 655 | } |
| 656 | rv = true; |
| 657 | } |
| 658 | } |
| 659 | { |
| 660 | NHPTimer::STime t1 = LastCheckTime; |
| 661 | if (NHPTimer::GetTimePassed(&t1) > CHANNEL_CHECK_INTERVAL) { |