6 Commits

Author SHA1 Message Date
fac7968405 Bound memory, and disable free list for now
All checks were successful
Tests / Release [gcc] total: 827, passed: 827
GNU C Compiler (gcc) |Total|New|Outstanding|Fixed|Trend |:-:|:-:|:-:|:-:|:-: |1|0|1|0|:zzz:
Tests / Release [gcc,aarch64] total: 826, passed: 826
Tests / Coverage total: 825, passed: 825
weaselab/conflict-set/pipeline/head This commit looks good
CC #9
2024-03-11 17:11:37 -07:00
e3f6fbe955 Create a Node0 when splitting existing partial key 2024-03-11 16:22:14 -07:00
219af68745 sizeof(Node0) also needs to be < kBytesPerKey
Also remove vestigial comment.

CC #9
2024-03-11 16:15:09 -07:00
52db15d8bd Check min number of children invariant
CC #9
2024-03-11 16:15:07 -07:00
b3b91ef860 Merge Node4 with one child into child if favorable
CC #9
2024-03-11 16:15:05 -07:00
13ee3c3a12 WIP downsize for Node{16,48,256} on erase
CC #9
2024-03-11 16:15:01 -07:00
983 changed files with 2119 additions and 5808 deletions

View File

@@ -1,2 +1,2 @@
CompileFlags:
Add: [-DENABLE_MAIN, -UNDEBUG, -DENABLE_FUZZ, -DTHREAD_TEST, -fexceptions]
Add: [-DENABLE_MAIN, -UNDEBUG, -DENABLE_FUZZ, -fexceptions]

1
.gitignore vendored
View File

@@ -1,3 +1,2 @@
.cache
__pycache__
build

View File

@@ -1,11 +1,11 @@
repos:
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: 6d365699efc33b1b432eab5b4ae331a19e1857de # frozen: v18.1.2
rev: b111689e7b5cba60be3c62d5db2bd1357f4d36ca
hooks:
- id: clang-format
exclude: ".*third_party/.*"
- repo: https://github.com/cheshirekow/cmake-format-precommit
rev: e2c2116d86a80e72e7146a06e68b7c228afc6319 # frozen: v0.6.13
rev: e2c2116d86a80e72e7146a06e68b7c228afc6319
hooks:
- id: cmake-format
- repo: local
@@ -13,7 +13,7 @@ repos:
- id: debug verbose check
name: disallow checking in DEBUG_VERBOSE=1
description: disallow checking in DEBUG_VERBOSE=1
entry: "^#define DEBUG_VERBOSE 1$"
entry: '^#define DEBUG_VERBOSE 1$'
language: pygrep
types: [c++]
- repo: local
@@ -21,14 +21,6 @@ repos:
- id: debug verbose check
name: disallow checking in SHOW_MEMORY=1
description: disallow checking in SHOW_MEMORY=1
entry: "^#define SHOW_MEMORY 1$"
entry: '^#define SHOW_MEMORY 1$'
language: pygrep
types: [c++]
- repo: https://github.com/shellcheck-py/shellcheck-py
rev: a23f6b85d0fdd5bb9d564e2579e678033debbdff # frozen: v0.10.0.1
hooks:
- id: shellcheck
- repo: https://github.com/psf/black
rev: 552baf822992936134cbd31a38f69c8cfe7c0f05 # frozen: 24.3.0
hooks:
- id: black

116
Bench.cpp
View File

@@ -34,7 +34,7 @@ ConflictSet::ReadRange singleton(Arena &arena, std::span<const uint8_t> key) {
std::span<uint8_t>(new (arena) uint8_t[key.size() + 1], key.size() + 1);
memcpy(r.data(), key.data(), key.size());
r[key.size()] = 0;
return {{key.data(), int(key.size())}, {r.data(), int(r.size())}, 0};
return {key.data(), int(key.size()), r.data(), int(r.size())};
}
ConflictSet::ReadRange prefixRange(Arena &arena, std::span<const uint8_t> key) {
@@ -52,7 +52,7 @@ ConflictSet::ReadRange prefixRange(Arena &arena, std::span<const uint8_t> key) {
auto r = std::span<uint8_t>(new (arena) uint8_t[index + 1], index + 1);
memcpy(r.data(), key.data(), index + 1);
r[r.size() - 1]++;
return {{key.data(), int(key.size())}, {r.data(), int(r.size())}, 0};
return {key.data(), int(key.size()), r.data(), int(r.size())};
}
void benchConflictSet() {
@@ -258,114 +258,4 @@ void benchConflictSet() {
}
}
constexpr int kKeyLenForWorstCase = 50;
ConflictSet worstCaseConflictSetForRadixRangeRead(int cardinality) {
ConflictSet cs{0};
for (int i = 0; i < kKeyLenForWorstCase; ++i) {
for (int j = 0; j < cardinality; ++j) {
auto b = std::vector<uint8_t>(i, 0);
b.push_back(j);
auto e = std::vector<uint8_t>(i, 255);
e.push_back(255 - j);
weaselab::ConflictSet::WriteRange w[] = {{
{b.data(), int(b.size())},
{nullptr, 0},
},
{
{e.data(), int(e.size())},
{nullptr, 0},
}};
std::sort(std::begin(w), std::end(w),
[](const auto &lhs, const auto &rhs) {
int cl = std::min(lhs.begin.len, rhs.begin.len);
if (cl > 0) {
int c = memcmp(lhs.begin.p, rhs.begin.p, cl);
if (c != 0) {
return c < 0;
}
}
return lhs.begin.len < rhs.begin.len;
});
cs.addWrites(w, sizeof(w) / sizeof(w[0]), 0);
}
}
// Defeat short-circuiting on the left
{
auto k = std::vector<uint8_t>(kKeyLenForWorstCase, 0);
weaselab::ConflictSet::WriteRange w[] = {
{
{k.data(), int(k.size())},
{nullptr, 0},
},
};
cs.addWrites(w, sizeof(w) / sizeof(w[0]), 1);
}
// Defeat short-circuiting on the right
{
auto k = std::vector<uint8_t>(kKeyLenForWorstCase, 255);
weaselab::ConflictSet::WriteRange w[] = {
{
{k.data(), int(k.size())},
{nullptr, 0},
},
};
cs.addWrites(w, sizeof(w) / sizeof(w[0]), 1);
}
return cs;
}
void benchWorstCaseForRadixRangeRead() {
ankerl::nanobench::Bench bench;
std::unique_ptr<ConflictSet> cs[256];
for (int i = 0; i < 256; ++i) {
cs[i] =
std::make_unique<ConflictSet>(worstCaseConflictSetForRadixRangeRead(i));
}
auto begin = std::vector<uint8_t>(kKeyLenForWorstCase - 1, 0);
begin.push_back(1);
auto end = std::vector<uint8_t>(kKeyLenForWorstCase - 1, 255);
end.push_back(254);
weaselab::ConflictSet::Result result;
weaselab::ConflictSet::ReadRange r{
{begin.data(), int(begin.size())}, {end.data(), int(end.size())}, 0};
bench.run("worst case for radix tree", [&]() {
for (int i = 0; i < 256; ++i) {
result = weaselab::ConflictSet::TooOld;
cs[i]->check(&r, &result, 1);
if (result != weaselab::ConflictSet::Commit) {
abort();
}
}
});
// for (int i = 0; i < 256; ++i) {
// bench.run("worst case for radix tree, span " + std::to_string(i), [&]() {
// result = weaselab::ConflictSet::TooOld;
// cs[i]->check(&r, &result, 1);
// if (result != weaselab::ConflictSet::Commit) {
// abort();
// }
// });
// }
}
void benchCreateAndDestroy() {
ankerl::nanobench::Bench bench;
bench.run("create and destroy", [&]() { ConflictSet cs{0}; });
}
int main(void) {
benchConflictSet();
benchWorstCaseForRadixRangeRead();
benchCreateAndDestroy();
}
int main(void) { benchConflictSet(); }

View File

@@ -1,22 +1,13 @@
cmake_minimum_required(VERSION 3.18)
project(
conflict-set
VERSION 0.0.12
conflict_set
VERSION 0.0.1
DESCRIPTION
"A data structure for optimistic concurrency control on ranges of bitwise-lexicographically-ordered keys."
HOMEPAGE_URL "https://git.weaselab.dev/weaselab/conflict-set"
LANGUAGES C CXX)
set(CMAKE_CXX_STANDARD 20)
file(WRITE ${CMAKE_CURRENT_BINARY_DIR}/version.txt ${PROJECT_VERSION})
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/version.txt.in
${CMAKE_CURRENT_SOURCE_DIR}/paper/version.txt)
include(CMakePushCheckState)
include(CheckCXXCompilerFlag)
include(CheckIncludeFileCXX)
include(CheckCXXSourceCompiles)
set(DEFAULT_BUILD_TYPE "Release")
if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
@@ -31,40 +22,16 @@ if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
"MinSizeRel" "RelWithDebInfo")
endif()
add_compile_options(-fdata-sections -ffunction-sections -Wswitch-enum
-Werror=switch-enum -fPIC)
if(NOT APPLE)
# This causes some versions of clang to crash on macos
add_compile_options(-g -fno-omit-frame-pointer)
endif()
set(full_relro_flags "-pie;LINKER:-z,relro,-z,now,-z,noexecstack")
cmake_push_check_state()
list(APPEND CMAKE_REQUIRED_LINK_OPTIONS ${full_relro_flags})
check_cxx_source_compiles("int main(){}" HAS_FULL_RELRO FAIL_REGEX "warning:")
if(HAS_FULL_RELRO)
add_link_options(${full_relro_flags})
endif()
cmake_pop_check_state()
set(version_script_flags
LINKER:--version-script=${CMAKE_CURRENT_SOURCE_DIR}/linker.map)
cmake_push_check_state()
list(APPEND CMAKE_REQUIRED_LINK_OPTIONS ${version_script_flags})
check_cxx_source_compiles("int main(){}" HAS_VERSION_SCRIPT FAIL_REGEX
"warning:")
cmake_pop_check_state()
option(USE_SIMD_FALLBACK
"Use fallback implementations of functions that use SIMD" OFF)
option(DISABLE_TSAN "Disable TSAN" OFF)
add_compile_options(-fdata-sections -ffunction-sections)
# This is encouraged according to
# https://valgrind.org/docs/manual/manual-core-adv.html#manual-core-adv.clientreq
include_directories(SYSTEM ${CMAKE_CURRENT_SOURCE_DIR}/third_party/valgrind)
include_directories(SYSTEM ${CMAKE_SOURCE_DIR}/third_party/valgrind)
add_compile_options($<$<COMPILE_LANGUAGE:CXX>:-Wno-invalid-offsetof>)
if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
add_compile_options(-Wno-maybe-uninitialized
$<$<COMPILE_LANGUAGE:CXX>:-Wno-invalid-offsetof>)
endif()
if(APPLE)
add_link_options(-Wl,-dead_strip)
@@ -72,93 +39,72 @@ else()
add_link_options(-Wl,--gc-sections)
endif()
if(EMSCRIPTEN)
# https://github.com/emscripten-core/emscripten/issues/15377#issuecomment-1285167486
add_link_options(-lnodefs.js -lnoderawfs.js)
add_link_options(-s ALLOW_MEMORY_GROWTH)
endif()
include(CheckIncludeFileCXX)
include(CMakePushCheckState)
if(NOT USE_SIMD_FALLBACK)
cmake_push_check_state()
list(APPEND CMAKE_REQUIRED_FLAGS -mavx)
check_include_file_cxx("immintrin.h" HAS_AVX)
if(HAS_AVX)
cmake_push_check_state()
list(APPEND CMAKE_REQUIRED_FLAGS -mavx)
check_include_file_cxx("immintrin.h" HAS_AVX)
if(HAS_AVX)
add_compile_options(-mavx)
add_compile_definitions(HAS_AVX)
endif()
cmake_pop_check_state()
endif()
cmake_pop_check_state()
check_include_file_cxx("arm_neon.h" HAS_ARM_NEON)
if(HAS_ARM_NEON)
check_include_file_cxx("arm_neon.h" HAS_ARM_NEON)
if(HAS_ARM_NEON)
add_compile_definitions(HAS_ARM_NEON)
endif()
endif()
set(CMAKE_CXX_IMPLICIT_LINK_LIBRARIES "")
add_library(${PROJECT_NAME}-object OBJECT ConflictSet.cpp)
target_compile_options(${PROJECT_NAME}-object PRIVATE -fno-exceptions
add_library(${PROJECT_NAME}_object OBJECT ConflictSet.cpp)
target_compile_options(${PROJECT_NAME}_object PRIVATE -fPIC -fno-exceptions
-fvisibility=hidden)
target_include_directories(${PROJECT_NAME}-object
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
target_include_directories(${PROJECT_NAME}_object
PRIVATE ${CMAKE_SOURCE_DIR}/include)
add_library(${PROJECT_NAME} SHARED $<TARGET_OBJECTS:${PROJECT_NAME}-object>)
add_library(${PROJECT_NAME} SHARED $<TARGET_OBJECTS:${PROJECT_NAME}_object>)
set_target_properties(
${PROJECT_NAME} PROPERTIES LIBRARY_OUTPUT_DIRECTORY
"${CMAKE_CURRENT_BINARY_DIR}/radix_tree")
"${CMAKE_BINARY_DIR}/radix_tree")
if(NOT CMAKE_BUILD_TYPE STREQUAL Debug)
set_target_properties(${PROJECT_NAME} PROPERTIES LINKER_LANGUAGE C)
endif()
if(HAS_VERSION_SCRIPT)
target_link_options(
${PROJECT_NAME} PRIVATE
LINKER:--version-script=${CMAKE_CURRENT_SOURCE_DIR}/linker.map)
if(NOT APPLE)
target_link_options(${PROJECT_NAME} PRIVATE
LINKER:--version-script=${CMAKE_SOURCE_DIR}/linker.map)
endif()
add_library(${PROJECT_NAME}-static STATIC
$<TARGET_OBJECTS:${PROJECT_NAME}-object>)
add_library(${PROJECT_NAME}_static STATIC
$<TARGET_OBJECTS:${PROJECT_NAME}_object>)
if(NOT CMAKE_BUILD_TYPE STREQUAL Debug)
set_target_properties(${PROJECT_NAME}-static PROPERTIES LINKER_LANGUAGE C)
set_target_properties(${PROJECT_NAME}_static PROPERTIES LINKER_LANGUAGE C)
endif()
if(APPLE)
if(NOT APPLE AND CMAKE_OBJCOPY)
add_custom_command(
TARGET ${PROJECT_NAME}-static
PRE_LINK
COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/privatize_symbols_macos.sh
$<TARGET_OBJECTS:${PROJECT_NAME}-object>)
else()
add_custom_command(
TARGET ${PROJECT_NAME}-static
TARGET conflict_set_static
POST_BUILD
COMMAND
${CMAKE_OBJCOPY}
--keep-global-symbols=${CMAKE_CURRENT_SOURCE_DIR}/symbol-exports.txt
$<TARGET_FILE:${PROJECT_NAME}-static> || echo
"Proceeding with all symbols global in static library")
${CMAKE_OBJCOPY} --keep-global-symbols=${CMAKE_SOURCE_DIR}/symbols.txt
$<TARGET_FILE:${PROJECT_NAME}_static>)
endif()
set(TEST_FLAGS -Wall -Wextra -Wunreachable-code -Wpedantic -UNDEBUG)
include(CTest)
# disable tests if this is being used through e.g. FetchContent
if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR AND BUILD_TESTING)
set(TEST_FLAGS -Wall -Wextra -Wunreachable-code -Wpedantic -UNDEBUG)
# corpus tests, which are tests curated by libfuzzer. The goal is to get broad
# coverage with a small number of tests.
file(GLOB CORPUS_TESTS ${CMAKE_CURRENT_SOURCE_DIR}/corpus/*)
if(BUILD_TESTING)
# Shared library version of FoundationDB's skip list implementation
add_library(skip_list SHARED SkipList.cpp)
target_compile_options(skip_list PRIVATE -fno-exceptions -fvisibility=hidden)
target_include_directories(skip_list
PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
set_target_properties(
skip_list PROPERTIES LIBRARY_OUTPUT_DIRECTORY
"${CMAKE_CURRENT_BINARY_DIR}/skip_list")
target_compile_options(skip_list PRIVATE -fPIC -fno-exceptions
-fvisibility=hidden)
target_include_directories(skip_list PUBLIC ${CMAKE_SOURCE_DIR}/include)
set_target_properties(skip_list PROPERTIES LIBRARY_OUTPUT_DIRECTORY
"${CMAKE_BINARY_DIR}/skip_list")
set_target_properties(skip_list PROPERTIES OUTPUT_NAME ${PROJECT_NAME})
set_target_properties(skip_list PROPERTIES VERSION ${PROJECT_VERSION}
SOVERSION ${PROJECT_VERSION_MAJOR})
@@ -166,27 +112,16 @@ if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR AND BUILD_TESTING)
# Shared library version of a std::unordered_map-based conflict set (point
# queries only)
add_library(hash_table SHARED HashTable.cpp)
target_compile_options(hash_table PRIVATE -fno-exceptions -fvisibility=hidden)
target_include_directories(hash_table
PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
set_target_properties(
hash_table PROPERTIES LIBRARY_OUTPUT_DIRECTORY
"${CMAKE_CURRENT_BINARY_DIR}/hash_table")
target_compile_options(hash_table PRIVATE -fPIC -fno-exceptions
-fvisibility=hidden)
target_include_directories(hash_table PUBLIC ${CMAKE_SOURCE_DIR}/include)
set_target_properties(hash_table PROPERTIES LIBRARY_OUTPUT_DIRECTORY
"${CMAKE_BINARY_DIR}/hash_table")
set_target_properties(hash_table PROPERTIES OUTPUT_NAME ${PROJECT_NAME})
set_target_properties(
hash_table PROPERTIES VERSION ${PROJECT_VERSION} SOVERSION
${PROJECT_VERSION_MAJOR})
add_executable(driver_skip_list TestDriver.cpp)
target_compile_options(driver_skip_list PRIVATE ${TEST_FLAGS})
target_link_libraries(driver_skip_list PRIVATE skip_list)
foreach(TEST ${CORPUS_TESTS})
get_filename_component(hash ${TEST} NAME)
add_test(NAME skip_list_${hash} COMMAND driver_skip_list ${TEST})
endforeach()
# ad hoc testing
add_executable(conflict_set_main ConflictSet.cpp)
target_include_directories(conflict_set_main
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
@@ -213,13 +148,12 @@ if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR AND BUILD_TESTING)
endif()
endif()
# whitebox tests
# corpus tests
file(GLOB CORPUS_TESTS ${CMAKE_SOURCE_DIR}/corpus/*)
add_executable(fuzz_driver ConflictSet.cpp FuzzTestDriver.cpp)
target_compile_options(fuzz_driver PRIVATE ${TEST_FLAGS})
if(NOT CMAKE_CROSSCOMPILING)
target_compile_options(fuzz_driver PRIVATE -fsanitize=address,undefined)
target_link_options(fuzz_driver PRIVATE -fsanitize=address,undefined)
endif()
target_compile_definitions(fuzz_driver PRIVATE ENABLE_FUZZ)
target_include_directories(fuzz_driver
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
@@ -228,65 +162,27 @@ if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR AND BUILD_TESTING)
add_test(NAME conflict_set_fuzz_${hash} COMMAND fuzz_driver ${TEST})
endforeach()
# tsan tests
if(NOT CMAKE_CROSSCOMPILING AND NOT DISABLE_TSAN)
add_executable(tsan_driver ConflictSet.cpp FuzzTestDriver.cpp)
target_compile_options(tsan_driver PRIVATE ${TEST_FLAGS} -fsanitize=thread)
target_link_options(tsan_driver PRIVATE -fsanitize=thread)
target_compile_definitions(tsan_driver PRIVATE ENABLE_FUZZ THREAD_TEST)
target_include_directories(tsan_driver
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
foreach(TEST ${CORPUS_TESTS})
get_filename_component(hash ${TEST} NAME)
add_test(NAME conflict_set_tsan_${hash} COMMAND tsan_driver ${TEST})
endforeach()
endif()
# blackbox tests
add_executable(driver TestDriver.cpp)
target_compile_options(driver PRIVATE ${TEST_FLAGS})
target_link_libraries(driver PRIVATE ${PROJECT_NAME})
foreach(TEST ${CORPUS_TESTS})
get_filename_component(hash ${TEST} NAME)
add_test(NAME conflict_set_blackbox_${hash} COMMAND driver ${TEST})
endforeach()
# scripted tests. Written manually to fill in anything libfuzzer couldn't
# find.
if(NOT CMAKE_CROSSCOMPILING)
find_package(Python3 REQUIRED COMPONENTS Interpreter)
set_property(
DIRECTORY
APPEND
PROPERTY CMAKE_CONFIGURE_DEPENDS
${CMAKE_CURRENT_SOURCE_DIR}/test_conflict_set.py)
execute_process(
COMMAND ${Python3_EXECUTABLE}
${CMAKE_CURRENT_SOURCE_DIR}/test_conflict_set.py list
OUTPUT_VARIABLE SCRIPT_TESTS)
foreach(TEST ${SCRIPT_TESTS})
add_test(
NAME script_test_${TEST}
COMMAND
${Python3_EXECUTABLE}
${CMAKE_CURRENT_SOURCE_DIR}/test_conflict_set.py test ${TEST}
--build-dir ${CMAKE_CURRENT_BINARY_DIR})
endforeach()
endif()
add_executable(driver_skip_list TestDriver.cpp)
target_compile_options(driver_skip_list PRIVATE ${TEST_FLAGS})
target_link_libraries(driver_skip_list PRIVATE skip_list)
find_program(VALGRIND_EXE valgrind)
if(VALGRIND_EXE AND NOT CMAKE_CROSSCOMPILING)
list(LENGTH CORPUS_TESTS len)
math(EXPR last "${len} - 1")
set(partition_size 100)
foreach(i RANGE 0 ${last} ${partition_size})
list(SUBLIST CORPUS_TESTS ${i} ${partition_size} partition)
add_test(NAME conflict_set_blackbox_valgrind_${i}
add_test(NAME conflict_set_blackbox_valgrind
COMMAND ${VALGRIND_EXE} --error-exitcode=99 --
$<TARGET_FILE:driver> ${partition})
endforeach()
$<TARGET_FILE:driver> ${CORPUS_TESTS})
endif()
foreach(TEST ${CORPUS_TESTS})
get_filename_component(hash ${TEST} NAME)
add_test(NAME conflict_set_blackbox_${hash} COMMAND driver ${TEST})
add_test(NAME skip_list_${hash} COMMAND driver_skip_list ${TEST})
endforeach()
# api smoke tests
# c90
@@ -307,56 +203,31 @@ if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR AND BUILD_TESTING)
PROPERTIES CXX_STANDARD_REQUIRED ON)
add_test(NAME conflict_set_cxx_api_test COMMAND conflict_set_cxx_api_test)
# symbol visibility tests
if(NOT CMAKE_BUILD_TYPE STREQUAL Debug)
if(APPLE)
set(symbol_exports ${CMAKE_CURRENT_SOURCE_DIR}/apple-symbol-exports.txt)
set(symbol_imports ${CMAKE_CURRENT_SOURCE_DIR}/apple-symbol-imports.txt)
else()
set(symbol_exports ${CMAKE_CURRENT_SOURCE_DIR}/symbol-exports.txt)
if(CMAKE_SYSTEM_PROCESSOR STREQUAL aarch64)
set(symbol_imports
${CMAKE_CURRENT_SOURCE_DIR}/aarch64-symbol-imports.txt)
else()
set(symbol_imports ${CMAKE_CURRENT_SOURCE_DIR}/symbol-imports.txt)
endif()
endif()
if(NOT APPLE AND NOT CMAKE_BUILD_TYPE STREQUAL Debug)
add_test(
NAME conflict_set_shared_symbols
COMMAND
${CMAKE_CURRENT_SOURCE_DIR}/test_symbols.sh
$<TARGET_FILE:${PROJECT_NAME}> ${symbol_exports} ${symbol_imports})
COMMAND ${CMAKE_SOURCE_DIR}/test_symbols.sh
$<TARGET_FILE:${PROJECT_NAME}> ${CMAKE_SOURCE_DIR}/symbols.txt)
add_test(
NAME conflict_set_static_symbols
COMMAND
${CMAKE_CURRENT_SOURCE_DIR}/test_symbols.sh
$<TARGET_FILE:${PROJECT_NAME}-static> ${symbol_exports}
${symbol_imports})
${CMAKE_SOURCE_DIR}/test_symbols.sh
$<TARGET_FILE:${PROJECT_NAME}_static> ${CMAKE_SOURCE_DIR}/symbols.txt)
endif()
# bench
add_executable(conflict_set_bench Bench.cpp)
target_link_libraries(conflict_set_bench PRIVATE ${PROJECT_NAME})
set_target_properties(conflict_set_bench PROPERTIES SKIP_BUILD_RPATH ON)
add_executable(real_data_bench RealDataBench.cpp)
target_link_libraries(real_data_bench PRIVATE ${PROJECT_NAME})
set_target_properties(real_data_bench PROPERTIES SKIP_BUILD_RPATH ON)
# fuzzer-based perf
add_executable(driver_perf TestDriver.cpp)
target_compile_definitions(driver_perf PRIVATE PERF_TEST=1)
target_link_libraries(driver_perf PRIVATE ${PROJECT_NAME})
# server bench
add_executable(server_bench ServerBench.cpp)
target_link_libraries(server_bench PRIVATE ${PROJECT_NAME})
set_target_properties(server_bench PROPERTIES SKIP_BUILD_RPATH ON)
endif()
# packaging
set(CPACK_PACKAGE_CONTACT andrew@weaselab.dev)
set(CMAKE_INSTALL_DEFAULT_COMPONENT_NAME all)
set(CPACK_PACKAGE_VENDOR "Weaselab")
set(CPACK_RESOURCE_FILE_LICENSE "${CMAKE_CURRENT_SOURCE_DIR}/LICENSE")
@@ -366,30 +237,6 @@ set(CPACK_RESOURCE_FILE_README "${CMAKE_CURRENT_SOURCE_DIR}/README.md")
set(CPACK_RPM_PACKAGE_ARCHITECTURE ${CMAKE_SYSTEM_PROCESSOR})
set(CPACK_RPM_SPEC_INSTALL_POST "/bin/true") # avoid stripping
set(CPACK_RPM_PACKAGE_LICENSE "Apache 2.0")
set(CPACK_RPM_FILE_NAME RPM-DEFAULT)
# deb
set(CPACK_DEBIAN_FILE_NAME DEB-DEFAULT)
# see *-imports.txt - dependency versions need to be synced with symbol versions
if(CMAKE_SYSTEM_PROCESSOR STREQUAL aarch64)
set(CPACK_DEBIAN_PACKAGE_DEPENDS "libc6 (>= 2.17)")
else()
set(CPACK_DEBIAN_PACKAGE_DEPENDS "libc6 (>= 2.14)")
endif()
# macos
set(CMAKE_OSX_DEPLOYMENT_TARGET 11.0)
if(APPLE)
find_program(PANDOC_EXE pandoc)
if(PANDOC_EXE)
execute_process(COMMAND ${PANDOC_EXE} ${CMAKE_CURRENT_SOURCE_DIR}/README.md
-o ${CMAKE_CURRENT_BINARY_DIR}/README.txt)
set(CPACK_RESOURCE_FILE_README ${CMAKE_CURRENT_BINARY_DIR}/README.txt)
endif()
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/LICENSE
${CMAKE_CURRENT_BINARY_DIR}/LICENSE.txt COPYONLY)
set(CPACK_RESOURCE_FILE_LICENSE ${CMAKE_CURRENT_BINARY_DIR}/LICENSE.txt)
endif()
include(CPack)
@@ -401,7 +248,7 @@ target_include_directories(
$<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME}>)
target_include_directories(
${PROJECT_NAME}-static
${PROJECT_NAME}_static
PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
$<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME}>)
@@ -410,16 +257,13 @@ set_target_properties(
SOVERSION ${PROJECT_VERSION_MAJOR})
install(
TARGETS ${PROJECT_NAME} ${PROJECT_NAME}-static
EXPORT ${PROJECT_NAME}Config
TARGETS ${PROJECT_NAME} ${PROJECT_NAME}_static
EXPORT ConflictSetConfig
ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR})
install(DIRECTORY include/
DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME})
install(EXPORT ${PROJECT_NAME}Config
DESTINATION ${CMAKE_INSTALL_DATAROOTDIR}/${PROJECT_NAME}/cmake)
cpack_add_component(all)
install(EXPORT ConflictSetConfig
DESTINATION ${CMAKE_INSTALL_DATAROOTDIR}/ConflictSet/cmake)

File diff suppressed because it is too large Load Diff

View File

@@ -2,12 +2,10 @@
#include <cstdint>
#include <fstream>
#include <sstream>
#include <thread>
extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size);
int main(int argc, char **argv) {
auto doTest = [&]() {
for (int i = 1; i < argc; ++i) {
std::ifstream t(argv[i], std::ios::binary);
std::stringstream buffer;
@@ -15,12 +13,4 @@ int main(int argc, char **argv) {
auto str = buffer.str();
LLVMFuzzerTestOneInput((const uint8_t *)str.data(), str.size());
}
};
#ifdef THREAD_TEST
std::thread thread2{doTest};
#endif
doTest();
#ifdef THREAD_TEST
thread2.join();
#endif
}

View File

@@ -96,22 +96,13 @@ void ConflictSet::setOldestVersion(int64_t oldestVersion) {
return impl->setOldestVersion(oldestVersion);
}
int64_t ConflictSet::getBytes() const { return -1; }
void ConflictSet::getMetricsV1(MetricsV1 **metrics, int *count) const {
*metrics = nullptr;
*count = 0;
}
double ConflictSet::MetricsV1::getValue() const { return 0; }
ConflictSet::ConflictSet(int64_t oldestVersion)
: impl(new(safe_malloc(sizeof(Impl))) Impl{oldestVersion}) {}
: impl(new (safe_malloc(sizeof(Impl))) Impl{oldestVersion}) {}
ConflictSet::~ConflictSet() {
if (impl) {
impl->~Impl();
safe_free(impl, sizeof(Impl));
free(impl);
}
}
@@ -151,11 +142,6 @@ ConflictSet_create(int64_t oldestVersion) {
__attribute__((__visibility__("default"))) void ConflictSet_destroy(void *cs) {
using Impl = ConflictSet::Impl;
((Impl *)cs)->~Impl();
safe_free(cs, sizeof(Impl));
}
__attribute__((__visibility__("default"))) int64_t
ConflictSet_getBytes(void *cs) {
using Impl = ConflictSet::Impl;
return -1;
free(cs);
}
}

View File

@@ -2,8 +2,6 @@
#include "ConflictSet.h"
using namespace weaselab;
#include <bit>
#include <cassert>
#include <compare>
@@ -12,12 +10,10 @@ using namespace weaselab;
#include <cstdlib>
#include <cstring>
#include <inttypes.h>
#include <latch>
#include <map>
#include <set>
#include <span>
#include <string>
#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>
@@ -39,74 +35,16 @@ operator<=>(const std::span<const uint8_t> &lhs,
return lhs.size() <=> rhs.size();
}
[[nodiscard]] inline auto operator<=>(const std::span<const uint8_t> &lhs,
const ConflictSet::Key &rhs) noexcept {
int cl = std::min<int>(lhs.size(), rhs.len);
if (cl > 0) {
if (auto c = memcmp(lhs.data(), rhs.p, cl) <=> 0; c != 0) {
return c;
}
}
return lhs.size() <=> size_t(rhs.len);
}
// This header contains code that we want to reuse outside of ConflictSet.cpp or
// want to exclude from coverage since it's only testing related.
// GCOVR_EXCL_START
#if SHOW_MEMORY
inline int64_t mallocBytes = 0;
inline int64_t peakMallocBytes = 0;
#endif
inline thread_local int64_t mallocBytesDelta = 0;
#ifndef NDEBUG
constexpr auto kMallocHeaderSize = 16;
#endif
// malloc that aborts on OOM and thus always returns a non-null pointer. Must be
// paired with `safe_free`.
__attribute__((always_inline)) inline void *safe_malloc(size_t s) {
mallocBytesDelta += s;
#if SHOW_MEMORY
mallocBytes += s;
if (mallocBytes > peakMallocBytes) {
peakMallocBytes = mallocBytes;
}
#endif
void *p = malloc(s
#ifndef NDEBUG
+ kMallocHeaderSize
#endif
);
if (p == nullptr) {
abort();
}
#ifndef NDEBUG
memcpy(p, &s, sizeof(s));
(char *&)p += kMallocHeaderSize;
#endif
if (void *p = malloc(s)) {
return p;
}
// Must be paired with `safe_malloc`.
//
// There's nothing safer about this than free. Only called safe_free for
// symmetry with safe_malloc.
__attribute__((always_inline)) inline void safe_free(void *p, size_t s) {
mallocBytesDelta -= s;
#if SHOW_MEMORY
mallocBytes -= s;
#endif
#ifndef NDEBUG
(char *&)p -= kMallocHeaderSize;
size_t expected;
memcpy(&expected, p, sizeof(expected));
assert(s == expected);
#endif
free(p);
}
abort();
}
// ==================== BEGIN ARENA IMPL ====================
@@ -152,7 +90,6 @@ inline void *operator new[](size_t size, std::align_val_t align,
/// align must be a power of two
template <class T> T *align_up(T *t, size_t align) {
assert(std::popcount(align) == 1);
auto unaligned = uintptr_t(t);
auto aligned = (unaligned + align - 1) & ~(align - 1);
return reinterpret_cast<T *>(reinterpret_cast<char *>(t) + aligned -
@@ -161,7 +98,6 @@ template <class T> T *align_up(T *t, size_t align) {
/// align must be a power of two
constexpr inline int align_up(uint32_t unaligned, uint32_t align) {
assert(std::popcount(align) == 1);
return (unaligned + align - 1) & ~(align - 1);
}
@@ -177,10 +113,12 @@ struct Arena::ArenaImpl {
uint8_t *begin() { return reinterpret_cast<uint8_t *>(this + 1); }
};
static_assert(sizeof(Arena::ArenaImpl) == 16);
static_assert(alignof(Arena::ArenaImpl) == 8);
inline Arena::Arena(int initialSize) : impl(nullptr) {
if (initialSize > 0) {
auto allocationSize =
align_up(initialSize + sizeof(ArenaImpl), alignof(ArenaImpl));
auto allocationSize = align_up(initialSize + sizeof(ArenaImpl), 16);
impl = (Arena::ArenaImpl *)safe_malloc(allocationSize);
impl->prev = nullptr;
impl->capacity = allocationSize - sizeof(ArenaImpl);
@@ -191,7 +129,7 @@ inline Arena::Arena(int initialSize) : impl(nullptr) {
inline void onDestroy(Arena::ArenaImpl *impl) {
while (impl) {
auto *prev = impl->prev;
safe_free(impl, sizeof(Arena::ArenaImpl) + impl->capacity);
free(impl);
impl = prev;
}
}
@@ -216,7 +154,7 @@ inline void *operator new(size_t size, std::align_val_t align, Arena &arena) {
(arena.impl ? std::max<int>(sizeof(Arena::ArenaImpl),
arena.impl->capacity * 2)
: 0)),
alignof(Arena::ArenaImpl));
16);
auto *impl = (Arena::ArenaImpl *)safe_malloc(allocationSize);
impl->prev = arena.impl;
impl->capacity = allocationSize - sizeof(Arena::ArenaImpl);
@@ -255,75 +193,10 @@ template <class T> struct ArenaAlloc {
void deallocate(T *, size_t) noexcept {}
};
template <class T> struct Vector {
static_assert(std::is_trivially_destructible_v<T>);
static_assert(std::is_trivially_copyable_v<T>);
explicit Vector(Arena *arena)
: arena(arena), t(nullptr), size_(0), capacity(0) {}
void append(std::span<const T> slice) {
if (size_ + int(slice.size()) > capacity) {
grow(std::max<int>(size_ + slice.size(), capacity * 2));
}
if (slice.size() > 0) {
memcpy(const_cast<std::remove_const_t<T> *>(t) + size_, slice.data(),
slice.size() * sizeof(T));
}
size_ += slice.size();
}
// Caller must write to the returned slice
std::span<T> unsafePrepareAppend(int appendSize) {
if (size_ + appendSize > capacity) {
grow(std::max<int>(size_ + appendSize, capacity * 2));
}
auto result = std::span<T>(t + size_, appendSize);
size_ += appendSize;
return result;
}
void push_back(const T &t) { append(std::span<const T>(&t, 1)); }
T *begin() { return t; }
T *end() { return t + size_; }
T *data() { return t; }
T &back() {
assert(size_ > 0);
return t[size_ - 1];
}
T &operator[](int i) {
assert(i >= 0 && i < size_);
return t[i];
}
void pop_back() {
assert(size_ > 0);
--size_;
}
int size() const { return size_; }
operator std::span<const T>() const { return std::span(t, size_); }
private:
void grow(int newCapacity) {
capacity = newCapacity;
auto old = std::span<const T>(*this);
t = (T *)new (std::align_val_t(alignof(T)), *arena)
uint8_t[capacity * sizeof(T)];
size_ = 0;
append(old);
}
Arena *arena;
T *t;
int size_;
int capacity;
};
template <class T> auto vector(Arena &arena) { return Vector<T>(&arena); }
template <class T> using Vector = std::vector<T, ArenaAlloc<T>>;
template <class T> auto vector(Arena &arena) {
return Vector<T>(ArenaAlloc<T>(&arena));
}
template <class T, class C> using Set = std::set<T, C, ArenaAlloc<T>>;
template <class T, class C = std::less<T>> auto set(Arena &arena) {
return Set<T, C>(ArenaAlloc<T>(&arena));
@@ -476,16 +349,42 @@ inline uint32_t Arbitrary::bounded(uint32_t s) {
// ==================== END ARBITRARY IMPL ====================
// ==================== BEGIN UTILITIES IMPL ====================
// Call Stepwise::step for each element of remaining until it returns true.
// Applies a permutation to `remaining` as a side effect.
template <class Stepwise> void runInterleaved(std::span<Stepwise> remaining) {
while (remaining.size() > 0) {
for (int i = 0; i < int(remaining.size());) {
bool done = remaining[i].step();
if (done) {
if (i != int(remaining.size()) - 1) {
using std::swap;
swap(remaining[i], remaining.back());
}
remaining = remaining.subspan(0, remaining.size() - 1);
} else {
++i;
}
}
}
};
template <class Stepwise> void runSequential(std::span<Stepwise> remaining) {
for (auto &r : remaining) {
while (!r.step()) {
}
}
}
struct ReferenceImpl {
explicit ReferenceImpl(int64_t oldestVersion)
: oldestVersion(oldestVersion), newestVersion(oldestVersion) {
explicit ReferenceImpl(int64_t oldestVersion) : oldestVersion(oldestVersion) {
writeVersionMap[""] = oldestVersion;
}
void check(const ConflictSet::ReadRange *reads, ConflictSet::Result *results,
int count) const {
for (int i = 0; i < count; ++i) {
if (reads[i].readVersion < oldestVersion ||
reads[i].readVersion < newestVersion - 2e9) {
if (reads[i].readVersion < oldestVersion) {
results[i] = ConflictSet::TooOld;
continue;
}
@@ -507,8 +406,6 @@ struct ReferenceImpl {
}
void addWrites(const ConflictSet::WriteRange *writes, int count,
int64_t writeVersion) {
assert(writeVersion >= newestVersion);
newestVersion = writeVersion;
for (int i = 0; i < count; ++i) {
auto begin =
std::string((const char *)writes[i].begin.p, writes[i].begin.len);
@@ -528,12 +425,11 @@ struct ReferenceImpl {
}
void setOldestVersion(int64_t oldestVersion) {
assert(oldestVersion >= this->oldestVersion);
assert(oldestVersion >= oldestVersion);
this->oldestVersion = oldestVersion;
}
int64_t oldestVersion;
int64_t newestVersion;
std::map<std::string, int64_t> writeVersionMap;
};
@@ -574,7 +470,23 @@ inline std::string printable(std::span<const uint8_t> key) {
return printable(std::string_view((const char *)key.data(), key.size()));
}
inline const char *resultToStr(ConflictSet::Result r) {
namespace {
template <class ConflictSetImpl> struct TestDriver {
Arbitrary arbitrary;
explicit TestDriver(const uint8_t *data, size_t size)
: arbitrary({data, size}) {}
int64_t writeVersion = 0;
int64_t oldestVersion = 0;
ConflictSetImpl cs{oldestVersion};
ReferenceImpl refImpl{oldestVersion};
constexpr static auto kMaxKeyLen = 64;
bool ok = true;
static const char *resultToStr(ConflictSet::Result r) {
switch (r) {
case ConflictSet::Commit:
return "commit";
@@ -584,56 +496,29 @@ inline const char *resultToStr(ConflictSet::Result r) {
return "too old";
}
abort();
}
namespace {
template <class ConflictSetImpl, bool kEnableAssertions = true>
struct TestDriver {
Arbitrary *arbitrary;
explicit TestDriver(Arbitrary &a) : arbitrary(&a) {
#if DEBUG_VERBOSE && !defined(NDEBUG)
fprintf(stderr, "%p Initial version: {%" PRId64 "}\n", this, writeVersion);
#endif
}
int64_t oldestVersion = arbitrary->next();
int64_t writeVersion = oldestVersion;
ConflictSetImpl cs{oldestVersion};
ReferenceImpl refImpl{oldestVersion};
constexpr static auto kMaxKeySuffixLen = 8;
bool ok = true;
const int prefixLen = arbitrary->bounded(512);
const int prefixByte = arbitrary->randT<uint8_t>();
// Call until it returns true, for "done". Check internal invariants etc
// between calls to next.
bool next() {
assert(cs.getBytes() >= 0);
if (!arbitrary->hasEntropy()) {
if (!arbitrary.hasEntropy()) {
return true;
}
Arena arena;
{
int numPointWrites = arbitrary->bounded(100);
int numRangeWrites = arbitrary->bounded(100);
int64_t v =
(writeVersion +=
arbitrary->bounded(10) ? arbitrary->bounded(10) : arbitrary->next());
int numPointWrites = arbitrary.bounded(100);
int numRangeWrites = arbitrary.bounded(100);
int64_t v = ++writeVersion;
auto *writes =
new (arena) ConflictSet::WriteRange[numPointWrites + numRangeWrites];
auto keys = set<std::string_view>(arena);
while (int(keys.size()) < numPointWrites + numRangeWrites * 2) {
if (!arbitrary->hasEntropy()) {
if (!arbitrary.hasEntropy()) {
return true;
}
int keyLen = prefixLen + arbitrary->bounded(kMaxKeySuffixLen);
int keyLen = arbitrary.bounded(kMaxKeyLen);
auto *begin = new (arena) uint8_t[keyLen];
memset(begin, prefixByte, prefixLen);
arbitrary->randomBytes(begin + prefixLen, keyLen - prefixLen);
arbitrary.randomBytes(begin, keyLen);
keys.insert(std::string_view((const char *)begin, keyLen));
}
@@ -643,7 +528,7 @@ struct TestDriver {
rangesRemaining = numRangeWrites;
pointsRemaining > 0 || rangesRemaining > 0; ++i) {
bool pointRead = pointsRemaining > 0 && rangesRemaining > 0
? bool(arbitrary->bounded(2))
? bool(arbitrary.bounded(2))
: pointsRemaining > 0;
if (pointRead) {
assert(pointsRemaining > 0);
@@ -662,123 +547,45 @@ struct TestDriver {
++iter;
--rangesRemaining;
}
#if DEBUG_VERBOSE && !defined(NDEBUG)
if (writes[i].end.len == 0) {
fprintf(stderr, "Write: {%s} -> %" PRId64 "\n",
printable(writes[i].begin).c_str(), writeVersion);
} else {
fprintf(stderr, "Write: [%s, %s) -> %" PRId64 "\n",
printable(writes[i].begin).c_str(),
printable(writes[i].end).c_str(), writeVersion);
}
#endif
}
assert(iter == keys.end());
assert(i == numPointWrites + numRangeWrites);
// Test non-canonical writes
if (numPointWrites > 0) {
int overlaps = arbitrary->bounded(numPointWrites);
for (int i = 0; i < numPointWrites + numRangeWrites && overlaps > 0;
++i) {
if (writes[i].end.len == 0) {
int keyLen = prefixLen + arbitrary->bounded(kMaxKeySuffixLen);
auto *begin = new (arena) uint8_t[keyLen];
memset(begin, prefixByte, prefixLen);
arbitrary->randomBytes(begin + prefixLen, keyLen - prefixLen);
writes[i].end.len = keyLen;
writes[i].end.p = begin;
auto c =
std::span<const uint8_t>(writes[i].begin.p,
writes[i].begin.len) <=>
std::span<const uint8_t>(writes[i].end.p, writes[i].end.len);
if (c > 0) {
using std::swap;
swap(writes[i].begin, writes[i].end);
} else if (c == 0) {
// It's a point write after all, I guess
writes[i].end.len = 0;
}
--overlaps;
}
}
}
if (arbitrary->bounded(2)) {
// Shuffle writes
for (int i = numPointWrites + numRangeWrites - 1; i > 0; --i) {
int j = arbitrary->bounded(i + 1);
if (i != j) {
using std::swap;
swap(writes[i], writes[j]);
}
}
}
oldestVersion +=
arbitrary->bounded(10) ? arbitrary->bounded(10) : arbitrary->next();
oldestVersion = std::min(oldestVersion, writeVersion);
#ifdef THREAD_TEST
std::latch ready{1};
std::thread thread2{[&]() {
ready.count_down();
ConflictSet::MetricsV1 *m;
int count;
cs.getMetricsV1(&m, &count);
for (int i = 0; i < count; ++i) {
m[i].getValue();
}
}};
ready.wait();
#endif
#if DEBUG_VERBOSE && !defined(NDEBUG)
for (int i = 0; i < numPointWrites + numRangeWrites; ++i) {
if (writes[i].end.len == 0) {
fprintf(stderr, "%p Write: {%s}\n", this,
printable(writes[i].begin).c_str());
} else {
fprintf(stderr, "%p Write: [%s, %s)\n", this,
printable(writes[i].begin).c_str(),
printable(writes[i].end).c_str());
}
}
fprintf(stderr, "%p Write @ %" PRId64 "\n", this, v);
#endif
CALLGRIND_START_INSTRUMENTATION;
cs.addWrites(writes, numPointWrites + numRangeWrites, v);
CALLGRIND_STOP_INSTRUMENTATION;
if constexpr (kEnableAssertions) {
refImpl.addWrites(writes, numPointWrites + numRangeWrites, v);
}
#if DEBUG_VERBOSE && !defined(NDEBUG)
fprintf(stderr, "%p Set oldest version: %" PRId64 "\n", this,
oldestVersion = std::max<int64_t>(writeVersion - arbitrary.bounded(10),
oldestVersion);
#endif
CALLGRIND_START_INSTRUMENTATION;
cs.setOldestVersion(oldestVersion);
CALLGRIND_STOP_INSTRUMENTATION;
if constexpr (kEnableAssertions) {
refImpl.setOldestVersion(oldestVersion);
}
#ifdef THREAD_TEST
thread2.join();
#endif
}
{
int numPointReads = arbitrary->bounded(100);
int numRangeReads = arbitrary->bounded(100);
int64_t v = std::max<int64_t>(writeVersion - (arbitrary->bounded(10)
? arbitrary->bounded(10)
: arbitrary->next()),
0);
int numPointReads = arbitrary.bounded(100);
int numRangeReads = arbitrary.bounded(100);
int64_t v = std::max<int64_t>(writeVersion - arbitrary.bounded(10), 0);
auto *reads =
new (arena) ConflictSet::ReadRange[numPointReads + numRangeReads];
auto keys = set<std::string_view>(arena);
while (int(keys.size()) < numPointReads + numRangeReads * 2) {
if (!arbitrary->hasEntropy()) {
if (!arbitrary.hasEntropy()) {
return true;
}
int keyLen = prefixLen + arbitrary->bounded(kMaxKeySuffixLen);
int keyLen = arbitrary.bounded(kMaxKeyLen);
auto *begin = new (arena) uint8_t[keyLen];
memset(begin, prefixByte, prefixLen);
arbitrary->randomBytes(begin + prefixLen, keyLen - prefixLen);
arbitrary.randomBytes(begin, keyLen);
keys.insert(std::string_view((const char *)begin, keyLen));
}
@@ -787,7 +594,7 @@ struct TestDriver {
for (int pointsRemaining = numPointReads, rangesRemaining = numRangeReads;
pointsRemaining > 0 || rangesRemaining > 0; ++i) {
bool pointRead = pointsRemaining > 0 && rangesRemaining > 0
? bool(arbitrary->bounded(2))
? bool(arbitrary.bounded(2))
: pointsRemaining > 0;
if (pointRead) {
assert(pointsRemaining > 0);
@@ -809,12 +616,12 @@ struct TestDriver {
reads[i].readVersion = v;
#if DEBUG_VERBOSE && !defined(NDEBUG)
if (reads[i].end.len == 0) {
fprintf(stderr, "%p Read: {%s} @ %" PRId64 "\n", this,
printable(reads[i].begin).c_str(), reads[i].readVersion);
fprintf(stderr, "Read: {%s} @ %d\n",
printable(reads[i].begin).c_str(), int(reads[i].readVersion));
} else {
fprintf(stderr, "%p Read: [%s, %s) @ %" PRId64 "\n", this,
fprintf(stderr, "Read: [%s, %s) @ %d\n",
printable(reads[i].begin).c_str(),
printable(reads[i].end).c_str(), reads[i].readVersion);
printable(reads[i].end).c_str(), int(reads[i].readVersion));
}
#endif
}
@@ -825,85 +632,32 @@ struct TestDriver {
auto *results2 =
new (arena) ConflictSet::Result[numPointReads + numRangeReads];
#ifdef THREAD_TEST
auto *results3 =
new (arena) ConflictSet::Result[numPointReads + numRangeReads];
std::latch ready{1};
std::thread thread2{[&]() {
ready.count_down();
// Call all const methods
cs.check(reads, results3, numPointReads + numRangeReads);
cs.getBytes();
ConflictSet::MetricsV1 *m;
int count;
cs.getMetricsV1(&m, &count);
for (int i = 0; i < count; ++i) {
m[i].getValue();
}
}};
ready.wait();
#endif
CALLGRIND_START_INSTRUMENTATION;
cs.check(reads, results1, numPointReads + numRangeReads);
CALLGRIND_STOP_INSTRUMENTATION;
if constexpr (kEnableAssertions) {
// Call remaining const methods
cs.getBytes();
ConflictSet::MetricsV1 *m;
int count;
cs.getMetricsV1(&m, &count);
for (int i = 0; i < count; ++i) {
m[i].getValue();
}
refImpl.check(reads, results2, numPointReads + numRangeReads);
}
auto compareResults = [reads, this](ConflictSet::Result *results1,
ConflictSet::Result *results2,
int count) {
for (int i = 0; i < count; ++i) {
for (int i = 0; i < numPointReads + numRangeReads; ++i) {
if (results1[i] != results2[i]) {
if (reads[i].end.len == 0) {
fprintf(stderr,
"Expected %s, got %s for read of {%s} at version %" PRId64
"\n",
resultToStr(results2[i]), resultToStr(results1[i]),
printable(reads[i].begin).c_str(), reads[i].readVersion);
} else {
fprintf(
stderr,
"%p Expected %s, got %s for read of {%s} at version %" PRId64
"Expected %s, got %s for read of [%s, %s) at version %" PRId64
"\n",
(void *)this, resultToStr(results2[i]),
resultToStr(results1[i]), printable(reads[i].begin).c_str(),
reads[i].readVersion);
} else {
fprintf(stderr,
"%p Expected %s, got %s for read of [%s, %s) at version "
"%" PRId64 "\n",
(void *)this, resultToStr(results2[i]),
resultToStr(results1[i]),
resultToStr(results2[i]), resultToStr(results1[i]),
printable(reads[i].begin).c_str(),
printable(reads[i].end).c_str(), reads[i].readVersion);
}
return false;
}
}
return true;
};
if constexpr (kEnableAssertions) {
if (!compareResults(results1, results2,
numPointReads + numRangeReads)) {
ok = false;
return true;
}
}
#ifdef THREAD_TEST
thread2.join();
if (!compareResults(results3, results2, numPointReads + numRangeReads)) {
ok = false;
return true;
}
#endif
}
return false;
}

55
Jenkinsfile vendored
View File

@@ -36,40 +36,6 @@ pipeline {
sh 'pre-commit run --all-files --show-diff-on-failure'
}
}
stage('Clang') {
agent {
dockerfile {
args '-v /home/jenkins/ccache:/ccache'
reuseNode true
}
}
steps {
CleanBuildAndTest("")
recordIssues(tools: [clang()])
}
}
stage('Debug') {
agent {
dockerfile {
args '-v /home/jenkins/ccache:/ccache'
reuseNode true
}
}
steps {
CleanBuildAndTest("-DCMAKE_BUILD_TYPE=Debug")
}
}
stage('SIMD fallback') {
agent {
dockerfile {
args '-v /home/jenkins/ccache:/ccache'
reuseNode true
}
}
steps {
CleanBuildAndTest("-DUSE_SIMD_FALLBACK=ON")
}
}
stage('Release [gcc]') {
agent {
dockerfile {
@@ -78,7 +44,7 @@ pipeline {
}
}
steps {
CleanBuildAndTest("-DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++ -DCMAKE_CXX_FLAGS=-DNVALGRIND")
CleanBuildAndTest("-DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++ -DCMAKE_BUILD_TYPE=Release")
recordIssues(tools: [gcc()])
sh '''
cd build
@@ -100,7 +66,7 @@ pipeline {
}
}
steps {
CleanBuildAndTest("-DCMAKE_TOOLCHAIN_FILE=../aarch64-toolchain.cmake -DCMAKE_CXX_FLAGS=-DNVALGRIND")
CleanBuildAndTest("-DCMAKE_TOOLCHAIN_FILE=../aarch64-toolchain.cmake")
sh '''
cd build
cpack -G DEB
@@ -117,18 +83,11 @@ pipeline {
}
}
steps {
script {
filter_args = "-f ConflictSet.cpp -f LongestCommonPrefix.h"
}
CleanBuildAndTest("-DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++ -DCMAKE_C_FLAGS=--coverage -DCMAKE_CXX_FLAGS=--coverage -DCMAKE_BUILD_TYPE=Debug -DDISABLE_TSAN=ON")
sh """
gcovr ${filter_args} --cobertura > build/coverage.xml
"""
recordCoverage qualityGates: [[criticality: 'NOTE', metric: 'MODULE']], tools: [[parser: 'COBERTURA', pattern: 'build/coverage.xml']]
sh """
gcovr ${filter_args}
gcovr ${filter_args} --fail-under-line 100 > /dev/null
"""
CleanBuildAndTest("-DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++ -DCMAKE_C_FLAGS=--coverage -DCMAKE_CXX_FLAGS=--coverage -DCMAKE_BUILD_TYPE=Debug")
sh '''
gcovr --exclude '.*third_party.*' --cobertura > build/coverage.xml
'''
cobertura autoUpdateHealth: false, autoUpdateStability: false, coberturaReportFile: 'build/coverage.xml', conditionalCoverageTargets: '70, 0, 0', failUnhealthy: false, failUnstable: false, lineCoverageTargets: '80, 0, 0', maxNumberOfBuilds: 0, methodCoverageTargets: '80, 0, 0', onlyStable: false, sourceEncoding: 'ASCII', zoomCoverageChart: false
}
}
}

View File

@@ -1,185 +0,0 @@
#pragma once
#include <assert.h>
#include <bit>
#include <stdint.h>
#include <string.h>
#ifdef HAS_AVX
#include <immintrin.h>
#elif HAS_ARM_NEON
#include <arm_neon.h>
#endif
#ifndef __SANITIZE_THREAD__
#if defined(__has_feature)
#if __has_feature(thread_sanitizer)
#define __SANITIZE_THREAD__
#endif
#endif
#endif
#if defined(HAS_AVX) || defined(HAS_ARM_NEON)
constexpr int kStride = 64;
#else
constexpr int kStride = 16;
#endif
constexpr int kUnrollFactor = 4;
inline bool compareStride(const uint8_t *ap, const uint8_t *bp) {
#if defined(HAS_ARM_NEON)
static_assert(kStride == 64);
uint8x16_t x[4]; // GCOVR_EXCL_LINE
for (int i = 0; i < 4; ++i) {
x[i] = vceqq_u8(vld1q_u8(ap + i * 16), vld1q_u8(bp + i * 16));
}
auto results = vreinterpretq_u16_u8(
vandq_u8(vandq_u8(x[0], x[1]), vandq_u8(x[2], x[3])));
bool eq = vget_lane_u64(vreinterpret_u64_u8(vshrn_n_u16(results, 4)), 0) ==
uint64_t(-1);
#elif defined(HAS_AVX)
static_assert(kStride == 64);
__m128i x[4]; // GCOVR_EXCL_LINE
for (int i = 0; i < 4; ++i) {
x[i] = _mm_cmpeq_epi8(_mm_loadu_si128((__m128i *)(ap + i * 16)),
_mm_loadu_si128((__m128i *)(bp + i * 16)));
}
auto eq =
_mm_movemask_epi8(_mm_and_si128(_mm_and_si128(x[0], x[1]),
_mm_and_si128(x[2], x[3]))) == 0xffff;
#else
// Hope it gets vectorized
auto eq = memcmp(ap, bp, kStride) == 0;
#endif
return eq;
}
// Precondition: ap[:kStride] != bp[:kStride]
inline int firstNeqStride(const uint8_t *ap, const uint8_t *bp) {
#if defined(HAS_AVX)
static_assert(kStride == 64);
uint64_t c[kStride / 16]; // GCOVR_EXCL_LINE
for (int i = 0; i < kStride; i += 16) {
const auto a = _mm_loadu_si128((__m128i *)(ap + i));
const auto b = _mm_loadu_si128((__m128i *)(bp + i));
const auto compared = _mm_cmpeq_epi8(a, b);
c[i / 16] = _mm_movemask_epi8(compared) & 0xffff;
}
return std::countr_zero(~(c[0] | c[1] << 16 | c[2] << 32 | c[3] << 48));
#elif defined(HAS_ARM_NEON)
static_assert(kStride == 64);
for (int i = 0; i < kStride; i += 16) {
// 0xff for each match
uint16x8_t results =
vreinterpretq_u16_u8(vceqq_u8(vld1q_u8(ap + i), vld1q_u8(bp + i)));
// 0xf for each mismatch
uint64_t bitfield =
~vget_lane_u64(vreinterpret_u64_u8(vshrn_n_u16(results, 4)), 0);
if (bitfield) {
return i + (std::countr_zero(bitfield) >> 2);
}
}
__builtin_unreachable(); // GCOVR_EXCL_LINE
#else
int i = 0;
for (; i < kStride - 1; ++i) {
if (*ap++ != *bp++) {
break;
}
}
return i;
#endif
}
// This gets covered in local development
// GCOVR_EXCL_START
#if defined(HAS_AVX) && !defined(__SANITIZE_THREAD__)
__attribute__((target("avx512f,avx512bw"))) inline int
longestCommonPrefix(const uint8_t *ap, const uint8_t *bp, int cl) {
int i = 0;
int end = cl & ~63;
while (i < end) {
const uint64_t eq =
_mm512_cmpeq_epi8_mask(_mm512_loadu_epi8(ap), _mm512_loadu_epi8(bp));
if (eq != uint64_t(-1)) {
return i + std::countr_one(eq);
}
i += 64;
ap += 64;
bp += 64;
}
if (i < cl) {
const uint64_t mask = (uint64_t(1) << (cl - i)) - 1;
const uint64_t eq = _mm512_cmpeq_epi8_mask(
_mm512_maskz_loadu_epi8(mask, ap), _mm512_maskz_loadu_epi8(mask, bp));
return i + std::countr_one(eq & mask);
}
assert(i == cl);
return i;
}
__attribute__((target("default")))
#endif
// GCOVR_EXCL_STOP
inline int
longestCommonPrefix(const uint8_t *ap, const uint8_t *bp, int cl) {
if (!(cl >= 0)) {
__builtin_unreachable(); // GCOVR_EXCL_LINE
}
int i = 0;
int end;
// kStride * kUnrollCount at a time
end = cl & ~(kStride * kUnrollFactor - 1);
while (i < end) {
for (int j = 0; j < kUnrollFactor; ++j) {
if (!compareStride(ap, bp)) {
return i + firstNeqStride(ap, bp);
}
i += kStride;
ap += kStride;
bp += kStride;
}
}
// kStride at a time
end = cl & ~(kStride - 1);
while (i < end) {
if (!compareStride(ap, bp)) {
return i + firstNeqStride(ap, bp);
}
i += kStride;
ap += kStride;
bp += kStride;
}
// word at a time
end = cl & ~(sizeof(uint64_t) - 1);
while (i < end) {
uint64_t a; // GCOVR_EXCL_LINE
uint64_t b; // GCOVR_EXCL_LINE
memcpy(&a, ap, 8);
memcpy(&b, bp, 8);
const auto mismatched = a ^ b;
if (mismatched) {
return i + std::countr_zero(mismatched) / 8;
}
i += 8;
ap += 8;
bp += 8;
}
// byte at a time
while (i < cl) {
if (*ap != *bp) {
break;
}
++ap;
++bp;
++i;
}
return i;
}

102
README.md
View File

@@ -1,38 +1,82 @@
A data structure for optimistic concurrency control on ranges of bitwise-lexicographically-ordered keys.
Intended as an alternative to FoundationDB's skip list.
Intended to replace FoundationDB's skip list.
Hardware for all benchmarks is an AMD Ryzen 9 7900 with (2x32GB) 5600MT/s CL28-34-34-89 1.35V RAM
# Microbenchmark
# FoundationDB's benchmark
## Skip list
| ns/op | op/s | err% | ins/op | cyc/op | IPC | bra/op | miss% | total | benchmark
|--------------------:|--------------------:|--------:|----------------:|----------------:|-------:|---------------:|--------:|----------:|:----------
| 172.03 | 5,812,791.77 | 0.4% | 3,130.62 | 879.00 | 3.562 | 509.23 | 0.0% | 0.01 | `point reads`
| 167.44 | 5,972,130.71 | 0.2% | 3,065.14 | 862.27 | 3.555 | 494.30 | 0.0% | 0.01 | `prefix reads`
| 238.77 | 4,188,130.84 | 0.9% | 3,589.93 | 1,259.30 | 2.851 | 637.12 | 0.0% | 0.01 | `range reads`
| 424.01 | 2,358,426.70 | 0.2% | 5,620.05 | 2,242.35 | 2.506 | 854.80 | 1.7% | 0.01 | `point writes`
| 418.45 | 2,389,780.56 | 0.4% | 5,525.07 | 2,211.05 | 2.499 | 831.71 | 1.7% | 0.01 | `prefix writes`
| 254.87 | 3,923,568.88 | 2.6% | 3,187.01 | 1,366.50 | 2.332 | 529.11 | 2.7% | 0.02 | `range writes`
| 675.96 | 1,479,374.50 | 3.3% | 7,735.41 | 3,468.60 | 2.230 | 1,386.02 | 1.8% | 0.01 | `monotonic increasing point writes`
| 137,986.20 | 7,247.10 | 0.6% | 789,752.33 | 699,462.00 | 1.129 | 144,824.14 | 0.0% | 0.01 | `worst case for radix tree`
| 21.63 | 46,231,564.03 | 1.0% | 448.00 | 107.14 | 4.181 | 84.00 | 0.0% | 0.01 | `create and destroy`
```
New conflict set: 2.404 sec
0.520 Mtransactions/sec
2.080 Mkeys/sec
Detect only: 2.266 sec
0.552 Mtransactions/sec
2.207 Mkeys/sec
Skiplist only: 1.594 sec
0.784 Mtransactions/sec
3.137 Mkeys/sec
Performance counters:
Build: 0.071
Add: 0.0641
Detect: 2.27
D.Sort: 0.44
D.Combine: 0.018
D.CheckRead: 0.855
D.CheckIntraBatch: 0.00903
D.MergeWrite: 0.739
D.RemoveBefore: 0.201
```
## Radix tree (this implementation)
| ns/op | op/s | err% | ins/op | cyc/op | IPC | bra/op | miss% | total | benchmark
|--------------------:|--------------------:|--------:|----------------:|----------------:|-------:|---------------:|--------:|----------:|:----------
| 11.18 | 89,455,125.34 | 0.6% | 185.37 | 57.08 | 3.248 | 41.51 | 0.4% | 0.01 | `point reads`
| 14.53 | 68,800,688.89 | 0.4% | 282.41 | 74.80 | 3.776 | 55.06 | 0.3% | 0.01 | `prefix reads`
| 36.54 | 27,367,576.87 | 0.2% | 798.06 | 188.90 | 4.225 | 141.69 | 0.2% | 0.01 | `range reads`
| 16.69 | 59,912,106.02 | 0.6% | 314.57 | 86.29 | 3.645 | 39.84 | 0.4% | 0.01 | `point writes`
| 30.09 | 33,235,744.07 | 0.5% | 591.33 | 155.92 | 3.793 | 82.69 | 0.2% | 0.01 | `prefix writes`
| 35.77 | 27,956,388.03 | 1.4% | 682.25 | 187.63 | 3.636 | 96.12 | 0.1% | 0.01 | `range writes`
| 74.04 | 13,505,408.41 | 2.7% | 1,448.95 | 392.10 | 3.695 | 260.53 | 0.1% | 0.01 | `monotonic increasing point writes`
| 330,984.50 | 3,021.29 | 1.9% | 3,994,153.50 | 1,667,309.00 | 2.396 | 806,019.50 | 0.0% | 0.01 | `worst case for radix tree`
| 92.46 | 10,814,961.65 | 0.5% | 1,800.00 | 463.41 | 3.884 | 297.00 | 0.0% | 0.01 | `create and destroy`
```
New conflict set: 1.743 sec
0.717 Mtransactions/sec
2.869 Mkeys/sec
Detect only: 1.611 sec
0.776 Mtransactions/sec
3.103 Mkeys/sec
Skiplist only: 0.919 sec
1.360 Mtransactions/sec
5.440 Mkeys/sec
Performance counters:
Build: 0.0657
Add: 0.0628
Detect: 1.61
D.Sort: 0.442
D.Combine: 0.0178
D.CheckRead: 0.395
D.CheckIntraBatch: 0.00776
D.MergeWrite: 0.524
D.RemoveBefore: 0.221
```
# Our benchmark
## Skip list
| ns/op | op/s | err% | total | benchmark
|--------------------:|--------------------:|--------:|----------:|:----------
| 270.07 | 3,702,706.03 | 0.4% | 0.01 | `point reads`
| 285.76 | 3,499,437.03 | 1.5% | 0.01 | `prefix reads`
| 532.54 | 1,877,794.90 | 0.7% | 0.01 | `range reads`
| 528.50 | 1,892,132.94 | 0.7% | 0.01 | `point writes`
| 516.53 | 1,935,978.22 | 0.9% | 0.01 | `prefix writes`
| 303.34 | 3,296,630.84 | 3.6% | 0.05 | `range writes`
| 502.88 | 1,988,553.24 | 2.0% | 0.01 | `monotonic increasing point writes`
## Radix tree (this implementation)
| ns/op | op/s | err% | total | benchmark
|--------------------:|--------------------:|--------:|----------:|:----------
| 14.52 | 68,850,842.99 | 1.2% | 0.01 | `point reads`
| 60.89 | 16,422,538.22 | 1.5% | 0.01 | `prefix reads`
| 226.89 | 4,407,362.98 | 0.5% | 0.01 | `range reads`
| 22.99 | 43,498,198.49 | 0.2% | 0.01 | `point writes`
| 50.51 | 19,799,864.54 | 1.0% | 0.01 | `prefix writes`
| 82.50 | 12,121,212.12 | 2.6% | 0.03 | `range writes`
| 119.94 | 8,337,354.54 | 2.1% | 0.01 | `monotonic increasing point writes`
# "Real data" test
@@ -41,13 +85,13 @@ Point queries only, best of three runs. Gc ratio is the ratio of time spent doin
## skip list
```
Check: 4.47891 seconds, 364.05 MB/s, Add: 4.55599 seconds, 123.058 MB/s, Gc ratio: 37.1145%
Check: 12.7863 seconds, 292.384 MB/s, Add: 19.8276 seconds, 35.4071 MB/s, Gc ratio: 23.5314%
```
## radix tree
```
Check: 0.953012 seconds, 1710.94 MB/s, Add: 1.30025 seconds, 431.188 MB/s, Gc ratio: 43.9816%, Peak idle memory: 2.28375e+06
Check: 3.60187 seconds, 1037.94 MB/s, Add: 3.03958 seconds, 230.966 MB/s, Gc ratio: 52.3876%
```
## hash table
@@ -55,5 +99,5 @@ Check: 0.953012 seconds, 1710.94 MB/s, Add: 1.30025 seconds, 431.188 MB/s, Gc ra
(The hash table implementation doesn't work on range queries, and its purpose is to provide an idea of how fast point queries can be)
```
Check: 0.804094 seconds, 2027.81 MB/s, Add: 0.652952 seconds, 858.645 MB/s, Gc ratio: 35.3885%
Check: 2.15925 seconds, 1731.4 MB/s, Add: 1.08519 seconds, 646.926 MB/s, Gc ratio: 52.1526%
```

View File

@@ -1,18 +1,13 @@
#include <ConflictSet.h>
#include <cerrno>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <fcntl.h>
#include <string_view>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <vector>
using namespace weaselab;
double now() {
return std::chrono::duration_cast<std::chrono::nanoseconds>(
std::chrono::steady_clock::now().time_since_epoch())
@@ -33,7 +28,7 @@ constexpr inline size_t rightAlign(size_t offset, size_t alignment) {
int main(int argc, const char **argv) {
// Use with this dataset https://snap.stanford.edu/data/memetracker9.html
// Preprocess the files with `sed -i'' '/^Q/d'`
// Preprocess the files with `sed -i '' '/^Q/d'`
double checkTime = 0;
double addTime = 0;
@@ -45,8 +40,6 @@ int main(int argc, const char **argv) {
int64_t version = 0;
double timer = 0;
int64_t peakMemory = 0;
for (int i = 1; i < argc; ++i) {
int fd = open(argv[i], O_RDONLY);
struct stat st;
@@ -116,10 +109,6 @@ int main(int argc, const char **argv) {
write = {};
reads.clear();
if (cs.getBytes() > peakMemory) {
peakMemory = cs.getBytes();
}
timer = now();
cs.setOldestVersion(version - 10000);
gcTime += now() - timer;
@@ -129,19 +118,8 @@ int main(int argc, const char **argv) {
close(fd);
}
ConflictSet::MetricsV1 *metrics;
int metricsCount;
cs.getMetricsV1(&metrics, &metricsCount);
for (int i = 0; i < metricsCount; ++i) {
printf("# HELP %s %s\n", metrics[i].name, metrics[i].help);
printf("# TYPE %s %s\n", metrics[i].name,
metrics[i].type == metrics[i].Counter ? "counter" : "gauge");
printf("%s %g\n", metrics[i].name, metrics[i].getValue());
}
printf("Check: %g seconds, %g MB/s, Add: %g seconds, %g MB/s, Gc ratio: "
"%g%%, Peak idle memory: %g\n",
printf(
"Check: %g seconds, %g MB/s, Add: %g seconds, %g MB/s, Gc ratio: %g%%\n",
checkTime, checkBytes / checkTime * 1e-6, addTime,
addBytes / addTime * 1e-6, gcTime / (gcTime + addTime) * 1e2,
double(peakMemory));
addBytes / addTime * 1e-6, gcTime / (gcTime + addTime) * 1e2);
}

View File

@@ -1,369 +0,0 @@
#include <atomic>
#include <errno.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <string>
#include <string_view>
#include <sys/ioctl.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <thread>
#include <unistd.h>
#include <utility>
#include <vector>
#include "ConflictSet.h"
#include "third_party/nadeau.h"
std::atomic<int64_t> transactions;
constexpr int kBaseSearchDepth = 32;
constexpr int kWindowSize = 10000000;
std::string numToKey(int64_t num) {
std::string result;
result.resize(kBaseSearchDepth + sizeof(int64_t));
memset(result.data(), 0, kBaseSearchDepth);
int64_t be = __builtin_bswap64(num);
memcpy(result.data() + kBaseSearchDepth, &be, sizeof(int64_t));
return result;
}
void workload(weaselab::ConflictSet *cs) {
int64_t version = kWindowSize;
cs->addWrites(nullptr, 0, version);
for (;; transactions.fetch_add(1, std::memory_order_relaxed)) {
// Reads
{
auto beginK = numToKey(version - kWindowSize);
auto endK = numToKey(version - 1);
auto pointRv = version - kWindowSize + rand() % kWindowSize + 1;
auto pointK = numToKey(pointRv);
weaselab::ConflictSet::ReadRange reads[] = {
{
{(const uint8_t *)pointK.data(), int(pointK.size())},
{nullptr, 0},
pointRv,
},
{
{(const uint8_t *)beginK.data(), int(beginK.size())},
{(const uint8_t *)endK.data(), int(endK.size())},
version - 2,
},
};
weaselab::ConflictSet::Result result[sizeof(reads) / sizeof(reads[0])];
cs->check(reads, result, sizeof(reads) / sizeof(reads[0]));
// for (int i = 0; i < sizeof(reads) / sizeof(reads[0]); ++i) {
// if (result[i] != weaselab::ConflictSet::Commit) {
// fprintf(stderr, "Unexpected conflict: [%s, %s) @ %" PRId64 "\n",
// printable(reads[i].begin).c_str(),
// printable(reads[i].end).c_str(), reads[i].readVersion);
// abort();
// }
// }
}
// Writes
{
weaselab::ConflictSet::WriteRange w;
auto k = numToKey(version);
w.begin.p = (const uint8_t *)k.data();
w.end.len = 0;
if (version % (kWindowSize / 2) == 0) {
for (int l = 0; l <= k.size(); ++l) {
w.begin.len = l;
cs->addWrites(&w, 1, version);
}
} else {
w.begin.len = k.size();
cs->addWrites(&w, 1, version);
int64_t beginN = version - kWindowSize + rand() % kWindowSize;
auto b = numToKey(beginN);
auto e = numToKey(beginN + 1000);
w.begin.p = (const uint8_t *)b.data();
w.begin.len = b.size();
w.end.p = (const uint8_t *)e.data();
w.end.len = e.size();
cs->addWrites(&w, 1, version);
}
}
// GC
cs->setOldestVersion(version - kWindowSize);
++version;
}
}
// Adapted from getaddrinfo man page
int getListenFd(const char *node, const char *service) {
struct addrinfo hints;
struct addrinfo *result, *rp;
int sfd, s;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* stream socket */
hints.ai_flags = AI_PASSIVE; /* For wildcard IP address */
hints.ai_protocol = 0; /* Any protocol */
hints.ai_canonname = nullptr;
hints.ai_addr = nullptr;
hints.ai_next = nullptr;
s = getaddrinfo(node, service, &hints, &result);
if (s != 0) {
fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s));
abort();
}
/* getaddrinfo() returns a list of address structures.
Try each address until we successfully bind(2).
If socket(2) (or bind(2)) fails, we (close the socket
and) try the next address. */
for (rp = result; rp != nullptr; rp = rp->ai_next) {
sfd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
if (sfd == -1) {
continue;
}
int val = 1;
setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
if (bind(sfd, rp->ai_addr, rp->ai_addrlen) == 0) {
break; /* Success */
}
close(sfd);
}
freeaddrinfo(result); /* No longer needed */
if (rp == nullptr) { /* No address succeeded */
fprintf(stderr, "Could not bind\n");
abort();
}
int rv = listen(sfd, SOMAXCONN);
if (rv) {
perror("listen()");
abort();
}
return sfd;
}
// HTTP response
//
std::string_view part1 =
"HTTP/1.1 200 OK \r\nContent-type: text/plain; version=0.0.4; "
"charset=utf-8; escaping=values\r\nContent-Length: ";
// Decimal content length
std::string_view part2 = "\r\n\r\n";
// Body
double toSeconds(timeval t) {
return double(t.tv_sec) + double(t.tv_usec) * 1e-6;
}
#ifdef __linux__
#include <linux/perf_event.h>
struct PerfCounter {
PerfCounter(int type, int config, const std::string &labels = {},
int groupLeaderFd = -1)
: labels(labels) {
struct perf_event_attr pe;
memset(&pe, 0, sizeof(pe));
pe.type = type;
pe.size = sizeof(pe);
pe.config = config;
pe.inherit = 1;
pe.exclude_kernel = 1;
pe.exclude_hv = 1;
fd = perf_event_open(&pe, 0, -1, groupLeaderFd, 0);
if (fd < 0 && errno != ENOENT && errno != EINVAL) {
perror(labels.c_str());
}
}
int64_t total() const {
int64_t count;
if (read(fd, &count, sizeof(count)) != sizeof(count)) {
perror("read instructions from perf");
abort();
}
return count;
}
PerfCounter(PerfCounter &&other)
: fd(std::exchange(other.fd, -1)), labels(std::move(other.labels)) {}
PerfCounter &operator=(PerfCounter &&other) {
fd = std::exchange(other.fd, -1);
labels = std::move(other.labels);
return *this;
}
~PerfCounter() {
if (fd >= 0) {
close(fd);
}
}
bool ok() const { return fd >= 0; }
const std::string &getLabels() const { return labels; }
int getFd() const { return fd; }
private:
int fd;
std::string labels;
static long perf_event_open(struct perf_event_attr *hw_event, pid_t pid,
int cpu, int group_fd, unsigned long flags) {
int ret;
ret = syscall(SYS_perf_event_open, hw_event, pid, cpu, group_fd, flags);
return ret;
}
};
#endif
int main(int argc, char **argv) {
if (argc != 3) {
goto fail;
}
{
int listenFd = getListenFd(argv[1], argv[2]);
weaselab::ConflictSet cs{0};
weaselab::ConflictSet::MetricsV1 *metrics;
int metricsCount;
cs.getMetricsV1(&metrics, &metricsCount);
#ifdef __linux__
PerfCounter instructions{PERF_TYPE_HARDWARE, PERF_COUNT_HW_INSTRUCTIONS};
PerfCounter cycles{PERF_TYPE_HARDWARE, PERF_COUNT_HW_CPU_CYCLES, "",
instructions.getFd()};
std::vector<PerfCounter> cacheCounters;
for (auto [id, idStr] : std::initializer_list<std::pair<int, std::string>>{
{PERF_COUNT_HW_CACHE_L1D, "l1d"},
{PERF_COUNT_HW_CACHE_L1I, "l1i"},
{PERF_COUNT_HW_CACHE_LL, "ll"},
{PERF_COUNT_HW_CACHE_DTLB, "dtlb"},
{PERF_COUNT_HW_CACHE_ITLB, "itlb"},
{PERF_COUNT_HW_CACHE_BPU, "bpu"},
{PERF_COUNT_HW_CACHE_NODE, "node"},
}) {
for (auto [op, opStr] :
std::initializer_list<std::pair<int, std::string>>{
{PERF_COUNT_HW_CACHE_OP_READ, "read"},
{PERF_COUNT_HW_CACHE_OP_WRITE, "write"},
{PERF_COUNT_HW_CACHE_OP_PREFETCH, "prefetch"},
}) {
int groupLeaderFd = -1;
for (auto [result, resultStr] :
std::initializer_list<std::pair<int, std::string>>{
{PERF_COUNT_HW_CACHE_RESULT_MISS, "miss"},
{PERF_COUNT_HW_CACHE_RESULT_ACCESS, "access"},
}) {
auto labels = "{id=\"" + idStr + "\", op=\"" + opStr +
"\", result=\"" + resultStr + "\"}";
cacheCounters.emplace_back(PERF_TYPE_HW_CACHE,
id | (op << 8) | (result << 16), labels,
groupLeaderFd);
if (!cacheCounters.back().ok()) {
cacheCounters.pop_back();
} else {
if (groupLeaderFd == -1) {
groupLeaderFd = cacheCounters.back().getFd();
}
}
}
}
}
#endif
auto w = std::thread{workload, &cs};
for (;;) {
struct sockaddr_storage peer_addr = {};
socklen_t peer_addr_len = sizeof(peer_addr);
const int connfd =
accept(listenFd, (struct sockaddr *)&peer_addr, &peer_addr_len);
std::string body;
rusage r;
getrusage(RUSAGE_SELF, &r);
body += "# HELP process_cpu_seconds_total Total user and system CPU time "
"spent in seconds.\n# TYPE process_cpu_seconds_total counter\n"
"process_cpu_seconds_total ";
body += std::to_string(toSeconds(r.ru_utime) + toSeconds(r.ru_stime));
body += "\n";
body += "# HELP process_resident_memory_bytes Resident memory size in "
"bytes.\n# TYPE process_resident_memory_bytes gauge\n"
"process_resident_memory_bytes ";
body += std::to_string(getCurrentRSS());
body += "\n";
body += "# HELP transactions_total Total number of transactions\n"
"# TYPE transactions_total counter\n"
"transactions_total ";
body += std::to_string(transactions.load(std::memory_order_relaxed));
body += "\n";
#ifdef __linux__
body += "# HELP instructions_total Total number of instructions\n"
"# TYPE instructions_total counter\n"
"instructions_total ";
body += std::to_string(instructions.total());
body += "\n";
body += "# HELP cycles_total Total number of cycles\n"
"# TYPE cycles_total counter\n"
"cycles_total ";
body += std::to_string(cycles.total());
body += "\n";
body += "# HELP cache_event_total Total number of cache events\n"
"# TYPE cache_event_total counter\n";
for (const auto &counter : cacheCounters) {
body += "cache_event_total" + counter.getLabels() + " " +
std::to_string(counter.total()) + "\n";
}
#endif
for (int i = 0; i < metricsCount; ++i) {
body += "# HELP ";
body += metrics[i].name;
body += " ";
body += metrics[i].help;
body += "\n";
body += "# TYPE ";
body += metrics[i].name;
body += " ";
body += metrics[i].type == metrics[i].Counter ? "counter" : "gauge";
body += "\n";
body += metrics[i].name;
body += " ";
body += std::to_string(metrics[i].getValue());
body += "\n";
}
auto len = std::to_string(body.size());
iovec iov[] = {
{(void *)part1.data(), part1.size()},
{(void *)len.data(), len.size()},
{(void *)part2.data(), part2.size()},
{(void *)body.data(), body.size()},
};
int written;
do {
written = writev(connfd, iov, sizeof(iov) / sizeof(iov[0]));
} while (written < 0 && errno == EINTR);
close(connfd);
}
}
fail:
fprintf(stderr, "Expected ./%s <host> <port>\n", argv[0]);
return 1;
}

View File

@@ -16,14 +16,10 @@
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* This source code is modified to compile outside of FoundationDB
*/
#include "ConflictSet.h"
#include "Internal.h"
#include <algorithm>
#include <span>
std::span<const uint8_t> keyAfter(Arena &arena, std::span<const uint8_t> key) {
@@ -42,7 +38,7 @@ std::span<const uint8_t> copyToArena(Arena &arena,
}
using Version = int64_t;
#define force_inline inline __attribute__((always_inline))
#define force_inline __attribute__((always_inline))
using StringRef = std::span<const uint8_t>;
struct KeyRangeRef {
@@ -54,135 +50,6 @@ struct KeyRangeRef {
: begin(begin), end(keyAfter(arena, begin)) {}
};
struct KeyInfo {
StringRef key;
bool begin;
bool write;
KeyInfo() = default;
KeyInfo(StringRef key, bool begin, bool write)
: key(key), begin(begin), write(write) {}
};
force_inline int extra_ordering(const KeyInfo &ki) {
return ki.begin * 2 + (ki.write ^ ki.begin);
}
// returns true if done with string
force_inline bool getCharacter(const KeyInfo &ki, int character,
int &outputCharacter) {
// normal case
if (character < ki.key.size()) {
outputCharacter = 5 + ki.key.begin()[character];
return false;
}
// termination
if (character == ki.key.size()) {
outputCharacter = 0;
return false;
}
if (character == ki.key.size() + 1) {
// end/begin+read/write relative sorting
outputCharacter = extra_ordering(ki);
return false;
}
outputCharacter = 0;
return true;
}
bool operator<(const KeyInfo &lhs, const KeyInfo &rhs) {
int i = std::min(lhs.key.size(), rhs.key.size());
int c = memcmp(lhs.key.data(), rhs.key.data(), i);
if (c != 0)
return c < 0;
// Always sort shorter keys before longer keys.
if (lhs.key.size() < rhs.key.size()) {
return true;
}
if (lhs.key.size() > rhs.key.size()) {
return false;
}
// When the keys are the same length, use the extra ordering constraint.
return extra_ordering(lhs) < extra_ordering(rhs);
}
bool operator==(const KeyInfo &lhs, const KeyInfo &rhs) {
return !(lhs < rhs || rhs < lhs);
}
void swapSort(std::vector<KeyInfo> &points, int a, int b) {
if (points[b] < points[a]) {
KeyInfo temp;
temp = points[a];
points[a] = points[b];
points[b] = temp;
}
}
struct SortTask {
int begin;
int size;
int character;
SortTask(int begin, int size, int character)
: begin(begin), size(size), character(character) {}
};
void sortPoints(std::vector<KeyInfo> &points) {
std::vector<SortTask> tasks;
std::vector<KeyInfo> newPoints;
std::vector<int> counts;
tasks.emplace_back(0, points.size(), 0);
while (tasks.size()) {
SortTask st = tasks.back();
tasks.pop_back();
if (st.size < 10) {
std::sort(points.begin() + st.begin, points.begin() + st.begin + st.size);
continue;
}
newPoints.resize(st.size);
counts.assign(256 + 5, 0);
// get counts
int c;
bool allDone = true;
for (int i = st.begin; i < st.begin + st.size; i++) {
allDone &= getCharacter(points[i], st.character, c);
counts[c]++;
}
if (allDone)
continue;
// calculate offsets from counts and build next level of tasks
int total = 0;
for (int i = 0; i < counts.size(); i++) {
int temp = counts[i];
if (temp > 1)
tasks.emplace_back(st.begin + total, temp, st.character + 1);
counts[i] = total;
total += temp;
}
// put in their places
for (int i = st.begin; i < st.begin + st.size; i++) {
getCharacter(points[i], st.character, c);
newPoints[counts[c]++] = points[i];
}
// copy back into original points array
for (int i = 0; i < st.size; i++)
points[st.begin + i] = newPoints[i];
}
}
static thread_local uint32_t g_seed = 0;
static inline int skfastrand() {
@@ -282,7 +149,7 @@ private:
setMaxVersion(level, v);
}
void destroy() { safe_free(this, getNodeSize()); }
void destroy() { free(this); }
private:
int getNodeSize() const {
@@ -401,21 +268,13 @@ public:
}
explicit SkipList(Version version = 0) {
#if DEBUG_VERBOSE
fprintf(stderr, "skip_list: create\n");
#endif
header = Node::create(StringRef(), MaxLevels - 1);
for (int l = 0; l < MaxLevels; l++) {
header->setNext(l, nullptr);
header->setMaxVersion(l, version);
}
}
~SkipList() {
#if DEBUG_VERBOSE
fprintf(stderr, "skip_list: destroy\n");
#endif
destroy();
}
~SkipList() { destroy(); }
SkipList(SkipList &&other) noexcept : header(other.header) {
other.header = nullptr;
}
@@ -708,8 +567,7 @@ struct SkipListConflictSet {};
struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
Impl(int64_t oldestVersion)
: oldestVersion(oldestVersion), newestVersion(oldestVersion),
skipList(oldestVersion) {}
: oldestVersion(oldestVersion), skipList(oldestVersion) {}
void check(const ConflictSet::ReadRange *reads, ConflictSet::Result *results,
int count) const {
Arena arena;
@@ -724,8 +582,7 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
}
skipList.detectConflicts(ranges, count, results);
for (int i = 0; i < count; ++i) {
if (reads[i].readVersion < oldestVersion ||
reads[i].readVersion < newestVersion - 2e9) {
if (reads[i].readVersion < oldestVersion) {
results[i] = TooOld;
}
}
@@ -733,40 +590,8 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
void addWrites(const ConflictSet::WriteRange *writes, int count,
int64_t writeVersion) {
auto points = std::vector<KeyInfo>(count * 2);
Arena arena;
for (int r = 0; r < count; r++) {
points.emplace_back(StringRef(writes[r].begin.p, writes[r].begin.len),
true, true);
points.emplace_back(
writes[r].end.len > 0
? StringRef{writes[r].end.p, size_t(writes[r].end.len)}
: keyAfter(arena, points.back().key),
false, true);
}
sortPoints(points);
int activeWriteCount = 0;
std::vector<std::pair<StringRef, StringRef>> combinedWriteConflictRanges;
for (const KeyInfo &point : points) {
if (point.write) {
if (point.begin) {
activeWriteCount++;
if (activeWriteCount == 1)
combinedWriteConflictRanges.emplace_back(point.key, StringRef());
} else /*if (point.end)*/ {
activeWriteCount--;
if (activeWriteCount == 0)
combinedWriteConflictRanges.back().second = point.key;
}
}
}
assert(writeVersion >= newestVersion);
newestVersion = writeVersion;
const int stringCount = combinedWriteConflictRanges.size() * 2;
const int stringCount = count * 2;
const int stripeSize = 16;
SkipList::Finger fingers[stripeSize];
@@ -777,10 +602,12 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
int ss = stringCount - (stripes - 1) * stripeSize;
for (int s = stripes - 1; s >= 0; s--) {
for (int i = 0; i * 2 < ss; ++i) {
const auto &w = combinedWriteConflictRanges[s * stripeSize / 2 + i];
values[i * 2] = w.first;
values[i * 2 + 1] = w.second;
keyUpdates += 3;
const auto &w = writes[s * stripeSize / 2 + i];
values[i * 2] = {w.begin.p, size_t(w.begin.len)};
values[i * 2 + 1] = w.end.len > 0
? StringRef{w.end.p, size_t(w.end.len)}
: keyAfter(arena, values[i * 2]);
keyUpdates += 2;
}
skipList.find(values, fingers, temp, ss);
skipList.addConflictRanges(fingers, ss / 2, writeVersion);
@@ -789,114 +616,46 @@ struct __attribute__((visibility("hidden"))) ConflictSet::Impl {
}
void setOldestVersion(int64_t oldestVersion) {
assert(oldestVersion >= this->oldestVersion);
this->oldestVersion = oldestVersion;
SkipList::Finger finger;
int temp;
std::span<const uint8_t> key = removalKey;
skipList.find(&key, &finger, &temp, 1);
skipList.removeBefore(oldestVersion, finger, std::exchange(keyUpdates, 10));
skipList.removeBefore(oldestVersion, finger, std::exchange(keyUpdates, 0));
removalArena = Arena();
removalKey = copyToArena(
removalArena, {finger.getValue().data(), finger.getValue().size()});
}
int64_t totalBytes = 0;
private:
int64_t keyUpdates = 10;
int64_t keyUpdates = 0;
Arena removalArena;
std::span<const uint8_t> removalKey;
int64_t oldestVersion;
int64_t newestVersion;
SkipList skipList;
};
// Internal entry points. Public entry points should just delegate to these
void internal_check(ConflictSet::Impl *impl,
const ConflictSet::ReadRange *reads,
ConflictSet::Result *results, int count) {
impl->check(reads, results, count);
}
void internal_addWrites(ConflictSet::Impl *impl,
const ConflictSet::WriteRange *writes, int count,
int64_t writeVersion) {
mallocBytesDelta = 0;
impl->addWrites(writes, count, writeVersion);
impl->totalBytes += mallocBytesDelta;
#if SHOW_MEMORY
if (impl->totalBytes != mallocBytes) {
abort();
}
#endif
}
void internal_setOldestVersion(ConflictSet::Impl *impl, int64_t oldestVersion) {
mallocBytesDelta = 0;
impl->setOldestVersion(oldestVersion);
impl->totalBytes += mallocBytesDelta;
#if SHOW_MEMORY
if (impl->totalBytes != mallocBytes) {
abort();
}
#endif
}
ConflictSet::Impl *internal_create(int64_t oldestVersion) {
mallocBytesDelta = 0;
auto *result = new (safe_malloc(sizeof(ConflictSet::Impl)))
ConflictSet::Impl{oldestVersion};
result->totalBytes += mallocBytesDelta;
return result;
}
void internal_destroy(ConflictSet::Impl *impl) {
impl->~Impl();
safe_free(impl, sizeof(ConflictSet::Impl));
}
int64_t internal_getBytes(ConflictSet::Impl *impl) { return impl->totalBytes; }
void internal_getMetricsV1(ConflictSet::Impl *impl,
ConflictSet::MetricsV1 **metrics, int *count) {
*metrics = nullptr;
*count = 0;
}
double internal_getMetricValue(const ConflictSet::MetricsV1 *metric) {
return 0;
}
void ConflictSet::check(const ReadRange *reads, Result *results,
int count) const {
internal_check(impl, reads, results, count);
return impl->check(reads, results, count);
}
void ConflictSet::addWrites(const WriteRange *writes, int count,
int64_t writeVersion) {
internal_addWrites(impl, writes, count, writeVersion);
return impl->addWrites(writes, count, writeVersion);
}
void ConflictSet::setOldestVersion(int64_t oldestVersion) {
internal_setOldestVersion(impl, oldestVersion);
}
int64_t ConflictSet::getBytes() const { return internal_getBytes(impl); }
void ConflictSet::getMetricsV1(MetricsV1 **metrics, int *count) const {
return internal_getMetricsV1(impl, metrics, count);
}
double ConflictSet::MetricsV1::getValue() const {
return internal_getMetricValue(this);
return impl->setOldestVersion(oldestVersion);
}
ConflictSet::ConflictSet(int64_t oldestVersion)
: impl(internal_create(oldestVersion)) {}
: impl(new (safe_malloc(sizeof(Impl))) Impl{oldestVersion}) {}
ConflictSet::~ConflictSet() {
if (impl) {
internal_destroy(impl);
impl->~Impl();
free(impl);
}
}
@@ -917,36 +676,25 @@ extern "C" {
__attribute__((__visibility__("default"))) void
ConflictSet_check(void *cs, const ConflictSet_ReadRange *reads,
ConflictSet_Result *results, int count) {
internal_check((ConflictSet::Impl *)cs, reads, results, count);
((ConflictSet::Impl *)cs)->check(reads, results, count);
}
__attribute__((__visibility__("default"))) void
ConflictSet_addWrites(void *cs, const ConflictSet_WriteRange *writes, int count,
int64_t writeVersion) {
internal_addWrites((ConflictSet::Impl *)cs, writes, count, writeVersion);
((ConflictSet::Impl *)cs)->addWrites(writes, count, writeVersion);
}
__attribute__((__visibility__("default"))) void
ConflictSet_setOldestVersion(void *cs, int64_t oldestVersion) {
internal_setOldestVersion((ConflictSet::Impl *)cs, oldestVersion);
((ConflictSet::Impl *)cs)->setOldestVersion(oldestVersion);
}
__attribute__((__visibility__("default"))) void *
ConflictSet_create(int64_t oldestVersion) {
return internal_create(oldestVersion);
return new (safe_malloc(sizeof(ConflictSet::Impl)))
ConflictSet::Impl{oldestVersion};
}
__attribute__((__visibility__("default"))) void ConflictSet_destroy(void *cs) {
internal_destroy((ConflictSet::Impl *)cs);
}
__attribute__((__visibility__("default"))) int64_t
ConflictSet_getBytes(void *cs) {
return internal_getBytes((ConflictSet::Impl *)cs);
using Impl = ConflictSet::Impl;
((Impl *)cs)->~Impl();
free(cs);
}
}
#if SHOW_MEMORY
struct __attribute__((visibility("default"))) PeakPrinter {
~PeakPrinter() {
printf("--- skip_list ---\n");
printf("malloc bytes: %g\n", double(mallocBytes));
printf("Peak malloc bytes: %g\n", double(peakMallocBytes));
}
} peakPrinter;
#endif

View File

@@ -3,68 +3,18 @@
#include <fstream>
#include <sstream>
#ifndef PERF_TEST
#define PERF_TEST 0
#endif
int main(int argc, char **argv) {
for (int i = 1; i < argc; ++i) {
printf("Running: %s\n", argv[i]);
std::ifstream t(argv[i], std::ios::binary);
std::stringstream buffer;
buffer << t.rdbuf();
auto str = buffer.str();
Arbitrary arbitrary({(const uint8_t *)str.data(), str.size()});
TestDriver<ConflictSet, !PERF_TEST> driver1{arbitrary};
TestDriver<ConflictSet, !PERF_TEST> driver2{arbitrary};
bool done1 = false;
bool done2 = false;
for (;;) {
if (!done1) {
done1 = driver1.next();
if (!driver1.ok) {
TestDriver<ConflictSet> driver{(const uint8_t *)str.data(), str.size()};
while (!driver.next())
;
if (!driver.ok) {
abort();
}
}
if (!done2) {
done2 = driver2.next();
if (!driver2.ok) {
abort();
}
}
if (done1 && done2) {
break;
}
}
{
ConflictSet::MetricsV1 *metrics;
int metricsCount;
driver1.cs.getMetricsV1(&metrics, &metricsCount);
printf("#################### METRICS for ConflictSet 1 for %s "
"####################\n",
argv[i]);
for (int i = 0; i < metricsCount; ++i) {
printf("# HELP %s %s\n", metrics[i].name, metrics[i].help);
printf("# TYPE %s %s\n", metrics[i].name,
metrics[i].type == metrics[i].Counter ? "counter" : "gauge");
printf("%s %g\n", metrics[i].name, metrics[i].getValue());
}
puts("");
}
{
ConflictSet::MetricsV1 *metrics;
int metricsCount;
driver2.cs.getMetricsV1(&metrics, &metricsCount);
printf("#################### METRICS for ConflictSet 2 for %s "
"####################\n",
argv[i]);
for (int i = 0; i < metricsCount; ++i) {
printf("# HELP %s %s\n", metrics[i].name, metrics[i].help);
printf("# TYPE %s %s\n", metrics[i].name,
metrics[i].type == metrics[i].Counter ? "counter" : "gauge");
printf("%s %g\n", metrics[i].name, metrics[i].getValue());
}
puts("");
}
}
}

View File

@@ -1,10 +0,0 @@
__aarch64_ldadd8_relax
__getauxval@GLIBC_2.17
__stack_chk_fail@GLIBC_2.17
__stack_chk_guard@GLIBC_2.17
abort@GLIBC_2.17
free@GLIBC_2.17
malloc@GLIBC_2.17
memcpy@GLIBC_2.17
memmove@GLIBC_2.17
memset@GLIBC_2.17

View File

@@ -1,19 +0,0 @@
_ConflictSet_addWrites
_ConflictSet_check
_ConflictSet_create
_ConflictSet_destroy
_ConflictSet_getBytes
_ConflictSet_setOldestVersion
__ZN8weaselab11ConflictSet16setOldestVersionEx
__ZN8weaselab11ConflictSet9addWritesEPKNS0_10WriteRangeEix
__ZN8weaselab11ConflictSetC1EOS0_
__ZN8weaselab11ConflictSetC1Ex
__ZN8weaselab11ConflictSetC2EOS0_
__ZN8weaselab11ConflictSetC2Ex
__ZN8weaselab11ConflictSetD1Ev
__ZN8weaselab11ConflictSetD2Ev
__ZN8weaselab11ConflictSetaSEOS0_
__ZNK8weaselab11ConflictSet12getMetricsV1EPPNS0_9MetricsV1EPi
__ZNK8weaselab11ConflictSet5checkEPKNS0_9ReadRangeEPNS0_6ResultEi
__ZNK8weaselab11ConflictSet8getBytesEv
__ZNK8weaselab11ConflictSet9MetricsV18getValueEv

View File

@@ -1,10 +0,0 @@
___stack_chk_fail
___stack_chk_guard
__tlv_bootstrap
_abort
_bzero
_free
_malloc
_memcpy
_memmove
dyld_stub_binder

View File

@@ -1,138 +0,0 @@
import ctypes
import enum
import os
from typing import Optional
class _Key(ctypes.Structure):
_fields_ = [("p", ctypes.POINTER(ctypes.c_ubyte)), ("len", ctypes.c_int)]
class ReadRange(ctypes.Structure):
_fields_ = [
("begin", _Key),
("end", _Key),
("readVersion", ctypes.c_int64),
]
class WriteRange(ctypes.Structure):
_fields_ = [("begin", _Key), ("end", _Key)]
class Result(enum.Enum):
COMMIT = 0
CONFLICT = 1
TOO_OLD = 2
def write(begin: bytes, end: Optional[bytes] = None) -> WriteRange:
b = (ctypes.c_ubyte * len(begin)).from_buffer(bytearray(begin))
if end is None:
e = (ctypes.c_ubyte * 0)()
else:
e = (ctypes.c_ubyte * len(end)).from_buffer(bytearray(end))
return WriteRange(_Key(b, len(b)), _Key(e, len(e)))
def read(version: int, begin: bytes, end: Optional[bytes] = None) -> ReadRange:
b = (ctypes.c_ubyte * len(begin)).from_buffer(bytearray(begin))
if end is None:
e = (ctypes.c_ubyte * 0)()
else:
e = (ctypes.c_ubyte * len(end)).from_buffer(bytearray(end))
return ReadRange(_Key(b, len(b)), _Key(e, len(e)), version)
class ConflictSet:
def __init__(
self,
version: int = 0,
build_dir: Optional[str] = None,
implementation: Optional[str] = None,
) -> None:
self._lib = None
if build_dir is None:
build_dir = os.path.dirname(__file__) + "/build"
if implementation is None:
implementation = "radix_tree"
for f in (
build_dir + "/" + implementation + "/libconflict-set.so.0",
os.path.dirname(__file__)
+ "/build/"
+ implementation
+ "/libconflict-set.0.dylib",
):
try:
self._lib = ctypes.cdll.LoadLibrary(f)
except:
pass
if self._lib is None:
import sys
print(
"Could not find libconflict-set implementation " + implementation,
file=sys.stderr,
)
sys.exit(1)
self._lib.ConflictSet_create.argtypes = (ctypes.c_int64,)
self._lib.ConflictSet_create.restype = ctypes.c_void_p
self._lib.ConflictSet_check.argtypes = (
ctypes.c_void_p,
ctypes.POINTER(ReadRange),
ctypes.POINTER(ctypes.c_int),
ctypes.c_int,
)
self._lib.ConflictSet_addWrites.argtypes = (
ctypes.c_void_p,
ctypes.POINTER(WriteRange),
ctypes.c_int,
ctypes.c_int64,
)
self._lib.ConflictSet_setOldestVersion.argtypes = (
ctypes.c_void_p,
ctypes.c_int64,
)
self._lib.ConflictSet_destroy.argtypes = (ctypes.c_void_p,)
self._lib.ConflictSet_getBytes.argtypes = (ctypes.c_void_p,)
self._lib.ConflictSet_getBytes.restype = ctypes.c_int64
self.p = self._lib.ConflictSet_create(version)
def addWrites(self, version: int, *writes: WriteRange):
self._lib.ConflictSet_addWrites(
self.p, (WriteRange * len(writes))(*writes), len(writes), version
)
def check(self, *reads: ReadRange) -> list[Result]:
r = (ctypes.c_int * len(reads))()
self._lib.ConflictSet_check(
self.p, (ReadRange * len(reads))(*reads), r, len(reads)
)
return [Result(x) for x in r]
def setOldestVersion(self, version: int) -> None:
self._lib.ConflictSet_setOldestVersion(self.p, version)
def getBytes(self) -> int:
return self._lib.ConflictSet_getBytes(self.p)
def __enter__(self):
return self
def close(self) -> None:
if self.p is not None:
self._lib.ConflictSet_destroy(self.p)
self.p = None
def __exit__(self, exception_type, exception_value, exception_traceback):
self.close()

View File

@@ -7,7 +7,6 @@ int main(void) {
ConflictSet_WriteRange w;
ConflictSet_Result result;
ConflictSet_ReadRange r;
int64_t bytes;
w.begin.p = (const uint8_t *)"0000";
w.begin.len = 4;
w.end.len = 0;
@@ -18,8 +17,6 @@ int main(void) {
r.readVersion = 0;
ConflictSet_check(cs, &r, &result, 1);
assert(result == ConflictSet_Conflict);
bytes = ConflictSet_getBytes(cs);
assert(bytes > 0);
ConflictSet_destroy(cs);
return 0;
}

View File

@@ -1,9 +1,6 @@
#include "ConflictSet.h"
#include <cassert>
#include <cstdio>
using namespace weaselab;
int main(void) {
ConflictSet cs(0);
@@ -20,16 +17,4 @@ int main(void) {
r.readVersion = 0;
cs.check(&r, &result, 1);
assert(result == ConflictSet::Conflict);
int64_t bytes = cs.getBytes();
assert(bytes > 0);
ConflictSet::MetricsV1 *metrics;
int metricsCount;
cs.getMetricsV1(&metrics, &metricsCount);
for (int i = 0; i < metricsCount; ++i) {
printf("# HELP %s %s\n", metrics[i].name, metrics[i].help);
printf("# TYPE %s %s\n", metrics[i].name,
metrics[i].type == metrics[i].Counter ? "counter" : "gauge");
printf("%s %g\n", metrics[i].name, metrics[i].getValue());
}
}

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Some files were not shown because too many files have changed in this diff Show More