| 763 | } |
| 764 | |
| 765 | void _processStep() { |
| 766 | LOG_DEBUG(_env, "processing step %s", _step); |
| 767 | // finished txns |
| 768 | for (const auto& [txnId, resp]: _step.finishedTxns) { |
| 769 | LOG_DEBUG(_env, "txn %s finished", txnId); |
| 770 | // we need to send the response back to the client |
| 771 | auto inFlight = _inFlightTxns.find(txnId); |
| 772 | if (inFlight->second.hasClient) { |
| 773 | _shared.timingsTotal[(int)inFlight->second.kind].add(ternNow() - inFlight->second.receivedAt); |
| 774 | _shared.errors[(int)inFlight->second.kind].add(resp.kind() != CDCMessageKind::ERROR ? TernError::NO_ERROR : resp.getError()); |
| 775 | CDCRespMsg respMsg; |
| 776 | respMsg.id = inFlight->second.cdcRequestId; |
| 777 | respMsg.body = std::move(resp); |
| 778 | LOG_DEBUG(_env, "sending response with req id %s, kind %s, back to %s", inFlight->second.cdcRequestId, inFlight->second.kind, inFlight->second.clientAddr); |
| 779 | _packCDCResponse(inFlight->second.sockIx, inFlight->second.clientAddr, inFlight->second.kind, respMsg); |
| 780 | _inFlightCDCReqs.erase(InFlightCDCRequestKey(inFlight->second.cdcRequestId, inFlight->second.clientAddr)); |
| 781 | } |
| 782 | _inFlightTxns.erase(inFlight); |
| 783 | _updateInFlightTxns(); |
| 784 | } |
| 785 | // in flight txns |
| 786 | for (const auto& [txnId, shardReq]: _step.runningTxns) { |
| 787 | CDCShardReq prevReq; |
| 788 | LOG_TRACE(_env, "txn %s needs shard %s, req %s", txnId, shardReq.shid, shardReq.req); |
| 789 | CdcToShardReqMsg shardReqMsg; |
| 790 | |
| 791 | // Do not allocate new req id for repeated requests, so that we'll just accept |
| 792 | // the first one that comes back. There's a chance for the txnId to not be here |
| 793 | // yet: if we have just restarted the CDC. In this case we fill it in here, but |
| 794 | // obviously without client addr. |
| 795 | auto inFlightTxn = _inFlightTxns.find(txnId); |
| 796 | if (inFlightTxn == _inFlightTxns.end()) { |
| 797 | LOG_INFO(_env, "Could not find in-flight transaction %s, this might be because the CDC was restarted in the middle of a transaction.", txnId); |
| 798 | InFlightCDCRequest req; |
| 799 | req.hasClient = false; |
| 800 | req.lastSentRequestId = _freshShardReqId(); |
| 801 | inFlightTxn = _inFlightTxns.emplace(txnId, req).first; |
| 802 | shardReqMsg.id = req.lastSentRequestId; |
| 803 | _updateInFlightTxns(); |
| 804 | } else if (shardReq.repeated) { |
| 805 | shardReqMsg.id = inFlightTxn->second.lastSentRequestId; |
| 806 | } else { |
| 807 | shardReqMsg.id = _freshShardReqId(); |
| 808 | } |
| 809 | shardReqMsg.body = shardReq.req; |
| 810 | // Pack |
| 811 | _shared.shardsMutex.lock(); |
| 812 | ShardInfo shardInfo = _shared.shards[shardReq.shid.u8]; |
| 813 | _shared.shardsMutex.unlock(); |
| 814 | |
| 815 | LOG_DEBUG(_env, "sending request for txn %s with req id %s to shard %s (%s)", txnId, shardReqMsg.id, shardReq.shid, shardInfo.addrs); |
| 816 | _shardSender.prepareOutgoingMessage(_env, _shared.socks[SHARD_SOCK].addr(), shardInfo.addrs, [this, &shardReqMsg](BincodeBuf& bbuf) { |
| 817 | shardReqMsg.pack(bbuf, _expandedCDCKey); |
| 818 | }); |
| 819 | // Record the in-flight req |
| 820 | _inFlightShardReqs.insert(shardReqMsg.id, InFlightShardRequest{ |
| 821 | .txnId = txnId, |
| 822 | .sentAt = ternNow(), |
no test coverage detected