protected visitation against spurious insertion sentinels

include/boost/unordered/detail/foa/concurrent_table.hpp

@@ -198,13 +198,13 @@ private:
 template<typename Integral>
 struct atomic_integral
 {
-  operator Integral()const{return n.load(std::memory_order_relaxed);}
-  void operator=(Integral m){n.store(m,std::memory_order_relaxed);}
-  void operator|=(Integral m){n.fetch_or(m,std::memory_order_relaxed);}
-  void operator&=(Integral m){n.fetch_and(m,std::memory_order_relaxed);}
+  operator Integral()const{return n.load(std::memory_order_acquire);}
+  void operator=(Integral m){n.store(m,std::memory_order_release);}
+  void operator|=(Integral m){n.fetch_or(m);}
+  void operator&=(Integral m){n.fetch_and(m);}
 
   atomic_integral& operator=(atomic_integral const& rhs) {
-    n.store(rhs.n.load(std::memory_order_relaxed),std::memory_order_relaxed);
+    n.store(rhs.n.load());
     return *this;
   }
 
@@ -1177,7 +1177,7 @@ private:
         do{
           auto n=unchecked_countr_zero(mask);
           if(BOOST_LIKELY(
-            pg->is_occupied(n)&&bool(this->pred()(x,this->key_from(p[n]))))){
+            is_occupied(pg,n)&&bool(this->pred()(x,this->key_from(p[n]))))){
             f(pg,n,p+n);
             return 1;
           }
@@ -1236,7 +1236,7 @@ private:
           do{
             auto n=unchecked_countr_zero(mask);
             if(BOOST_LIKELY(
-              pg->is_occupied(n)&&
+              is_occupied(pg,n)&&
               bool(this->pred()(*it,this->key_from(p[n]))))){
               f(cast_for(access_mode,type_policy::value_from(p[n])));
               ++res;
@@ -1436,10 +1436,103 @@ private:
     bool        commit_=false;
   };
 
+  struct latch_free_reserve_slot
+  {
+    latch_free_reserve_slot(group_type* pg_,std::size_t pos_):pg{pg_},pos{pos_}
+    {
+      unsigned char expected=0;
+      succeeded_=reinterpret_cast<std::atomic<unsigned char>*>(pg)[pos].
+        compare_exchange_weak(expected,1);
+    }
+
+    ~latch_free_reserve_slot()
+    {
+      if(succeeded_&&!commit_)pg->reset(pos);
+    }
+
+    bool succeeded()const{return succeeded_;}
+
+    void commit(){commit_=true;}
+
+    group_type  *pg;
+    std::size_t pos;
+    bool        succeeded_;
+    bool        commit_=false;
+  };
+
+  struct assign_insert_counter_on_exit
+  {
+    ~assign_insert_counter_on_exit(){counter=x;}
+
+    group_insert_counter_type &counter;
+    boost::uint32_t            x;
+  };
+
   template<typename GroupAccessMode,typename F,typename... Args>
   BOOST_FORCEINLINE int
   unprotected_norehash_emplace_or_visit(
     GroupAccessMode access_mode,F&& f,Args&&... args)
+  {
+    return unprotected_norehash_emplace_or_visit_with_layout(
+      std::integral_constant<bool,group_type::regular_layout>{},
+      access_mode,std::forward<F>(f),std::forward<Args>(args)...);
+  }
+
+  template<typename GroupAccessMode,typename F,typename... Args>
+  BOOST_FORCEINLINE int
+  unprotected_norehash_emplace_or_visit_with_layout(
+    std::true_type, /* regular layout, almost-latch-free */
+    GroupAccessMode access_mode,F&& f,Args&&... args)
+  {
+    const auto &k=this->key_from(std::forward<Args>(args)...);
+    auto        hash=this->hash_for(k);
+    auto        pos0=this->position_for(hash);
+
+    for(;;){
+    startover:
+      boost::uint32_t counter=0;
+      while(BOOST_UNLIKELY((counter=insert_counter(pos0))%2==1)){}
+      if(unprotected_visit(
+        access_mode,k,pos0,hash,std::forward<F>(f)))return 0;
+
+      reserve_size rsize(*this);
+      if(BOOST_LIKELY(rsize.succeeded())){
+        for(prober pb(pos0);;pb.next(this->arrays.groups_size_mask)){
+          auto pos=pb.get();
+          auto pg=this->arrays.groups()+pos;
+          auto mask=pg->match_available();
+          if(BOOST_LIKELY(mask!=0)){
+            auto n=unchecked_countr_zero(mask);
+            latch_free_reserve_slot rslot{pg,n};
+            if(BOOST_UNLIKELY(!rslot.succeeded())){
+              /* slot wasn't empty */
+              goto startover;
+            }
+            if(BOOST_UNLIKELY(
+              !insert_counter(pos0).compare_exchange_weak(counter,counter+1))){
+              /* other thread inserted from pos0, need to start over */
+              goto startover;
+            }
+            assign_insert_counter_on_exit a{insert_counter(pos0),counter+2};
+            auto p=this->arrays.elements()+pos*N+n;
+            this->construct_element(p,std::forward<Args>(args)...);
+            pg->set(n,hash);
+            rslot.commit();
+            rsize.commit();
+            return 1;
+          }
+          pg->mark_overflow(hash);
+        }
+      }
+      else return -1;
+    }
+  }
+
+  template<typename GroupAccessMode,typename F,typename... Args>
+  BOOST_FORCEINLINE int
+  unprotected_norehash_emplace_or_visit_with_layout(
+    std::false_type, /* non-regular layout, latched */
+    GroupAccessMode access_mode,F&& f,Args&&... args)
   {
     const auto &k=this->key_from(std::forward<Args>(args)...);
     auto        hash=this->hash_for(k);
@@ -1486,6 +1579,46 @@ private:
     }
   }
 
+  /* check occupation with previous unsynced match */
+
+  static bool is_occupied(group_type* pg,std::size_t pos)
+  {
+    return is_occupied(
+      std::integral_constant<bool,group_type::regular_layout>{},pg,pos);
+  }
+
+  static bool is_occupied(std::true_type,group_type* pg,std::size_t pos)
+  {
+    /* regular layout, almost-latch-free insertion -> possible reserved slot */
+    return reinterpret_cast<std::atomic<unsigned char>*>(pg)[pos]>1;
+  }
+
+  static bool is_occupied(std::false_type,group_type* pg,std::size_t pos)
+  {
+    /* non-regular layout, latched insertion -> no reserved slots */
+    return pg->is_occupied(pos);
+  }
+
+  /* check occupation with previous synced match */
+
+  static bool is_really_occupied(group_type* pg,std::size_t pos)
+  {
+    return is_really_occupied(
+      std::integral_constant<bool,group_type::regular_layout>{},pg,pos);
+  }
+
+  static bool is_really_occupied(std::true_type,group_type* pg,std::size_t pos)
+  {
+    /* regular layout, almost-latch-free insertion -> possible reserved slot */
+    return reinterpret_cast<std::atomic<unsigned char>*>(pg)[pos]>1;
+  }
+
+  static bool is_really_occupied(std::false_type,group_type*,std::size_t)
+  {
+    /* non-regular layout, latched insertion -> no false positives */
+    return true;
+  }
+
   template<typename GroupAccessMode,typename F>
   auto for_all_elements(GroupAccessMode access_mode,F f)const
     ->decltype(f(nullptr),void())
@@ -1523,7 +1656,7 @@ private:
         auto mask=this->match_really_occupied(pg,last);
         while(mask){
           auto n=unchecked_countr_zero(mask);
-          if(!f(pg,n,p+n))return false;
+          if(BOOST_LIKELY(is_really_occupied(pg,n))&&!f(pg,n,p+n))return false;
           mask&=mask-1;
         }
       }
@@ -1558,7 +1691,7 @@ private:
         auto mask=this->match_really_occupied(&g,last);
         while(mask){
           auto n=unchecked_countr_zero(mask);
-          f(&g,n,p+n);
+          if(BOOST_LIKELY(is_really_occupied(&g,n)))f(&g,n,p+n);
           mask&=mask-1;
         }
       }
@@ -1580,7 +1713,7 @@ private:
         auto mask=this->match_really_occupied(&g,last);
         while(mask){
           auto n=unchecked_countr_zero(mask);
-          if(!f(p+n))return false;
+          if(BOOST_LIKELY(is_really_occupied(&g,n))&&!f(p+n))return false;
           mask&=mask-1;
         }
         return true;