working_set.cc

这是一个著名的应用层组播中间件的源码

//Copyright (c) 2004, Charles Killian, Adolfo Rodriguez, Dejan Kostic, Sooraj Bhat, and Amin Vahdat
//All rights reserved.
//
//Redistribution and use in source and binary forms, with or without
//modification, are permitted provided that the following conditions are met:
//
//   * Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//   * Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//   * Neither the names of Duke University nor The University of
//     California, San Diego, nor the names of its contributors
//     may be used to endorse or promote products derived from
//     this software without specific prior written permission.
//
//THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
//IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
//DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE
//FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
//DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
//SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
//OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
//USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "working_set.h"
#include "macedon.h"
#include "limits.h"

void debug (int level, char *format, ... )     ;

int random_integer( int maximum)
  ;

working_set::working_set(int digest_size, int digest_ratio)
  : sketch(UNIVERSE_SIZE ), duplicates( 0), keep( KEEP)
#if USE_WS_STORE
  ,file_store( 0)
#endif
#if USE_BLOOM
  ,local_digest(( digest_size? digest_size : BLOOM_SIZE), 4, digest_ratio)
#endif


{
  last_seq=0;
  earliest_seq=0;
}

working_set::~working_set()
{
}
#if USE_WS_STORE

// ---------------------------------------------- 
// set_store
// ---------------------------------------------- 

void  working_set::set_store (int elements,int size, char* name )
{
  int check_store_size = (int)(elements*size);

  file_store = new object_file_store ( name, size, check_store_size, 0);
  ssize = size;
}
#endif

// ---------------------------------------------- 
// get_last
// ---------------------------------------------- 

int  working_set::get_last()
{
  return last_seq;
}

// ---------------------------------------------- 
// contains
// ---------------------------------------------- 

int  working_set::contains (int key)
{
  map <int, message*, less<int> >::const_iterator element
    = contents.find( key);
  if (element != contents.end())
    {
      return 1;
    }

  return 0;
}

// ---------------------------------------------- 
// insert
// ---------------------------------------------- 

int  working_set::insert (int key, int shallow, unsigned char * data, int size)
{
  if (contains( key))
    {
      duplicates++;
      return 0;
    }
  
  if (last_seq < key)
    last_seq = key;

  message* plug = NULL;
  if (data)
    {
      plug = new message ();
      if (!plug)
	{
	 printf("plug: %d\n", plug);
	 abort();
	}
#if USE_WS_STORE
      if (file_store && file_store->within_range (key-1) && size == ssize)
	{
	  file_store->lock_write( key);
	  plug-> data = file_store->get (key, size);
	}
      else
#endif

	{     
//  	  printf ("new key: %d size: %d\n", key, size);
//  	  fflush (stdout );
	  plug-> data = new unsigned char [size];
	}
      bcopy( data, plug-> data, size);
#if USE_WS_STORE
      if (file_store && file_store->within_range (key-1) && size == ssize)
	{
	  file_store->unlock( key);
	}
#endif

      plug->size = size;
    }

 contents [key]= plug;

  if (!shallow)
    {
      sketch.insert ( key);
      // truncate if needed
      earliest_seq = local_digest.insert( key, 0);
    }
  if (contents.size()> KEEP)
    {
      map <int, message*, less<int> >::iterator element
	=contents.begin();
      int vk = (*element).first;
      message* plug = (*element).second;
      if (plug)
	{
#if USE_WS_STORE
	  if (!(file_store && file_store->within_range (key-1) && size == ssize))
#endif

	    {
//  	      printf ("delete key: %d size: %d\n", key, size);
//  	      fflush (stdout );
	      delete [] plug->data;
	    }

	  delete plug;
	}

      contents. erase( element );
      sketch.remove ( vk, contents);

//    printf("removing element front: %d\n", *element);
    }

  return 1;
}
// ---------------------------------------------- 
// get_false_positives
// ---------------------------------------------- 

unsigned char *working_set::get_message(int item, int &size)
{
  unsigned char *temp;
  map <int, message*, less<int> >::const_iterator element
    = contents.find( item);
  if (element == contents.end())
    return NULL;

  message* found = (*element).second;
  if (found) 
    {
      size = found->size;
      temp = new unsigned char [found->size];
      if (!temp)
	{
	 printf("temp: %d\n", temp);
	 abort();
	}
      bcopy(found->data, temp, size);
      return temp;
  }
  else
    return NULL;
}

// ---------------------------------------------- 
// get_false_positives
// ---------------------------------------------- 

void  working_set::get_false_positives 
(list < int>& result)
{
//  printf("get_false_positives from %d to %d\n", earliest_seq, last_seq);

  //the goal is to identify sequence numbers that will deemed as false positives when the other side receives are bloom filter
//    for (int value = earliest_seq+1; value < last_seq ;value++ )
 int limit =  last_seq - ASKING_WINDOW;

 if (limit < 0) limit = 0;
 if (limit < earliest_seq) limit = earliest_seq;

  for (int value = last_seq; value > limit; value-- )
    {
//  printf("considering: %d\n", value);
      if (!contains( value) && local_digest.contains( value))
	{
	  result.push_back( value);
	}
    }
}

// ---------------------------------------------- 
// get_asking
// ---------------------------------------------- 

void  working_set::get_asking (int* asking, int count, list < int>& result)
{
  for (int counter = 0; counter < count ;counter++ )
    {
      if (contains( asking[counter]))
	{
	  result.push_back( counter);
	}
    }
}


// ---------------------------------------------- 
// export_sketch
// ---------------------------------------------- 

unsigned char*  working_set::export_sketch (int & size)
{
  unsigned char* buffer = new unsigned char [sketch.size_compacted_in_bytes()];
//  printf("sketch.size_compacted_in_bytes(): %d\n", sketch.size_compacted_in_bytes());

  sketch.serialize( buffer);
  //printf("contents . size(): %d\n", contents . size());

 size = sketch.size_compacted_in_bytes();
  return buffer;
}

// ---------------------------------------------- 
// export_digest
// ---------------------------------------------- 

unsigned char*   working_set::export_digest (int & size)
{
  unsigned char* buffer = new unsigned char [local_digest.size_compacted_in_bytes()];
  local_digest.serialize( buffer);
  size = local_digest.size_compacted_in_bytes();
  return buffer;
}


// ---------------------------------------------- 
// get_different_keys
// ---------------------------------------------- 

int working_set::get_different_keys 
(list < int>& result, 
 int starter, 
 digest& source_digest, 
 int high, int low, int maximum_count)
{
  //    source_digest.dump_state();
  
  debug(1, "get_different_keys %d high %d low %d\n", starter, high,  low);
  int max_stories=1000;
  int story[max_stories];
  int num_stories=0;
  int chosen;
  
  int last = -1;
  
  map <int, message*, less<int> >::const_iterator element;
  
  for(element=contents.begin();element!=contents.end();element++)
    {
      int value =(*element).first;

      //  printf("considering: %d\n", value);
      if (!source_digest.contains( value) && value < high && value > low)
	{
	  if (num_stories<max_stories)
	    {
	      story[num_stories] = value;
	      num_stories++;
	    }
	  else
	    {
	      chosen = random_integer(num_stories);
	      story[chosen] = value;
	    }
	}
    }

  int back = maximum_count;
  while (num_stories && back > 0)
    {
      chosen = random_integer(num_stories);
      result.push_back(story[chosen]);
      if (chosen != num_stories-1)
	story[chosen] = story[num_stories-1];
      num_stories--;
      back--;
    } 
  return 0;
}

// ---------------------------------------------- 
// get_modulo_keys
// ---------------------------------------------- 

int working_set::get_modulo_keys 
(list < int>& result, 
 digest& source_digest, int low, int high, 
 int mod, int mod_max, int maximum_count)
{
  //    source_digest.dump_state();
  
  debug(1, "get_modulo_keys mod %d max %d\n", mod, mod_max);
  int max_stories=2000;
  int story[max_stories];
  int num_stories=0;
  int chosen;
  
  int last = -1;
  
  map <int, message*, less<int> >::const_iterator element;
  
  for(element=contents.begin();element!=contents.end();element++)
    {
      int value = (*element).first;

      if (!source_digest.contains( value) && 
	  //	  value < high && 
	  value > low && 
	  mod_max !=0 && 
	  (((value % mod_max) == mod) || (mod == -1))
	  )
	{
	  if (num_stories<max_stories)
	    {
	      story[num_stories] = value;
	      num_stories++;
	    }
	  else
	    {
	      chosen = random_integer(num_stories);
	      story[chosen] = value;
	    }
	}
    }

  int back = maximum_count;
  while (num_stories && back > 0)
    {
      chosen = random_integer(num_stories);
//        if (story[chosen] > 1000000) {
//  	printf("REPLAY Something is wrong with this value %d, size %d\n", story[chosen], result.size());
//  	fflush(stdout);
//  	exit(2);
//        }
      result.push_back(story[chosen]);
      if (chosen != num_stories-1)
	story[chosen] = story[num_stories-1];
      num_stories--;
      back--;
    }
  return 0;
}


// ---------------------------------------------- 
// computer_resemblance
// ---------------------------------------------- 

double  working_set::compute_resemblance (unsigned char * buffer)
{
  bullet_summary_ticket* source_ticket = new bullet_summary_ticket(working_set::UNIVERSE_SIZE);
  source_ticket->import( buffer);

  double resemblance = compute_resemblance(*source_ticket);
  delete source_ticket;
  return resemblance;
}

// ---------------------------------------------- 
// compute_resemblance
// ---------------------------------------------- 

double  working_set::compute_resemblance (bullet_summary_ticket& ticket)
{
  double resemblance = ticket % sketch;
//   printf("resemblance: %lf\n", resemblance);

  return resemblance;

}

// ---------------------------------------------- 
// Test
// ---------------------------------------------- 

void  working_set::Testspeed ()
{
  int inserted = 0;
  int  removed = 0;
printf("testing working set: \n");

    double now = Scheduler::instance().clock();

  for (int number = 0; number < KEEP*1000;number++ )
    {
      int  chosen = number;
      if ( contains(chosen))
	{
	  printf("dup: %d\n", chosen);
	}

      insert (chosen);
	inserted++;
      }
    double finished = Scheduler::instance().clock();
    printf("finished - now: %f\n", finished - now);

}

//  void main()
//  {
//      working_set dd;
//      dd.Testspeed ();
//  }

// ---------------------------------------------- 
// dump_stats
// ---------------------------------------------- 

void  working_set::dump_stats ()
{
  printf("REPLAY working_set: contents. size(): %d  duplicates %d \n", contents . size(), duplicates);

}

// ---------------------------------------------- 
// get_earliest
// ---------------------------------------------- 

int  working_set::get_earliest ()
{
  return earliest_seq;
}

// ---------------------------------------------- 
// get_first
// ---------------------------------------------- 

int  working_set::get_first ()
{
  map <int, message*, less<int> >::const_iterator element
    = contents. begin();
  if (element == contents.end())
    return 0;

  return (*element). first;
}

// ---------------------------------------------- 
// stats
// ---------------------------------------------- 

int working_set::stats (int max, int ending)
{
    const int BULLET_MAX_PEERS = 10;

//      int big_contains[BULLET_MAX_PEERS] = {0};
//      int destination_contains;
//      for (int number = local_digest.get_lowest_sequence (); number < ending ;number++ )
//        {
//  	if (local_digest.contains
//  	    (number))
//  	  {
//  	  destination_contains++;
//  	  int mod_to = number % max;
//  	  big_contains[mod_to]++;
//  	  }
//       }
//      printf("destination_registers: %f %d out of  %d \n", destination_contains/(double) local_digest.get_items(), destination_contains, local_digest.get_items());
//      for (int counter = 0; counter <  max ;counter++ )
//        {
//  	int mod_to = counter % max;
//  	printf(" big_contains[%d]: %d from %d to %d \n", counter, big_contains[counter], local_digest.get_lowest_sequence (), ending);
//        }
    return last_seq;
}

// ---------------------------------------------- 
// reset
// ---------------------------------------------- 

int working_set::reset ()
{
  local_digest.reset();
  sketch.reset();
  map <int, message*, less<int> >::const_iterator element;
  
  for(element=contents.begin();element!=contents.end();element++)
    {
      int value = (*element).first;
      message* plug = (*element).second;
      if (plug)
	{
#if USE_WS_STORE
	  if (!(file_store && file_store->within_range (key-1) && size == ssize))
#endif

	    {
//  	      printf ("delete key: %d size: %d\n", key, size);
//  	      fflush (stdout );
	      delete [] plug->data;
	    }

	  delete plug;
	}
    }
  contents.clear();
}