#ifndef __ARPEGGIO_H
#define __ARPEGGIO_H

#include "chordfingerpns.h"
#include "chord.h"

#include <vector>
#include <string>
#include <iostream>
#include <sstream>
#include <map>
#include <ext/hash_map>
#include <ext/hash_set>

using namespace std;


struct Metadata
{
  vector<string> keywords;    // must be sorted!
  string filename;
  string keywordsAsString() const {
    return keywordsVectorToString(keywords);
  }
  
  static string keywordsVectorToString(const vector<string> &k) {
    vector<string> keywords = k;
    ostringstream out;
    sort(keywords.begin(), keywords.end());
    for (vector<string>::const_iterator it = keywords.begin();
         it != keywords.end(); it++) {
      out << *it << "+";
    }
    return out.str();
  };

  u_long bwcost() const {
    u_long r = 0;
    for (vector<string>::const_iterator it = keywords.begin();
         it != keywords.end(); it++) {
      r += it->size() + 1;
    }
    r+= filename.size() + 1;
    return r;
  }
  
  friend bool operator==(const Metadata & x, const Metadata & y);
};

class Arpeggio: public ChordFingerPNS {
  
public:
  typedef ConsistentHash::CHID CHID;

  Arpeggio(IPAddress i, Args& a, LocTable *l = NULL, const char *name=NULL);
  ~Arpeggio() {};
  string proto_name() { return "Arpeggio"; }
  virtual void lookup(Args*);
  virtual void insert(Args*);
  virtual void crash(Args*);
  virtual void join(Args*);
  virtual void insertAndFree(Args*);
  virtual void lookupAndFree(Args*);
  
  struct query_args
  {
    string indexName;
    vector<string> keywords;
  };
  
  struct query_result
  {
    vector<Metadata> results;
    IDMap source;
  };

  struct insert_args
  {
    Metadata entry;
  };

  struct insert_result
  {
    
  };

  struct offerBlocks_args
  {
    set<size_t> hashes;
    IDMap source;
  };

  struct offerBlocks_result
  {
    
  };

  struct requestBlocks_args
  {
    vector<size_t> hashes;
    IDMap source;
  };

  struct requestBlocks_result
  {
    vector<Metadata> entries;
  };

  struct gatewayInsert_args
  {
    Metadata entry;
  };

  struct gatewayInsert_result
  {
    
  };
  
  void remoteQuery(query_args *a, query_result *r);
  void remoteInsert(insert_args *a, insert_result *r);
  void remoteOfferBlocks(offerBlocks_args *a, offerBlocks_result *r);
  void remoteRequestBlocks(requestBlocks_args *a,
                           requestBlocks_result *r);
  void remoteGatewayInsert(gatewayInsert_args *a,
                           gatewayInsert_result *r);
  
protected:
  hash_multimap<string, Metadata> mdIndex;
  hash_map<CHID, string> keywordsHashMap;
  hash_map<size_t, Metadata> metadataHashMap;
  hash_set<string> gatewaySeen;
  int numReplicas;
  int kssK;
  uint syncTimer;
  bool syncRunning;
  uint syncOutstanding;
  
  
  template<class BT, class AT, class RT> 
  bool replicaRPC(const string & indexname, void (BT::* fn)(AT *, RT *),
                  AT *args, RT *ret, u_long type, u_long size,
                  Time timeout = 0, int numReplicas = 0);
  void addMetadataToIndexes(Metadata md);
  ConsistentHash::CHID replicatedKeyspace(ConsistentHash::CHID id,
                                          int numReplicas);
  vector<string> replicatedIndexes(ConsistentHash::CHID id, int numReplicas);
  void replSendOfferBlocks(IDMap node);
  void replGetMissingBlocks(const IDMap & remote,
                            const set<size_t> & blockIDs);
  void replOfferToAll();
  void rescheduleSynchronize(void *);
  vector<vector<string> > generateKeywordSets(const vector<string> & keywords,
                                              int K);
  void testKSS();
  void findAndAddToIndex(const string & indexName,
                         const Metadata & md);
  void recursiveAdd(const Metadata & md);
  void recursiveAddWithIG(const Metadata & md);
  vector<string> tokenize(const string& str,const string& delimiters);
  vector<CHID> getPreds(CHID id, int n);
  void findAndAddToIndexInternal(pair<string, Metadata> args);
  void simulateLeaf(pair<int, int> args);
  void simulateLeafInsert(pair<int, pair<int, int> > args);
  Time next_exponential( u_int mean );
  double pareto(double a, u_int b);
  string getRandomQuery();
  string getFilename(int i);
  int numberOfFilesForHost();
  vector<long> fileRanksForHost(int n);
  void printTime(void *a);
  


  static set<CHID> allArpeggioIDs; // For cheating about predecessors
  static hash_map<ConsistentHash::CHID, Chord::IDMap> allArpeggioIDsToIDMap;
  static FILE *queryFile;
  static FILE *resultsFile;
  
  vector<int> leafStates;
  int epoch;
  Time leafOfflineTime;
  vector<double> leafClassProb;
  vector<Time> leafClassIdleTime;
  vector<double> leafClassOfflineProb;
  vector<int> leafInsertEpoch;
  vector<vector<long> > leafInsertQueue;
  vector<bool> leafInsertGenerated;
};

namespace __gnu_cxx
{
  template<>
  class hash<Metadata>
  {
  public:
    size_t operator()(const Metadata & x)
    {
      hash<string> h;
      return h(x.filename) * 37 + h(x.keywordsAsString());
    }
  };
}

  
#endif