client.c

chord 源码 http://pdos.csail.mit.edu/chord/

/*
 *
 * Copyright (C) 2001  Frank Dabek ([e-mail]), 
 *                     Frans Kaashoek ([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 <sys/time.h>
#include <stdlib.h>

#include <sfsmisc.h>
#include <arpc.h>
#include <crypt.h>

#include <chord_types.h>
#include <chord.h>
#include <route.h>
#include <configurator.h>
#include <location.h>
#include <locationtable.h>
#include <merkle_misc.h>

#include "dhash_common.h"
#include "dhash.h"
#include "dhashcli.h"
#include "dhblock.h"
#include "download.h"
#include "dhash_store.h"

#include <coord.h>
#include <misc_utils.h>
#include <modlogger.h>

// HACK global indicator variable for whether or not to
//      transfer data over TCP pipes.
bool dhash_tcp_transfers = false;

#define warning modlogger ("dhashcli", modlogger::WARNING)
#define info  modlogger ("dhashcli", modlogger::INFO)
#define trace modlogger ("dhashcli", modlogger::TRACE)

#ifdef DMALLOC
#include <dmalloc.h>
#endif
#include <ida.h>

#include "succopt.h"

static void
order_succs (ptr<locationtable> locations,
	     const Coord &me, const vec<chord_node> &succs,
	     vec<chord_node> &out, u_long max = 0);

static struct dhashcli_config_init {
  dhashcli_config_init ();
} dcci;

dhashcli_config_init::dhashcli_config_init ()
{
  bool ok = true;

#define set_int Configurator::only ().set_int

  /** Whether or not to order successors by expected latency */
  ok = ok && set_int ("dhashcli.order_successors", 1);

  assert (ok);
#undef set_int
}


// ---------------------------------------------------------------------------
// DHASHCLI

dhashcli::dhashcli (ptr<vnode> node, ptr<dhash> dh)
  : clntnode (node), ordersucc_ (true), dh (dh)
{
  int ordersucc = 1;
  Configurator::only ().get_int ("dhashcli.order_successors", ordersucc);
  ordersucc_ = (ordersucc > 0);
}


void
dhashcli::retrieve (blockID blockID, cb_ret cb, int options, 
		    ptr<chordID> guess)
{
  chordID myID = clntnode->my_ID ();

  trace << myID << ": retrieve (" << blockID << "): new retrieve.\n";
  
  
  ptr<rcv_state> rs = New refcounted<rcv_state> (blockID, cb);
  
  ptr<dhblock> block = allocate_dhblock (blockID.ctype);

  // We would like to obtain enough successors to provide maximal
  // choice to the client when doing the expensive fetch phase.
  // Unfortunately, we are currently hurt by the fact that there
  // are holes in our successor list: nodes without the block
  // that Chord can't tell us about.  Maybe we need up-calls.
  clntnode->find_succlist (blockID.ID,
			   block->num_fetch (),
			   wrap (this, &dhashcli::retrieve_lookup_cb,rs,block),
			   guess);
}



void
dhashcli::retrieve_lookup_cb (ptr<rcv_state> rs, ptr<dhblock> block,
			      vec<chord_node> succs,
			      route r, chordstat status)
{
  chordID myID = clntnode->my_ID ();
  rs->timemark ();
  rs->r = r;
  
  if (status) {
    trace << myID << ": retrieve (" << rs->key 
          << "): lookup failure: " << status << "\n";
    rs->complete (DHASH_CHORDERR, NULL); // failure
    rs = NULL;
    return;
  }
  strbuf s;
  s << myID << ": retrieve_verbose (" << rs->key << "): route";
  for (size_t i = 0; i < r.size (); i++)
    s << " " << r[i]->id ();
  s << "\n";
  trace << s;
  s.tosuio ()->clear ();
  s << myID << ": retrieve_verbose (" << rs->key << "): succs";
  for (size_t i = 0; i < succs.size (); i++)
    s << " " << succs[i].x;
  s << "\n";
  trace << s;

  doassemble (rs, block, succs);

}

void
dhashcli::doassemble (ptr<rcv_state> rs, ptr<dhblock> block, 
		      vec<chord_node> succs)
{
  chordID myID = clntnode->my_ID ();

  if (succs.size () < block->min_fetch ()) {
    warning << myID << ": retrieve (" << rs->key << "): "
	    << "insufficient number of successors returned!\n";
    rs->complete (DHASH_CHORDERR, NULL); // failure
    rs = NULL;
    return;
  }

  if (ordersucc_) {
    ptr<locationtable> lt = NULL;
    if (rs->succopt)
      lt = clntnode->locations;
    
    // Store list of successors ordered by expected distance.
    // fetch_frag will pull from this list in order.
    order_succs (lt, clntnode->my_location ()->coords (),
		 succs, rs->succs, block->num_put ());
  } else {
    rs->succs = succs;
  }

  // Dispatch min_fetch parallel requests, even though we don't know
  // how many fragments will truly be needed.
  u_int tofetch = block->min_fetch () + 0;
  if (tofetch > rs->succs.size ()) tofetch = rs->succs.size ();
  for (u_int i = 0; i < tofetch; i++)
    fetch_frag (rs, block);
}

void
dhashcli::fetch_frag (ptr<rcv_state> rs, ptr<dhblock> b)
{
  if (rs->completed) return;
  register size_t i = rs->nextsucc;

  chordID myID = clntnode->my_ID ();

  // Ugh. No more successors available.
  if (i >= rs->succs.size ()) {
    // If there are outstanding fragments, there is still hope.
    // XXX Actually, this is a lie. If we know that they will not
    //     provide us with enough total fragments to reconstruct the
    //     block,  e.g. incoming_rpcs + frags.size < NUM_DFRAGS,
    //     we should just give up to the user now.  However for
    //     book keeping purposes, we don't do this.
    if (rs->incoming_rpcs > 0)
      return;
    
    // Should we try harder? Like, try and get more successors and
    // check out the swath? No, let's just fail and have the higher
    // level know that they should retry.
    trace << myID << ": retrieve (" << rs->key 
          << "): out of successors; failing.\n";
    rs->complete (DHASH_NOENT, NULL);
    rs = NULL;
    return;
  }
  
  rs->incoming_rpcs += 1;

  dhash_download::execute (clntnode, dh, rs->succs[i], 
			   blockID(rs->key.ID, rs->key.ctype),
			   wrap (this, &dhashcli::retrieve_fetch_cb, rs, i, b),
			   wrap (this, &dhashcli::on_timeout, rs, b));
  rs->nextsucc += 1;
}

bool
dhashcli::on_timeout (ptr<rcv_state> rs, 
		      ptr<dhblock> b,
		      chord_node dest,
		      int retry_num) 
{
  trace << clntnode->my_ID () << ": retrieve (" << rs->key
	<< "): timeout " << retry_num << " on " << dest << "\n";
  if (retry_num == 0)
    fetch_frag (rs, b);
  return false;
}

void
dhashcli::retrieve_fetch_cb (ptr<rcv_state> rs, u_int i,
			     ptr<dhblock> block_t,
			     ptr<dhash_block> block)
{
  chordID myID = clntnode->my_ID ();

  rs->incoming_rpcs -= 1;
  if (rs->completed) {
    // Here it might just be that we got a fragment back after we'd
    // already gotten enough to reconstruct the block.
    trace << myID << ": retrieve (" << rs->key
          << "): unexpected fragment from " << rs->succs[i] << ", discarding.\n";
    return;
  }

  if (!block) {
    trace << myID << ": retrieve (" << rs->key
	  << "): failed from successor " << rs->succs[i] << "\n";
    rs->errors++;
    fetch_frag (rs, block_t);
    return;
  }

  trace << myID << ": retrieve_verbose (" << rs->key << "): read from "
	<< rs->succs[i].x << "\n";
    
  
#ifdef VERBOSE_LOG  
  bigint h = compute_hash (block->data, block->len);
  trace << myID << ": retrieve (" << rs->key << ") got frag " << i
	<< " with hash " << h << " " << res->compl_res->res.size () << "\n";
#endif /* VERBOSE_LOG */
  
  int err = block_t->process_download (rs->key, block->data);
  if (err) {
    rs->errors++;
    trace << myID << ": retrieve_verbose (" << rs->key << "): err \n";
    fetch_frag (rs, block_t);
  }

  if (block_t->done ()) {
    rs->timemark ();
    str data = block_t->produce_block_data ();
    ptr<dhash_block> ret_block = New refcounted<dhash_block> (data.cstr(),
							      data.len (),
							      rs->key.ctype);
    ret_block->ID = rs->key.ID;
    ret_block->hops = rs->r.size ();
    ret_block->errors = rs->errors;
    ret_block->retries = block->errors;
    
    for (size_t i = 1; i < rs->times.size (); i++) {
      timespec diff = rs->times[i] - rs->times[i - 1];
      ret_block->times.push_back (diff.tv_sec * 1000 +
			      int (diff.tv_nsec/1000000));
    }