/*

 * Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 *

 */

#ifndef SHARE_UTILITIES_CONCURRENTHASHTABLE_INLINE_HPP

#define SHARE_UTILITIES_CONCURRENTHASHTABLE_INLINE_HPP

#include "memory/allocation.inline.hpp"

#include "runtime/atomic.hpp"

#include "runtime/orderAccess.hpp"

#include "runtime/prefetch.inline.hpp"

#include "utilities/concurrentHashTable.hpp"

#include "utilities/globalCounter.inline.hpp"

#include "utilities/numberSeq.hpp"

#include "utilities/spinYield.hpp"

// 2^30 = 1G buckets

#define SIZE_BIG_LOG2 30

// 2^5  = 32 buckets

#define SIZE_SMALL_LOG2 5

// Number from spinYield.hpp. In some loops SpinYield would be unfair.

#define SPINPAUSES_PER_YIELD 8192

#ifdef ASSERT

#ifdef _LP64

// Two low bits are not usable.

static const void* POISON_PTR = (void*)UCONST64(0xfbadbadbadbadbac);

#else

// Two low bits are not usable.

static const void* POISON_PTR = (void*)0xffbadbac;

#endif

#endif

// Node

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::Node*

ConcurrentHashTable<CONFIG, F>::

  Node::next() const

{

  return Atomic::load_acquire(&_next);

}

// Bucket

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::Node*

ConcurrentHashTable<CONFIG, F>::

  Bucket::first_raw() const

{

  return Atomic::load_acquire(&_first);

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  Bucket::release_assign_node_ptr(

    typename ConcurrentHashTable<CONFIG, F>::Node* const volatile * dst,

    typename ConcurrentHashTable<CONFIG, F>::Node* node) const

{

  // Due to this assert this methods is not static.

  assert(is_locked(), "Must be locked.");

  Node** tmp = (Node**)dst;

  Atomic::release_store(tmp, clear_set_state(node, *dst));

}

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::Node*

ConcurrentHashTable<CONFIG, F>::

  Bucket::first() const

{

  // We strip the states bit before returning the ptr.

  return clear_state(Atomic::load_acquire(&_first));

}

template <typename CONFIG, MEMFLAGS F>

inline bool ConcurrentHashTable<CONFIG, F>::

  Bucket::have_redirect() const

{

  return is_state(first_raw(), STATE_REDIRECT_BIT);

}

template <typename CONFIG, MEMFLAGS F>

inline bool ConcurrentHashTable<CONFIG, F>::

  Bucket::is_locked() const

{

  return is_state(first_raw(), STATE_LOCK_BIT);

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  Bucket::lock()

{

  int i = 0;

  // SpinYield would be unfair here

  while (!this->trylock()) {

    if ((++i) == SPINPAUSES_PER_YIELD) {

      // On contemporary OS yielding will give CPU to another runnable thread if

      // there is no CPU available.

      os::naked_yield();

      i = 0;

    } else {

      SpinPause();

    }

  }

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  Bucket::release_assign_last_node_next(

     typename ConcurrentHashTable<CONFIG, F>::Node* node)

{

  assert(is_locked(), "Must be locked.");

  Node* const volatile * ret = first_ptr();

  while (clear_state(*ret) != NULL) {

    ret = clear_state(*ret)->next_ptr();

  }

  release_assign_node_ptr(ret, node);

}

template <typename CONFIG, MEMFLAGS F>

inline bool ConcurrentHashTable<CONFIG, F>::

  Bucket::cas_first(typename ConcurrentHashTable<CONFIG, F>::Node* node,

                    typename ConcurrentHashTable<CONFIG, F>::Node* expect

                    )

{

  if (is_locked()) {

    return false;

  }

    return true;

  }

  return false;

}

template <typename CONFIG, MEMFLAGS F>

inline bool ConcurrentHashTable<CONFIG, F>::

  Bucket::trylock()

{

  if (is_locked()) {

    return false;

  }

  // We will expect a clean first pointer.

  Node* tmp = first();

    return true;

  }

  return false;

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  Bucket::unlock()

{

  assert(is_locked(), "Must be locked.");

  assert(!have_redirect(),

         "Unlocking a bucket after it has reached terminal state.");

  Atomic::release_store(&_first, clear_state(first()));

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  Bucket::redirect()

{

  assert(is_locked(), "Must be locked.");

  Atomic::release_store(&_first, set_state(_first, STATE_REDIRECT_BIT));

}

// InternalTable

template <typename CONFIG, MEMFLAGS F>

inline ConcurrentHashTable<CONFIG, F>::

  InternalTable::InternalTable(size_t log2_size)

    : _log2_size(log2_size), _size(((size_t)1ul) << _log2_size),

      _hash_mask(~(~((size_t)0) << _log2_size))

{

  assert(_log2_size >= SIZE_SMALL_LOG2 && _log2_size <= SIZE_BIG_LOG2,

         "Bad size");

  _buckets = NEW_C_HEAP_ARRAY(Bucket, _size, F);

  // Use placement new for each element instead of new[] which could use more

  // memory than allocated.

  for (size_t i = 0; i < _size; ++i) {

    new (_buckets + i) Bucket();

  }

}

template <typename CONFIG, MEMFLAGS F>

inline ConcurrentHashTable<CONFIG, F>::

  InternalTable::~InternalTable()

{

  FREE_C_HEAP_ARRAY(Bucket, _buckets);

}

// ScopedCS

template <typename CONFIG, MEMFLAGS F>

inline ConcurrentHashTable<CONFIG, F>::

  ScopedCS::ScopedCS(Thread* thread, ConcurrentHashTable<CONFIG, F>* cht)

    : _thread(thread),

      _cht(cht),

      _cs_context(GlobalCounter::critical_section_begin(_thread))

{

  // This version is published now.

  if (Atomic::load_acquire(&_cht->_invisible_epoch) != NULL) {

    Atomic::release_store_fence(&_cht->_invisible_epoch, (Thread*)NULL);

  }

}

template <typename CONFIG, MEMFLAGS F>

inline ConcurrentHashTable<CONFIG, F>::

  ScopedCS::~ScopedCS()

{

  GlobalCounter::critical_section_end(_thread, _cs_context);

}

template <typename CONFIG, MEMFLAGS F>

template <typename LOOKUP_FUNC>

inline typename CONFIG::Value* ConcurrentHashTable<CONFIG, F>::

  MultiGetHandle::get(LOOKUP_FUNC& lookup_f, bool* grow_hint)

{

  return ScopedCS::_cht->internal_get(ScopedCS::_thread, lookup_f, grow_hint);

}

// HaveDeletables

template <typename CONFIG, MEMFLAGS F>

template <typename EVALUATE_FUNC>

inline bool ConcurrentHashTable<CONFIG, F>::

  HaveDeletables<true, EVALUATE_FUNC>::have_deletable(Bucket* bucket,

                                                      EVALUATE_FUNC& eval_f,

                                                      Bucket* prefetch_bucket)

{

  // Instantiated for pointer type (true), so we can use prefetch.

  // When visiting all Nodes doing this prefetch give around 30%.

  Node* pref = prefetch_bucket != NULL ? prefetch_bucket->first() : NULL;

  for (Node* next = bucket->first(); next != NULL ; next = next->next()) {

    if (pref != NULL) {

      Prefetch::read(*pref->value(), 0);

      pref = pref->next();

    }

    // Read next() Node* once.  May be racing with a thread moving the next

    // pointers.

    Node* next_pref = next->next();

    if (next_pref != NULL) {

      Prefetch::read(*next_pref->value(), 0);

    }

    if (eval_f(next->value())) {

      return true;

    }

  }

  return false;

}

template <typename CONFIG, MEMFLAGS F>

template <bool b, typename EVALUATE_FUNC>

inline bool ConcurrentHashTable<CONFIG, F>::

  HaveDeletables<b, EVALUATE_FUNC>::have_deletable(Bucket* bucket,

                                                   EVALUATE_FUNC& eval_f,

                                                   Bucket* preb)

{

  for (Node* next = bucket->first(); next != NULL ; next = next->next()) {

    if (eval_f(next->value())) {

      return true;

    }

  }

  return false;

}

// ConcurrentHashTable

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  write_synchonize_on_visible_epoch(Thread* thread)

{

  assert(_resize_lock_owner == thread, "Re-size lock not held");

  OrderAccess::fence(); // Prevent below load from floating up.

  // If no reader saw this version we can skip write_synchronize.

  if (Atomic::load_acquire(&_invisible_epoch) == thread) {

    return;

  }

  assert(_invisible_epoch == NULL, "Two thread doing bulk operations");

  // We set this/next version that we are synchronizing for to not published.

  // A reader will zero this flag if it reads this/next version.

  Atomic::release_store(&_invisible_epoch, thread);

  GlobalCounter::write_synchronize();

}

template <typename CONFIG, MEMFLAGS F>

inline bool ConcurrentHashTable<CONFIG, F>::

  try_resize_lock(Thread* locker)

{

  if (_resize_lock->try_lock()) {

    if (_resize_lock_owner != NULL) {

      assert(locker != _resize_lock_owner, "Already own lock");

      // We got mutex but internal state is locked.

      _resize_lock->unlock();

      return false;

    }

  } else {

    return false;

  }

  _invisible_epoch = 0;

  _resize_lock_owner = locker;

  return true;

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  lock_resize_lock(Thread* locker)

{

  size_t i = 0;

  // If lock is hold by some other thread, the chances that it is return quick

  // is low. So we will prefer yielding.

  SpinYield yield(1, 512);

  do {

    _resize_lock->lock_without_safepoint_check();

    // If holder of lock dropped mutex for safepoint mutex might be unlocked,

    // and _resize_lock_owner will contain the owner.

    if (_resize_lock_owner != NULL) {

      assert(locker != _resize_lock_owner, "Already own lock");

      // We got mutex but internal state is locked.

      _resize_lock->unlock();

      yield.wait();

    } else {

      break;

    }

  } while(true);

  _resize_lock_owner = locker;

  _invisible_epoch = 0;

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  unlock_resize_lock(Thread* locker)

{

  _invisible_epoch = 0;

  assert(locker == _resize_lock_owner, "Not unlocked by locker.");

  _resize_lock_owner = NULL;

  _resize_lock->unlock();

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  free_nodes()

{

  // We assume we are not MT during freeing.

  for (size_t node_it = 0; node_it < _table->_size; node_it++) {

    Bucket* bucket = _table->get_buckets() + node_it;

    Node* node = bucket->first();

    while (node != NULL) {

      Node* free_node = node;

      node = node->next();

      Node::destroy_node(free_node);

    }

  }

}

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::InternalTable*

ConcurrentHashTable<CONFIG, F>::

  get_table() const

{

  return Atomic::load_acquire(&_table);

}

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::InternalTable*

ConcurrentHashTable<CONFIG, F>::

  get_new_table() const

{

  return Atomic::load_acquire(&_new_table);

}

template <typename CONFIG, MEMFLAGS F>

inline typename ConcurrentHashTable<CONFIG, F>::InternalTable*

ConcurrentHashTable<CONFIG, F>::

  set_table_from_new()

{

  InternalTable* old_table = _table;

  // Publish the new table.

  Atomic::release_store(&_table, _new_table);

  // All must see this.

  GlobalCounter::write_synchronize();

  // _new_table not read any more.

  _new_table = NULL;

  DEBUG_ONLY(_new_table = (InternalTable*)POISON_PTR;)

  return old_table;

}

template <typename CONFIG, MEMFLAGS F>

inline void ConcurrentHashTable<CONFIG, F>::

  internal_grow_range(Thread* thread, size_t start, size_t stop)

{

  assert(stop <= _table->_size, "Outside backing array");

  assert(_new_table != NULL, "Grow not proper setup before start");

  // The state is also copied here. Hence all buckets in new table will be

  // locked. I call the siblings odd/even, where even have high bit 0 and odd

  // have high bit 1.

  for (size_t even_index = start; even_index < stop; even_index++) {

    Bucket* bucket = _table->get_bucket(even_index);

    bucket->lock();

    size_t odd_index = even_index + _table->_size;

    _new_table->get_buckets()[even_index] = *bucket;

    _new_table->get_buckets()[odd_index] = *bucket;

    // Moves lockers go to new table, where they will wait until unlock() below.

    bucket->redirect(); /* Must release stores above */

    // When this is done we have separated the nodes into corresponding buckets

    // in new table.

    if (!unzip_bucket(thread, _table, _new_table, even_index, odd_index)) {

      // If bucket is empty, unzip does nothing.

      // We must make sure readers go to new table before we poison the bucket.

      DEBUG_ONLY(GlobalCounter::write_synchronize();)

    }

    // Unlock for writes into the new table buckets.

    _new_table->get_bucket(even_index)->unlock();

    _new_table->get_bucket(odd_index)->unlock();

    DEBUG_ONLY(

       bucket->release_assign_node_ptr(

          _table->get_bucket(even_index)->first_ptr(), (Node*)POISON_PTR);

    )

  }

}

template <typename CONFIG, MEMFLAGS F>

template <typename LOOKUP_FUNC, typename DELETE_FUNC>

inline bool ConcurrentHashTable<CONFIG, F>::

  internal_remove(Thread* thread, LOOKUP_FUNC& lookup_f, DELETE_FUNC& delete_f)

{

  Bucket* bucket = get_bucket_locked(thread, lookup_f.get_hash());

  assert(bucket->is_locked(), "Must be locked.");

  Node* const volatile * rem_n_prev = bucket->first_ptr();

  Node* rem_n = bucket->first();

  bool have_dead = false;

  while (rem_n != NULL) {

    if (lookup_f.equals(rem_n->value(), &have_dead)) {

      bucket->release_assign_node_ptr(rem_n_prev, rem_n->next());

      break;

    } else {

      rem_n_prev = rem_n->next_ptr();

      rem_n = rem_n->next();

    }

  }

  bucket->unlock();

  if (rem_n == NULL) {

    return false;

  }

  // Publish the deletion.

  GlobalCounter::write_synchronize();

  delete_f(rem_n->value());

  Node::destroy_node(rem_n);

  JFR_ONLY(_stats_rate.remove();)

  return true;

}

template <typename CONFIG, MEMFLAGS F>

template <typename EVALUATE_FUNC, typename DELETE_FUNC>

inline void ConcurrentHashTable<CONFIG, F>::

  do_bulk_delete_locked_for(Thread* thread, size_t start_idx, size_t stop_idx,

                            EVALUATE_FUNC& eval_f, DELETE_FUNC& del_f, bool is_mt)

{

  // Here we have resize lock so table is SMR safe, and there is no new

  // table. Can do this in parallel if we want.

  assert((is_mt && _resize_lock_owner != NULL) ||

         (!is_mt && _resize_lock_owner == thread), "Re-size lock not held");

  Node* ndel[BULK_DELETE_LIMIT];

  InternalTable* table = get_table();

  assert(start_idx < stop_idx, "Must be");

  assert(stop_idx <= _table->_size, "Must be");

  // Here manual do critical section since we don't want to take the cost of

  // locking the bucket if there is nothing to delete. But we can have

  // concurrent single deletes. The _invisible_epoch can only be used by the

  // owner of _resize_lock, us here. There we should not changed it in our

  // own read-side.

  GlobalCounter::CSContext cs_context = GlobalCounter::critical_section_begin(thread);

  for (size_t bucket_it = start_idx; bucket_it < stop_idx; bucket_it++) {

    Bucket* bucket = table->get_bucket(bucket_it);

    Bucket* prefetch_bucket = (bucket_it+1) < stop_idx ?

                              table->get_bucket(bucket_it+1) : NULL;

    if (!HaveDeletables<IsPointer<VALUE>::value, EVALUATE_FUNC>::

        have_deletable(bucket, eval_f, prefetch_bucket)) {

        // Nothing to remove in this bucket.

        continue;

    }

    GlobalCounter::critical_section_end(thread, cs_context);

    // We left critical section but the bucket cannot be removed while we hold

    // the _resize_lock.

    bucket->lock();

    size_t nd = delete_check_nodes(bucket, eval_f, BULK_DELETE_LIMIT, ndel);