Const iterator works.

Reviewers: florijan, mislav.bradac

Reviewed By: mislav.bradac

Subscribers: buda, pullbot

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

src/data_structures/concurrent/common.hpp

@@ -75,5 +75,5 @@ class AccessorBase {
   size_t size() const { return accessor.size(); }
 
  protected:
-  typename list::Accessor accessor;
+  typename list::template Accessor<SkipList<T>> accessor;
 };

src/data_structures/concurrent/skiplist.hpp

@@ -3,6 +3,7 @@
 #include <algorithm>
 #include <cmath>
 #include <memory>
+#include <type_traits>
 
 #include "utils/assert.hpp"
 #include "utils/crtp.hpp"
@@ -451,10 +452,11 @@ class SkipList : private Lockable<lock_t> {
 
   friend class Accessor;
 
+  template <typename TSkipList>
   class Accessor {
     friend class SkipList;
 
-    Accessor(SkipList *skiplist)
+    Accessor(TSkipList *skiplist)
         : skiplist(skiplist), status_(skiplist->gc.CreateNewAccessor()) {
       debug_assert(skiplist != nullptr, "Skiplist is nullptr.");
     }
@@ -473,15 +475,37 @@ class SkipList : private Lockable<lock_t> {
       status_.alive_ = false;
     }
 
-    Iterator begin() { return skiplist->begin(); }
+    // TODO(dgleich): Remove after C++17 compile flag.
+    template <class B>
+    struct negation : std::integral_constant<bool, !bool(B::value)> {};
 
-    ConstIterator begin() const { return skiplist->cbegin(); }
+    template <typename TIsConst = TSkipList>
+    Iterator begin(typename std::enable_if<
+                       negation<std::is_const<TIsConst>>::value>::type * = 0) {
+      return skiplist->begin();
+    }
+
+    template <typename TIsConst = TSkipList>
+    ConstIterator begin(
+        typename std::enable_if<std::is_const<TIsConst>::value>::type * =
+            0) const {
+      return skiplist->cbegin();
+    }
 
     ConstIterator cbegin() const { return skiplist->cbegin(); }
 
-    Iterator end() { return skiplist->end(); }
+    template <typename TIsConst = TSkipList>
+    Iterator end(typename std::enable_if<
+                     negation<std::is_const<TIsConst>>::value>::type * = 0) {
+      return skiplist->end();
+    }
 
-    ConstIterator end() const { return skiplist->cend(); }
+    template <typename TIsConst = TSkipList>
+    ConstIterator end(
+        typename std::enable_if<std::is_const<TIsConst>::value>::type * =
+            0) const {
+      return skiplist->cend();
+    }
 
     ConstIterator cend() const { return skiplist->cend(); }
 
@@ -517,25 +541,25 @@ class SkipList : private Lockable<lock_t> {
       return skiplist->reverse(item);
     }
 
-  /**
-   * Returns an iterator pointing to the element equal to
-   * item, or the first larger element.
-   *
-   * @param item An item that is comparable to skiplist element type.
-   * @tparam TItem item type
-   */
+    /**
+     * Returns an iterator pointing to the element equal to
+     * item, or the first larger element.
+     *
+     * @param item An item that is comparable to skiplist element type.
+     * @tparam TItem item type
+     */
     template <class TItem>
     Iterator find_or_larger(const TItem &item) {
-      return skiplist->find_or_larger<Iterator, TItem>(item);
+      return skiplist->template find_or_larger<Iterator, TItem>(item);
     }
 
-  /**
-   * Returns an iterator pointing to the element equal to
-   * item, or the first larger element.
-   *
-   * @param item An item that is comparable to skiplist element type.
-   * @tparam TItem item type
-   */
+    /**
+     * Returns an iterator pointing to the element equal to
+     * item, or the first larger element.
+     *
+     * @param item An item that is comparable to skiplist element type.
+     * @tparam TItem item type
+     */
     template <class TItem>
     ConstIterator find_or_larger(const TItem &item) const {
       return static_cast<const SkipList &>(*skiplist)
@@ -621,7 +645,7 @@ class SkipList : private Lockable<lock_t> {
       // now we need to estimate the count of elements equal to item
       // we'll do that by looking for the first element that is greater
       // than item, and counting how far we have to look
-      
+
       // first find the rightmost (highest) succ that has value == item
       int count_level = 0;
       for (int i = position_level; i >= 0; i--)
@@ -669,14 +693,16 @@ class SkipList : private Lockable<lock_t> {
     }
 
    private:
-    SkipList *skiplist;
+    TSkipList *skiplist;
     Node *preds[H], *succs[H];
     typename SkipListGC<Node>::AccessorStatus &status_;
   };
 
-  Accessor access() { return Accessor(this); }
+  Accessor<SkipList> access() { return Accessor<SkipList>(this); }
 
-  const Accessor access() const { return Accessor(this); }
+  Accessor<const SkipList> caccess() const {
+    return Accessor<const SkipList>(this);
+  }
 
  private:
   using guard_t = std::unique_lock<lock_t>;
@@ -1022,5 +1048,5 @@ class SkipList : private Lockable<lock_t> {
    */
   std::atomic<size_t> count{0};
   Node *header;
-  SkipListGC<Node> gc;
+  mutable SkipListGC<Node> gc;
 };

src/database/indexes/index_common.hpp

@@ -17,7 +17,7 @@ namespace IndexUtils {
  * iterating, i.e. it allows us to iterate over the suffix of some skiplist
  * hence the name SkipListSuffix.
  */
-template <class TIterator, class TValue>
+template <class TIterator, class TValue, typename TAccessor>
 class SkipListSuffix {
  public:
   class Iterator {
@@ -39,15 +39,16 @@ class SkipListSuffix {
     TIterator current_;
   };
 
-  SkipListSuffix(const TIterator begin,
-                 typename SkipList<TValue>::Accessor &&accessor)
+  SkipListSuffix(
+      const TIterator begin,
+      typename SkipList<TValue>::template Accessor<TAccessor> &&accessor)
       : begin_(begin), accessor_(std::move(accessor)) {}
 
   Iterator begin() const { return Iterator(begin_); }
   Iterator end() { return Iterator(accessor_.end()); }
 
   TIterator begin_;
-  typename SkipList<TValue>::Accessor accessor_;
+  typename SkipList<TValue>::template Accessor<TAccessor> accessor_;
 };
 
 /**
@@ -67,21 +68,24 @@ class SkipListSuffix {
  *    ignored).
  * @Tparam TIndexEntry - index entry inside skiplist
  * @Tparam TRecord - type of record under index (edge/vertex usually.)
+ * @Tparam TAccessor - type of accessor to use (const skiplist/non const
+ * skiplist).
  * @return iterable collection of distinct vlist records<TRecord> for which
  * exists function evaluates as true
  */
-template <class TIterator, class TIndexEntry, class TRecord>
+template <class TIterator, class TIndexEntry, class TRecord, typename TAccessor>
 static auto GetVlists(
-    typename SkipList<TIndexEntry>::Accessor &&skiplist_accessor,
+    typename SkipList<TIndexEntry>::template Accessor<TAccessor>
+        &&skiplist_accessor,
     TIterator begin,
     const std::function<bool(const TIndexEntry &entry)> predicate,
     const tx::Transaction &t,
     const std::function<bool(const TIndexEntry &, const TRecord *)> exists,
     bool current_state = false) {
   TIndexEntry *prev = nullptr;
-  auto range =
-      iter::takewhile(predicate, SkipListSuffix<TIterator, TIndexEntry>(
-                                     begin, std::move(skiplist_accessor)));
+  auto range = iter::takewhile(
+      predicate, SkipListSuffix<TIterator, TIndexEntry, TAccessor>(
+                     begin, std::move(skiplist_accessor)));
   auto filtered = iter::filter(
       [&t, exists, prev, current_state](TIndexEntry &entry) mutable {
         // Check if the current entry could offer new possible return value
@@ -176,4 +180,4 @@ static void Refresh(
     }
   }
 }
-};  // IndexUtils
+};  // namespace IndexUtils

src/database/indexes/label_property_index.hpp

@@ -86,7 +86,7 @@ class LabelPropertyIndex {
    * be populated with, and can be used from this moment forward without missing
    * any records.
    * @param key - index which finished being populated.
-  */
+   */
   void IndexFinishedBuilding(const Key &key) {
     ready_for_use_.access().insert(key);
   }
@@ -172,7 +172,7 @@ class LabelPropertyIndex {
     auto access = GetKeyStorage(key)->access();
     auto begin = access.begin();
     return IndexUtils::GetVlists<typename SkipList<IndexEntry>::Iterator,
-                                 IndexEntry, Vertex>(
+                                 IndexEntry, Vertex, SkipList<IndexEntry>>(
         std::move(access), begin, [](const IndexEntry &) { return true; }, t,
         [key](const IndexEntry &entry, const Vertex *const vertex) {
           return LabelPropertyIndex::Exists(key, entry.value_, vertex);
@@ -269,10 +269,9 @@ class LabelPropertyIndex {
     auto access = GetKeyStorage(key)->access();
 
     // create the iterator startpoint based on the lower bound
-    auto start_iter = lower
-                          ? access.find_or_larger(make_index_bound(
-                                lower, lower.value().IsInclusive()))
-                          : access.begin();
+    auto start_iter = lower ? access.find_or_larger(make_index_bound(
+                                  lower, lower.value().IsInclusive()))
+                            : access.begin();
 
     // a function that defines if an entry staisfies the filtering predicate.
     // since we already handled the lower bound, we only need to deal with the
@@ -394,8 +393,7 @@ class LabelPropertyIndex {
    */
   std::vector<Key> Keys() {
     std::vector<Key> keys;
-    for (auto &kv : indices_.access())
-      keys.push_back(kv.first);
+    for (auto &kv : indices_.access()) keys.push_back(kv.first);
     return keys;
   }
 

tests/unit/skiplist_access.cpp

@@ -0,0 +1,47 @@
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include <vector>
+
+#include "data_structures/concurrent/skiplist.hpp"
+
+TEST(SkipList, Access) {
+  SkipList<int> input;
+  {
+    auto accessor = input.access();
+    accessor.insert(1);
+    accessor.insert(2);
+    accessor.insert(3);
+  }
+
+  auto accessor = input.access();
+  std::vector<int> results;
+  for (auto it = accessor.begin(); it != accessor.end(); ++it)
+    results.push_back(*it);
+
+  EXPECT_THAT(results, testing::ElementsAre(1, 2, 3));
+}
+
+TEST(SkipList, ConstAccess) {
+  SkipList<int> input;
+  {
+    auto accessor = input.access();
+    accessor.insert(1);
+    accessor.insert(2);
+    accessor.insert(3);
+  }
+
+  const SkipList<int> &skiplist = input;
+  auto accessor = skiplist.caccess();
+
+  std::vector<int> results;
+  for (auto it = accessor.begin(); it != accessor.end(); ++it)
+    results.push_back(*it);
+
+  EXPECT_THAT(results, testing::ElementsAre(1, 2, 3));
+}
+
+int main(int argc, char **argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

tests/unit/skiplist_suffix.cpp

@@ -17,8 +17,9 @@ int Count(TIterable &collection) {
 TEST(SkipListSuffix, EmptyRange) {
   SkipList<int> V;
   auto access = V.access();
-  auto r1 = IndexUtils::SkipListSuffix<typename SkipList<int>::Iterator, int>(
-      access.begin(), std::move(access));
+  auto r1 = IndexUtils::SkipListSuffix<typename SkipList<int>::Iterator, int,
+                                       SkipList<int>>(access.begin(),
+                                                      std::move(access));
   EXPECT_EQ(Count(r1), 0);
 }
 
@@ -28,8 +29,9 @@ TEST(SkipListSuffix, NonEmptyRange) {
   access.insert(1);
   access.insert(5);
   access.insert(3);
-  auto r1 = IndexUtils::SkipListSuffix<typename SkipList<int>::Iterator, int>(
-      access.begin(), std::move(access));
+  auto r1 = IndexUtils::SkipListSuffix<typename SkipList<int>::Iterator, int,
+                                       SkipList<int>>(access.begin(),
+                                                      std::move(access));
   EXPECT_EQ(Count(r1), 3);
   auto iter = r1.begin();
   EXPECT_EQ(*iter, 1);