blob: ba11b6732f35a077f804135967f7da1b3fb0c54d [file] [log] [blame]
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -08001/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
2/*
3 * Copyright (c) 2013 University of California, Los Angeles
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License version 2 as
7 * published by the Free Software Foundation;
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
14 * You should have received a copy of the GNU General Public License
15 * along with this program; if not, write to the Free Software
16 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17 *
18 * Zhenkai Zhu <zhenkai@cs.ucla.edu>
19 * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
20 */
21
22#include "dispatcher.h"
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080023#include "logging.h"
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080024#include "ccnx-discovery.h"
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080025
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080026#include <boost/make_shared.hpp>
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080027#include <boost/lexical_cast.hpp>
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080028
29using namespace Ccnx;
30using namespace std;
31using namespace boost;
32
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080033INIT_LOGGER ("Dispatcher");
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080034
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080035static const string BROADCAST_DOMAIN = "/ndn/broadcast/chronoshare";
36
37Dispatcher::Dispatcher(const filesystem::path &path, const std::string &localUserName,
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080038 const std::string &sharedFolder,
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080039 const filesystem::path &rootDir, Ccnx::CcnxWrapperPtr ccnx,
Alexander Afanasyevd94a8c62013-01-24 13:53:40 -080040 int poolSize)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080041 : m_ccnx(ccnx)
42 , m_core(NULL)
43 , m_rootDir(rootDir)
44 , m_executor(poolSize)
45 , m_objectManager(ccnx, rootDir)
46 , m_localUserName(localUserName)
47 , m_sharedFolder(sharedFolder)
Zhenkai Zhu1dcbbab2013-01-22 16:03:20 -080048 , m_server(NULL)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080049{
Alexander Afanasyevcbda9922013-01-22 11:21:12 -080050 m_syncLog = make_shared<SyncLog>(path, localUserName);
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080051 m_actionLog = make_shared<ActionLog>(m_ccnx, path, m_syncLog, sharedFolder,
52 // bind (&Dispatcher::Did_ActionLog_ActionApply_AddOrModify, this, _1, _2, _3, _4, _5, _6, _7),
53 ActionLog::OnFileAddedOrChangedCallback (), // don't really need this callback
54 bind (&Dispatcher::Did_ActionLog_ActionApply_Delete, this, _1));
Alexander Afanasyev1b15cc62013-01-22 17:00:40 -080055 Name syncPrefix = Name(BROADCAST_DOMAIN)(sharedFolder);
Alexander Afanasyevcbda9922013-01-22 11:21:12 -080056
Zhenkai Zhu1dcbbab2013-01-22 16:03:20 -080057 m_server = new ContentServer(m_ccnx, m_actionLog, rootDir);
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080058 m_server->registerPrefix(Name ("/"));
Zhenkai Zhu1dcbbab2013-01-22 16:03:20 -080059 m_server->registerPrefix(syncPrefix);
60
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080061 m_core = new SyncCore (m_syncLog, localUserName, Name ("/"), syncPrefix,
Alexander Afanasyevd94a8c62013-01-24 13:53:40 -080062 bind(&Dispatcher::Did_SyncLog_StateChange, this, _1), ccnx);
Zhenkai Zhu1dcbbab2013-01-22 16:03:20 -080063
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080064 m_actionFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), 3);
65 m_fileFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), 3);
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080066
67 Ccnx::CcnxDiscovery::registerCallback (TaggedFunction (bind (&Dispatcher::Did_LocalPrefix_Updated, this, _1), "dispatcher"));
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080068}
69
70Dispatcher::~Dispatcher()
71{
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080072 Ccnx::CcnxDiscovery::deregisterCallback (TaggedFunction (bind (&Dispatcher::Did_LocalPrefix_Updated, this, _1), "dispatcher"));
73
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080074 if (m_core != NULL)
75 {
76 delete m_core;
77 m_core = NULL;
78 }
Zhenkai Zhu1dcbbab2013-01-22 16:03:20 -080079
80 if (m_server != NULL)
81 {
82 delete m_server;
83 m_server = NULL;
84 }
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -080085}
86
Alexander Afanasyev758f51b2013-01-24 13:48:18 -080087void
88Dispatcher::Did_LocalPrefix_Updated (const Ccnx::Name &prefix)
89{
90 Name oldLocalPrefix = m_syncLog->LookupLocalLocator ();
91 _LOG_DEBUG ("LocalPrefix changed from: " << oldLocalPrefix << " to: " << prefix);
92
93 m_server->deregisterPrefix(prefix);
94 m_syncLog->UpdateLocalLocator (prefix);
95 m_server->deregisterPrefix(oldLocalPrefix);
96}
97
Alexander Afanasyevf9978f82013-01-23 16:30:31 -080098/////////////////////////////////////////////////////////////////////////////////////////////////////
99/////////////////////////////////////////////////////////////////////////////////////////////////////
100/////////////////////////////////////////////////////////////////////////////////////////////////////
101
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800102void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800103Dispatcher::Did_LocalFile_AddOrModify (const filesystem::path &relativeFilePath)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800104{
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800105 m_executor.execute (bind (&Dispatcher::Did_LocalFile_AddOrModify_Execute, this, relativeFilePath));
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800106}
107
108void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800109Dispatcher::Did_LocalFile_AddOrModify_Execute (filesystem::path relativeFilePath)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800110{
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800111 filesystem::path absolutePath = m_rootDir / relativeFilePath;
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800112 if (!filesystem::exists(absolutePath))
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800113 {
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800114 BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Update non exist file: " + absolutePath.string() ));
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800115 }
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800116
117 FileItemPtr currentFile = m_actionLog->LookupFile (relativeFilePath.generic_string ());
118 if (currentFile &&
119 *Hash::FromFileContent (absolutePath) == Hash (currentFile->file_hash ().c_str (), currentFile->file_hash ().size ()) &&
120 last_write_time (absolutePath) == currentFile->mtime () &&
121 status (absolutePath).permissions () == static_cast<filesystem::perms> (currentFile->mode ()))
122 {
123 _LOG_ERROR ("Got notification about the same file [" << relativeFilePath << "]");
124 return;
125 }
126
127 int seg_num;
128 HashPtr hash;
129 tie (hash, seg_num) = m_objectManager.localFileToObjects (absolutePath, m_localUserName);
130
131 m_actionLog->AddLocalActionUpdate (relativeFilePath.generic_string(),
132 *hash,
133 last_write_time (absolutePath), status (absolutePath).permissions (), seg_num);
134
135 // notify SyncCore to propagate the change
136 m_core->localStateChanged();
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800137}
138
139void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800140Dispatcher::Did_LocalFile_Delete (const filesystem::path &relativeFilePath)
141{
142 m_executor.execute (bind (&Dispatcher::Did_LocalFile_Delete_Execute, this, relativeFilePath));
143}
144
145void
146Dispatcher::Did_LocalFile_Delete_Execute (filesystem::path relativeFilePath)
147{
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800148 filesystem::path absolutePath = m_rootDir / relativeFilePath;
149 if (!filesystem::exists(absolutePath))
150 {
151 BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Delete notification but file exists: " + absolutePath.string() ));
152 }
153
154 FileItemPtr currentFile = m_actionLog->LookupFile (relativeFilePath.generic_string ());
155 if (!currentFile)
156 {
157 _LOG_ERROR ("File already deleted [" << relativeFilePath << "]");
158 return;
159 }
160
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800161 m_actionLog->AddLocalActionDelete (relativeFilePath.generic_string());
162 // notify SyncCore to propagate the change
163 m_core->localStateChanged();
164}
165
166/////////////////////////////////////////////////////////////////////////////////////////////////////
167/////////////////////////////////////////////////////////////////////////////////////////////////////
168/////////////////////////////////////////////////////////////////////////////////////////////////////
169
170/**
171 * Callbacks:
172 *
173 * - from SyncLog: when state changes -> to fetch missing actions
174 *
175 * - from FetchManager/Actions: when action is fetched -> to request a file, specified by the action
176 * -> to add action to the action log
177 *
178 * - from ActionLog/Delete: when action applied (file state changed, file deleted) -> to delete local file
179 *
180 * - from ActionLog/AddOrUpdate: when action applied (file state changes, file added or modified) -> do nothing?
181 *
182 * - from FetchManager/Files: when file segment is retrieved -> save it in ObjectDb
183 * when file fetch is completed -> if file belongs to FileState, then assemble it to filesystem. Don't do anything otherwise
184 */
185
186void
187Dispatcher::Did_SyncLog_StateChange (const SyncStateMsgPtr &stateMsg)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800188{
189 int size = stateMsg->state_size();
190 int index = 0;
191 // iterate and fetch the actions
192 while (index < size)
193 {
194 SyncState state = stateMsg->state(index);
195 if (state.has_old_seq() && state.has_seq())
196 {
197 uint64_t oldSeq = state.old_seq();
198 uint64_t newSeq = state.seq();
199 Name userName = state.name();
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800200
201 // fetch actions with oldSeq + 1 to newSeq (inclusive)
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800202 Name actionNameBase = Name(userName)("action")(m_sharedFolder);
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800203
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800204 m_actionFetcher->Enqueue (userName, actionNameBase,
205 bind (&Dispatcher::Did_FetchManager_ActionFetch, this, _1, _2, _3, _4), FetchManager::FinishCallback (),
206 oldSeq + 1, newSeq, FetchManager::PRIORITY_HIGH);
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800207 }
208 }
209}
210
211void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800212Dispatcher::Did_FetchManager_ActionFetch (const Ccnx::Name &deviceName, const Ccnx::Name &actionName, uint32_t seqno, Ccnx::PcoPtr actionPco)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800213{
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800214 /// @todo Errors and exception checking
215 _LOG_DEBUG ("Received action deviceName: " << deviceName << ", actionName: " << actionName << ", seqno: " << seqno);
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800216
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800217 ActionItemPtr action = m_actionLog->AddRemoteAction (deviceName, seqno, actionPco);
218 // trigger may invoke Did_ActionLog_ActionApply_Delete or Did_ActionLog_ActionApply_AddOrModify callbacks
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800219
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800220 if (action->action () == ActionItem::UPDATE)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800221 {
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800222 Hash hash (action->file_hash ().c_str(), action->file_hash ().size ());
223
224 Name fileNameBase = Name (deviceName)("file")(hash.GetHash (), hash.GetHashBytes ());
225
226 if (m_objectDbMap.find (hash) == m_objectDbMap.end ())
227 {
228 m_objectDbMap [hash] = make_shared<ObjectDb> (m_rootDir, lexical_cast<string> (hash));
229 }
230
231 m_fileFetcher->Enqueue (deviceName, fileNameBase,
232 bind (&Dispatcher::Did_FetchManager_FileSegmentFetch, this, _1, _2, _3, _4),
233 bind (&Dispatcher::Did_FetchManager_FileFetchComplete, this, _1, _2),
234 0, action->seg_num (), FetchManager::PRIORITY_NORMAL);
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800235 }
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800236}
237
238void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800239Dispatcher::Did_ActionLog_ActionApply_Delete (const std::string &filename)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800240{
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800241 m_executor.execute (bind (&Dispatcher::Did_ActionLog_ActionApply_Delete_Execute, this, filename));
242}
243
244void
245Dispatcher::Did_ActionLog_ActionApply_Delete_Execute (std::string filename)
246{
247 _LOG_DEBUG ("Action to delete " << filename);
248
249 filesystem::path absolutePath = m_rootDir / filename;
250 if (filesystem::exists(absolutePath))
251 {
252 // need some protection from local detection of removal
253 remove (absolutePath);
254 }
255 // don't exist
256}
257
258void
259Dispatcher::Did_FetchManager_FileSegmentFetch (const Ccnx::Name &deviceName, const Ccnx::Name &fileSegmentName, uint32_t segment, Ccnx::PcoPtr fileSegmentPco)
260{
261 m_executor.execute (bind (&Dispatcher::Did_FetchManager_FileSegmentFetch_Execute, this, deviceName, fileSegmentName, segment, fileSegmentPco));
262}
263
264void
265Dispatcher::Did_FetchManager_FileSegmentFetch_Execute (Ccnx::Name deviceName, Ccnx::Name fileSegmentName, uint32_t segment, Ccnx::PcoPtr fileSegmentPco)
266{
267 const Bytes &hashBytes = fileSegmentName.getCompFromBack (1);
268 Hash hash (head(hashBytes), hashBytes.size());
269
270 _LOG_DEBUG ("Received segment deviceName: " << deviceName << ", segmentName: " << fileSegmentName << ", segment: " << segment);
271
272 map<Hash, ObjectDbPtr>::iterator db = m_objectDbMap.find (hash);
273 if (db != m_objectDbMap.end())
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800274 {
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800275 db->second->saveContentObject(deviceName, segment, fileSegmentPco->buf ());
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800276 }
277 else
278 {
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800279 _LOG_ERROR ("no db available for this content object: " << fileSegmentName << ", size: " << fileSegmentPco->buf ().size());
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800280 }
281}
282
283void
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800284Dispatcher::Did_FetchManager_FileFetchComplete (const Ccnx::Name &deviceName, const Ccnx::Name &fileBaseName)
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800285{
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800286 m_executor.execute (bind (&Dispatcher::Did_FetchManager_FileFetchComplete_Execute, this, deviceName, fileBaseName));
287}
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800288
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800289void
290Dispatcher::Did_FetchManager_FileFetchComplete_Execute (Ccnx::Name deviceName, Ccnx::Name fileBaseName)
291{
292 _LOG_DEBUG ("Finished fetching " << deviceName << ", fileBaseName: " << fileBaseName);
Alexander Afanasyevf9978f82013-01-23 16:30:31 -0800293
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800294 const Bytes &hashBytes = fileBaseName.getCompFromBack (1);
295 Hash hash (head (hashBytes), hashBytes.size ());
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800296
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800297 FileItemsPtr filesToAssemble = m_actionLog->LookupFilesForHash (hash);
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800298
Alexander Afanasyevf2c16e02013-01-23 18:08:04 -0800299 for (FileItems::iterator file = filesToAssemble->begin ();
300 file != filesToAssemble->end ();
301 file++)
302 {
303 boost::filesystem::path filePath = m_rootDir / file->filename ();
304 m_objectManager.objectsToLocalFile (deviceName, hash, filePath);
305
306 last_write_time (filePath, file->mtime ());
307 permissions (filePath, static_cast<filesystem::perms> (file->mode ()));
308 }
309
310 if (m_objectDbMap.find (hash) != m_objectDbMap.end())
311 {
312 // remove the db handle
313 m_objectDbMap.erase (hash);
314 }
315 else
316 {
317 _LOG_ERROR ("no db available for this file: " << hash);
318 }
Zhenkai Zhuc3fd51e2013-01-22 10:45:54 -0800319}