BufferPool-v1.0:Functionality correct but performance issues still exist

Author: whzruc

cpp/CMakeLists.txt

@@ -20,7 +20,7 @@ include_directories(${CMAKE_CURRENT_BINARY_DIR})
 
 add_subdirectory(pixels-common)
 add_subdirectory(pixels-core)
-add_subdirectory(pixels-cli)
+#add_subdirectory(pixels-cli)
 add_subdirectory(third-party/googletest)
 add_subdirectory(tests)
 

cpp/Makefile

@@ -57,4 +57,8 @@ debug: deps
 
 release: deps
 	cmake $(GENERATOR) $(FORCE_COLOR) $(EXTENSION_FLAGS) ${CLIENT_FLAGS} -DEXTENSION_STATIC_BUILD=1 -DCMAKE_BUILD_TYPE=Release ${BUILD_FLAGS} -S pixels-duckdb/duckdb -B build/release && \
-	cmake --build build/release --config Release
+	cmake --build build/release --config Release
+
+relWithDebInfo: deps
+	cmake $(GENERATOR) $(FORCE_COLOR) $(EXTENSION_FLAGS) ${CLIENT_FLAGS} -DEXTENSION_STATIC_BUILD=1 -DCMAKE_BUILD_TYPE=RelWithDebInfo ${BUILD_FLAGS} -S pixels-duckdb/duckdb -B build/relWithDebInfo && \
+	cmake --build build/relWithDebInfo --config RelWithDebInfo

cpp/include/PixelsReadGlobalState.hpp

@@ -40,6 +40,9 @@ namespace duckdb
     {
         mutex lock;
 
+        atomic<int> active_threads;  // 活跃线程数
+        atomic<bool> all_done;       // 是否所有线程都已完成
+
         //! The initial reader from the bind phase
         std::shared_ptr <PixelsReader> initialPixelsReader;
 

cpp/pixels-common/CMakeLists.txt

@@ -4,50 +4,21 @@ set(CMAKE_CXX_STANDARD 17)
 
 include(ExternalProject)
 
-set(pixels_common_cxx
-        lib/physical/storage/LocalFS.cpp
-		lib/physical/storage/LocalFSProvider.cpp
-		lib/physical/storage/PhysicalLocalWriter.cpp
-		lib/physical/PhysicalWriterOption.cpp
-		lib/physical/Status.cpp
-        lib/physical/Storage.cpp
-		lib/physical/FilePath.cpp
-        lib/physical/natives/PixelsRandomAccessFile.cpp
-        lib/physical/natives/DirectRandomAccessFile.cpp
-        lib/physical/natives/ByteBuffer.cpp
-        lib/physical/io/PhysicalLocalReader.cpp
-        lib/physical/StorageFactory.cpp
-        lib/physical/Request.cpp
-        lib/physical/RequestBatch.cpp
-        lib/physical/scheduler/NoopScheduler.cpp
-        lib/physical/SchedulerFactory.cpp
-        lib/exception/InvalidArgumentException.cpp
-        lib/utils/Constants.cpp
-        lib/utils/String.cpp
-        include/physical/natives/DirectIoLib.h
-        lib/physical/natives/DirectIoLib.cpp
-        include/utils/ConfigFactory.h
-        lib/utils/ConfigFactory.cpp
-        include/physical/MergedRequest.h
-        include/physical/scheduler/SortMergeScheduler.h
-        lib/physical/scheduler/SortMergeScheduler.cpp
-        lib/MergedRequest.cpp include/profiler/TimeProfiler.h
-        lib/profiler/TimeProfiler.cpp
-        include/profiler/CountProfiler.h
-        lib/profiler/CountProfiler.cpp
-        include/profiler/AbstractProfiler.h
-        include/physical/allocator/Allocator.h
-        include/physical/allocator/OrdinaryAllocator.h
-        lib/physical/allocator/OrdinaryAllocator.cpp
-        include/physical/allocator/BufferPoolAllocator.h
-        lib/physical/allocator/BufferPoolAllocator.cpp
-        include/physical/BufferPool.h
-        lib/physical/BufferPool.cpp
-        include/physical/natives/DirectUringRandomAccessFile.h
-        lib/physical/natives/DirectUringRandomAccessFile.cpp
-		include/utils/ColumnSizeCSVReader.h lib/utils/ColumnSizeCSVReader.cpp
-        include/physical/StorageArrayScheduler.h lib/physical/StorageArrayScheduler.cpp
-		include/physical/natives/ByteOrder.h
+
+file(GLOB_RECURSE pixels_common_cxx
+		"lib/physical/*.cpp"
+		"lib/physical/*.h"
+		"lib/exception/*.cpp"
+		"lib/exception/*.h"
+		"lib/utils/*.cpp"
+		"lib/utils/*.h"
+		"lib/profiler/*.cpp"
+		"lib/profiler/*.h"
+		"include/physical/*.h"
+		"include/profiler/*.h"
+		"include/utils/*.h"
+		"include/physical/BufferPool/*.h"
+		"lib/MergedRequest.cpp"
 )
 
 include_directories(include)

cpp/pixels-common/include/physical/BufferPool.h

@@ -33,41 +33,154 @@
 #include "exception/InvalidArgumentException.h"
 #include "utils/ColumnSizeCSVReader.h"
 #include <map>
-
+#include "physical/BufferPool/Bitmap.h"
+#include "physical/BufferPool/BufferPoolEntry.h"
+#include <mutex>
+#include <thread>
+#include <cstdio>
 // when allocating buffer pool, we use the size of the first pxl file. Consider that
 // the remaining pxl file has larger size than the first file, we allocate some extra
 // size (10MB) to each column.
 // TODO: how to evaluate the maximal pool size
-#define EXTRA_POOL_SIZE 3*1024*1024
+#define EXTRA_POOL_SIZE 10*1024*1024
 
 class DirectUringRandomAccessFile;
 // This class is global class. The variable is shared by each thread
 class BufferPool
 {
 public:
+  // 嵌套子类，用于管理缓冲区池条目及其属性
+  class BufferPoolManagedEntry {
+  public:
+    enum class State{
+      InitizaledNotAllocated,
+      AllocatedAndInUse,
+      UselessButNotFree
+    };
+  private:
+    std::shared_ptr<BufferPoolEntry> bufferPoolEntry;  // 指向缓冲区池条目的智能指针
+    int ring_index;                                   // 环形缓冲区索引
+    size_t current_size;                              // 当前使用大小
+    int offset;                                     // 偏移量
+    State state;
+
+
+
+  public:
+
+    BufferPoolManagedEntry(std::shared_ptr<BufferPoolEntry> entry,
+                          int ringIdx,
+                          size_t currSize,
+                          off_t off)
+        : bufferPoolEntry(std::move(entry)),
+          ring_index(ringIdx),
+          current_size(currSize),
+          offset(off) ,
+          state(State::InitizaledNotAllocated){}
+
+    std::shared_ptr<BufferPoolEntry> getBufferPoolEntry() const {
+      return bufferPoolEntry;
+    }
+
+    int getRingIndex() const {
+      return ring_index;
+    }
+
+    void setRingIndex(int index) {
+      ring_index = index;
+    }
+
+    size_t getCurrentSize() const {
+      return current_size;
+    }
+
+    void setCurrentSize(size_t size) {
+      current_size = size;
+    }
+
+    int getOffset() const {
+      return offset;
+    }
+
+    void setOffset(int off) {
+      offset = off;
+    }
+
+    // 获取当前状态
+    State getStatus() const {
+      return state;
+    }
+
+    // 设置状态
+    void setStatus(State newStatus) {
+      state = newStatus;
+    }
+  };
+
     static void
     Initialize(std::vector <uint32_t> colIds, std::vector <uint64_t> bytes, std::vector <std::string> columnNames);
 
-    static std::shared_ptr <ByteBuffer> GetBuffer(uint32_t colId);
+    static void
+    InitializeBuffers();
+
+    static std::shared_ptr <ByteBuffer> GetBuffer(uint32_t colId,uint64_t byte,std::string columnName);
 
-    static int64_t GetBufferId(uint32_t index);
+    static int64_t GetBufferId();
 
     static void Switch();
 
     static void Reset();
 
+    static std::shared_ptr<BufferPoolEntry> AddNewBuffer(size_t size);
+
+    static int getRingIndex(uint32_t colId);
+
+    static std::shared_ptr<ByteBuffer> AllocateNewBuffer(std::shared_ptr<BufferPoolManagedEntry> currentBufferManagedEntry, uint32_t colId,uint64_t byte,std::string columnName);
+
+    static std::shared_ptr<ByteBuffer> ReusePreviousBuffer(std::shared_ptr<BufferPoolManagedEntry> currentBufferManagedEntry,uint32_t colId,uint64_t byte,std::string columnName);
+
+  static void PrintStats() {
+    // 打印当前线程 ID
+    std::thread::id tid = std::this_thread::get_id();
+
+    printf("线程 %zu -> 全局缓冲区使用: %ld / %ld\n",
+           std::hash<std::thread::id>{}(tid), // 转换成整数便于阅读
+           global_used_size, global_free_size);
+
+    // 线程局部统计
+    printf("线程 %zu -> Buffer0使用: %zu, 缓冲区数量: %d\n",
+           std::hash<std::thread::id>{}(tid),
+           thread_local_used_size[0], thread_local_buffer_count[0]);
+
+    printf("线程 %zu -> Buffer1使用: %zu, 缓冲区数量: %d\n",
+           std::hash<std::thread::id>{}(tid),
+           thread_local_used_size[1], thread_local_buffer_count[1]);
+  }
 private:
     BufferPool() = default;
+  // global
+  static std::mutex bufferPoolMutex;
 
+  // thread local
+  static thread_local bool isInitialized;
+  static thread_local std::vector <std::shared_ptr<BufferPoolEntry>> registeredBuffers[2];
+  static thread_local long global_used_size;
+  static thread_local long global_free_size;
+  static thread_local std::shared_ptr <DirectIoLib> directIoLib;
+  static thread_local int nextRingIndex;
+  static thread_local std::shared_ptr<BufferPoolEntry> nextEmptyBufferPoolEntry[2];
     static thread_local int colCount;
-    static thread_local std::map<uint32_t, uint64_t>
-    nrBytes;
-    static thread_local bool isInitialized;
-    static thread_local std::map<uint32_t, std::shared_ptr < ByteBuffer>>
-    buffers[2];
-    static std::shared_ptr <DirectIoLib> directIoLib;
+
     static thread_local int currBufferIdx;
     static thread_local int nextBufferIdx;
+    static thread_local std::map <uint32_t, std::shared_ptr<ByteBuffer>> buffersAllocated[2];
     friend class DirectUringRandomAccessFile;
+
+    static thread_local std::unordered_map<uint32_t, std::shared_ptr<BufferPoolManagedEntry>> ringBufferMap[2];
+
+
+
+  static thread_local size_t thread_local_used_size[2];  // 线程已使用大小
+  static thread_local int thread_local_buffer_count[2];   // 线程持有的缓冲区数量
 };
 #endif // DUCKDB_BUFFERPOOL_H

cpp/pixels-common/include/physical/BufferPool/Bitmap.h

@@ -0,0 +1,48 @@
+/*
+/* @author whz 
+/* @create 7/30/25.
+*/
+
+#ifndef BITMAP_H
+#define BITMAP_H
+#include <cstddef>
+#include "exception/InvalidArgumentException.h"
+#include <iostream>
+
+class Bitmap {
+public:
+
+  explicit Bitmap(size_t size)
+      : bits((size + 7) / 8, 0), num_bits(size) {}
+
+  void set(size_t index) {
+    if (index >= num_bits) throw InvalidArgumentException("Bitmap::set: index out of range");
+    bits[index / 8] |= (1 << (index % 8));
+  }
+
+  void clear(size_t index) {
+    if (index >= num_bits) throw InvalidArgumentException("Bitmap::clear: index out of range");
+    bits[index / 8] &= ~(1 << (index % 8));
+  }
+
+  bool test(size_t index) const {
+    if (index >= num_bits) throw InvalidArgumentException("Bitmap::test: index out of range");
+    return bits[index / 8] & (1 << (index % 8));
+  }
+
+  size_t size() const { return num_bits; }
+
+  void print() const {
+    for (size_t i = 0; i < num_bits; ++i) {
+      std::cout << (test(i) ? '1' : '0');
+      if ((i + 1) % 8 == 0) std::cout << ' ';
+    }
+    std::cout << '\n';
+  }
+
+private:
+  std::vector<uint8_t> bits;
+  size_t num_bits;
+};
+
+#endif //BITMAP_H

cpp/pixels-common/include/physical/BufferPool/BufferPoolEntry.h

@@ -0,0 +1,46 @@
+/*
+/* @author whz 
+/* @create 7/30/25.
+*/
+
+#ifndef BUFFERPOOLENTRY_H
+#define BUFFERPOOLENTRY_H
+#include <memory>
+#include <physical/natives/ByteBuffer.h>
+#include <physical/natives/DirectIoLib.h>
+#include "physical/BufferPool/Bitmap.h"
+class BufferPoolEntry {
+public:
+explicit BufferPoolEntry(size_t size,int slice_size,std::shared_ptr<DirectIoLib> direct_lib,int offset,int ring_index);
+  size_t getSize() const;
+  std::shared_ptr<Bitmap> getBitmap() const;
+  std::shared_ptr<ByteBuffer> getBuffer() const;
+  bool isFull() const;
+  int getNextFreeIndex() const;
+  int setNextFreeIndex(int index);
+  ~BufferPoolEntry();
+  uint64_t checkCol(uint32_t) const;
+  void addCol(uint32_t colId,uint64_t bytes);
+  bool isInUse() const;
+  void setInUse(bool in_use);
+  int getOffsetInBuffers() const;
+  void setOffsetInBuffers(int offset);
+  bool getIsRegistered() const;
+  void setIsRegistered(bool registered);
+  int getRingIndex() const;
+  void setRingIndex(int ring_index);
+  void reset();
+private:
+size_t size_;
+std::shared_ptr<Bitmap> bitmap_;
+std::shared_ptr<ByteBuffer> buffer_;
+  bool is_full_;
+  int next_free_;
+  std::map<uint32_t, uint64_t> nr_bytes_;
+  bool is_in_use_;
+  int offset_in_buffers_;
+  bool is_registered;
+  int ring_index;
+};
+
+#endif //BUFFERPOOLENTRY_H

cpp/pixels-common/include/physical/Request.h

@@ -34,6 +34,7 @@ class Request
     uint64_t queryId;
     uint64_t start;
     uint64_t length;
+    int ring_index;
 
     Request(uint64_t queryId_, uint64_t start_, uint64_t length_,
             int64_t bufferId = -1);

cpp/pixels-common/include/physical/RequestBatch.h

@@ -42,6 +42,8 @@ class RequestBatch
 
     void add(Request request);
 
+    Request& getRequest(int index);
+
     int getSize();
 
     std::vector <Request> getRequests();

cpp/pixels-common/include/physical/Scheduler.h

@@ -28,6 +28,7 @@
 #include "physical/PhysicalReader.h"
 #include "physical/RequestBatch.h"
 #include "profiler/TimeProfiler.h"
+#include <mutex>
 
 class Scheduler
 {
@@ -46,5 +47,6 @@ class Scheduler
                                                                    RequestBatch batch,
                                                                    std::vector <std::shared_ptr<ByteBuffer>> reuseBuffers,
                                                                    long queryId) = 0;
+    std::mutex mtx;
 };
 #endif //PIXELS_SCHEDULER_H