/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 **/

#ifndef QUICKSTEP_UTILITY_BLOOM_FILTER_HPP
#define QUICKSTEP_UTILITY_BLOOM_FILTER_HPP

#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <utility>
#include <vector>

#include "storage/StorageConstants.hpp"
#include "threading/Mutex.hpp"
#include "threading/SharedMutex.hpp"
#include "threading/SpinSharedMutex.hpp"
#include "utility/BloomFilter.pb.h"
#include "utility/Macros.hpp"

#include "glog/logging.h"

namespace quickstep {
/** \addtogroup Utility
 *  @{
 */

/**
 * @brief A simple Bloom Filter implementation with basic primitives
 *        based on Partow's Bloom Filter implementation.
 **/
class BloomFilter {
 public:
  static const uint32_t kNumBitsPerByte = 8;

  /**
   * @brief Constructor.
   * @note When no bit_array is being passed to the constructor,
   *       then the bit_array is owned and managed by this class.
   *
   * @param random_seed A random_seed that generates unique hash functions.
   * @param hash_fn_count The number of hash functions used by this bloom filter.
   * @param bit_array_size_in_bytes Size of the bit array.
   **/
  BloomFilter(const std::uint64_t random_seed,
              const std::size_t hash_fn_count,
              const std::uint64_t bit_array_size_in_bytes)
      : random_seed_(random_seed),
        hash_fn_count_(hash_fn_count),
        array_size_in_bytes_(bit_array_size_in_bytes),
        array_size_(array_size_in_bytes_ * kNumBitsPerByte),
        bit_array_(new std::uint8_t[array_size_in_bytes_]),
        is_bit_array_owner_(true) {
    reset();
    generate_unique_hash_fn();
  }

  /**
   * @brief Constructor.
   * @note When a bit_array is passed as an argument to the constructor,
   *       then the ownership of the bit array lies with the caller.
   *
   * @param random_seed A random_seed that generates unique hash functions.
   * @param hash_fn_count The number of hash functions used by this bloom filter.
   * @param bit_array_size_in_bytes Size of the bit array.
   * @param bit_array A pointer to the memory region that is used to store bit array.
   * @param is_initialized A boolean that indicates whether to zero-out the region
   *                       before use or not.
   **/
  BloomFilter(const std::uint64_t random_seed,
              const std::size_t hash_fn_count,
              const std::uint64_t bit_array_size_in_bytes,
              std::uint8_t *bit_array,
              const bool is_initialized)
      : random_seed_(random_seed),
        hash_fn_count_(hash_fn_count),
        array_size_in_bytes_(bit_array_size_in_bytes),
        array_size_(bit_array_size_in_bytes * kNumBitsPerByte),
        bit_array_(bit_array),  // Owned by the calling method.
        is_bit_array_owner_(false) {
    if (!is_initialized) {
      reset();
    }
    generate_unique_hash_fn();
  }

  /**
   * @brief Constructor.
   * @note When a bloom filter proto is passed as an initializer,
   *       then the bit_array is owned and managed by this class.
   *
   * @param bloom_filter_proto The protobuf representation of a
   *        bloom filter configuration.
   **/
  explicit BloomFilter(const serialization::BloomFilter &bloom_filter_proto)
      : random_seed_(bloom_filter_proto.bloom_filter_seed()),
        hash_fn_count_(bloom_filter_proto.number_of_hashes()),
        array_size_in_bytes_(bloom_filter_proto.bloom_filter_size()),
        array_size_(array_size_in_bytes_ * kNumBitsPerByte),
        bit_array_(new std::uint8_t[array_size_in_bytes_]),
        is_bit_array_owner_(true) {
    reset();
    generate_unique_hash_fn();
  }

  /**
   * @brief Destructor.
   **/
  ~BloomFilter() {
    if (is_bit_array_owner_) {
      bit_array_.reset();
    } else {
      bit_array_.release();
    }
  }

  static bool ProtoIsValid(const serialization::BloomFilter &bloom_filter_proto) {
    return bloom_filter_proto.IsInitialized();
  }

  /**
   * @brief Zeros out the contents of the bit array.
   **/
  inline void reset() {
    // Initialize the bit_array with all zeros.
    std::fill_n(bit_array_.get(), array_size_in_bytes_, 0x00);
    inserted_element_count_ = 0;
  }

  /**
   * @brief Get the random seed that was used to initialize this bloom filter.
   *
   * @return Returns the random seed.
   **/
  inline std::uint64_t getRandomSeed() const {
    return random_seed_;
  }

  /**
   * @brief Get the number of hash functions used in this bloom filter.
   *
   * @return Returns the number of hash functions.
   **/
  inline std::uint32_t getNumberOfHashes() const {
    return hash_fn_count_;
  }

  /**
   * @brief Get the size of the bit array in bytes for this bloom filter.
   *
   * @return Returns the bit array size (in bytes).
   **/
  inline std::uint64_t getBitArraySize() const {
    return array_size_in_bytes_;
  }

  /**
   * @brief Get the constant pointer to the bit array for this bloom filter
   *
   * @return Returns constant pointer to the bit array.
   **/
  inline const std::uint8_t* getBitArray() const {
    return bit_array_.get();
  }

  /**
   * @brief Inserts a given value into the bloom filter in a thread-safe manner.
   *
   * @param key_begin A pointer to the value being inserted.
   * @param length Size of the value being inserted in bytes.
   */
  inline void insert(const std::uint8_t *key_begin, const std::size_t length) {
    // Locks are needed during insertion, when multiple workers may be modifying the
    // bloom filter concurrently. However, locks are not required during membership test.
    std::size_t bit_index = 0;
    std::size_t bit = 0;
    std::vector<std::pair<std::size_t, std::size_t>> modified_bit_positions;
    std::vector<bool> is_bit_position_correct;

    // Determine all the bit positions that are required to be set.
    for (std::size_t i = 0; i < hash_fn_count_; ++i) {
      compute_indices(hash_ap(key_begin, length, hash_fn_[i]), &bit_index, &bit);
      modified_bit_positions.push_back(std::make_pair(bit_index, bit));
    }

    // Acquire a reader lock and check which of the bit positions are already set.
    {
      SpinSharedMutexSharedLock<false> shared_reader_lock(bloom_filter_insert_mutex_);
      for (std::size_t i = 0; i < hash_fn_count_; ++i) {
        bit_index = modified_bit_positions[i].first;
        bit = modified_bit_positions[i].second;
        if (((bit_array_.get())[bit_index / kNumBitsPerByte] & (1 << bit)) != (1 << bit)) {
          is_bit_position_correct.push_back(false);
        } else {
          is_bit_position_correct.push_back(true);
        }
      }
    }

    // Acquire a writer lock and set the bit positions are which are not set.
    {
      SpinSharedMutexExclusiveLock<false> exclusive_writer_lock(bloom_filter_insert_mutex_);
      for (std::size_t i = 0; i < hash_fn_count_; ++i) {
        if (!is_bit_position_correct[i]) {
          bit_index = modified_bit_positions[i].first;
          bit = modified_bit_positions[i].second;
          (bit_array_.get())[bit_index / kNumBitsPerByte] |= (1 << bit);
        }
      }
    }
    ++inserted_element_count_;
  }

  /**
   * @brief Inserts a given value into the bloom filter.
   * @Warning This is a faster thread-unsafe version of the insert() function.
   *          The caller needs to ensure the thread safety.
   *
   * @param key_begin A pointer to the value being inserted.
   * @param length Size of the value being inserted in bytes.
   */
  inline void insertUnSafe(const std::uint8_t *key_begin, const std::size_t length) {
    std::size_t bit_index = 0;
    std::size_t bit = 0;

    for (std::size_t i = 0; i < hash_fn_count_; ++i) {
      compute_indices(hash_ap(key_begin, length, hash_fn_[i]), &bit_index, &bit);
      (bit_array_.get())[bit_index / kNumBitsPerByte] |= (1 << bit);
    }

    ++inserted_element_count_;
  }

  /**
   * @brief Test membership of a given value in the bloom filter.
   *        If true is returned, then a value may or may not be present in the bloom filter.
   *        If false is returned, a value is certainly not present in the bloom filter.
   *
   * @note The membersip test does not require any locks, because the assumption is that
   *       the bloom filter will only be used after it has been built.
   *
   * @param key_begin A pointer to the value being tested for membership.
   * @param length Size of the value being inserted in bytes.
   */
  inline bool contains(const std::uint8_t *key_begin, const std::size_t length) const {
    std::size_t bit_index = 0;
    std::size_t bit = 0;
    for (std::size_t i = 0; i < hash_fn_count_; ++i) {
      compute_indices(hash_ap(key_begin, length, hash_fn_[i]), &bit_index, &bit);
      if (((bit_array_.get())[bit_index / kNumBitsPerByte] & (1 << bit)) != (1 << bit)) {
        return false;
      }
    }
    return true;
  }

  /**
   * @brief Perform a bitwise-OR of the given Bloom filter with this bloom filter.
   *        Essentially, it does a union of this bloom filter with the passed bloom filter.
   *
   * @param bloom_filter A const pointer to the bloom filter object to do bitwise-OR with.
   */
  inline void bitwiseOr(const BloomFilter *bloom_filter) {
    SpinSharedMutexExclusiveLock<false> exclusive_writer_lock(bloom_filter_insert_mutex_);
    for (std::size_t byte_index = 0; byte_index < bloom_filter->getBitArraySize(); ++byte_index) {
      (bit_array_.get())[byte_index] |= bloom_filter->getBitArray()[byte_index];
    }
  }

  /**
   * @brief Return the number of elements currently inserted into bloom filter.
   *
   * @return The number of elements inserted into bloom filter.
   **/
  inline std::size_t element_count() const {
    return inserted_element_count_;
  }

 protected:
  inline void compute_indices(const std::uint32_t &hash, std::size_t *bit_index, std::size_t *bit) const {
    *bit_index = hash % array_size_;
    *bit = *bit_index % kNumBitsPerByte;
  }

  void generate_unique_hash_fn() {
    hash_fn_.reserve(hash_fn_count_);
    const std::uint32_t predef_hash_fn_count = 128;
    static const std::uint32_t predef_hash_fn[predef_hash_fn_count] = {
       0xAAAAAAAA, 0x55555555, 0x33333333, 0xCCCCCCCC,
       0x66666666, 0x99999999, 0xB5B5B5B5, 0x4B4B4B4B,
       0xAA55AA55, 0x55335533, 0x33CC33CC, 0xCC66CC66,
       0x66996699, 0x99B599B5, 0xB54BB54B, 0x4BAA4BAA,
       0xAA33AA33, 0x55CC55CC, 0x33663366, 0xCC99CC99,
       0x66B566B5, 0x994B994B, 0xB5AAB5AA, 0xAAAAAA33,
       0x555555CC, 0x33333366, 0xCCCCCC99, 0x666666B5,
       0x9999994B, 0xB5B5B5AA, 0xFFFFFFFF, 0xFFFF0000,
       0xB823D5EB, 0xC1191CDF, 0xF623AEB3, 0xDB58499F,
       0xC8D42E70, 0xB173F616, 0xA91A5967, 0xDA427D63,
       0xB1E8A2EA, 0xF6C0D155, 0x4909FEA3, 0xA68CC6A7,
       0xC395E782, 0xA26057EB, 0x0CD5DA28, 0x467C5492,
       0xF15E6982, 0x61C6FAD3, 0x9615E352, 0x6E9E355A,
       0x689B563E, 0x0C9831A8, 0x6753C18B, 0xA622689B,
       0x8CA63C47, 0x42CC2884, 0x8E89919B, 0x6EDBD7D3,
       0x15B6796C, 0x1D6FDFE4, 0x63FF9092, 0xE7401432,
       0xEFFE9412, 0xAEAEDF79, 0x9F245A31, 0x83C136FC,
       0xC3DA4A8C, 0xA5112C8C, 0x5271F491, 0x9A948DAB,
       0xCEE59A8D, 0xB5F525AB, 0x59D13217, 0x24E7C331,
       0x697C2103, 0x84B0A460, 0x86156DA9, 0xAEF2AC68,
       0x23243DA5, 0x3F649643, 0x5FA495A8, 0x67710DF8,
       0x9A6C499E, 0xDCFB0227, 0x46A43433, 0x1832B07A,
       0xC46AFF3C, 0xB9C8FFF0, 0xC9500467, 0x34431BDF,
       0xB652432B, 0xE367F12B, 0x427F4C1B, 0x224C006E,
       0x2E7E5A89, 0x96F99AA5, 0x0BEB452A, 0x2FD87C39,
       0x74B2E1FB, 0x222EFD24, 0xF357F60C, 0x440FCB1E,
       0x8BBE030F, 0x6704DC29, 0x1144D12F, 0x948B1355,
       0x6D8FD7E9, 0x1C11A014, 0xADD1592F, 0xFB3C712E,
       0xFC77642F, 0xF9C4CE8C, 0x31312FB9, 0x08B0DD79,
       0x318FA6E7, 0xC040D23D, 0xC0589AA7, 0x0CA5C075,
       0xF874B172, 0x0CF914D5, 0x784D3280, 0x4E8CFEBC,
       0xC569F575, 0xCDB2A091, 0x2CC016B4, 0x5C5F4421
    };
    if (hash_fn_count_ <= predef_hash_fn_count) {
      std::copy(predef_hash_fn, predef_hash_fn + hash_fn_count_, hash_fn_.begin());
      for (std::uint32_t i = 0; i < hash_fn_.size(); ++i) {
        hash_fn_[i] = hash_fn_[i] * hash_fn_[(i + 3) % hash_fn_count_] + static_cast<std::uint32_t>(random_seed_);
      }
    } else {
      LOG(FATAL) << "Requested number of hash functions is too large.";
    }
  }

  inline std::uint32_t hash_ap(const std::uint8_t *begin, std::size_t remaining_length, std::uint32_t hash) const {
    const std::uint8_t *itr = begin;
    std::uint32_t loop = 0;
    while (remaining_length >= 8) {
      const std::uint32_t &i1 = *(reinterpret_cast<const std::uint32_t*>(itr)); itr += sizeof(std::uint32_t);
      const std::uint32_t &i2 = *(reinterpret_cast<const std::uint32_t*>(itr)); itr += sizeof(std::uint32_t);
      hash ^= (hash <<  7) ^  i1 * (hash >> 3) ^ (~((hash << 11) + (i2 ^ (hash >> 5))));
      remaining_length -= 8;
    }
    if (remaining_length) {
      if (remaining_length >= 4) {
        const std::uint32_t &i = *(reinterpret_cast<const std::uint32_t*>(itr));
        if (loop & 0x01) {
          hash ^= (hash <<  7) ^  i * (hash >> 3);
        } else {
          hash ^= (~((hash << 11) + (i ^ (hash >> 5))));
        }
        ++loop;
        remaining_length -= 4;
        itr += sizeof(std::uint32_t);
      }
      if (remaining_length >= 2) {
        const std::uint16_t &i = *(reinterpret_cast<const std::uint16_t*>(itr));
        if (loop & 0x01) {
          hash ^= (hash <<  7) ^  i * (hash >> 3);
        } else {
          hash ^= (~((hash << 11) + (i ^ (hash >> 5))));
        }
        ++loop;
        remaining_length -= 2;
        itr += sizeof(std::uint16_t);
      }
      if (remaining_length) {
        hash += ((*itr) ^ (hash * 0xA5A5A5A5)) + loop;
      }
    }
    return hash;
  }

 private:
  const std::uint64_t random_seed_;
  std::vector<std::uint32_t> hash_fn_;
  const std::uint32_t hash_fn_count_;
  std::uint64_t array_size_in_bytes_;
  std::uint64_t array_size_;
  std::unique_ptr<std::uint8_t> bit_array_;
  std::uint32_t inserted_element_count_;
  const bool is_bit_array_owner_;

  alignas(kCacheLineBytes) mutable SpinSharedMutex<false> bloom_filter_insert_mutex_;

  DISALLOW_COPY_AND_ASSIGN(BloomFilter);
};

/** @} */

}  // namespace quickstep

#endif  // QUICKSTEP_UTILITY_BLOOM_FILTER_HPP