| 1082 | } |
| 1083 | |
| 1084 | Reference<ILogSystem::IPeekCursor> TagPartitionedLogSystem::peekTxs(UID dbgid, |
| 1085 | Version begin, |
| 1086 | int8_t peekLocality, |
| 1087 | Version localEnd, |
| 1088 | bool canDiscardPopped) { |
| 1089 | Version end = getEnd(); |
| 1090 | if (!tLogs.size()) { |
| 1091 | TraceEvent("TLogPeekTxsNoLogs", dbgid).log(); |
| 1092 | return makeReference<ILogSystem::ServerPeekCursor>( |
| 1093 | Reference<AsyncVar<OptionalInterface<TLogInterface>>>(), txsTag, begin, end, false, false); |
| 1094 | } |
| 1095 | TraceEvent("TLogPeekTxs", dbgid) |
| 1096 | .detail("Begin", begin) |
| 1097 | .detail("End", end) |
| 1098 | .detail("LocalEnd", localEnd) |
| 1099 | .detail("PeekLocality", peekLocality) |
| 1100 | .detail("CanDiscardPopped", canDiscardPopped); |
| 1101 | |
| 1102 | int maxTxsTags = txsTags; |
| 1103 | bool needsOldTxs = tLogs[0]->tLogVersion < TLogVersion::V4; |
| 1104 | for (auto& it : oldLogData) { |
| 1105 | maxTxsTags = std::max<int>(maxTxsTags, it.txsTags); |
| 1106 | needsOldTxs = needsOldTxs || it.tLogs[0]->tLogVersion < TLogVersion::V4; |
| 1107 | } |
| 1108 | |
| 1109 | if (peekLocality < 0 || localEnd == invalidVersion || localEnd <= begin) { |
| 1110 | std::vector<Reference<ILogSystem::IPeekCursor>> cursors; |
| 1111 | cursors.reserve(maxTxsTags); |
| 1112 | for (int i = 0; i < maxTxsTags; i++) { |
| 1113 | cursors.push_back(peekAll(dbgid, begin, end, Tag(tagLocalityTxs, i), true)); |
| 1114 | } |
| 1115 | // SOMEDAY: remove once upgrades from 6.2 are no longer supported |
| 1116 | if (needsOldTxs) { |
| 1117 | cursors.push_back(peekAll(dbgid, begin, end, txsTag, true)); |
| 1118 | } |
| 1119 | |
| 1120 | return makeReference<ILogSystem::BufferedCursor>(cursors, begin, end, false, false, canDiscardPopped); |
| 1121 | } |
| 1122 | |
| 1123 | try { |
| 1124 | if (localEnd >= end) { |
| 1125 | std::vector<Reference<ILogSystem::IPeekCursor>> cursors; |
| 1126 | cursors.reserve(maxTxsTags); |
| 1127 | for (int i = 0; i < maxTxsTags; i++) { |
| 1128 | cursors.push_back(peekLocal(dbgid, Tag(tagLocalityTxs, i), begin, end, true, peekLocality)); |
| 1129 | } |
| 1130 | // SOMEDAY: remove once upgrades from 6.2 are no longer supported |
| 1131 | if (needsOldTxs) { |
| 1132 | cursors.push_back(peekLocal(dbgid, txsTag, begin, end, true, peekLocality)); |
| 1133 | } |
| 1134 | |
| 1135 | return makeReference<ILogSystem::BufferedCursor>(cursors, begin, end, false, false, canDiscardPopped); |
| 1136 | } |
| 1137 | |
| 1138 | std::vector<Reference<ILogSystem::IPeekCursor>> cursors; |
| 1139 | std::vector<LogMessageVersion> epochEnds; |
| 1140 | |
| 1141 | cursors.resize(2); |