Newer bits on head chunk in DynamicBitset

Reviewers: florijan, buda

Reviewed By: florijan

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D813

src/data_structures/bitset/dynamic_bitset.hpp

@@ -6,208 +6,185 @@
 #include "threading/sync/spinlock.hpp"
 #include "utils/assert.hpp"
 
-/** A sequentially ordered non-unique
+/**
- * lock-free concurrent collection of bits.
+ * A sequentially ordered non-unique lock-free concurrent collection of bits.
  *
- * Grows dynamically to accomodate the maximum
+ * Grows dynamically to accomodate the maximum set-bit position. Does not
- * set-bit position. Does not dynamically
+ * dynamically decrease in size.
- * decrease in size.
  *
- * Bits can be set, retrieved and cleared
+ * Bits can be set, retrieved and cleared in groups. Note that all group
- * in groups. Note that all group operations
+ * operations fail if the group spans multiple basic storage units. For example,
- * fail if the group spans multiple basic
+ * if basic storage is in 8 bits, calling at(5, 2) is legal, but calling at(7,
- * storage units. For example, if basic storage
+ * 2) is not.
- * is in 8 bits, calling at(5, 2) is legal,
- * but calling at(7, 2) is not.
  *
- * Organizes bits into chunks. Finding the
+ * @tparam block_t - Basic storage type. Must have lock-free atomic. Should
- * right chunk has has O(n) lookup performance,
+ * probably always be uint_8.
- * so use large chunks for better speed in large
- * bitsets. At the same time larger chunks
- * mean coarser-grained memory allocation.
- *
- * @tparam block_t - Basic storage type. Must have
- * lock-free atomic. Should probably always be uint_8.
  * @tparam chunk_size - number of bits in one chunk
  */
 template <class block_t = uint8_t, size_t chunk_size = 32768>
-class DynamicBitset : Lockable<SpinLock> {
+class DynamicBitset {
-  // basic storage unit
+  // Basic storage unit.
   struct Block {
     Block() = default;
 
-    // the number of bits in one Block
+    Block(const Block &) = delete;
-    static constexpr size_t size = sizeof(block_t) * 8;
+    Block(Block &&) = delete;
+    Block &operator=(const Block &) = delete;
+    Block &operator=(Block &&) = delete;
+
+    // The number of bits in one Block.
+    static constexpr size_t kSize = sizeof(block_t) * 8;
 
     block_t at(size_t k, size_t n) const {
-      debug_assert(k + n - 1 < size, "Invalid index.");
+      debug_assert(k + n - 1 < kSize, "Invalid index.");
-      return (block.load() >> k) & bitmask(n);
+      return (block_.load() >> k) & bitmask(n);
     }
 
     void set(size_t k, size_t n) {
-      debug_assert(k + n - 1 < size, "Invalid index.");
+      debug_assert(k + n - 1 < kSize, "Invalid index.");
-      block.fetch_or(bitmask(n) << k);
+      block_.fetch_or(bitmask(n) << k);
     }
 
     void clear(size_t k, size_t n) {
-      debug_assert(k + n - 1 < size, "Invalid index.");
+      debug_assert(k + n - 1 < kSize, "Invalid index.");
-      block.fetch_and(~(bitmask(n) << k));
+      block_.fetch_and(~(bitmask(n) << k));
     }
 
    private:
-    std::atomic<block_t> block{0};
+    std::atomic<block_t> block_{0};
 
     constexpr block_t bitmask(size_t group_size) const {
-      return (block_t)(-1) >> (size - group_size);
+      return (block_t)(-1) >> (kSize - group_size);
     }
   };
 
   struct Chunk {
-    Chunk() : next(nullptr) {
+    Chunk(Chunk *next, int64_t chunk_id) : next_(next), chunk_id_(chunk_id) {
-      static_assert(chunk_size % sizeof(block_t) == 0,
+      static_assert(chunk_size % Block::kSize == 0,
                     "chunk size not divisible by block size");
     }
 
-    Chunk(Chunk &) = delete;
+    Chunk(const Chunk &) = delete;
     Chunk(Chunk &&) = delete;
+    Chunk &operator=(const Chunk &) = delete;
+    Chunk &operator=(Chunk &&) = delete;
 
-    // the number of bits in one chunk
+    // The number of bits in one chunk.
-    static constexpr size_t size = chunk_size * Block::size;
+    static constexpr size_t kSize = chunk_size;
-    // the number of blocks in one chunk
+    // The number of blocks_ in one chunk.
-    static constexpr size_t n_blocks = chunk_size / sizeof(block_t);
+    static constexpr size_t kNumBlocks = chunk_size / Block::kSize;
 
     block_t at(size_t k, size_t n) const {
-      return blocks[k / Block::size].at(k % Block::size, n);
+      return blocks_[k / Block::kSize].at(k % Block::kSize, n);
     }
 
     void set(size_t k, size_t n) {
-      blocks[k / Block::size].set(k % Block::size, n);
+      blocks_[k / Block::kSize].set(k % Block::kSize, n);
     }
 
     void clear(size_t k, size_t n) {
-      blocks[k / Block::size].clear(k % Block::size, n);
+      blocks_[k / Block::kSize].clear(k % Block::kSize, n);
     }
 
-    Block blocks[n_blocks];
+    // Range of the bits stored in this chunk is [low, high>.
-    std::atomic<Chunk *> next;
+    int64_t low() const { return chunk_id_ * kSize; }
+
+    int64_t high() const { return (chunk_id_ + 1) * kSize; }
+
+    Block blocks_[kNumBlocks];
+    std::atomic<Chunk *> next_;
+    const int64_t chunk_id_;
   };
 
  public:
-  DynamicBitset(){};
+  DynamicBitset() {}
 
-  // can't move nor copy a DynamicBitset because of atomic
+  // Can't move nor copy a DynamicBitset because of atomic head and locking.
-  // head and locking
   DynamicBitset(const DynamicBitset &) = delete;
   DynamicBitset(DynamicBitset &&) = delete;
   DynamicBitset &operator=(const DynamicBitset &) = delete;
   DynamicBitset &operator=(DynamicBitset &&) = delete;
 
   ~DynamicBitset() {
-    auto now = head.load();
+    auto now = head_.load();
     while (now != nullptr) {
-      auto next = now->next.load();
+      auto next = now->next_.load();
       delete now;
       now = next;
     }
   }
 
-  /** Gets the block of bit starting at bit k
+  /**
-   * and containing the the following n bits.
+   * Gets the block of bit starting at bit k and containing the the following n
-   * The bit index with k in this bitset is
+   * bits. The bit index with k in this bitset is zeroth bit in the returned
-   * zeroth bit in the returned value.
+   * value.
    */
   block_t at(size_t k, size_t n) const {
-    if (k >= Chunk::size * chunk_count) return 0;
+    if (k >= head_.load()->high()) return 0;
 
-    auto &chunk = find_chunk(k);
+    const auto &chunk = FindChunk(k);
     return chunk.at(k, n);
   }
 
   /** Returns k-th bit's value. */
   bool at(size_t k) const { return at(k, 1); }
 
-  /** Set all the bits in the group of size `n`, starting from
+  /**
-   * bit `k`.
+   * Set all the bits in the group of size `n`, starting from bit `k`.
    */
   void set(size_t k, size_t n = 1) {
-    auto &chunk = find_or_create_chunk(k);
+    auto &chunk = FindOrCreateChunk(k);
     return chunk.set(k, n);
   }
 
-  /** Clears all the bits in the group of size `n`, starting
+  /**
-   * from bit `k`.
+   * Clears all the bits in the group of size `n`, starting from bit `k`.
-   * */
+   */
   void clear(size_t k, size_t n = 1) {
-    // if desired bit is out of bounds, it's already clear
+    // If desired bit is out of bounds, it's already clear.
-    if (k >= Chunk::size * chunk_count) return;
+    if (k >= head_.load()->high()) return;
 
-    auto &chunk = find_or_create_chunk(k);
+    auto &chunk = FindOrCreateChunk(k);
     return chunk.clear(k, n);
   }
 
  private:
-  // finds the chunk to which k-th bit belong. if k is
+  // Finds the chunk to which k-th bit belongs fails if k is out of bounds.
-  // out of bounds, the chunk for it is created
+  const Chunk &FindChunk(size_t &k) const {
-  Chunk &find_or_create_chunk(size_t &k) {
+    debug_assert(k < head_.load()->high(), "Index out of bounds");
-    Chunk *chunk = head.load(), *next = nullptr;
+    Chunk *chunk = head_;
 
-    // while i'm not in the right chunk
+    while (k < chunk->low()) {
-    // (my index is bigger than the size of this chunk)
+      chunk = chunk->next_;
-    while (k >= Chunk::size) {
+      debug_assert(chunk != nullptr, "chunk is nullptr");
-      next = chunk->next.load();
-
-      // if a next chunk exists, switch to it and decrement my
-      // pointer by the size of the current chunk
-      if (next != nullptr) {
-        chunk = next;
-        k -= Chunk::size;
-        continue;
-      }
-
-      // the next chunk does not exist and we need it. take an exclusive
-      // lock to prevent others that also want to create a new chunk
-      // from creating it
-      auto guard = acquire_unique();
-
-      // double-check locking. if the chunk exists now, some other thread
-      // has just created it, continue searching for my chunk
-      if (chunk->next.load() != nullptr) continue;
-
-      chunk->next.store(new Chunk());
-      chunk_count++;
     }
-
+    k -= chunk->low();
-    debug_assert(chunk != nullptr, "Chunk is nullptr.");
     return *chunk;
   }
 
-  // finds the chunk to which k-th bit belongs.
+  /**
-  // fails if k is out of bounds
+   * Finds the chunk to which k-th bit belong. If k is out of bounds, the chunk
-  const Chunk &find_chunk(size_t &k) const {
+   * for it is created.
-    debug_assert(k < chunk_size * Chunk::size, "Index out  of bounds");
+   */
-    Chunk *chunk = head.load(), *next = nullptr;
+  Chunk &FindOrCreateChunk(size_t &k) {
+    Chunk *head = head_;
 
-    // while i'm not in the right chunk
+    while (k >= head->high()) {
-    // (my index is bigger than the size of this chunk)
+      // The next chunk does not exist and we need it, so we will try to create
-    while (k >= Chunk::size) {
+      // it.
-      next = chunk->next.load();
+      Chunk *new_head = new Chunk(head, head->chunk_id_ + 1);
-
+      if (!head_.compare_exchange_strong(head, new_head)) {
-      // if a next chunk exists, switch to it and decrement my
+        // Other thread updated head_ before us, so we need to delete new_head.
-      // pointer by the size of the current chunk
+        head = head_;
-      if (next != nullptr) {
+        delete new_head;
-        chunk = next;
-        k -= Chunk::size;
         continue;
       }
-
-      // the next chunk does not exist, this is illegal state
-      permanent_fail("Out of bounds");
     }
 
-    debug_assert(chunk != nullptr, "Chunk is nullptr.");
+    // Now we are sure chunk exists and we can call find function.
-    return *chunk;
+    // const_cast is used to avoid code duplication.
+    return const_cast<Chunk &>(FindChunk(k));
   }
 
-  std::atomic<Chunk *> head{new Chunk()};
+  std::atomic<Chunk *> head_{new Chunk(nullptr, 0)};
-  std::atomic<int64_t> chunk_count{1};
 };

src/transactions/commit_log.hpp

@@ -52,9 +52,6 @@ class CommitLog {
 
   Info fetch_info(transaction_id_t id) const { return Info{log.at(2 * id, 2)}; }
 
-  // TODO: Searching the log will take more and more time the more and more
-  // transactoins are done. This could be awerted if DynamicBitset is changed
-  // to point to largest chunk instead of the smallest.
   DynamicBitset<uint8_t, 32768> log;
 };
 }

tests/unit/dynamic_bitset.cpp

@@ -1,9 +1,18 @@
+#include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
 #include "data_structures/bitset/dynamic_bitset.hpp"
 
-TEST(DynamicBitset, BasicAtAndSet) {
+namespace {
-  DynamicBitset<> db;
+
+template <typename T>
+class DynamicBitsetTest : public ::testing::Test {};
+typedef ::testing::Types<DynamicBitset<>, DynamicBitset<uint8_t, 8>>
+    DynamicBitsetTypes;
+TYPED_TEST_CASE(DynamicBitsetTest, DynamicBitsetTypes);
+
+TYPED_TEST(DynamicBitsetTest, BasicAtAndSet) {
+  TypeParam db;
 
   EXPECT_EQ(db.at(17, 1), 0);
   EXPECT_EQ(db.at(17), false);
@@ -12,8 +21,8 @@ TEST(DynamicBitset, BasicAtAndSet) {
   EXPECT_EQ(db.at(17), true);
 }
 
-TEST(DynamicBitset, GroupAt) {
+TYPED_TEST(DynamicBitsetTest, GroupAt) {
-  DynamicBitset<> db;
+  TypeParam db;
 
   db.set(0, 1);
   db.set(1, 1);
@@ -27,8 +36,8 @@ TEST(DynamicBitset, GroupAt) {
   EXPECT_EQ(db.at(1, 3), 1 | 4);
 }
 
-TEST(DynamicBitset, GroupSet) {
+TYPED_TEST(DynamicBitsetTest, GroupSet) {
-  DynamicBitset<> db;
+  TypeParam db;
   EXPECT_EQ(db.at(0, 3), 0);
   db.set(1, 2);
   EXPECT_FALSE(db.at(0));
@@ -78,3 +87,4 @@ TEST(DynamicBitset, ConstBitset) {
   dbs.set(17);
   const_accepting(dbs);
 }
+}