ci: use full ASAN

.github/workflows/ci-workflow.yml

@@ -94,16 +94,10 @@ jobs:
         run: npm install --include=dev
         working-directory: tools/nodejs_api
 
-      - name: Build debug
-        run: CC=gcc CXX=g++ make alldebug NUM_THREADS=32
-
-      - name: Run test with ASan
-        run: ctest --output-on-failure -j 10
+      - name: Test with ASAN
+        run: CC=gcc CXX=g++ make test ASAN=1 NUM_THREADS=32
         env:
-          LD_PRELOAD: "/usr/lib/x86_64-linux-gnu/libasan.so.6"
           ASAN_OPTIONS: "detect_leaks=1:log_path=/tmp/asan.log"
-        working-directory: ./build/debug/test
-        continue-on-error: true
 
       - name: Report ASan log
         run: |

CMakeLists.txt

@@ -245,7 +245,6 @@ include_directories(third_party/pyparse)
 include_directories(third_party/utf8proc/include)
 include_directories(third_party/pybind11/include)
 include_directories(third_party/re2/include)
-include_directories(third_party/concurrentqueue)
 include_directories(third_party/serd/include)
 include_directories(third_party/miniparquet/src)
 include_directories(third_party/fast_float/include)

Makefile

@@ -113,24 +113,20 @@ clang-tidy-ci:
 	$(call mkdirp,build/release) && cd build/release && \
 	cmake $(GENERATOR) $(FORCE_COLOR) $(SANITIZER_FLAG) -DCMAKE_EXPORT_COMPILE_COMMANDS=1 ../..
 
-pytest:
-	$(MAKE) release
+pytest: release
 	cd $(ROOT_DIR)/tools/python_api/test && \
 	python3 -m pytest -v test_main.py
 
-nodejstest:
-	$(MAKE) nodejs
+nodejstest: nodejs
 	cd $(ROOT_DIR)/tools/nodejs_api/ && \
 	npm test
 
-javatest: arrow
+javatest: arrow java
 ifeq ($(OS),Windows_NT)
-	$(MAKE) java
 	$(call mkdirp,tools/java_api/build/test)  && cd tools/java_api/ && \
 	javac -d build/test -cp ".;build/kuzu_java.jar;third_party/junit-platform-console-standalone-1.9.3.jar"  -sourcepath src/test/java/com/kuzudb/test/*.java && \
 	java -jar third_party/junit-platform-console-standalone-1.9.3.jar -cp ".;build/kuzu_java.jar;build/test/" --scan-classpath --include-package=com.kuzudb.java_test --details=verbose
 else
-	$(MAKE) java
 	$(call mkdirp,tools/java_api/build/test)  && cd tools/java_api/ && \
 	javac -d build/test -cp ".:build/kuzu_java.jar:third_party/junit-platform-console-standalone-1.9.3.jar"  -sourcepath src/test/java/com/kuzudb/test/*.java && \
 	java -jar third_party/junit-platform-console-standalone-1.9.3.jar -cp ".:build/kuzu_java.jar:build/test/" --scan-classpath --include-package=com.kuzudb.java_test --details=verbose

src/include/storage/buffer_manager/buffer_manager.h

@@ -1,14 +1,11 @@
 #pragma once
 
 #include <functional>
+#include <queue>
 #include <vector>
 
-#include "concurrentqueue.h"
 #include "storage/buffer_manager/bm_file_handle.h"
-
-namespace spdlog {
-class logger;
-}
+#include "storage/buffer_manager/locked_queue.h"
 
 namespace kuzu {
 namespace storage {
@@ -35,13 +32,16 @@ struct EvictionCandidate {
     PageState* pageState = nullptr;
     // The version of the corresponding page at the time the candidate is enqueued.
     uint64_t pageVersion = -1u;
+
+    inline bool operator==(const EvictionCandidate& other) {
+        return fileHandle == other.fileHandle && pageIdx == other.pageIdx &&
+               pageState == other.pageState && pageVersion == other.pageVersion;
+    }
 };
 
 class EvictionQueue {
 public:
-    explicit EvictionQueue(uint64_t capacity) : capacity{capacity} {
-        queue = std::make_unique<moodycamel::ConcurrentQueue<EvictionCandidate>>(capacity);
-    }
+    EvictionQueue() { queue = std::make_unique<LockedQueue<EvictionCandidate>>(); }
 
     inline void enqueue(EvictionCandidate& candidate) {
         std::shared_lock sLck{mtx};
@@ -63,8 +63,7 @@ class EvictionQueue {
 
 private:
     std::shared_mutex mtx;
-    uint64_t capacity;
-    std::unique_ptr<moodycamel::ConcurrentQueue<EvictionCandidate>> queue;
+    std::unique_ptr<LockedQueue<EvictionCandidate>> queue;
 };
 
 /**
@@ -193,7 +192,7 @@ class BufferManager {
     inline common::frame_group_idx_t addNewFrameGroup(common::PageSizeClass pageSizeClass) {
         return vmRegions[pageSizeClass]->addNewFrameGroup();
     }
-    inline void clearEvictionQueue() { evictionQueue = std::make_unique<EvictionQueue>(0); }
+    inline void clearEvictionQueue() { evictionQueue = std::make_unique<EvictionQueue>(); }
 
 private:
     bool claimAFrame(
@@ -221,7 +220,6 @@ class BufferManager {
     }
 
 private:
-    std::shared_ptr<spdlog::logger> logger;
     std::atomic<uint64_t> usedMemory;
     std::atomic<uint64_t> bufferPoolSize;
     std::atomic<uint64_t> numEvictionQueueInsertions;

src/include/storage/buffer_manager/locked_queue.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <mutex>
+#include <queue>
+
+template<typename T>
+class LockedQueue {
+    std::mutex mtx;
+    std::queue<T> q;
+
+public:
+    LockedQueue() {}
+
+    void enqueue(T x) {
+        std::scoped_lock lck{mtx};
+        q.push(std::move(x));
+    }
+
+    bool try_dequeue(T& x) {
+        std::scoped_lock lck{mtx};
+        if (!q.empty()) {
+            x = q.front();
+            q.pop();
+            return true;
+        }
+        return false;
+    }
+
+    size_t size() {
+        std::scoped_lock lck{mtx};
+        return q.size();
+    }
+};

src/storage/buffer_manager/buffer_manager.cpp

@@ -53,7 +53,7 @@ void EvictionQueue::removeCandidatesForFile(kuzu::storage::BMFileHandle& fileHan
     std::unique_lock xLck{mtx};
     EvictionCandidate candidate;
     uint64_t loopedCandidateIdx = 0;
-    auto numCandidatesInQueue = queue->size_approx();
+    auto numCandidatesInQueue = queue->size();
     while (loopedCandidateIdx < numCandidatesInQueue && queue->try_dequeue(candidate)) {
         if (candidate.fileHandle != &fileHandle) {
             queue->enqueue(candidate);
@@ -63,18 +63,15 @@ void EvictionQueue::removeCandidatesForFile(kuzu::storage::BMFileHandle& fileHan
 }
 
 BufferManager::BufferManager(uint64_t bufferPoolSize)
-    : logger{LoggerUtils::getLogger(LoggerConstants::LoggerEnum::BUFFER_MANAGER)}, usedMemory{0},
-      bufferPoolSize{bufferPoolSize}, numEvictionQueueInsertions{0} {
-    logger->info("Done initializing buffer manager.");
+    : usedMemory{0}, bufferPoolSize{bufferPoolSize}, numEvictionQueueInsertions{0} {
     if (bufferPoolSize < BufferPoolConstants::PAGE_4KB_SIZE) {
         throw BufferManagerException("The given buffer pool size should be at least 4KB.");
     }
     vmRegions.resize(2);
     vmRegions[0] = std::make_unique<VMRegion>(
         PageSizeClass::PAGE_4KB, BufferPoolConstants::DEFAULT_VM_REGION_MAX_SIZE);
     vmRegions[1] = std::make_unique<VMRegion>(PageSizeClass::PAGE_256KB, bufferPoolSize);
-    evictionQueue =
-        std::make_unique<EvictionQueue>(bufferPoolSize / BufferPoolConstants::PAGE_4KB_SIZE);
+    evictionQueue = std::make_unique<EvictionQueue>();
 }
 
 // Important Note: Pin returns a raw pointer to the frame. This is potentially very dangerous and