Merge branch 'devel' of github.com:arangodb/arangodb into devel

arangod/Agency/Node.cpp

@@ -22,6 +22,7 @@
 ////////////////////////////////////////////////////////////////////////////////
 
 #include "Node.h"
+#include "Store.h"
 
 #include "Basics/StringUtils.h"
 
@@ -58,13 +59,20 @@ inline std::vector<std::string> split(const std::string& value, char separator)
 }
 
 // Construct with node name
-Node::Node (std::string const& name) : _node_name(name), _parent(nullptr) {
+Node::Node (std::string const& name) : _node_name(name), _parent(nullptr),
+                                       _store(nullptr) {
   _value.clear();
 }
 
 // Construct with node name in tree structure
 Node::Node (std::string const& name, Node* parent) :
-  _node_name(name), _parent(parent) {
+  _node_name(name), _parent(parent), _store(nullptr) {
+  _value.clear();
+}
+
+// Construct for store
+Node::Node (std::string const& name, Store* store) :
+  _node_name(name), _parent(nullptr), _store(store) {
   _value.clear();
 }
 
@@ -215,6 +223,14 @@ Node& Node::root() {
   return *tmp;
 }
 
+Store& Node::store() {
+  return *(root()._store);
+}
+
+Store const& Node::store() const {
+  return *(root()._store);
+}
+
 // velocypack value type of this node
 ValueType Node::valueType() const {
   return slice().type();
@@ -224,7 +240,7 @@ ValueType Node::valueType() const {
 bool Node::addTimeToLive (long millis) {
   auto tkey = std::chrono::system_clock::now() +
     std::chrono::milliseconds(millis);
-  root()._timeTable.insert(
+  store().timeTable().insert(
     std::pair<TimePoint,std::shared_ptr<Node>>(
       tkey, _parent->_children[_node_name]));
   _ttl = tkey;
@@ -234,10 +250,10 @@ bool Node::addTimeToLive (long millis) {
 // remove time to live entry for this node
 bool Node::removeTimeToLive () {
   if (_ttl != std::chrono::system_clock::time_point()) {
-    auto ret = root()._timeTable.equal_range(_ttl);
+    auto ret = store().timeTable().equal_range(_ttl);
     for (auto it = ret.first; it!=ret.second;) {
       if (it->second == _parent->_children[_node_name]) {
-        root()._timeTable.erase(it);
+        store().timeTable().erase(it);
         break;
       }
       ++it;
@@ -247,7 +263,7 @@ bool Node::removeTimeToLive () {
 }
 
 inline bool Node::observedBy (std::string const& url) const {
-  auto ret = root()._observerTable.equal_range(url);
+  auto ret = store().observerTable().equal_range(url);
   for (auto it = ret.first; it!=ret.second; ++it) {
     if (it->second == uri()) {
       return true;
@@ -406,8 +422,8 @@ template<> bool Node::handle<OBSERVE> (VPackSlice const& slice) {
   
   // check if such entry exists
   if (!observedBy(url)) {
-    root()._observerTable.emplace(std::pair<std::string,std::string>(url,uri));
-    root()._observedTable.emplace(std::pair<std::string,std::string>(uri,url));
+    store().observerTable().emplace(std::pair<std::string,std::string>(url,uri));
+    store().observedTable().emplace(std::pair<std::string,std::string>(uri,url));
     return true;
   }
 
@@ -426,17 +442,17 @@ template<> bool Node::handle<UNOBSERVE> (VPackSlice const& slice) {
 
   // delete in both cases a single entry (ensured above)
   // breaking the iterators is fine then
-  auto ret = root()._observerTable.equal_range(url);
+  auto ret = store().observerTable().equal_range(url);
   for (auto it = ret.first; it!=ret.second; ++it) {
     if (it->second == uri) {
-      root()._observerTable.erase(it);
+      store().observerTable().erase(it);
       break;
     }
   }
-  ret = root()._observedTable.equal_range(uri);
+  ret = store().observedTable().equal_range(uri);
   for (auto it = ret.first; it!=ret.second; ++it) {
     if (it->second == url) {
-      root()._observedTable.erase(it);
+      store().observedTable().erase(it);
       return true;
     }
   }
@@ -555,12 +571,6 @@ std::ostream& Node::print (std::ostream& o) const {
     o << std::endl;
   }
 
-  if (!_timeTable.empty()) {
-    for (auto const& i : _timeTable) {
-      o << i.second.get() << std::endl;
-    }
-  }
-
   return o;
 }
 

arangod/Agency/Node.h

@@ -64,6 +64,8 @@ class Node;
 typedef std::chrono::system_clock::time_point TimePoint;
 typedef std::multimap<TimePoint, std::shared_ptr<Node>> TimeTable;
 
+class Store;
+
 /// @brief Simple tree implementation
 class Node {
   
@@ -80,6 +82,9 @@ public:
   /// @brief Construct with name and introduce to tree under parent
   Node (std::string const& name, Node* parent);
   
+  /// @brief Construct with name and introduce to tree under parent
+  Node (std::string const& name, Store* store);
+
   /// @brief Default dtor
   virtual ~Node ();
   
@@ -157,6 +162,9 @@ public:
   /// @brief Is this node being observed by url
   bool observedBy (std::string const& url) const;
 
+  Store& store();
+  Store const& store() const;
+
 protected:
 
   /// @brief Add time to live entry
@@ -168,17 +176,11 @@ protected:
   std::string _node_name;              /**< @brief my name */
 
   Node* _parent;                       /**< @brief parent */
+  Store* _store;                       /**< @brief Store */
   Children _children;                  /**< @brief child nodes */
   TimePoint _ttl;                      /**< @brief my expiry */
   Buffer<uint8_t> _value;              /**< @brief my value */
 
-  /// @brief Table of expiries in tree (only used in root node)
-  std::multimap<TimePoint, std::shared_ptr<Node>> _timeTable;
-
-  /// @brief Table of observers in tree (only used in root node)
-  std::multimap <std::string,std::string> _observerTable;
-  std::multimap <std::string,std::string> _observedTable;
-  
 };
 
 inline std::ostream& operator<< (std::ostream& o, Node const& n) {

arangod/Agency/Store.cpp

@@ -99,7 +99,7 @@ inline static bool endpointPathFromUrl (
 }
 
 // Create with name
-Store::Store (std::string const& name) : Node(name), Thread(name) {}
+Store::Store (std::string const& name) : Thread(name), _node(name,this) {}
 
 // Default ctor
 Store::~Store () {}
@@ -459,3 +459,53 @@ void Store::run() {
   
 }
 
+
+bool Store::applies (arangodb::velocypack::Slice const& slice) {
+  return _node.applies(slice);
+}
+
+
+void Store::toBuilder (Builder& b) const {
+  _node.toBuilder(b);
+}
+
+Node Store::operator ()(std::vector<std::string> const& pv) {
+  return _node(pv);
+}
+
+Node const Store::operator ()(std::vector<std::string> const& pv) const {
+  return _node(pv);
+}
+  
+
+Node Store::operator ()(std::string const& path) {
+  return _node(path);
+}
+
+Node const Store::operator ()(std::string const& path) const {
+  return _node(path);
+}
+
+
+std::multimap<TimePoint, std::shared_ptr<Node>>& Store::timeTable () {
+  return _timeTable;
+}
+
+const std::multimap<TimePoint, std::shared_ptr<Node>>& Store::timeTable () const {
+  return _timeTable;
+}
+
+std::multimap <std::string,std::string>& Store::observerTable() {
+  return _observerTable;
+}
+std::multimap <std::string,std::string> const& Store::observerTable() const {
+  return _observerTable;
+}
+
+std::multimap <std::string,std::string>& Store::observedTable() {
+  return _observedTable;
+}
+
+std::multimap <std::string,std::string> const& Store::observedTable() const {
+  return _observedTable;
+}

arangod/Agency/Store.h

@@ -36,7 +36,7 @@ namespace consensus {
 class Agent;
 
 /// @brief Key value tree 
-class Store : public Node, public arangodb::Thread {
+class Store : public arangodb::Thread {
   
 public:
 
@@ -76,7 +76,33 @@ public:
   /// @brief See how far the path matches anything in store
   size_t matchPath (std::vector<std::string> const& pv) const;
 
+  /// @brief Get node specified by path vector  
+  Node operator ()(std::vector<std::string> const& pv);
+  /// @brief Get node specified by path vector  
+  Node const operator ()(std::vector<std::string> const& pv) const;
+  
+  /// @brief Get node specified by path string  
+  Node operator ()(std::string const& path);
+  /// @brief Get node specified by path string  
+  Node const operator ()(std::string const& path) const;
+
+  /// @brief Apply single slice
+  bool applies (arangodb::velocypack::Slice const&);
+
+  /// @brief Create Builder representing this store
+  void toBuilder (Builder&) const;
+
+  friend class Node;
+
 private:
+
+  std::multimap<TimePoint, std::shared_ptr<Node>>& timeTable ();
+  std::multimap<TimePoint, std::shared_ptr<Node>> const& timeTable () const;
+  std::multimap <std::string,std::string>& observerTable();
+  std::multimap <std::string,std::string> const& observerTable() const;
+  std::multimap <std::string,std::string>& observedTable();
+  std::multimap <std::string,std::string> const& observedTable() const;
+  
   /// @brief Read individual entry specified in slice into builder
   bool read  (arangodb::velocypack::Slice const&,
               arangodb::velocypack::Builder&) const;
@@ -99,6 +125,16 @@ private:
   /// @brief My own agent
   Agent* _agent;
 
+  /// @brief Table of expiries in tree (only used in root node)
+  std::multimap<TimePoint, std::shared_ptr<Node>> _timeTable;
+  
+  /// @brief Table of observers in tree (only used in root node)
+  std::multimap <std::string,std::string> _observerTable;
+  std::multimap <std::string,std::string> _observedTable;
+
+  /// @brief Root node
+  Node _node;
+
 };
 
 }}

arangod/Aql/AstNode.cpp

@@ -1697,6 +1697,17 @@ bool AstNode::isAttributeAccessForVariable(
   return true;
 }
 
+/// @brief recursively clear flags
+void AstNode::clearFlagsRecursive() {
+  clearFlags();
+  size_t const n = numMembers();
+
+  for (size_t i = 0; i < n; ++i) {
+    auto member = getMemberUnchecked(i);
+    member->clearFlagsRecursive();
+  }
+}
+
 /// @brief whether or not a node is simple enough to be used in a simple
 /// expression
 bool AstNode::isSimple() const {

arangod/Aql/AstNode.h

@@ -319,6 +319,9 @@ struct AstNode {
 
   /// @brief reset flags in case a node is changed drastically
   inline void clearFlags() { flags = 0; }
+  
+  /// @brief recursively clear flags
+  void clearFlagsRecursive();
 
   /// @brief set a flag for the node
   inline void setFlag(AstNodeFlagType flag) const {

arangod/Aql/Expression.cpp

@@ -205,6 +205,7 @@ void Expression::replaceVariableReference(Variable const* variable,
     // must rebuild the expression completely, as it may have changed drastically
     _built = false;
     _type = UNPROCESSED;
+    _node->clearFlagsRecursive(); // recursively delete the node's flags
   }
 
   const_cast<AstNode*>(_node)->clearFlags();
@@ -224,7 +225,7 @@ void Expression::invalidate() {
       _func = nullptr;
       _built = false;
     }
-  }
+  } 
   // we do not need to invalidate the other expression type
   // expression data will be freed in the destructor
 }

arangod/V8Server/V8Traverser.cpp

@@ -473,6 +473,7 @@ bool NeighborsOptions::matchesVertex(std::string const& id) const {
     // Nothing to do
     return true;
   }
+
   std::vector<std::string> parts =
       arangodb::basics::StringUtils::split(id, "/");
   TRI_ASSERT(parts.size() == 2);
@@ -745,6 +746,7 @@ static void AnyNeighbors(std::vector<EdgeCollectionInfo*>& collectionInfos,
                 nextDepth.emplace_back(tmp);
               }
               visited.emplace(std::move(tmp));
+              continue;
             }
             v = edge.get(TRI_VOC_ATTRIBUTE_FROM).getString(l);
             if (visited.find(std::string(v, l)) == visited.end()) {
@@ -782,6 +784,12 @@ void TRI_RunNeighborsSearch(std::vector<EdgeCollectionInfo*>& collectionInfos,
   std::unordered_set<std::string> visited;
   startVertices.emplace_back(opts.start);
   visited.emplace(opts.start);
+  if (!result.empty()) {
+    // We have a continuous search. Mark previous result as visited
+    for (auto const& r : result) {
+      visited.emplace(r);
+    }
+  }
 
   switch (opts.direction) {
     case TRI_EDGE_IN: