dbm.c

基于DHT的对等协议

/*
 *
 * Copyright (C) 2003--2007 Emil Sit ([e-mail]),
 * Copyright (C) 2001--2004 Frank Dabek ([e-mail]), 
 *            Massachusetts Institute of Technology
 * 
 *
 *  Permission is hereby granted, free of charge, to any person obtaining
 *  a copy of this software and associated documentation files (the
 *  "Software"), to deal in the Software without restriction, including
 *  without limitation the rights to use, copy, modify, merge, publish,
 *  distribute, sublicense, and/or sell copies of the Software, and to
 *  permit persons to whom the Software is furnished to do so, subject to
 *  the following conditions:
 *
 *  The above copyright notice and this permission notice shall be
 *  included in all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 *  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 *  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 *  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 *  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 *  OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 *  WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 *
 */

#include <async.h>
#include <dhash_common.h>
#include <dhashclient.h>
#include <dhblock.h>
#include <dhblock_keyhash.h>
#include <sfscrypt.h>
#include <sys/time.h>

// {{{ Declarations
class harness_t {
public:
  enum test_mode_t { FETCH, STORE };

protected:
  str ctlsock;	        // How to connect to lsd/dhash
  dhash_ctype ctype;    // Type of objects to store
  unsigned int nobj;    // Number of objects to store
  unsigned int objsize; // Size of each object stored
  unsigned int lifetime;// How long before objects expire
  unsigned int window;  // Number of operations to run in parallel
  unsigned int seed;    // Random seed

  unsigned int niters;  // Mutable objects may go multiple rounds

  dhashclient *dhash;
  ptr<option_block> options;

private:
  FILE *outfile;
  FILE *bwfile;

  test_mode_t mode;

  int bps;
  timecb_t *measurecb;
  void measure_bw ();

  struct timeval start;
  struct timeval end;
  unsigned int nundone;
  unsigned int lastsent;
  unsigned int out;
  vec<bool> done;

  void tcp_connect_cb (int fd);
  void connected ();
  void go (unsigned int iter);
  void init_iter ();
  void storecb (int iter, unsigned int dx, u_int64_t start,
      dhash_stat status, ptr<insert_info> i);
  void fetchcb (int iter, unsigned int i, u_int64_t start,
      dhash_stat stat, ptr<dhash_block> blk, vec<chordID> path);
  void complete ();

  void eofhandler ();

protected:
  vec<chordID> IDs;
  virtual void prepare_test_data () = 0;
  virtual void store_one (unsigned int iter, unsigned int i, cbinsertgw_t cb) = 0;
  virtual void fetch_one (unsigned int iter, unsigned int i, cb_cret cb) = 0;
  virtual bool verify_one (unsigned int iter, unsigned int i, ptr<dhash_block> blk) = 0;

public:
  static harness_t *alloc (str cs, str ctype);
  harness_t (str cs, dhash_ctype ct);
  virtual ~harness_t ();

  virtual bool parse_argv (const vec<str> &argv);

  virtual void run (test_mode_t mode);
}; 

struct chash_harness_t : public harness_t {
  chash_harness_t (str cs) : harness_t (cs, DHASH_CONTENTHASH) {}
  ~chash_harness_t () {}
  vec<str> data;
  void prepare_test_data ();
  void store_one (unsigned int iter, unsigned int i, cbinsertgw_t cb);
  void fetch_one (unsigned int iter, unsigned int i, cb_cret cb);
  bool verify_one (unsigned int iter, unsigned int i, ptr<dhash_block> blk);
};

struct noauth_harness_t : public harness_t {
  noauth_harness_t (str cs) : harness_t (cs, DHASH_NOAUTH) {}
  ~noauth_harness_t () {}
  vec<vec<str> > data;
  void prepare_test_data ();
  void store_one (unsigned int iter, unsigned int i, cbinsertgw_t cb);
  void fetch_one (unsigned int iter, unsigned int i, cb_cret cb);
  bool verify_one (unsigned int iter, unsigned int i, ptr<dhash_block> blk);
};

struct keyhash_harness_t : public harness_t {
  static str rawkey;
  keyhash_harness_t (str cs) :
    harness_t (cs, DHASH_KEYHASH), sk (NULL) {}
  ~keyhash_harness_t () {}
  vec<vec<keyhash_payload> > data;
  ptr<sfspriv> sk;
  sfs_pubkey2 pk;
  void prepare_test_data ();
  void store_one (unsigned int iter, unsigned int i, cbinsertgw_t cb);
  void fetch_one (unsigned int iter, unsigned int i, cb_cret cb);
  bool verify_one (unsigned int iter, unsigned int i, ptr<dhash_block> blk);
};
// }}}

// {{{ Utility functions
void
usage (int stat) 
{
  warnx << "Usage: " << progname << " dhashsock store|fetch <ctype>"
    << " <nobj> <objsize> [option=value ...]\n";
  warnx << "Valid options include:\n"
    "\tlifetime=<count>[s|m|h|d|w|M|y]\n"
    "\tniters=<iterations>\n"
    "\tnops=<rpc window size>\n"
    "\tseed=<randseed>\n"
    "\tout=<outfile>\n"
    "\tbw=<enablebw+logfile>\n";
  if (stat < 0)
    return;
  exit (stat);
}

u_int64_t
getusec ()
{
  timeval tv;
  gettimeofday (&tv, NULL);
  return tv.tv_sec * INT64(1000000) + tv.tv_usec;
}

chordID
make_data (mstr &d)
{
  // size must be word sized
  //  assert (datasize % sizeof (long) == 0);
  
  char *rd = d.cstr ();
  for (unsigned int i = 0; i < d.len (); i++) 
    rd[i] = random ();
  rd[d.len () - 1] = 0;

  return compute_hash (rd, d.len ());
}

void
cleanup (harness_t *harness)
{
  if (harness)
    delete harness;
  exit (0);
}
// }}}

// {{{ harness_t
// {{{ Object initialization and cleanup
harness_t *
harness_t::alloc (str cs, str ctype)
{
  if (ctype == "chash" || ctype == "contenthash") {
    return New chash_harness_t (cs);
  }
  else if (ctype == "noauth") {
    return New noauth_harness_t (cs);
  }
  else if (ctype == "keyhash") {
    return New keyhash_harness_t (cs);
  }
  return NULL;
}

harness_t::harness_t (str cs, dhash_ctype ct) :
  ctlsock (cs),
  ctype (ct),
  nobj (32),
  objsize (8192),
  lifetime (0),
  window (1),
  seed (0),
  niters (1),
  dhash (NULL),
  options (NULL),
  outfile (stdout),
  bwfile (stdout),
  mode (STORE),
  bps (0),
  measurecb (NULL),
  nundone (nobj),
  lastsent (0),
  out (0)
{
}

harness_t::~harness_t ()
{
  if (dhash) {
    delete dhash;
    dhash = NULL;
  }
  for (size_t i = 0; i < done.size (); i++) {
    if (!done[i]) 
      warn << (IDs[i]>>144) << " not done\n";
  }
  if (outfile) {
    fclose (outfile);
  }
  if (bwfile) {
    fclose (bwfile);
  }
}

void
harness_t::eofhandler () 
{
  warn << "Unexpected EOF on transport: block too large?\n";
  complete ();
}

bool
harness_t::parse_argv (const vec<str> &argv)
{
  if (argv.size () < 2)
    usage (1);
  nobj = atoi (argv[0]);
  objsize = atoi (argv[1]);

  if (argv.size () > 2) {
    for (size_t i = 2; i < argv.size (); i++) {
      char *eoff = strchr (argv[i].cstr (), '=');
      if (!eoff)
	usage (1);
      str name = substr (argv[i], 0, eoff - argv[i].cstr ());
      str val  = str (eoff + 1);
      if (name == "nops") {
	window = atoi (val.cstr ());
      }
      else if (name == "seed") {
	seed = strtoul (val.cstr (), NULL, 10);
      }
      else if (name == "niters") {
	niters = atoi (val.cstr ());
      }
      else if (name == "out") {
	if (val != "-")
	  outfile = fopen (val, "w");
      }
      else if (name == "bw") {
	if (val != "-")
	  bwfile = fopen (val, "w");
	measurecb = delaycb (1, wrap (this, &harness_t::measure_bw));
      }
      else if (name == "lifetime") {
	u_int32_t factor = 1; // Default of seconds.
	char f = val[val.len () - 1];
	if (!isdigit (f))
	  val = substr (val, 0, val.len () - 1);
	char *err = NULL;
	lifetime = strtoul (val.cstr (), &err, 10);
	if (*err != '\0') {
	  warnx << "Bad lifetime " << val << "\n";
	  usage (1);
	}
	switch (f) {
	  case 'y': factor = 365 * 86400 + 6 * 3600; break;
	  case 'M': factor = 30 * 86400;     break;
	  case 'w': factor = 7 * 86400;      break;
	  case 'd': factor = 86400;          break;
	  case 'h': factor = 3600;	     break;
	  case 'm': factor = 60;	     break;
	  case 's': factor = 1;              break;
	  default:
	    if (!isdigit (f)) {
	      warnx ("Bad life factor '%c'\n", f); 
	      usage (1);
	    }
	}
	lifetime *= factor;
	options = New refcounted<option_block> ();
	options->flags = DHASHCLIENT_EXPIRATION_SUPPLIED;
	options->expiration = time (NULL) + lifetime;
      }
      else {
	warnx << "Unrecognized option " << name << "\n";
	usage (1);
      }
    }
  }

  srandom (seed);
  nundone = nobj;
  return true;
}
// }}}
// {{{ Bandwidth measurement
void
harness_t::measure_bw (void)
{
  float bw = objsize * bps; // we get called every second
  bw /= 1024; // convert to K.
  unsigned long long usecs = (getusec ()/1000);
  fprintf (bwfile, "%llu\t%6.2f KB/s\n", usecs, bw);
  bps = 0;
  measurecb = delaycb (1, wrap (this, &harness_t::measure_bw));
}
// }}}
// {{{ Initial connection
void
harness_t::run (test_mode_t m)
{
  mode = m;
  const char *cstr = ctlsock.cstr ();
  if (strchr (cstr, ':')) {
    char *port = strchr (cstr, ':');
    str host = substr (ctlsock, 0, port - cstr);