merkle_tree_bdb.c

基于DHT的对等协议

{
  merkle_hash prefix (key);
  prefix.clear_suffix (depth);
  DBT pfx; prefix_to_dbt (depth, prefix, &pfx);

  // warnx << "del_node of " << hexdump (pfx.data, pfx.size) << "\n";

  int flags = 0;
  if (!t)
    flags = DB_AUTO_COMMIT;
  int r = nodedb->del (nodedb, t, &pfx, flags);
  if (r && r != DB_NOTFOUND)
    warner ("merkle_tree_bdb::del_node", "nodedb->del", r);

  return r;
}
// }}}
// {{{ merkle_tree_bdb::check_key
bool
merkle_tree_bdb::check_key (const merkle_hash &key, DB_TXN *t)
{
  DBT dkey; mhash_to_dbt (key, &dkey);
  DBT data; bzero (&data, sizeof (data)); data.flags = DB_DBT_PARTIAL;

  int r = keydb->get (keydb, t, &dkey, &data, 0);
  if (r) {
    if (r != DB_NOTFOUND)
      warner ("merkle_tree_bdb::check_key", "keydb->get", r);
    return false;
  }
  return true;
}
// }}}
// {{{ merkle_tree_bdb::insert_key
int
merkle_tree_bdb::insert_key (const merkle_hash &key, DB_TXN *t)
{
  DBT dkey; mhash_to_dbt (key, &dkey);
  DBT data; bzero (&data, sizeof (data));
  int flags = DB_NOOVERWRITE;
  if (!t)
    flags |= DB_AUTO_COMMIT;

  int r = keydb->put (keydb, t, &dkey, &data, flags);
  if (r) {
    if (r == DB_KEYEXIST)
      warner ("merkle_tree_bdb::insert_key", "keydb->put", r);
  }
  return r;
}
// }}}
// {{{ merkle_tree_bdb::remove_key
int
merkle_tree_bdb::remove_key (const merkle_hash &key, DB_TXN *t)
{
  DBT dkey; mhash_to_dbt (key, &dkey);
  int flags = 0;
  if (!t)
    flags = DB_AUTO_COMMIT;
  int r = keydb->del (keydb, t, &dkey, flags);
  if (r && r != DB_NOTFOUND)
    warner ("merkle_tree_bdb::remove_key", "keydb->del", r);
  return r;
}
// }}}
// {{{ merkle_tree_bdb::get_root
merkle_node *
merkle_tree_bdb::get_root ()
{
  static merkle_hash h (0);
  merkle_node_bdb *node = read_node (0, h);
  return node;
}
// }}}
// {{{ merkle_tree_bdb::insert
// Duplicates code more or less from merkle_tree.C
int
merkle_tree_bdb::insert (const chordID &id, DB_TXN *t)
{
  merkle_hash mkey (id);
  return insert (mkey, t);
}

int
merkle_tree_bdb::insert (const chordID &id, const u_int32_t aux, DB_TXN *t)
{
  // When there is auxiliary information, we use the low-order bytes
  // of the key to hold the information.  This is used mostly for mutable
  // data that needs to distinguish whether or not the remote node has
  // the same version as the local node.
  chordID key = id;
  key >>= 32;
  key <<= 32;
  assert (key > 0);
  key |= aux;
  merkle_hash mkey (key);
  return insert (mkey, t);
}

int
merkle_tree_bdb::insert (merkle_hash &key, DB_TXN *parent)
{
  // Run this (potentially) in a nested transaction
  DB_TXN *t = NULL;
  dbe->txn_begin (dbe, parent, &t, 0);

  merkle_hash last_h;
  // Find the nodes that will need to be rehashed.
  vec<merkle_node_bdb *> nodes;
  u_int depth = 0;
  merkle_node_bdb *n = read_node (depth, key, t, DB_RMW);
  while (n != NULL) {
    nodes.push_back (n);
    depth++;
    n = read_node (depth, key, t, DB_RMW);
  }
  n = nodes.back ();
  if (!n->isleaf ()) {
    dbfe_txn_abort (dbe, t);
    fatal << "merkle_tree_bdb::insert: bottom of tree is not a leaf?\n";
  }
  int r = 0;
  while (n->leaf_is_full ()) {
    r = n->leaf2internal (t); // Creates all the children
    if (r)
      goto insert_cleanup;
    n = read_node (n->depth + 1, key, t, DB_RMW);
    nodes.push_back (n);
    assert (n->isleaf ());
  }
  // n is a leaf with enough room for another key

  r = insert_key (key, t);
  if (r) {
    if (r != DB_KEYEXIST && r != ENOSPC)
      fatal << "merkle_tree_bdb::insert: unexpected error.\n";
    goto insert_cleanup;
  }

  // Rehash this path and return to disk.
  while (nodes.size ()) {
    n = nodes.pop_back ();
    assert (n->depth == nodes.size ());
    n->count += 1;
    sha1ctx sc;
    if (n->isleaf ()) {
      assert (n->count > 0 && n->count <= 64);
      vec<merkle_hash> keys;
      get_hash_list (keys, n->depth, n->prefix, t);
      for (u_int i = 0; i < keys.size (); i++)
	sc.update (keys[i].bytes, keys[i].size);
    } else {
      u_int32_t branch = key.read_slot (n->depth);
      n->_child_hash[branch] = last_h;
      for (u_int i = 0; i < 64; i++)
	sc.update (n->_child_hash[i].bytes, n->_child_hash[i].size);
    }
    sc.final (n->hash.bytes);
    last_h = n->hash;

    r = write_node (n, t);
    delete n;
    if (r)
      goto insert_cleanup;
  }
  assert (!r);
  dbfe_txn_commit (dbe, t);
  return r;

insert_cleanup:
  while (nodes.size ())
    delete nodes.pop_back ();
  dbfe_txn_abort (dbe, t);
  return r;
}
// }}}
// {{{ merkle_tree_bdb::remove
int
merkle_tree_bdb::remove (const chordID &id, DB_TXN *t)
{
  merkle_hash mkey (id);
  return remove (mkey, t);
}

int
merkle_tree_bdb::remove (const chordID &id, const u_int32_t aux, DB_TXN *t)
{
  chordID key = id;
  key >>= 32;
  key <<= 32;
  assert (key > 0);
  key |= aux;
  merkle_hash mkey (key);
  return remove (mkey, t);
}
int
merkle_tree_bdb::remove (merkle_hash &key, DB_TXN *parent)
{
  DB_TXN *t = NULL;
  dbe->txn_begin (dbe, parent, &t, 0);

  int r = remove_key (key, t);
  if (r) {
    dbfe_txn_abort (dbe, t);
    return r;
  }

  // Find the nodes that will need to be rehashed.
  vec<merkle_node_bdb *> nodes;
  u_int depth = 0;
  merkle_node_bdb *n = read_node (depth, key, t, DB_RMW);
  while (n != NULL) {
    nodes.push_back (n);
    depth++;
    n = read_node (depth, key, t, DB_RMW);
  }
  n = nodes.back ();
  if (!n->isleaf ()) {
    dbfe_txn_abort (dbe, t);
    fatal << "merkle_tree_bdb::remove: bottom of tree is not a leaf?\n";
  }

  // Rehash this path and return to disk.
  r = 0;
  merkle_hash last_h;
  while (nodes.size ()) {
    n = nodes.pop_back ();
    assert (n->depth == nodes.size ());
    assert (n->count != 0);
    n->count -= 1;
    if (!n->isleaf () && n->count <= 64) {
      r = n->internal2leaf (t);
    }
    if (r) {
      delete n;
      goto remove_cleanup;
    }

    sha1ctx sc;
    if (n->isleaf ()) {
      if (n->count == 0) {
	n->hash = merkle_hash (0);
      } else {
	assert (n->count > 0 && n->count <= 64);
	vec<merkle_hash> keys;
	get_hash_list (keys, n->depth, n->prefix, t);
	assert (keys.size () == n->count);
	for (u_int i = 0; i < keys.size (); i++)
	  sc.update (keys[i].bytes, keys[i].size);
	sc.final (n->hash.bytes);
      }
    } else {
      u_int32_t branch = key.read_slot (n->depth);
      n->_child_hash[branch] = last_h;
      for (u_int i = 0; i < 64; i++)
	sc.update (n->_child_hash[i].bytes, n->_child_hash[i].size);
      sc.final (n->hash.bytes);
    }
    last_h = n->hash;

    r = write_node (n, t);
    delete n;
    if (r)
      goto remove_cleanup;
  }
  assert (!r);
  dbfe_txn_commit (dbe, t);
  return r;
remove_cleanup:
  while (nodes.size ())
    delete nodes.pop_back ();
  dbfe_txn_abort (dbe, t);
  return r;
}
// }}}
// {{{ merkle_tree_bdb::key_exists
bool
merkle_tree_bdb::key_exists (chordID key)
{
  return check_key (key);
}
// }}}
// {{{ merkle_tree_bdb::get_hash_list
int
merkle_tree_bdb::get_hash_list (vec<merkle_hash> &keys,
      u_int depth, const merkle_hash &prefix, DB_TXN *t)
{
  merkle_hash h = prefix;
  h.clear_suffix (depth);
  DBT key; mhash_to_dbt (h, &key);
  DBT content; bzero (&content, sizeof (content)); // Irrelevant

  DBC *cursor;
  int r = keydb->cursor (keydb, t, &cursor, 0);
  if (r) {
    warner ("merkle_tree_bdb::get_hash_list", "cursor open", r);
    (void) cursor->c_close (cursor);
    return r;
  }
  r = cursor->c_get (cursor, &key, &content, DB_SET_RANGE);
  while (!r) {
    merkle_hash h = dbt_to_mhash (key);
    if (!prefix_match (depth, prefix, h))
      break;
    keys.push_back (h);
    bzero (&key, sizeof (key));
    bzero (&content, sizeof (content));
    r = cursor->c_get (cursor, &key, &content, DB_NEXT);
  }
  if (r && r != DB_NOTFOUND)
    warner ("merkle_tree_bdb::get_hash_list", "cursor c_get", r);
  (void) cursor->c_close (cursor);
  return r;
}
// }}}
// {{{ merkle_tree_bdb::database_get_keys
vec<merkle_hash>
merkle_tree_bdb::database_get_keys (u_int depth, const merkle_hash &prefix)
{
  DB_TXN *t;
  dbfe_txn_begin (dbe, &t);
  vec<merkle_hash> keys;
  get_hash_list (keys, depth, prefix, t);
  dbfe_txn_commit (dbe, t);
  return keys;
}
// }}}
// {{{ merkle_tree_bdb::get_keyrange_nowrap
void
merkle_tree_bdb::get_keyrange_nowrap (const chordID &min,
    const chordID &max, u_int n, vec<chordID> &keys)
{
  merkle_hash h (min);
  DBT key; mhash_to_dbt (h, &key);
  DBT content; bzero (&content, sizeof (content)); // Irrelevant

  // This cursor is not transaction protected:
  // get_keyrange is typically used for key exchange
  // in merkle_server, or debugging in merkledump.
  DBC *cursor;
  int r = keydb->cursor (keydb, NULL, &cursor, 0);
  if (r) {
    warner ("merkle_tree_bdb::get_keyrange", "cursor open", r);
    (void) cursor->c_close (cursor);
    return;
  }
  r = cursor->c_get (cursor, &key, &content, DB_SET_RANGE);
  while (!r && keys.size () < n) {
    merkle_hash h = dbt_to_mhash (key);
    chordID c = static_cast<bigint> (h);
    if (!betweenbothincl (min, max, c))
      break;
    keys.push_back (c);
    bzero (&key, sizeof (key));
    bzero (&content, sizeof (content));
    r = cursor->c_get (cursor, &key, &content, DB_NEXT);
  }
  if (r && r != DB_NOTFOUND)
    warner ("merkle_tree_bdb::get_keyrange", "cursor c_get", r);
  (void) cursor->c_close (cursor);
}
// }}}
// {{{ merkle_tree_bdb::lookup_exact
merkle_node *
merkle_tree_bdb::lookup_exact (u_int depth, const merkle_hash &key)
{
  return read_node (depth, key);
}
// }}}
// {{{ merkle_tree_bdb::lookup (no max depth)
merkle_node *
merkle_tree_bdb::lookup (u_int depth, const merkle_hash &key)
{
  return lookup (&depth, MAX_DEPTH, key);
}
// }}}
// {{{ merkle_tree_bdb::lookup (max depth)
merkle_node *
merkle_tree_bdb::lookup (u_int *depth, u_int max_depth, const merkle_hash &key)
{
  DB_TXN *t = NULL;
  dbfe_txn_begin (dbe, &t);
  merkle_node_bdb *n = NULL;
  // Start at the deepest allowed position and search upwards.
  for (*depth = max_depth; *depth >= 0; (*depth)--) {
    n = read_node (*depth, key, t);
    if (n || !*depth)
      break;
  }
  dbfe_txn_commit (dbe, t);
  return n;
}
// }}}
// {{{ merkle_tree_bdb::lookup_release
void
merkle_tree_bdb::lookup_release (merkle_node *n)
{
  delete n;
}
// }}}
// {{{ merkle_tree_bdb::check_invariants
void
merkle_tree_bdb::check_invariants ()
{
  // Must override merkle_tree's implementation to place whole
  // check in a transaction and to ensure that extra invariants hold.
  DB_TXN *t = NULL;
  dbfe_txn_begin (dbe, &t);
  merkle_node_bdb *root = read_node (0, 0, t);
  verify_subtree (root, t);
  delete root;
  dbfe_txn_commit (dbe, t);
}

void
merkle_tree_bdb::verify_subtree (merkle_node_bdb *n, DB_TXN *t)
{
  u_int64_t ncount (0);
  sha1ctx sc;
  if (!n->isleaf ()) {
    merkle_hash nprefix (n->prefix);
    for (int i = 0; i < 64; i++) {
      nprefix.write_slot (n->depth, i);
      merkle_node_bdb *child = read_node (n->depth + 1, nprefix, t);
      ncount += child->count;
      verify_subtree (child, t);
      if (n->_child_hash[i] != child->hash) {
	fatal << "merkle_tree: child hash mismatch "
	      << n->_child_hash[i] << " vs " << child->hash
	      << " at "
	      << (n->isleaf () ? "leaf" : "non-leaf")
	      << " at depth " << n->depth << " and prefix "
	      << n->prefix << "\n";
      }
      sc.update (child->hash.bytes, child->hash.size);
      delete child;
    }
  } else {
    vec<merkle_hash> keys;
    get_hash_list (keys, n->depth, n->prefix, t);
    ncount = keys.size ();
    assert (n->count <= 64);
    for (u_int i = 0; i < keys.size (); i++) {
      sc.update (keys[i].bytes, keys[i].size);
    }
  }
  if (ncount != n->count) {
    fatal << "merkle_tree: counted " << ncount
          << " children but had recorded " << n->count << " at "
          << (n->isleaf () ? "leaf" : "non-leaf")
	  << " at depth " << n->depth << " and prefix "
          << n->prefix << "\n";
  }

  merkle_hash nhash;
  if (n->count)
    sc.final (nhash.bytes);
  if (nhash != n->hash) {
    warn << "nhash   = " << nhash << "\n";
    warn << "n->hash = " << n->hash << "\n";
    warn << "n->count= " << n->count << "\n";
    fatal << "nhash of "
	  << (n->isleaf () ? "leaf" : "non-leaf")
	  << " didn't match at depth " << n->depth << " and prefix "
	  << n->prefix << "\n";
  }
}
// }}}
// }}}

/* vim:set foldmethod=marker: */