diff --git a/ConflictSet.cpp b/ConflictSet.cpp index 5910681..dbf4a9b 100644 --- a/ConflictSet.cpp +++ b/ConflictSet.cpp @@ -1082,106 +1082,145 @@ struct CheckRangeLeftSide { // Return true if the max version among all keys that start with key[:prefixLen] // that are < key is <= readVersion -bool checkRangeRightSide(Node *n, std::span key, int prefixLen, - int64_t readVersion) { +struct CheckRangeRightSide { + CheckRangeRightSide(Node *n, std::span key, int prefixLen, + int64_t readVersion) + : n(n), key(key), remaining(key), prefixLen(prefixLen), + readVersion(readVersion) { #if DEBUG_VERBOSE && !defined(NDEBUG) - fprintf(stderr, "Check range right side to %s for keys starting with %s\n", - printable(key).c_str(), printable(key.subspan(0, prefixLen)).c_str()); + fprintf(stderr, "Check range right side to %s for keys starting with %s\n", + printable(key).c_str(), + printable(key.subspan(0, prefixLen)).c_str()); #endif - auto remaining = key; + } + + Node *n; + std::span key; + std::span remaining; + int prefixLen; + int64_t readVersion; int searchPathLen = 0; - for (;;) { - if (n->maxVersion <= readVersion) { - return true; - } - if (remaining.size() == 0) { - assert(searchPathLen >= prefixLen); - return true; - } + bool ok; - if (searchPathLen >= prefixLen) { - if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) { - return false; - } - } + enum Phase { Search, DownLeftSpine } phase = Search; - if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) && - n->entryPresent && n->entry.pointVersion > readVersion) { - return false; - } - - int c = getChildGeq(n, remaining[0]); - if (c == remaining[0]) { - n = getChildExists(n, c); - remaining = remaining.subspan(1, remaining.size() - 1); - ++searchPathLen; - } else { - if (c >= 0) { - if (searchPathLen < prefixLen) { - n = getChildExists(n, c); - goto downLeftSpine; - } - if (maxBetweenExclusive(n, -1, c) > readVersion) { - return false; - } + bool step() { + switch (phase) { + case Search: { + if (n->maxVersion <= readVersion) { + ok = true; + return true; + } + if (remaining.size() == 0) { + assert(searchPathLen >= prefixLen); + ok = true; return true; - } else { - n = nextSibling(n); - goto downLeftSpine; } - } - if (n->partialKeyLen > 0) { - int commonLen = std::min(n->partialKeyLen, remaining.size()); - for (int i = 0; i < commonLen; ++i) { - auto c = n->partialKey[i] <=> remaining[i]; - if (c == 0) { - ++searchPathLen; - continue; + if (searchPathLen >= prefixLen) { + if (maxBetweenExclusive(n, -1, remaining[0]) > readVersion) { + ok = false; + return true; } - if (c > 0) { + } + + if (searchPathLen >= prefixLen && searchPathLen < int(key.size()) && + n->entryPresent && n->entry.pointVersion > readVersion) { + ok = false; + return true; + } + + int c = getChildGeq(n, remaining[0]); + if (c == remaining[0]) { + n = getChildExists(n, c); + remaining = remaining.subspan(1, remaining.size() - 1); + ++searchPathLen; + } else { + if (c >= 0) { if (searchPathLen < prefixLen) { - goto downLeftSpine; - } - if (n->entryPresent && n->entry.rangeVersion > readVersion) { + n = getChildExists(n, c); + phase = DownLeftSpine; return false; } + if (maxBetweenExclusive(n, -1, c) > readVersion) { + ok = false; + return true; + } + ok = true; return true; } else { - if (searchPathLen >= prefixLen && n->maxVersion > readVersion) { - return false; - } n = nextSibling(n); - goto downLeftSpine; - } - } - if (commonLen == n->partialKeyLen) { - // partial key matches - remaining = remaining.subspan(commonLen, remaining.size() - commonLen); - } else if (n->partialKeyLen > int(remaining.size())) { - if (searchPathLen < prefixLen) { - goto downLeftSpine; - } - if (n->entryPresent && n->entry.rangeVersion > readVersion) { + phase = DownLeftSpine; return false; } + } + + if (n->partialKeyLen > 0) { + int commonLen = std::min(n->partialKeyLen, remaining.size()); + for (int i = 0; i < commonLen; ++i) { + auto c = n->partialKey[i] <=> remaining[i]; + if (c == 0) { + ++searchPathLen; + continue; + } + if (c > 0) { + if (searchPathLen < prefixLen) { + phase = DownLeftSpine; + return false; + } + if (n->entryPresent && n->entry.rangeVersion > readVersion) { + ok = false; + return true; + } + ok = true; + return true; + } else { + if (searchPathLen >= prefixLen && n->maxVersion > readVersion) { + ok = false; + return true; + } + n = nextSibling(n); + phase = DownLeftSpine; + return false; + } + } + if (commonLen == n->partialKeyLen) { + // partial key matches + remaining = + remaining.subspan(commonLen, remaining.size() - commonLen); + } else if (n->partialKeyLen > int(remaining.size())) { + if (searchPathLen < prefixLen) { + phase = DownLeftSpine; + return false; + } + if (n->entryPresent && n->entry.rangeVersion > readVersion) { + ok = false; + return true; + } + ok = true; + return true; + } + } + } break; + case DownLeftSpine: + if (n == nullptr) { + ok = true; return true; } + for (;;) { + if (n->entryPresent) { + ok = n->entry.rangeVersion <= readVersion; + return true; + } + int c = getChildGeq(n, 0); + assert(c >= 0); + n = getChildExists(n, c); + } + break; } + return false; } -downLeftSpine: - if (n == nullptr) { - return true; - } - for (;;) { - if (n->entryPresent) { - return n->entry.rangeVersion <= readVersion; - } - int c = getChildGeq(n, 0); - assert(c >= 0); - n = getChildExists(n, c); - } -} +}; bool checkRangeRead(Node *n, std::span begin, std::span end, int64_t readVersion) { @@ -1238,7 +1277,11 @@ bool checkRangeRead(Node *n, std::span begin, return false; } - if (!checkRangeRightSide(n, end, lcp + 1, readVersion)) { + CheckRangeRightSide checkRangeRightSide{n, end, lcp + 1, readVersion}; + while (!checkRangeRightSide.step()) + ; + + if (!checkRangeRightSide.ok) { return false; }