<commit_before>#include <chord.h>
#include "merkle_syncer.h"
#include "qhash.h"
#include "async.h"
#include <id_utils.h>
#include <comm.h>

#include <modlogger.h>
#define warning modlogger ("merkle", modlogger::WARNING)
#define info  modlogger ("merkle", modlogger::INFO)
#define trace modlogger ("merkle", modlogger::TRACE)

// {{{ General utility functions
void
format_rpcnode (merkle_tree *ltree, u_int depth, const merkle_hash &prefix,
		merkle_node *node, merkle_rpc_node *rpcnode)
{
  rpcnode->depth = depth;
  rpcnode->prefix = prefix;
  rpcnode->count = node->count;
  rpcnode->hash = node->hash;
  rpcnode->isleaf = node->isleaf ();

  if (!node->isleaf ()) {
    rpcnode->child_hash.setsize (64);
    for (int i = 0; i < 64; i++)
      rpcnode->child_hash[i] = node->child_hash (i);
  } else {
    vec<merkle_hash> keys = ltree->database_get_keys (depth, prefix);

    if (keys.size () != rpcnode->count) {
      warn << "\n\n\n----------------------------------------------------------\n";
      warn << "BUG BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG BUG\n";
      warn << "Send this output to <EMAIL>\n";
      warn << "BUG: " << depth << " " << prefix << "\n";
      warn << "BUG: " << keys.size () << " != " << rpcnode->count << "\n";
      ltree->check_invariants ();
      warn << "BUG BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG BUG\n";
      panic << "----------------------------------------------------------\n\n\n";
    }

    rpcnode->child_hash.setsize (keys.size ());
    for (u_int i = 0; i < keys.size (); i++) {
      rpcnode->child_hash[i] = keys[i];
    }
  }
}

// Check whether [l1, r1] overlaps [l2, r2] on the circle.
static bool
overlap (const bigint &l1, const bigint &r1, const bigint &l2, const bigint &r2)
{
  // There might be a more efficient way to do this..
  return (betweenbothincl (l1, r1, l2) || betweenbothincl (l1, r1, r2)
	  || betweenbothincl (l2, r2, l1) || betweenbothincl (l2, r2, r1));
}

static void
compare_keylists (const vec<chordID> &lkeys,
		  const vec<chordID> &vrkeys,
		  chordID rngmin, chordID rngmax,
		  missingfnc_t missingfnc)
{
  // populate a hash table with the remote keys
  qhash<chordID, int, hashID> rkeys;
  for (u_int i = 0; i < vrkeys.size (); i++) {
    if (betweenbothincl (rngmin, rngmax, vrkeys[i])) {
      // trace << "remote key: " << vrkeys[i] << "\n";
      rkeys.insert (vrkeys[i], 1);
    }
  }

  // do I have something he doesn't have?
  for (unsigned int i = 0; i < lkeys.size (); i++) {
    if (!rkeys[lkeys[i]] &&
	betweenbothincl (rngmin, rngmax, lkeys[i])) {
      trace << "remote missing [" << rngmin << ", "
	    << rngmax << "] key=" << lkeys[i] << "\n";
      (*missingfnc) (lkeys[i], false);
    } else {
      if (rkeys[lkeys[i]]) trace << "remote has " << lkeys[i] << "\n";
      else trace << "out of range: " << lkeys[i] << "\n";
      rkeys.remove (lkeys[i]);
    }
  }

  //anything left: he has and I don't
  qhash_slot<chordID, int> *slot = rkeys.first ();
  while (slot) {
    trace << "local missing [" << rngmin << ", "
	  << rngmax << "] key=" << slot->key << "\n";
    (*missingfnc) (slot->key, true);
    slot = rkeys.next (slot);
  }

}
// }}}

// {{{ merkle_getkeyrange
// {{{ merkle_getkeyrange declarations
class merkle_getkeyrange {
private:
  uint vnode;
  dhash_ctype ctype;
  bigint rngmin;
  bigint rngmax;
  bigint current;
  missingfnc_t missing;
  rpcfnc_t rpcfnc;
  vec<chordID> lkeys;

  cbv completecb;

  void finish ();
  void doRPC (int procno, ptr<void> in, void *out, aclnt_cb cb);
  void go ();
  void getkeys_cb (ref<getkeys_arg> arg, ref<getkeys_res> res, clnt_stat err);

public:
  ~merkle_getkeyrange () {}
  merkle_getkeyrange (uint vnode, dhash_ctype ctype,
		      bigint rngmin, bigint rngmax,
		      const vec<chordID> &plkeys,
		      missingfnc_t missing, rpcfnc_t rpcfnc,
		      cbv completecb)
    : vnode (vnode), ctype (ctype), rngmin (rngmin),
      rngmax (rngmax), current (rngmin),
      missing (missing), rpcfnc (rpcfnc), lkeys (plkeys),
      completecb (completecb)
    { go (); }
};
// }}}
// {{{ merkle_getkeyrange utility
void
merkle_getkeyrange::finish ()
{
  delaycb (0, completecb);
  delete this;
}

void
merkle_getkeyrange::doRPC (int procno, ptr<void> in, void *out, aclnt_cb cb)
{
  // Must resort to bundling all values into one argument since
  // async/callback.h is configured with too few parameters.
  struct RPC_delay_args args (merklesync_program_1, procno, in, out, cb,
			      NULL);
  (*rpcfnc) (&args);
}
// }}}
void
merkle_getkeyrange::go ()
{
  if (!betweenbothincl (rngmin, rngmax, current)) {
    trace << "merkle_getkeyrange::go () ==> DONE\n";
    finish ();
    return;
  }

  ref<getkeys_arg> arg = New refcounted<getkeys_arg> ();
  arg->ctype = ctype;
  arg->vnode = vnode;
  arg->rngmin = current;
  arg->rngmax = rngmax;
  ref<getkeys_res> res = New refcounted<getkeys_res> ();
  doRPC (MERKLESYNC_GETKEYS, arg, res,
	 wrap (this, &merkle_getkeyrange::getkeys_cb, arg, res));
}

void
merkle_getkeyrange::getkeys_cb (ref<getkeys_arg> arg, ref<getkeys_res> res,
				clnt_stat err)

{
  if (err) {
    warn << "GETKEYS: rpc error " << err << "\n";
    finish ();
    return;
  } else if (res->status != MERKLE_OK) {
    warn << "GETKEYS: protocol error " << err2str (res->status) << "\n";
    finish ();
    return;
  }
//  warn << "getkeys_cb: [" << arg->rngmin << "," << arg->rngmax << "] "
//       << res->resok->keys.size () << " keys, with "
//       << (res->resok->morekeys ? "more" : "no") << " keys left.\n";

  // Assuming keys are sent back in increasing clockwise order
  vec<chordID> rkeys;
  for (u_int i = 0; i < res->resok->keys.size (); i++)
    rkeys.push_back (res->resok->keys[i]);

  chordID sentmax = rngmax;
  if (res->resok->keys.size () > 0)
    sentmax = res->resok->keys.back ();
  compare_keylists (lkeys, rkeys, current, sentmax, missing);

  current = incID (sentmax);
  if (!res->resok->morekeys) {
    finish ();
    return;
  }
  go ();
}
// }}}

// {{{ merkle_syncer
// {{{ merkle_syncer utility
merkle_syncer::merkle_syncer (uint vnode, dhash_ctype ctype,
			      ptr<merkle_tree> ltree,
			      rpcfnc_t rpcfnc, missingfnc_t missingfnc)
  : vnode (vnode), ctype (ctype), ltree (ltree), rpcfnc (rpcfnc),
    missingfnc (missingfnc), completecb (cbi_null),
    outstanding_sendnodes (0),
    outstanding_keyranges (0)
{
  deleted = New refcounted<bool> (false);
  fatal_err = NULL;<commit_msg>Correctly set sentmax for last getkeyrange in merkle_syncer.<commit_after>#include <chord.h>
#include "merkle_syncer.h"
#include "qhash.h"
#include "async.h"
#include <id_utils.h>
#include <comm.h>

#include <modlogger.h>
#define warning modlogger ("merkle", modlogger::WARNING)
#define info  modlogger ("merkle", modlogger::INFO)
#define trace modlogger ("merkle", modlogger::TRACE)

// {{{ General utility functions
void
format_rpcnode (merkle_tree *ltree, u_int depth, const merkle_hash &prefix,
		merkle_node *node, merkle_rpc_node *rpcnode)
{
  rpcnode->depth = depth;
  rpcnode->prefix = prefix;
  rpcnode->count = node->count;
  rpcnode->hash = node->hash;
  rpcnode->isleaf = node->isleaf ();

  if (!node->isleaf ()) {
    rpcnode->child_hash.setsize (64);
    for (int i = 0; i < 64; i++)
      rpcnode->child_hash[i] = node->child_hash (i);
  } else {
    vec<merkle_hash> keys = ltree->database_get_keys (depth, prefix);

    if (keys.size () != rpcnode->count) {
      warn << "\n\n\n----------------------------------------------------------\n";
      warn << "BUG BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG BUG\n";
      warn << "Send this output to <EMAIL>\n";
      warn << "BUG: " << depth << " " << prefix << "\n";
      warn << "BUG: " << keys.size () << " != " << rpcnode->count << "\n";
      ltree->check_invariants ();
      warn << "BUG BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG  BUG BUG\n";
      panic << "----------------------------------------------------------\n\n\n";
    }

    rpcnode->child_hash.setsize (keys.size ());
    for (u_int i = 0; i < keys.size (); i++) {
      rpcnode->child_hash[i] = keys[i];
    }
  }
}

// Check whether [l1, r1] overlaps [l2, r2] on the circle.
static bool
overlap (const bigint &l1, const bigint &r1, const bigint &l2, const bigint &r2)
{
  // There might be a more efficient way to do this..
  return (betweenbothincl (l1, r1, l2) || betweenbothincl (l1, r1, r2)
	  || betweenbothincl (l2, r2, l1) || betweenbothincl (l2, r2, r1));
}

static void
compare_keylists (const vec<chordID> &lkeys,
		  const vec<chordID> &vrkeys,
		  chordID rngmin, chordID rngmax,
		  missingfnc_t missingfnc)
{
  // populate a hash table with the remote keys
  qhash<chordID, int, hashID> rkeys;
  for (u_int i = 0; i < vrkeys.size (); i++) {
    if (betweenbothincl (rngmin, rngmax, vrkeys[i])) {
      // trace << "remote key: " << vrkeys[i] << "\n";
      rkeys.insert (vrkeys[i], 1);
    }
  }

  // do I have something he doesn't have?
  for (unsigned int i = 0; i < lkeys.size (); i++) {
    if (!rkeys[lkeys[i]] &&
	betweenbothincl (rngmin, rngmax, lkeys[i])) {
      trace << "remote missing [" << rngmin << ", "
	    << rngmax << "] key=" << lkeys[i] << "\n";
      (*missingfnc) (lkeys[i], false);
    } else {
      if (rkeys[lkeys[i]]) trace << "remote has " << lkeys[i] << "\n";
      else trace << "out of range: " << lkeys[i] << "\n";
      rkeys.remove (lkeys[i]);
    }
  }

  //anything left: he has and I don't
  qhash_slot<chordID, int> *slot = rkeys.first ();
  while (slot) {
    trace << "local missing [" << rngmin << ", "
	  << rngmax << "] key=" << slot->key << "\n";
    (*missingfnc) (slot->key, true);
    slot = rkeys.next (slot);
  }

}
// }}}

// {{{ merkle_getkeyrange
// {{{ merkle_getkeyrange declarations
class merkle_getkeyrange {
private:
  uint vnode;
  dhash_ctype ctype;
  bigint rngmin;
  bigint rngmax;
  bigint current;
  missingfnc_t missing;
  rpcfnc_t rpcfnc;
  vec<chordID> lkeys;

  cbv completecb;

  void finish ();
  void doRPC (int procno, ptr<void> in, void *out, aclnt_cb cb);
  void go ();
  void getkeys_cb (ref<getkeys_arg> arg, ref<getkeys_res> res, clnt_stat err);

public:
  ~merkle_getkeyrange () {}
  merkle_getkeyrange (uint vnode, dhash_ctype ctype,
		      bigint rngmin, bigint rngmax,
		      const vec<chordID> &plkeys,
		      missingfnc_t missing, rpcfnc_t rpcfnc,
		      cbv completecb)
    : vnode (vnode), ctype (ctype), rngmin (rngmin),
      rngmax (rngmax), current (rngmin),
      missing (missing), rpcfnc (rpcfnc), lkeys (plkeys),
      completecb (completecb)
    { go (); }
};
// }}}
// {{{ merkle_getkeyrange utility
void
merkle_getkeyrange::finish ()
{
  delaycb (0, completecb);
  delete this;
}

void
merkle_getkeyrange::doRPC (int procno, ptr<void> in, void *out, aclnt_cb cb)
{
  // Must resort to bundling all values into one argument since
  // async/callback.h is configured with too few parameters.
  struct RPC_delay_args args (merklesync_program_1, procno, in, out, cb,
			      NULL);
  (*rpcfnc) (&args);
}
// }}}
void
merkle_getkeyrange::go ()
{
  if (!betweenbothincl (rngmin, rngmax, current)) {
    trace << "merkle_getkeyrange::go () ==> DONE\n";
    finish ();
    return;
  }

  ref<getkeys_arg> arg = New refcounted<getkeys_arg> ();
  arg->ctype = ctype;
  arg->vnode = vnode;
  arg->rngmin = current;
  arg->rngmax = rngmax;
  ref<getkeys_res> res = New refcounted<getkeys_res> ();
  doRPC (MERKLESYNC_GETKEYS, arg, res,
	 wrap (this, &merkle_getkeyrange::getkeys_cb, arg, res));
}

void
merkle_getkeyrange::getkeys_cb (ref<getkeys_arg> arg, ref<getkeys_res> res,
				clnt_stat err)

{
  if (err) {
    warn << "GETKEYS: rpc error " << err << "\n";
    finish ();
    return;
  } else if (res->status != MERKLE_OK) {
    warn << "GETKEYS: protocol error " << err2str (res->status) << "\n";
    finish ();
    return;
  }
//  warn << "getkeys_cb: [" << arg->rngmin << "," << arg->rngmax << "] "
//       << res->resok->keys.size () << " keys, with "
//       << (res->resok->morekeys ? "more" : "no") << " keys left.\n";

  // Assuming keys are sent back in increasing clockwise order
  vec<chordID> rkeys;
  for (u_int i = 0; i < res->resok->keys.size (); i++)
    rkeys.push_back (res->resok->keys[i]);

  chordID sentmax = rngmax;
  if (res->resok->morekeys && res->resok->keys.size () > 0)
    sentmax = res->resok->keys.back ();
  compare_keylists (lkeys, rkeys, current, sentmax, missing);

  current = incID (sentmax);
  if (!res->resok->morekeys) {
    finish ();
    return;
  }
  go ();
}
// }}}

// {{{ merkle_syncer
// {{{ merkle_syncer utility
merkle_syncer::merkle_syncer (uint vnode, dhash_ctype ctype,
			      ptr<merkle_tree> ltree,
			      rpcfnc_t rpcfnc, missingfnc_t missingfnc)
  : vnode (vnode), ctype (ctype), ltree (ltree), rpcfnc (rpcfnc),
    missingfnc (missingfnc), completecb (cbi_null),
    outstanding_sendnodes (0),
    outstanding_keyranges (0)
{
  deleted = New refcounted<bool> (false);
  fatal_err = NULL;
{'fraction_non_alphanumeric': 0.18364828384111068, 'fraction_numerical': 0.005553413035094486, 'mean_word_length': 2.9760809567617295, 'pattern_counts': {'":': 0, '<': 279, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4701918', 'n_tokens_mistral': 5047, 'n_tokens_neox': 4935, 'n_words': 1213}