/** @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.
 */

/****************************************************************************

  Cluster.h


****************************************************************************/

#ifndef _P_Cluster_Cache_h
#define _P_Cluster_Cache_h

//*****************************************************************************
// Initially derived from Cluster.h "1.77.2.11 1999/01/21 03:24:10"
//*****************************************************************************

/****************************************************************************/
// #define LOCAL_CLUSTER_TEST_MODE 1
// ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
//  Set the above #define to enable local clustering.  "Local clustering"
//  is a test only mode where all cluster nodes reside on the same host.
//
//  Configuration notes:
//   - For "cluster.config" entries, always use "127.0.0.1" as the IP
//     address and select a host unique cluster port.
//
//  Restrictions:
//   1) Does not work with the manager.  You must only run with the server
//      and hand configure "cluster.config".
//   2) Currently, this has only been tested in a two node configuration.
//
/****************************************************************************/

#include "P_ClusterMachine.h"
#include "clusterinterface.h"
//
// Cluster Processor
//
// - monitors the status of the cluster
// - provides communication between machines in the cluster
// - provides callbacks to other processors when the cluster configuration
//   changes
//
#define CLUSTER_MAJOR_VERSION               6
#define CLUSTER_MINOR_VERSION               0

// Lowest supported major/minor cluster version
#define MIN_CLUSTER_MAJOR_VERSION	    CLUSTER_MAJOR_VERSION
#define MIN_CLUSTER_MINOR_VERSION	    CLUSTER_MINOR_VERSION


#define DEFAULT_CLUSTER_PORT_NUMBER         0
#define DEFAULT_NUMBER_OF_CLUSTER_THREADS   1
#define DEFAULT_CLUSTER_HOST                ""

#define MAX_CLUSTER_SEND_LENGTH             INT_MAX

#define CLUSTER_MAX_MACHINES                256
// less than 1% disparity at 255 machines, 32707 is prime less than 2^15
#define CLUSTER_HASH_TABLE_SIZE             32707

// after timeout the configuration is "dead"
#define CLUSTER_CONFIGURATION_TIMEOUT       HRTIME_DAY
// after zombie the configuration is deleted
#define CLUSTER_CONFIGURATION_ZOMBIE        (HRTIME_DAY*2)


// the number of configurations into the past we probe for data
// one allows a new machine to come into or fall out of the
// cluster without loss of data.  If the data is redistributed within
// one day, no data will be lost.
#define CONFIGURATION_HISTORY_PROBE_DEPTH   1


// move these to a central event definition file (Event.h)
#define CLUSTER_EVENT_CHANGE            (CLUSTER_EVENT_EVENTS_START)
#define CLUSTER_EVENT_CONFIGURATION     (CLUSTER_EVENT_EVENTS_START+1)
#define CLUSTER_EVENT_OPEN              (CLUSTER_EVENT_EVENTS_START+2)
#define CLUSTER_EVENT_OPEN_EXISTS       (CLUSTER_EVENT_EVENTS_START+3)
#define CLUSTER_EVENT_OPEN_FAILED       (CLUSTER_EVENT_EVENTS_START+4)

// internal event code
#define CLUSTER_EVENT_STEAL_THREAD      (CLUSTER_EVENT_EVENTS_START+50)

#ifdef DEBUG
extern int64_t num_of_cluster_cachevc;
#endif
//////////////////////////////////////////////////////////////
// Miscellaneous byte swap routines
//////////////////////////////////////////////////////////////
inline void
ats_swap16(uint16_t * d)
{
  unsigned char *p = (unsigned char *) d;
  *d = ((p[1] << 8) | p[0]);
}

inline uint16_t
ats_swap16(uint16_t d)
{
  ats_swap16(&d);
  return d;
}

inline void
ats_swap32(uint32_t * d)
{
  unsigned char *p = (unsigned char *) d;
  *d = ((p[3] << 24) | (p[2] << 16) | (p[1] << 8) | p[0]);
}

inline uint32_t
ats_swap32(uint32_t d)
{
  ats_swap32(&d);
  return d;
}

inline void
ats_swap64(uint64_t * d)
{
  unsigned char *p = (unsigned char *) d;
  *d = (((uint64_t) p[7] << 56) | ((uint64_t) p[6] << 48) |
        ((uint64_t) p[5] << 40) | ((uint64_t) p[4] << 32) |
        ((uint64_t) p[3] << 24) | ((uint64_t) p[2] << 16) | ((uint64_t) p[1] << 8) | (uint64_t) p[0]);
}

inline uint64_t
ats_swap64(uint64_t d)
{
  ats_swap64(&d);
  return d;
}

//////////////////////////////////////////////////////////////

struct ClusterConfiguration
{
  int n_machines;
  ClusterMachine *machines[CLUSTER_MAX_MACHINES];

  ClusterMachine *machine_hash(unsigned int hash_value)
  {
    return machines[hash_table[hash_value % CLUSTER_HASH_TABLE_SIZE]];
  }

  ClusterMachine *find(unsigned int ip, int port = 0) {
    for (int i = 0; i < n_machines; i++)
      if (ip == machines[i]->ip && (!port || !machines[i]->cluster_port || machines[i]->cluster_port == port))
        return machines[i];
    return NULL;
  }

  int find_idx(unsigned int ip, int port = 0) {
    for (int i = 0; i < n_machines; i++)
      if (ip == machines[i]->ip && (!port || !machines[i]->cluster_port || machines[i]->cluster_port == port))
        return i;
    return -1;
  }

  //
  // Private
  //
  ClusterConfiguration();
  unsigned char hash_table[CLUSTER_HASH_TABLE_SIZE];
  ink_hrtime changed;
  SLINK(ClusterConfiguration, link);
};

inline bool
machine_in_vector(ClusterMachine * m, ClusterMachine ** mm, int len)
{
  for (int i = 0; i < len; i++)
    if (m == mm[i])
      return true;
  return false;
}

//
//   Return a machine list based on the given hash which is ordered from
//   now to the past.  Returned list contains no duplicates and depth_list[0]
//   entry (current configuration) is always valid.  Entries following
//   depth_list[0] are considered online.
//   Returns 0 on Success else non-zero on error.
//
int
cluster_machine_depth_list(unsigned int hash, ClusterMachine ** depth_list,
                           int depth_list_size = (CONFIGURATION_HISTORY_PROBE_DEPTH + 1));

//
// Returns either a machine or NULL.
// Finds a machine starting at probe_depth going up to
//    CONFIGURATION_HISTORY_PROBE_DEPTH
// which is up, not the current machine and has not yet been probed.
// Updates: probe_depth and past_probes.
//
inkcoreapi ClusterMachine *cluster_machine_at_depth(unsigned int hash, int *probe_depth = NULL,
                                                    ClusterMachine ** past_probes = NULL);

//
// Returns either a machine or NULL.
// Finds a machine according the default cluster configuration
//
inkcoreapi ClusterMachine *cluster_machine_by_default(unsigned int hash);

//
// Returns the latest left_time according the owner_ip.
//
inkcoreapi ink_hrtime owner_left_time_get(unsigned int owner_ip);

//
// Cluster
//   A cluster of machines which act as a single cache.
//
struct Cluster
{
  //
  // Public Interface
  //

  //
  // Cluster Hash Function
  //

  // Takes a hash value to a machine.  The hash function has the following
  // properties:
  //   1 - it divides input domain into the output range evenly (within 1%)
  //   2 - it tends to produce the same Machine for the same hash_value's
  //       for different configurations
  //   3 - it produces the hash same function for a given configuration of
  //       machines independent of the order they were added or removed
  //       from the cluster.  (it is a pure function of the configuration)
  //   Thread-safe
  //
  ClusterMachine *machine_hash(unsigned int hash_value)
  {
    return current_configuration()->machine_hash(hash_value);
  }

  //
  // A hash table used to record the latest left time of each machine.
  // Key: ip, data type: unsigned int
  // Val: the latest left time, data type: ink_hrtime
  //
  InkHashTable *machines_left_time_ht;

  //
  // Cluster Configuration
  //
  void init_default_configuration();

  // Register callback for a cluster configuration change.
  // calls cont->handleEvent(EVENT_CLUSTER_CHANGE);
  //   Thread-safe
  //
  void cluster_change_callback(Continuation * cont);

  // Return the current configuration
  //   Thread-safe
  //
  ClusterConfiguration *current_configuration()
  {
    return configurations.head;
  }

  // Return the previous configuration.
  // Use from within the cluster_change_callback.
  //   Thread-safe
  //
  ClusterConfiguration *previous_configuration()
  {
    return configurations.head->link.next;
  }

  //
  // Private
  //
  // The configurations are updated only in the thread which is
  // accepting cluster connections.
  //
  SLL<ClusterConfiguration> configurations;
  ClusterConfiguration *default_configuration;

  Cluster();
};

//
// ClusterVCToken
//   An token passed between nodes to represent a virtualized connection.
//   (see ClusterProcessor::alloc_remote() and attach_remote() below)
//
struct ClusterVCToken
{
  //
  // Marshal this data to send the token across the cluster
  //
  uint32_t ip_created;
  uint32_t ch_id;
  uint32_t sequence_number;

  bool is_clear()
  {
    return !ip_created;
  }
  void clear()
  {
    ip_created = 0;
    sequence_number = 0;
  }

  ClusterVCToken(unsigned int aip = 0, unsigned int id = 0, unsigned int aseq = 0)
:  ip_created(aip), ch_id(id), sequence_number(aseq) {
  }
  //
  // Private
  //
  void alloc();

  inline void SwapBytes()
  {
    ats_swap32(&ch_id);
    ats_swap32(&sequence_number);
  }
};

//
// ClusterFunctionPtr
//   A pointer to a procedure which can be invoked accross the cluster.
//   This must be registered.
//
typedef void ClusterFunction(ClusterHandler * ch, void *data, int len);

typedef void ClusterFunctionExt(ClusterSession cs, void *context, void *data);
typedef ClusterFunction *ClusterFunctionPtr;

struct ClusterVConnectionBase;

struct ClusterVConnState
{
  //
  // Private
  //
  volatile int enabled;
  // multiples of XXX_PERIOD, high = less often
  int priority;
  VIO vio;
  void *queue;
  int ifd;
  Event *delay_timeout;
  Link<ClusterVConnectionBase> link;

  // void enqueue(void * q, ClusterVConnection * vc);
  ClusterVConnState();
};

struct ClusterVConnectionBase: public CacheVConnection
{
  //
  // Initiate an IO operation.
  // "data" is unused.
  // Only one READ and one WRITE may be active at one time.
  //   THREAD-SAFE, may be called when not handling an event from
  //                the ClusterVConnectionBase, or the ClusterVConnectionBase
  //                creation callback.
  //

  virtual VIO *do_io_read(Continuation * c, int64_t nbytes, MIOBuffer * buf);
  virtual VIO *do_io_write(Continuation * c, int64_t nbytes, IOBufferReader * buf, bool owner = false);
  virtual void do_io_shutdown(ShutdownHowTo_t howto)
  {
    (void) howto;
    ink_assert(!"shutdown of cluster connection");
  }
  virtual void do_io_close(int lerrno = -1);
  virtual VIO* do_io_pread(Continuation*, int64_t, MIOBuffer*, int64_t);

  // Set the timeouts associated with this connection.
  // active_timeout is for the total elasped time of the connection.
  // inactivity_timeout is the elapsed time *while an operation was
  //   enabled* during which the connection was unable to sink/provide data.
  // calling these functions repeatedly resets the timeout.
  //   NOT THREAD-SAFE, may only be called when handing an event from this
  //                    ClusterVConnectionBase, or the ClusterVConnectionBase
  //                    creation callback.
  //
  void set_active_timeout(ink_hrtime timeout_in);
  void set_inactivity_timeout(ink_hrtime timeout_in);
  void cancel_active_timeout();
  void cancel_inactivity_timeout();

  ClusterVConnectionBase();

#ifdef DEBUG
  // Class static data
  static int enable_debug_trace;
#endif
  Action action_;
  EThread *thread;
  volatile int closed;
  ClusterVConnState read;
  ClusterVConnState write;
  LINKM(ClusterVConnectionBase, read, link)
  LINKM(ClusterVConnectionBase, write, link)
  ink_hrtime inactivity_timeout_in;
  ink_hrtime active_timeout_in;
  Event *inactivity_timeout;
  Event *active_timeout;

  virtual void reenable(VIO *);
  virtual void reenable_re(VIO *);
};

inline void
ClusterVConnectionBase::set_active_timeout(ink_hrtime timeout)
{
  active_timeout_in = timeout;
  if (active_timeout) {
    ink_assert(!active_timeout->cancelled);
    if (active_timeout->ethread == this_ethread())
      active_timeout->schedule_in(timeout);
    else {
      active_timeout->cancel(this);
      active_timeout = thread->schedule_in(this, timeout);
    }
  } else {
    if (thread) {
      active_timeout = thread->schedule_in(this, timeout);
    }
  }
}

inline void
ClusterVConnectionBase::set_inactivity_timeout(ink_hrtime timeout)
{
  inactivity_timeout_in = timeout;
  if (inactivity_timeout) {
    ink_assert(!inactivity_timeout->cancelled);
    if (inactivity_timeout->ethread == this_ethread())
      inactivity_timeout->schedule_in(timeout);
    else {
      inactivity_timeout->cancel(this);
      inactivity_timeout = thread->schedule_in(this, timeout);
    }
  } else {
    if (thread) {
      inactivity_timeout = thread->schedule_in(this, timeout);
    }
  }
}

inline void
ClusterVConnectionBase::cancel_active_timeout()
{
  if (active_timeout) {
    active_timeout->cancel(this);
    active_timeout = NULL;
    active_timeout_in = 0;
  }
}

inline void
ClusterVConnectionBase::cancel_inactivity_timeout()
{
  if (inactivity_timeout) {
    inactivity_timeout->cancel(this);
    inactivity_timeout = NULL;
    inactivity_timeout_in = 0;
  }
}

// Data debt owed to VC which is deferred due to lock miss
class ByteBankDescriptor
{
public:
  ByteBankDescriptor()
  {
  }
  IOBufferBlock *get_block()
  {
    return block;
  }

  static ByteBankDescriptor *ByteBankDescriptor_alloc(IOBufferBlock *);
  static void ByteBankDescriptor_free(ByteBankDescriptor *);

public:
  LINK(ByteBankDescriptor, link);

private:
  Ptr<IOBufferBlock> block;  // holder of bank bytes
};

//
// ClusterVConnection
//
struct ClusterVConnection: public ClusterVConnectionBase
{
  //
  // Public Interface (included from ClusterVConnectionBase)
  //
  // Thread-safe  (see Net.h for details)
  //
  // virtual VIO * do_io(
  //   int                   op,
  //   Continuation        * c = NULL,
  //   int                   nbytes = INT64_MAX,
  //   MIOBuffer           * buf = 0,
  //   int                   whence = SEEK_CUR);
  //
  // NOT Thread-safe (see Net.h for details)
  //
  // void set_active_timeout(ink_hrtime timeout_in);
  // void set_inactivity_timeout(ink_hrtime timeout_in);

  //
  // Private
  //

  int startEvent(int event, Event * e);
  int mainEvent(int event, Event * e);

  // 0 on success -1 on failure
  int start(EThread * t);       // New connect protocol

  ClusterVConnection(int is_new_connect_read = 0);
  ~ClusterVConnection();
  void free();                  // Destructor actions (we are using ClassAllocator)

  virtual bool is_read_from_writer() {
    return false;
  }
  virtual void do_io_close(int lerrno = -1);

  ClusterHandler *ch;
  //
  //  Read Channel: (new_connect_read == true)
  //     - open_local()    caller is reader
  //     - connect_local() caller is writer
  //
  //  Write Channel: (new_connect_read == false)
  //     - open_local()    caller is writer
  //     - connect_local() caller is reader
  //
  int new_connect_read;         // Data flow direction wrt origin node
  int remote_free;
  int last_local_free;
  int channel;
  ClusterVCToken token;
  volatile int close_disabled;
  volatile int remote_closed;
  volatile int remote_close_disabled;
  volatile int remote_lerrno;
  int was_closed();
  void allow_close();
  void disable_close();
  int was_remote_closed();
  void allow_remote_close();
  bool schedule_write();
  void set_type(int);
  ink_hrtime start_time;
  ink_hrtime last_activity_time;
  Queue<ByteBankDescriptor> byte_bank_q;   // done awaiting completion
  int n_set_data_msgs;          // # pending set_data() msgs on VC
  int n_recv_set_data_msgs;     // # set_data() msgs received on VC
  volatile int pending_remote_fill;     // Remote fill pending on connection
  Ptr<IOBufferBlock> read_block;   // Hold current data for open read
  bool have_all_data;           // All data in read_block
  int initial_data_bytes;       // bytes in open_read buffer
  Ptr<IOBufferBlock> remote_write_block;   // Write side data for remote fill
  void *current_cont;           // Track current continuation (debug)

#define CLUSTER_IOV_NOT_OPEN               -2
#define CLUSTER_IOV_NONE                   -1
  int iov_map;                  // which iov?

  Ptr<ProxyMutex> read_locked;
  Ptr<ProxyMutex> write_locked;
  SLINK(ClusterVConnection, ready_alink);

  // Data buffer for unmarshaled objects from remote node.
  Ptr<IOBufferData> marshal_buf;

  // Pending write data
  Ptr<IOBufferBlock> write_list;
  IOBufferBlock *write_list_tail;
  int write_list_bytes;
  int write_bytes_in_transit;

  CacheHTTPInfo alternate;
  time_t time_pin;
  int disk_io_priority;
  void set_remote_fill_action(Action *);

  // For VC(s) established via OPEN_READ, we are passed a CacheHTTPInfo
  //  in the reply.
  virtual bool get_data(int id, void *data);    // backward compatibility
  virtual void get_http_info(CacheHTTPInfo **);
  virtual int64_t get_object_size();

  // For VC(s) established via the HTTP version of OPEN_WRITE, additional
  //  data for the VC is passed in a second message.  This additional
  //  data has a lifetime equal to the cache VC
  virtual void set_http_info(CacheHTTPInfo *);

  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();
  bool is_ram_cache_hit()
  {
    return 0;
  }
  virtual int get_header(void **ptr, int *len);
  virtual int set_header(void *ptr, int len);
  virtual int get_single_data(void **ptr, int *len);
  virtual void reenable(VIO *);
  virtual bool is_pread_capable();
};

//
// Cluster operation options
//
#define CLUSTER_OPT_STEAL             0x0001    // allow thread stealing
#define CLUSTER_OPT_IMMEDIATE         0x0002    // require immediate response
#define CLUSTER_OPT_ALLOW_IMMEDIATE   0x0004    // allow immediate response
#define CLUSTER_OPT_DELAY             0x0008    // require delayed response
#define CLUSTER_OPT_CONN_READ         0x0010    // new conn read
#define CLUSTER_OPT_CONN_WRITE        0x0020    // new conn write
#define CLUSTER_OPT_DATA_IS_OCONTROL  0x0040    // data in OutgoingControl
#define CLUSTER_FUNCTION_MALLOCED     -1

struct ClusterRemoteDataHeader
{
  int32_t cluster_function;
};
//
// ClusterProcessor
//
class ClusterAccept;

struct ClusterProcessor
{
  //
  // Public Interface
  //

  // Invoke a function on a remote node
  //   marshal your own data, provide a continuation for timeouts and errors
  //
  // Options: CLUSTER_OPT_DELAY, CLUSTER_OPT_STEAL, CLUSTER_OPT_DATA_IS_OCONTROL
  // Returns: 1 for immediate send, 0 for delayed, -1 for error

  int invoke_remote(ClusterHandler *ch, int cluster_fn_index, void *data, int len, int options = CLUSTER_OPT_STEAL);

  int invoke_remote_data(ClusterHandler *ch, int cluster_fn_index,
                         void *data, int data_len,
                         IOBufferBlock * buf,
                         int logical_channel, ClusterVCToken * token,
                         void (*bufdata_free) (void *), void *bufdata_free_arg, int options = CLUSTER_OPT_STEAL, bool zero_body = false);

  // Pass the data in as a malloc'ed block to be freed by callee
  int invoke_remote_malloced(ClusterHandler *ch, ClusterRemoteDataHeader * data, int len /* including header */ )
  {
    return invoke_remote(ch, CLUSTER_FUNCTION_MALLOCED, data, len);
  }
  void free_remote_data(char *data, int len);

  // Allocate the local side of a remote VConnection.
  // returns a token which can be passed to the remote side
  // through an existing link and passed to attach_remoteVC()
  // if CLUSTER_OPT_IMMEDIATE is set, CLUSTER_DELAYED_OPEN will not be returned
  //
  // Options: CLUSTER_OPT_IMMEDIATE, CLUSTER_OPT_ALLOW_IMMEDIATE
  // Returns: pointer for CLUSTER_OPT_IMMEDIATE
  //            or CLUSTER_DELAYED_OPEN on success,
  //          NULL on failure
  // calls:  cont->handleEvent( CLUSTER_EVENT_OPEN, ClusterVConnection *)
  //         on delayed success.
  //
  // NOTE: the CLUSTER_EVENT_OPEN may be called before "open/connect" returns

#define CLUSTER_DELAYED_OPEN       ((ClusterVConnection*)-1)
#define CLUSTER_NODE_DOWN          ((ClusterVConnection*)-2)
  ClusterVConnection *open_local(Continuation * cont, ClusterMachine * mp, ClusterVCToken & token, int options = 0);

  // Get the other side of a remote VConnection which was previously
  // allocated with open.
  //
  // Options: CLUSTER_OPT_IMMEDIATE, CLUSTER_OPT_ALLOW_IMMEDIATE
  // return a pointer or CLUSTER_DELAYED_OPEN success, NULL on failure
  //
  ClusterVConnection *connect_local(Continuation * cont, ClusterVCToken * token, int channel, int options = 0);
  inkcoreapi bool disable_remote_cluster_ops(ClusterMachine *);

  //
  // Processor interface
  //
  virtual int init();
  virtual int start();

  ClusterProcessor();
  virtual ~ ClusterProcessor();

  //
  // Private
  //
  ClusterAccept *accept_handler;
  Cluster *this_cluster;
  // Connect to a new cluster machine
  void connect(char *hostname, int16_t id = -1);
  void connect(unsigned int ip, int port = 0, int16_t id = -1, bool delay = false);
  // send the list of known machines to new machine
  void send_machine_list(ClusterMachine * m);
  void compute_cluster_mode();
  // Internal invoke_remote interface
  int internal_invoke_remote(ClusterHandler * m, int cluster_fn, void *data, int len, int options, void *cmsg, bool zero_body = false, void *cc = NULL);
};

inkcoreapi extern ClusterProcessor clusterProcessor;

inline Cluster *
this_cluster()
{
  return clusterProcessor.this_cluster;
}

//
// Set up a thread to receive events from the ClusterProcessor
// This function should be called for all threads created to
// accept such events by the EventProcesor.
//
void initialize_thread_for_cluster(EThread * thread);

//
// ClusterFunction Registry
//
//   Declare an instance of this class here to register
//   a function.   In order to allow older versions of software
//   to co-exist with newer versions, always add to the bottom
//   of the list.
//

extern ClusterFunction *ptest_ClusterFunction;

extern ClusterFunction test_ClusterFunction;
extern ClusterFunction ping_ClusterFunction;
extern ClusterFunction ping_reply_ClusterFunction;
extern ClusterFunction machine_list_ClusterFunction;
extern ClusterFunction close_channel_ClusterFunction;
extern ClusterFunction get_hostinfo_ClusterFunction;
extern ClusterFunction put_hostinfo_ClusterFunction;
extern ClusterFunction cache_lookup_ClusterFunction;
//extern ClusterFunction cache_op_ClusterFunction;
extern ClusterFunction cache_op_malloc_ClusterFunction;
//extern ClusterFunction cache_op_result_ClusterFunction;
extern ClusterFunction set_channel_data_ClusterFunction;
extern ClusterFunction post_setchan_send_ClusterFunction;
extern ClusterFunction set_channel_pin_ClusterFunction;
extern ClusterFunction post_setchan_pin_ClusterFunction;
extern ClusterFunction set_channel_priority_ClusterFunction;
extern ClusterFunction post_setchan_priority_ClusterFunction;
extern ClusterFunction default_api_ClusterFunction;

extern ClusterFunctionExt cache_op_ClusterFunction;
extern ClusterFunctionExt cache_op_result_ClusterFunction;

struct ClusterFunctionDescriptor
{
  bool fMalloced;               // the function will free the data
  bool ClusterFunc;             // Process incoming message only
  //   in ET_CLUSTER thread.
  int q_priority;               // lower is higher priority
  ClusterFunctionPtr pfn;
  ClusterFunctionPtr post_pfn;  // msg queue/send callout
};

#define CLUSTER_CMSG_QUEUES		2
#define CMSG_MAX_PRI			0
#define CMSG_LOW_PRI			(CLUSTER_CMSG_QUEUES-1)

#ifndef DEFINE_CLUSTER_FUNCTIONS
extern
#endif
ClusterFunctionDescriptor clusterFunction[0]
#if 0
#ifdef DEFINE_CLUSTER_FUNCTIONS
  = {
  {false, true, CMSG_LOW_PRI, test_ClusterFunction, 0},
  {false, true, CMSG_LOW_PRI, ping_ClusterFunction, 0},
  {false, true, CMSG_LOW_PRI, ping_reply_ClusterFunction, 0},
  {false, true, CMSG_LOW_PRI, machine_list_ClusterFunction, 0},
  {false, true, CMSG_LOW_PRI, close_channel_ClusterFunction, 0},
  {false, false, CMSG_LOW_PRI,
   get_hostinfo_ClusterFunction, 0},    // in HostDB.cc
  {false, false, CMSG_LOW_PRI,
   put_hostinfo_ClusterFunction, 0},    // in HostDB.cc
  {false, true, CMSG_LOW_PRI,
   cache_lookup_ClusterFunction, 0},    // in CacheCont.cc
  {true, true, CMSG_LOW_PRI, cache_op_malloc_ClusterFunction, 0},
  {false, true, CMSG_LOW_PRI, cache_op_ClusterFunction, 0},
  {false, false, CMSG_LOW_PRI, cache_op_result_ClusterFunction, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},   // OBSOLETE
  {false, false, CMSG_LOW_PRI, 0, 0},   // OBSOLETE
  {false, false, CMSG_LOW_PRI, 0, 0},   // OBSOLETE
  {false, true, CMSG_MAX_PRI, set_channel_data_ClusterFunction,
   post_setchan_send_ClusterFunction},
  {false, true, CMSG_MAX_PRI, set_channel_pin_ClusterFunction,
   post_setchan_pin_ClusterFunction},
  {false, true, CMSG_MAX_PRI, set_channel_priority_ClusterFunction,
   post_setchan_priority_ClusterFunction},
   /********************************************
    * RESERVED for future cluster internal use *
    ********************************************/
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},
  {false, false, CMSG_LOW_PRI, 0, 0},

   /*********************************************
    * RESERVED for Cluster RPC API use		*
    *********************************************/
#ifdef NON_MODULAR
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0},
  {true, false, CMSG_LOW_PRI, default_api_ClusterFunction, 0}
#endif
  // ********** ADD NEW ENTRIES ABOVE THIS LINE ************
}
#endif
#endif
;
extern int SIZE_clusterFunction;        // clusterFunction[] entries

//////////////////////////////////////////////////////////////
// Map from Cluster Function code to send queue priority
//////////////////////////////////////////////////////////////
inline int
ClusterFuncToQpri(int cluster_func)
{
  if (cluster_func < 0) {
    return CMSG_LOW_PRI;
  } else {
    return clusterFunction[cluster_func].q_priority;
  }
}

//
// This table had better match the above list
//
#define TEST_CLUSTER_FUNCTION                        0
#define PING_CLUSTER_FUNCTION                        1
#define PING_REPLY_CLUSTER_FUNCTION                  2
#define MACHINE_LIST_CLUSTER_FUNCTION                3
#define CLOSE_CHANNEL_CLUSTER_FUNCTION               4
#define GET_HOSTINFO_CLUSTER_FUNCTION                5
#define PUT_HOSTINFO_CLUSTER_FUNCTION                6
#define CACHE_LOOKUP_CLUSTER_FUNCTION                7
#define CACHE_OP_MALLOCED_CLUSTER_FUNCTION           8
#define CACHE_OP_CLUSTER_FUNCTION                    9
#define CACHE_OP_RESULT_CLUSTER_FUNCTION             10
#define SET_CHANNEL_DATA_CLUSTER_FUNCTION      	     14
#define SET_CHANNEL_PIN_CLUSTER_FUNCTION      	     15
#define SET_CHANNEL_PRIORITY_CLUSTER_FUNCTION  	     16

/********************************************
 * RESERVED for future cluster internal use *
 ********************************************/
#define INTERNAL_RESERVED1_CLUSTER_FUNCTION  	     17
#define INTERNAL_RESERVED2_CLUSTER_FUNCTION  	     18
#define INTERNAL_RESERVED3_CLUSTER_FUNCTION  	     19
#define INTERNAL_RESERVED4_CLUSTER_FUNCTION  	     20
#define INTERNAL_RESERVED5_CLUSTER_FUNCTION  	     21
#define INTERNAL_RESERVED6_CLUSTER_FUNCTION  	     22
#define INTERNAL_RESERVED7_CLUSTER_FUNCTION  	     23
#define INTERNAL_RESERVED8_CLUSTER_FUNCTION  	     24
#define INTERNAL_RESERVED9_CLUSTER_FUNCTION  	     25
#define INTERNAL_RESERVED10_CLUSTER_FUNCTION  	     26
#define INTERNAL_RESERVED11_CLUSTER_FUNCTION  	     27
#define INTERNAL_RESERVED12_CLUSTER_FUNCTION  	     28
#define INTERNAL_RESERVED13_CLUSTER_FUNCTION  	     29
#define INTERNAL_RESERVED14_CLUSTER_FUNCTION  	     30
#define INTERNAL_RESERVED15_CLUSTER_FUNCTION  	     31
#define INTERNAL_RESERVED16_CLUSTER_FUNCTION  	     32
#define INTERNAL_RESERVED17_CLUSTER_FUNCTION  	     33
#define INTERNAL_RESERVED18_CLUSTER_FUNCTION  	     34
#define INTERNAL_RESERVED19_CLUSTER_FUNCTION  	     35
#define INTERNAL_RESERVED20_CLUSTER_FUNCTION  	     36
#define INTERNAL_RESERVED21_CLUSTER_FUNCTION  	     37
#define INTERNAL_RESERVED22_CLUSTER_FUNCTION  	     38
#define INTERNAL_RESERVED23_CLUSTER_FUNCTION  	     39
#define INTERNAL_RESERVED24_CLUSTER_FUNCTION  	     40
#define INTERNAL_RESERVED25_CLUSTER_FUNCTION  	     41
#define INTERNAL_RESERVED26_CLUSTER_FUNCTION  	     42
#define INTERNAL_RESERVED27_CLUSTER_FUNCTION  	     43
#define INTERNAL_RESERVED28_CLUSTER_FUNCTION  	     44
#define INTERNAL_RESERVED29_CLUSTER_FUNCTION  	     45
#define INTERNAL_RESERVED30_CLUSTER_FUNCTION  	     46
#define INTERNAL_RESERVED31_CLUSTER_FUNCTION  	     47
#define INTERNAL_RESERVED32_CLUSTER_FUNCTION  	     48
#define INTERNAL_RESERVED33_CLUSTER_FUNCTION  	     49
#define INTERNAL_RESERVED34_CLUSTER_FUNCTION  	     50

/****************************************************************************
 * Cluster RPC API definitions.						    *
 *									    *
 ****************************************************************************
 * Note: All of the following must be kept in sync with INKClusterRPCKey_t  *
 * 	 definition in ts/ts.h and ts/experimental.h			    *
 ****************************************************************************/

/************************************************
 * RESERVED for Wireless Group			*
 ************************************************/
#define API_F01_CLUSTER_FUNCTION  	     	     51
#define API_F02_CLUSTER_FUNCTION  	     	     52
#define API_F03_CLUSTER_FUNCTION  	     	     53
#define API_F04_CLUSTER_FUNCTION  	     	     54
#define API_F05_CLUSTER_FUNCTION  	     	     55
#define API_F06_CLUSTER_FUNCTION  	     	     56
#define API_F07_CLUSTER_FUNCTION  	     	     57
#define API_F08_CLUSTER_FUNCTION  	     	     58
#define API_F09_CLUSTER_FUNCTION  	     	     59
#define API_F10_CLUSTER_FUNCTION  	     	     60

/************************************************
 * RESERVED for future use			*
 ************************************************/
#define API_F11_CLUSTER_FUNCTION  	     	     61
#define API_F12_CLUSTER_FUNCTION  	     	     62
#define API_F13_CLUSTER_FUNCTION  	     	     63
#define API_F14_CLUSTER_FUNCTION  	     	     64
#define API_F15_CLUSTER_FUNCTION  	     	     65
#define API_F16_CLUSTER_FUNCTION  	     	     66
#define API_F17_CLUSTER_FUNCTION  	     	     67
#define API_F18_CLUSTER_FUNCTION  	     	     68
#define API_F19_CLUSTER_FUNCTION  	     	     69
#define API_F20_CLUSTER_FUNCTION  	     	     70

#define API_F21_CLUSTER_FUNCTION  	     	     71
#define API_F22_CLUSTER_FUNCTION  	     	     72
#define API_F23_CLUSTER_FUNCTION  	     	     73
#define API_F24_CLUSTER_FUNCTION  	     	     74
#define API_F25_CLUSTER_FUNCTION  	     	     75
#define API_F26_CLUSTER_FUNCTION  	     	     76
#define API_F27_CLUSTER_FUNCTION  	     	     77
#define API_F28_CLUSTER_FUNCTION  	     	     78
#define API_F29_CLUSTER_FUNCTION  	     	     79
#define API_F30_CLUSTER_FUNCTION  	     	     80

#define CLUSTER_CACHE_OP_CLUSTER_FUNCTION      (CLUSTER_MSG_START+81)
#define CLUSTER_CACHE_DATA_READ_BEGIN          (CLUSTER_MSG_START+82)
#define CLUSTER_CACHE_DATA_READ_REENABLE       (CLUSTER_MSG_START+83)
#define CLUSTER_CACHE_DATA_WRITE_BEGIN         (CLUSTER_MSG_START+84)
#define CLUSTER_CACHE_HEADER_ONLY_UPDATE       (CLUSTER_MSG_START+85)
#define CLUSTER_CACHE_DATA_CLOSE               (CLUSTER_MSG_START+86)
#define CLUSTER_CACHE_DATA_ABORT               (CLUSTER_MSG_START+87)
#define CLUSTER_CACHE_DATA_WRITE_DONE          (CLUSTER_MSG_START+88)

#define CLUSTER_CACHE_OP_RESULT_CLUSTER_FUNCTION (CLUSTER_MSG_START+89)
#define CLUSTER_CACHE_DATA_READ_DONE           (CLUSTER_MSG_START+90)
#define CLUSTER_CACHE_DATA_ERROR               (CLUSTER_MSG_START+91)

#define CLUSTER_INTERNEL_ERROR                 (CLUSTER_MSG_START+100)
#define CLUSTER_PING_CLUSTER_FUNCTION          (CLUSTER_MSG_START+101)                        1
#define CLUSTER_PING_REPLY_CLUSTER_FUNCTION    (CLUSTER_MSG_START+102)

#define API_STARECT_CLUSTER_FUNCTION           API_F01_CLUSTER_FUNCTION
#define API_END_CLUSTER_FUNCTION               API_F30_CLUSTER_FUNCTION

#define UNDEFINED_CLUSTER_FUNCTION             0xFDEFFDEF

//////////////////////////////////////////////
// Initial cluster connect exchange message
//////////////////////////////////////////////
struct ClusterHelloMessage
{
  uint16_t _NativeByteOrder;
  uint16_t _major;
  uint16_t _minor;
  uint16_t _min_major;
  uint16_t _min_minor;
  int16_t _id;
#ifdef LOCAL_CLUSTER_TEST_MODE
  int16_t _port;
  char _pad[114];               // pad out to 128 bytes
#else
  char _pad[116];               // pad out to 128 bytes
#endif

    ClusterHelloMessage():_NativeByteOrder(1)
  {
    _major = CLUSTER_MAJOR_VERSION;
    _minor = CLUSTER_MINOR_VERSION;
    _min_major = MIN_CLUSTER_MAJOR_VERSION;
    _min_minor = MIN_CLUSTER_MINOR_VERSION;
    _id = 0;
    memset(_pad, '\0', sizeof(_pad));
  }
  int NativeByteOrder()
  {
    return (_NativeByteOrder == 1);
  }
  void AdjustByteOrder()
  {
    if (!NativeByteOrder()) {
      ats_swap16(&_major);
      ats_swap16(&_minor);
      ats_swap16(&_min_major);
      ats_swap16(&_min_minor);
    }
  }
};

///////////////////////////////////////////////////////////////////
// Cluster message header definition.
///////////////////////////////////////////////////////////////////
struct ClusterMessageHeader
{
  uint16_t _InNativeByteOrder;    // always non-zero
  uint16_t _MsgVersion;           // always non-zero

  void _init(uint16_t msg_version)
  {
    _InNativeByteOrder = 1;
    _MsgVersion = msg_version;
  }
  ClusterMessageHeader():_InNativeByteOrder(0), _MsgVersion(0)
  {
  }
  ClusterMessageHeader(uint16_t msg_version) {
    _init(msg_version);
  }
  int MsgInNativeByteOrder()
  {
    return (_InNativeByteOrder == 1);
  }
  int NeedByteSwap()
  {
    return (_InNativeByteOrder != 1);
  }
  int GetMsgVersion()
  {
    if (NeedByteSwap()) {
      return ats_swap16(_MsgVersion);
    } else {
      return _MsgVersion;
    }
  }
};

///////////////////////////////////////////////////////////////////

//
// cluster_ping
//
typedef void (*PingReturnFunction) (ClusterHandler *, void *data, int len);

struct PingMessage:public ClusterMessageHeader
{
  PingReturnFunction fn;        // Note: Pointer to a func
  char data[1];                 // start of data

  enum
  {
    MIN_VERSION = 1,
    MAX_VERSION = 1,
    PING_MESSAGE_VERSION = MAX_VERSION
  };

    PingMessage(uint16_t vers = PING_MESSAGE_VERSION)
:  ClusterMessageHeader(vers), fn(NULL) {
    data[0] = '\0';
  }
  /////////////////////////////////////////////////////////////////////////////
  static int protoToVersion(int protoMajor)
  {
    (void) protoMajor;
    return PING_MESSAGE_VERSION;
  }
  static int sizeof_fixedlen_msg()
  {
    PingMessage *p = 0;
    // Maybe use offsetof here instead. /leif
    return (uintptr_t) (&p->data[0]);
  }
  void init(uint16_t vers = PING_MESSAGE_VERSION) {
    _init(vers);
  }
  inline void SwapBytes()
  {
  }                             // No action, message is always reflected back
  /////////////////////////////////////////////////////////////////////////////
};

inline void
cluster_ping(ClusterHandler *ch, PingReturnFunction fn, void *data, int len)
{
  PingMessage *msg = (PingMessage *)alloca(PingMessage::sizeof_fixedlen_msg() + len);
  msg->init();
  msg->fn = fn;
  memcpy(msg->data, data, len);
  clusterProcessor.invoke_remote(ch, PING_CLUSTER_FUNCTION, (void *) msg, (msg->sizeof_fixedlen_msg() + len));
}

// filled with 0's
extern char channel_dummy_output[DEFAULT_MAX_BUFFER_SIZE];

//
// Private (for testing)
//
ClusterConfiguration *configuration_add_machine(ClusterConfiguration * c, ClusterMachine * m);
ClusterConfiguration *configuration_remove_machine(ClusterConfiguration * c, ClusterMachine * m);
extern bool machineClusterHash;
extern bool boundClusterHash;
extern bool randClusterHash;

void build_cluster_hash_table(ClusterConfiguration *);

inline void
ClusterVC_enqueue_read(Queue<ClusterVConnectionBase, ClusterVConnectionBase::Link_read_link> &q, ClusterVConnectionBase * vc)
{
  ClusterVConnState * cs = &vc->read;
  ink_assert(!cs->queue);
  cs->queue = &q;
  q.enqueue(vc);
}

inline void
ClusterVC_enqueue_write(Queue<ClusterVConnectionBase, ClusterVConnectionBase::Link_write_link> &q, ClusterVConnectionBase * vc)
{
  ClusterVConnState * cs = &vc->write;
  ink_assert(!cs->queue);
  cs->queue = &q;
  q.enqueue(vc);
}

inline void
ClusterVC_remove_read(ClusterVConnectionBase * vc)
{
  ClusterVConnState * cs = &vc->read;
  ink_assert(cs->queue);
  ((Queue<ClusterVConnectionBase, ClusterVConnectionBase::Link_read_link> *)cs->queue)->remove(vc);
  cs->queue = NULL;
}

inline void
ClusterVC_remove_write(ClusterVConnectionBase * vc)
{
  ClusterVConnState * cs = &vc->write;
  ink_assert(cs->queue);
  ((Queue<ClusterVConnectionBase, ClusterVConnectionBase::Link_write_link> *)cs->queue)->remove(vc);
  cs->queue = NULL;
}


struct ClusterCacheVC: public CacheVConnection
{
  static int size_to_init;
  Action _action;
  Ptr<IOBufferData> buf;          // for read
  Ptr<IOBufferData> first_buf;    // the head fragment
  Ptr<IOBufferBlock> blocks; // data available to write

  CacheHTTPInfo alternate;

  VIO vio;
  ink_hrtime start_time;
  CacheFragType frag_type;
  int64_t seek_to;                // pread offset
  int64_t offset;                 // offset into 'blocks' of data to write
  int64_t length;                 // length of data available to write
  int64_t total_len;
  int64_t data_sent;
  int64_t doc_len;

  int doc_pos;                // read position in 'buf'
  int d_len;                  // the length of data in 'buf'

  int closed;
  int recursive;
  int disk_io_priority;
  int probe_depth;
  MessagePriority priority;

  time_t time_pin;
  EThread *initial_thread;  // initial thread open_XX was called on
  ClusterSession cs;
  Event *trigger;
  ContinuationHandler save_handler;


  bool in_progress; //
  bool remote_closed;
  bool session_closed;

  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;
  };
  ClusterCacheVC();
  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) {
    if (enable_cache_empty_http_doc) {
      MIMEField *field = info->m_alt->m_response_hdr.field_find(
          MIME_FIELD_CONTENT_LENGTH, MIME_LEN_CONTENT_LENGTH);
      if (field && !field->value_get_int64())
        f.force_empty = 1;
      else
        f.force_empty = 0;
    } else
      f.force_empty = 0;
    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();
  }
  virtual bool is_read_from_writer()
  {
    return f.read_from_writer_called;
  }
  virtual bool is_pread_capable()
  {
    return !f.read_from_writer_called;
  }
  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);
  int removeEvent(int event, void *data);
};



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
{

};


struct CacheDiffuser : public Continuation
{
  INK_MD5 key;
  MIOBuffer buffer;
#ifdef HTTP_CACHE
  CacheHTTPInfo alternate;
  URL url;
#endif
  CacheVConnection *remote_read_vc;
  CacheVConnection *local_write_vc;
  VIO *read_vio;
  VIO *write_vio;
  char *hostname;
  int64_t doc_size;
  int len;
  CacheFragType frag_type;
  bool terminate;
  ContinuationHandler current_handler;
  ContinuationHandler read_handler;
  ContinuationHandler write_handler;

  CacheDiffuser():remote_read_vc(0), local_write_vc(0), read_vio(0), write_vio(0),
      hostname(0), doc_size(-1), len(0), frag_type(CACHE_FRAG_TYPE_NONE),
      terminate(false), current_handler(0)
  {
    SET_HANDLER(&CacheDiffuser::main_handler);
    read_handler = (ContinuationHandler)&CacheDiffuser::cacheRemoteReadHandler;
    write_handler = (ContinuationHandler)&CacheDiffuser::cacheLocalWriteHandler;
  }

  int cacheLocalWriteHandler(int event, void *e);
  int cacheRemoteReadHandler(int event, void *e);
  Action *do_cache_remote_read();
  Action *do_cache_local_write();
  static Action *do_cache_diffuse(ClusterMachine *m, int opcode, INK_MD5 *key, URL *url, CacheHTTPHdr *request, CacheLookupHttpConfig *params, CacheFragType frag_type);
  int main_handler(int event, void *e);
};

extern ClassAllocator<CacheDiffuser> cacheDiffuserAllocator;

CacheDiffuser * new_CacheDiffuser();
void free_CacheDiffuser(CacheDiffuser *cd);

#define DIFFUSE_BUCKET_SIZE 1021


struct DiffuseTable {
  Que(CacheDiffuser, link) buckts[DIFFUSE_BUCKET_SIZE];
  ink_spinlock locks[DIFFUSE_BUCKET_SIZE];

  DiffuseTable() {
    for(int i = 0; i < DIFFUSE_BUCKET_SIZE; ++i)
      ink_spinlock_init(&locks[i]);
  }

  ~DiffuseTable() {
    for (int i = 0; i < DIFFUSE_BUCKET_SIZE; ++i)
      ink_spinlock_destroy(&locks[i]);
  }
  CacheDiffuser* register_diffuser(INK_MD5 *key) {
    uint32_t indx = ((uint32_t) key->b[0]) % DIFFUSE_BUCKET_SIZE;
    bool found = false;
    CacheDiffuser *ret = NULL;
    ink_spinlock_acquire(&locks[indx]);

    forl_LL(CacheDiffuser, cd, buckts[indx]) {
      if (cd->key == *key) {
        found = true;
        break;
      }
    }

    if (!found) {
      ret = new_CacheDiffuser();
      ret->key = *key;
      buckts[indx].enqueue(ret);
    }

    ink_spinlock_release(&locks[indx]);

    return ret;
  }

  void unregister_diffuser(CacheDiffuser *cd)
  {
    uint32_t indx = ((uint32_t) cd->key.b[0]) % DIFFUSE_BUCKET_SIZE;

    ink_spinlock_acquire(&locks[indx]);
    buckts[indx].remove(cd);
    ink_spinlock_release(&locks[indx]);

    free_CacheDiffuser(cd);
  }
};

inline IOBufferBlock *
clone_IOBufferBlockList(IOBufferBlock *ab, int64_t offset, int64_t len)
{
  IOBufferBlock *b = ab;
  IOBufferBlock *head = NULL;
  IOBufferBlock *clone = NULL;

  while (b && len >= 0) {
    int64_t max_bytes = b->read_avail();
    max_bytes -= offset;
    if (max_bytes <= 0) {
      offset = -max_bytes;
      b = b->next;
      continue;
    }

    if (!head) {
      head = b->clone();
      head->consume(offset);
      clone = head;
    } else {
      clone->next = b->clone();
      clone = clone->next;
    }

    len -= max_bytes;
    b = b->next;
    offset = 0;
  }
  if (clone && len < 0)
    clone->fill(len);
  return head;
}

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 && !in_progress) {
    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 && !in_progress) {
    free_ClusterCacheVC(this);
    return EVENT_DONE;
  } else if (vio.vc_server)
    handleEvent(EVENT_IMMEDIATE, 0);
  return EVENT_DONE;
}

extern ClassAllocator<ClusterCacheVC> clusterCacheVCAllocator;

inline ClusterCacheVC *
new_ClusterCacheVC(Continuation *cont)
{
#ifdef DEBUG
  ink_atomic_increment(&num_of_cluster_cachevc, 1);
#endif
  EThread *t = cont->mutex->thread_holding;
  ClusterCacheVC *c = THREAD_ALLOC(clusterCacheVCAllocator, t);
  c->_action = cont;
  c->initial_thread = t;
  c->mutex = cont->mutex;
  c->start_time = ink_get_hrtime();
  ink_assert(c->trigger == NULL);

  Debug("cluster_cache_new", "new %p", c);
  return c;
}

inline void
free_ClusterCacheVC(ClusterCacheVC *cont)
{
#ifdef DEBUG
  ink_atomic_increment(&num_of_cluster_cachevc, -1);
#endif
  Debug("cluster_cache_free", "free %p", cont);
  ink_debug_assert(cont->mutex->thread_holding == this_ethread());

  if (cont->trigger)
    cont->trigger->cancel();
  ink_assert(!cont->in_progress);

  if (!cont->session_closed)
    cluster_close_session(cont->cs);

  cont->vio.buffer.clear();
  cont->vio.mutex.clear();
#ifdef HTTP_CACHE
  if (cont->vio.op == VIO::WRITE)
    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();

  memset((char *) &cont->vio, 0, cont->size_to_init);

  THREAD_FREE_TO(cont, clusterCacheVCAllocator, this_ethread(), MAX_CACHE_VCS_PER_THREAD);
}
#endif /* _Cluster_h */
