model/sync-logic.cc

/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
/*
 * Copyright (c) 2012 University of California, Los Angeles
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation;
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
 *         卞超轶 Chaoyi Bian <bcy@pku.edu.cn>
 *	   Alexander Afanasyev <alexander.afanasyev@ucla.edu>
 */

#include "sync-logic.h"
#include "sync-diff-leaf.h"
#include "sync-full-leaf.h"
#include <boost/make_shared.hpp>
#include <boost/foreach.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <vector>

using namespace std;
using namespace boost;

namespace Sync
{

const boost::posix_time::time_duration SyncLogic::m_delayedCheckTime = boost::posix_time::seconds (4.0);


SyncLogic::SyncLogic (const string &syncPrefix,
                      LogicCallback fetch,
                      CcnxWrapperPtr ccnxHandle)
  : m_syncPrefix (syncPrefix)
  , m_fetch (fetch)
  , m_ccnxHandle (ccnxHandle)
  , m_delayedCheckThreadRunning (true)
{
  srandom(time(NULL));
  m_ccnxHandle->setInterestFilter (syncPrefix,
                                   bind (&SyncLogic::respondSyncInterest, this, _1));

  m_delayedCheckThread = thread (&SyncLogic::delayedChecksLoop, this);
}

SyncLogic::~SyncLogic ()
{
  m_delayedCheckThreadRunning = false;
  // cout << "Requested stop" << this_thread::get_id () << endl;
  m_delayedCheckThread.interrupt ();
  m_delayedCheckThread.join ();
}

void
SyncLogic::delayedChecksLoop ()
{
  while (m_delayedCheckThreadRunning)
    {
      try
        {
          DelayedChecksList::value_type tuple;
          
          {
            unique_lock<mutex> lock (m_listChecksMutex);
            while (m_delayedCheckThreadRunning && m_listChecks.size () == 0)
              {
                m_listChecksCondition.wait (lock);
                // cout << "Got something" << endl;
              }

            if (m_listChecks.size () == 0) continue;

            tuple = m_listChecks.front ();
            m_listChecks.pop_front ();
            // cout << "pop" << endl;
            // release the mutex
          }

          // waiting and calling
          
          // cout << "Duration: " << tuple.get<0> () - get_system_time () << endl;
          this_thread::sleep (tuple.get<0> ()); 
          
          if (!m_delayedCheckThreadRunning) continue;
          tuple.get<1> () (); // call the scheduled function
        }
      catch (thread_interrupted e)
        {
          // cout << "interrupted: " << this_thread::get_id () << endl;
          // do nothing
        }
    }
  // cout << "Exited...\n";
}



void
SyncLogic::respondSyncInterest (const string &interest)
{
  string hash = interest.substr(interest.find_last_of("/") + 1);
  DigestPtr digest = make_shared<Digest> ();
  try
    {
      istringstream is (hash);
      is >> *digest;
    }
  catch (Error::DigestCalculationError &e)
    {
      // log error. ignoring it for now, later we should log it
      return;
    }

  processSyncInterest (digest, interest);
}

void
SyncLogic::processSyncInterest (DigestConstPtr digest, const std::string &interestName, bool timedProcessing/*=false*/)
{
  // cout << "SyncLogic::processSyncInterest " << timedProcessing << endl;
    
  if (*m_state.getDigest() == *digest)
  {
    m_syncInterestTable.insert (interestName);
    return;
  }

  DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);

  if (stateInDiffLog != m_log.end ())
  {
    m_ccnxHandle->publishData (interestName,
                               lexical_cast<string> (*(*stateInDiffLog)->diff ()),
                               m_syncResponseFreshness);
    return;
  }

  if (!timedProcessing)
    {
      {
        // Alex: Should we ignore interests if interest with the same digest is already in the wait queue?
        
        lock_guard<mutex> lock (m_listChecksMutex);
        system_time delay = get_system_time () + m_delayedCheckTime;
        // do we need randomization??
        // delay += boost::ptime::milliseconds (rand() % 80 + 20);
      
        m_listChecks.push_back (make_tuple (delay,
                                            bind (&SyncLogic::processSyncInterest, this, digest, interestName, true))
                                );
      }
      m_listChecksCondition.notify_one ();
    }
  else
    {
      m_ccnxHandle->publishData (interestName + "/state",
                                 lexical_cast<string> (m_state),
                                 m_syncResponseFreshness);
    }
}

void
SyncLogic::processSyncData (const string &name, const string &dataBuffer)
{
  string last = name.substr(name.find_last_of("/") + 1);
  stringstream ss(dataBuffer);

  const LeafContainer &fullLc = m_state.getLeaves();
  DiffStatePtr diffLog = make_shared<DiffState>();

  if (last == "state")
  {
    FullState full;
    ss >> full;
    BOOST_FOREACH (LeafConstPtr leaf, full.getLeaves().get<ordered>())
    {
      shared_ptr<const FullLeaf> fullLeaf = dynamic_pointer_cast<const FullLeaf>(leaf);
      const NameInfo &info = fullLeaf->getInfo();
      LeafContainer::iterator it = fullLc.find(info);
      NameInfoConstPtr pInfo = StdNameInfo::FindOrCreate(info.toString());
      SeqNo seq = fullLeaf->getSeq();

      if (it == fullLc.end())
      {
	string prefix = info.toString();
	prefix += "/";
	prefix += seq.getSession();
	m_fetch(prefix, 1, seq.getSeq());
	m_state.update(pInfo, seq);
	diffLog->update(pInfo, seq);
      }
      else
      {
	SeqNo currSeq = (*it)->getSeq();
	if (currSeq < seq)
	{
	  string prefix = info.toString();
	  prefix += "/";
	  prefix += seq.getSession();

	  if (currSeq.getSession() == seq.getSession())
	    m_fetch(prefix, currSeq.getSeq() + 1, seq.getSeq());
	  else
	    m_fetch(prefix, 1, seq.getSeq());

	  m_state.update(pInfo, seq);
	  diffLog->update(pInfo, seq);
	}
      }
    }
  }
  else
  {
    DiffState diff;
    ss >> diff;
    BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>())
    {
      shared_ptr<const DiffLeaf> diffLeaf = dynamic_pointer_cast<const DiffLeaf>(leaf);
      const NameInfo &info = diffLeaf->getInfo();
      LeafContainer::iterator it = fullLc.find(info);
      SeqNo seq = diffLeaf->getSeq();

      switch (diffLeaf->getOperation())
      {
	case UPDATE:
	  if (it == fullLc.end())
	  {
	    string prefix = info.toString();
	    prefix += "/";
	    prefix += seq.getSession();
	    m_fetch(prefix, 1, seq.getSeq());

	    NameInfoConstPtr pInfo = StdNameInfo::FindOrCreate(info.toString());
	    m_state.update(pInfo, seq);
	    diffLog->update(pInfo, seq);
	  }
	  else
	  {
	    SeqNo currSeq = (*it)->getSeq();
	    if (currSeq < seq)
	    {
	      string prefix = info.toString();
	      prefix += "/";
	      prefix += seq.getSession();

	      if (currSeq.getSession() == seq.getSession())
	        m_fetch(prefix, currSeq.getSeq() + 1, seq.getSeq());
	      else
	        m_fetch(prefix, 1, seq.getSeq());

	      NameInfoConstPtr pInfo = StdNameInfo::FindOrCreate(info.toString());
	      m_state.update(pInfo, seq);
	      diffLog->update(pInfo, seq);
	    }
	  }
	  break;

	case REMOVE:
	  if (it != fullLc.end())
	  {
	    NameInfoConstPtr pInfo = StdNameInfo::FindOrCreate(info.toString());
  	    m_state.remove(pInfo);
	    diffLog->remove(pInfo);
	  }
	  break;

	default:
	  break;
      }
    }
  }

  diffLog->setDigest(m_state.getDigest());
  m_log.insert(diffLog);
  sendSyncInterest();
}

void
SyncLogic::addLocalNames (const string &prefix, uint32_t session, uint32_t seq)
{
  NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix);
  SeqNo seqN(session, seq);
  DiffStatePtr diff = make_shared<DiffState>();
  diff->update(info, seqN);
  m_state.update(info, seqN);
  diff->setDigest(m_state.getDigest());
  m_log.insert(diff);

  vector<string> pis = m_syncInterestTable.fetchAll ();
  stringstream ss;
  ss << *diff;
  for (vector<string>::iterator ii = pis.begin(); ii != pis.end(); ++ii)
  {
    m_ccnxHandle->publishData(*ii, ss.str(), m_syncResponseFreshness);
  }
}

void
SyncLogic::sendSyncInterest ()
{
  ostringstream os;
  os << m_syncPrefix << "/" << m_state.getDigest();

  m_ccnxHandle->sendInterest (os.str (),
                              bind (&SyncLogic::processSyncData, this, _1, _2));
}

}