| 56 | } |
| 57 | |
| 58 | folly::SemiFuture<Response> request( |
| 59 | uint32_t maxBytes, |
| 60 | uint32_t maxWaitSeconds) override { |
| 61 | ++numRequests_; |
| 62 | |
| 63 | auto promise = BoltPromise<Response>("LocalExchangeSource::request"); |
| 64 | auto future = promise.getSemiFuture(); |
| 65 | |
| 66 | promise_ = std::move(promise); |
| 67 | |
| 68 | auto buffers = OutputBufferManager::getInstance().lock(); |
| 69 | BOLT_CHECK_NOT_NULL(buffers, "invalid OutputBufferManager"); |
| 70 | BOLT_CHECK(requestPending_); |
| 71 | auto requestedSequence = sequence_; |
| 72 | auto self = shared_from_this(); |
| 73 | // Since this lambda may outlive 'this', we need to capture a |
| 74 | // shared_ptr to the current object (self). |
| 75 | auto resultCallback = [self, requestedSequence, buffers, this]( |
| 76 | std::vector<std::unique_ptr<folly::IOBuf>> data, |
| 77 | int64_t sequence) { |
| 78 | { |
| 79 | std::lock_guard<std::mutex> l(timeoutMutex_); |
| 80 | // This function is called either for a result or timeout. Only the |
| 81 | // first of these calls has an effect. |
| 82 | auto iter = timeouts_.find(self); |
| 83 | if (iter != timeouts_.end()) { |
| 84 | timeouts_.erase(iter); |
| 85 | } else { |
| 86 | return; |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | if (requestedSequence > sequence && !data.empty()) { |
| 91 | VLOG(2) << "Receives earlier sequence than requested: task " << taskId_ |
| 92 | << ", destination " << destination_ << ", requested " |
| 93 | << sequence << ", received " << requestedSequence; |
| 94 | int64_t nExtra = requestedSequence - sequence; |
| 95 | BOLT_CHECK(nExtra < data.size()); |
| 96 | data.erase(data.begin(), data.begin() + nExtra); |
| 97 | sequence = requestedSequence; |
| 98 | } |
| 99 | if (data.empty()) { |
| 100 | sequence = requestedSequence; |
| 101 | } |
| 102 | std::vector<std::unique_ptr<SerializedPage>> pages; |
| 103 | bool atEnd = false; |
| 104 | int64_t totalBytes = 0; |
| 105 | for (auto& inputPage : data) { |
| 106 | if (!inputPage) { |
| 107 | atEnd = true; |
| 108 | // Keep looping, there could be extra end markers. |
| 109 | continue; |
| 110 | } |
| 111 | totalBytes += inputPage->length(); |
| 112 | inputPage->unshare(); |
| 113 | pages.push_back(std::make_unique<SerializedPage>(std::move(inputPage))); |
| 114 | inputPage = nullptr; |
| 115 | } |
nothing calls this directly
no test coverage detected