| 578 | } |
| 579 | |
| 580 | void ParallelChannel::CallMethod( |
| 581 | const google::protobuf::MethodDescriptor* method, |
| 582 | google::protobuf::RpcController* cntl_base, |
| 583 | const google::protobuf::Message* request, |
| 584 | google::protobuf::Message* response, |
| 585 | google::protobuf::Closure* done) { |
| 586 | Controller* cntl = static_cast<Controller*>(cntl_base); |
| 587 | cntl->OnRPCBegin(butil::gettimeofday_us()); |
| 588 | // Make sure cntl->sub_count() always equal #sub-channels |
| 589 | const int nchan = _chans.size(); |
| 590 | cntl->_pchan_sub_count = nchan; |
| 591 | |
| 592 | const CallId cid = cntl->call_id(); |
| 593 | const int rc = bthread_id_lock(cid, NULL); |
| 594 | if (rc != 0) { |
| 595 | CHECK_EQ(EINVAL, rc); |
| 596 | if (!cntl->FailedInline()) { |
| 597 | cntl->SetFailed(EINVAL, "Fail to lock call_id=%" PRId64, cid.value); |
| 598 | } |
| 599 | LOG_IF(ERROR, cntl->is_used_by_rpc()) |
| 600 | << "Controller=" << cntl << " was used by another RPC before. " |
| 601 | "Did you forget to Reset() it before reuse?"; |
| 602 | // Have to run done in-place. |
| 603 | // Read comment in CallMethod() in channel.cpp for details. |
| 604 | if (done) { |
| 605 | done->Run(); |
| 606 | } |
| 607 | return; |
| 608 | } |
| 609 | cntl->set_used_by_rpc(); |
| 610 | |
| 611 | ParallelChannelDone* d = NULL; |
| 612 | int ndone = nchan; |
| 613 | int fail_limit = 1; |
| 614 | int success_limit = 1; |
| 615 | Controller::ClientSettings settings{}; |
| 616 | DEFINE_SMALL_ARRAY(SubCall, aps, nchan, 64); |
| 617 | |
| 618 | if (cntl->FailedInline()) { |
| 619 | // The call_id is cancelled before RPC. |
| 620 | goto FAIL; |
| 621 | } |
| 622 | // we don't support http whose response is NULL. |
| 623 | if (response == NULL) { |
| 624 | cntl->SetFailed(EINVAL, "response must be non-NULL"); |
| 625 | goto FAIL; |
| 626 | } |
| 627 | if (nchan == 0) { |
| 628 | cntl->SetFailed(EPERM, "No channels added"); |
| 629 | goto FAIL; |
| 630 | } |
| 631 | |
| 632 | for (int i = 0; i < nchan; ++i) { |
| 633 | SubChan& sub_chan = _chans[i]; |
| 634 | if (sub_chan.call_mapper != NULL) { |
| 635 | aps[i] = sub_chan.call_mapper->Map(i, nchan, method, request, response); |
| 636 | // Test is_skip first because it implies is_bad. |
| 637 | if (aps[i].is_skip()) { |
nothing calls this directly
no test coverage detected