Merge remote-tracking branch 'git.irl/dispatcher'
Conflicts:
src/sync-core.cc
src/sync-core.h
diff --git a/ccnx/ccnx-closure.cpp b/ccnx/ccnx-closure.cpp
index bc9b244..312af6d 100644
--- a/ccnx/ccnx-closure.cpp
+++ b/ccnx/ccnx-closure.cpp
@@ -1,54 +1,63 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#include "ccnx-closure.h"
namespace Ccnx {
Closure::Closure(const DataCallback &dataCallback, const TimeoutCallback &timeoutCallback)
- : m_timeoutCallback(NULL), m_dataCallback(NULL)
+ : m_timeoutCallback (timeoutCallback)
+ , m_dataCallback (dataCallback)
{
- m_timeoutCallback = new TimeoutCallback (timeoutCallback);
- m_dataCallback = new DataCallback (dataCallback);
-}
-
-Closure::Closure(const Closure &other)
- : m_timeoutCallback(NULL), m_dataCallback(NULL)
-{
- m_timeoutCallback = new TimeoutCallback(*(other.m_timeoutCallback));
- m_dataCallback = new DataCallback(*(other.m_dataCallback));
}
Closure::~Closure ()
{
- delete m_dataCallback;
- delete m_timeoutCallback;
- m_dataCallback = NULL;
- m_timeoutCallback = NULL;
}
Closure::TimeoutCallbackReturnValue
Closure::runTimeoutCallback(const Name &interest)
{
- if ((*m_timeoutCallback).empty())
- {
- return RESULT_OK;
- }
+ if (m_timeoutCallback.empty ())
+ {
+ return RESULT_OK;
+ }
- return (*m_timeoutCallback)(interest);
+ return m_timeoutCallback (interest);
}
void
-Closure::runDataCallback(const Name &name, const Bytes &content)
+Closure::runDataCallback(const Name &name, PcoPtr content)
{
- if (m_dataCallback != NULL) {
- (*m_dataCallback)(name, content);
- }
+ if (!m_dataCallback.empty ())
+ {
+ m_dataCallback (name, content);
+ }
}
Closure *
-Closure::dup() const
+Closure::dup () const
{
- Closure *closure = new Closure(*this);
- return closure;
+ return new Closure (*this);
}
} // Ccnx
diff --git a/ccnx/ccnx-closure.h b/ccnx/ccnx-closure.h
index 1bcaa06..343d8be 100644
--- a/ccnx/ccnx-closure.h
+++ b/ccnx/ccnx-closure.h
@@ -1,17 +1,39 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#ifndef CCNX_CLOSURE_H
#define CCNX_CLOSURE_H
#include "ccnx-common.h"
#include "ccnx-name.h"
-using namespace std;
-
namespace Ccnx {
+class ParsedContentObject;
+typedef boost::shared_ptr<ParsedContentObject> PcoPtr;
+
class Closure
{
public:
- typedef boost::function<void (const Name &, const Bytes &)> DataCallback;
+ typedef boost::function<void (const Name &, PcoPtr pco)> DataCallback;
typedef enum
{
@@ -22,10 +44,11 @@
typedef boost::function<TimeoutCallbackReturnValue (const Name &)> TimeoutCallback;
Closure(const DataCallback &dataCallback, const TimeoutCallback &timeoutCallback = TimeoutCallback());
- Closure(const Closure &other);
virtual ~Closure();
+
virtual void
- runDataCallback(const Name &name, const Bytes &content);
+ runDataCallback(const Name &name, Ccnx::PcoPtr pco);
+
virtual TimeoutCallbackReturnValue
runTimeoutCallback(const Name &interest);
@@ -33,8 +56,8 @@
dup() const;
protected:
- TimeoutCallback *m_timeoutCallback;
- DataCallback *m_dataCallback;
+ TimeoutCallback m_timeoutCallback;
+ DataCallback m_dataCallback;
};
} // Ccnx
diff --git a/ccnx/ccnx-common.h b/ccnx/ccnx-common.h
index 41d7814..985965c 100644
--- a/ccnx/ccnx-common.h
+++ b/ccnx/ccnx-common.h
@@ -86,6 +86,16 @@
return BytesPtr ();
}
+template<class Msg>
+inline BytesPtr
+serializeMsg(const Msg &msg)
+{
+ int size = msg->ByteSize ();
+ BytesPtr bytes (new Bytes (size));
+ msg->SerializeToArray (head(*bytes), size);
+ return bytes;
+}
+
// --- Bytes operations end ---
// Exceptions
diff --git a/ccnx/ccnx-name.cpp b/ccnx/ccnx-name.cpp
index a59cf8a..848019a 100644
--- a/ccnx/ccnx-name.cpp
+++ b/ccnx/ccnx-name.cpp
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#include "ccnx-name.h"
#include <boost/lexical_cast.hpp>
#include <ctype.h>
@@ -155,6 +176,14 @@
}
Name &
+Name::appendComp(const Name &comp)
+{
+ m_comps.insert (m_comps.end (),
+ comp.m_comps.begin (), comp.m_comps.end ());
+ return *this;
+}
+
+Name &
Name::appendComp(const Bytes &comp)
{
m_comps.push_back(comp);
@@ -169,16 +198,6 @@
}
Name &
-Name::appendComp(const Name &name)
-{
- for (vector<Bytes>::const_iterator i = name.m_comps.begin (); i != name.m_comps.end (); i++)
- {
- appendComp (*i);
- }
- return *this;
-}
-
-Name &
Name::appendComp (const void *buf, size_t size)
{
Bytes comp (reinterpret_cast<const unsigned char*> (buf), reinterpret_cast<const unsigned char*> (buf) + size);
@@ -190,7 +209,7 @@
{
Bytes comp;
comp.push_back (0);
-
+
while (number > 0)
{
comp.push_back (static_cast<unsigned char> (number & 0xFF));
diff --git a/ccnx/ccnx-name.h b/ccnx/ccnx-name.h
index a6b6228..92c3054 100644
--- a/ccnx/ccnx-name.h
+++ b/ccnx/ccnx-name.h
@@ -51,7 +51,7 @@
size_t
length () const
{ return m_buf->length; }
-
+
private:
void init(ccn_charbuf *buf);
@@ -80,16 +80,16 @@
operator CcnxCharbufPtr () const { return toCcnxCharbuf (); }
Name &
+ appendComp(const Name &comp);
+
+ Name &
appendComp(const Bytes &comp);
Name &
appendComp(const string &compStr);
Name &
- appendComp(const Name &name);
-
- Name &
- appendComp(const void *buf, size_t size);
+ appendComp(const void *buf, size_t size);
/**
* Append int component
@@ -100,7 +100,14 @@
*/
Name &
appendComp(uint64_t number);
-
+
+ template<class T>
+ Name &
+ operator ()(const T &comp) { return appendComp (comp); }
+
+ Name &
+ operator ()(const void *buf, size_t size) { return appendComp (buf, size); }
+
int
size() const {return m_comps.size();}
diff --git a/ccnx/ccnx-pco.cpp b/ccnx/ccnx-pco.cpp
index 8c2fd3f..2c4e351 100644
--- a/ccnx/ccnx-pco.cpp
+++ b/ccnx/ccnx-pco.cpp
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#include "ccnx-pco.h"
namespace Ccnx {
@@ -5,13 +26,14 @@
void
ParsedContentObject::init(const unsigned char *data, size_t len)
{
+ readRaw(m_bytes, data, len);
+
m_comps = ccn_indexbuf_create();
- int res = ccn_parse_ContentObject(data, len, &m_pco, m_comps);
+ int res = ccn_parse_ContentObject(head (m_bytes), len, &m_pco, m_comps);
if (res < 0)
{
boost::throw_exception(MisformedContentObjectException());
}
- readRaw(m_bytes, data, len);
}
ParsedContentObject::ParsedContentObject(const unsigned char *data, size_t len)
@@ -43,13 +65,13 @@
{
const unsigned char *content;
size_t len;
- Bytes bytes;
int res = ccn_content_get_value(head(m_bytes), m_pco.offset[CCN_PCO_E], &m_pco, &content, &len);
if (res < 0)
{
boost::throw_exception(MisformedContentObjectException());
}
+ Bytes bytes;
readRaw(bytes, content, len);
return bytes;
}
diff --git a/ccnx/ccnx-pco.h b/ccnx/ccnx-pco.h
index 0c52c7a..cecc763 100644
--- a/ccnx/ccnx-pco.h
+++ b/ccnx/ccnx-pco.h
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#ifndef CCNX_CONTENT_OBJECT_H
#define CCNX_CONTENT_OBJECT_H
@@ -5,16 +26,15 @@
#include "ccnx-common.h"
#include "ccnx-name.h"
-using namespace std;
-
namespace Ccnx {
-struct MisformedContentObjectException : virtual boost::exception, virtual exception { };
+struct MisformedContentObjectException : virtual boost::exception, virtual std::exception { };
class ParsedContentObject
{
public:
ParsedContentObject(const unsigned char *data, size_t len);
+ ParsedContentObject(const unsigned char *data, const ccn_parsed_ContentObject &pco);
ParsedContentObject(const Bytes &bytes);
ParsedContentObject(const ParsedContentObject &other);
virtual ~ParsedContentObject();
@@ -24,7 +44,7 @@
BytesPtr
contentPtr() const;
-
+
Name
name() const;
diff --git a/ccnx/ccnx-selectors.h b/ccnx/ccnx-selectors.h
index 1275217..420f33e 100644
--- a/ccnx/ccnx-selectors.h
+++ b/ccnx/ccnx-selectors.h
@@ -62,7 +62,7 @@
answerOriginKind(AOK kind) {m_answerOriginKind = kind; return *this;}
inline Selectors &
- interestLifetime(int lifetime) {m_interestLifetime = lifetime; return *this;}
+ interestLifetime(double lifetime) {m_interestLifetime = lifetime; return *this;}
inline Selectors &
scope(int scope) {m_scope = scope; return *this;}
diff --git a/ccnx/ccnx-wrapper.cpp b/ccnx/ccnx-wrapper.cpp
index b014a07..7f2e567 100644
--- a/ccnx/ccnx-wrapper.cpp
+++ b/ccnx/ccnx-wrapper.cpp
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#include "ccnx-wrapper.h"
extern "C" {
#include <ccn/fetch.h>
@@ -6,6 +27,7 @@
#include <boost/throw_exception.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/random.hpp>
+#include <boost/make_shared.hpp>
#include <sstream>
typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
@@ -265,26 +287,28 @@
return CCN_UPCALL_RESULT_OK;
}
- 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"));
- }
+ PcoPtr pco = make_shared<ParsedContentObject> (info->content_ccnb, info->pco->offset[CCN_PCO_E]);
- Name name(info->content_ccnb, info->content_comps);
+ // 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"));
+ // }
- 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);
+ // Name name(info->content_ccnb, info->content_comps);
- cp->runDataCallback(name, content);
+ // 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);
+
+ cp->runDataCallback (pco->name (), pco);
return CCN_UPCALL_RESULT_OK;
}
-int CcnxWrapper::sendInterest (const Name &interest, const Closure *closure, const Selectors &selectors)
+int CcnxWrapper::sendInterest (const Name &interest, const Closure &closure, const Selectors &selectors)
{
UniqueRecLock lock(m_mutex);
if (!m_running || !m_connected)
@@ -294,7 +318,7 @@
ccn_charbuf *pname = namePtr->getBuf();
ccn_closure *dataClosure = new ccn_closure;
- Closure *myClosure = closure->dup();
+ Closure *myClosure = closure.dup();
dataClosure->data = (void *)myClosure;
dataClosure->p = &incomingData;
diff --git a/ccnx/ccnx-wrapper.h b/ccnx/ccnx-wrapper.h
index baa22b1..cfd030b 100644
--- a/ccnx/ccnx-wrapper.h
+++ b/ccnx/ccnx-wrapper.h
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#ifndef CCNX_WRAPPER_H
#define CCNX_WRAPPER_H
@@ -9,23 +30,15 @@
#include "ccnx-name.h"
#include "ccnx-selectors.h"
#include "ccnx-closure.h"
-
-using namespace std;
+#include "ccnx-pco.h"
namespace Ccnx {
-struct CcnxOperationException : virtual boost::exception, virtual exception { };
+struct CcnxOperationException : virtual boost::exception, virtual std::exception { };
class CcnxWrapper
{
public:
- typedef boost::shared_mutex Lock;
- typedef boost::unique_lock<Lock> WriteLock;
- typedef boost::shared_lock<Lock> ReadLock;
-
- typedef boost::recursive_mutex RecLock;
- typedef boost::unique_lock<RecLock> UniqueRecLock;
-
typedef boost::function<void (const Name &)> InterestCallback;
CcnxWrapper();
@@ -38,13 +51,13 @@
clearInterestFilter (const Name &prefix);
virtual int
- sendInterest (const Name &interest, const Closure *closure, const Selectors &selector = Selectors());
+ sendInterest (const Name &interest, const Closure &closure, const Selectors &selector = Selectors());
virtual int
- publishData (const Name &name, const unsigned char *buf, size_t len, int freshness);
+ publishData (const Name &name, const unsigned char *buf, size_t len, int freshness = 2147/* max value for ccnx*/);
int
- publishData (const Name &name, const Bytes &content, int freshness);
+ publishData (const Name &name, const Bytes &content, int freshness = 2147/* max value for ccnx*/);
static Name
getLocalPrefix ();
@@ -69,6 +82,13 @@
/// @endcond
protected:
+ typedef boost::shared_mutex Lock;
+ typedef boost::unique_lock<Lock> WriteLock;
+ typedef boost::shared_lock<Lock> ReadLock;
+
+ typedef boost::recursive_mutex RecLock;
+ typedef boost::unique_lock<RecLock> UniqueRecLock;
+
ccn* m_handle;
RecLock m_mutex;
boost::thread m_thread;
diff --git a/ccnx/ccnx-tunnel.cpp b/disabled/ccnx-tunnel.cpp
similarity index 65%
rename from ccnx/ccnx-tunnel.cpp
rename to disabled/ccnx-tunnel.cpp
index 0808663..def125a 100644
--- a/ccnx/ccnx-tunnel.cpp
+++ b/disabled/ccnx-tunnel.cpp
@@ -1,3 +1,24 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#include "ccnx-tunnel.h"
#include "ccnx-pco.h"
@@ -5,8 +26,8 @@
{
CcnxTunnel::CcnxTunnel()
- : CcnxWrapper()
- , m_localPrefix("/")
+ : CcnxWrapper()
+ , m_localPrefix("/")
{
refreshLocalPrefix();
}
@@ -28,12 +49,13 @@
}
int
-CcnxTunnel::sendInterest (const Name &interest, const Closure *closure, const Selectors &selectors)
+CcnxTunnel::sendInterest (const Name &interest, const Closure &closure, const Selectors &selectors)
{
Name tunneledInterest = queryRoutableName(interest);
- Closure *cp = new TunnelClosure(closure, this, interest);
- CcnxWrapper::sendInterest(tunneledInterest, cp, selectors);
- delete cp;
+
+ CcnxWrapper::sendInterest(tunneledInterest,
+ TunnelClosure(closure, *this, interest),
+ selectors);
return 0;
}
@@ -118,33 +140,30 @@
m_rit.erase(prefix);
}
-TunnelClosure::TunnelClosure(const DataCallback &dataCallback, CcnxTunnel *tunnel, const Name &originalInterest, const TimeoutCallback &timeoutCallback)
- : Closure(dataCallback, timeoutCallback)
- , m_tunnel(tunnel)
- , m_originalInterest(originalInterest)
+TunnelClosure::TunnelClosure(const DataCallback &dataCallback, CcnxTunnel &tunnel,
+ const Name &originalInterest, const TimeoutCallback &timeoutCallback)
+ : Closure(dataCallback, timeoutCallback)
+ , m_tunnel(tunnel)
+ , m_originalInterest(originalInterest)
{
}
-TunnelClosure::TunnelClosure(const Closure *closure, CcnxTunnel *tunnel, const Name &originalInterest)
- : Closure(*closure)
- , m_tunnel(tunnel)
+TunnelClosure::TunnelClosure(const Closure &closure, CcnxTunnel &tunnel, const Name &originalInterest)
+ : Closure(closure)
+ , m_tunnel(tunnel)
{
}
Closure *
TunnelClosure::dup() const
{
- Closure *closure = new TunnelClosure(*m_dataCallback, m_tunnel, m_originalInterest, *m_timeoutCallback);
- return closure;
+ return new TunnelClosure (*this);
}
void
TunnelClosure::runDataCallback(const Name &name, const Bytes &content)
{
- if (m_tunnel != NULL)
- {
- m_tunnel->handleTunneledData(name, content, (*m_dataCallback));
- }
+ m_tunnel.handleTunneledData(name, content, m_dataCallback);
}
Closure::TimeoutCallbackReturnValue
diff --git a/ccnx/ccnx-tunnel.h b/disabled/ccnx-tunnel.h
similarity index 63%
rename from ccnx/ccnx-tunnel.h
rename to disabled/ccnx-tunnel.h
index 05f981a..01dbf5e 100644
--- a/ccnx/ccnx-tunnel.h
+++ b/disabled/ccnx-tunnel.h
@@ -1,12 +1,28 @@
+/* -*- 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
+ *
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ */
+
#ifndef CCNX_TUNNEL_H
#define CCNX_TUNNEL_H
-#include <boost/variant.hpp>
-
-#include "ccnx-common.h"
#include "ccnx-wrapper.h"
-#include "ccnx-name.h"
-#include "ccnx-selectors.h"
#define _OVERRIDE
#ifdef __GNUC__
@@ -16,8 +32,6 @@
#endif // __GNUC__ version
#endif // __GNUC__
-using namespace std;
-
// Eventually, Sync::CcnxWrapper should be moved to this namespace.
// It has nothing to do with Sync.
namespace Ccnx
@@ -26,8 +40,8 @@
class CcnxTunnel : public CcnxWrapper
{
public:
- typedef multimap<Name, InterestCallback> RegisteredInterestTable;
- typedef multimap<Name, InterestCallback>::iterator RitIter;
+ typedef std::multimap<Name, InterestCallback> RegisteredInterestTable;
+ typedef std::multimap<Name, InterestCallback>::iterator RitIter;
CcnxTunnel();
@@ -41,7 +55,7 @@
publishContentObject(const Name &name, const Bytes &contentObject, int freshness);
virtual int
- sendInterest (const Name &interest, const Closure *closure, const Selectors &selectors = Selectors());
+ sendInterest (const Name &interest, const Closure &closure, const Selectors &selectors = Selectors());
// prefix is topology-independent
@@ -89,9 +103,10 @@
class TunnelClosure : public Closure
{
public:
- TunnelClosure(const DataCallback &dataCallback, CcnxTunnel *tunnel, const Name &originalInterest, const TimeoutCallback &timeoutCallback = TimeoutCallback());
+ TunnelClosure(const DataCallback &dataCallback, CcnxTunnel &tunnel,
+ const Name &originalInterest, const TimeoutCallback &timeoutCallback = TimeoutCallback());
- TunnelClosure(const Closure *closure, CcnxTunnel *tunnel, const Name &originalInterest);
+ TunnelClosure(const Closure &closure, CcnxTunnel &tunnel, const Name &originalInterest);
virtual void
runDataCallback(const Name &name, const Bytes &content) _OVERRIDE;
@@ -100,10 +115,10 @@
runTimeoutCallback(const Name &interest) _OVERRIDE;
virtual Closure *
- dup() const _OVERRIDE;
+ dup() const;
private:
- CcnxTunnel *m_tunnel;
+ CcnxTunnel &m_tunnel;
Name m_originalInterest;
};
diff --git a/test/test-ccnx-tunnel.cc b/disabled/test-ccnx-tunnel.cc
similarity index 85%
rename from test/test-ccnx-tunnel.cc
rename to disabled/test-ccnx-tunnel.cc
index 6c2ab0d..724d6e5 100644
--- a/test/test-ccnx-tunnel.cc
+++ b/disabled/test-ccnx-tunnel.cc
@@ -82,21 +82,16 @@
g_dc_o = 0;
t1->publishData(Name(inner), (const unsigned char *)inner.c_str(), inner.size(), 5);
usleep(100000);
- Closure *outerClosure = new Closure(bind(outerCallback, _1, _2));
- c1->sendInterest(Name("/local/hello"), outerClosure);
+
+ c1->sendInterest(Name("/local/hello"), Closure(bind(outerCallback, _1, _2)));
usleep(100000);
// it is indeed published as /local/hello
BOOST_CHECK_EQUAL(g_dc_o, 1);
g_dc_i = 0;
- Closure *innerClosure = new Closure(bind(innerCallback, _1, _2));
- t2->sendInterest(Name(inner), innerClosure);
+ t2->sendInterest(Name(inner), Closure(bind(innerCallback, _1, _2)));
usleep(100000);
BOOST_CHECK_EQUAL(g_dc_i, 1);
-
- delete outerClosure;
- delete innerClosure;
-
}
BOOST_AUTO_TEST_CASE (CcnxTunnelRegister)
@@ -107,7 +102,7 @@
t3.overridePrefix();
t3.setInterestFilter(Name("/t3"), bind(interestCallback, _1));
usleep(100000);
- Closure *innerClosure = new Closure(bind(innerCallback, _1, _2));
+ Closure innerClosure (bind(innerCallback, _1, _2));
t1->sendInterest(Name("/t3/hello"), innerClosure);
usleep(100000);
BOOST_CHECK_EQUAL(g_dc_i, 1);
@@ -119,7 +114,6 @@
usleep(100000);
BOOST_CHECK_EQUAL(g_dc_i, 1);
BOOST_CHECK_EQUAL(g_ic, 1);
- delete innerClosure;
}
diff --git a/gui/chronosharegui.cpp b/gui/chronosharegui.cpp
index f252370..5b49909 100644
--- a/gui/chronosharegui.cpp
+++ b/gui/chronosharegui.cpp
@@ -22,7 +22,7 @@
ChronoShareGui::ChronoShareGui(QWidget *parent) :
QWidget(parent),
- m_settingsFilePath(QDir::homePath() + ".cronoShare")
+ m_settingsFilePath(QDir::homePath() + "/.cronoshare")
{
// load settings
if(!loadSettings())
@@ -95,8 +95,7 @@
// create the "quit program" action
m_quitProgram = new QAction(tr("&Quit"), this);
- connect(m_quitProgram, SIGNAL(triggered()), this, SLOT(quit()));
-
+ connect(m_quitProgram, SIGNAL(triggered()), qApp, SLOT(quit()));
}
void ChronoShareGui::createTrayIcon()
@@ -191,6 +190,9 @@
// Load Settings
QSettings settings(m_settingsFilePath, QSettings::NativeFormat);
+
+ qDebug() << settings.allKeys();
+
if(settings.contains("dirPath"))
{
m_dirPath = settings.value("dirPath", QDir::homePath()).toString();
@@ -201,12 +203,17 @@
m_dirPath = QDir::homePath();
successful = false;
}
+
+ qDebug() << "success: " << successful;
return successful;
}
void ChronoShareGui::saveSettings()
{
+ qDebug() << m_settingsFilePath;
+ qDebug() << m_dirPath;
+
// Save Settings
QSettings settings(m_settingsFilePath, QSettings::NativeFormat);
settings.setValue("dirPath", m_dirPath);
@@ -221,5 +228,4 @@
#if WAF
#include "chronosharegui.moc"
#include "chronosharegui.cpp.moc"
-#endif
-
+#endif
\ No newline at end of file
diff --git a/gui/chronosharegui.h b/gui/chronosharegui.h
index 48e2bc3..76b73ba 100644
--- a/gui/chronosharegui.h
+++ b/gui/chronosharegui.h
@@ -31,6 +31,7 @@
#include <QFileDialog>
#include <QCloseEvent>
#include <QMessageBox>
+#include <QApplication>
class ChronoShareGui : public QWidget
{
diff --git a/log4cxx.properties b/log4cxx.properties
new file mode 100644
index 0000000..89f14d6
--- /dev/null
+++ b/log4cxx.properties
@@ -0,0 +1,22 @@
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=ERROR, A1
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+# A1 uses PatternLayout.
+log4j.appender.A1.layout=org.apache.log4j.PatternLayout
+log4j.appender.A1.target=System.err
+#log4j.appender.A1.layout.ConversionPattern=%d{dd-MMM HH:MM:SS,SSS} %p %c %m%n
+#log4j.appender.A1.layout.ConversionPattern=%d{hh:mm:ss,SSS} %-14t %-14c %m%n
+log4j.appender.A1.layout.ConversionPattern=%d{ss,SSS} %-5p %-12c %m%n
+
+log4j.logger.Sync = DEBUG
+log4j.logger.Sync.Log = ERROR
+#log4j.logger.SyncInterestTable = TRACE
+#log4j.logger.AppDataFetch = TRACE
+log4j.logger.Test = TRACE
+#log4j.logger.bgpparser=TRACE
+#log4j.logger.bgpparser.AttributeType=ERROR
+#log4j.logger.bgpparser.MRTCommonHeader=ERROR
+
diff --git a/scheduler/scheduler.cc b/scheduler/scheduler.cc
index 2806075..ee6a124 100644
--- a/scheduler/scheduler.cc
+++ b/scheduler/scheduler.cc
@@ -20,9 +20,14 @@
*/
#include "scheduler.h"
+#include "one-time-task.h"
+#include "periodic-task.h"
+
#include <utility>
+#include <boost/make_shared.hpp>
using namespace std;
+using namespace boost;
#define EVLOOP_NO_EXIT_ON_EMPTY 0x04
@@ -91,24 +96,45 @@
m_thread.join();
}
-bool
-Scheduler::addTask(const TaskPtr &task)
+TaskPtr
+Scheduler::scheduleOneTimeTask (SchedulerPtr scheduler, double delay,
+ const Task::Callback &callback, const Task::Tag &tag)
{
- TaskPtr newTask = task;
+ TaskPtr task = make_shared<OneTimeTask> (callback, tag, scheduler, delay);
+ if (scheduler->addTask (task))
+ return task;
+ else
+ return TaskPtr ();
+}
+TaskPtr
+Scheduler::schedulePeriodicTask (SchedulerPtr scheduler, IntervalGeneratorPtr delayGenerator,
+ const Task::Callback &callback, const Task::Tag &tag)
+{
+ TaskPtr task = make_shared<PeriodicTask> (callback, tag, scheduler, delayGenerator);
+
+ if (scheduler->addTask (task))
+ return task;
+ else
+ return TaskPtr ();
+}
+
+bool
+Scheduler::addTask(TaskPtr newTask)
+{
if (addToMap(newTask))
{
newTask->reset();
int res = evtimer_add(newTask->ev(), newTask->tv());
if (res < 0)
{
- cout << "evtimer_add failed for " << task->tag() << endl;
+ cout << "evtimer_add failed for " << newTask->tag() << endl;
}
return true;
}
else
{
- cout << "fail to add task: " << task->tag() << endl;
+ cout << "fail to add task: " << newTask->tag() << endl;
}
return false;
diff --git a/scheduler/scheduler.h b/scheduler/scheduler.h
index 83b9926..e783cf0 100644
--- a/scheduler/scheduler.h
+++ b/scheduler/scheduler.h
@@ -38,6 +38,10 @@
#include <sys/time.h>
#include "task.h"
+#include "interval-generator.h"
+
+class Scheduler;
+typedef boost::shared_ptr<Scheduler> SchedulerPtr;
/**
* @brief Scheduler class
@@ -56,9 +60,18 @@
virtual void
shutdown();
+ // helper method to schedule one-time task
+ static TaskPtr
+ scheduleOneTimeTask (SchedulerPtr scheduler, double delay, const Task::Callback &callback, const Task::Tag &tag);
+
+ // helper method to schedule periodic task
+ static TaskPtr
+ schedulePeriodicTask (SchedulerPtr scheduler, IntervalGeneratorPtr delayGenerator,
+ const Task::Callback &callback, const Task::Tag &tag);
+
// if task with the same tag exists, the task is not added and return false
virtual bool
- addTask(const TaskPtr &task);
+ addTask(TaskPtr task);
// delete task by task->tag, regardless of whether it's invoked or not
virtual void
@@ -117,9 +130,6 @@
boost::thread m_thread;
};
-class Scheduler;
-typedef boost::shared_ptr<Scheduler> SchedulerPtr;
-
struct SchedulerException : virtual boost::exception, virtual std::exception { };
#endif // SCHEDULER_H
diff --git a/src/action-log.cc b/src/action-log.cc
index 7c5934e..5c4f592 100644
--- a/src/action-log.cc
+++ b/src/action-log.cc
@@ -20,17 +20,104 @@
*/
#include "action-log.h"
+#include "logging.h"
using namespace boost;
using namespace std;
using namespace Ccnx;
+INIT_LOGGER ("ActionLog");
+
+const std::string INIT_DATABASE = "\
+CREATE TABLE ActionLog ( \n\
+ device_id INTEGER NOT NULL, \n\
+ seq_no INTEGER NOT NULL, \n\
+ \n\
+ action CHAR(1) NOT NULL, /* 0 for \"update\", 1 for \"delete\". */ \n\
+ filename TEXT NOT NULL, \n\
+ \n\
+ version INTEGER NOT NULL, \n\
+ action_timestamp TIMESTAMP NOT NULL, \n\
+ \n\
+ file_hash BLOB, /* NULL if action is \"delete\" */ \n\
+ file_atime TIMESTAMP, \n\
+ file_mtime TIMESTAMP, \n\
+ file_ctime TIMESTAMP, \n\
+ file_chmod INTEGER, \n\
+ file_seg_num INTEGER, /* NULL if action is \"delete\" */ \n\
+ \n\
+ parent_device_id INTEGER, \n\
+ parent_seq_no INTEGER, \n\
+ \n\
+ action_name TEXT, \n\
+ action_content_object BLOB, \n\
+ \n\
+ PRIMARY KEY (device_id, seq_no), \n\
+ \n\
+ FOREIGN KEY (parent_device_id, parent_seq_no) \n\
+ REFERENCES ActionLog (device_id, seq_no) \n\
+ ON UPDATE RESTRICT \n\
+ ON DELETE SET NULL \n\
+); \n\
+ \n\
+CREATE INDEX ActionLog_filename_version ON ActionLog (filename,version); \n\
+CREATE INDEX ActionLog_parent ON ActionLog (parent_device_id, parent_seq_no); \n\
+CREATE INDEX ActionLog_action_name ON ActionLog (action_name); \n\
+ \n\
+CREATE TRIGGER ActionLogInsert_trigger \n\
+ AFTER INSERT ON ActionLog \n\
+ FOR EACH ROW \n\
+ WHEN (SELECT device_id \n\
+ FROM ActionLog \n\
+ WHERE filename=NEW.filename AND \n\
+ version > NEW.version) IS NULL AND \n\
+ (SELECT a.device_id \n\
+ FROM ActionLog a \n\
+ LEFT JOIN SyncNodes s ON s.device_id=a.device_id \n\
+ WHERE filename=NEW.filename AND \n\
+ version = NEW.version AND \n\
+ a.device_id != NEW.device_id AND \n\
+ s.device_name > (SELECT device_name \n\
+ FROM SyncNodes \n\
+ WHERE device_id=NEW.device_id)) IS NULL \n\
+ BEGIN \n\
+ SELECT apply_action ((SELECT device_name FROM SyncNodes where device_id=NEW.device_id), \
+ NEW.device_id, NEW.seq_no, \
+ NEW.action,NEW.filename,NEW.file_hash, \
+ strftime('%s', NEW.file_atime),strftime('%s', NEW.file_mtime),strftime('%s', NEW.file_ctime), \
+ NEW.file_chmod, NEW.file_seg_num); /* function that applies action and adds record the FileState */ \n \
+ END; \n\
+ \n\
+CREATE TABLE FileState ( \n\
+ type INTEGER NOT NULL, /* 0 - newest, 1 - oldest */ \n\
+ filename TEXT NOT NULL, \n\
+ device_id INTEGER NOT NULL, \n\
+ seq_no INTEGER NOT NULL, \n\
+ file_hash BLOB, /* NULL if action is \"delete\" */ \n\
+ file_atime TIMESTAMP, \n\
+ file_mtime TIMESTAMP, \n\
+ file_ctime TIMESTAMP, \n\
+ file_chmod INTEGER, \n\
+ file_seg_num INTEGER, \n\
+ \n\
+ PRIMARY KEY (type, filename) \n\
+); \n\
+ \n\
+CREATE INDEX FileState_device_id_seq_no ON FileState (device_id, seq_no); \n\
+";
+
+
ActionLog::ActionLog (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &path,
+ SyncLogPtr syncLog,
const std::string &localName, const std::string &sharedFolder)
- : SyncLog (path, localName)
+ : DbHelper (path)
+ , m_syncLog (syncLog)
, m_ccnx (ccnx)
, m_sharedFolderName (sharedFolder)
{
+ sqlite3_exec (m_db, INIT_DATABASE.c_str (), NULL, NULL, NULL);
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, sqlite3_errmsg (m_db));
+
int res = sqlite3_create_function (m_db, "apply_action", -1, SQLITE_ANY, reinterpret_cast<void*> (this),
ActionLog::apply_action_xFun,
0, 0);
@@ -61,9 +148,9 @@
sqlite3_int64 parent_device_id = -1;
sqlite3_int64 parent_seq_no = -1;
string parent_device_name;
-
+
sqlite3_bind_text (stmt, 1, filename.c_str (), filename.size (), SQLITE_STATIC);
- if (sqlite3_step (stmt) == SQLITE_ROW)
+ if (sqlite3_step (stmt) == SQLITE_ROW)
{
version = sqlite3_column_int64 (stmt, 0) + 1;
@@ -74,7 +161,7 @@
parent_device_name = string(reinterpret_cast<const char*> (sqlite3_column_blob (stmt, 4)), sqlite3_column_bytes (stmt, 4));
}
}
-
+
sqlite3_finalize (stmt);
return make_tuple (version, parent_device_id, parent_seq_no, parent_device_name);
}
@@ -88,17 +175,17 @@
int seg_num)
{
sqlite3_exec (m_db, "BEGIN TRANSACTION;", 0,0,0);
-
- sqlite3_int64 seq_no = GetNextLocalSeqNo ();
+
+ sqlite3_int64 seq_no = m_syncLog->GetNextLocalSeqNo ();
sqlite3_int64 version = 0;
sqlite3_int64 parent_device_id = -1;
string parent_device_name;
sqlite3_int64 parent_seq_no = -1;
sqlite3_int64 action_time = time (0);
-
+
tie (version, parent_device_id, parent_seq_no, parent_device_name) = GetExistingRecord (filename);
-
+
sqlite3_stmt *stmt;
int res = sqlite3_prepare_v2 (m_db, "INSERT INTO ActionLog "
"(device_id, seq_no, action, filename, version, action_timestamp, "
@@ -117,7 +204,7 @@
// "VALUES (?, ?, ?, ?, ?, datetime(?, 'unixepoch'),"
// " ?, datetime(?, 'unixepoch'), datetime(?, 'unixepoch'), datetime(?, 'unixepoch'), ?,"
// " ?, ?)" << endl;
-
+
if (res != SQLITE_OK)
{
BOOST_THROW_EXCEPTION (Error::Db ()
@@ -126,16 +213,16 @@
// << errmsg_info_str ("Some error with prepare AddActionUpdate"));
}
-
- sqlite3_bind_int64 (stmt, 1, m_localDeviceId);
+
+ sqlite3_bind_int64 (stmt, 1, m_syncLog->GetLocalSyncNodeId ());
sqlite3_bind_int64 (stmt, 2, seq_no);
sqlite3_bind_int (stmt, 3, 0);
sqlite3_bind_text (stmt, 4, filename.c_str (), filename.size (), SQLITE_TRANSIENT);
sqlite3_bind_int64 (stmt, 5, version);
sqlite3_bind_int64 (stmt, 6, action_time);
-
+
sqlite3_bind_blob (stmt, 7, hash.GetHash (), hash.GetHashBytes (), SQLITE_TRANSIENT);
-
+
// sqlite3_bind_int64 (stmt, 8, atime); // NULL
sqlite3_bind_int64 (stmt, 9, wtime);
// sqlite3_bind_int64 (stmt, 10, ctime); // NULL
@@ -152,7 +239,7 @@
sqlite3_bind_null (stmt, 13);
sqlite3_bind_null (stmt, 14);
}
-
+
// missing part: creating ContentObject for the action !!!
ActionItem item;
@@ -171,7 +258,7 @@
{
cout << Name (reinterpret_cast<const unsigned char *> (parent_device_name.c_str ()),
parent_device_name.size ()) << endl;
-
+
item.set_parent_device_name (parent_device_name);
item.set_parent_seq_no (parent_seq_no);
}
@@ -180,22 +267,18 @@
string item_msg;
item.SerializeToString (&item_msg);
- Name actionName (m_localName);
- actionName
- .appendComp ("action")
- .appendComp (m_sharedFolderName)
- .appendComp (seq_no);
-
+ Name actionName = Name (m_syncLog->GetLocalName ())("action")(m_sharedFolderName)(seq_no);
+
Bytes actionData = m_ccnx->createContentObject (actionName, item_msg.c_str (), item_msg.size ());
CcnxCharbufPtr namePtr = actionName.toCcnxCharbuf ();
-
+
sqlite3_bind_blob (stmt, 14, namePtr->buf (), namePtr->length (), SQLITE_TRANSIENT);
sqlite3_bind_blob (stmt, 15, &actionData[0], actionData.size (), SQLITE_TRANSIENT);
-
+
sqlite3_step (stmt);
- sqlite3_finalize (stmt);
-
+ sqlite3_finalize (stmt);
+
sqlite3_exec (m_db, "END TRANSACTION;", 0,0,0);
}
@@ -211,22 +294,22 @@
ActionLog::AddActionDelete (const std::string &filename)
{
sqlite3_exec (m_db, "BEGIN TRANSACTION;", 0,0,0);
-
+
sqlite3_int64 version = 0;
sqlite3_int64 parent_device_id = -1;
string parent_device_name;
sqlite3_int64 parent_seq_no = -1;
sqlite3_int64 action_time = time (0);
-
+
tie (version, parent_device_id, parent_seq_no, parent_device_name) = GetExistingRecord (filename);
if (parent_device_id < 0) // no records exist or file was already deleted
{
- sqlite3_exec (m_db, "END TRANSACTION;", 0,0,0);
+ sqlite3_exec (m_db, "END TRANSACTION;", 0,0,0);
return;
}
- sqlite3_int64 seq_no = GetNextLocalSeqNo ();
+ sqlite3_int64 seq_no = m_syncLog->GetNextLocalSeqNo ();
sqlite3_stmt *stmt;
sqlite3_prepare_v2 (m_db, "INSERT INTO ActionLog "
@@ -237,7 +320,7 @@
" ?, ?,"
" ?, ?)", -1, &stmt, 0);
- sqlite3_bind_int64 (stmt, 1, m_localDeviceId);
+ sqlite3_bind_int64 (stmt, 1, m_syncLog->GetLocalSyncNodeId ());
sqlite3_bind_int64 (stmt, 2, seq_no);
sqlite3_bind_int (stmt, 3, 1);
sqlite3_bind_text (stmt, 4, filename.c_str (), filename.size (), SQLITE_TRANSIENT);
@@ -258,28 +341,24 @@
string item_msg;
item.SerializeToString (&item_msg);
- Name actionName (m_localName);
- actionName
- .appendComp ("action")
- .appendComp (m_sharedFolderName)
- .appendComp (seq_no);
-
+ Name actionName = Name (m_syncLog->GetLocalName ())("action")(m_sharedFolderName)(seq_no);
+
Bytes actionData = m_ccnx->createContentObject (actionName, item_msg.c_str (), item_msg.size ());
CcnxCharbufPtr namePtr = actionName.toCcnxCharbuf ();
-
+
sqlite3_bind_blob (stmt, 9, namePtr->buf (), namePtr->length (), SQLITE_TRANSIENT);
sqlite3_bind_blob (stmt, 10, &actionData[0], actionData.size (), SQLITE_TRANSIENT);
-
+
sqlite3_step (stmt);
-
+
// cout << Ccnx::Name (reinterpret_cast<const unsigned char *> (parent_device_name.c_str ()),
// parent_device_name.size ()) << endl;
-
+
// assign name to the action, serialize action, and create content object
-
- sqlite3_finalize (stmt);
-
- sqlite3_exec (m_db, "END TRANSACTION;", 0,0,0);
+
+ sqlite3_finalize (stmt);
+
+ sqlite3_exec (m_db, "END TRANSACTION;", 0,0,0);
}
@@ -299,7 +378,7 @@
// cout << "device_name: " << sqlite3_value_text (argv[0]) << endl;
// cout << "action: " << sqlite3_value_int (argv[1]) << endl;
// cout << "filename: " << sqlite3_value_text (argv[2]) << endl;
-
+
string device_name (reinterpret_cast<const char*> (sqlite3_value_blob (argv[0])), sqlite3_value_bytes (argv[0]));
sqlite3_int64 device_id = sqlite3_value_int64 (argv[1]);
sqlite3_int64 seq_no = sqlite3_value_int64 (argv[2]);
@@ -338,11 +417,11 @@
sqlite3_bind_int (stmt, 7, mode);
sqlite3_bind_int (stmt, 8, seg_num);
sqlite3_bind_text (stmt, 9, filename.c_str (), -1, SQLITE_TRANSIENT);
-
+
sqlite3_step (stmt);
// cout << sqlite3_errmsg (the->m_db) << endl;
-
+
sqlite3_finalize (stmt);
int affected_rows = sqlite3_changes (the->m_db);
@@ -362,7 +441,7 @@
sqlite3_bind_int64 (stmt, 6, mtime);
sqlite3_bind_int64 (stmt, 7, ctime);
sqlite3_bind_int (stmt, 8, mode);
-
+
sqlite3_step (stmt);
// cout << sqlite3_errmsg (the->m_db) << endl;
sqlite3_finalize (stmt);
@@ -375,11 +454,11 @@
sqlite3_bind_text (stmt, 1, filename.c_str (), -1, SQLITE_STATIC);
cout << "Delete " << filename << endl;
-
+
sqlite3_step (stmt);
sqlite3_finalize (stmt);
}
-
+
sqlite3_result_null (context);
}
diff --git a/src/action-log.h b/src/action-log.h
index 7a154c1..1654215 100644
--- a/src/action-log.h
+++ b/src/action-log.h
@@ -22,7 +22,9 @@
#ifndef ACTION_LOG_H
#define ACTION_LOG_H
+#include "db-helper.h"
#include "sync-log.h"
+
#include <boost/tuple/tuple.hpp>
#include <action-item.pb.h>
#include <ccnx-wrapper.h>
@@ -30,10 +32,11 @@
class ActionLog;
typedef boost::shared_ptr<ActionLog> ActionLogPtr;
-class ActionLog : public SyncLog
+class ActionLog : public DbHelper
{
public:
ActionLog (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &path,
+ SyncLogPtr syncLog,
const std::string &localName, const std::string &sharedFolder);
void
@@ -45,7 +48,7 @@
void
AddActionMove (const std::string &oldFile, const std::string &newFile);
-
+
void
AddActionDelete (const std::string &filename);
@@ -58,8 +61,10 @@
static void
apply_action_xFun (sqlite3_context *context, int argc, sqlite3_value **argv);
-
+
private:
+ SyncLogPtr m_syncLog;
+
Ccnx::CcnxWrapperPtr m_ccnx;
Ccnx::Name m_sharedFolderName;
};
diff --git a/src/chronoshare-client.ice b/src/chronoshare-client.ice
deleted file mode 100644
index c2e056f..0000000
--- a/src/chronoshare-client.ice
+++ /dev/null
@@ -1,34 +0,0 @@
-/* -*- 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: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
- * Zhenkai Zhu <zhenkai@cs.ucla.edu>
- */
-
-module ChronoshareClient
-{
- sequence<byte> HashBytes;
-
- interface Notify
- {
- void updateFile (string filename, ["cpp:array"] HashBytes fileHash, long wtime, int mode);
-
- void moveFile (string origFilename, string newFilename);
-
- void deleteFile (string filename);
- };
-};
diff --git a/src/db-helper.cc b/src/db-helper.cc
index 82feb89..95ccc7a 100644
--- a/src/db-helper.cc
+++ b/src/db-helper.cc
@@ -20,163 +20,33 @@
*/
#include "db-helper.h"
-// #include "sync-log.h"
+#include "logging.h"
+
#include <boost/make_shared.hpp>
#include <boost/ref.hpp>
#include <boost/throw_exception.hpp>
+INIT_LOGGER ("DbHelper");
+
using namespace boost;
namespace fs = boost::filesystem;
const std::string INIT_DATABASE = "\
PRAGMA foreign_keys = ON; \n\
- \n\
-CREATE TABLE \n\
- SyncNodes( \n\
- device_id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
- device_name BLOB NOT NULL, \n\
- description TEXT, \n\
- seq_no INTEGER NOT NULL, \n\
- last_known_locator BLOB, \n\
- last_update TIMESTAMP \n\
- ); \n\
- \n\
-CREATE TRIGGER SyncNodesUpdater_trigger \n\
- BEFORE INSERT ON SyncNodes \n\
- FOR EACH ROW \n\
- WHEN (SELECT device_id \n\
- FROM SyncNodes \n\
- WHERE device_name=NEW.device_name) \n\
- IS NOT NULL \n\
- BEGIN \n\
- UPDATE SyncNodes \n\
- SET seq_no=max(seq_no,NEW.seq_no) \n\
- WHERE device_name=NEW.device_name; \n\
- SELECT RAISE(IGNORE); \n\
- END; \n\
- \n\
-CREATE INDEX SyncNodes_device_name ON SyncNodes (device_name); \n\
- \n\
-CREATE TABLE SyncLog( \n\
- state_id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
- state_hash BLOB NOT NULL UNIQUE, \n\
- last_update TIMESTAMP NOT NULL \n\
- ); \n\
- \n\
-CREATE TABLE \n\
- SyncStateNodes( \n\
- id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
- state_id INTEGER NOT NULL \n\
- REFERENCES SyncLog (state_id) ON UPDATE CASCADE ON DELETE CASCADE, \n\
- device_id INTEGER NOT NULL \n\
- REFERENCES SyncNodes (device_id) ON UPDATE CASCADE ON DELETE CASCADE, \n\
- seq_no INTEGER NOT NULL \n\
- ); \n\
- \n\
-CREATE INDEX SyncStateNodes_device_id ON SyncStateNodes (device_id); \n\
-CREATE INDEX SyncStateNodes_state_id ON SyncStateNodes (state_id); \n\
-CREATE INDEX SyncStateNodes_seq_no ON SyncStateNodes (seq_no); \n\
- \n\
-CREATE TRIGGER SyncLogGuard_trigger \n\
- BEFORE INSERT ON SyncLog \n\
- FOR EACH ROW \n\
- WHEN (SELECT state_hash \n\
- FROM SyncLog \n\
- WHERE state_hash=NEW.state_hash) \n\
- IS NOT NULL \n\
- BEGIN \n\
- DELETE FROM SyncLog WHERE state_hash=NEW.state_hash; \n\
- END; \n\
- \n\
-CREATE TABLE ActionLog ( \n\
- device_id INTEGER NOT NULL, \n\
- seq_no INTEGER NOT NULL, \n\
- \n\
- action CHAR(1) NOT NULL, /* 0 for \"update\", 1 for \"delete\". */ \n\
- filename TEXT NOT NULL, \n\
- \n\
- version INTEGER NOT NULL, \n\
- action_timestamp TIMESTAMP NOT NULL, \n\
- \n\
- file_hash BLOB, /* NULL if action is \"delete\" */ \n\
- file_atime TIMESTAMP, \n\
- file_mtime TIMESTAMP, \n\
- file_ctime TIMESTAMP, \n\
- file_chmod INTEGER, \n\
- file_seg_num INTEGER, /* NULL if action is \"delete\" */ \n\
- \n\
- parent_device_id INTEGER, \n\
- parent_seq_no INTEGER, \n\
- \n\
- action_name TEXT, \n\
- action_content_object BLOB, \n\
- \n\
- PRIMARY KEY (device_id, seq_no), \n\
- \n\
- FOREIGN KEY (parent_device_id, parent_seq_no) \n\
- REFERENCES ActionLog (device_id, seq_no) \n\
- ON UPDATE RESTRICT \n\
- ON DELETE SET NULL \n\
-); \n\
- \n\
-CREATE INDEX ActionLog_filename_version ON ActionLog (filename,version); \n\
-CREATE INDEX ActionLog_parent ON ActionLog (parent_device_id, parent_seq_no); \n\
-CREATE INDEX ActionLog_action_name ON ActionLog (action_name); \n\
- \n\
-CREATE TRIGGER ActionLogInsert_trigger \n\
- AFTER INSERT ON ActionLog \n\
- FOR EACH ROW \n\
- WHEN (SELECT device_id \n\
- FROM ActionLog \n\
- WHERE filename=NEW.filename AND \n\
- version > NEW.version) IS NULL AND \n\
- (SELECT a.device_id \n\
- FROM ActionLog a \n\
- LEFT JOIN SyncNodes s ON s.device_id=a.device_id \n\
- WHERE filename=NEW.filename AND \n\
- version = NEW.version AND \n\
- a.device_id != NEW.device_id AND \n\
- s.device_name > (SELECT device_name \n\
- FROM SyncNodes \n\
- WHERE device_id=NEW.device_id)) IS NULL \n\
- BEGIN \n\
- SELECT apply_action ((SELECT device_name FROM SyncNodes where device_id=NEW.device_id), \
- NEW.device_id, NEW.seq_no, \
- NEW.action,NEW.filename,NEW.file_hash, \
- strftime('%s', NEW.file_atime),strftime('%s', NEW.file_mtime),strftime('%s', NEW.file_ctime), \
- NEW.file_chmod, NEW.file_seg_num); /* function that applies action and adds record the FileState */ \n \
- END; \n\
- \n\
-CREATE TABLE FileState ( \n\
- type INTEGER NOT NULL, /* 0 - newest, 1 - oldest */ \n\
- filename TEXT NOT NULL, \n\
- device_id INTEGER NOT NULL, \n\
- seq_no INTEGER NOT NULL, \n\
- file_hash BLOB, /* NULL if action is \"delete\" */ \n\
- file_atime TIMESTAMP, \n\
- file_mtime TIMESTAMP, \n\
- file_ctime TIMESTAMP, \n\
- file_chmod INTEGER, \n\
- file_seg_num INTEGER, \n\
- \n\
- PRIMARY KEY (type, filename) \n\
-); \n\
- \n\
-CREATE INDEX FileState_device_id_seq_no ON FileState (device_id, seq_no); \n\
";
DbHelper::DbHelper (const fs::path &path)
{
fs::path chronoshareDirectory = path / ".chronoshare";
fs::create_directories (chronoshareDirectory);
-
+
int res = sqlite3_open((chronoshareDirectory / "state.db").c_str (), &m_db);
if (res != SQLITE_OK)
{
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str ("Cannot open/create dabatabase: [" + (chronoshareDirectory / "state.db").string () + "]"));
}
-
+
res = sqlite3_create_function (m_db, "hash", 2, SQLITE_ANY, 0, 0,
DbHelper::hash_xStep, DbHelper::hash_xFinal);
if (res != SQLITE_OK)
@@ -187,14 +57,8 @@
// Alex: determine if tables initialized. if not, initialize... not sure what is the best way to go...
// for now, just attempt to create everything
-
- char *errmsg = 0;
- res = sqlite3_exec (m_db, INIT_DATABASE.c_str (), NULL, NULL, &errmsg);
- if (res != SQLITE_OK && errmsg != 0)
- {
- std::cerr << "DEBUG: " << errmsg << std::endl;
- sqlite3_free (errmsg);
- }
+ sqlite3_exec (m_db, INIT_DATABASE.c_str (), NULL, NULL, NULL);
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, sqlite3_errmsg (m_db));
}
DbHelper::~DbHelper ()
@@ -222,7 +86,7 @@
sqlite3_result_error (context, "Hash expects (blob,integer) parameters", -1);
return;
}
-
+
EVP_MD_CTX **hash_context = reinterpret_cast<EVP_MD_CTX **> (sqlite3_aggregate_context (context, sizeof (EVP_MD_CTX *)));
if (hash_context == 0)
@@ -236,7 +100,7 @@
*hash_context = EVP_MD_CTX_create ();
EVP_DigestInit_ex (*hash_context, HASH_FUNCTION (), 0);
}
-
+
int nameBytes = sqlite3_value_bytes (argv[0]);
const void *name = sqlite3_value_blob (argv[0]);
sqlite3_int64 seqno = sqlite3_value_int64 (argv[1]);
@@ -262,7 +126,7 @@
sqlite3_result_blob (context, &charNullResult, 1, SQLITE_TRANSIENT); //SQLITE_TRANSIENT forces to make a copy
return;
}
-
+
unsigned char *hash = new unsigned char [EVP_MAX_MD_SIZE];
unsigned int hashLength = 0;
@@ -271,7 +135,7 @@
sqlite3_result_blob (context, hash, hashLength, SQLITE_TRANSIENT); //SQLITE_TRANSIENT forces to make a copy
delete [] hash;
-
+
EVP_MD_CTX_destroy (*hash_context);
}
diff --git a/src/dispatcher.cc b/src/dispatcher.cc
index c815258..d48bb43 100644
--- a/src/dispatcher.cc
+++ b/src/dispatcher.cc
@@ -29,7 +29,7 @@
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, const Ccnx::CcnxWrapperPtr &ccnx, const SchedulerPtr &scheduler, int poolSize)
+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)
@@ -38,9 +38,12 @@
, m_localUserName(localUserName)
, m_sharedFolder(sharedFolder)
{
- m_log = make_shared<ActionLog>(m_ccnx, path, localUserName, 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_log, localUserName, localPrefix, syncPrefix, bind(&Dispatcher::syncStateChanged, this, _1), ccnx, scheduler);
+ m_core = new SyncCore (m_syncLog, localUserName, localPrefix, syncPrefix,
+ bind(&Dispatcher::syncStateChanged, this, _1), ccnx, scheduler);
}
Dispatcher::~Dispatcher()
@@ -95,11 +98,11 @@
{
case UPDATE:
{
- filesystem::path absolutePath = m_rootDir + relativeFilePath;
+ filesystem::path absolutePath = m_rootDir / relativeFilePath;
if (filesystem::exists(absolutePath))
{
HashPtr hash = Hash::FromFileContent(absolutePath);
- if (m_log->KnownFileState(relativeFilePath.string(), *hash))
+ 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
@@ -113,7 +116,7 @@
time_t wtime = filesystem::last_write_time(absolutePath);
filesystem::file_status stat = filesystem::status(absolutePath);
int mode = stat.permissions();
- m_log->AddActionUpdate (relativeFilePath.string(), *hash, wtime, mode, seg_num);
+ 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
@@ -128,7 +131,7 @@
}
case DELETE:
{
- m_log->AddActionDelete (relativeFilePath.string());
+ m_actionLog->AddActionDelete (relativeFilePath.string());
// notify SyncCore to propagate the change
m_core->localStateChanged();
break;
@@ -178,7 +181,7 @@
{
case ActionItem::UPDATE:
{
- // TODO:
+ // @TODO
// need a function in ActionLog to apply received action, i.e. record remote action in ActionLog
string hashBytes = actionItem->file_hash();
@@ -197,7 +200,7 @@
{
string filename = actionItem->filename();
// TODO:
- // m_log->AddRemoteActionDelete(filename);
+ // m_actionLog->AddRemoteActionDelete(filename);
break;
}
default:
diff --git a/src/dispatcher.h b/src/dispatcher.h
index 579ff79..7c3220e 100644
--- a/src/dispatcher.h
+++ b/src/dispatcher.h
@@ -49,7 +49,9 @@
// 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, const Ccnx::CcnxWrapperPtr &ccnx, const SchedulerPtr &scheduler, int poolSize = 2);
+ 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 -------
@@ -91,9 +93,11 @@
fileReady(const Ccnx::Name &fileNamePrefix);
private:
+ Ccnx::CcnxWrapperPtr m_ccnx;
SyncCore *m_core;
- ActionLogPtr m_log;
- CcnxWrapperPtr m_ccnx;
+ SyncLogPtr m_syncLog;
+ ActionLogPtr m_actionLog;
+
boost::filesystem::path m_rootDir;
Executor m_executor;
ObjectManager m_objectManager;
diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
new file mode 100644
index 0000000..7df2687
--- /dev/null
+++ b/src/fetch-manager.cc
@@ -0,0 +1,134 @@
+/* -*- 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: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#include "fetch-manager.h"
+#include <boost/make_shared.hpp>
+#include <boost/ref.hpp>
+#include <boost/throw_exception.hpp>
+
+using namespace boost;
+using namespace std;
+using namespace Ccnx;
+
+//The disposer object function
+struct fetcher_disposer { void operator() (Fetcher *delete_this) { delete delete_this; } };
+
+FetchManager::FetchManager (CcnxWrapperPtr ccnx, SyncLogPtr sync, uint32_t parallelFetches/* = 3*/)
+ : m_ccnx (ccnx)
+ , m_sync (sync)
+ , m_maxParallelFetches (parallelFetches)
+ , m_currentParallelFetches (0)
+
+{
+ m_scheduler = make_shared<Scheduler> ();
+ m_scheduler->start ();
+}
+
+FetchManager::~FetchManager ()
+{
+ m_fetchList.clear_and_dispose (fetcher_disposer ());
+
+ m_scheduler->shutdown ();
+ m_scheduler.reset ();
+}
+
+void
+FetchManager::Enqueue (const Ccnx::Name &deviceName, uint32_t minSeqNo, uint32_t maxSeqNo, int priority/*=PRIORITY_NORMAL*/)
+{
+ // we may need to guarantee that LookupLocator will gives an answer and not throw exception...
+ Name forwardingHint;
+ try {
+ forwardingHint = m_sync->LookupLocator (deviceName);
+ }
+ catch (Error::Db &exception) {
+ // just ignore for now
+ }
+
+ Fetcher &fetcher = *(new Fetcher (m_ccnx,
+ bind (&FetchManager::DidDataSegmentFetched, this, _1, _2, _3, _4, _5),
+ bind (&FetchManager::DidFetchComplete, this, _1),
+ bind (&FetchManager::DidNoDataTimeout, this, _1),
+ deviceName, minSeqNo, maxSeqNo
+ /* Alex: should or should not include hint initially?*/));
+
+ switch (priority)
+ {
+ case PRIORITY_HIGH:
+ m_fetchList.push_front (fetcher);
+ break;
+
+ case PRIORITY_NORMAL:
+ default:
+ m_fetchList.push_back (fetcher);
+ break;
+ }
+
+ ScheduleFetches (); // will start a fetch if m_currentParallelFetches is less than max, otherwise does nothing
+}
+
+void
+FetchManager::ScheduleFetches ()
+{
+ unique_lock<mutex> lock (m_parellelFetchMutex);
+
+ for (FetchList::iterator item = m_fetchList.begin ();
+ m_currentParallelFetches < m_maxParallelFetches && item != m_fetchList.end ();
+ item++)
+ {
+ if (item->IsActive ())
+ continue;
+
+ m_currentParallelFetches ++;
+ item->RestartPipeline ();
+ }
+}
+
+void
+FetchManager::DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, Ccnx::PcoPtr data)
+{
+ // do something
+}
+
+void
+FetchManager::DidNoDataTimeout (Fetcher &fetcher)
+{
+ fetcher.SetForwardingHint (Ccnx::Name ("/ndn/broadcast"));
+ {
+ unique_lock<mutex> lock (m_parellelFetchMutex);
+ m_currentParallelFetches --;
+ // no need to do anything with the m_fetchList
+ }
+
+ ScheduleFetches ();
+}
+
+void
+FetchManager::DidFetchComplete (Fetcher &fetcher)
+{
+ {
+ unique_lock<mutex> lock (m_parellelFetchMutex);
+ m_currentParallelFetches --;
+ m_fetchList.erase_and_dispose (FetchList::s_iterator_to (fetcher), fetcher_disposer ());
+ }
+
+ // ? do something else
+}
diff --git a/src/fetch-manager.h b/src/fetch-manager.h
new file mode 100644
index 0000000..cfceef5
--- /dev/null
+++ b/src/fetch-manager.h
@@ -0,0 +1,113 @@
+/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2012-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
+ *
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#ifndef FETCH_MANAGER_H
+#define FETCH_MANAGER_H
+
+#include <boost/exception/all.hpp>
+#include <boost/shared_ptr.hpp>
+#include <string>
+#include <list>
+#include <stdint.h>
+#include "scheduler.h"
+
+#include "ccnx-wrapper.h"
+#include "sync-log.h"
+
+#include "fetcher.h"
+
+class FetchManager
+{
+ enum
+ {
+ PRIORITY_NORMAL,
+ PRIORITY_HIGH
+ };
+
+public:
+ FetchManager (Ccnx::CcnxWrapperPtr ccnx, SyncLogPtr sync, uint32_t parallelFetches = 3);
+ virtual ~FetchManager ();
+
+ void
+ Enqueue (const Ccnx::Name &deviceName,
+ uint32_t minSeqNo, uint32_t maxSeqNo, int priority=PRIORITY_NORMAL);
+
+ inline Ccnx::CcnxWrapperPtr
+ GetCcnx ();
+
+ inline SchedulerPtr
+ GetScheduler ();
+
+ // Fetch Events
+ void
+ DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, Ccnx::PcoPtr data);
+
+ void
+ DidNoDataTimeout (Fetcher &fetcher);
+
+ void
+ DidFetchComplete (Fetcher &fetcher);
+
+private:
+ void
+ ScheduleFetches ();
+
+private:
+
+private:
+ Ccnx::CcnxWrapperPtr m_ccnx;
+ SyncLogPtr m_sync; // to access forwarding hints
+ SchedulerPtr m_scheduler;
+
+ uint32_t m_maxParallelFetches;
+ uint32_t m_currentParallelFetches;
+ boost::mutex m_parellelFetchMutex;
+
+ // optimized list structure for fetch queue
+ typedef boost::intrusive::member_hook< Fetcher,
+ boost::intrusive::list_member_hook<>, &Fetcher::m_managerListHook> MemberOption;
+ typedef boost::intrusive::list<Fetcher, MemberOption> FetchList;
+
+ FetchList m_fetchList;
+};
+
+Ccnx::CcnxWrapperPtr
+FetchManager::GetCcnx ()
+{
+ return m_ccnx;
+}
+
+SchedulerPtr
+FetchManager::GetScheduler ()
+{
+ return m_scheduler;
+}
+
+typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
+namespace Error {
+struct FetchManager : virtual boost::exception, virtual std::exception { };
+}
+
+typedef boost::shared_ptr<FetchManager> FetchManagerPtr;
+
+
+#endif // FETCHER_H
diff --git a/src/fetcher.cc b/src/fetcher.cc
index e021540..f009819 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -20,51 +20,153 @@
*/
#include "fetcher.h"
+#include "fetch-manager.h"
+#include "ccnx-pco.h"
+
#include <boost/make_shared.hpp>
#include <boost/ref.hpp>
#include <boost/throw_exception.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost;
using namespace std;
-namespace fs = boost::filesystem;
+using namespace Ccnx;
-const std::string INIT_DATABASE = "\
-PRAGMA foreign_keys = ON; \
- \
-CREATE TABLE \
- Queue( \
- name BLOB NOT NULL PRIMARY KEY, \
- seq_no_rcvd INTEGER, \
- seq_no_available INTEGER, \
- ); \
-";
+Fetcher::Fetcher (CcnxWrapperPtr ccnx,
+ OnDataSegmentCallback onDataSegment,
+ OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
+ const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+ boost::posix_time::time_duration timeout/* = boost::posix_time::seconds (30)*/,
+ const Ccnx::Name &forwardingHint/* = Ccnx::Name ()*/)
+ : m_ccnx (ccnx)
-Fetcher::Fetcher (const fs::path &path, const string &name)
+ , m_onDataSegment (onDataSegment)
+ , m_onFetchComplete (onFetchComplete)
+ , m_onFetchFailed (onFetchFailed)
+
+ , m_active (false)
+ , m_name (name)
+ , m_forwardingHint (forwardingHint)
+ , m_maximumNoActivityPeriod (timeout)
+
+ , m_minSendSeqNo (-1)
+ , m_maxInOrderRecvSeqNo (-1)
+ , m_minSeqNo (minSeqNo)
+ , m_maxSeqNo (maxSeqNo)
+
+ , m_pipeline (6) // initial "congestion window"
+ , m_activePipeline (0)
{
- fs::path chronoshareDirectory = path / ".chronoshare";
- fs::create_directories (chronoshareDirectory);
-
- int res = sqlite3_open((chronoshareDirectory / (name+".db")).c_str (), &m_db);
- if (res != SQLITE_OK)
- {
- BOOST_THROW_EXCEPTION (Error::Fetcher ()
- << errmsg_info_str ("Cannot open/create dabatabase: [" + (chronoshareDirectory / (name+".db")).string () + "]"));
- }
-
- char *errmsg = 0;
- res = sqlite3_exec (m_db, INIT_DATABASE.c_str (), NULL, NULL, &errmsg);
- if (res != SQLITE_OK && errmsg != 0)
- {
- // std::cerr << "DEBUG: " << errmsg << std::endl;
- sqlite3_free (errmsg);
- }
}
Fetcher::~Fetcher ()
{
- int res = sqlite3_close (m_db);
- if (res != SQLITE_OK)
+}
+
+void
+Fetcher::RestartPipeline ()
+{
+ m_active = true;
+ m_minSendSeqNo = m_maxInOrderRecvSeqNo;
+ // cout << "Restart: " << m_minSendSeqNo << endl;
+ m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
+
+ FillPipeline ();
+}
+
+void
+Fetcher::SetForwardingHint (const Ccnx::Name &forwardingHint)
+{
+ m_forwardingHint = forwardingHint;
+}
+
+void
+Fetcher::FillPipeline ()
+{
+ for (; m_minSendSeqNo < m_maxSeqNo && m_activePipeline < m_pipeline; m_minSendSeqNo++)
{
- // complain
+ if (m_outOfOrderRecvSeqNo.find (m_minSendSeqNo+1) != m_outOfOrderRecvSeqNo.end ())
+ continue;
+
+ // cout << ">>> " << m_minSendSeqNo+1 << endl;
+ m_ccnx->sendInterest (Name (m_forwardingHint)(m_name)(m_minSendSeqNo+1),
+ Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
+ bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)),
+ Selectors().interestLifetime (1)); // Alex: this lifetime should be changed to RTO
+
+ m_activePipeline ++;
}
}
+
+void
+Fetcher::OnData (uint32_t seqno, const Ccnx::Name &name, PcoPtr data)
+{
+ if (m_forwardingHint == Name ())
+ m_onDataSegment (*this, seqno, m_name, name, data);
+ else
+ {
+ try {
+ PcoPtr pco = make_shared<ParsedContentObject> (*data->contentPtr ());
+ m_onDataSegment (*this, seqno, m_name, pco->name (), pco);
+ }
+ catch (MisformedContentObjectException &e)
+ {
+ cerr << "MisformedContentObjectException..." << endl;
+ // no idea what should do...
+ // let's ignore for now
+ }
+ }
+
+ m_activePipeline --;
+ m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
+
+ ////////////////////////////////////////////////////////////////////////////
+ m_outOfOrderRecvSeqNo.insert (seqno);
+ set<int32_t>::iterator inOrderSeqNo = m_outOfOrderRecvSeqNo.begin ();
+ for (; inOrderSeqNo != m_outOfOrderRecvSeqNo.end ();
+ inOrderSeqNo++)
+ {
+ if (*inOrderSeqNo == m_maxInOrderRecvSeqNo+1)
+ {
+ m_maxInOrderRecvSeqNo = *inOrderSeqNo;
+ }
+ else
+ break;
+ }
+ m_outOfOrderRecvSeqNo.erase (m_outOfOrderRecvSeqNo.begin (), inOrderSeqNo);
+ ////////////////////////////////////////////////////////////////////////////
+
+ if (m_maxInOrderRecvSeqNo == m_maxSeqNo)
+ {
+ m_active = false;
+ m_onFetchComplete (*this);
+ }
+ else
+ {
+ FillPipeline ();
+ }
+}
+
+Closure::TimeoutCallbackReturnValue
+Fetcher::OnTimeout (uint32_t seqno, const Ccnx::Name &name)
+{
+ // cout << "Fetcher::OnTimeout: " << name << endl;
+ // cout << "Last: " << m_lastPositiveActivity << ", config: " << m_maximumNoActivityPeriod
+ // << ", now: " << date_time::second_clock<boost::posix_time::ptime>::universal_time()
+ // << ", oldest: " << (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod) << endl;
+
+ if (m_lastPositiveActivity <
+ (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod))
+ {
+ m_activePipeline --;
+ if (m_activePipeline == 0)
+ {
+ m_active = false;
+ m_onFetchFailed (*this);
+ // this is not valid anymore, but we still should be able finish work
+ }
+ return Closure::RESULT_OK;
+ }
+ else
+ return Closure::RESULT_REEXPRESS;
+}
diff --git a/src/fetcher.h b/src/fetcher.h
index 8cb4664..12ad67f 100644
--- a/src/fetcher.h
+++ b/src/fetcher.h
@@ -22,31 +22,95 @@
#ifndef FETCHER_H
#define FETCHER_H
-#include <stdint.h>
-#include <sqlite3.h>
-#include <boost/exception/all.hpp>
-#include <string>
-#include <boost/filesystem.hpp>
+#include "ccnx-wrapper.h"
+#include "ccnx-name.h"
-typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
+#include "scheduler.h"
+#include <boost/intrusive/list.hpp>
+#include <boost/date_time/posix_time/posix_time_types.hpp>
+
+class FetchManager;
class Fetcher
{
public:
- Fetcher (const boost::filesystem::path &path, const std::string &name);
- virtual ~Fetcher ();
-
-private:
+ typedef boost::function<void (Fetcher &, uint32_t /*requested seqno*/, const Ccnx::Name & /*requested base name*/,
+ const Ccnx::Name & /*actual name*/, Ccnx::PcoPtr /*content object*/)> OnDataSegmentCallback;
+ typedef boost::function<void (Fetcher &)> OnFetchCompleteCallback;
+ typedef boost::function<void (Fetcher &)> OnFetchFailedCallback;
-protected:
- sqlite3 *m_db;
+ Fetcher (Ccnx::CcnxWrapperPtr ccnx,
+ OnDataSegmentCallback onDataSegment,
+ OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
+ const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+ boost::posix_time::time_duration timeout = boost::posix_time::seconds (30), // this time is not precise, but sets min bound
+ // actual time depends on how fast Interests timeout
+ const Ccnx::Name &forwardingHint = Ccnx::Name ());
+ virtual ~Fetcher ();
+
+ inline bool
+ IsActive () const;
+
+ void
+ RestartPipeline ();
+
+ void
+ SetForwardingHint (const Ccnx::Name &forwardingHint);
+
+private:
+ void
+ FillPipeline ();
+
+ void
+ OnData (uint32_t seqno, const Ccnx::Name &name, Ccnx::PcoPtr data);
+
+ Ccnx::Closure::TimeoutCallbackReturnValue
+ OnTimeout (uint32_t seqno, const Ccnx::Name &name);
+
+public:
+ boost::intrusive::list_member_hook<> m_managerListHook;
+
+private:
+ Ccnx::CcnxWrapperPtr m_ccnx;
+
+ OnDataSegmentCallback m_onDataSegment;
+ OnFetchCompleteCallback m_onFetchComplete;
+ OnFetchFailedCallback m_onFetchFailed;
+
+ bool m_active;
+
+ Ccnx::Name m_name;
+ Ccnx::Name m_forwardingHint;
+
+ boost::posix_time::time_duration m_maximumNoActivityPeriod;
+
+ int32_t m_minSendSeqNo;
+ int32_t m_maxInOrderRecvSeqNo;
+ std::set<int32_t> m_outOfOrderRecvSeqNo;
+
+ int32_t m_minSeqNo;
+ int32_t m_maxSeqNo;
+
+ uint32_t m_pipeline;
+ uint32_t m_activePipeline;
+
+ boost::posix_time::ptime m_lastPositiveActivity;
+
};
+typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
+
namespace Error {
struct Fetcher : virtual boost::exception, virtual std::exception { };
}
typedef boost::shared_ptr<Fetcher> FetcherPtr;
+bool
+Fetcher::IsActive () const
+{
+ return m_active;
+}
+
#endif // FETCHER_H
diff --git a/src/logging.cc b/src/logging.cc
new file mode 100644
index 0000000..28d965b
--- /dev/null
+++ b/src/logging.cc
@@ -0,0 +1,61 @@
+/* -*- 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 "logging.h"
+
+#ifdef HAVE_LOG4CXX
+
+#include <log4cxx/logger.h>
+#include <log4cxx/basicconfigurator.h>
+#include <log4cxx/consoleappender.h>
+#include <log4cxx/patternlayout.h>
+#include <log4cxx/level.h>
+#include <log4cxx/propertyconfigurator.h>
+#include <log4cxx/defaultconfigurator.h>
+#include <log4cxx/helpers/exception.h>
+using namespace log4cxx;
+using namespace log4cxx::helpers;
+
+#include <unistd.h>
+
+void
+INIT_LOGGERS ()
+{
+ static bool configured = false;
+
+ if (configured) return;
+
+ if (access ("log4cxx.properties", R_OK)==0)
+ PropertyConfigurator::configureAndWatch ("log4cxx.properties");
+ else
+ {
+ PatternLayoutPtr layout (new PatternLayout ("%d{HH:mm:ss} %p %c{1} - %m%n"));
+ ConsoleAppenderPtr appender (new ConsoleAppender (layout));
+
+ BasicConfigurator::configure( appender );
+ Logger::getRootLogger()->setLevel (log4cxx::Level::getInfo ());
+ }
+
+ configured = true;
+}
+
+#endif
diff --git a/src/logging.h b/src/logging.h
new file mode 100644
index 0000000..849d2e8
--- /dev/null
+++ b/src/logging.h
@@ -0,0 +1,84 @@
+/* -*- 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: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#ifndef LOGGING_H
+#define LOGGING_H
+
+#include "config.h"
+
+#ifdef HAVE_LOG4CXX
+
+#include <log4cxx/logger.h>
+
+#define INIT_LOGGER(name) \
+ static log4cxx::LoggerPtr staticModuleLogger = log4cxx::Logger::getLogger (name);
+
+#define _LOG_DEBUG(x) \
+ LOG4CXX_DEBUG(staticModuleLogger, x);
+
+#define _LOG_TRACE(x) \
+ LOG4CXX_TRACE(staticModuleLogger, x);
+
+#define _LOG_FUNCTION(x) \
+ LOG4CXX_TRACE(staticModuleLogger, __FUNCTION__ << "(" << x << ")");
+
+#define _LOG_FUNCTION_NOARGS \
+ LOG4CXX_TRACE(staticModuleLogger, __FUNCTION__ << "()");
+
+#define _LOG_ERROR(x) \
+ LOG4CXX_ERROR(staticModuleLogger, x);
+
+#define _LOG_ERROR_COND(cond,x) \
+ if (cond) { _LOG_ERROR(x) }
+
+#define _LOG_DEBUG_COND(cond,x) \
+ if (cond) { _LOG_DEBUG(x) }
+
+void
+INIT_LOGGERS ();
+
+#else // else HAVE_LOG4CXX
+
+#define INIT_LOGGER(name)
+#define _LOG_FUNCTION(x)
+#define _LOG_FUNCTION_NOARGS
+#define _LOG_TRACE(x)
+#define INIT_LOGGERS(x)
+#define _LOG_ERROR(x)
+#define _LOG_ERROR_COND(cond,x)
+#define _LOG_DEBUG_COND(cond,x)
+
+#ifdef _DEBUG
+
+#include <boost/thread/thread.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
+#include <iostream>
+
+#define _LOG_DEBUG(x) \
+ std::clog << boost::get_system_time () << " " << boost::this_thread::get_id () << " " << x << endl;
+
+#else
+#define _LOG_DEBUG(x)
+#endif
+
+#endif // HAVE_LOG4CXX
+
+#endif // LOGGING_H
diff --git a/src/object-manager.cc b/src/object-manager.cc
index 7f66b2a..bb19842 100644
--- a/src/object-manager.cc
+++ b/src/object-manager.cc
@@ -55,7 +55,7 @@
{
HashPtr fileHash = Hash::FromFileContent (file);
ObjectDb fileDb (m_folder, lexical_cast<string> (*fileHash));
-
+
fs::ifstream iff (file, std::ios::in | std::ios::binary);
sqlite3_int64 segment = 0;
while (iff.good ())
@@ -63,21 +63,17 @@
char buf[MAX_FILE_SEGMENT_SIZE];
iff.read (buf, MAX_FILE_SEGMENT_SIZE);
- Name name (deviceName);
- name
- .appendComp ("file")
- .appendComp (fileHash->GetHash (), fileHash->GetHashBytes ())
- .appendComp (segment);
+ Name name = Name (deviceName)("file")(fileHash->GetHash (), fileHash->GetHashBytes ())(segment);
// cout << *fileHash << endl;
// cout << name << endl;
Bytes data = m_ccnx->createContentObject (name, buf, iff.gcount ());
fileDb.saveContentObject (deviceName, segment, data);
-
+
segment ++;
}
-
+
return fileHash;
}
@@ -103,12 +99,12 @@
BytesPtr data = obj.contentPtr ();
off.write (reinterpret_cast<const char*> (head(*data)), data->size());
-
+
segment ++;
bytes = fileDb.fetchSegment (deviceName, segment);
}
// permission and timestamp should be assigned somewhere else (ObjectManager has no idea about that)
-
+
return true;
}
diff --git a/src/sync-core.cc b/src/sync-core.cc
index 79226b2..d85093b 100644
--- a/src/sync-core.cc
+++ b/src/sync-core.cc
@@ -20,136 +20,80 @@
*/
#include "sync-core.h"
-
-#include "one-time-task.h"
+#include "sync-state-helper.h"
+#include "logging.h"
#include "random-interval-generator.h"
+#include <boost/lexical_cast.hpp>
+
+INIT_LOGGER ("Sync.Core");
+
const string SyncCore::RECOVER = "RECOVER";
const double SyncCore::WAIT = 0.05;
const double SyncCore::RANDOM_PERCENT = 0.5;
-// for debugging
-static void
-printMsg(SyncStateMsgPtr &msg)
-{
- cout << " ===== start Msg ======" << endl;
- int size = msg->state_size();
- if (size > 0)
- {
- int index = 0;
- while (index < size)
- {
- SyncState state = msg->state(index);
- string strName = state.name();
- string strLocator = state.locator();
- sqlite3_int64 seq = state.seq();
- cout << "Name: " << Name((const unsigned char *)strName.c_str(), strName.size());
- cout << ", Locator: " << Name((const unsigned char *)strLocator.c_str(), strLocator.size());
- cout << ", seq: " << seq << endl;
- index ++;
- }
- }
- else
- {
- cout << "Msg size 0" << endl;
- }
- cout << " ++++++++ end Msg ++++++++ \n\n" << endl;
-}
+using namespace boost;
+using namespace Ccnx;
-SyncCore::SyncCore(SyncLogPtr syncLog, const Name &userName, const Name &localPrefix, const Name &syncPrefix, const StateMsgCallback &callback, const CcnxWrapperPtr &handle, const SchedulerPtr &scheduler)
- : m_log(syncLog)
+SyncCore::SyncCore(SyncLogPtr syncLog, const Name &userName, const Name &localPrefix, const Name &syncPrefix,
+ const StateMsgCallback &callback, CcnxWrapperPtr ccnx, SchedulerPtr scheduler)
+ : m_ccnx (ccnx)
+ , m_log(syncLog)
, m_scheduler(scheduler)
, m_stateMsgCallback(callback)
- , m_userName(userName)
- , m_localPrefix(localPrefix)
, m_syncPrefix(syncPrefix)
- , m_handle(handle)
, m_recoverWaitGenerator(new RandomIntervalGenerator(WAIT, RANDOM_PERCENT, RandomIntervalGenerator::UP))
{
m_rootHash = m_log->RememberStateInStateLog();
- m_syncClosure = new Closure(boost::bind(&SyncCore::handleSyncData, this, _1, _2), boost::bind(&SyncCore::handleSyncInterestTimeout, this, _1));
- m_recoverClosure = new Closure(boost::bind(&SyncCore::handleRecoverData, this, _1, _2), boost::bind(&SyncCore::handleRecoverInterestTimeout, this, _1));
- m_handle->setInterestFilter(m_syncPrefix, boost::bind(&SyncCore::handleInterest, this, _1));
- m_log->initYP(m_yp);
+
+ m_ccnx->setInterestFilter(m_syncPrefix, boost::bind(&SyncCore::handleInterest, this, _1));
+ // m_log->initYP(m_yp);
+ m_log->UpdateLocalLocator (localPrefix);
+
m_scheduler->start();
sendSyncInterest();
}
SyncCore::~SyncCore()
{
- if (m_syncClosure != 0)
- {
- delete m_syncClosure;
- m_syncClosure = 0;
- }
-
- if (m_recoverClosure != 0)
- {
- delete m_recoverClosure;
- m_recoverClosure = 0;
- }
-}
-
-Name
-SyncCore::yp(const Name &deviceName)
-{
- Name locator;
- ReadLock lock(m_ypMutex);
- if (m_yp.find(deviceName) != m_yp.end())
- {
- locator = m_yp[deviceName];
- }
- else
- {
- cout << "self: " << m_userName << ", deviceName: " << deviceName << " not found in yp " << endl;
- }
- return locator;
+ // need to "deregister" closures
}
void
-SyncCore::updateLocalPrefix(const Name &localPrefix)
+SyncCore::updateLocalPrefix (const Name &localPrefix)
{
- m_localPrefix = localPrefix;
- // optionally, we can have a sync action to announce the new prefix
- // we are not doing this for now
+ m_log->UpdateLocalLocator (localPrefix);
}
void
SyncCore::updateLocalState(sqlite3_int64 seqno)
{
- m_log->UpdateDeviceSeqNo(m_userName, seqno);
+ m_log->UpdateLocalSeqNo (seqno);
localStateChanged();
-
}
void
SyncCore::localStateChanged()
{
- // choose to update locator everytime
- m_log->UpdateLocator(m_userName, m_localPrefix);
- {
- WriteLock lock(m_ypMutex);
- m_yp[m_userName] = m_localPrefix;
- }
HashPtr oldHash = m_rootHash;
m_rootHash = m_log->RememberStateInStateLog();
SyncStateMsgPtr msg = m_log->FindStateDifferences(*oldHash, *m_rootHash);
// reply sync Interest with oldHash as last component
- Name syncName = constructSyncName(oldHash);
- Bytes syncData;
- msgToBytes(msg, syncData);
- m_handle->publishData(syncName, syncData, FRESHNESS);
- cout << m_userName << " publishes: " << *oldHash << endl;
- printMsg(msg);
+ Name syncName = Name (m_syncPrefix)(oldHash->GetHash(), oldHash->GetHashBytes());
+ BytesPtr syncData = serializeMsg (msg);
+
+ m_ccnx->publishData(syncName, *syncData, FRESHNESS);
+ _LOG_TRACE (m_log->GetLocalName () << " publishes: " << *oldHash);
+ _LOG_TRACE (msg);
// 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
- ostringstream ss;
- ss << *m_rootHash;
- TaskPtr task(new OneTimeTask(boost::bind(&SyncCore::sendSyncInterest, this), ss.str(), m_scheduler, 0.05));
- m_scheduler->addTask(task);
+ Scheduler::scheduleOneTimeTask (m_scheduler, 0.05,
+ bind(&SyncCore::sendSyncInterest, this),
+ lexical_cast<string> (*m_rootHash));
+
sendSyncInterest();
}
@@ -180,38 +124,14 @@
{
// we know the hash, should reply everything
SyncStateMsgPtr msg = m_log->FindStateDifferences(*(Hash::Origin), *m_rootHash);
- // DEBUG
- /*
- assert(msg->state_size() > 0);
- int size = msg->state_size();
- int index = 0;
- cout << "Reply recover interest with: " << endl;
- while (index < size)
- {
- SyncState state = msg->state(index);
- string strName = state.name();
- string strLoc = state.locator();
- cout << "Name: " << Name((const unsigned char *)strName.c_str(), strName.size()) << ", Loc: " << Name((const unsigned char *)strLoc.c_str(), strLoc.size()) << ", seq: " << state.seq() << endl;
- if (state.type() == SyncState::UPDATE)
- {
- cout << "Action: update" << endl;
+
+ BytesPtr syncData = serializeMsg (msg);
+ m_ccnx->publishData(name, *syncData, FRESHNESS);
+ _LOG_TRACE (m_log->GetLocalName () << " publishes " << hash);
+ _LOG_TRACE (msg);
}
else
{
- cout << "Action: delete" << endl;
- }
- index++;
- }
- */
- // END DEBUG
- Bytes syncData;
- msgToBytes(msg, syncData);
- m_handle->publishData(name, syncData, FRESHNESS);
- cout << m_userName << " publishes " << hash << endl;
- printMsg(msg);
- }
- else
- {
// we don't recognize this hash, can not help
}
}
@@ -224,32 +144,30 @@
if (*hash == *m_rootHash)
{
// we have the same hash; nothing needs to be done
- cout << "same as root hash: " << *hash << endl;
+ _LOG_TRACE ("same as root hash: " << *hash);
return;
}
else if (m_log->LookupSyncLog(*hash) > 0)
{
// we know something more
- cout << "found hash in sync log" << endl;
+ _LOG_TRACE ("found hash in sync log");
SyncStateMsgPtr msg = m_log->FindStateDifferences(*hash, *m_rootHash);
- Bytes syncData;
- msgToBytes(msg, syncData);
- m_handle->publishData(name, syncData, FRESHNESS);
- cout << m_userName << " publishes: " << *hash << endl;
- printMsg(msg);
+ BytesPtr syncData = serializeMsg (msg);
+ m_ccnx->publishData(name, *syncData, FRESHNESS);
+ _LOG_TRACE (m_log->GetLocalName () << " publishes: " << *hash);
+ _LOG_TRACE (msg);
}
else
{
// we don't recognize the hash, send recover Interest if still don't know the hash after a randomized wait period
- ostringstream ss;
- ss << *hash;
double wait = m_recoverWaitGenerator->nextInterval();
- cout << m_userName << ", rootHash: " << *m_rootHash << ", hash: " << *hash << endl;
- cout << "recover task scheduled after wait: " << wait << endl;
- TaskPtr task(new OneTimeTask(boost::bind(&SyncCore::recover, this, hash), ss.str(), m_scheduler, wait));
- m_scheduler->addTask(task);
+ _LOG_TRACE (m_log->GetLocalName () << ", rootHash: " << *m_rootHash << ", hash: " << *hash);
+ _LOG_TRACE ("recover task scheduled after wait: " << wait);
+ Scheduler::scheduleOneTimeTask (m_scheduler,
+ wait, boost::bind(&SyncCore::recover, this, hash),
+ "r-"+lexical_cast<string> (*hash));
}
}
@@ -272,18 +190,18 @@
}
void
-SyncCore::handleRecoverData(const Name &name, const Bytes &content)
+SyncCore::handleRecoverData(const Name &name, PcoPtr content)
{
- //cout << "handle recover data" << endl;
- handleStateData(content);
+ //cout << "handle recover data" << end;
+ handleStateData(*content->contentPtr ());
sendSyncInterest();
}
void
-SyncCore::handleSyncData(const Name &name, const Bytes &content)
+SyncCore::handleSyncData(const Name &name, PcoPtr content)
{
// suppress recover in interest - data out of order case
- handleStateData(content);
+ handleStateData(*content->contentPtr ());
// resume outstanding sync interest
sendSyncInterest();
@@ -297,12 +215,12 @@
if(!success)
{
// ignore misformed SyncData
- cerr << "Misformed SyncData"<< endl;
+ _LOG_ERROR ("Misformed SyncData");
return;
}
- cout << m_userName << " receives Msg " << endl;
- printMsg (msg);
+ _LOG_TRACE (m_log->GetLocalName () << " receives Msg ");
+ _LOG_TRACE (msg);
int size = msg->state_size();
int index = 0;
while (index < size)
@@ -322,14 +240,13 @@
Name locatorName((const unsigned char *)locStr.c_str(), locStr.size());
// cout << ", Got loc: " << locatorName << endl;
m_log->UpdateLocator(deviceName, locatorName);
- WriteLock lock(m_ypMutex);
- m_yp[deviceName] = locatorName;
- cout << "self: " << m_userName << ", device: " << deviceName << " < == > " << locatorName << endl;
+
+ _LOG_TRACE ("self: " << m_log->GetLocalName () << ", device: " << deviceName << " < == > " << locatorName);
}
}
else
{
- cout << "nani" << endl;
+ _LOG_ERROR ("Receive SYNC DELETE, but we don't support it yet");
deregister(deviceName);
}
index++;
@@ -350,9 +267,13 @@
void
SyncCore::sendSyncInterest()
{
- Name syncInterest = constructSyncName(m_rootHash);
- m_handle->sendInterest(syncInterest, m_syncClosure);
- cout << m_userName << " send SYNC interest: " << *m_rootHash << endl;
+ Name syncInterest = Name (m_syncPrefix)(m_rootHash->GetHash(), m_rootHash->GetHashBytes());
+
+ m_ccnx->sendInterest(syncInterest,
+ Closure (boost::bind(&SyncCore::handleSyncData, this, _1, _2),
+ boost::bind(&SyncCore::handleSyncInterestTimeout, this, _1)));
+
+ _LOG_TRACE (m_log->GetLocalName () << " send SYNC interest: " << *m_rootHash);
}
void
@@ -360,16 +281,19 @@
{
if (!(*hash == *m_rootHash) && m_log->LookupSyncLog(*hash) <= 0)
{
- cout << m_userName << ", Recover for: " << *hash << endl;
+ _LOG_TRACE (m_log->GetLocalName () << ", Recover for: " << *hash);
// unfortunately we still don't recognize this hash
Bytes bytes;
readRaw(bytes, (const unsigned char *)hash->GetHash(), hash->GetHashBytes());
- Name recoverInterest = m_syncPrefix;
- recoverInterest.appendComp(RECOVER);
+
// append the unknown hash
- recoverInterest.appendComp(bytes);
- m_handle->sendInterest(recoverInterest, m_recoverClosure);
- cout << m_userName << " send RECOVER Interest: " << *hash << endl;
+ Name recoverInterest = Name (m_syncPrefix)(RECOVER)(bytes);
+
+ m_ccnx->sendInterest(recoverInterest,
+ Closure (boost::bind(&SyncCore::handleRecoverData, this, _1, _2),
+ boost::bind(&SyncCore::handleRecoverInterestTimeout, this, _1)));
+
+ _LOG_TRACE (m_log->GetLocalName () << " send RECOVER Interest: " << *hash);
}
else
{
@@ -384,24 +308,6 @@
// TODO: handle deregistering
}
-Name
-SyncCore::constructSyncName(const HashPtr &hash)
-{
- Bytes bytes;
- readRaw(bytes, (const unsigned char*)hash->GetHash(), hash->GetHashBytes());
- Name syncName = m_syncPrefix;
- syncName.appendComp(bytes);
- return syncName;
-}
-
-void
-SyncCore::msgToBytes(const SyncStateMsgPtr &msg, Bytes &bytes)
-{
- int size = msg->ByteSize();
- bytes.resize(size);
- msg->SerializeToArray(head(bytes), size);
-}
-
sqlite3_int64
SyncCore::seq(const Name &name)
{
diff --git a/src/sync-core.h b/src/sync-core.h
index edee04a..bd7c5e8 100644
--- a/src/sync-core.h
+++ b/src/sync-core.h
@@ -15,8 +15,8 @@
* 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>
+ * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
*/
#ifndef SYNC_CORE_H
@@ -25,22 +25,13 @@
#include "sync-log.h"
#include "ccnx-wrapper.h"
#include "scheduler.h"
-#include "interval-generator.h"
#include <boost/function.hpp>
-#include <boost/thread/shared_mutex.hpp>
-
-using namespace std;
-using namespace Ccnx;
class SyncCore
{
public:
typedef boost::function<void (const SyncStateMsgPtr & stateMsg) > StateMsgCallback;
- typedef map<Name, Name> YellowPage;
- typedef boost::shared_mutex Mutex;
- typedef boost::shared_lock<Mutex> ReadLock;
- typedef boost::unique_lock<Mutex> WriteLock;
static const int FRESHNESS = 2; // seconds
static const string RECOVER;
@@ -49,91 +40,77 @@
public:
SyncCore(SyncLogPtr syncLog
- , const Name &userName
- , const Name &localPrefix // routable name used by the local user
- , const Name &syncPrefix // the prefix for the sync collection
+ , const Ccnx::Name &userName
+ , const Ccnx::Name &localPrefix // routable name used by the local user
+ , const Ccnx::Name &syncPrefix // the prefix for the sync collection
, const StateMsgCallback &callback // callback when state change is detected
- , const CcnxWrapperPtr &handle
- , const SchedulerPtr &scheduler);
+ , Ccnx::CcnxWrapperPtr ccnx
+ , SchedulerPtr scheduler);
~SyncCore();
- // some other code should call this fuction when local prefix
- // changes; e.g. when wake up in another network
void
- updateLocalPrefix(const Name &localPrefix);
+ updateLocalPrefix (const Ccnx::Name &localPrefix);
void
- localStateChanged();
-
- Name
- yp(const Name &name);
+ localStateChanged ();
void
- handleInterest(const Name &name);
+ updateLocalState (sqlite3_int64);
+
+// ------------------ only used in test -------------------------
+public:
+ HashPtr
+ root() const { return m_rootHash; }
+
+ sqlite3_int64
+ seq (const Ccnx::Name &name);
+
+private:
+ void
+ handleInterest(const Ccnx::Name &name);
void
- handleSyncData(const Name &name, const Bytes &content);
+ handleSyncData(const Ccnx::Name &name, Ccnx::PcoPtr content);
void
- handleRecoverData(const Name &name, const Bytes &content);
+ handleRecoverData(const Ccnx::Name &name, Ccnx::PcoPtr content);
- Closure::TimeoutCallbackReturnValue
- handleSyncInterestTimeout(const Name &name);
+ Ccnx::Closure::TimeoutCallbackReturnValue
+ handleSyncInterestTimeout(const Ccnx::Name &name);
- Closure::TimeoutCallbackReturnValue
- handleRecoverInterestTimeout(const Name &name);
+ Ccnx::Closure::TimeoutCallbackReturnValue
+ handleRecoverInterestTimeout(const Ccnx::Name &name);
void
- deregister(const Name &name);
+ deregister(const Ccnx::Name &name);
void
recover(const HashPtr &hash);
-// ------------------ only used in test -------------------------
- HashPtr
- root() { return m_rootHash; }
-
- sqlite3_int64
- seq(const Name &name);
-
- void
- updateLocalState(sqlite3_int64);
-
-protected:
+private:
void
sendSyncInterest();
void
- handleSyncInterest(const Name &name);
+ handleSyncInterest(const Ccnx::Name &name);
void
- handleRecoverInterest(const Name &name);
+ handleRecoverInterest(const Ccnx::Name &name);
void
- handleStateData(const Bytes &content);
+ handleStateData(const Ccnx::Bytes &content);
- Name
- constructSyncName(const HashPtr &hash);
+private:
+ Ccnx::CcnxWrapperPtr m_ccnx;
- static void
- msgToBytes(const SyncStateMsgPtr &msg, Bytes &bytes);
-
-protected:
SyncLogPtr m_log;
SchedulerPtr m_scheduler;
StateMsgCallback m_stateMsgCallback;
- Name m_userName;
- Name m_localPrefix;
- Name m_syncPrefix;
+
+ Ccnx::Name m_syncPrefix;
HashPtr m_rootHash;
- YellowPage m_yp;
- Mutex m_ypMutex;
- CcnxWrapperPtr m_handle;
- Closure *m_syncClosure;
- Closure *m_recoverClosure;
IntervalGeneratorPtr m_recoverWaitGenerator;
-
};
#endif // SYNC_CORE_H
diff --git a/src/sync-log.cc b/src/sync-log.cc
index eb421a9..5623a83 100644
--- a/src/sync-log.cc
+++ b/src/sync-log.cc
@@ -20,11 +20,14 @@
*/
#include "sync-log.h"
+#include "logging.h"
#include <utility>
#include <boost/make_shared.hpp>
#include <boost/thread.hpp>
+INIT_LOGGER ("Sync.Log");
+
using namespace boost;
using namespace std;
using namespace Ccnx;
@@ -34,13 +37,75 @@
cout << q << endl;
}
+const std::string INIT_DATABASE = "\
+CREATE TABLE \n\
+ SyncNodes( \n\
+ device_id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
+ device_name BLOB NOT NULL, \n\
+ description TEXT, \n\
+ seq_no INTEGER NOT NULL, \n\
+ last_known_locator BLOB, \n\
+ last_update TIMESTAMP \n\
+ ); \n\
+ \n\
+CREATE TRIGGER SyncNodesUpdater_trigger \n\
+ BEFORE INSERT ON SyncNodes \n\
+ FOR EACH ROW \n\
+ WHEN (SELECT device_id \n\
+ FROM SyncNodes \n\
+ WHERE device_name=NEW.device_name) \n\
+ IS NOT NULL \n\
+ BEGIN \n\
+ UPDATE SyncNodes \n\
+ SET seq_no=max(seq_no,NEW.seq_no) \n\
+ WHERE device_name=NEW.device_name; \n\
+ SELECT RAISE(IGNORE); \n\
+ END; \n\
+ \n\
+CREATE INDEX SyncNodes_device_name ON SyncNodes (device_name); \n\
+ \n\
+CREATE TABLE SyncLog( \n\
+ state_id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
+ state_hash BLOB NOT NULL UNIQUE, \n\
+ last_update TIMESTAMP NOT NULL \n\
+ ); \n\
+ \n\
+CREATE TABLE \n\
+ SyncStateNodes( \n\
+ id INTEGER PRIMARY KEY AUTOINCREMENT, \n\
+ state_id INTEGER NOT NULL \n\
+ REFERENCES SyncLog (state_id) ON UPDATE CASCADE ON DELETE CASCADE, \n\
+ device_id INTEGER NOT NULL \n\
+ REFERENCES SyncNodes (device_id) ON UPDATE CASCADE ON DELETE CASCADE, \n\
+ seq_no INTEGER NOT NULL \n\
+ ); \n\
+ \n\
+CREATE INDEX SyncStateNodes_device_id ON SyncStateNodes (device_id); \n\
+CREATE INDEX SyncStateNodes_state_id ON SyncStateNodes (state_id); \n\
+CREATE INDEX SyncStateNodes_seq_no ON SyncStateNodes (seq_no); \n\
+ \n\
+CREATE TRIGGER SyncLogGuard_trigger \n\
+ BEFORE INSERT ON SyncLog \n\
+ FOR EACH ROW \n\
+ WHEN (SELECT state_hash \n\
+ FROM SyncLog \n\
+ WHERE state_hash=NEW.state_hash) \n\
+ IS NOT NULL \n\
+ BEGIN \n\
+ DELETE FROM SyncLog WHERE state_hash=NEW.state_hash; \n\
+ END; \n\
+";
+
SyncLog::SyncLog (const boost::filesystem::path &path, const std::string &localName)
: DbHelper (path)
, m_localName (localName)
{
+ sqlite3_exec (m_db, INIT_DATABASE.c_str (), NULL, NULL, NULL);
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, sqlite3_errmsg (m_db));
+
UpdateDeviceSeqNo (localName, 0);
-
+
sqlite3_stmt *stmt;
int res = sqlite3_prepare_v2 (m_db, "SELECT device_id, seq_no FROM SyncNodes WHERE device_name=?", -1, &stmt, 0);
@@ -59,25 +124,25 @@
sqlite3_finalize (stmt);
}
-void
-SyncLog::initYP(map<Name, Name> &yp)
-{
- sqlite3_stmt *stmt;
- sqlite3_prepare_v2(m_db, "SELECT device_name, last_known_locator FROM SyncNodes;", -1, &stmt, 0);
+// void
+// SyncLog::initYP(map<Name, Name> &yp)
+// {
+// sqlite3_stmt *stmt;
+// sqlite3_prepare_v2(m_db, "SELECT device_name, last_known_locator FROM SyncNodes;", -1, &stmt, 0);
- while (sqlite3_step(stmt) == SQLITE_ROW)
- {
- Name deviceName((const unsigned char *)sqlite3_column_blob(stmt, 0), sqlite3_column_bytes(stmt, 0));
- Name locator;
- if (sqlite3_column_type(stmt, 1) == SQLITE_BLOB)
- {
- locator = Name((const unsigned char *)sqlite3_column_blob(stmt, 1), sqlite3_column_bytes(stmt, 1));
- }
- yp.insert(make_pair(deviceName, locator));
- }
+// while (sqlite3_step(stmt) == SQLITE_ROW)
+// {
+// Name deviceName((const unsigned char *)sqlite3_column_blob(stmt, 0), sqlite3_column_bytes(stmt, 0));
+// Name locator;
+// if (sqlite3_column_type(stmt, 1) == SQLITE_BLOB)
+// {
+// locator = Name((const unsigned char *)sqlite3_column_blob(stmt, 1), sqlite3_column_bytes(stmt, 1));
+// }
+// yp.insert(make_pair(deviceName, locator));
+// }
- sqlite3_finalize(stmt);
-}
+// sqlite3_finalize(stmt);
+// }
sqlite3_int64
SyncLog::GetNextLocalSeqNo ()
@@ -91,12 +156,12 @@
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str ("Impossible thing in ActionLog::AddActionUpdate"));
}
-
- sqlite3_int64 seq_no = sqlite3_column_int64 (stmt_seq, 0) + 1;
+
+ sqlite3_int64 seq_no = sqlite3_column_int64 (stmt_seq, 0) + 1;
sqlite3_finalize (stmt_seq);
UpdateDeviceSeqNo (m_localDeviceId, seq_no);
-
+
return seq_no;
}
@@ -116,16 +181,17 @@
ORDER BY device_name); \
", 0,0,0);
- std::cout << sqlite3_errmsg (m_db) << std::endl;
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, "DbError: " << sqlite3_errmsg (m_db));
+
if (res != SQLITE_OK)
{
sqlite3_exec (m_db, "ROLLBACK TRANSACTION;", 0,0,0);
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str (sqlite3_errmsg(m_db)));
}
-
+
sqlite3_int64 rowId = sqlite3_last_insert_rowid (m_db);
-
+
sqlite3_stmt *insertStmt;
res += sqlite3_prepare (m_db, "\
INSERT INTO SyncStateNodes \
@@ -137,7 +203,7 @@
res += sqlite3_bind_int64 (insertStmt, 1, rowId);
sqlite3_step (insertStmt);
- // std::cout << sqlite3_errmsg (m_db) << std::endl;
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, "DbError: " << sqlite3_errmsg (m_db));
if (res != SQLITE_OK)
{
sqlite3_exec (m_db, "ROLLBACK TRANSACTION;", 0,0,0);
@@ -145,7 +211,7 @@
<< errmsg_info_str (sqlite3_errmsg(m_db)));
}
sqlite3_finalize (insertStmt);
-
+
sqlite3_stmt *getHashStmt;
res += sqlite3_prepare (m_db, "\
SELECT state_hash FROM SyncLog WHERE state_id = ?\
@@ -163,7 +229,7 @@
{
sqlite3_exec (m_db, "ROLLBACK TRANSACTION;", 0,0,0);
- // std::cout << sqlite3_errmsg (m_db) << std::endl;
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, "DbError: " << sqlite3_errmsg (m_db));
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str ("Not a valid hash in rememberStateInStateLog"));
}
@@ -176,7 +242,7 @@
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str ("Some error with rememberStateInStateLog"));
}
-
+
return retval;
}
@@ -192,7 +258,7 @@
sqlite3_stmt *stmt;
int res = sqlite3_prepare (m_db, "SELECT state_id FROM SyncLog WHERE state_hash = ?",
-1, &stmt, 0);
-
+
if (res != SQLITE_OK)
{
BOOST_THROW_EXCEPTION (Error::Db ()
@@ -223,14 +289,14 @@
{
sqlite3_stmt *stmt;
// update is performed using trigger
- int res = sqlite3_prepare (m_db, "INSERT INTO SyncNodes (device_name, seq_no) VALUES (?,?);",
+ int res = sqlite3_prepare (m_db, "INSERT INTO SyncNodes (device_name, seq_no) VALUES (?,?);",
-1, &stmt, 0);
Ccnx::CcnxCharbufPtr nameBuf = name;
res += sqlite3_bind_blob (stmt, 1, nameBuf->buf (), nameBuf->length (), SQLITE_STATIC);
res += sqlite3_bind_int64 (stmt, 2, seqNo);
sqlite3_step (stmt);
-
+
if (res != SQLITE_OK)
{
BOOST_THROW_EXCEPTION (Error::Db ()
@@ -240,17 +306,23 @@
}
void
+SyncLog::UpdateLocalSeqNo (sqlite3_int64 seqNo)
+{
+ return UpdateDeviceSeqNo (m_localDeviceId, seqNo);
+}
+
+void
SyncLog::UpdateDeviceSeqNo (sqlite3_int64 deviceId, sqlite3_int64 seqNo)
{
sqlite3_stmt *stmt;
// update is performed using trigger
- int res = sqlite3_prepare (m_db, "UPDATE SyncNodes SET seq_no=MAX(seq_no,?) WHERE device_id=?;",
+ int res = sqlite3_prepare (m_db, "UPDATE SyncNodes SET seq_no=MAX(seq_no,?) WHERE device_id=?;",
-1, &stmt, 0);
res += sqlite3_bind_int64 (stmt, 1, seqNo);
res += sqlite3_bind_int64 (stmt, 2, deviceId);
sqlite3_step (stmt);
-
+
if (res != SQLITE_OK)
{
BOOST_THROW_EXCEPTION (Error::Db ()
@@ -303,6 +375,13 @@
sqlite3_finalize(stmt);
}
+void
+SyncLog::UpdateLocalLocator (const Ccnx::Name &locator)
+{
+ return UpdateLocator (m_localName, locator);
+}
+
+
SyncStateMsgPtr
SyncLog::FindStateDifferences (const std::string &oldHash, const std::string &newHash, bool includeOldSeq)
{
@@ -365,7 +444,7 @@
SyncStateMsgPtr msg = make_shared<SyncStateMsg> ();
// sqlite3_trace(m_db, xTrace, NULL);
-
+
while (sqlite3_step (stmt) == SQLITE_ROW)
{
SyncState *state = msg->add_state ();
diff --git a/src/sync-log.h b/src/sync-log.h
index 549d759..1ab25e9 100644
--- a/src/sync-log.h
+++ b/src/sync-log.h
@@ -35,9 +35,17 @@
public:
SyncLog (const boost::filesystem::path &path, const std::string &localName);
- // fill in the map with device-name <-> locator pairs
- void
- initYP(map<Ccnx::Name, Ccnx::Name> &yp);
+ /**
+ * @brief Get local username
+ */
+ inline const Ccnx::Name &
+ GetLocalName () const;
+
+ /**
+ * @brief Get database ID of the local sync node (make sense only for the local database)
+ */
+ inline const sqlite3_int64
+ GetLocalSyncNodeId () const;
sqlite3_int64
GetNextLocalSeqNo (); // side effect: local seq_no will be increased
@@ -46,12 +54,18 @@
void
UpdateDeviceSeqNo (const Ccnx::Name &name, sqlite3_int64 seqNo);
+ void
+ UpdateLocalSeqNo (sqlite3_int64 seqNo);
+
Ccnx::Name
LookupLocator (const Ccnx::Name &deviceName);
void
UpdateLocator (const Ccnx::Name &deviceName, const Ccnx::Name &locator);
+ void
+ UpdateLocalLocator (const Ccnx::Name &locator);
+
// done
/**
* Create an entry in SyncLog and SyncStateNodes corresponding to the current state of SyncNodes
@@ -72,7 +86,7 @@
FindStateDifferences (const std::string &oldHash, const std::string &newHash, bool includeOldSeq = false);
SyncStateMsgPtr
- FindStateDifferences (const Hash &oldHash, const Hash &newHash, bool includeOldSeq = false);
+ FindStateDifferences (const Hash &oldHash, const Hash &newHash, bool includeOldSeq = false);
//-------- only used in test -----------------
sqlite3_int64
@@ -81,21 +95,32 @@
protected:
void
UpdateDeviceSeqNo (sqlite3_int64 deviceId, sqlite3_int64 seqNo);
-
+
protected:
- // std::string m_localName;
Ccnx::Name m_localName;
-
+
sqlite3_int64 m_localDeviceId;
typedef boost::mutex Mutex;
typedef boost::unique_lock<Mutex> WriteLock;
-
- Mutex m_stateUpdateMutex;
+
+ Mutex m_stateUpdateMutex;
};
typedef boost::shared_ptr<SyncLog> SyncLogPtr;
+const Ccnx::Name &
+SyncLog::GetLocalName () const
+{
+ return m_localName;
+}
+
+const sqlite3_int64
+SyncLog::GetLocalSyncNodeId () const
+{
+ return m_localDeviceId;
+}
+
#endif // SYNC_LOG_H
diff --git a/src/sync-state-helper.h b/src/sync-state-helper.h
new file mode 100644
index 0000000..435ef9a
--- /dev/null
+++ b/src/sync-state-helper.h
@@ -0,0 +1,59 @@
+/* -*- 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
+ *
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#ifndef SYNC_STATE_HELPER_H
+#define SYNC_STATE_HELPER_H
+
+#include "sync-state.pb.h"
+
+inline std::ostream &
+operator << (std::ostream &os, const SyncStateMsgPtr &msg)
+{
+ os << " ===== start Msg ======" << std::endl;
+
+ int size = msg->state_size();
+ if (size > 0)
+ {
+ int index = 0;
+ while (index < size)
+ {
+ SyncState state = msg->state(index);
+ string strName = state.name();
+ string strLocator = state.locator();
+ sqlite3_int64 seq = state.seq();
+
+ os << "Name: " << Ccnx::Name((const unsigned char *)strName.c_str(), strName.size())
+ << ", Locator: " << Ccnx::Name((const unsigned char *)strLocator.c_str(), strLocator.size())
+ << ", seq: " << seq << std::endl;
+ index ++;
+ }
+ }
+ else
+ {
+ os << "Msg size 0" << std::endl;
+ }
+ os << " ++++++++ end Msg ++++++++ " << std::endl;
+
+ return os;
+}
+
+
+#endif // SYNC_STATE_HELPER_H
diff --git a/test/test-ccnx-wrapper.cc b/test/test-ccnx-wrapper.cc
index 651692f..d5188de 100644
--- a/test/test-ccnx-wrapper.cc
+++ b/test/test-ccnx-wrapper.cc
@@ -31,9 +31,10 @@
c2->publishData(name, (const unsigned char*)content.c_str(), content.size(), 5);
}
-void dataCallback(const Name &name, const Bytes &content)
+void dataCallback(const Name &name, Ccnx::PcoPtr pco)
{
- string msg((const char*)&content[0], content.size());
+ BytesPtr content = pco->contentPtr ();
+ string msg(reinterpret_cast<const char *> (head (*content)), content->size());
g_dataCallback_counter ++;
BOOST_CHECK_EQUAL(name, msg);
}
@@ -53,7 +54,7 @@
usleep(100000);
c2->setInterestFilter(prefix2, bind(publish2, _1));
- Closure *closure = new Closure(bind(dataCallback, _1, _2), bind(timeout, _1));
+ Closure closure (bind(dataCallback, _1, _2), bind(timeout, _1));
c1->sendInterest(Name("/c2/hi"), closure);
usleep(100000);
@@ -63,13 +64,12 @@
// reset
g_dataCallback_counter = 0;
g_timeout_counter = 0;
- delete closure;
}
BOOST_AUTO_TEST_CASE (CcnxWrapperSelector)
{
- Closure *closure = new Closure(bind(dataCallback, _1, _2), bind(timeout, _1));
+ Closure closure (bind(dataCallback, _1, _2), bind(timeout, _1));
Selectors selectors;
selectors.interestLifetime(1);
@@ -94,7 +94,6 @@
// reset
g_dataCallback_counter = 0;
g_timeout_counter = 0;
- delete closure;
}
BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/test-fetch-manager.cc b/test/test-fetch-manager.cc
new file mode 100644
index 0000000..276fe0c
--- /dev/null
+++ b/test/test-fetch-manager.cc
@@ -0,0 +1,201 @@
+/* -*- 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: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#include "fetch-manager.h"
+#include "fetcher.h"
+#include "ccnx-wrapper.h"
+#include <boost/test/unit_test.hpp>
+#include <boost/make_shared.hpp>
+
+using namespace Ccnx;
+using namespace std;
+using namespace boost;
+
+BOOST_AUTO_TEST_SUITE(TestFetchManager)
+
+struct FetcherTestData
+{
+ set<uint32_t> recvData;
+ set<uint32_t> recvContent;
+
+ set<Name> differentNames;
+ set<Name> segmentNames;
+
+ bool m_done;
+ bool m_failed;
+
+ FetcherTestData ()
+ : m_done (false)
+ , m_failed (false)
+ {
+ }
+
+ void
+ onData (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, Ccnx::PcoPtr pco)
+ {
+ recvData.insert (seqno);
+ differentNames.insert (basename);
+ segmentNames.insert (name);
+
+ BytesPtr data = pco->contentPtr ();
+
+ if (data->size () == sizeof(int))
+ {
+ recvContent.insert (*reinterpret_cast<const int*> (head(*data)));
+ }
+
+ // cout << "<<< " << basename << ", " << name << ", " << seqno << endl;
+ }
+
+ void
+ onComplete (Fetcher &fetcher)
+ {
+ m_done = true;
+ // cout << "Done" << endl;
+ }
+
+ void
+ onFail (Fetcher &fetcher)
+ {
+ m_failed = true;
+ // cout << "Failed" << endl;
+ }
+};
+
+
+BOOST_AUTO_TEST_CASE (TestFetcher)
+{
+ CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
+
+ Name baseName ("/base");
+ /* publish seqnos: 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, <gap 5>, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, <gap 1>, 26 */
+ // this will allow us to test our pipeline of 6
+ for (int i = 0; i < 10; i++)
+ {
+ ccnx->publishData (Name (baseName)(i), reinterpret_cast<const unsigned char*> (&i), sizeof(int), 30);
+ }
+
+ for (int i = 15; i < 25; i++)
+ {
+ ccnx->publishData (Name (baseName)(i), reinterpret_cast<const unsigned char*> (&i), sizeof(int), 30);
+ }
+
+ int oneMore = 26;
+ ccnx->publishData (Name (baseName)(oneMore), reinterpret_cast<const unsigned char*> (&oneMore), sizeof(int), 30);
+
+ FetcherTestData data;
+
+ Fetcher fetcher (ccnx,
+ bind (&FetcherTestData::onData, &data, _1, _2, _3, _4, _5),
+ bind (&FetcherTestData::onComplete, &data, _1),
+ bind (&FetcherTestData::onFail, &data, _1),
+ Name ("/base"), 0, 26,
+ boost::posix_time::seconds (5)); // this time is not precise
+
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), false);
+ fetcher.RestartPipeline ();
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), true);
+
+ usleep(7000000);
+ BOOST_CHECK_EQUAL (data.m_failed, true);
+ BOOST_CHECK_EQUAL (data.differentNames.size (), 1);
+ BOOST_CHECK_EQUAL (data.segmentNames.size (), 20);
+ BOOST_CHECK_EQUAL (data.recvData.size (), 20);
+ BOOST_CHECK_EQUAL (data.recvContent.size (), 20);
+
+ {
+ ostringstream recvData;
+ for (set<uint32_t>::iterator i = data.recvData.begin (); i != data.recvData.end (); i++)
+ recvData << *i << ", ";
+
+ ostringstream recvContent;
+ for (set<uint32_t>::iterator i = data.recvContent.begin (); i != data.recvContent.end (); i++)
+ recvContent << *i << ", ";
+
+ BOOST_CHECK_EQUAL (recvData.str (), "0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, ");
+ BOOST_CHECK_EQUAL (recvData.str (), recvContent.str ());
+ }
+
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), false);
+ fetcher.RestartPipeline ();
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), true);
+
+ usleep(7000000);
+ BOOST_CHECK_EQUAL (data.m_failed, true);
+
+ // publishing missing pieces
+ for (int i = 0; i < 27; i++)
+ {
+ ccnx->publishData (Name (baseName)(i), reinterpret_cast<const unsigned char*> (&i), sizeof(int), 1);
+ }
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), false);
+ fetcher.RestartPipeline ();
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), true);
+
+ usleep(1000000);
+ BOOST_CHECK_EQUAL (data.m_done, true);
+
+ {
+ ostringstream recvData;
+ for (set<uint32_t>::iterator i = data.recvData.begin (); i != data.recvData.end (); i++)
+ recvData << *i << ", ";
+
+ ostringstream recvContent;
+ for (set<uint32_t>::iterator i = data.recvContent.begin (); i != data.recvContent.end (); i++)
+ recvContent << *i << ", ";
+
+ BOOST_CHECK_EQUAL (recvData.str (), "0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, ");
+ BOOST_CHECK_EQUAL (recvData.str (), recvContent.str ());
+ }
+}
+
+// BOOST_AUTO_TEST_CASE (CcnxWrapperSelector)
+// {
+
+// Closure closure (bind(dataCallback, _1, _2), bind(timeout, _1));
+
+// Selectors selectors;
+// selectors.interestLifetime(1);
+
+// string n1 = "/random/01";
+// c1->sendInterest(Name(n1), closure, selectors);
+// sleep(2);
+// c2->publishData(Name(n1), (const unsigned char *)n1.c_str(), n1.size(), 4);
+// usleep(100000);
+// BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+// BOOST_CHECK_EQUAL(g_dataCallback_counter, 0);
+
+// string n2 = "/random/02";
+// selectors.interestLifetime(2);
+// c1->sendInterest(Name(n2), closure, selectors);
+// sleep(1);
+// c2->publishData(Name(n2), (const unsigned char *)n2.c_str(), n2.size(), 4);
+// usleep(100000);
+// BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+// BOOST_CHECK_EQUAL(g_dataCallback_counter, 1);
+
+// // reset
+// g_dataCallback_counter = 0;
+// g_timeout_counter = 0;
+// }
+
+BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/test-sync-core.cc b/test/test-sync-core.cc
index 354e2b4..a7ab3c0 100644
--- a/test/test-sync-core.cc
+++ b/test/test-sync-core.cc
@@ -1,13 +1,17 @@
#include "sync-core.h"
-
+#include "logging.h"
#include <boost/test/unit_test.hpp>
#include <boost/filesystem.hpp>
+#include <boost/make_shared.hpp>
using namespace std;
using namespace Ccnx;
+using namespace boost;
using namespace boost::filesystem;
+INIT_LOGGER("Test.Sync.Core");
+
BOOST_AUTO_TEST_SUITE(SyncCoreTests)
void callback(const SyncStateMsgPtr &msg)
@@ -22,7 +26,7 @@
BOOST_CHECK(state.old_seq() >= 0);
if (state.seq() != 0)
{
- BOOST_CHECK(state.old_seq() != state.seq());
+ BOOST_CHECK(state.old_seq() != state.seq());
}
index++;
}
@@ -35,6 +39,8 @@
BOOST_AUTO_TEST_CASE(SyncCoreTest)
{
+ INIT_LOGGERS();
+
string dir = "./SyncCoreTest";
// clean the test dir
path d(dir);
@@ -55,48 +61,52 @@
SyncLogPtr log1(new SyncLog(dir1, user1.toString()));
SyncLogPtr log2(new SyncLog(dir2, user2.toString()));
- SchedulerPtr scheduler(new Scheduler());
+ // should not have used the same scheduler...
+ SchedulerPtr scheduler1 = make_shared<Scheduler> ();
+ SchedulerPtr scheduler2 = make_shared<Scheduler> ();
-
- SyncCore *core1 = new SyncCore(log1, user1, loc1, syncPrefix, bind(callback, _1), c1, scheduler);
+ SyncCore *core1 = new SyncCore(log1, user1, loc1, syncPrefix, bind(callback, _1), c1, scheduler1);
usleep(10000);
- SyncCore *core2 = new SyncCore(log2, user2, loc2, syncPrefix, bind(callback, _1), c2, scheduler);
+ SyncCore *core2 = new SyncCore(log2, user2, loc2, syncPrefix, bind(callback, _1), c2, scheduler2);
usleep(1000000);
checkRoots(core1->root(), core2->root());
- cout << "\n\n\n\n\n\n----------\n";
+ // _LOG_TRACE ("\n\n\n\n\n\n----------\n");
+
core1->updateLocalState(1);
usleep(100000);
checkRoots(core1->root(), core2->root());
BOOST_CHECK_EQUAL(core2->seq(user1), 1);
- BOOST_CHECK_EQUAL(core2->yp(user1), loc1);
+ BOOST_CHECK_EQUAL(log2->LookupLocator (user1), loc1);
core1->updateLocalState(5);
usleep(100000);
checkRoots(core1->root(), core2->root());
BOOST_CHECK_EQUAL(core2->seq(user1), 5);
- BOOST_CHECK_EQUAL(core2->yp(user1), loc1);
+ BOOST_CHECK_EQUAL(log2->LookupLocator (user1), loc1);
core2->updateLocalState(10);
usleep(100000);
checkRoots(core1->root(), core2->root());
BOOST_CHECK_EQUAL(core1->seq(user2), 10);
- BOOST_CHECK_EQUAL(core1->yp(user2), loc2);
+ BOOST_CHECK_EQUAL(log1->LookupLocator (user2), loc2);
// simple simultaneous data generation
- cout << "\n\n\n\n\n\n----------Simultaneous\n";
+ // _LOG_TRACE ("\n\n\n\n\n\n----------Simultaneous\n");
+ _LOG_TRACE ("Simultaneous");
+
core1->updateLocalState(11);
usleep(100);
core2->updateLocalState(15);
- usleep(1000000);
+ usleep(2000000);
checkRoots(core1->root(), core2->root());
BOOST_CHECK_EQUAL(core1->seq(user2), 15);
BOOST_CHECK_EQUAL(core2->seq(user1), 11);
- BOOST_CHECK_EQUAL(core1->yp(user1), loc1);
- BOOST_CHECK_EQUAL(core1->yp(user2), loc2);
- BOOST_CHECK_EQUAL(core2->yp(user1), loc1);
- BOOST_CHECK_EQUAL(core2->yp(user2), loc2);
+ BOOST_CHECK_EQUAL(log1->LookupLocator (user1), loc1);
+ BOOST_CHECK_EQUAL(log1->LookupLocator (user2), loc2);
+ BOOST_CHECK_EQUAL(log2->LookupLocator (user1), loc1);
+ BOOST_CHECK_EQUAL(log2->LookupLocator (user2), loc2);
// clean the test dir
if (exists(d))
diff --git a/test/database-test.cc b/test/test-sync-log.cc
similarity index 97%
rename from test/database-test.cc
rename to test/test-sync-log.cc
index 3663a41..311f937 100644
--- a/test/database-test.cc
+++ b/test/test-sync-log.cc
@@ -22,6 +22,7 @@
#include <boost/test/unit_test.hpp>
#include <boost/lexical_cast.hpp>
+#include "logging.h"
#include <unistd.h>
#include "action-log.h"
#include <iostream>
@@ -33,11 +34,13 @@
using namespace Ccnx;
namespace fs = boost::filesystem;
-BOOST_AUTO_TEST_SUITE(DatabaseTest)
+BOOST_AUTO_TEST_SUITE(TestSyncLog)
BOOST_AUTO_TEST_CASE (BasicDatabaseTest)
{
+ INIT_LOGGERS ();
+
fs::path tmpdir = fs::unique_path (fs::temp_directory_path () / "%%%%-%%%%-%%%%-%%%%");
SyncLog db (tmpdir, "/alex");
@@ -73,7 +76,7 @@
BOOST_CHECK_EQUAL (msg->state_size(), 1);
BOOST_CHECK_EQUAL (msg->state (0).type (), SyncState::UPDATE);
BOOST_CHECK_EQUAL (msg->state (0).seq (), 2);
-
+
msg = db.FindStateDifferences ("2ff304769cdb0125ac039e6fe7575f8576dceffc62618a431715aaf6eea2bf1c", "00");
BOOST_CHECK_EQUAL (msg->state_size(), 1);
BOOST_CHECK_EQUAL (msg->state (0).type (), SyncState::DELETE);
@@ -89,7 +92,7 @@
BOOST_CHECK_EQUAL (msg->state_size(), 1);
BOOST_CHECK_EQUAL (msg->state (0).type (), SyncState::UPDATE);
BOOST_CHECK_EQUAL (msg->state (0).seq (), 0);
-
+
db.UpdateDeviceSeqNo (Name ("/bob"), 1);
hash = db.RememberStateInStateLog ();
BOOST_CHECK_EQUAL (lexical_cast<string> (*hash), "5df5affc07120335089525e82ec9fda60c6dccd7addb667106fb79de80610519");
diff --git a/wscript b/wscript
index f46e331..ef4c437 100644
--- a/wscript
+++ b/wscript
@@ -9,6 +9,7 @@
opt.add_option('--debug',action='store_true',default=False,dest='debug',help='''debugging mode''')
opt.add_option('--test', action='store_true',default=False,dest='_test',help='''build unit tests''')
opt.add_option('--yes',action='store_true',default=False) # for autoconf/automake/make compatibility
+ opt.add_option('--log4cxx', action='store_true',default=False,dest='log4cxx',help='''Compile with log4cxx logging support''')
opt.load('compiler_cxx boost ccnx protoc ice_cxx qt4')
@@ -31,6 +32,10 @@
if not conf.get_define ("HAVE_SSL"):
conf.fatal ("Cannot find SSL libraries")
+ if conf.options.log4cxx:
+ conf.check_cfg(package='liblog4cxx', args=['--cflags', '--libs'], uselib_store='LOG4CXX', mandatory=True)
+ conf.define ("HAVE_LOG4CXX", 1)
+
conf.load ('ccnx')
conf.load('protoc')
@@ -51,9 +56,10 @@
if conf.options.debug:
conf.define ('_DEBUG', 1)
- conf.env.append_value('CXXFLAGS', ['-O0', '-Wall', '-Wno-unused-variable', '-fcolor-diagnostics', '-g3'])
+ conf.env.append_value('CXXFLAGS', ['-O0', '-Wall', '-Wno-unused-variable',
+ '-fcolor-diagnostics', '-g3', '-Qunused-arguments'])
else:
- conf.env.append_value('CXXFLAGS', ['-O3', '-g'])
+ conf.env.append_value('CXXFLAGS', ['-O3', '-g', '-Qunused-arguments'])
if conf.options._test:
conf.env.TEST = 1
@@ -65,33 +71,33 @@
target = "scheduler",
features = ["cxx"],
source = bld.path.ant_glob(['scheduler/**/*.cc']),
- use = 'BOOST BOOST_THREAD LIBEVENT LIBEVENT_PTHREADS',
- includes = ['scheduler'],
+ use = 'BOOST BOOST_THREAD LIBEVENT LIBEVENT_PTHREADS LOG4CXX',
+ includes = "scheduler",
)
libccnx = bld (
target="ccnx",
features=['cxx'],
source = bld.path.ant_glob(['ccnx/**/*.cc', 'ccnx/**/*.cpp']),
- use = 'BOOST BOOST_THREAD SSL CCNX scheduler',
- includes = ['ccnx', 'scheduler'],
+ use = 'BOOST BOOST_THREAD SSL CCNX LOG4CXX scheduler',
+ includes = "ccnx scheduler",
)
chornoshare = bld (
target="chronoshare",
features=['cxx'],
source = bld.path.ant_glob(['src/**/*.cc', 'src/**/*.cpp', 'src/**/*.proto']),
- use = "BOOST BOOST_FILESYSTEM SQLITE3 scheduler ccnx",
+ use = "BOOST BOOST_FILESYSTEM SQLITE3 LOG4CXX scheduler ccnx",
includes = "ccnx scheduler src",
)
-
+
# Unit tests
if bld.env['TEST']:
unittests = bld.program (
target="unit-tests",
source = bld.path.ant_glob(['test/*.cc']),
features=['cxx', 'cxxprogram'],
- use = 'BOOST_TEST BOOST_FILESYSTEM ccnx database chronoshare',
+ use = 'BOOST_TEST BOOST_FILESYSTEM LOG4CXX ccnx database chronoshare',
includes = "ccnx scheduler src",
)
@@ -101,14 +107,14 @@
defines = "WAF",
source = bld.path.ant_glob(['filesystemwatcher/*.cpp']),
includes = "filesystemwatcher . ",
- use = "QTCORE QTGUI"
+ use = "QTCORE QTGUI LOG4CXX"
)
qt = bld (
target = "chronoshare-gui",
features = "qt4 cxx cxxprogram",
defines = "WAF",
- source = bld.path.ant_glob(['gui/*.cpp']),
+ source = bld.path.ant_glob(['gui/*.cpp', 'gui/*.qrc']),
includes = "gui . ",
- use = "QTCORE QTGUI"
+ use = "QTCORE QTGUI LOG4CXX"
)