<commit_before>#include <dhash.h>
#include <dhash_prot.h>
#include <chord.h>
#include <chord_prot.h>
#include <chord_util.h>
#include <dbfe.h>
#include <arpc.h>
#ifdef DMALLOC
#include <dmalloc.h>
#endif


dhash::dhash(str dbname, vnode *node, int k, int ss, int cs) :
  host_node (node), key_store(ss), key_cache(cs) {

  db = New dbfe();
  nreplica = k;
  rc_delay = 7;
  kc_delay = 11;

  //set up the options we want
  dbOptions opts;
  opts.addOption("opt_async", 1);
  opts.addOption("opt_cachesize", 80000);
  opts.addOption("opt_nodesize", 4096);

  if (int err = db->opendb(const_cast < char *>(dbname.cstr()), opts)) {
    warn << "open returned: " << strerror(err) << err << "\n";
    exit (-1);
  }
  
  assert(host_node);

  key_store.set_flushcb (wrap (this, &dhash::store_flush));
  key_replicate.set_flushcb (wrap (this, &dhash::store_flush));  
  key_cache.set_flushcb (wrap (this, &dhash::cache_flush));

  /* statistics */
  keys_stored = 0;
  bytes_stored = 0;
  bytes_served = 0;
  keys_replicated = 0;
  keys_cached = 0;

  //  init_key_status ();
  update_replica_list ();
  install_replica_timer ();
  install_keycheck_timer ();
  transfer_initial_keys ();

  // RPC demux
  host_node->addHandler (DHASH_PROGRAM, wrap(this, &dhash::dispatch));
}

void
dhash::dispatch (unsigned long procno, 
		 chord_RPC_arg *arg,
		 unsigned long rpc_id) 
{

  char *marshalled_arg = arg->marshalled_args.base ();
  int arg_len = arg->marshalled_args.size ();

  xdrmem x (marshalled_arg, arg_len, XDR_DECODE);
  xdrproc_t proc = dhash_program_1.tbl[procno].xdr_arg;

  switch (procno) {
  case DHASHPROC_FETCH:
    {

      dhash_fetch_arg *farg = New dhash_fetch_arg ();
      if (!proc (x.xdrp (), farg)) {
	warn << "DHASH: error unmarshalling arguments\n";
	return;
      }
      warnt ("DHASH: FETCH_request");
      fetch (farg->key, wrap (this, &dhash::fetchsvc_cb, rpc_id, farg));
    }
    break;

  case DHASHPROC_FETCHITER:
    {
      warnt ("DHASH: fetchiter_request");

      dhash_fetch_arg *farg = New dhash_fetch_arg ();
      if (!proc (x.xdrp (), farg)) {
	warn << "DHASH: error unmarshalling arguments\n";
	return;
      }

      dhash_fetchiter_res *res = New dhash_fetchiter_res (DHASH_CONTINUE);

      if (key_status (farg->key) != DHASH_NOTPRESENT) {
	//fetch the key and return it, end of story
	fetch (farg->key, wrap (this, &dhash::fetchiter_svc_cb,
				rpc_id, farg));
	delete res;
	return;
      } else if (responsible (farg->key))  {
	//no where else to go, return NOENT or RETRY?
	res->set_status (DHASH_NOENT);
      } else {
	res->set_status (DHASH_CONTINUE);
	chordID succ = host_node->my_succ ();
	chordID mID = host_node->my_ID ();
	chordID nid;
	if (betweenrightincl (mID,
			      succ,
			      farg->key)) 
	  nid = succ;
	else 
	  nid = host_node->lookup_closestpred (farg->key);
	
	res->cont_res->next.x = nid;
	res->cont_res->next.r = 
	  host_node->chordnode->locations->getaddress (nid);

	vec<chord_node> succ_list;
	chordID last = nid;
	for (int i = 0; i < nreplica; i++) {
	  chord_node nsucc;
	  nsucc.x = host_node->lookup_closestsucc (last);
	  nsucc.r = host_node->chordnode->locations->getaddress (nsucc.x);
	  succ_list.push_back (nsucc);
	  last = nsucc.x;
	}
	res->cont_res->succ_list.set (succ_list.base (), succ_list.size ());

      }

      dhash_reply (rpc_id, DHASHPROC_FETCHITER, res);
      delete res;
    }
    break;
  case DHASHPROC_STORE:
    {

      warnt("DHASH: STORE_request");

      dhash_insertarg *sarg = New dhash_insertarg ();
      if (!proc (x.xdrp (), sarg)) {<commit_msg>fix core dump under linux<commit_after>#include <dhash.h>
#include <dhash_prot.h>
#include <chord.h>
#include <chord_prot.h>
#include <chord_util.h>
#include <dbfe.h>
#include <arpc.h>
#ifdef DMALLOC
#include <dmalloc.h>
#endif


dhash::dhash(str dbname, vnode *node, int k, int ss, int cs) :
  host_node (node), key_store(ss), key_cache(cs) {

  db = New dbfe();
  nreplica = k;
  rc_delay = 7;
  kc_delay = 11;

  //set up the options we want
  dbOptions opts;
  opts.addOption("opt_async", 1);
  opts.addOption("opt_cachesize", 80000);
  opts.addOption("opt_nodesize", 4096);

  if (int err = db->opendb(const_cast < char *>(dbname.cstr()), opts)) {
    warn << "open returned: " << strerror(err) << err << "\n";
    exit (-1);
  }
  
  assert(host_node);

  key_store.set_flushcb (wrap (this, &dhash::store_flush));
  key_replicate.set_flushcb (wrap (this, &dhash::store_flush));  
  key_cache.set_flushcb (wrap (this, &dhash::cache_flush));

  /* statistics */
  keys_stored = 0;
  bytes_stored = 0;
  bytes_served = 0;
  keys_replicated = 0;
  keys_cached = 0;

  //  init_key_status ();
  update_replica_list ();
  install_replica_timer ();
  install_keycheck_timer ();
  transfer_initial_keys ();

  // RPC demux
  host_node->addHandler (DHASH_PROGRAM, wrap(this, &dhash::dispatch));
}

void
dhash::dispatch (unsigned long procno, 
		 chord_RPC_arg *arg,
		 unsigned long rpc_id) 
{

  char *marshalled_arg = arg->marshalled_args.base ();
  int arg_len = arg->marshalled_args.size ();

  xdrmem x (marshalled_arg, arg_len, XDR_DECODE);
  xdrproc_t proc = dhash_program_1.tbl[procno].xdr_arg;

  switch (procno) {
  case DHASHPROC_FETCH:
    {

      dhash_fetch_arg *farg = New dhash_fetch_arg ();
      if (!proc (x.xdrp (), farg)) {
	warn << "DHASH: error unmarshalling arguments\n";
	return;
      }
      warnt ("DHASH: FETCH_request");
      fetch (farg->key, wrap (this, &dhash::fetchsvc_cb, rpc_id, farg));
    }
    break;

  case DHASHPROC_FETCHITER:
    {
      warnt ("DHASH: fetchiter_request");

      dhash_fetch_arg *farg = New dhash_fetch_arg ();
      if (!proc (x.xdrp (), farg)) {
	warn << "DHASH: error unmarshalling arguments\n";
	return;
      }

      dhash_fetchiter_res *res = New dhash_fetchiter_res (DHASH_CONTINUE);

      if (key_status (farg->key) != DHASH_NOTPRESENT) {
	//fetch the key and return it, end of story
	fetch (farg->key, wrap (this, &dhash::fetchiter_svc_cb,
				rpc_id, farg));
	delete res;
	return;
      } else if (responsible (farg->key))  {
	//no where else to go, return NOENT or RETRY?
	res->set_status (DHASH_NOENT);
      } else {
	res->set_status (DHASH_CONTINUE);
	chordID succ = host_node->my_succ ();
	chordID mID = host_node->my_ID ();
	chordID nid;
	if (betweenrightincl (mID,
			      succ,
			      farg->key)) 
	  nid = succ;
	else 
	  nid = host_node->lookup_closestpred (farg->key);
	
	res->cont_res->next.x = nid;
	res->cont_res->next.r = 
	  host_node->chordnode->locations->getaddress (nid);

	res->cont_res->succ_list.setsize (nreplica);

	chordID last = nid;
	for (int i = 0; i < nreplica; i++) {
	  chord_node nsucc;
	  nsucc.x = host_node->lookup_closestsucc (last);
	  nsucc.r = host_node->chordnode->locations->getaddress (nsucc.x);
	  res->cont_res->succ_list[i] = nsucc;
	  last = nsucc.x;
	}

      }

      dhash_reply (rpc_id, DHASHPROC_FETCHITER, res);
      delete res;
    }
    break;
  case DHASHPROC_STORE:
    {

      warnt("DHASH: STORE_request");

      dhash_insertarg *sarg = New dhash_insertarg ();
      if (!proc (x.xdrp (), sarg)) {
{'fraction_non_alphanumeric': 0.14678123679391464, 'fraction_numerical': 0.0059163262431328354, 'mean_word_length': 3.49936628643853, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 80, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2523353', 'n_tokens_mistral': 2976, 'n_tokens_neox': 2775, 'n_words': 589}