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