server.c

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

    nfsres.resok->count = 0;
    nfsres.resok->eof = 1;
    nfsres.resok->file_attributes.set_present(1);
    *nfsres.resok->file_attributes.attributes = ro2nfsattr(data->inode);
    sbp->reply(&nfsres);
  }
  else {
    //XXX if a read straddles two blocks we will do a short read 
    ///   i.e. only read the first block
    size_t block = ra->offset / fsinfo.blocksize;
    read_file_data (false, block, data->inode->reg, 
		    wrap (this, &chord_server::read_data_cb,
			  sbp, data, ID));

    size_t ftch_lim = block + PREFETCH_BLOCKS;
    size_t file_bytes = roundup (data->inode->reg->size, fsinfo.blocksize);
    size_t file_blks = file_bytes / fsinfo.blocksize;

    for (size_t b = block + 1; b <= ftch_lim && b < file_blks; b++)
      read_file_data (true, b, data->inode->reg, wrap (this, &chord_server::ignore_data_cb));
  }
}

void
chord_server::ignore_data_cb (ptr<sfsro_data> data)
{
}

void
chord_server::read_data_cb (nfscall *sbp, ptr<sfsro_data> inode, chordID inodeID,
			    ptr<sfsro_data> data)
{
  if (!data)
    sbp->error (NFS3ERR_STALE);
  else if (data->type != SFSRO_FILEBLK)
    sbp->error (NFS3ERR_STALE);		
  else {
    char *buf = data->data->base();
    size_t size = data->data->size();

    read3args *ra = sbp->Xtmpl getarg<read3args> ();
    read3res nfsres (NFS3_OK);
    
    fattr3 fa = ro2nfsattr(inode->inode);
    unsigned int blocksize = (unsigned int)fsinfo.blocksize;
    
    size_t start = ra->offset % blocksize;
    size_t n = MIN (MIN (ra->count, size), size - start);
    
    nfsres.resok->count = n;
    nfsres.resok->eof = (fa.size >= ra->offset + n) ? 1 : 0;

    // XXX avoid a data copy by marshalling directly into the xdr buffer
    nfsres.resok->data.setsize(n);
    memcpy (nfsres.resok->data.base(), buf + start, 
	    nfsres.resok->data.size ()); 
    nfsres.resok->file_attributes.set_present(1);
    *nfsres.resok->file_attributes.attributes = fa;
    sbp->reply (&nfsres);
  }
}






// ------------------------ util -------------------


chordID 
chord_server::nfsfh_to_chordid (nfs_fh3 *nfh) 
{
  chordID n;
  mpz_set_rawmag_be (&n, nfh->data.base (), nfh->data.size ());
  return n;
}

void
chord_server::chordid_to_nfsfh (chordID *n, nfs_fh3 *nfh)
{
  str raw = n->getraw ();
  nfh->data.setsize (raw.len ());
  memcpy (nfh->data.base (), raw.cstr (), raw.len ());
}

chordID
chord_server::sfshash_to_chordid (sfs_hash *fh) 
{
  bigint n;
  mpz_set_rawmag_be (&n, fh->base (), fh->size ());
  return n;
}

static fattr3 
ro2nfsattr (sfsro_inode *si)
{
  fattr3 ni;
  // XXX this is duplicated code.  It 
  // should be fixed in the spec file.

  if (si->type == SFSROLNK) {
    ni.nlink = si->lnk->nlink;
    ni.size = si->lnk->dest.len ();
    ni.used = 0;
    ni.mtime = si->lnk->mtime;
    ni.ctime = si->lnk->ctime;
    ni.atime = si->lnk->mtime;
    ni.fileid = path2fileid(si->lnk->path);
  } else {
    ni.nlink = si->reg->nlink;
    ni.size = si->reg->size;
    ni.used = si->reg->used;
    ni.mtime = si->reg->mtime;
    ni.ctime = si->reg->ctime;
    ni.atime = si->reg->mtime;
    ni.fileid = path2fileid(si->reg->path);
  }

  /* Below are synthesized attributes */
  ni.mode = 0444;
  
  switch (si->type) {
  case SFSROREG_EXEC:
    ni.mode = 0555;
  case SFSROREG:
    ni.type = NF3REG;
    break;
  case SFSRODIR:
    ni.mode = 0555;
    ni.type = NF3DIR;
    break;
  case SFSRODIR_OPAQ:
    ni.mode = 0111;
    ni.type = NF3DIR;
    break;
  case SFSROLNK:
    ni.type = NF3LNK;
    break;
  default:
    fatal ("server::ro2nfsattr: Unknown si->type %X\n",
	   si->type);
    break;
  }
    
  ni.uid = sfs_uid;
  ni.gid = sfs_gid;
  ni.rdev.minor = 0;
  ni.rdev.major = 0;
  ni.fsid = 0;

  return ni;
}


// ------------------------ lookup ------------------

struct lookup_state {
  ptr<sfsro_data> dirdata;
  sfsro_inode_reg *dirinode;
  str component;
  cblookup_t cb;

  ssize_t curblock; // linear scan state variables
  ssize_t maxblock;

  lookup_state (ptr<sfsro_data> dirdata, str component, cblookup_t cb,
		size_t blocksize) :
    dirdata (dirdata), component (component), cb (cb)
  {
    assert (dirdata->type == SFSRO_INODE); 
    assert (dirdata->inode->type == SFSRODIR ||
	    dirdata->inode->type == SFSRODIR_OPAQ);
    dirinode = dirdata->inode->reg;

    curblock = 0;
    maxblock = -1; 
    if (dirinode->direct.size() > 0)
      maxblock = (dirinode->direct.size() - 1)  / blocksize;
  }
};


void
chord_server::lookup(ptr<sfsro_data> dirdata, chordID dirID, str component, cblookup_t cb)
{
  if (!dirdata) {
    (*cb) (NULL, 0, NFS3ERR_STALE);
  } else if (dirdata->type != SFSRO_INODE) {
    (*cb) (NULL, 0, NFS3ERR_IO);
  } else if (dirdata->inode->type != SFSRODIR && 
	     dirdata->inode->type != SFSRODIR_OPAQ) {  
    (*cb) (NULL, 0, NFS3ERR_IO);    
  } else if (component == ".") {
    (*cb) (dirdata, dirID, NFS3_OK);
  } else if (component == "..") {
    namei (dirdata->inode->reg->path, cb);
  } else {
    ref<lookup_state> st =
      New refcounted<lookup_state> (dirdata, component, cb, fsinfo.blocksize);
    lookup_scandir_nextblock(st);
  }
}


void
chord_server::lookup_scandir_nextblock(ref<lookup_state> st)
{
  if (st->curblock > st->maxblock) {
    (*st->cb) (NULL, 0, NFS3ERR_NOENT);
  } else {
    size_t curblock = st->curblock;
    st->curblock++;
    read_file_data (false, curblock, st->dirinode, 
    		    wrap (this, &chord_server::lookup_scandir_nextblock_cb, st));
  }
}


void
chord_server::lookup_scandir_nextblock_cb(ref<lookup_state> st, ptr<sfsro_data> dat)
{
  if (dat == NULL || dat->type != SFSRO_DIRBLK) 
    (*st->cb) (NULL, 0, NFS3ERR_STALE);
  else {
    sfsro_dirent *dirent = dirent_lookup (st->component, dat->dir);
    if (dirent) {
      chordID componentID = sfshash_to_chordid (&dirent->fh);
      fetch_data (false, componentID,
		  wrap (this, &chord_server::lookup_component_inode_cb, st, componentID));
    } else {
      lookup_scandir_nextblock(st);
    }
  }
}



void
chord_server::lookup_component_inode_cb (ref<lookup_state> st, chordID ID, ptr<sfsro_data> data)
{
  if (!data) 
    (*st->cb) (NULL, 0, NFS3ERR_STALE);
  else if (data->type != SFSRO_INODE) 
    (*st->cb) (NULL, 0, NFS3ERR_IO);
  else {
    (*st->cb) (data, ID, NFS3_OK);
  }
}


sfsro_dirent *
chord_server::dirent_lookup (str name, sfsro_directory *dir)
{
  assert (name != ".");
  assert (name != "..");

  sfsro_dirent *e = NULL, *e_prev = NULL;

  for (e = e_prev = dir->entries; e; e = e->nextentry) {

     if (name == e->name)
	 return e;

     if ((e_prev->name < name) && (name < e->name))
	 return NULL;
     e_prev = e;
  }
  return NULL;
}




// ---------------------------- namei -----------------------

struct namei_state {
  str path;
  vec<str> components;
  cbnamei_t cb;

  namei_state (str path, cbnamei_t cb) : path (path), cb (cb) {
    splitpath (components, path);
  };
};

// NAMEI translate a path to an inode
void
chord_server::namei (str path, cbnamei_t cb)
{
  ref<namei_state> st = New refcounted<namei_state>(path, cb);
  namei_iter (st, this->rootdir, this->rootdirID);
}


void
chord_server::namei_iter (ref<namei_state> st, ptr<sfsro_data> inode, chordID inodeID)
{
  if (!st->components.size ()) {
    (*st->cb) (inode, inodeID, NFS3_OK);
  } else if (inode->inode->type != SFSRODIR &&
	     inode->inode->type != SFSRODIR_OPAQ) {  
    (*st->cb) (NULL, 0, NFS3ERR_IO);
  } else {
    str component = st->components.pop_front ();
    lookup (inode, inodeID, component, wrap (this, &chord_server::namei_iter_cb, st));
  }
}


void
chord_server::namei_iter_cb (ref<namei_state> st,
			     ptr<sfsro_data> data, chordID dataID, nfsstat3 status)
{
  if (status != NFS3_OK) {
    (*st->cb) (NULL, 0, NFS3ERR_STALE);
  } else {
    namei_iter (st, data, dataID);
  }
}


static void
splitpath (vec<str> &out, str in)
{
  const char *p = in.cstr ();
  const char *e = p + in.len ();
  const char *n;

  for (;;) {
    while (*p == '/')
      p++;
    for (n = p; n < e && *n != '/'; n++)
      ;
    if (n == p)
      return;
    out.push_back (str (p, n - p));
    p = n;
  }
    
}


// -------------------------- bmap --------------------------

void
chord_server::bmap (bool pfonly, size_t block, sfsro_inode_reg *inode, cbbmap_t cb)
{
  chordID ID;

  // XXX this is the same calculation that is at the end of sfsrodb/sfsrodb.C
  u_int32_t nfh = (fsinfo.blocksize - 100) / (20*2);

  if (block < SFSRO_NDIR) {
    ID = sfshash_to_chordid (&(inode->direct[block]));
    (*cb) (ID, true);
  }
  else if (block < SFSRO_NDIR + nfh) {
    ID = sfshash_to_chordid (&(inode->indirect));
    unsigned int slotno1 = block - SFSRO_NDIR;
    bmap_recurse (pfonly, cb, slotno1, ID, true);
  }
  else if (block < SFSRO_NDIR + nfh * nfh + nfh) {
   // dindirect starts SFSRO_NDIR + nfh
    unsigned int index_in_dindirect = block - (SFSRO_NDIR + nfh);
    unsigned int slotno1 = index_in_dindirect / nfh;
    unsigned int slotno2 = index_in_dindirect % nfh;

    ID = sfshash_to_chordid (&(inode->double_indirect));
    bmap_recurse (pfonly, wrap (this, &chord_server::bmap_recurse, pfonly, cb, slotno2),
		  slotno1, ID, true);
  }
  else if (block < SFSRO_NDIR + nfh * nfh * nfh + nfh * nfh + nfh) {
    // tindirect starts SFSRO_NDIR + nfh * nfh + nfh
    unsigned int index_in_tindirect = block - (SFSRO_NDIR + nfh * nfh + nfh);
    unsigned int slotno1 = index_in_tindirect / (nfh * nfh);
    unsigned int slotno2 = (index_in_tindirect % (nfh * nfh)) / nfh;
    unsigned int slotno3 = index_in_tindirect % nfh;

    ID = sfshash_to_chordid (&(inode->triple_indirect));
    bmap_recurse (pfonly, wrap (this, &chord_server::bmap_recurse, pfonly,
				wrap (this, &chord_server::bmap_recurse, pfonly,
				      cb, slotno3), slotno2), slotno1, ID, true);
  }
  else {
    warn << "bmap: block out of range: " << block << "\n";
    (*cb) (0, false);
  }
}

void
chord_server::bmap_recurse(bool pfonly, cbbmap_t cb, u_int32_t slotno, chordID ID, bool success)
{
  if (success) {
    fetch_data (pfonly, ID,
		wrap (this, &chord_server::bmap_recurse_data_cb, pfonly, cb, slotno));
  } else {
    (*cb) (0, false);
  }
}

void
chord_server::bmap_recurse_data_cb (bool pfonly, cbbmap_t cb,
				    u_int32_t slotno, ptr<sfsro_data> dat)
{
  if (dat) {
    assert (dat->type == SFSRO_INDIR);
    sfsro_indirect *indirect =  dat->indir;
    sfs_hash *fh = &indirect->handles[slotno];
    chordID ID =  sfshash_to_chordid (fh);
    (*cb) (ID, true);
  } else {
    (*cb) (0, false);
  }
}





// ------------------- read block from file (or directory) --------------------

// XXX should this call check the type of the block? 
// XXX the ref counting is questionable here.. 
void
chord_server::read_file_data (bool pfonly, size_t block, sfsro_inode_reg *inode, cbdata_t cb)
{
  // the caller must ensure that block is within the size of the file

  // convert logical file block to chordID
  bmap (pfonly, block, inode, wrap (this, &chord_server::read_file_data_bmap_cb, pfonly, cb));
}



void
chord_server::read_file_data_bmap_cb (bool pfonly, cbdata_t cb, chordID ID, bool success)
{
  if (success) {
    fetch_data (pfonly, ID, cb);
  } else {
    (*cb) (NULL);
  }
}


// ------------------------ raw data fetch ------------------

// pfonly:
//   -- if the data is cached, then call it back.
//   -- otherwise, ensure the data is read in (it might already be incoming),
//      but don't call it back.
//
void
chord_server::fetch_data (bool pfonly, chordID ID, cbdata_t cb)
{
  fetch_data (pfonly, ID, DHASH_CONTENTHASH, cb);
}

void
chord_server::fetch_data (bool pfonly, chordID ID, dhash_ctype ct, cbdata_t cb)
{
  if (ptr<sfsro_data> dat = data_cache [ID]) {
    (*cb) (dat);
  }
  else if (vec<cbdata_t> *l = pf_waiters[ID]) {
    if (!pfonly) {
      l->push_back (cb);
    }
  }
  else {
    pf_waiters.insert (ID);
    vec<cbdata_t> *l = pf_waiters[ID];
    l->push_back (cb);
    dh.retrieve (ID, ct, wrap (this, &chord_server::fetch_data_cb, ID, ct, cb));
  }
}

void
chord_server::fetch_data_cb (chordID ID, dhash_ctype ct, cbdata_t cb, 
			     dhash_stat stat,
			     ptr<dhash_block> blk,
			     vec<chordID> path)
{
  ptr<sfsro_data> data = NULL;

  if (blk) {
    switch (ct) {
    case DHASH_CONTENTHASH:
      {
	data = New refcounted<sfsro_data>;
	xdrmem x (blk->data, blk->data.len (), XDR_DECODE);
	if (xdr_sfsro_data (x.xdrp (), data)) {
	  data_cache.insert (ID, data);
	} else {
	  warn << "Couldn't unmarshall data\n";
	}
	break;
      }
    case DHASH_KEYHASH:
      {
	data = New refcounted<sfsro_data>;
	ptr<keyhash_payload> p = keyhash_payload::decode (blk);
	xdrmem x (p->buf ().cstr (), p->buf ().len (), XDR_DECODE);
	if (xdr_sfsro_data (x.xdrp (), data)) {
	  data_cache.insert (ID, data);
	} else {
	  warn << "Couldn't unmarshall pk data\n";
	}
	break;
      }
    default:
      warn << "WTF\n";
      break;
    }
  }

  vec<cbdata_t> *l = pf_waiters[ID];
  assert (l); 
  
  while (l->size ()) {
    cbdata_t cb = l->pop_back (); 
    (*cb) (data);
  }
  
  pf_waiters.remove (ID);
}