//////////////////////////////////////////////////////////////////////////////// /// DISCLAIMER /// /// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany /// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany /// /// Licensed under the Apache License, Version 2.0 (the "License"); /// you may not use this file except in compliance with the License. /// You may obtain a copy of the License at /// /// http://www.apache.org/licenses/LICENSE-2.0 /// /// Unless required by applicable law or agreed to in writing, software /// distributed under the License is distributed on an "AS IS" BASIS, /// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. /// See the License for the specific language governing permissions and /// limitations under the License. /// /// Copyright holder is ArangoDB GmbH, Cologne, Germany /// /// @author Michael Hackstein //////////////////////////////////////////////////////////////////////////////// #include "ClusterTraverser.h" #include "Basics/StaticStrings.h" #include "Basics/VelocyPackHelper.h" #include "Cluster/ClusterComm.h" #include "Cluster/ClusterMethods.h" #include "Graph/BreadthFirstEnumerator.h" #include "Graph/ClusterTraverserCache.h" #include "Graph/TraverserCache.h" #include "Transaction/Helpers.h" #include #include using namespace arangodb; using namespace arangodb::graph; using ClusterTraverser = arangodb::traverser::ClusterTraverser; ClusterTraverser::ClusterTraverser(arangodb::traverser::TraverserOptions* opts, std::unordered_map const* engines, std::string const& dbname, transaction::Methods* trx) : Traverser(opts, trx), _dbname(dbname), _engines(engines) { _opts->linkTraverser(this); } void ClusterTraverser::setStartVertex(std::string const& vid) { _verticesToFetch.clear(); _startIdBuilder.clear(); _startIdBuilder.add(VPackValue(vid)); VPackSlice idSlice = _startIdBuilder.slice(); auto it = _vertices.find(arangodb::velocypack::StringRef(vid)); if (it == _vertices.end()) { size_t firstSlash = vid.find("/"); if (firstSlash == std::string::npos || vid.find("/", firstSlash + 1) != std::string::npos) { // We can stop here. The start vertex is not a valid _id traverserCache()->increaseFilterCounter(); _done = true; return; } } if (!vertexMatchesConditions(arangodb::velocypack::StringRef(vid), 0)) { // Start vertex invalid _done = true; return; } arangodb::velocypack::StringRef persId = traverserCache()->persistString(arangodb::velocypack::StringRef(vid)); _vertexGetter->reset(persId); if (_opts->useBreadthFirst) { _enumerator.reset(new arangodb::graph::BreadthFirstEnumerator(this, idSlice, _opts)); } else { _enumerator.reset(new arangodb::traverser::DepthFirstEnumerator(this, vid, _opts)); } _done = false; } bool ClusterTraverser::getVertex(VPackSlice edge, std::vector& result) { bool res = _vertexGetter->getVertex(edge, result); if (res) { arangodb::velocypack::StringRef const& other = result.back(); if (_vertices.find(other) == _vertices.end()) { // Vertex not yet cached. Prepare it. _verticesToFetch.emplace(other); } } return res; } bool ClusterTraverser::getSingleVertex(arangodb::velocypack::Slice edge, arangodb::velocypack::StringRef const sourceVertexId, uint64_t depth, arangodb::velocypack::StringRef& targetVertexId) { bool res = _vertexGetter->getSingleVertex(edge, sourceVertexId, depth, targetVertexId); if (res) { if (_vertices.find(targetVertexId) == _vertices.end()) { // Vertex not yet cached. Prepare it. _verticesToFetch.emplace(targetVertexId); } } return res; } void ClusterTraverser::fetchVertices() { auto ch = static_cast(traverserCache()); ch->insertedDocuments() += _verticesToFetch.size(); transaction::BuilderLeaser lease(_trx); fetchVerticesFromEngines(_dbname, _engines, _verticesToFetch, _vertices, *(lease.get())); _verticesToFetch.clear(); if (_enumerator != nullptr) { _enumerator->incHttpRequests(_engines->size()); } } aql::AqlValue ClusterTraverser::fetchVertexData(arangodb::velocypack::StringRef idString) { // TRI_ASSERT(idString.isString()); auto cached = _vertices.find(idString); if (cached == _vertices.end()) { // Vertex not yet cached. Prepare for load. _verticesToFetch.emplace(idString); fetchVertices(); cached = _vertices.find(idString); } // Now all vertices are cached!! TRI_ASSERT(cached != _vertices.end()); return aql::AqlValue((*cached).second->data()); } ////////////////////////////////////////////////////////////////////////////// /// @brief Function to add the real data of a vertex into a velocypack builder ////////////////////////////////////////////////////////////////////////////// void ClusterTraverser::addVertexToVelocyPack(arangodb::velocypack::StringRef vid, VPackBuilder& result) { auto cached = _vertices.find(vid); if (cached == _vertices.end()) { // Vertex not yet cached. Prepare for load. _verticesToFetch.emplace(vid); fetchVertices(); cached = _vertices.find(vid); } // Now all vertices are cached!! TRI_ASSERT(cached != _vertices.end()); result.add(VPackSlice((*cached).second->data())); } void ClusterTraverser::destroyEngines() { // We have to clean up the engines in Coordinator Case. auto cc = ClusterComm::instance(); if (cc != nullptr) { // nullptr only happens on controlled server shutdown std::string const url( "/_db/" + arangodb::basics::StringUtils::urlEncode(_trx->vocbase().name()) + "/_internal/traverser/"); if (_enumerator != nullptr) { _enumerator->incHttpRequests(_engines->size()); } for (auto const& it : *_engines) { arangodb::CoordTransactionID coordTransactionID = TRI_NewTickServer(); std::unordered_map headers; auto res = cc->syncRequest(coordTransactionID, "server:" + it.first, RequestType::DELETE_REQ, url + arangodb::basics::StringUtils::itoa(it.second), "", headers, 30.0); if (res->status != CL_COMM_SENT) { // Note If there was an error on server side we do not have // CL_COMM_SENT std::string message("Could not destroy all traversal engines"); if (!res->errorMessage.empty()) { message += std::string(": ") + res->errorMessage; } LOG_TOPIC("8a7a0", ERR, arangodb::Logger::FIXME) << message; } } } }