// Copyright 2010-2013 RethinkDB, all rights reserved.

#ifndef BUFFER_CACHE_SEMANTIC_CHECKING_HPP_

#define BUFFER_CACHE_SEMANTIC_CHECKING_HPP_

#include <utility>

#include "utils.hpp"

#include <boost/crc.hpp>

#include "buffer_cache/mirrored/config.hpp"

#include "buffer_cache/types.hpp"

#include "concurrency/coro_fifo.hpp"

#include "concurrency/fifo_checker.hpp"

#include "concurrency/rwi_lock.hpp"

#include "containers/scoped.hpp"

#include "perfmon/types.hpp"

#include "repli_timestamp.hpp"

// TODO: Have the semantic checking cache make sure that the

// repli_timestamp_ts are correct.

/* The semantic-checking cache (scc_cache_t) is a wrapper around another cache that will

make sure that the inner cache obeys the proper semantics. */

template<class inner_cache_t> class scc_buf_lock_t;

template<class inner_cache_t> class scc_transaction_t;

template<class inner_cache_t> class scc_cache_t;

typedef uint32_t crc_t;

class serializer_t;

/* Buf */

template<class inner_cache_t>

class scc_buf_lock_t {

public:

scc_buf_lock_t(

scc_transaction_t<inner_cache_t> *txn, block_id_t block_id, access_t mode,

buffer_cache_order_mode_t order_mode = buffer_cache_order_mode_check,

lock_in_line_callback_t *call_when_in_line = 0);

explicit scc_buf_lock_t(scc_transaction_t<inner_cache_t> *txn);

scc_buf_lock_t();

~scc_buf_lock_t();

void swap(scc_buf_lock_t<inner_cache_t> &swapee);

void release();

void release_if_acquired();

block_id_t get_block_id() const;

const void *get_data_read() const;

void *get_data_write();

void *get_data_write(uint32_t cache_block_size);

void mark_deleted();

void touch_recency(repli_timestamp_t timestamp);

bool is_acquired() const;

bool is_deleted() const;

repli_timestamp_t get_recency() const;

private:

bool snapshotted;

bool has_been_changed;

scoped_ptr_t<typename inner_cache_t::buf_lock_type> internal_buf_lock;

scc_cache_t<inner_cache_t> *cache;

private:

crc_t compute_crc() {

boost::crc_optimal<32, 0x04C11DB7, 0xFFFFFFFF, 0xFFFFFFFF, true, true> crc_computer;

crc_computer.process_bytes(internal_buf_lock->get_data_read(), internal_buf_lock->cache_block_size());

return crc_computer.checksum();

}

public:

eviction_priority_t get_eviction_priority() {

return internal_buf_lock->get_eviction_priority();

}

void set_eviction_priority(eviction_priority_t val) {

internal_buf_lock->set_eviction_priority(val);

}

};

/* Transaction */

template<class inner_cache_t>

class scc_transaction_t :

public home_thread_mixin_debug_only_t

{

public:

scc_transaction_t(scc_cache_t<inner_cache_t> *cache, access_t access, int expected_change_count, repli_timestamp_t recency_timestamp, order_token_t _order_token, write_durability_t durability);

scc_transaction_t(scc_cache_t<inner_cache_t> *cache, access_t access, order_token_t order_token);

~scc_transaction_t();

// TODO: Implement semantic checking for snapshots!

void snapshot() {

snapshotted = true;

inner_transaction.snapshot();

}

access_t get_access() const {

return inner_transaction.get_access();

}

write_durability_t get_durability() const {

return inner_transaction.get_durability();

}

void set_account(typename inner_cache_t::cache_account_type *cache_account);

void get_subtree_recencies(block_id_t *block_ids, size_t num_block_ids, repli_timestamp_t *recencies_out, get_subtree_recencies_callback_t *cb);

scc_cache_t<inner_cache_t> *get_cache() const { return cache; }

scc_cache_t<inner_cache_t> *cache;

const order_token_t order_token;

void set_token_pair(write_token_pair_t *token_pair) {

inner_transaction.set_token_pair(token_pair);

}

private:

bool snapshotted; // Disables CRC checks

friend class scc_buf_lock_t<inner_cache_t>;

friend class scc_cache_t<inner_cache_t>;

access_t access;

typename inner_cache_t::transaction_type inner_transaction;

};

/* Cache */

template<class inner_cache_t>

class scc_cache_t : public home_thread_mixin_debug_only_t, public serializer_read_ahead_callback_t {

public:

typedef scc_buf_lock_t<inner_cache_t> buf_lock_type;

typedef scc_transaction_t<inner_cache_t> transaction_type;

typedef typename inner_cache_t::cache_account_type cache_account_type;

static void create(serializer_t *serializer);

scc_cache_t(serializer_t *serializer,

const mirrored_cache_config_t &dynamic_config,

perfmon_collection_t *parent);

block_size_t get_block_size();

void create_cache_account(

int priority,

scoped_ptr_t<typename inner_cache_t::cache_account_type> *out);

void offer_read_ahead_buf(block_id_t block_id,

scoped_malloc_t<ser_buffer_t> *buf,

const counted_t<standard_block_token_t> &token,

repli_timestamp_t recency_timestamp);

bool contains_block(block_id_t block_id);

unsigned int num_blocks();

coro_fifo_t &co_begin_coro_fifo() { return inner_cache.co_begin_coro_fifo(); }

private:

inner_cache_t inner_cache;

private:

friend class scc_transaction_t<inner_cache_t>;

friend class scc_buf_lock_t<inner_cache_t>;

/* CRC checking stuff */

two_level_array_t<crc_t> crc_map;

/* order checking stuff */

two_level_nevershrink_array_t<plain_sink_t> sink_map;

};

#include "buffer_cache/semantic_checking.tcc"

#endif /* BUFFER_CACHE_SEMANTIC_CHECKING_HPP_ */