From 75f7f42d41d8659b43a839bd31570a81e562eb7e Mon Sep 17 00:00:00 2001
From: Adam Kupczyk <akupczyk@redhat.com>
Date: Wed, 4 Oct 2017 18:03:29 -0700
Subject: [PATCH] Added CPU prefetch for skiplist

Summary:
This change causes following changes result of test:
./db_bench --writes 10000000 --benchmarks="fillrandom" --compression_type none
from
fillrandom   :       3.177 micros/op 314804 ops/sec;   34.8 MB/s
to
fillrandom   :       2.777 micros/op 360087 ops/sec;   39.8 MB/s
Closes https://github.com/facebook/rocksdb/pull/2961

Differential Revision: D5977822

Pulled By: yiwu-arbug

fbshipit-source-id: 1ea77707bffa978b1592b0c5d0fe76bfa1930f8d
---
 memtable/inlineskiplist.h | 25 ++++++++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)

diff --git a/memtable/inlineskiplist.h b/memtable/inlineskiplist.h
index 505b73d28..6262fc691 100644
--- a/memtable/inlineskiplist.h
+++ b/memtable/inlineskiplist.h
@@ -239,6 +239,7 @@ class InlineSkipList {
   // point to a node that is before the key, and after should point to
   // a node that is after the key.  after should be nullptr if a good after
   // node isn't conveniently available.
+  template<bool prefetch_before>
   void FindSpliceForLevel(const char* key, Node* before, Node* after, int level,
                           Node** out_prev, Node** out_next);
 
@@ -446,6 +447,9 @@ InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
   Node* last_bigger = nullptr;
   while (true) {
     Node* next = x->Next(level);
+    if (next != nullptr) {
+      PREFETCH(next->Next(level), 0, 1);
+    }
     // Make sure the lists are sorted
     assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
     // Make sure we haven't overshot during our search
@@ -485,6 +489,9 @@ InlineSkipList<Comparator>::FindLessThan(const char* key, Node** prev,
   while (true) {
     assert(x != nullptr);
     Node* next = x->Next(level);
+    if (next != nullptr) {
+      PREFETCH(next->Next(level), 0, 1);
+    }
     assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
     assert(x == head_ || KeyIsAfterNode(key, x));
     if (next != last_not_after && KeyIsAfterNode(key, next)) {
@@ -535,6 +542,9 @@ uint64_t InlineSkipList<Comparator>::EstimateCount(const char* key) const {
   while (true) {
     assert(x == head_ || compare_(x->Key(), key) < 0);
     Node* next = x->Next(level);
+    if (next != nullptr) {
+      PREFETCH(next->Next(level), 0, 1);
+    }
     if (next == nullptr || compare_(next->Key(), key) >= 0) {
       if (level == 0) {
         return count;
@@ -642,12 +652,21 @@ void InlineSkipList<Comparator>::InsertWithHint(const char* key, void** hint) {
 }
 
 template <class Comparator>
+template <bool prefetch_before>
 void InlineSkipList<Comparator>::FindSpliceForLevel(const char* key,
                                                     Node* before, Node* after,
                                                     int level, Node** out_prev,
                                                     Node** out_next) {
   while (true) {
     Node* next = before->Next(level);
+    if (next != nullptr) {
+      PREFETCH(next->Next(level), 0, 1);
+    }
+    if (prefetch_before == true) {
+      if (next != nullptr && level>0) {
+        PREFETCH(next->Next(level-1), 0, 1);
+      }
+    }
     assert(before == head_ || next == nullptr ||
            KeyIsAfterNode(next->Key(), before));
     assert(before == head_ || KeyIsAfterNode(key, before));
@@ -668,7 +687,7 @@ void InlineSkipList<Comparator>::RecomputeSpliceLevels(const char* key,
   assert(recompute_level > 0);
   assert(recompute_level <= splice->height_);
   for (int i = recompute_level - 1; i >= 0; --i) {
-    FindSpliceForLevel(key, splice->prev_[i + 1], splice->next_[i + 1], i,
+    FindSpliceForLevel<true>(key, splice->prev_[i + 1], splice->next_[i + 1], i,
                        &splice->prev_[i], &splice->next_[i]);
   }
 }
@@ -796,7 +815,7 @@ void InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
         // search, because it should be unlikely that lots of nodes have
         // been inserted between prev[i] and next[i]. No point in using
         // next[i] as the after hint, because we know it is stale.
-        FindSpliceForLevel(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
+        FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
                            &splice->next_[i]);
 
         // Since we've narrowed the bracket for level i, we might have
@@ -811,7 +830,7 @@ void InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
     for (int i = 0; i < height; ++i) {
       if (i >= recompute_height &&
           splice->prev_[i]->Next(i) != splice->next_[i]) {
-        FindSpliceForLevel(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
+        FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
                            &splice->next_[i]);
       }
       assert(splice->next_[i] == nullptr ||