Merge remote-tracking branch 'git.irl/dispatcher'
Conflicts:
src/sync-core.cc
src/sync-core.h
diff --git a/src/action-log.cc b/src/action-log.cc
index f49c69e..5c4f592 100644
--- a/src/action-log.cc
+++ b/src/action-log.cc
@@ -461,3 +461,18 @@
sqlite3_result_null (context);
}
+
+bool
+ActionLog::KnownFileState(const std::string &filename, const Hash &hash)
+{
+ sqlite3_stmt *stmt;
+ sqlite3_prepare_v2 (m_db, "SELECT * FROM FileState WHERE filename = ? AND file_hash = ?;", -1, &stmt, 0);
+ sqlite3_bind_text(stmt, 1, filename.c_str(), -1, SQLITE_STATIC);
+ sqlite3_bind_blob(stmt, 2, hash.GetHash (), hash.GetHashBytes (), SQLITE_STATIC);
+ if (sqlite3_step (stmt) == SQLITE_ROW)
+ {
+ return true;
+ }
+
+ return false;
+}
diff --git a/src/action-log.h b/src/action-log.h
index 33da6d1..1654215 100644
--- a/src/action-log.h
+++ b/src/action-log.h
@@ -52,6 +52,9 @@
void
AddActionDelete (const std::string &filename);
+ bool
+ KnownFileState(const std::string &filename, const Hash &hash);
+
private:
boost::tuple<sqlite3_int64, sqlite3_int64, sqlite3_int64, std::string>
GetExistingRecord (const std::string &filename);
diff --git a/src/dispatcher.cc b/src/dispatcher.cc
new file mode 100644
index 0000000..d48bb43
--- /dev/null
+++ b/src/dispatcher.cc
@@ -0,0 +1,257 @@
+/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2013 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
+ *
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
+#include "dispatcher.h"
+#include <boost/make_shared.hpp>
+
+using namespace Ccnx;
+using namespace std;
+using namespace boost;
+
+static const string BROADCAST_DOMAIN = "/ndn/broadcast/chronoshare";
+static const int MAX_FILE_SEGMENT_SIZE = 1024;
+
+Dispatcher::Dispatcher(const filesystem::path &path, const std::string &localUserName, const Ccnx::Name &localPrefix, const std::string &sharedFolder, const filesystem::path &rootDir, Ccnx::CcnxWrapperPtr ccnx, SchedulerPtr scheduler, int poolSize)
+ : m_ccnx(ccnx)
+ , m_core(NULL)
+ , m_rootDir(rootDir)
+ , m_executor(poolSize)
+ , m_objectManager(ccnx, rootDir)
+ , m_localUserName(localUserName)
+ , m_sharedFolder(sharedFolder)
+{
+ m_syncLog = make_shared<SyncLog>(path, localUserName);
+ m_actionLog = make_shared<ActionLog>(m_ccnx, path, m_syncLog, localUserName, sharedFolder);
+
+ Name syncPrefix(BROADCAST_DOMAIN + sharedFolder);
+ m_core = new SyncCore (m_syncLog, localUserName, localPrefix, syncPrefix,
+ bind(&Dispatcher::syncStateChanged, this, _1), ccnx, scheduler);
+}
+
+Dispatcher::~Dispatcher()
+{
+ if (m_core != NULL)
+ {
+ delete m_core;
+ m_core = NULL;
+ }
+}
+
+void
+Dispatcher::fileChangedCallback(const filesystem::path &relativeFilePath, ActionType type)
+{
+ Executor::Job job = bind(&Dispatcher::fileChanged, this, relativeFilePath, type);
+ m_executor.execute(job);
+}
+
+void
+Dispatcher::syncStateChangedCallback(const SyncStateMsgPtr &stateMsg)
+{
+ Executor::Job job = bind(&Dispatcher::syncStateChanged, this, stateMsg);
+ m_executor.execute(job);
+}
+
+void
+Dispatcher::actionReceivedCallback(const ActionItemPtr &actionItem)
+{
+ Executor::Job job = bind(&Dispatcher::actionReceived, this, actionItem);
+ m_executor.execute(job);
+}
+
+void
+Dispatcher::fileSegmentReceivedCallback(const Ccnx::Name &name, const Ccnx::Bytes &content)
+{
+ Executor::Job job = bind(&Dispatcher::fileSegmentReceived, this, name, content);
+ m_executor.execute(job);
+}
+
+void
+Dispatcher::fileReadyCallback(const Ccnx::Name &fileNamePrefix)
+{
+ Executor::Job job = bind(&Dispatcher::fileReady, this, fileNamePrefix);
+ m_executor.execute(job);
+}
+
+void
+Dispatcher::fileChanged(const filesystem::path &relativeFilePath, ActionType type)
+{
+
+ switch (type)
+ {
+ case UPDATE:
+ {
+ filesystem::path absolutePath = m_rootDir / relativeFilePath;
+ if (filesystem::exists(absolutePath))
+ {
+ HashPtr hash = Hash::FromFileContent(absolutePath);
+ if (m_actionLog->KnownFileState(relativeFilePath.string(), *hash))
+ {
+ // the file state is known; i.e. the detected changed file is identical to
+ // the file state kept in FileState table
+ // it is the case that backend puts the file fetched from remote;
+ // we should not publish action for this.
+ }
+ else
+ {
+ uintmax_t fileSize = filesystem::file_size(absolutePath);
+ int seg_num = fileSize / MAX_FILE_SEGMENT_SIZE + ((fileSize % MAX_FILE_SEGMENT_SIZE == 0) ? 0 : 1);
+ time_t wtime = filesystem::last_write_time(absolutePath);
+ filesystem::file_status stat = filesystem::status(absolutePath);
+ int mode = stat.permissions();
+ m_actionLog->AddActionUpdate (relativeFilePath.string(), *hash, wtime, mode, seg_num);
+ // publish the file
+ m_objectManager.localFileToObjects(relativeFilePath, m_localUserName);
+ // notify SyncCore to propagate the change
+ m_core->localStateChanged();
+ }
+ break;
+ }
+ else
+ {
+ BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Update non exist file: " + absolutePath.string() ));
+ }
+ }
+ case DELETE:
+ {
+ m_actionLog->AddActionDelete (relativeFilePath.string());
+ // notify SyncCore to propagate the change
+ m_core->localStateChanged();
+ break;
+ }
+ default:
+ break;
+ }
+}
+
+void
+Dispatcher::syncStateChanged(const SyncStateMsgPtr &stateMsg)
+{
+ int size = stateMsg->state_size();
+ int index = 0;
+ // iterate and fetch the actions
+ while (index < size)
+ {
+ SyncState state = stateMsg->state(index);
+ if (state.has_old_seq() && state.has_seq())
+ {
+ uint64_t oldSeq = state.old_seq();
+ uint64_t newSeq = state.seq();
+ Name userName = state.name();
+ Name locator = state.locator();
+
+ // fetch actions with oldSeq + 1 to newSeq (inclusive)
+ Name actionNameBase(userName);
+ actionNameBase.appendComp("action")
+ .appendComp(m_sharedFolder);
+
+ for (uint64_t seqNo = oldSeq + 1; seqNo <= newSeq; seqNo++)
+ {
+ Name actionName = actionNameBase;
+ actionName.appendComp(seqNo);
+
+ // TODO:
+ // use fetcher to fetch the name with callback "actionRecieved"
+ }
+ }
+ }
+}
+
+void
+Dispatcher::actionReceived(const ActionItemPtr &actionItem)
+{
+ switch (actionItem->action())
+ {
+ case ActionItem::UPDATE:
+ {
+ // @TODO
+ // need a function in ActionLog to apply received action, i.e. record remote action in ActionLog
+
+ string hashBytes = actionItem->file_hash();
+ Hash hash(hashBytes.c_str(), hashBytes.size());
+ ostringstream oss;
+ oss << hash;
+ string hashString = oss.str();
+ ObjectDbPtr db = make_shared<ObjectDb>(m_rootDir, hashString);
+ m_objectDbMap[hashString] = db;
+
+ // TODO:
+ // user fetcher to fetch the file with callback "fileSegmentReceived" for segment callback and "fileReady" for file ready callback
+ break;
+ }
+ case ActionItem::DELETE:
+ {
+ string filename = actionItem->filename();
+ // TODO:
+ // m_actionLog->AddRemoteActionDelete(filename);
+ break;
+ }
+ default:
+ break;
+ }
+}
+
+void
+Dispatcher::fileSegmentReceived(const Ccnx::Name &name, const Ccnx::Bytes &content)
+{
+ int size = name.size();
+ uint64_t segment = name.getCompAsInt(size - 1);
+ Bytes hashBytes = name.getComp(size - 2);
+ Hash hash(head(hashBytes), hashBytes.size());
+ ostringstream oss;
+ oss << hash;
+ string hashString = oss.str();
+ if (m_objectDbMap.find(hashString) != m_objectDbMap.end())
+ {
+ ObjectDbPtr db = m_objectDbMap[hashString];
+ // get the device name
+ // Name deviceName = name.getPartialName();
+ // db->saveContenObject(deviceName, segment, content);
+ }
+ else
+ {
+ cout << "no db available for this content object: " << name << ", size: " << content.size() << endl;
+ }
+}
+
+void
+Dispatcher::fileReady(const Ccnx::Name &fileNamePrefix)
+{
+ int size = fileNamePrefix.size();
+ Bytes hashBytes = fileNamePrefix.getComp(size - 1);
+ Hash hash(head(hashBytes), hashBytes.size());
+ ostringstream oss;
+ oss << hash;
+ string hashString = oss.str();
+
+ if (m_objectDbMap.find(hashString) != m_objectDbMap.end())
+ {
+ // remove the db handle
+ m_objectDbMap.erase(hashString);
+ }
+ else
+ {
+ cout << "no db available for this file: " << fileNamePrefix << endl;
+ }
+
+ // query the action table to get the path on local file system
+ // m_objectManager.objectsToLocalFile(deviceName, hash, relativeFilePath);
+
+}
diff --git a/src/dispatcher.h b/src/dispatcher.h
new file mode 100644
index 0000000..7c3220e
--- /dev/null
+++ b/src/dispatcher.h
@@ -0,0 +1,118 @@
+/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2013 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
+ *
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
+#ifndef DISPATCHER_H
+#define DISPATCHER_H
+
+#include "action-log.h"
+#include "sync-core.h"
+#include "ccnx-wrapper.h"
+#include "executor.h"
+#include "object-db.h"
+#include "object-manager.h"
+#include <boost/function.hpp>
+#include <boost/filesystem.hpp>
+#include <boost/shared_ptr.hpp>
+#include <map>
+
+typedef boost::shared_ptr<ActionItem> ActionItemPtr;
+
+// TODO:
+// This class lacks a permanent table to store the files in fetching process
+// and fetch the missing pieces for those in the table after the application launches
+class Dispatcher
+{
+public:
+ typedef enum
+ {
+ UPDATE = 0,
+ DELETE = 1
+ } ActionType;
+
+ // sharedFolder is the name to be used in NDN name;
+ // rootDir is the shared folder dir in local file system;
+ Dispatcher(const boost::filesystem::path &path, const std::string &localUserName, const Ccnx::Name &localPrefix,
+ const std::string &sharedFolder, const boost::filesystem::path &rootDir,
+ Ccnx::CcnxWrapperPtr ccnx, SchedulerPtr scheduler, int poolSize = 2);
+ ~Dispatcher();
+
+ // ----- Callbacks, they only submit the job to executor and immediately return so that event processing thread won't be blocked for too long -------
+
+ // callback to process local file change
+ void
+ fileChangedCallback(const boost::filesystem::path &relativeFilepath, ActionType type);
+
+ // callback to process remote sync state change
+ void
+ syncStateChangedCallback(const SyncStateMsgPtr &stateMsg);
+
+ // callback to process remote action data
+ void
+ actionReceivedCallback(const ActionItemPtr &actionItem);
+
+ // callback to porcess file data
+ void
+ fileSegmentReceivedCallback(const Ccnx::Name &name, const Ccnx::Bytes &content);
+
+ // callback to assemble file
+ void
+ fileReadyCallback(const Ccnx::Name &fileNamePrefix);
+
+private:
+ void
+ fileChanged(const boost::filesystem::path &relativeFilepath, ActionType type);
+
+ void
+ syncStateChanged(const SyncStateMsgPtr &stateMsg);
+
+ void
+ actionReceived(const ActionItemPtr &actionItem);
+
+ void
+ fileSegmentReceived(const Ccnx::Name &name, const Ccnx::Bytes &content);
+
+ void
+ fileReady(const Ccnx::Name &fileNamePrefix);
+
+private:
+ Ccnx::CcnxWrapperPtr m_ccnx;
+ SyncCore *m_core;
+ SyncLogPtr m_syncLog;
+ ActionLogPtr m_actionLog;
+
+ boost::filesystem::path m_rootDir;
+ Executor m_executor;
+ ObjectManager m_objectManager;
+ Ccnx::Name m_localUserName;
+ // maintain object db ptrs so that we don't need to create them
+ // for every fetched segment of a file
+ map<Ccnx::Name, ObjectDbPtr> m_objectDbMap;
+ std::string m_sharedFolder;
+};
+
+namespace Error
+{
+ struct Dispatcher : virtual boost::exception, virtual std::exception {};
+ typedef boost::error_info<struct tag_errmsg, std::string> error_info_str;
+}
+
+#endif // DISPATCHER_H
+
diff --git a/src/object-db.h b/src/object-db.h
index 9ecd753..8591096 100644
--- a/src/object-db.h
+++ b/src/object-db.h
@@ -27,6 +27,7 @@
#include <ccnx-common.h>
#include <ccnx-name.h>
#include <boost/filesystem.hpp>
+#include <boost/shared_ptr.hpp>
class ObjectDb
{
@@ -51,4 +52,6 @@
sqlite3 *m_db;
};
+typedef boost::shared_ptr<ObjectDb> ObjectDbPtr;
+
#endif // OBJECT_DB_H
diff --git a/src/sync-core.cc b/src/sync-core.cc
index c0dc45b..d85093b 100644
--- a/src/sync-core.cc
+++ b/src/sync-core.cc
@@ -39,10 +39,10 @@
const StateMsgCallback &callback, CcnxWrapperPtr ccnx, SchedulerPtr scheduler)
: m_ccnx (ccnx)
, m_log(syncLog)
- , m_scheduler(scheduler)
- , m_stateMsgCallback(callback)
- , m_syncPrefix(syncPrefix)
- , m_recoverWaitGenerator(new RandomIntervalGenerator(WAIT, RANDOM_PERCENT, RandomIntervalGenerator::UP))
+ , m_scheduler(scheduler)
+ , m_stateMsgCallback(callback)
+ , m_syncPrefix(syncPrefix)
+ , m_recoverWaitGenerator(new RandomIntervalGenerator(WAIT, RANDOM_PERCENT, RandomIntervalGenerator::UP))
{
m_rootHash = m_log->RememberStateInStateLog();
@@ -60,12 +60,23 @@
}
void
+SyncCore::updateLocalPrefix (const Name &localPrefix)
+{
+ m_log->UpdateLocalLocator (localPrefix);
+}
+
+void
SyncCore::updateLocalState(sqlite3_int64 seqno)
{
m_log->UpdateLocalSeqNo (seqno);
+ localStateChanged();
+}
+void
+SyncCore::localStateChanged()
+{
HashPtr oldHash = m_rootHash;
- m_rootHash = m_log->RememberStateInStateLog ();
+ m_rootHash = m_log->RememberStateInStateLog();
SyncStateMsgPtr msg = m_log->FindStateDifferences(*oldHash, *m_rootHash);
@@ -79,7 +90,7 @@
// no hurry in sending out new Sync Interest; if others send the new Sync Interest first, no problem, we know the new root hash already;
// this is trying to avoid the situation that the order of SyncData and new Sync Interest gets reversed at receivers
- Scheduler::scheduleOneTimeTask (m_scheduler, 0.1,
+ Scheduler::scheduleOneTimeTask (m_scheduler, 0.05,
bind(&SyncCore::sendSyncInterest, this),
lexical_cast<string> (*m_rootHash));
@@ -118,9 +129,9 @@
m_ccnx->publishData(name, *syncData, FRESHNESS);
_LOG_TRACE (m_log->GetLocalName () << " publishes " << hash);
_LOG_TRACE (msg);
- }
- else
- {
+ }
+ else
+ {
// we don't recognize this hash, can not help
}
}
@@ -227,7 +238,7 @@
{
string locStr = state.locator();
Name locatorName((const unsigned char *)locStr.c_str(), locStr.size());
- // cout << ", Got loc: " << locatorName << endl;
+ // cout << ", Got loc: " << locatorName << endl;
m_log->UpdateLocator(deviceName, locatorName);
_LOG_TRACE ("self: " << m_log->GetLocalName () << ", device: " << deviceName << " < == > " << locatorName);
diff --git a/src/sync-core.h b/src/sync-core.h
index 63a0f8e..bd7c5e8 100644
--- a/src/sync-core.h
+++ b/src/sync-core.h
@@ -46,7 +46,13 @@
, const StateMsgCallback &callback // callback when state change is detected
, Ccnx::CcnxWrapperPtr ccnx
, SchedulerPtr scheduler);
- ~SyncCore ();
+ ~SyncCore();
+
+ void
+ updateLocalPrefix (const Ccnx::Name &localPrefix);
+
+ void
+ localStateChanged ();
void
updateLocalState (sqlite3_int64);