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 "sync-log.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;

INIT_LOGGER ("SyncLogic");

namespace Sync
{

SyncLogic::SyncLogic (const std::string &syncPrefix,
                      LogicUpdateCallback onUpdate,
                      LogicRemoveCallback onRemove)
  : m_syncPrefix (syncPrefix)
  , m_onUpdate (onUpdate)
  , m_onRemove (onRemove)
  , m_ccnxHandle(new CcnxWrapper())
  , m_randomGenerator (static_cast<unsigned int> (std::time (0)))
  , m_rangeUniformRandom (m_randomGenerator, uniform_int<> (20,80))
{
#ifdef _DEBUG
#ifdef HAVE_LOG4CXX
  // _LOG_FUNCTION (syncPrefix);
  static int id = 0;
  staticModuleLogger = log4cxx::Logger::getLogger ("SyncLogic." + lexical_cast<string> (id));
  id ++;
#endif
#endif
  
  m_ccnxHandle->setInterestFilter (m_syncPrefix,
                                   bind (&SyncLogic::respondSyncInterest, this, _1));

  m_scheduler.schedule (posix_time::seconds (0),
                        bind (&SyncLogic::sendSyncInterest, this),
                        REEXPRESSING_INTEREST);
}

SyncLogic::~SyncLogic ()
{
  // _LOG_FUNCTION (this);
  // cout << "SyncLogic::~SyncLogic ()" << endl;

  m_ccnxHandle.reset ();
}

void
SyncLogic::respondSyncInterest (const string &interest)
{
  _LOG_TRACE ("<< I " << interest);
  //cout << "Respond Sync Interest" << endl;
  string hash = interest.substr(interest.find_last_of("/") + 1);
  // cout << "Received Sync Interest: " << hash << endl;
  
  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, false);
}

void
SyncLogic::processSyncInterest (DigestConstPtr digest, const std::string &interestName, bool timedProcessing/*=false*/)
{
  recursive_mutex::scoped_lock lock (m_stateMutex);

  // Special case when state is not empty and we have received request with zero-root digest
  if (digest->isZero () && !m_state.getDigest()->isZero ())
    {
      _LOG_TRACE (">> D " << interestName << "/state" << " (zero)");

      m_syncInterestTable.remove (interestName + "/state");
      m_ccnxHandle->publishData (interestName + "/state",
                                 lexical_cast<string> (m_state),
                                 m_syncResponseFreshness);
      if (m_outstandingInterest == interestName)
        {
          m_scheduler.schedule (posix_time::seconds (0),
                                bind (&SyncLogic::sendSyncInterest, this),
                                REEXPRESSING_INTEREST);
        }
      return;
    }

  if (*m_state.getDigest() == *digest)
    {
      // cout << interestName << "\n";
      if (digest->isZero ())
        {
          _LOG_TRACE ("processSyncInterest (): Digest is zero, adding /state to PIT");
          m_syncInterestTable.insert (interestName + "/state");
        }
      else
        {
          _LOG_TRACE ("processSyncInterest (): Same state. Adding to PIT");
          m_syncInterestTable.insert (interestName);
        }
      return;
    }
  
  DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);

  if (stateInDiffLog != m_log.end ())
  {
    DiffStateConstPtr stateDiff = (*stateInDiffLog)->diff ();
    // string state = lexical_cast<string> (*stateDiff);
    // erase_all (state, "\n");
    // _LOG_TRACE (">> D " << interestName << ", state: " << state);
    // _LOG_DEBUG ("Log size: " << m_log.size ());

    // BOOST_FOREACH (DiffStateConstPtr ds, m_log.get<sequenced> ())
    //   {
    //     string state = lexical_cast<string> (*ds);
    //     erase_all (state, "\n");
    //     _LOG_DEBUG ("   " << state << ", " << *ds->getDigest ());
    //   }

    m_syncInterestTable.remove (interestName);
    m_ccnxHandle->publishData (interestName,
                               lexical_cast<string> (*stateDiff),
                               m_syncResponseFreshness);
    if (m_outstandingInterest == interestName)
      {
        m_scheduler.schedule (posix_time::seconds (0),
                              bind (&SyncLogic::sendSyncInterest, this),
                              REEXPRESSING_INTEREST);
      }
    return;
  }

  if (!timedProcessing)
    {
      m_scheduler.schedule (posix_time::milliseconds (m_rangeUniformRandom ()) /*from 20 to 100ms*/,
                            bind (&SyncLogic::processSyncInterest, this, digest, interestName, true),
                            DELAYED_INTEREST_PROCESSING);
      
    }
  else
    {
      _LOG_TRACE (">> D " << interestName << "/state" << " (timed processing)");

      m_syncInterestTable.remove (interestName + "/state");
      m_ccnxHandle->publishData (interestName + "/state",
                                 lexical_cast<string> (m_state),
                                 m_syncResponseFreshness);

      if (m_outstandingInterest == interestName)
        {
          m_scheduler.schedule (posix_time::seconds (0),
                                bind (&SyncLogic::sendSyncInterest, this),
                                REEXPRESSING_INTEREST);
        }
    }
}

void
SyncLogic::processSyncData (const string &name, const string &dataBuffer)
{
  _LOG_TRACE ("<< D " << name);
  DiffStatePtr diffLog = make_shared<DiffState> ();
  
  try
    {
      recursive_mutex::scoped_lock lock (m_stateMutex);
      
      string last = name.substr(name.find_last_of("/") + 1);
      istringstream ss (dataBuffer);

      if (last == "state")
        {
          FullState full;
          ss >> full;
          BOOST_FOREACH (LeafConstPtr leaf, full.getLeaves()) // order doesn't matter
            {
              NameInfoConstPtr info = leaf->getInfo ();
              SeqNo seq = leaf->getSeq ();

              bool inserted = false;
              bool updated = false;
              SeqNo oldSeq;
              tie (inserted, updated, oldSeq) = m_state.update (info, seq);

              if (inserted || updated)
                {
                  diffLog->update (info, seq);
                  m_onUpdate (info->toString (), seq, oldSeq);
                }
            }
        }
      else
        {
          DiffState diff;
          ss >> diff;
          BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>())
            {
              DiffLeafConstPtr diffLeaf = dynamic_pointer_cast<const DiffLeaf> (leaf);
              BOOST_ASSERT (diffLeaf != 0);

              NameInfoConstPtr info = diffLeaf->getInfo();
              if (diffLeaf->getOperation() == UPDATE)
                {
                  SeqNo seq = diffLeaf->getSeq();

                  bool inserted = false;
                  bool updated = false;
                  SeqNo oldSeq;
                  tie (inserted, updated, oldSeq) = m_state.update (info, seq);

                  if (inserted || updated)
                    {
                      diffLog->update (info, seq);
                      m_onUpdate (info->toString (), seq, oldSeq);
                    }
                }
              else if (diffLeaf->getOperation() == REMOVE)
                {
                  if (m_state.remove (info))
                    {
                      diffLog->remove (info);
                      m_onRemove (info->toString ());
                    }
                }
              else
                {
                  BOOST_ASSERT (false); // just in case
                }
            }
        }

      insertToDiffLog (diffLog);
    }
  catch (Error::SyncXmlDecodingFailure &e)
    {
      diffLog.reset ();
      // don't do anything
    }

  // if state has changed, then it is safe to express a new interest
  if (diffLog->getLeaves ().size () > 0)
    {
      m_scheduler.schedule (posix_time::seconds (0),
                            bind (&SyncLogic::sendSyncInterest, this),
                            REEXPRESSING_INTEREST);
    }
  else
    {
      // should not reexpress the same interest. Need at least wait for data lifetime
      // Otherwise we will get immediate reply from the local daemon and there will be 100% utilization
      m_scheduler.cancel (REEXPRESSING_INTEREST);
      // m_scheduler.schedule (posix_time::seconds (0),
      m_scheduler.schedule (posix_time::seconds (m_syncResponseFreshness) + posix_time::milliseconds (1),
                            bind (&SyncLogic::sendSyncInterest, this),
                            REEXPRESSING_INTEREST);
    }
}

void
SyncLogic::satisfyPendingSyncInterests (DiffStatePtr diffLog)
{
  vector<string> pis = m_syncInterestTable.fetchAll ();
  if (pis.size () > 0)
    {
      stringstream ss;
      ss << *diffLog;
      bool satisfiedOwnInterest = false;
      
      for (vector<string>::iterator ii = pis.begin(); ii != pis.end(); ++ii)
        {
          _LOG_TRACE (">> D " << *ii);
          m_ccnxHandle->publishData (*ii, ss.str(), m_syncResponseFreshness);

          {
            recursive_mutex::scoped_lock lock (m_stateMutex);
            // _LOG_DEBUG (*ii << " == " << m_outstandingInterest << " = " << (*ii == m_outstandingInterest));
            satisfiedOwnInterest = satisfiedOwnInterest || (*ii == m_outstandingInterest) || (*ii == (m_outstandingInterest + "/state"));
          }
        }

      if (satisfiedOwnInterest)
        {
          _LOG_DEBUG ("Have satisfied our own interest. Scheduling interest reexpression");
          // we need to reexpress interest only if we satisfied our own interest
          m_scheduler.schedule (posix_time::milliseconds (0),
                                bind (&SyncLogic::sendSyncInterest, this),
                                REEXPRESSING_INTEREST);
        }
    }
}

void
SyncLogic::insertToDiffLog (DiffStatePtr diffLog) 
{
  //cout << "Process Pending Interests" <<endl;
  diffLog->setDigest (m_state.getDigest());  
  if (m_log.size () > 0)
    {
      m_log.get<sequenced> ().front ()->setNext (diffLog);
    }
  m_log.erase (m_state.getDigest()); // remove diff state with the same digest.  next pointers are still valid
  /// @todo Optimization
  m_log.get<sequenced> ().push_front (diffLog);
  // _LOG_DEBUG (*diffLog->getDigest () << " " << m_log.size ());
}

void
SyncLogic::addLocalNames (const string &prefix, uint32_t session, uint32_t seq)
{
  DiffStatePtr diff;
  {
    //cout << "Add local names" <<endl;
    recursive_mutex::scoped_lock lock (m_stateMutex);
    NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix);

    SeqNo seqN (session, seq);
    m_state.update(info, seqN);

    _LOG_DEBUG ("addLocalNames (): new state " << *m_state.getDigest ());
    
    diff = make_shared<DiffState>();
    diff->update(info, seqN);
    insertToDiffLog (diff);
  }

  // _LOG_DEBUG ("PIT size: " << m_syncInterestTable.size ());
  satisfyPendingSyncInterests (diff);  
}

void
SyncLogic::remove(const string &prefix) 
{
  DiffStatePtr diff;
  {
    recursive_mutex::scoped_lock lock (m_stateMutex);
    NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix);
    m_state.remove(info);	

    diff = make_shared<DiffState>();
    diff->remove(info);

    insertToDiffLog (diff);
  }

  satisfyPendingSyncInterests (diff);  
}

void
SyncLogic::sendSyncInterest ()
{
  ostringstream os;

  {
    // cout << "Sending Sync Interest" << endl;
    recursive_mutex::scoped_lock lock (m_stateMutex);

    os << m_syncPrefix << "/" << *m_state.getDigest();

    _LOG_TRACE (">> I " << os.str ());

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

  m_scheduler.cancel (REEXPRESSING_INTEREST);
  m_scheduler.schedule (posix_time::seconds (4),
                        bind (&SyncLogic::sendSyncInterest, this),
                        REEXPRESSING_INTEREST);
}

}