//
// Created by 马也驰 on 2024/12/9.
//

#ifndef LEVELDB_SLOTPAGE_H
#define LEVELDB_SLOTPAGE_H

#include <cassert>
#include <cstdint>
#include <cstdlib>
#include <vector>
#include <iostream>
#include <fstream>
#include <mutex>
#include <iostream>
#include <unordered_map>

// bitmap: || bitmap_size(size_t) | first_empty_slot(size_t) || bits... |

#define BITMAP_SIZE 8192
#define BITS_PER_BYTE 8

#define POSMASK(off) (0x01 << (BITS_PER_BYTE-(off)-1))
#define SETBIT(byte, off) (*(byte) |= POSMASK(off))
#define RESETBIT(byte, off) (*(byte) &= (0xff ^ POSMASK(off)))
#define HASFREESLOT(byte) ((byte) != ~0x0)

struct slot_content {
  uint32_t vlog_num;
  uint32_t value_offset;
  slot_content() {}
  slot_content(uint32_t vn, uint32_t vo) {
    vlog_num = vn;
    value_offset = vo;
  }
};

// test passed
class SlotCache {
// slot number -> slot content
#define BLOCK_NUM 16
#define BLOCK_SIZE 4096
#define SLOT_PER_BLOCK (BLOCK_SIZE/sizeof(slot_content))
#define SLOT_OFFSET_IN_BLOCK(slot_num) ((slot_num)%SLOT_PER_BLOCK)
public:
  SlotCache(const std::string &slotpage_fname) {
    this->slotpage_fname = slotpage_fname;
//    this->slotpage_handler = std::fstream(slotpage_fname, std::ios::in | std::ios::out);
    this->slotpage_handler = std::fstream(slotpage_fname, std::ios::in | std::ios::out);
    if (!this->slotpage_handler.is_open()) {
      // 文件不存在，尝试创建
      this->slotpage_handler = std::fstream(slotpage_fname, std::ios::out);
      this->slotpage_handler.close();
      // 重新以读写模式打开
      this->slotpage_handler = std::fstream(slotpage_fname, std::ios::in | std::ios::out);
    }
    if (!this->slotpage_handler.is_open()) {
      std::cerr << "Failed to open or create the file: " << slotpage_fname << std::endl;
      std::exit(EXIT_FAILURE);
    }
    for (auto i = 0; i < BLOCK_NUM; i++) {
      block_cache[i] = static_cast<struct slot_content*>(malloc(BLOCK_SIZE));
      access_time[i] = 0;
      info[i] = block_info();
    }
  }

  ~SlotCache() {
    for (auto i = 0; i < BLOCK_NUM; i++) {
      latches_[i].lock();
      if (info[i].is_dirty) {
        write_back_block(info[i].block_num);
      }
      latches_[i].unlock();
    }
    this->slotpage_handler.close();
  }

  /// methods only for test
  void write_for_test(char *sc, size_t bytes) {
    slotpage_handler.seekp(0);
    slotpage_handler.write(reinterpret_cast<const char*>(sc), bytes);
    slotpage_handler.flush();
  }

  void flush_all_blk() {
    for (auto blkinfo : info) {
      if (blkinfo.is_dirty) {
        write_back_block(blkinfo.block_num);
      }
    }
    this->slotpage_handler.flush();
  }

  void get_slot(size_t slot_num, struct slot_content *sc) {
    auto block_num = slotnum_hash2_blocknum(slot_num);
    auto blockcache_num = block_num % BLOCK_NUM;
    latches_[blockcache_num].lock();
    if (!info[blockcache_num].used || info[blockcache_num].block_num != block_num) {  // cache miss
      if (info[blockcache_num].is_dirty) {
        write_back_block(blockcache_num);
      }
      read_in_block(blockcache_num, block_num);
      access_time[blockcache_num] = 0;
      info[blockcache_num] = block_info(block_num, false, true);
    }
    read_slot(sc, blockcache_num, SLOT_OFFSET_IN_BLOCK(slot_num));
    access_time[blockcache_num]++;
    latches_[blockcache_num].unlock();
  }

  void set_slot(size_t slot_num, struct slot_content *sc) {
    auto block_num = slotnum_hash2_blocknum(slot_num);
    auto blockcache_num = block_num % BLOCK_NUM;
    latches_[blockcache_num].lock();
    if (!info[blockcache_num].used || info[blockcache_num].block_num != block_num) {
      if (info[blockcache_num].is_dirty) {
        write_back_block(blockcache_num);
      }
      read_in_block(blockcache_num, block_num);
      access_time[blockcache_num] = 0;
      info[blockcache_num] = block_info(block_num, false, true);
    }
    set_slot(sc, blockcache_num, SLOT_OFFSET_IN_BLOCK(slot_num));
    access_time[blockcache_num]++;
    info[blockcache_num].is_dirty = true;
    latches_[blockcache_num].unlock();
  }

private:
 struct block_info {
  bool used;
  size_t block_num;
  bool is_dirty;
  block_info() { used = false; is_dirty = false; }
  block_info(size_t block_num, bool is_dirty=false, bool used = false) {
    this->block_num = block_num;
    this->is_dirty = is_dirty;
    this->used = used;
  }
};

private:
  inline static size_t slotnum_hash2_blocknum(size_t slot_num) {
    return slot_num * sizeof(slot_content) / BLOCK_SIZE;
  }
  void write_back_block(size_t blockcache_num) {
    auto write_pos = info[blockcache_num].block_num * BLOCK_SIZE;
    slotpage_handler.seekp(write_pos);
    slotpage_handler.write(
        reinterpret_cast<const char*>(block_cache[blockcache_num]), BLOCK_SIZE);
    slotpage_handler.flush();
  }
  void read_in_block(size_t blockcache_num, size_t block_num) {
    slotpage_handler.seekp(block_num*BLOCK_SIZE);
    slotpage_handler.read(reinterpret_cast<char*>(block_cache[blockcache_num]), BLOCK_SIZE);
  }
  inline void read_slot(struct slot_content *sc, size_t blockcache_num, size_t slot_num_offset) {
    auto src = &block_cache[blockcache_num][slot_num_offset];
    memcpy(sc, src, sizeof(slot_content));
  }
  inline void set_slot(struct slot_content *sc, size_t blockcache_num, size_t slot_num_offset) {
    auto src = &block_cache[blockcache_num][slot_num_offset];
    memcpy(src, sc, sizeof(slot_content));
  }

private:
  std::string slotpage_fname;
  std::fstream slotpage_handler;
  std::mutex latches_[BLOCK_NUM];
  struct slot_content *block_cache[BLOCK_NUM];
  size_t access_time[BLOCK_NUM];
  struct block_info info[BLOCK_NUM];
};

/// test passed
class BitMap {
  // in memory bitmap
public:
  BitMap() {
    char *bitmap = static_cast<char*>(malloc(BITMAP_SIZE*sizeof(char)));
    memset(bitmap, 0, BITMAP_SIZE * sizeof(char));
    bitmaps_.push_back(bitmap);
    size = BITMAP_SIZE;
    first_empty_slot = 0;
  }

  BitMap(const std::string &dbname) {
    name = dbname + "_bitmap";
    auto bitmap_handler = std::fstream(name, std::ios::in | std::ios::out);
    if (!bitmap_handler.is_open()) {
      // 文件不存在，尝试创建
      bitmap_handler = std::fstream(name, std::ios::out);
      bitmap_handler.close();
      // 重新以读写模式打开
      bitmap_handler = std::fstream(name, std::ios::in | std::ios::out);
      bitmap_handler.seekp(0);
      size_t tmp[2] = {2*sizeof(size_t), 0};
      bitmap_handler.write(reinterpret_cast<const char*>(tmp), sizeof(tmp));
      bitmap_handler.close();

      // init bitmap
      char *bitmap = static_cast<char*>(malloc(BITMAP_SIZE*sizeof(char)));
      memset(bitmap, 0, BITMAP_SIZE * sizeof(char));
      bitmaps_.push_back(bitmap);
      size = BITMAP_SIZE;
      first_empty_slot = 0;
    } else {
      // read in bitmap content
      size_t bitmap_header[2];
      bitmap_handler.seekp(0);
      bitmap_handler.read(reinterpret_cast<char*>(bitmap_header), 2*sizeof(size_t));
      this->first_empty_slot = bitmap_header[1];
      this->size = bitmap_header[0];

      const size_t page_num = (size - 2*sizeof(size_t)) / BITMAP_SIZE;
      for (auto i = 0; i < page_num; i++) {
        char *bitmap = static_cast<char*>(malloc(BITMAP_SIZE*sizeof(char)));
        bitmap_handler.seekp(2*sizeof(size_t) + i*BITMAP_SIZE);
        bitmap_handler.read(bitmap, BITMAP_SIZE);
        bitmaps_.push_back(bitmap);
      }

      // init bitmap
      if (page_num == 0) {
        char *bitmap = static_cast<char*>(malloc(BITMAP_SIZE*sizeof(char)));
        memset(bitmap, 0, BITMAP_SIZE * sizeof(char));
        bitmaps_.push_back(bitmap);
        size = BITMAP_SIZE;
        first_empty_slot = 0;
      }
    }
  }

  ~BitMap() {
    auto bitmap_handler = std::fstream(name, std::ios::in | std::ios::out);
    assert(bitmap_handler.is_open());
    size_t tmp[2] = {size, first_empty_slot};
    bitmap_handler.seekp(0);
    bitmap_handler.write(reinterpret_cast<const char*>(tmp), sizeof(tmp));
    size_t off = 2 * sizeof(size_t);
    for (auto bitmap : bitmaps_) {
      bitmap_handler.seekp(off);
      bitmap_handler.write(bitmap, BITMAP_SIZE);
      off += BITMAP_SIZE;
    }
    assert(off == size);
    bitmap_handler.flush();
  }

  /** methods for test **/
  void show_allocated_slot() {
    for (int i = 0; i < this->size; i++) {
      auto byte = *get_bitmap_byte(i);
      for (int b = 0; b < sizeof(byte) << 3; b++) {
        if (byte & 0x80) {
          std::cout << (i<<3)+b << ' ';
        }
        byte <<= 1;
      }
//      std::cout << std::endl;
    }
//    std::cout << std::endl;
  }

  void dealloc_slot(size_t slot_num) {
    mtx.lock();
    const size_t byte = slot2byte(slot_num);
    const size_t off  = slot2offset(slot_num);
    char *target_byte = get_bitmap_byte(byte);
    assert(*target_byte & POSMASK(off));
    RESETBIT(target_byte, off);
//    set_bitmap_byte(byte, target_byte);
    first_empty_slot = first_empty_slot < slot_num ? first_empty_slot:slot_num;
    mtx.unlock();
  }

  size_t alloc_slot() {
    mtx.lock();
    size_t target_slot = first_empty_slot;
    char *start_byte = get_bitmap_byte(slot2byte(first_empty_slot));
    const size_t off = slot2offset(first_empty_slot);
    SETBIT(start_byte, off);
    // find the next free slot
    if (HASFREESLOT(*start_byte)) {
      auto bit_off = find_first_free_slot_inbyte(*start_byte);
      first_empty_slot += bit_off - off;
      if (slot2byte(first_empty_slot) >= size) {
        alloc_new_bitmap();
      }
    } else {
      size_t i;
      for (i = slot2byte(first_empty_slot)+1; i < size; i++) {
        char *byte  = get_bitmap_byte(i);
        if (HASFREESLOT(*byte)) {
          // FIXME: pack four bytes to do free slot finding
          auto bit_off = find_first_free_slot_inbyte(*byte);
          first_empty_slot = byte2slot(i) + bit_off;
          break;
        }
      }
      // scale the bitmap
      if (i >= size) {
        alloc_new_bitmap();
//        char *byte = get_bitmap_byte(i);
//        SETBIT(byte, 0);
        first_empty_slot = byte2slot(i) + 1;
      }
    }
    mtx.unlock();
    return target_slot;
  }

private:
  static inline size_t slot2byte(size_t slot_num) { return slot_num / BITS_PER_BYTE; }
  static inline size_t byte2slot(size_t byte_num) { return byte_num * BITS_PER_BYTE; }
  static inline size_t slot2offset(size_t slot_num) { return slot_num % BITS_PER_BYTE; }
  inline char *get_bitmap_byte(size_t byte_num) {
    char *bitmap = bitmaps_.at(byte_num / BITMAP_SIZE);
    return &bitmap[byte_num % BITMAP_SIZE];
  }
//  inline void set_bitmap_byte(size_t byte_num, char byte) {
//    char *bitmap = bitmaps_.at(byte_num / BITMAP_SIZE);
//    bitmap[byte_num % BITMAP_SIZE] = byte;
//  }
  static inline size_t find_first_free_slot_inbyte(uint8_t byte) {
    uint32_t rbyte = reverse_bits(byte);  // 使用 32 位变量
    return __builtin_ctz(~rbyte);  // 使用 GCC 内建函数找到最低位 1 的位置
  }
  static inline uint8_t reverse_bits(uint8_t byte) {
    uint8_t reversed = 0;

    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 1st bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 2nd bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 3rd bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 4th bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 5th bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 6th bit
    reversed = (reversed << 1) | (byte & 1); byte >>= 1; // 7th bit
    reversed = (reversed << 1) | (byte & 1);            // 8th bit (final step)

    return reversed;
  }
  void alloc_new_bitmap() {
    char *bitmap = static_cast<char*>(malloc(BITMAP_SIZE*sizeof(char)));
    memset(bitmap, 0, BITMAP_SIZE * sizeof(char));
    bitmaps_.push_back(bitmap);
    size += BITMAP_SIZE;
  }

private:
  std::string name;
  std::vector<char *> bitmaps_;
  size_t size;
  size_t first_empty_slot;
  std::mutex mtx;
};


//SlotCache * get_slotcache(const std::string &dbname);

// read and write slot in disk
// first version with no cache
class SlotPage {
public:
  SlotPage(const std::string &dbname) {
    slotpage_fname = slotpage_handler_name(dbname);
    bitmap = new BitMap(dbname);
    slotcache = new SlotCache(slotpage_fname);
    assert(slotcache);
  }

public:
  void get_slot(size_t slot_num, struct slot_content *sc) {
    slotcache->get_slot(slot_num, sc);
  }
  void set_slot(size_t slot_num, struct slot_content *sc) {
    slotcache->set_slot(slot_num, sc);
  }
  size_t alloc_slot() { return bitmap->alloc_slot(); }
  void dealloc_slot(size_t slot_num) { bitmap->dealloc_slot(slot_num); }

private:
  static std::string slotpage_handler_name(const std::string &dbname) {
    return dbname + "_slotpage";
  }
  inline static size_t slotpage_pos(size_t slot_num) {
    return slot_num * sizeof(slot_content);
  }

private:
  std::string slotpage_fname;
  BitMap *bitmap;
//  static SlotCache *slotcache;
  SlotCache *slotcache;
};


//#ifndef ENABLE_SLOT_CACHES
//#define ENABLE_SLOT_CACHES
//std::unordered_map<std::string, SlotCache *> slotcaches_;
//#endif
//
//SlotCache * get_slotcache(const std::string &dbname) {
//  if (slotcaches_.find(dbname) == slotcaches_.end()) {
//    slotcaches_[dbname] = new SlotCache(dbname);
//  }
//  return slotcaches_[dbname];
//}

#endif  // LEVELDB_SLOTPAGE_H