| /* -*- 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 "logging.h" |
| #include "ccnx-discovery.h" |
| #include "fetch-task-db.h" |
| |
| #include <boost/make_shared.hpp> |
| #include <boost/lexical_cast.hpp> |
| |
| using namespace Ccnx; |
| using namespace std; |
| using namespace boost; |
| |
| INIT_LOGGER ("Dispatcher"); |
| |
| static const string CHRONOSHARE_APP = "chronoshare"; |
| static const string BROADCAST_DOMAIN = "/ndn/broadcast"; |
| |
| static const int CONTENT_FRESHNESS = 1800; // seconds |
| const static double DEFAULT_SYNC_INTEREST_INTERVAL = 10.0; // seconds; |
| |
| Dispatcher::Dispatcher(const std::string &localUserName |
| , const std::string &sharedFolder |
| , const filesystem::path &rootDir |
| , Ccnx::CcnxWrapperPtr ccnx |
| , bool enablePrefixDiscovery |
| ) |
| : m_ccnx(ccnx) |
| , m_core(NULL) |
| , m_rootDir(rootDir) |
| , m_executor(1) // creates problems with file assembly. need to ensure somehow that FinishExectute is called after all Segment_Execute finished |
| , m_objectManager(ccnx, rootDir, CHRONOSHARE_APP) |
| , m_localUserName(localUserName) |
| , m_sharedFolder(sharedFolder) |
| , m_server(NULL) |
| , m_enablePrefixDiscovery(enablePrefixDiscovery) |
| { |
| m_syncLog = make_shared<SyncLog>(m_rootDir, localUserName); |
| m_actionLog = make_shared<ActionLog>(m_ccnx, m_rootDir, m_syncLog, sharedFolder, CHRONOSHARE_APP, |
| // bind (&Dispatcher::Did_ActionLog_ActionApply_AddOrModify, this, _1, _2, _3, _4, _5, _6, _7), |
| ActionLog::OnFileAddedOrChangedCallback (), // don't really need this callback |
| bind (&Dispatcher::Did_ActionLog_ActionApply_Delete, this, _1)); |
| m_fileState = m_actionLog->GetFileState (); |
| |
| Name syncPrefix = Name(BROADCAST_DOMAIN)(CHRONOSHARE_APP)(sharedFolder); |
| |
| // m_server needs a different ccnx face |
| m_server = new ContentServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName, m_sharedFolder, CHRONOSHARE_APP, CONTENT_FRESHNESS); |
| m_server->registerPrefix(Name("/")); |
| m_server->registerPrefix(Name(BROADCAST_DOMAIN)); |
| |
| m_stateServer = new StateServer (make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName, m_sharedFolder, CHRONOSHARE_APP, m_objectManager, CONTENT_FRESHNESS); |
| // no need to register, right now only listening on localhost prefix |
| |
| m_core = new SyncCore (m_syncLog, localUserName, Name("/"), syncPrefix, |
| bind(&Dispatcher::Did_SyncLog_StateChange, this, _1), ccnx, DEFAULT_SYNC_INTEREST_INTERVAL); |
| |
| FetchTaskDbPtr actionTaskDb = make_shared<FetchTaskDb>(m_rootDir, "action"); |
| m_actionFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), |
| Name(BROADCAST_DOMAIN), // no appname suffix now |
| 3, |
| bind (&Dispatcher::Did_FetchManager_ActionFetch, this, _1, _2, _3, _4), FetchManager::FinishCallback(), actionTaskDb); |
| |
| FetchTaskDbPtr fileTaskDb = make_shared<FetchTaskDb>(m_rootDir, "file"); |
| m_fileFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), |
| Name(BROADCAST_DOMAIN), // no appname suffix now |
| 3, |
| bind (&Dispatcher::Did_FetchManager_FileSegmentFetch, this, _1, _2, _3, _4), |
| bind (&Dispatcher::Did_FetchManager_FileFetchComplete, this, _1, _2), |
| fileTaskDb); |
| |
| |
| if (m_enablePrefixDiscovery) |
| { |
| _LOG_DEBUG("registering prefix discovery in Dispatcher"); |
| string tag = "dispatcher" + m_localUserName.toString(); |
| Ccnx::CcnxDiscovery::registerCallback (TaggedFunction (bind (&Dispatcher::Did_LocalPrefix_Updated, this, _1), tag)); |
| } |
| |
| m_executor.start (); |
| } |
| |
| Dispatcher::~Dispatcher() |
| { |
| _LOG_DEBUG ("Enter destructor of dispatcher"); |
| m_executor.shutdown (); |
| |
| // _LOG_DEBUG (">>"); |
| |
| if (m_enablePrefixDiscovery) |
| { |
| _LOG_DEBUG("deregistering prefix discovery in Dispatcher"); |
| string tag = "dispatcher" + m_localUserName.toString(); |
| Ccnx::CcnxDiscovery::deregisterCallback (TaggedFunction (bind (&Dispatcher::Did_LocalPrefix_Updated, this, _1), tag)); |
| } |
| |
| if (m_core != NULL) |
| { |
| delete m_core; |
| m_core = NULL; |
| } |
| |
| if (m_server != NULL) |
| { |
| delete m_server; |
| m_server = NULL; |
| } |
| |
| if (m_stateServer != NULL) |
| { |
| delete m_stateServer; |
| m_stateServer = NULL; |
| } |
| } |
| |
| void |
| Dispatcher::Did_LocalPrefix_Updated (const Ccnx::Name &forwardingHint) |
| { |
| Name effectiveForwardingHint; |
| if (m_localUserName.size () >= forwardingHint.size () && |
| m_localUserName.getPartialName (0, forwardingHint.size ()) == forwardingHint) |
| { |
| effectiveForwardingHint = Name ("/"); // "directly" accesible |
| } |
| else |
| { |
| effectiveForwardingHint = forwardingHint; |
| } |
| |
| Name oldLocalPrefix = m_syncLog->LookupLocalLocator (); |
| |
| if (oldLocalPrefix == effectiveForwardingHint) |
| { |
| _LOG_DEBUG ("Got notification about prefix change from " << oldLocalPrefix << " to: " << forwardingHint << ", but effective prefix didn't change"); |
| return; |
| } |
| |
| if (effectiveForwardingHint == Name ("/") || |
| effectiveForwardingHint == Name (BROADCAST_DOMAIN)) |
| { |
| _LOG_DEBUG ("Basic effective prefix [" << effectiveForwardingHint << "]. Updating local prefix, but don't reregister"); |
| m_syncLog->UpdateLocalLocator (effectiveForwardingHint); |
| return; |
| } |
| |
| _LOG_DEBUG ("LocalPrefix changed from: " << oldLocalPrefix << " to: " << effectiveForwardingHint); |
| |
| m_server->registerPrefix(effectiveForwardingHint); |
| m_syncLog->UpdateLocalLocator (effectiveForwardingHint); |
| |
| if (oldLocalPrefix == Name ("/") || |
| oldLocalPrefix == Name (BROADCAST_DOMAIN)) |
| { |
| _LOG_DEBUG ("Don't deregister basic prefix: " << oldLocalPrefix); |
| return; |
| } |
| m_server->deregisterPrefix(oldLocalPrefix); |
| } |
| |
| // moved to state-server |
| // void |
| // Dispatcher::Restore_LocalFile (FileItemPtr file) |
| // { |
| // m_executor.execute (bind (&Dispatcher::Restore_LocalFile_Execute, this, file)); |
| // } |
| |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| |
| void |
| Dispatcher::Did_LocalFile_AddOrModify (const filesystem::path &relativeFilePath) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_LocalFile_AddOrModify_Execute, this, relativeFilePath)); |
| } |
| |
| void |
| Dispatcher::Did_LocalFile_AddOrModify_Execute (filesystem::path relativeFilePath) |
| { |
| _LOG_DEBUG(m_localUserName << " calls LocalFile_AddOrModify_Execute"); |
| filesystem::path absolutePath = m_rootDir / relativeFilePath; |
| if (!filesystem::exists(absolutePath)) |
| { |
| //BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Update non exist file: " + absolutePath.string() )); |
| _LOG_DEBUG("Update non exist file: " << absolutePath.string()); |
| return; |
| } |
| |
| FileItemPtr currentFile = m_fileState->LookupFile (relativeFilePath.generic_string ()); |
| if (currentFile && |
| *Hash::FromFileContent (absolutePath) == Hash (currentFile->file_hash ().c_str (), currentFile->file_hash ().size ()) |
| // The following two are commented out to prevent front end from reporting intermediate files |
| // should enable it if there is other way to prevent this |
| // && last_write_time (absolutePath) == currentFile->mtime () |
| // && status (absolutePath).permissions () == static_cast<filesystem::perms> (currentFile->mode ()) |
| ) |
| { |
| _LOG_ERROR ("Got notification about the same file [" << relativeFilePath << "]"); |
| return; |
| } |
| |
| if (currentFile && |
| !currentFile->is_complete ()) |
| { |
| _LOG_ERROR ("Got notification about incomplete file [" << relativeFilePath << "]"); |
| return; |
| } |
| |
| |
| int seg_num; |
| HashPtr hash; |
| tie (hash, seg_num) = m_objectManager.localFileToObjects (absolutePath, m_localUserName); |
| |
| try |
| { |
| m_actionLog->AddLocalActionUpdate (relativeFilePath.generic_string(), |
| *hash, |
| last_write_time (absolutePath), status (absolutePath).permissions (), seg_num); |
| |
| // notify SyncCore to propagate the change |
| m_core->localStateChangedDelayed (); |
| } |
| catch (filesystem::filesystem_error &error) |
| { |
| _LOG_ERROR ("File operations failed on [" << relativeFilePath << "] (ignoring)"); |
| } |
| } |
| |
| void |
| Dispatcher::Did_LocalFile_Delete (const filesystem::path &relativeFilePath) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_LocalFile_Delete_Execute, this, relativeFilePath)); |
| } |
| |
| void |
| Dispatcher::Did_LocalFile_Delete_Execute (filesystem::path relativeFilePath) |
| { |
| filesystem::path absolutePath = m_rootDir / relativeFilePath; |
| if (filesystem::exists(absolutePath)) |
| { |
| //BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Delete notification but file exists: " + absolutePath.string() )); |
| _LOG_ERROR("DELETE command, but file still exists: " << absolutePath.string()); |
| return; |
| } |
| |
| m_actionLog->AddLocalActionDelete (relativeFilePath.generic_string()); |
| // notify SyncCore to propagate the change |
| m_core->localStateChangedDelayed(); |
| } |
| |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| ///////////////////////////////////////////////////////////////////////////////////////////////////// |
| |
| /** |
| * Callbacks: |
| * |
| * - from SyncLog: when state changes -> to fetch missing actions |
| * |
| * - from FetchManager/Actions: when action is fetched -> to request a file, specified by the action |
| * -> to add action to the action log |
| * |
| * - from ActionLog/Delete: when action applied (file state changed, file deleted) -> to delete local file |
| * |
| * - from ActionLog/AddOrUpdate: when action applied (file state changes, file added or modified) -> do nothing? |
| * |
| * - from FetchManager/Files: when file segment is retrieved -> save it in ObjectDb |
| * when file fetch is completed -> if file belongs to FileState, then assemble it to filesystem. Don't do anything otherwise |
| */ |
| |
| void |
| Dispatcher::Did_SyncLog_StateChange (SyncStateMsgPtr stateMsg) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_SyncLog_StateChange_Execute, this, stateMsg)); |
| } |
| |
| void |
| Dispatcher::Did_SyncLog_StateChange_Execute (SyncStateMsgPtr stateMsg) |
| { |
| int size = stateMsg->state_size(); |
| int index = 0; |
| // iterate and fetch the actions |
| for (; index < size; index++) |
| { |
| 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 (reinterpret_cast<const unsigned char *> (state.name ().c_str ()), state.name ().size ()); |
| |
| // fetch actions with oldSeq + 1 to newSeq (inclusive) |
| Name actionNameBase = Name ("/")(userName)(CHRONOSHARE_APP)("action")(m_sharedFolder); |
| |
| m_actionFetcher->Enqueue (userName, actionNameBase, |
| std::max<uint64_t> (oldSeq + 1, 1), newSeq, FetchManager::PRIORITY_HIGH); |
| } |
| } |
| } |
| |
| |
| void |
| Dispatcher::Did_FetchManager_ActionFetch (const Ccnx::Name &deviceName, const Ccnx::Name &actionBaseName, uint32_t seqno, Ccnx::PcoPtr actionPco) |
| { |
| /// @todo Errors and exception checking |
| _LOG_DEBUG ("Received action deviceName: " << deviceName << ", actionBaseName: " << actionBaseName << ", seqno: " << seqno); |
| |
| ActionItemPtr action = m_actionLog->AddRemoteAction (deviceName, seqno, actionPco); |
| if (!action) |
| { |
| _LOG_ERROR ("AddRemoteAction did not insert action, ignoring"); |
| return; |
| } |
| // trigger may invoke Did_ActionLog_ActionApply_Delete or Did_ActionLog_ActionApply_AddOrModify callbacks |
| |
| if (action->action () == ActionItem::UPDATE) |
| { |
| Hash hash (action->file_hash ().c_str(), action->file_hash ().size ()); |
| |
| Name fileNameBase = Name ("/")(deviceName)(CHRONOSHARE_APP)("file")(hash.GetHash (), hash.GetHashBytes ()); |
| |
| string hashStr = lexical_cast<string> (hash); |
| if (ObjectDb::DoesExist (m_rootDir / ".chronoshare", deviceName, hashStr)) |
| { |
| _LOG_DEBUG ("File already exists in the database. No need to refetch, just directly applying the action"); |
| Did_FetchManager_FileFetchComplete (deviceName, fileNameBase); |
| } |
| else |
| { |
| if (m_objectDbMap.find (hash) == m_objectDbMap.end ()) |
| { |
| _LOG_DEBUG ("create ObjectDb for " << hash); |
| m_objectDbMap [hash] = make_shared<ObjectDb> (m_rootDir / ".chronoshare", hashStr); |
| } |
| |
| m_fileFetcher->Enqueue (deviceName, fileNameBase, |
| 0, action->seg_num () - 1, FetchManager::PRIORITY_NORMAL); |
| } |
| } |
| // if necessary (when version number is the highest) delete will be applied through the trigger in m_actionLog->AddRemoteAction call |
| } |
| |
| void |
| Dispatcher::Did_ActionLog_ActionApply_Delete (const std::string &filename) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_ActionLog_ActionApply_Delete_Execute, this, filename)); |
| } |
| |
| void |
| Dispatcher::Did_ActionLog_ActionApply_Delete_Execute (std::string filename) |
| { |
| _LOG_DEBUG ("Action to delete " << filename); |
| |
| filesystem::path absolutePath = m_rootDir / filename; |
| try |
| { |
| if (filesystem::exists(absolutePath)) |
| { |
| // need some protection from local detection of removal |
| remove (absolutePath); |
| |
| // hack to remove empty parent dirs |
| filesystem::path parentPath = absolutePath.parent_path(); |
| while (parentPath > m_rootDir) |
| { |
| if (filesystem::is_empty(parentPath)) |
| { |
| filesystem::remove(parentPath); |
| parentPath = parentPath.parent_path(); |
| } |
| else |
| { |
| break; |
| } |
| } |
| } |
| // don't exist |
| } |
| catch (filesystem::filesystem_error &error) |
| { |
| _LOG_ERROR ("File operations failed when removing [" << absolutePath << "] (ignoring)"); |
| } |
| } |
| |
| void |
| Dispatcher::Did_FetchManager_FileSegmentFetch (const Ccnx::Name &deviceName, const Ccnx::Name &fileSegmentBaseName, uint32_t segment, Ccnx::PcoPtr fileSegmentPco) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_FetchManager_FileSegmentFetch_Execute, this, deviceName, fileSegmentBaseName, segment, fileSegmentPco)); |
| } |
| |
| void |
| Dispatcher::Did_FetchManager_FileSegmentFetch_Execute (Ccnx::Name deviceName, Ccnx::Name fileSegmentBaseName, uint32_t segment, Ccnx::PcoPtr fileSegmentPco) |
| { |
| // fileSegmentBaseName: /<device_name>/<appname>/file/<hash> |
| |
| const Bytes &hashBytes = fileSegmentBaseName.getCompFromBack (0); |
| Hash hash (head(hashBytes), hashBytes.size()); |
| |
| _LOG_DEBUG ("Received segment deviceName: " << deviceName << ", segmentBaseName: " << fileSegmentBaseName << ", segment: " << segment); |
| |
| // _LOG_DEBUG ("Looking up objectdb for " << hash); |
| |
| map<Hash, ObjectDbPtr>::iterator db = m_objectDbMap.find (hash); |
| if (db != m_objectDbMap.end()) |
| { |
| db->second->saveContentObject(deviceName, segment, fileSegmentPco->buf ()); |
| } |
| else |
| { |
| _LOG_ERROR ("no db available for this content object: " << fileSegmentBaseName << ", size: " << fileSegmentPco->buf ().size()); |
| } |
| |
| // ObjectDb objectDb (m_rootDir / ".chronoshare", lexical_cast<string> (hash)); |
| // objectDb.saveContentObject(deviceName, segment, fileSegmentPco->buf ()); |
| } |
| |
| void |
| Dispatcher::Did_FetchManager_FileFetchComplete (const Ccnx::Name &deviceName, const Ccnx::Name &fileBaseName) |
| { |
| m_executor.execute (bind (&Dispatcher::Did_FetchManager_FileFetchComplete_Execute, this, deviceName, fileBaseName)); |
| } |
| |
| void |
| Dispatcher::Did_FetchManager_FileFetchComplete_Execute (Ccnx::Name deviceName, Ccnx::Name fileBaseName) |
| { |
| // fileBaseName: /<device_name>/<appname>/file/<hash> |
| |
| _LOG_DEBUG ("Finished fetching " << deviceName << ", fileBaseName: " << fileBaseName); |
| |
| const Bytes &hashBytes = fileBaseName.getCompFromBack (0); |
| Hash hash (head (hashBytes), hashBytes.size ()); |
| _LOG_DEBUG ("Extracted hash: " << hash.shortHash ()); |
| |
| if (m_objectDbMap.find (hash) != m_objectDbMap.end()) |
| { |
| // remove the db handle |
| m_objectDbMap.erase (hash); // to commit write |
| } |
| else |
| { |
| _LOG_ERROR ("no db available for this file: " << hash); |
| } |
| |
| FileItemsPtr filesToAssemble = m_fileState->LookupFilesForHash (hash); |
| |
| for (FileItems::iterator file = filesToAssemble->begin (); |
| file != filesToAssemble->end (); |
| file++) |
| { |
| boost::filesystem::path filePath = m_rootDir / file->filename (); |
| |
| try |
| { |
| if (filesystem::exists (filePath) && |
| filesystem::last_write_time (filePath) == file->mtime () && |
| filesystem::status (filePath).permissions () == static_cast<filesystem::perms> (file->mode ()) && |
| *Hash::FromFileContent (filePath) == hash) |
| { |
| _LOG_DEBUG ("Asking to assemble a file, but file already exists on a filesystem"); |
| continue; |
| } |
| } |
| catch (filesystem::filesystem_error &error) |
| { |
| _LOG_ERROR ("File operations failed on [" << filePath << "] (ignoring)"); |
| } |
| |
| if (ObjectDb::DoesExist (m_rootDir / ".chronoshare", deviceName, boost::lexical_cast<string>(hash))) |
| { |
| bool ok = m_objectManager.objectsToLocalFile (deviceName, hash, filePath); |
| if (ok) |
| { |
| last_write_time (filePath, file->mtime ()); |
| permissions (filePath, static_cast<filesystem::perms> (file->mode ())); |
| |
| m_fileState->SetFileComplete (file->filename ()); |
| } |
| else |
| { |
| _LOG_ERROR ("Notified about complete fetch, but file cannot be restored from the database: [" << filePath << "]"); |
| } |
| } |
| else |
| { |
| _LOG_ERROR (filePath << " supposed to have all segments, but not"); |
| // should abort for debugging |
| } |
| } |
| } |
| |
| // moved to state-server |
| // void |
| // Dispatcher::Restore_LocalFile_Execute (FileItemPtr file) |
| // { |
| // _LOG_DEBUG ("Got request to restore local file [" << file->filename () << "]"); |
| // // the rest will gracefully fail if object-db is missing or incomplete |
| |
| // boost::filesystem::path filePath = m_rootDir / file->filename (); |
| // Name deviceName (file->device_name ().c_str (), file->device_name ().size ()); |
| // Hash hash (file->file_hash ().c_str (), file->file_hash ().size ()); |
| |
| // try |
| // { |
| // if (filesystem::exists (filePath) && |
| // filesystem::last_write_time (filePath) == file->mtime () && |
| // filesystem::status (filePath).permissions () == static_cast<filesystem::perms> (file->mode ()) && |
| // *Hash::FromFileContent (filePath) == hash) |
| // { |
| // _LOG_DEBUG ("Asking to assemble a file, but file already exists on a filesystem"); |
| // return; |
| // } |
| // } |
| // catch (filesystem::filesystem_error &error) |
| // { |
| // _LOG_ERROR ("File operations failed on [" << filePath << "] (ignoring)"); |
| // } |
| |
| // m_objectManager.objectsToLocalFile (deviceName, hash, filePath); |
| |
| // last_write_time (filePath, file->mtime ()); |
| // permissions (filePath, static_cast<filesystem::perms> (file->mode ())); |
| |
| // } |