Re: [boost] GSOC 2015 : Project on Concurrent Hash Tables

Subject: Re: [boost] GSOC 2015 : Project on Concurrent Hash Tables
From: Amarnath V A (me_at_[hidden])
Date: 2015-03-10 09:13:43

Hello,

I have made some changes to the move constructor and also tried
writing the copy constructor. Please comment on the implementation of
the constructors. Does these meet the requirements and please point
out if I have missed some crucial steps.

concurrent_unordered_map(const concurrent_unordered_map &old_map) :
          _hasher(old_map._hasher),
          _key_equal(old_map._key_equal),
          _allocator(old_map._allocator),
          _max_load_factor(old_map._max_load_factor),
          _min_bucket_capacity(old_map._min_bucket_capacity),
          _oldbucketit(_oldbuckets.begin())
      {
          _oldbuckets.fill(nullptr);
          buckets_type *tempbuckets=old_map._buckets.load(memory_order_consume);
          size_t n = tempbuckets->size();

          buckets_type *buckets=new buckets_type(n);
          buckets=_buckets.exchange(buckets, memory_order_acq_rel);
          for(const auto &ob : *tempbuckets)
          {
            if(ob.count.load(memory_order_relaxed))
            {
              for(auto &i : ob.items)
              {
                if(i.p)
                {
                  size_t hash=_hasher(i.p->first);
                  auto itb=_get_bucket(hash);
                  bucket_type &b=*itb;
                  if(b.items.size()==b.items.capacity())
                  {
                    size_t newcapacity=b.items.capacity()*2;
                    b.items.reserve(newcapacity ? newcapacity : 1);
                  }
                  b.items.push_back(item_type(hash, i.p));
                  b.count.fetch_add(1, memory_order_relaxed);
                }
              }
            }
          }
      }

      concurrent_unordered_map(concurrent_unordered_map &&old_map)
BOOST_NOEXCEPT :
          _hasher(std::move(old_map._hasher)),
          _key_equal(std::move(old_map._key_equal)),
          _allocator(std::move(old_map._allocator)),
          _max_load_factor(std::move(old_map._max_load_factor)),
          _min_bucket_capacity(std::move(old_map._min_bucket_capacity)),
          _oldbucketit(_oldbuckets.begin())
      {
          _oldbuckets.fill(nullptr);

          typedef decltype(_rehash_lock) rehash_lock_t;
          lock_guard<rehash_lock_t> guard(old_map._rehash_lock,
adopt_lock_t());
          size_t n = old_map._buckets.load(memory_order_relaxed)->size();

          buckets_type *tempbuckets=new buckets_type(13);
          auto untempbuckets=undoer([&tempbuckets]{ delete tempbuckets; });
          buckets_type *buckets=new buckets_type(n);

          // Swap old buckets with new buckets
          tempbuckets=old_map._buckets.exchange(tempbuckets,
memory_order_acq_rel);
          buckets=_buckets.exchange(buckets, memory_order_acq_rel);
          // Tell all threads using old buckets to start reloading the
bucket list
          for(auto &b : *tempbuckets)
            b.lock.store(2);
          // If it's a simple buckets cycle, simply move over all
items as it's noexcept
          if(_buckets.load(memory_order_relaxed)->size()==tempbuckets->size())
          {
            // Simply move the old buckets into new buckets as-is
            for(auto obit=tempbuckets->begin(),
bit=_buckets.load(memory_order_relaxed)->begin();
obit!=tempbuckets->end(); ++obit, ++bit)
            {
              auto &ob=*obit;
              auto &b=*bit;
              if(ob.count.load(memory_order_relaxed))
              {
                b.items=std::move(ob.items);
                b.count.store(ob.count.load(memory_order_consume),
memory_order_release);
                ob.count.store(0, memory_order_release);
              }
            }
          }
          // Stop old buckets deleting stuff
          _reset_buckets(*tempbuckets);
          // Hold onto old buckets for a while
          typename old_buckets_type::iterator
myoldbucket=old_map._oldbucketit++;
          if(old_map._oldbucketit==old_map._oldbuckets.end())
old_map._oldbucketit=old_map._oldbuckets.begin();
          delete *myoldbucket;
          *myoldbucket=tempbuckets;
          tempbuckets=nullptr;
          untempbuckets.dismissed=true;
      }

But I have peculiar issue. When I run basic unit test for the copy
constructor, it fails saying it encountered a SIGABRT to memory
corruption. The unit test is pretty basic and is as follows:

BOOST_AUTO_TEST_CASE(works/concurrent_unordered_map/copyconstructor/basic,
"Tests that concurrent_unordered_map works as expected")
{
  printf("\n=== concurrent_unordered_map basic copy constructor ===\n");
  boost::spinlock::concurrent_unordered_map<int, int> map1;
  map1.reserve(100);    // test dense map
  BOOST_CHECK(map1.empty());
  BOOST_CHECK(map1.size()==0);
  for(int n=-200; n<=200; n+=2)
  {
    map1.emplace(n, n);
  }
  BOOST_CHECK(!map1.empty());
  BOOST_CHECK(map1.size()==201);
  printf("Load factor for map1 is %f\n", map1.load_factor());

  boost::spinlock::concurrent_unordered_map<int, int> map2(map1);
  BOOST_CHECK(!map2.empty());
  BOOST_CHECK(!map1.empty());
  BOOST_CHECK(map2.size()==201);
  BOOST_CHECK(map1.size()==201);

  printf("Load factor for map1 is %f\n", map1.load_factor());
  printf("Load factor for map2 is %f\n", map2.load_factor());

  map1.clear();
  map2.clear();
  BOOST_CHECK(map1.empty());
  BOOST_CHECK(map2.empty());
  BOOST_CHECK(map1.size()==0);
  BOOST_CHECK(map2.size()==0);
}

The second last BOOST_CHECK is were I hit on this issue. And the error is

Error in `/home/amarnath/work/boost.spinlock/test/unittests_1':
corrupted double-linked list: 0x0000000000689200 ***

But, when I remove the BOOST_CHECK and just run map1.size() and
map2.size() calls, I have no issues.I tried to understand the problem
but could not make out why it is failing? Any help is welcome.

Thanks,
Amarnath