/** @file

  A brief file description

  @section license License

  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 _P_CACHE_INTERNAL_H__
#define _P_CACHE_INTERNAL_H__

#include "libts.h"

#ifdef HTTP_CACHE
#include "HTTP.h"
#include "P_CacheHttp.h"
#endif

struct EvacuationBlock;

// Compilation Options

#define HIT_EVACUATE                    1
#define ALTERNATES                      1
// #define CACHE_LOCK_FAIL_RATE         0.001
// #define CACHE_AGG_FAIL_RATE          0.005
// #define CACHE_INSPECTOR_PAGES
#define MAX_CACHE_VCS_PER_THREAD        500

#define INTEGRAL_FRAGS                  4

#ifdef CACHE_INSPECTOR_PAGES
#ifdef DEBUG
#define CACHE_STAT_PAGES
#endif
#endif

#ifdef DEBUG
#define DDebug Debug
#else
#define DDebug if (0) dummy_debug
#endif

#define AIO_SOFT_FAILURE                -100000
// retry read from writer delay
#define WRITER_RETRY_DELAY  HRTIME_MSECONDS(50)

#define CACHE_READY(_x) (CacheProcessor::cache_ready & (1 << (_x)))

#ifndef CACHE_LOCK_FAIL_RATE
#define CACHE_TRY_LOCK(_l, _m, _t) MUTEX_TRY_LOCK(_l, _m, _t)
#else
#define CACHE_TRY_LOCK(_l, _m, _t)                             \
  MUTEX_TRY_LOCK(_l, _m, _t);                                  \
  if ((uint32_t)_t->generator.random() <                         \
     (uint32_t)(UINT_MAX *CACHE_LOCK_FAIL_RATE))                 \
    CACHE_MUTEX_RELEASE(_l)
#endif


#define VC_LOCK_RETRY_EVENT() \
  do { \
    trigger = mutex->thread_holding->schedule_in_local(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay), event); \
    return EVENT_CONT; \
  } while (0)

#define VC_SCHED_LOCK_RETRY() \
  do { \
    trigger = mutex->thread_holding->schedule_in_local(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay)); \
    return EVENT_CONT; \
  } while (0)

#define CONT_SCHED_LOCK_RETRY_RET(_c) \
  do { \
    _c->mutex->thread_holding->schedule_in_local(_c, HRTIME_MSECONDS(cache_config_mutex_retry_delay)); \
    return EVENT_CONT; \
  } while (0)

#define CONT_SCHED_LOCK_RETRY(_c) \
  _c->mutex->thread_holding->schedule_in_local(_c, HRTIME_MSECONDS(cache_config_mutex_retry_delay))

#define VC_SCHED_WRITER_RETRY() \
  do { \
    ink_assert(!trigger); \
    writer_lock_retry++; \
    ink_hrtime _t = WRITER_RETRY_DELAY; \
    if (writer_lock_retry > 2) \
      _t = WRITER_RETRY_DELAY * 2; \
    else if (writer_lock_retry > 5) \
      _t = WRITER_RETRY_DELAY * 10; \
    else if (writer_lock_retry > 10) \
      _t = WRITER_RETRY_DELAY * 100; \
    trigger = mutex->thread_holding->schedule_in_local(this, _t); \
    return EVENT_CONT; \
  } while (0)

#define VC_SCHED_RWW_WAIT_TIMEOUT(vc, v) \
  vc->trigger = vc->mutex->thread_holding->schedule_in_local(vc, HRTIME_MSECONDS(v))
  // cache stats definitions
enum
{
  cache_bytes_used_stat,
  cache_bytes_total_stat,
  cache_ram_cache_bytes_stat,
  cache_ram_cache_bytes_total_stat,
  cache_direntries_total_stat,
  cache_direntries_used_stat,
  cache_ram_cache_hits_stat,
  cache_ram_cache_misses_stat,
  cache_pread_count_stat,
  cache_percent_full_stat,
  cache_lookup_active_stat,
  cache_lookup_success_stat,
  cache_lookup_failure_stat,
  cache_read_active_stat,
  cache_read_success_stat,
  cache_read_failure_stat,
#ifdef SSD_CACHE
  cache_ssd_read_success_stat,
  cache_sas_read_success_stat,
  cache_ram_read_success_stat,
#endif
  cache_write_active_stat,
  cache_write_success_stat,
  cache_write_failure_stat,
  cache_write_backlog_failure_stat,
  cache_update_active_stat,
  cache_update_success_stat,
  cache_update_failure_stat,
  cache_remove_active_stat,
  cache_remove_success_stat,
  cache_remove_failure_stat,
  cache_evacuate_active_stat,
  cache_evacuate_success_stat,
  cache_evacuate_failure_stat,
  cache_scan_active_stat,
  cache_scan_success_stat,
  cache_scan_failure_stat,
  cache_directory_collision_count_stat,
  cache_single_fragment_document_count_stat,
  cache_two_fragment_document_count_stat,
  cache_three_plus_plus_fragment_document_count_stat,
  cache_read_busy_success_stat,
  cache_read_busy_failure_stat,
  cache_gc_bytes_evacuated_stat,
  cache_gc_frags_evacuated_stat,
  cache_write_bytes_stat,
  cache_hdr_vector_marshal_stat,
  cache_hdr_marshal_stat,
  cache_hdr_marshal_bytes_stat,
  cache_stat_count
};


extern RecRawStatBlock *cache_rsb;

#define GLOBAL_CACHE_SET_DYN_STAT(x,y) \
	RecSetGlobalRawStatSum(cache_rsb, (x), (y))

#define CACHE_SET_DYN_STAT(x,y) \
	RecSetGlobalRawStatSum(cache_rsb, (x), (y)) \
	RecSetGlobalRawStatSum(vol->cache_vol->vol_rsb, (x), (y))

#define CACHE_INCREMENT_DYN_STAT(x) \
	RecIncrRawStat(cache_rsb, mutex->thread_holding, (int) (x), 1); \
	RecIncrRawStat(vol->cache_vol->vol_rsb, mutex->thread_holding, (int) (x), 1);

#define CACHE_DECREMENT_DYN_STAT(x) \
	RecIncrRawStat(cache_rsb, mutex->thread_holding, (int) (x), -1); \
	RecIncrRawStat(vol->cache_vol->vol_rsb, mutex->thread_holding, (int) (x), -1);

#define CACHE_VOL_SUM_DYN_STAT(x,y) \
        RecIncrRawStat(vol->cache_vol->vol_rsb, mutex->thread_holding, (int) (x), (int64_t) y);

#define CACHE_SUM_DYN_STAT(x, y) \
	RecIncrRawStat(cache_rsb, mutex->thread_holding, (int) (x), (int64_t) (y)); \
	RecIncrRawStat(vol->cache_vol->vol_rsb, mutex->thread_holding, (int) (x), (int64_t) (y));

#define CACHE_SUM_DYN_STAT_THREAD(x, y) \
	RecIncrRawStat(cache_rsb, this_ethread(), (int) (x), (int64_t) (y)); \
	RecIncrRawStat(vol->cache_vol->vol_rsb, this_ethread(), (int) (x), (int64_t) (y));

#define GLOBAL_CACHE_SUM_GLOBAL_DYN_STAT(x, y) \
	RecIncrGlobalRawStatSum(cache_rsb,(x),(y))

#define CACHE_SUM_GLOBAL_DYN_STAT(x, y) \
	RecIncrGlobalRawStatSum(cache_rsb,(x),(y)) \
	RecIncrGlobalRawStatSum(vol->cache_vol->vol_rsb,(x),(y))

#define CACHE_CLEAR_DYN_STAT(x) \
do { \
	RecSetRawStatSum(cache_rsb, (x), 0); \
	RecSetRawStatCount(cache_rsb, (x), 0); \
	RecSetRawStatSum(vol->cache_vol->vol_rsb, (x), 0); \
	RecSetRawStatCount(vol->cache_vol->vol_rsb, (x), 0); \
} while (0);

// Configuration
extern int cache_config_dir_sync_frequency;
extern int cache_config_http_max_alts;
extern int cache_config_permit_pinning;
extern int cache_config_select_alternate;
extern int cache_config_vary_on_user_agent;
extern int cache_config_max_doc_size;
extern int cache_config_min_average_object_size;
extern int cache_config_agg_write_backlog;
extern int cache_config_enable_checksum;
extern int cache_config_alt_rewrite_max_size;
extern int cache_config_read_while_writer;
extern char cache_system_config_directory[PATH_NAME_MAX + 1];
extern int cache_clustering_enabled;
extern int cache_config_agg_write_backlog;
extern int cache_config_ram_cache_compress;
extern int cache_config_ram_cache_compress_percent;
extern int cache_config_ram_cache_use_seen_filter;
#ifdef HIT_EVACUATE
extern int cache_config_hit_evacuate_percent;
extern int cache_config_hit_evacuate_size_limit;
#endif
extern int cache_config_force_sector_size;
extern int cache_config_target_fragment_size;
extern int cache_config_mutex_retry_delay;
extern int cache_config_rww_max_delay;
extern int cache_rww_max_doc_size;

#ifdef HTTP_CACHE
extern int enable_cache_empty_http_doc;
#endif
#ifdef SSD_CACHE
extern int good_ssd_disks;
extern int64_t transistor_range_threshold;
#endif

struct CacheWriterTable;
struct CacheWriterEntry;

struct CacheVC: public CacheVConnection
{
  CacheVC();

//  bool get_entry(CacheWriterTable *table);
  bool add_entry(CacheWriterTable *table);
  void clear_entry(CacheWriterTable *table);

  VIO *do_io_read(Continuation *c, int64_t nbytes, MIOBuffer *buf);
  VIO *do_io_pread(Continuation *c, int64_t nbytes, MIOBuffer *buf, int64_t offset);
  VIO *do_io_write(Continuation *c, int64_t nbytes, IOBufferReader *buf, bool owner = false);
  void do_io_close(int lerrno = -1);
  void reenable(VIO *avio);
  void reenable_re(VIO *avio);
  bool get_data(int i, void *data);
  bool set_data(int i, void *data);

  bool is_ram_cache_hit()
  {
    ink_assert(vio.op == VIO::READ);
    return !f.not_from_ram_cache;
  }
  int get_header(void **ptr, int *len)
  {
    if (first_buf.m_ptr) {
      Doc *doc = (Doc*)first_buf->data();
      *ptr = doc->hdr();
      *len = doc->hlen;
      return 0;
    } else
      return -1;
  }
  int set_header(void *ptr, int len)
  {
    header_to_write = ptr;
    header_to_write_len = len;
    return 0;
  }
  int get_single_data(void **ptr, int *len)
  {
    if (first_buf.m_ptr) {
      Doc *doc = (Doc*)first_buf->data();
      if (doc->data_len() == doc->total_len) {
        *ptr = doc->data();
        *len = doc->data_len();
        return 0;
      }
    }
    return -1;
  }

  bool writer_done();
  int calluser(int event);
  int callcont(int event);
  int die();
  int dead(int event, Event *e);

  int handleReadDone(int event, Event *e);
  int handleRead(int event, Event *e);
  int do_read_call(CacheKey *akey);
  int handleWrite(int event, Event *e);
  int handleWriteLock(int event, Event *e);
  int do_write_call();
  int do_write_lock();
  int do_write_lock_call();
  int do_sync(uint32_t target_write_serial);

  int openReadClose(int event, Event *e);
  int openReadReadDone(int event, Event *e);
  int openReadMain(int event, Event *e);
  int openReadStartEarliest(int event, Event *e);
#ifdef HTTP_CACHE
  int openReadVecWrite(int event, Event *e);
#endif
  int openReadStartHead(int event, Event *e);
  int openReadFromWriter(int event, Event *e);
  int openReadFromWriterHead(int event, Event *e);
//  int openReadFromWriterData(int event, Event *e);
  int handleReadFromWriter(int event, Event *e);
  int openReadFromWriterMain(int event, Event *e);
  int openReadFromWriterFailure(int event, Event *);
  int openReadChooseWriter(int event, Event *e);

  int openWriteCloseDir(int event, Event *e);
  int openWriteCloseHeadDone(int event, Event *e);
  int openWriteCloseHead(int event, Event *e);
  int openWriteCloseDataDone(int event, Event *e);
  int openWriteClose(int event, Event *e);
  int openWriteRemoveVector(int event, Event *e);
  int openWriteWriteDone(int event, Event *e);
  int openWriteOverwrite(int event, Event *e);
  int openWriteMain(int event, Event *e);
  int openWriteStartDone(int event, Event *e);
  int openWriteStartBegin(int event, Event *e);

  int updateVector(int event, Event *e);
  int updateReadDone(int event, Event *e);
  int updateVecWrite(int event, Event *e);

  int removeEvent(int event, Event *e);

  int linkWrite(int event, Event *e);
  int derefRead(int event, Event *e);

  int scanVol(int event, Event *e);
  int scanObject(int event, Event *e);
  int scanUpdateDone(int event, Event *e);
  int scanOpenWrite(int event, Event *e);
  int scanRemoveDone(int event, Event *e);

  int is_io_in_progress()
  {
    return io.aiocb.aio_fildes != AIO_NOT_IN_PROGRESS;
  }
  void set_io_not_in_progress()
  {
    io.aiocb.aio_fildes = AIO_NOT_IN_PROGRESS;
  }
  void set_agg_write_in_progress()
  {
    io.aiocb.aio_fildes = AIO_AGG_WRITE_IN_PROGRESS;
  }
  void set_read_while_write_in_progress()
  {
    io.aiocb.aio_fildes = READ_WHILE_WRITE_IN_PROGRESS;
  }
  int evacuateDocDone(int event, Event *e);
  int evacuateReadHead(int event, Event *e);

  void cancel_trigger();
  virtual int64_t get_object_size();
#ifdef HTTP_CACHE
  virtual void set_http_info(CacheHTTPInfo *info);
  virtual void get_http_info(CacheHTTPInfo ** info);
#endif
  virtual bool set_pin_in_cache(time_t time_pin);
  virtual time_t get_pin_in_cache();
  virtual bool set_disk_io_priority(int priority);
  virtual int get_disk_io_priority();
  virtual bool is_read_from_writer() {
    return f.read_from_writer_called;
  }
  virtual bool is_pread_capable() {
    return !f.read_from_writer_called;
  }

  // offsets from the base stat
#define CACHE_STAT_ACTIVE  0
#define CACHE_STAT_SUCCESS 1
#define CACHE_STAT_FAILURE 2

  // number of bytes to memset to 0 in the CacheVC when we free
  // it. All member variables starting from vio are memset to 0.
  // This variable is initialized in CacheVC constructor.
  static int size_to_init;

  // Start Region A
  // This set of variables are not reset when the cacheVC is freed.
  // A CacheVC must set these to the correct values whenever needed
  // These are variables that are always set to the correct values
  // before being used by the CacheVC
  CacheKey key, first_key, earliest_key, update_key;
  Dir dir, earliest_dir, overwrite_dir, first_dir;
  // end Region A

  // Start Region B
  // These variables are individually cleared or reset when the
  // CacheVC is freed. All these variables must be reset/cleared
  // in free_CacheVC.
  Action _action;
#ifdef HTTP_CACHE
  CacheHTTPHdr request;
#endif
  CacheHTTPInfoVector vector;
  CacheHTTPInfo alternate;
  Ptr<IOBufferData> buf;
  Ptr<IOBufferData> first_buf;
  Ptr<IOBufferBlock> blocks; // data available to write
  Ptr<IOBufferBlock> writer_buf;

  Ptr<CacheWriterEntry> cw;

  OpenDirEntry *od;
  AIOCallbackInternal io;
  int alternate_index;          // preferred position in vector
  LINK(CacheVC, opendir_link);
  LINK(CacheVC, signal_link);
#ifdef CACHE_STAT_PAGES
  LINK(CacheVC, stat_link);
#endif
  // end Region B

  // Start Region C
  // These variables are memset to 0 when the structure is freed.
  // The size of this region is size_to_init which is initialized
  // in the CacheVC constuctor. It assumes that vio is the start
  // of this region.
  // NOTE: NOTE: NOTE: If vio is NOT the start, then CHANGE the
  // size_to_init initialization
  VIO vio;
  EThread *initial_thread;  // initial thread open_XX was called on
  CacheFragType frag_type;
  CacheHTTPInfo *info;
  CacheHTTPInfoVector *write_vector;
#ifdef HTTP_CACHE
  CacheLookupHttpConfig *params;
#endif
  int header_len;       // for communicating with agg_copy
  uint32_t frag_len;         //
  uint32_t write_len;     // for communicating with agg_copy
  uint32_t agg_len;       // for communicating with aggWrite
  uint32_t write_serial;  // serial of the final write for SYNC
  Vol *vol;
  Dir *last_collision;
  Event *trigger;
  CacheKey *read_key;
  ContinuationHandler save_handler;
  uint32_t pin_in_cache;
  ink_hrtime start_time;
  int base_stat;
  int recursive;
  int closed;
  int64_t seek_to;                // pread offset
  int64_t offset;                 // offset into 'blocks' of data to write
  int64_t writer_offset;          // offset of the writer for reading from a writer
  int64_t length;                 // length of data available to write
  int64_t doc_pos;                // read position in 'buf'
  uint64_t write_pos;             // length written
  uint64_t total_len;             // total length written and available to write
  uint64_t doc_len;               // total_length (of the selected alternate for HTTP)
  uint64_t update_len;
  int fragment;
  int scan_msec_delay;
  CacheVC *write_vc;
  char *hostname;
  int host_len;
  int header_to_write_len;
  void *header_to_write;
  short writer_lock_retry;
#ifdef SSD_CACHE
  SSDVol *ssd_vol;
  MigrateToSSD *mts;
  uint64_t dir_off;
#endif
  union
  {
    uint32_t flags;
    struct
    {
      unsigned int use_first_key:1;
      unsigned int overwrite:1; // overwrite first_key Dir if it exists
      unsigned int close_complete:1; // WRITE_COMPLETE is final
      unsigned int sync:1; // write to be committed to durable storage before WRITE_COMPLETE
      unsigned int evacuator:1;
      unsigned int single_fragment:1;
      unsigned int evac_vector:1;
      unsigned int lookup:1;
      unsigned int update:1;
      unsigned int remove:1;
      unsigned int remove_aborted_writers:1;
      unsigned int open_read_timeout:1; // UNUSED
      unsigned int data_done:1;
      unsigned int read_from_writer_called:1;
      unsigned int not_from_ram_cache:1;        // entire object was from ram cache
      unsigned int rewrite_resident_alt:1;
      unsigned int readers:1;
      unsigned int doc_from_ram_cache:1;
#ifdef HIT_EVACUATE
      unsigned int hit_evacuate:1;
#endif
#ifdef HTTP_CACHE
      unsigned int force_empty:1; // used for cache empty http document
#endif
#ifdef SSD_CACHE
      unsigned int read_from_ssd:1;
      unsigned int write_into_ssd:1;
      unsigned int ram_fixup:1;
      unsigned int transistor:1;
#endif
    } f;
  };
  // BTF optimization used to skip reading stuff in cache partition that doesn't contain any
  // dir entries.
  char *scan_vol_map; 
  // BTF fix to handle objects that overlapped over two different reads,
  // this is how much we need to back up the buffer to get the start of the overlapping object.
  off_t scan_fix_buffer_offset;
  //end region C
};

#define PUSH_HANDLER(_x) do {                                           \
    save_handler = handler; handler = (ContinuationHandler)(_x);        \
} while (0)

#define POP_HANDLER do {                                          \
    handler = save_handler;                                       \
  } while (0)

struct CacheSignaler: public Continuation
{
  Action action;
  Ptr<IOBufferBlock> blocks;
  int size;

  CacheSignaler(): size(0) {
    SET_HANDLER(&CacheSignaler::mainEvent);
  }

  int mainEvent(int event, void *e);
};

extern ClassAllocator<CacheSignaler> cacheSignalerAllocator;

inline int
CacheSignaler::mainEvent(int event, void *e) {
  action.continuation->handleEvent(event, this);
  mutex.clear();
  action.mutex.clear();
  blocks = NULL;
  cacheSignalerAllocator.free(this);
  return EVENT_DONE;
}

struct CacheWriterEntry: public RefCountObj
{
  ink_mutex *mutex;
  CacheKey key; // for writer set
  MIOBuffer buffer; // for writer set
  LINK(CacheWriterEntry, link);
  CacheHTTPInfo alternate; // for writer set
  IOBufferReader *r;
  CacheVC *writer; // backpointer to write_vc;
  int64_t doc_len;
  int64_t total_len;
  int writer_closed;
  bool header_only_update;
  bool not_rww;
  Ptr<IOBufferData> first_buf;
  Queue<CacheVC, CacheVC::Link_signal_link> sq_readers; // reader for signal
  enum WriterSignalType {
    WRITER_META, WRITER_DATA, WRITER_CLOSE, WRITER_ABORT, WRITER_ALT_REMOVE
  };

  CacheWriterEntry();
  void free();
  void signal_reader(WriterSignalType type);

  void reserve_writer_head(CacheVC *vc);
  void set_writer_meta(CacheVC *vc);
  int get_writer_meta(CacheVC *vc, bool *header_only);
  int get_writer_data(CacheVC *vc);
  void add_writer_data(IOBufferReader *reader, int64_t size) {
    ink_debug_assert(reader->read_avail() >= size);
    IOBufferBlock *b = iobufferblock_clone(reader->get_current_block(),
        reader->start_offset, size);

    ink_mutex_acquire(mutex);
    buffer.append_block(b);
    total_len += size;
    signal_reader(WRITER_DATA);
    ink_mutex_release(mutex);
  }

  void set_writer_close(int close);
  bool in_and_remove(CacheVC *vc) {
    ink_debug_assert(this == vc->cw.m_ptr && vc->mutex->thread_holding == this_ethread());
    ink_assert(vc->is_io_in_progress());

    bool result;
    ink_mutex_acquire(mutex);
    result = sq_readers.in(vc);
    if (result) {
      sq_readers.remove(vc);
      vc->cw = NULL;
    }
    ink_mutex_release(mutex);
    return result;
  }
};

extern ClassAllocator<CacheWriterEntry> cacheWriterEntryAllocator;

#define CACHE_WRITER_BUCKET_SIZE  8191
struct CacheWriterBucket
{
  DLL<CacheWriterEntry> writers;
  ink_mutex the_mutex;
  CacheWriterBucket()
  {
    ink_mutex_init(&the_mutex, "WriterBucketMutex");
  }
};

struct CacheWriterTable
{
  CacheWriterBucket buckets[CACHE_WRITER_BUCKET_SIZE];
  bool probe_entry(INK_MD5 *key, Ptr<CacheWriterEntry> *_ref) {
    int indx = key->word(1) % CACHE_WRITER_BUCKET_SIZE;
    ink_mutex *mutex = &buckets[indx].the_mutex;

    ink_mutex_acquire(mutex);
    for (CacheWriterEntry *entry = buckets[indx].writers.head; entry;
        entry = entry->link.next) {
      if (*key == entry->key) {
        *_ref = entry;
        ink_mutex_release(mutex);
        return true;
      }
    }
    ink_mutex_release(mutex);
    *_ref = NULL;
    return false;
  }
};

extern CacheWriterTable writerTable;

struct CacheRemoveCont: public Continuation
{
  int event_handler(int event, void *data);

  CacheRemoveCont()
    : Continuation(NULL)
  { }
};

struct DummyClusterAIOCallback: Continuation
{

};


// Global Data
extern ClassAllocator<CacheVC> cacheVConnectionAllocator;
extern CacheKey zero_key;
extern CacheSync *cacheDirSync;
// Function Prototypes
#ifdef HTTP_CACHE
int cache_write(CacheVC *, CacheHTTPInfoVector *);
int get_alternate_index(CacheHTTPInfoVector *cache_vector, CacheKey key);
#endif
CacheVC *new_DocEvacuator(int nbytes, Vol *d);

#ifdef SSD_CACHE
extern ClassAllocator<MigrateToSSD> migrateToSSDAllocator;
#endif
// inline Functions

TS_INLINE CacheVC *
new_CacheVC(Continuation *cont)
{
  EThread *t = cont->mutex->thread_holding;
  CacheVC *c = THREAD_ALLOC(cacheVConnectionAllocator, t);
#ifdef HTTP_CACHE
  c->vector.data.data = &c->vector.data.fast_data[0];
#endif
  c->_action = cont;
  c->initial_thread = t->tt == DEDICATED ? NULL : t;
  c->mutex = cont->mutex;
  c->start_time = ink_get_hrtime();
  ink_assert(c->trigger == NULL);
  Debug("cache_new", "new %p", c);
#ifdef CACHE_STAT_PAGES
  ink_assert(!c->stat_link.next);
  ink_assert(!c->stat_link.prev);
#endif
  dir_clear(&c->dir);
  return c;
}

TS_INLINE int
free_CacheVC(CacheVC *cont)
{
  Debug("cache_free", "free %p", cont);
  ProxyMutex *mutex = cont->mutex;
  Vol *vol = cont->vol;
  if (vol) {
    CACHE_DECREMENT_DYN_STAT(cont->base_stat + CACHE_STAT_ACTIVE);
    if (cont->closed > 0) {
      CACHE_INCREMENT_DYN_STAT(cont->base_stat + CACHE_STAT_SUCCESS);
#ifdef  SSD_CACHE
      if (cont->vio.op == VIO::READ) {
        if (cont->f.doc_from_ram_cache) {
          CACHE_INCREMENT_DYN_STAT(cache_ram_read_success_stat);
        } else if (cont->f.read_from_ssd) {
          CACHE_INCREMENT_DYN_STAT(cache_ssd_read_success_stat);
        } else {
          CACHE_INCREMENT_DYN_STAT(cache_sas_read_success_stat);
        }
      }
#endif
    }                             // else abort,cancel
  }
  if (cont->cw) {
    if (cont->vio.op == VIO::WRITE) {
      cont->cw->set_writer_close(-1);
      cont->clear_entry(&writerTable);
    }
    cont->cw = NULL;
  }
  ink_debug_assert(mutex->thread_holding == this_ethread());
  if (cont->trigger)
    cont->trigger->cancel();
  ink_assert(!cont->is_io_in_progress());
  ink_assert(!cont->od);
  ink_assert(!cont->signal_link.next && !cont->signal_link.prev);
  /* calling cont->io.action = NULL causes compile problem on 2.6 solaris
     release build....wierd??? For now, null out continuation and mutex
     of the action separately */
  cont->io.action.continuation = NULL;
  cont->io.action.mutex = NULL;
  cont->io.mutex.clear();
  cont->io.aio_result = 0;
  cont->io.aiocb.aio_nbytes = 0;
  cont->io.aiocb.aio_reqprio = AIO_DEFAULT_PRIORITY;
#ifdef SSD_CACHE
  ink_assert(!cont->mts);
#endif
#ifdef HTTP_CACHE
  cont->request.reset();
  cont->vector.clear();
#endif
  cont->vio.buffer.clear();
  cont->vio.mutex.clear();
#ifdef HTTP_CACHE
  if (cont->vio.op == VIO::WRITE && cont->alternate_index == CACHE_ALT_INDEX_DEFAULT)
    cont->alternate.destroy();
  else
    cont->alternate.clear();
#endif
  cont->_action.cancelled = 0;
  cont->_action.mutex.clear();
  cont->mutex.clear();
  cont->buf.clear();
  cont->first_buf.clear();
  cont->blocks.clear();
  cont->writer_buf.clear();
  cont->alternate_index = CACHE_ALT_INDEX_DEFAULT;
  if (cont->scan_vol_map)
    ats_free(cont->scan_vol_map);
  memset((char *) &cont->vio, 0, cont->size_to_init);
#ifdef CACHE_STAT_PAGES
  ink_assert(!cont->stat_link.next && !cont->stat_link.prev);
#endif
#ifdef DEBUG
  SET_CONTINUATION_HANDLER(cont, &CacheVC::dead);
#endif
  THREAD_FREE_TO(cont, cacheVConnectionAllocator, this_ethread(), MAX_CACHE_VCS_PER_THREAD);
  return EVENT_DONE;
}

TS_INLINE int
CacheVC::calluser(int event)
{
  recursive++;
  ink_debug_assert(!vol || this_ethread() != vol->mutex->thread_holding);
  vio._cont->handleEvent(event, (void *) &vio);
  recursive--;
  if (closed) {
    die();
    return EVENT_DONE;
  }
  return EVENT_CONT;
}

TS_INLINE int
CacheVC::callcont(int event)
{
  recursive++;
  ink_debug_assert(!vol || this_ethread() != vol->mutex->thread_holding);
  _action.continuation->handleEvent(event, this);
  recursive--;
  if (closed)
    die();
  else if (vio.vc_server)
    handleEvent(EVENT_IMMEDIATE, 0);
  return EVENT_DONE;
}

TS_INLINE int
CacheVC::do_read_call(CacheKey *akey)
{
  doc_pos = 0;
  read_key = akey;
  io.aiocb.aio_nbytes = dir_approx_size(&dir);
#ifdef SSD_CACHE
  ssd_vol = NULL;
  ink_assert(mts == NULL);
  mts = NULL;
  f.write_into_ssd = 0;
  f.ram_fixup = 0;
  f.transistor = 0;
  f.read_from_ssd = dir_inssd(&dir);

  if (!f.read_from_ssd && vio.op == VIO::READ && good_ssd_disks > 0){
    vol->history.put_key(read_key);
    if (vol->history.is_hot(read_key) && !vol->migrate_probe(read_key, NULL) && !od) {
      f.write_into_ssd = 1;
    }
  }
  if (f.read_from_ssd) {
    ssd_vol = &vol->ssd_vols[dir_get_index(&dir)];
    if (vio.op == VIO::READ && vol_transistor_range_valid(ssd_vol, &dir, transistor_range_threshold) &&
        ssd_vol->cos.is_seen(dir_offset(&dir) - 1) && !vol->migrate_probe(read_key, NULL) && !od)
      f.transistor = 1;
  }
  if (f.write_into_ssd || f.transistor) {
    mts = migrateToSSDAllocator.alloc();
    mts->vc = this;
    mts->key = *read_key;
    mts->rewrite = (f.transistor == 1);
    dir_assign(&mts->dir, &dir);
    vol->set_migrate_in_progress(mts);
  }
#endif
  PUSH_HANDLER(&CacheVC::handleRead);
  return handleRead(EVENT_CALL, 0);
}

TS_INLINE int
CacheVC::do_write_call()
{
  PUSH_HANDLER(&CacheVC::handleWrite);
  return handleWrite(EVENT_CALL, 0);
}

TS_INLINE void
CacheVC::cancel_trigger()
{
  if (trigger) {
    trigger->cancel_action();
    trigger = NULL;
  }
}

TS_INLINE int
CacheVC::die()
{
  if (vio.op == VIO::WRITE) {
    cw->set_writer_close(closed);
#ifdef HTTP_CACHE
    if (f.update && total_len) {
      alternate.object_key_set(earliest_key);
    }
#endif
    if (!is_io_in_progress()) {
      SET_HANDLER(&CacheVC::openWriteClose);
      if (!recursive)
        openWriteClose(EVENT_NONE, NULL);
    }                           // else catch it at the end of openWriteWriteDone
    return EVENT_CONT;
  } else {
    if (is_io_in_progress())
      save_handler = (ContinuationHandler) & CacheVC::openReadClose;
    else {
      SET_HANDLER(&CacheVC::openReadClose);
      if (!recursive)
        openReadClose(EVENT_NONE, NULL);
    }
    return EVENT_CONT;
  }
}

TS_INLINE int
CacheVC::handleWriteLock(int event, Event *e)
{
  NOWARN_UNUSED(event);
  cancel_trigger();
  int ret = 0;
  {
    CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding);
    if (!lock) {
      set_agg_write_in_progress();
      trigger = mutex->thread_holding->schedule_in_local(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay));
      return EVENT_CONT;
    }
    ret = handleWrite(EVENT_CALL, e);
  }
  if (ret == EVENT_RETURN)
    return handleEvent(AIO_EVENT_DONE, 0);
  return EVENT_CONT;
}

TS_INLINE int
CacheVC::do_write_lock()
{
  PUSH_HANDLER(&CacheVC::handleWriteLock);
  return handleWriteLock(EVENT_NONE, 0);
}

TS_INLINE int
CacheVC::do_write_lock_call()
{
  PUSH_HANDLER(&CacheVC::handleWriteLock);
  return handleWriteLock(EVENT_CALL, 0);
}

TS_INLINE bool
CacheVC::writer_done()
{
  OpenDirEntry *cod = od;
  if (!cod)
    cod = vol->open_read(&first_key);
  CacheVC *w = (cod) ? cod->writers.head : NULL;
  // If the write vc started after the reader, then its not the
  // original writer, since we never choose a writer that started
  // after the reader. The original writer was deallocated and then
  // reallocated for the same first_key
  for (; w && (w != write_vc || w->start_time > start_time); w = (CacheVC *) w->opendir_link.next);
  if (!w)
    return true;
  return false;
}

TS_INLINE int
Vol::close_write(CacheVC *cont)
{

#ifdef CACHE_STAT_PAGES
  ink_assert(stat_cache_vcs.head);
  stat_cache_vcs.remove(cont, cont->stat_link);
  ink_assert(!cont->stat_link.next && !cont->stat_link.prev);
#endif
  return open_dir.close_write(cont);
}

// Returns 0 on success or a positive error code on failure
TS_INLINE int
Vol::open_write(CacheVC *cont, int allow_if_writers, int max_writers)
{
  Vol *vol = this;
  bool agg_error = false;
  if (!cont->f.remove) {
    agg_error = (!cont->f.update && agg_todo_size > cache_config_agg_write_backlog);
#ifdef CACHE_AGG_FAIL_RATE
    agg_error = agg_error || ((uint32_t) mutex->thread_holding->generator.random() <
                              (uint32_t) (UINT_MAX * CACHE_AGG_FAIL_RATE));
#endif
  }
  if (agg_error) {
    CACHE_INCREMENT_DYN_STAT(cache_write_backlog_failure_stat);
    return ECACHE_WRITE_FAIL;
  }
#ifdef SSD_CACHE
  MigrateToSSD *m_result = NULL;
  if (vol->migrate_probe(&cont->first_key, &m_result)) {
    m_result->notMigrate = true;
  }
#endif
  if (open_dir.open_write(cont, allow_if_writers, max_writers)) {
#ifdef CACHE_STAT_PAGES
    ink_debug_assert(cont->mutex->thread_holding == this_ethread());
    ink_assert(!cont->stat_link.next && !cont->stat_link.prev);
    stat_cache_vcs.enqueue(cont, cont->stat_link);
#endif
    return 0;
  }
  return ECACHE_DOC_BUSY;
}

TS_INLINE int
Vol::close_write_lock(CacheVC *cont)
{
  EThread *t = cont->mutex->thread_holding;
  CACHE_TRY_LOCK(lock, mutex, t);
  if (!lock)
    return -1;
  return close_write(cont);
}

TS_INLINE int
Vol::open_write_lock(CacheVC *cont, int allow_if_writers, int max_writers)
{
  EThread *t = cont->mutex->thread_holding;
  CACHE_TRY_LOCK(lock, mutex, t);
  if (!lock)
    return -1;
  return open_write(cont, allow_if_writers, max_writers);
}

TS_INLINE OpenDirEntry *
Vol::open_read_lock(INK_MD5 *key, EThread *t)
{
  CACHE_TRY_LOCK(lock, mutex, t);
  if (!lock)
    return NULL;
  return open_dir.open_read(key);
}

TS_INLINE int
Vol::begin_read_lock(CacheVC *cont)
{
  // no need for evacuation as the entire document is already in memory
#ifndef  CACHE_STAT_PAGES
  if (cont->f.single_fragment)
    return 0;
#endif
  // VC is enqueued in stat_cache_vcs in the begin_read call
  EThread *t = cont->mutex->thread_holding;
  CACHE_TRY_LOCK(lock, mutex, t);
  if (!lock)
    return -1;
  return begin_read(cont);
}

TS_INLINE int
Vol::close_read_lock(CacheVC *cont)
{
  EThread *t = cont->mutex->thread_holding;
  CACHE_TRY_LOCK(lock, mutex, t);
  if (!lock)
    return -1;
  return close_read(cont);
}

TS_INLINE int
dir_delete_lock(CacheKey *key, Vol *d, ProxyMutex *m, Dir *del)
{
  EThread *thread = m->thread_holding;
  CACHE_TRY_LOCK(lock, d->mutex, thread);
  if (!lock)
    return -1;
  return dir_delete(key, d, del);
}

TS_INLINE int
dir_insert_lock(CacheKey *key, Vol *d, Dir *to_part, ProxyMutex *m)
{
  EThread *thread = m->thread_holding;
  CACHE_TRY_LOCK(lock, d->mutex, thread);
  if (!lock)
    return -1;
  return dir_insert(key, d, to_part);
}

TS_INLINE int
dir_overwrite_lock(CacheKey *key, Vol *d, Dir *to_part, ProxyMutex *m, Dir *overwrite, bool must_overwrite = true)
{
  EThread *thread = m->thread_holding;
  CACHE_TRY_LOCK(lock, d->mutex, thread);
  if (!lock)
    return -1;
  return dir_overwrite(key, d, to_part, overwrite, must_overwrite);
}

void TS_INLINE
rand_CacheKey(CacheKey *next_key, ProxyMutex *mutex)
{
  next_key->b[0] = mutex->thread_holding->generator.random();
  next_key->b[1] = mutex->thread_holding->generator.random();
}

extern uint8_t CacheKey_next_table[];
void TS_INLINE
next_CacheKey(CacheKey *next_key, CacheKey *key)
{
  uint8_t *b = (uint8_t *) next_key;
  uint8_t *k = (uint8_t *) key;
  b[0] = CacheKey_next_table[k[0]];
  for (int i = 1; i < 16; i++)
    b[i] = CacheKey_next_table[(b[i - 1] + k[i]) & 0xFF];
}
extern uint8_t CacheKey_prev_table[];
void TS_INLINE
prev_CacheKey(CacheKey *prev_key, CacheKey *key)
{
  uint8_t *b = (uint8_t *) prev_key;
  uint8_t *k = (uint8_t *) key;
  for (int i = 15; i > 0; i--)
    b[i] = 256 + CacheKey_prev_table[k[i]] - k[i - 1];
  b[0] = CacheKey_prev_table[k[0]];
}

TS_INLINE unsigned int
next_rand(unsigned int *p)
{
  unsigned int seed = *p;
  seed = 1103515145 * seed + 12345;
  *p = seed;
  return seed;
}

extern ClassAllocator<CacheRemoveCont> cacheRemoveContAllocator;

TS_INLINE CacheRemoveCont *
new_CacheRemoveCont()
{
  CacheRemoveCont *cache_rm = cacheRemoveContAllocator.alloc();

  cache_rm->mutex = new_ProxyMutex();
  SET_CONTINUATION_HANDLER(cache_rm, &CacheRemoveCont::event_handler);
  return cache_rm;
}

TS_INLINE void
free_CacheRemoveCont(CacheRemoveCont *cache_rm)
{
  cache_rm->mutex = NULL;
  cacheRemoveContAllocator.free(cache_rm);
}

TS_INLINE int
CacheRemoveCont::event_handler(int event, void *data)
{
  (void) event;
  (void) data;
  free_CacheRemoveCont(this);
  return EVENT_DONE;
}

int64_t cache_bytes_used(void);
int64_t cache_bytes_total(void);

#ifdef DEBUG
#define CACHE_DEBUG_INCREMENT_DYN_STAT(_x) CACHE_INCREMENT_DYN_STAT(_x)
#define CACHE_DEBUG_SUM_DYN_STAT(_x,_y) CACHE_SUM_DYN_STAT(_x,_y)
#else
#define CACHE_DEBUG_INCREMENT_DYN_STAT(_x)
#define CACHE_DEBUG_SUM_DYN_STAT(_x,_y)
#endif

struct CacheHostRecord;
struct Vol;
class CacheHostTable;

struct Cache
{
  volatile int cache_read_done;
  volatile int total_good_nvol;
  volatile int total_nvol;
  volatile int ready;
  int64_t cache_size;             //in store block size
  CacheHostTable *hosttable;
  volatile int total_initialized_vol;
  int scheme;

  int open(bool reconfigure, bool fix);
  int close();

  Action *lookup(Continuation *cont, CacheKey *key, CacheFragType type, char *hostname, int host_len);
  inkcoreapi Action *open_read(Continuation *cont, CacheKey *key, CacheFragType type, char *hostname, int len);
  inkcoreapi Action *open_write(Continuation *cont, CacheKey *key,
                                CacheFragType frag_type, int options = 0,
                                time_t pin_in_cache = (time_t) 0, char *hostname = 0, int host_len = 0);
  inkcoreapi Action *remove(Continuation *cont, CacheKey *key,
                            CacheFragType type = CACHE_FRAG_TYPE_HTTP,
                            bool user_agents = true, bool link = false,
                            char *hostname = 0, int host_len = 0);
  Action *scan(Continuation *cont, char *hostname = 0, int host_len = 0, int KB_per_second = 2500);

#ifdef HTTP_CACHE
  Action *lookup(Continuation *cont, URL *url, CacheFragType type);
  inkcoreapi Action *open_read(Continuation *cont, CacheKey *key,
                               CacheHTTPHdr *request,
                               CacheLookupHttpConfig *params, CacheFragType type, char *hostname, int host_len);
  Action *open_read(Continuation *cont, URL *url, CacheHTTPHdr *request,
                    CacheLookupHttpConfig *params, CacheFragType type);
  Action *open_write(Continuation *cont, CacheKey *key,
                     CacheHTTPInfo *old_info, time_t pin_in_cache = (time_t) 0,
                     CacheKey *key1 = NULL,
                     CacheFragType type = CACHE_FRAG_TYPE_HTTP, char *hostname = 0, int host_len = 0);
  Action *open_write(Continuation *cont, URL *url, CacheHTTPHdr *request,
                     CacheHTTPInfo *old_info, time_t pin_in_cache = (time_t) 0,
                     CacheFragType type = CACHE_FRAG_TYPE_HTTP);
  static void generate_key(INK_MD5 *md5, URL *url);
#endif

  Action *link(Continuation *cont, CacheKey *from, CacheKey *to, CacheFragType type, char *hostname, int host_len);
  Action *deref(Continuation *cont, CacheKey *key, CacheFragType type, char *hostname, int host_len);

  void vol_initialized(bool result);

  int open_done();

  Vol *key_to_vol(CacheKey *key, char *hostname, int host_len);

  Cache()
    : cache_read_done(0), total_good_nvol(0), total_nvol(0), ready(CACHE_INITIALIZING), cache_size(0),  // in store block size
      hosttable(NULL), total_initialized_vol(0), scheme(CACHE_NONE_TYPE)
    { }
};

extern Cache *theCache;
extern Cache *theStreamCache;
inkcoreapi extern Cache *caches[NUM_CACHE_FRAG_TYPES];

#ifdef HTTP_CACHE
TS_INLINE Action *
Cache::open_read(Continuation *cont, CacheURL *url, CacheHTTPHdr *request,
                 CacheLookupHttpConfig *params, CacheFragType type)
{
  INK_MD5 md5;
  int len;
  url->MD5_get(&md5);
  const char *hostname = url->host_get(&len);
  return open_read(cont, &md5, request, params, type, (char *) hostname, len);
}

TS_INLINE void
Cache::generate_key(INK_MD5 *md5, URL *url)
{
  url->MD5_get(md5);
}

TS_INLINE Action *
Cache::open_write(Continuation *cont, CacheURL *url, CacheHTTPHdr *request,
                  CacheHTTPInfo *old_info, time_t pin_in_cache, CacheFragType type)
{
  (void) request;
  INK_MD5 url_md5;
  url->MD5_get(&url_md5);
  int len;
  const char *hostname = url->host_get(&len);

  return open_write(cont, &url_md5, old_info, pin_in_cache, NULL, type, (char *) hostname, len);
}
#endif

TS_INLINE unsigned int
cache_hash(INK_MD5 & md5)
{
  uint64_t f = md5.fold();
  unsigned int mhash = (unsigned int) (f >> 32);
  return mhash;
}

#ifdef HTTP_CACHE
#define CLUSTER_CACHE
#endif
#ifdef CLUSTER_CACHE
#include "P_Net.h"
#include "P_ClusterInternal.h"
// Note: This include must occur here in order to avoid numerous forward
//       reference problems.
#include "P_ClusterInline.h"
#include "clusterinterface.h"
#endif

//struct ClusterCacheVC: public CacheVConnection
//{
//  Action _action;
//  int64_t seek_to;                // pread offset
//  int64_t offset;                 // offset into 'blocks' of data to write
//  int64_t writer_offset;          // offset of the writer for reading from a writer
//  int64_t length;                 // length of data available to write
//  int64_t total_len;
//  int64_t doc_len;
//
//  int doc_pos;                // read position in 'buf'
//  int d_len;
//
//  int closed;
//  int recursive;
//  time_t time_pin;
//  int disk_io_priority;
//  MessagePriority priority;
//
//  EThread *initial_thread;  // initial thread open_XX was called on
//  ClusterSession cs;
//  Event *trigger;
//  ContinuationHandler save_handler;
//
//  int probe_depth;
//  bool in_progress; //
//  bool data_done;
//  bool remote_closed;
//  bool trigger_remote_close;
//
//  VIO vio;
//  Ptr<IOBufferData> buf;          // for read
//  Ptr<IOBufferData> first_buf;
//  Ptr<IOBufferBlock> blocks; // data available to write
//  Ptr<IOBufferBlock> writer_buf;
//
//  DummyClusterAIOCallback io;
//  CacheHTTPInfo alternate;
//
//  union
//  {
//    uint32_t flags;
//    struct
//    {
//      unsigned int use_first_key:1;
//      unsigned int overwrite:1; // overwrite first_key Dir if it exists
//      unsigned int close_complete:1; // WRITE_COMPLETE is final
//      unsigned int sync:1; // write to be committed to durable storage before WRITE_COMPLETE
//      unsigned int evacuator:1;
//      unsigned int single_fragment:1;
//      unsigned int evac_vector:1;
//      unsigned int lookup:1;
//      unsigned int update:1;
//      unsigned int remove:1;
//      unsigned int remove_aborted_writers:1;
//      unsigned int open_read_timeout:1; // UNUSED
//      unsigned int data_done:1;
//      unsigned int read_from_writer_called:1;
//      unsigned int not_from_ram_cache:1;        // entire object was from ram cache
//      unsigned int rewrite_resident_alt:1;
//      unsigned int readers:1;
//      unsigned int doc_from_ram_cache:1;
//#ifdef HIT_EVACUATE
//      unsigned int hit_evacuate:1;
//#endif
//#ifdef HTTP_CACHE
//      unsigned int force_empty:1; // used for cache empty http document
//#endif
//#ifdef SSD_CACHE
//      unsigned int read_from_ssd:1;
//      unsigned int write_into_ssd:1;
//      unsigned int ram_fixup:1;
//      unsigned int transistor:1;
//#endif
//    } f;
//  };
//  VIO *do_io_read(Continuation *c, int64_t nbytes, MIOBuffer *buf); // invoke remote
//  VIO *do_io_pread(Continuation *c, int64_t nbytes, MIOBuffer *buf, int64_t offset); // invoke remote
//  VIO *do_io_write(Continuation *c, int64_t nbytes, IOBufferReader *buf, bool owner = false); // invoke remote
//  void do_io_close(int lerrno = -1); // invoke remote ?
//  void reenable(VIO *avio); // invoke remote ?
//  void reenable_re(VIO *avio); // invoke remote ?
//
//  void do_remote_close(); // invoke remote, for cancel or error
//
//  virtual int get_header(void **ptr, int *len)
//  {
//    NOWARN_UNUSED(ptr);
//    NOWARN_UNUSED(len);
//    ink_assert(!"implemented");
//    return -1;
//  }
//  virtual int set_header(void *ptr, int len)
//  {
//    NOWARN_UNUSED(ptr);
//    NOWARN_UNUSED(len);
//    ink_assert(!"implemented");
//    return -1;
//  }
//  virtual int get_single_data(void **ptr, int *len)
//  {
//    NOWARN_UNUSED(ptr);
//    NOWARN_UNUSED(len);
//    ink_assert(!"implemented");
//    return -1;
//  }
//
//#ifdef HTTP_CACHE
//  virtual void set_http_info(CacheHTTPInfo *info) {
//    alternate.copy_shallow(info);
//    info->clear();
//  }
//  virtual void get_http_info(CacheHTTPInfo ** info) {
//    *info = &alternate;
//  }
//#endif
//
//  bool is_ram_cache_hit() {
//    ink_assert(vio.op == VIO::READ);
//    return !f.not_from_ram_cache;
//  }
//  virtual bool set_disk_io_priority(int priority)
//  {
//    disk_io_priority = priority;
//    return true;
//  }
//  virtual int get_disk_io_priority() {
//    return disk_io_priority;
//  }
//  virtual bool set_pin_in_cache(time_t t) {
//    time_pin = t;
//    return true;
//  }
//  virtual time_t get_pin_in_cache() {
//    return time_pin;
//  }
//  virtual int64_t get_object_size()
//  {
//    return alternate.object_size_get();
//  }
//
//  void
//  cancel_trigger()
//  {
//    if (trigger) {
//      trigger->cancel_action();
//      trigger = NULL;
//    }
//  }
//
//  int calluser(int event);
//  int callcont(int event);
//  int handleRead(int event, void *data);
//  int openReadReadDone(int event, void *data);
//  int handleWrite(int event, void *data);
//  int openWriteWriteDone(int event, void *data);
//  int openReadStart(int event, void *data);
//  int openWriteStart(int event, void *data);
//  int openReadMain(int event, void *data);
//  int openWriteMain(int event, void *data);
//};

TS_INLINE Action *
CacheProcessor::lookup(Continuation *cont, CacheKey *key, bool cluster_cache_local, bool local_only,
                       CacheFragType frag_type, char *hostname, int host_len)
{
  (void) local_only;
#ifdef CLUSTER_CACHE
  // Try to send remote, if not possible, handle locally
  if ((cache_clustering_enabled > 0) && !cluster_cache_local && !local_only) {
    Action *a = Cluster_lookup(cont, key, frag_type, hostname, host_len);
    if (a) {
      return a;
    }
  }
#endif
  return caches[frag_type]->lookup(cont, key, frag_type, hostname, host_len);
}

TS_INLINE int
CacheProcessor::belong_to_me(CacheKey *key)
{
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0) {
    INK_MD5 url_md5 = *key;

    if (cluster_machine_at_depth(cache_hash(url_md5)))
      return 0;
  }
#endif
  return 1;
}

TS_INLINE inkcoreapi Action *
CacheProcessor::open_read(Continuation *cont, CacheKey *key, int cluster_cache_local, CacheFragType frag_type, char *hostname, int host_len)
{
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0 && cluster_cache_local != CACHE_CONTROL_LOCAL) {
    return open_read_internal(CACHE_OPEN_READ, cont, (MIOBuffer *) 0,
                              (CacheURL *) 0, (CacheHTTPHdr *) 0,
                              (CacheLookupHttpConfig *) 0, key, 0, frag_type, hostname, host_len, (cluster_cache_local == CACHE_CONTROL_MIGRATE));
  }
#endif
  return caches[frag_type]->open_read(cont, key, frag_type, hostname, host_len);
}

TS_INLINE Action *
CacheProcessor::open_read_buffer(Continuation *cont, MIOBuffer *buf, CacheKey *key, CacheFragType frag_type,
                                 char *hostname, int host_len)
{
  (void) buf;
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0) {
    return open_read_internal(CACHE_OPEN_READ_BUFFER, cont, buf,
                              (CacheURL *) 0, (CacheHTTPHdr *) 0,
                              (CacheLookupHttpConfig *) 0, key, 0, frag_type, hostname, host_len);
  }
#endif
  return caches[frag_type]->open_read(cont, key, frag_type, hostname, host_len);
}


TS_INLINE inkcoreapi Action *
CacheProcessor::open_write(Continuation *cont, CacheKey *key, bool cluster_cache_local, CacheFragType frag_type,
                           int expected_size, int options, time_t pin_in_cache,
                           char *hostname, int host_len)
{
  (void) expected_size;
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0 && !cluster_cache_local) {
    ClusterMachine *m = cluster_machine_at_depth(cache_hash(*key));
    if (m)
      return Cluster_write(cont, expected_size, (MIOBuffer *) 0, m,
                         key, frag_type, options, pin_in_cache,
                         CACHE_OPEN_WRITE, key, (CacheURL *) 0,
                         (CacheHTTPHdr *) 0, (CacheHTTPInfo *) 0, hostname, host_len);
  }
#endif
  return caches[frag_type]->open_write(cont, key, frag_type, options, pin_in_cache, hostname, host_len);
}

TS_INLINE Action *
CacheProcessor::open_write_buffer(Continuation *cont, MIOBuffer *buf, CacheKey *key,
                                  CacheFragType frag_type, int options, time_t pin_in_cache,
                                  char *hostname, int host_len)
{
  NOWARN_UNUSED(pin_in_cache);
  (void)cont;
  (void)buf;
  (void)key;
  (void)frag_type;
  (void)options;
  (void)hostname;
  (void)host_len;
  ink_assert(!"implemented");
  return NULL;
}

TS_INLINE Action *
CacheProcessor::remove(Continuation *cont, CacheKey *key, bool cluster_cache_local, CacheFragType frag_type,
                       bool rm_user_agents, bool rm_link, char *hostname, int host_len)
{
  Debug("cache_remove", "[CacheProcessor::remove] Issuing cache delete for %u", cache_hash(*key));
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0 && !cluster_cache_local) {
    ClusterMachine *m = cluster_machine_at_depth(cache_hash(*key));

    if (m) {
      return Cluster_remove(m, cont, key, rm_user_agents, rm_link, frag_type, hostname, host_len);
    }
  }
#endif
  return caches[frag_type]->remove(cont, key, frag_type, rm_user_agents, rm_link, hostname, host_len);
}

TS_INLINE Action *
scan(Continuation *cont, char *hostname = 0, int host_len = 0, int KB_per_second = 2500)
{
  return caches[CACHE_FRAG_TYPE_HTTP]->scan(cont, hostname, host_len, KB_per_second);
}

#ifdef HTTP_CACHE
TS_INLINE Action *
CacheProcessor::lookup(Continuation *cont, URL *url, bool cluster_cache_local, bool local_only, CacheFragType frag_type)
{
  (void) local_only;
  INK_MD5 md5;
  url->MD5_get(&md5);
  int host_len = 0;
  const char *hostname = url->host_get(&host_len);

  return lookup(cont, &md5, cluster_cache_local, local_only, frag_type, (char *) hostname, host_len);
}

TS_INLINE Action *
CacheProcessor::open_read_buffer(Continuation *cont, MIOBuffer *buf,
                                 URL *url, CacheHTTPHdr *request, CacheLookupHttpConfig *params, CacheFragType type)
{
  (void) buf;
#ifdef CLUSTER_CACHE
  if (cache_clustering_enabled > 0) {
    return open_read_internal(CACHE_OPEN_READ_BUFFER_LONG, cont, buf, url,
                              request, params, (CacheKey *) 0, 0, type, (char *) 0, 0);
  }
#endif
  return caches[type]->open_read(cont, url, request, params, type);
}

TS_INLINE Action *
CacheProcessor::open_write_buffer(Continuation * cont, MIOBuffer * buf, URL * url,
                                  CacheHTTPHdr * request, CacheHTTPHdr * response, CacheFragType type)
{
  (void) cont;
  (void) buf;
  (void) url;
  (void) request;
  (void) response;
  (void) type;
  ink_assert(!"implemented");
  return NULL;
}

#endif


#ifdef CLUSTER_CACHE
TS_INLINE Action *
CacheProcessor::open_read_internal(int opcode,
                                   Continuation *cont, MIOBuffer *buf,
                                   CacheURL *url,
                                   CacheHTTPHdr *request,
                                   CacheLookupHttpConfig *params,
                                   CacheKey *key,
                                   time_t pin_in_cache, CacheFragType frag_type, char *hostname, int host_len, bool migrate)
{
  INK_MD5 url_md5;
  if ((opcode == CACHE_OPEN_READ_LONG) || (opcode == CACHE_OPEN_READ_BUFFER_LONG)) {
    Cache::generate_key(&url_md5, url);
  } else {
    url_md5 = *key;
  }
  ClusterMachine *m = cluster_machine_at_depth(cache_hash(url_md5));

  if (m) {
    return Cluster_read(m, opcode, cont, buf, url,
                        request, params, &url_md5, pin_in_cache, frag_type, hostname, host_len, migrate);
  } else {
    if ((opcode == CACHE_OPEN_READ_LONG)
        || (opcode == CACHE_OPEN_READ_BUFFER_LONG)) {
      if (migrate && cache_migrate) {
        char cache_url[2048];
        int length;
        url->string_get_buf(cache_url, sizeof(cache_url), &length);
        cache_migrate(cache_url, length);
      }
      return caches[frag_type]->open_read(cont, &url_md5, request, params, frag_type, hostname, host_len);
    } else {
      return caches[frag_type]->open_read(cont, &url_md5, frag_type, hostname, host_len);
    }
  }
}
#endif

#ifdef CLUSTER_CACHE
TS_INLINE Action *
CacheProcessor::link(Continuation *cont, CacheKey *from, CacheKey *to, bool cluster_cache_local,
                     CacheFragType type, char *hostname, int host_len)
{
  if (cache_clustering_enabled > 0 && !cluster_cache_local) {
    // Use INK_MD5 in "from" to determine target machine
    ClusterMachine *m = cluster_machine_at_depth(cache_hash(*from));
    if (m) {
      return Cluster_link(m, cont, from, to, type, hostname, host_len);
    }
  }
  return caches[type]->link(cont, from, to, type, hostname, host_len);
}

TS_INLINE Action *
CacheProcessor::deref(Continuation *cont, CacheKey *key, bool cluster_cache_local, CacheFragType type, char *hostname, int host_len)
{
  if (cache_clustering_enabled > 0 && !cluster_cache_local) {
    ClusterMachine *m = cluster_machine_at_depth(cache_hash(*key));
    if (m) {
      return Cluster_deref(m, cont, key, type, hostname, host_len);
    }
  }
  return caches[type]->deref(cont, key, type, hostname, host_len);
}
#endif

TS_INLINE Action *
CacheProcessor::scan(Continuation *cont, char *hostname, int host_len, int KB_per_second)
{
  return caches[CACHE_FRAG_TYPE_HTTP]->scan(cont, hostname, host_len, KB_per_second);
}

TS_INLINE int
CacheProcessor::IsCacheEnabled()
{
  return CacheProcessor::initialized;
}

TS_INLINE int
CacheProcessor::IsCacheClustering()
{
  return cache_clustering_enabled;
}

TS_INLINE unsigned int
CacheProcessor::IsCacheReady(CacheFragType type)
{
  if (IsCacheEnabled() != CACHE_INITIALIZED)
    return 0;
  return (cache_ready & (1 << type));
}

TS_INLINE Cache *
local_cache()
{
  return theCache;
}

LINK_DEFINITION(CacheVC, opendir_link)

//inline bool
//CacheVC::get_entry(CacheWriterTable *table) {
//  int indx = first_key.word(1) % CACHE_WRITER_BUCKET_SIZE;
//  ink_mutex *mutex = &table->buckets[indx].the_mutex;
//
//  ink_mutex_acquire(mutex);
//  for (CacheWriterEntry *entry = table->buckets[indx].writers.head; entry; entry = entry->link.next) {
//    if (first_key == entry->key) {
//      cw = entry;
//      ink_mutex_release(mutex);
//      return true;
//    }
//  }
//  ink_mutex_release(mutex);
//  return false;
//}

inline bool
CacheVC::add_entry(CacheWriterTable *table) {
  int indx = first_key.word(1) % CACHE_WRITER_BUCKET_SIZE;
  ink_mutex *mutex = &table->buckets[indx].the_mutex;
  bool found = false;

  ink_mutex_acquire(mutex);

  for (CacheWriterEntry *entry = table->buckets[indx].writers.head; entry; entry = entry->link.next) {
    if (first_key == entry->key) {
      found = true;
      break;
    }
  }
  if (!found) {
    CacheWriterEntry *entry = cacheWriterEntryAllocator.alloc();
    entry->mutex = mutex;
    entry->key = first_key;
    entry->writer = this;
    entry->not_rww = !cache_config_read_while_writer;
    entry->r = entry->buffer.alloc_reader();
    table->buckets[indx].writers.push(entry);
    cw = entry;
  }

  ink_mutex_release(mutex);
  return !found;
}

inline void
CacheVC::clear_entry(CacheWriterTable *table) {
  int indx = first_key.word(1) % CACHE_WRITER_BUCKET_SIZE;
  ink_mutex *mutex = &table->buckets[indx].the_mutex;

  ink_assert(cw && cw->writer == this && mutex == cw->mutex);

  ink_mutex_acquire(cw->mutex);
  table->buckets[indx].writers.remove(cw);
  ink_mutex_release(cw->mutex);

}
//struct SetIOReadMessage: public ClusterMessageHeader
//{
//  int64_t nbytes;
//  int64_t offset;
//};
//
//struct SetIOWriteMessage: public ClusterMessageHeader
//{
//  int32_t hdr_len;
//  int64_t nbytes;
//};
//
//struct SetIOCloseMessage: public ClusterMessageHeader
//{
//  int h_len;
//  int d_len;
//  int64_t total_len;
//};
//
//struct SetIOReenableMessage: public ClusterMessageHeader
//{
//  int reenable;
//};
//struct SetResponseMessage: public ClusterMessageHeader
//{
//
//};
//
//ClusterCacheVC *new_ClusterCacheVC();
//void free_ClusterCacheVC(ClusterCacheVC *ccvc);
//
//inline int
//ClusterCacheVC::calluser(int event)
//{
//  recursive++;
//  ink_debug_assert(this_ethread() != vio._cont->mutex->thread_holding);
//  vio._cont->handleEvent(event, (void *) &vio);
//  recursive--;
//  if (closed) {
//    free_ClusterCacheVC(this);
//    return EVENT_DONE;
//  }
//  return EVENT_CONT;
//}
//
//inline int
//ClusterCacheVC::callcont(int event)
//{
//  recursive++;
//  ink_debug_assert(this_ethread() != _action.mutex->thread_holding);
//  _action.continuation->handleEvent(event, this);
//  recursive--;
//  if (closed) {
//    free_ClusterCacheVC(this);
//    return EVENT_DONE;
//  } else if (vio.vc_server)
//    handleEvent(EVENT_IMMEDIATE, 0);
//  return EVENT_DONE;
//}
//
//extern ClassAllocator<ClusterCacheVC> clusterCacheVCAllocator;
#endif /* _P_CACHE_INTERNAL_H__ */
