Merge changes Id564ca35,I6082a335,I54210671
* changes:
Printing error when file is requested to be assembled, but something is wrong with assembly
Delaying saving of the local state change from dispatcher (useful when many changes happen at about the same time)
ccnx: Update tracing in ccnx-wrapper
diff --git a/ccnx/ccnx-wrapper.cpp b/ccnx/ccnx-wrapper.cpp
index 2867ecc..d1c4672 100644
--- a/ccnx/ccnx-wrapper.cpp
+++ b/ccnx/ccnx-wrapper.cpp
@@ -262,8 +262,6 @@
tuple<CcnxWrapper::InterestCallback *, ExecutorPtr> *realData = reinterpret_cast< tuple<CcnxWrapper::InterestCallback *, ExecutorPtr>* > (selfp->data);
tie (f, executor) = *realData;
- _LOG_TRACE (">> incomingInterest upcall");
-
switch (kind)
{
case CCN_UPCALL_FINAL: // effective in unit tests
@@ -275,10 +273,11 @@
return CCN_UPCALL_RESULT_OK;
case CCN_UPCALL_INTEREST:
+ _LOG_TRACE (">> incomingInterest upcall: " << Name(info->interest_ccnb, info->interest_comps));
break;
default:
- _LOG_TRACE ("<< incomingInterest with CCN_UPCALL_RESULT_OK");
+ _LOG_TRACE ("<< incomingInterest with CCN_UPCALL_RESULT_OK: " << Name(info->interest_ccnb, info->interest_comps));
return CCN_UPCALL_RESULT_OK;
}
@@ -290,7 +289,6 @@
// (*f) (interest);
// closure->runInterestCallback(interest);
- _LOG_TRACE ("<< incomingInterest");
return CCN_UPCALL_RESULT_OK;
}
@@ -313,8 +311,6 @@
tuple<Closure *, ExecutorPtr, Selectors> *realData = reinterpret_cast< tuple<Closure*, ExecutorPtr, Selectors>* > (selfp->data);
tie (cp, executor, selectors) = *realData;
- _LOG_TRACE (">> incomingData upcall");
-
switch (kind)
{
case CCN_UPCALL_FINAL: // effecitve in unit tests
@@ -326,18 +322,19 @@
return CCN_UPCALL_RESULT_OK;
case CCN_UPCALL_CONTENT:
+ _LOG_TRACE (">> incomingData content upcall: " << Name (info->content_ccnb, info->content_comps));
break;
case CCN_UPCALL_INTEREST_TIMED_OUT: {
if (cp != NULL)
{
- _LOG_TRACE ("<< incomingData timeout");
Name interest(info->interest_ccnb, info->interest_comps);
+ _LOG_TRACE ("<< incomingData timeout: " << Name (info->interest_ccnb, info->interest_comps));
executor->execute (bind (&Closure::runTimeoutCallback, cp, interest, *cp, selectors));
}
else
{
- _LOG_TRACE ("<< incomingData timeout, but callback is not set...");
+ _LOG_TRACE ("<< incomingData timeout, but callback is not set...: " << Name (info->interest_ccnb, info->interest_comps));
}
return CCN_UPCALL_RESULT_OK;
}
@@ -348,24 +345,8 @@
PcoPtr pco = make_shared<ParsedContentObject> (info->content_ccnb, info->pco->offset[CCN_PCO_E]);
- // const unsigned char *pcontent;
- // size_t len;
- // if (ccn_content_get_value(info->content_ccnb, info->pco->offset[CCN_PCO_E], info->pco, &pcontent, &len) < 0)
- // {
- // // BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("decode ContentObject failed"));
- // }
-
- // Name name(info->content_ccnb, info->content_comps);
-
- // Bytes content;
- // // copy content and do processing on the copy
- // // otherwise the pointed memory may have been changed during the processing
- // readRaw(content, pcontent, len);
-
// this will be run in executor
executor->execute (bind (&Closure::runDataCallback, cp, pco->name (), pco));
- // cp->runDataCallback (pco->name (), pco);
-
_LOG_TRACE (">> incomingData");
return CCN_UPCALL_RESULT_OK;
@@ -373,7 +354,7 @@
int CcnxWrapper::sendInterest (const Name &interest, const Closure &closure, const Selectors &selectors)
{
- _LOG_TRACE (">> sendInterest");
+ _LOG_TRACE (">> sendInterest: " << interest);
{
UniqueRecLock lock(m_mutex);
if (!m_running || !m_connected)
@@ -406,8 +387,6 @@
_LOG_ERROR ("<< sendInterest: ccn_express_interest FAILED!!!");
}
- _LOG_TRACE ("<< sendInterest");
-
return 0;
}
diff --git a/src/dispatcher.cc b/src/dispatcher.cc
index 0a2bb82..9c611cb 100644
--- a/src/dispatcher.cc
+++ b/src/dispatcher.cc
@@ -168,6 +168,14 @@
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);
@@ -179,7 +187,7 @@
last_write_time (absolutePath), status (absolutePath).permissions (), seg_num);
// notify SyncCore to propagate the change
- m_core->localStateChanged();
+ m_core->localStateChangedDelayed ();
}
catch (filesystem::filesystem_error &error)
{
@@ -213,7 +221,7 @@
m_actionLog->AddLocalActionDelete (relativeFilePath.generic_string());
// notify SyncCore to propagate the change
- m_core->localStateChanged();
+ m_core->localStateChangedDelayed();
}
/////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -300,6 +308,7 @@
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
@@ -402,12 +411,18 @@
if (ObjectDb::DoesExist (m_rootDir / ".chronoshare", deviceName, boost::lexical_cast<string>(hash)))
{
- m_objectManager.objectsToLocalFile (deviceName, hash, filePath);
+ bool ok = m_objectManager.objectsToLocalFile (deviceName, hash, filePath);
+ if (ok)
+ {
+ last_write_time (filePath, file->mtime ());
+ permissions (filePath, static_cast<filesystem::perms> (file->mode ()));
- last_write_time (filePath, file->mtime ());
- permissions (filePath, static_cast<filesystem::perms> (file->mode ()));
-
- m_actionLog->SetFileComplete (file->filename ());
+ m_actionLog->SetFileComplete (file->filename ());
+ }
+ else
+ {
+ _LOG_ERROR ("Notified about complete fetch, but file cannot be restored from the database: [" << filePath << "]");
+ }
}
else
{
diff --git a/src/sync-core.cc b/src/sync-core.cc
index 4aaf1a5..f744f26 100644
--- a/src/sync-core.cc
+++ b/src/sync-core.cc
@@ -38,6 +38,8 @@
const std::string SYNC_INTEREST_TAG = "send-sync-interest";
const std::string SYNC_INTEREST_TAG2 = "send-sync-interest2";
+const std::string LOCAL_STATE_CHANGE_DELAYED_TAG = "local-state-changed";
+
using namespace boost;
using namespace Ccnx;
@@ -112,6 +114,15 @@
}
void
+SyncCore::localStateChangedDelayed ()
+{
+ // many calls to localStateChangedDelayed within 0.5 second will be suppressed to one localStateChanged calls
+ Scheduler::scheduleOneTimeTask (m_scheduler, 0.5,
+ bind (&SyncCore::localStateChanged, this),
+ LOCAL_STATE_CHANGE_DELAYED_TAG);
+}
+
+void
SyncCore::handleInterest(const Name &name)
{
int size = name.size();
diff --git a/src/sync-core.h b/src/sync-core.h
index b804ac5..c2b427b 100644
--- a/src/sync-core.h
+++ b/src/sync-core.h
@@ -56,6 +56,15 @@
void
localStateChanged ();
+ /**
+ * @brief Schedule an event to update local state with a small delay
+ *
+ * This call is preferred to localStateChanged if many local state updates
+ * are anticipated within a short period of time
+ */
+ void
+ localStateChangedDelayed ();
+
void
updateLocalState (sqlite3_int64);