Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

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

This commit is contained in:
Jan Steemann 2016-09-21 08:57:03 +02:00
parent 243476e454 049471d91f
commit 06d0b19a70
8 changed files with 157 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
arangod/Cluster/AgencyCallback.cpp
View File

@ -65,6 +65,8 @@ void AgencyCallback::refetchAndUpdate(bool needToAcquireMutex) {
AgencyCommResult result = _agency.getValues(key);
if (!result.successful()) {
LOG(ERR) << "Callback getValues to agency was not successful: "
<< result.errorCode() << " " << result.errorMessage();
return;
}
@ -124,18 +126,11 @@ bool AgencyCallback::execute(std::shared_ptr<VPackBuilder> newData) {
void AgencyCallback::executeByCallbackOrTimeout(double maxTimeout) {
// One needs to acquire the mutex of the condition variable
// before entering this function!
auto compareBuilder = std::make_shared<VPackBuilder>();
if (_lastData) {
compareBuilder = _lastData;
}
if (!_cv.wait(static_cast<uint64_t>(maxTimeout * 1000000.0))) {
if (!_lastData || !_lastData->slice().equals(compareBuilder->slice())) {
LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< "Waiting done and nothing happended. Refetching to be sure";
// mop: watches have not triggered during our sleep...recheck to be sure
refetchAndUpdate(false);
}
LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< "Waiting done and nothing happended. Refetching to be sure";
// mop: watches have not triggered during our sleep...recheck to be sure
refetchAndUpdate(false);
}
}

95
arangod/Cluster/ClusterInfo.cpp
View File

@ -839,7 +839,6 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
*dbServerResult = TRI_ERROR_CLUSTER_COULD_NOT_CREATE_DATABASE;
return true;
}
loadCurrent(); // update our cache
*dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, *errMsg);
}
return true;
@ -897,6 +896,7 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
}
if (*dbServerResult >= 0) {
loadCurrent(); // update our cache
return *dbServerResult;
}
@ -1062,7 +1062,6 @@ int ClusterInfo::createCollectionCoordinator(std::string const& databaseName,
}
}
}
loadCurrent();
if (tmpHaveError) {
*errMsg = "Error in creation of collection:" + tmpMsg;
*dbServerResult = TRI_ERROR_CLUSTER_COULD_NOT_CREATE_COLLECTION;
@ -1120,8 +1119,9 @@ int ClusterInfo::createCollectionCoordinator(std::string const& databaseName,
while (true) {
errorMsg = *errMsg;
if (*dbServerResult >= 0) {
loadCurrent();
return *dbServerResult;
}
@ -1151,37 +1151,21 @@ int ClusterInfo::dropCollectionCoordinator(std::string const& databaseName,
double const realTimeout = getTimeout(timeout);
double const endTime = TRI_microtime() + realTimeout;
double const interval = getPollInterval();
auto dbServerResult = std::make_shared<int>(-1);
auto errMsg = std::make_shared<std::string>();
std::function<bool(VPackSlice const& result)> dbServerChanged =
[=](VPackSlice const& result) {
AgencyComm ac;
if (result.isObject() && result.length() == 0) {
// ...remove the entire directory for the collection
AgencyCommResult res;
res = ac.removeValues(
"Current/Collections/" + databaseName + "/" + collectionID, true);
if (res.successful()) {
*dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, *errMsg);
return true;
}
*dbServerResult = setErrormsg(
TRI_ERROR_CLUSTER_COULD_NOT_REMOVE_COLLECTION_IN_CURRENT,
*errMsg);
return true;
loadCurrent();
*dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, *errMsg);
return true;
}
return true;
};
// monitor the entry for the collection
std::string const where =
"Current/Collections/" + databaseName + "/" + collectionID;
// ATTENTION: The following callback calls the above closure in a
// different thread. Nevertheless, the closure accesses some of our
// local variables. Therefore we have to protect all accesses to them
@ -1203,25 +1187,29 @@ int ClusterInfo::dropCollectionCoordinator(std::string const& databaseName,
AgencyWriteTransaction trans(
{delPlanCollection, incrementVersion},precondition);
res = ac.sendTransactionWithFailover(trans);
// Update our own cache:
loadPlan();
{
CONDITION_LOCKER(locker, agencyCallback->_cv);
while (true) {
errorMsg = *errMsg;
if (*dbServerResult >= 0) {
// ...remove the entire directory for the collection
ac.removeValues(
"Current/Collections/" + databaseName + "/" + collectionID, true);
loadCurrent();
return *dbServerResult;
}
if (TRI_microtime() > endTime) {
return setErrormsg(TRI_ERROR_CLUSTER_TIMEOUT, errorMsg);
}
agencyCallback->executeByCallbackOrTimeout(interval);
}
}
@ -1618,15 +1606,13 @@ int ClusterInfo::ensureIndexCoordinator(
}
resBuilder->add("isNewlyCreated", VPackValue(true));
}
loadCurrent();
*dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, *errMsg);
return true;
}
}
return true;
};
// ATTENTION: The following callback calls the above closure in a
// different thread. Nevertheless, the closure accesses some of our
@ -1644,19 +1630,19 @@ int ClusterInfo::ensureIndexCoordinator(
"Plan/Version", AgencySimpleOperationType::INCREMENT_OP);
AgencyPrecondition oldValue(key, AgencyPrecondition::VALUE, collection);
AgencyWriteTransaction trx ({newValue, incrementVersion}, oldValue);
AgencyCommResult result = ac.sendTransactionWithFailover(trx, 0.0);
if (!result.successful()) {
resultBuilder = *resBuilder;
return setErrormsg(TRI_ERROR_CLUSTER_COULD_NOT_CREATE_COLLECTION_IN_PLAN,
errorMsg);
}
loadPlan();
TRI_ASSERT(*numberOfShards > 0);
{
CONDITION_LOCKER(locker, agencyCallback->_cv);
@ -1664,15 +1650,16 @@ int ClusterInfo::ensureIndexCoordinator(
errorMsg = *errMsg;
resultBuilder = *resBuilder;
if (*dbServerResult >= 0) {
loadCurrent();
return *dbServerResult;
}
if (TRI_microtime() > endTime) {
return setErrormsg(TRI_ERROR_CLUSTER_TIMEOUT, errorMsg);
}
agencyCallback->executeByCallbackOrTimeout(interval);
}
}
@ -1741,7 +1728,7 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
}
VPackObjectIterator shards(current);
if (shards.size() == (size_t)localNumberOfShards) {
bool found = false;
for (auto const& shard : shards) {
@ -1767,13 +1754,12 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
}
if (!found) {
loadCurrent();
*dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, *errMsg);
}
}
return true;
};
// ATTENTION: The following callback calls the above closure in a
// different thread. Nevertheless, the closure accesses some of our
// local variables. Therefore we have to protect all accesses to them
@ -1789,30 +1775,30 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
// and the write lock we acquire below something has changed. Therefore
// we first get the previous value and then do a compare and swap operation.
VPackBuilder tmp;
VPackSlice indexes;
{
std::shared_ptr<LogicalCollection> c =
getCollection(databaseName, collectionID);
READ_LOCKER(readLocker, _planProt.lock);
if (c == nullptr) {
return setErrormsg(TRI_ERROR_ARANGO_COLLECTION_NOT_FOUND, errorMsg);
}
c->getIndexesVPack(tmp, false);
indexes = tmp.slice();
if (!indexes.isArray()) {
// no indexes present, so we can't delete our index
return setErrormsg(TRI_ERROR_ARANGO_INDEX_NOT_FOUND, errorMsg);
}
MUTEX_LOCKER(guard, *numberOfShardsMutex);
*numberOfShards = c->numberOfShards();
}
bool found = false;
VPackBuilder newIndexes;
{
@ -1822,14 +1808,14 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
for (auto const& indexSlice: VPackArrayIterator(indexes)) {
VPackSlice id = indexSlice.get("id");
VPackSlice type = indexSlice.get("type");
if (!id.isString() || !type.isString()) {
continue;
}
if (idString == id.copyString()) {
// found our index, ignore it when copying
found = true;
std::string const typeString = type.copyString();
if (typeString == "primary" || typeString == "edge") {
return setErrormsg(TRI_ERROR_FORBIDDEN, errorMsg);
@ -1842,7 +1828,7 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
if (!found) {
return setErrormsg(TRI_ERROR_ARANGO_INDEX_NOT_FOUND, errorMsg);
}
VPackBuilder newCollectionBuilder;
{
VPackObjectBuilder newCollectionObjectBuilder(&newCollectionBuilder);
@ -1862,15 +1848,15 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
AgencyPrecondition prec(key, AgencyPrecondition::VALUE, previous);
AgencyWriteTransaction trx ({newVal, incrementVersion}, prec);
AgencyCommResult result = ac.sendTransactionWithFailover(trx, 0.0);
if (!result.successful()) {
return setErrormsg(TRI_ERROR_CLUSTER_COULD_NOT_CREATE_COLLECTION_IN_PLAN,
errorMsg);
}
// load our own cache:
loadPlan();
{
MUTEX_LOCKER(guard, *numberOfShardsMutex);
TRI_ASSERT(*numberOfShards > 0);
@ -1882,8 +1868,9 @@ int ClusterInfo::dropIndexCoordinator(std::string const& databaseName,
while (true) {
errorMsg = *errMsg;
if (*dbServerResult >= 0) {
loadCurrent();
return *dbServerResult;
}
@ -1919,7 +1906,7 @@ void ClusterInfo::loadServers() {
AgencyCommResult result = _agency.getValues(prefixServers);
if (result.successful()) {
velocypack::Slice serversRegistered =
result.slice()[0].get(std::vector<std::string>(
{AgencyComm::prefix(), "Current", "ServersRegistered"}));

4
arangod/Cluster/TraverserEngine.cpp
View File

@ -338,3 +338,7 @@ TraverserEngine::~TraverserEngine() {
void TraverserEngine::smartSearch(VPackSlice, VPackBuilder&) {
THROW_ARANGO_EXCEPTION(TRI_ERROR_ONLY_ENTERPRISE);
}
void TraverserEngine::smartSearchBFS(VPackSlice, VPackBuilder&) {
THROW_ARANGO_EXCEPTION(TRI_ERROR_ONLY_ENTERPRISE);
}

8
arangod/Cluster/TraverserEngine.h
View File

@ -75,6 +75,9 @@ class BaseTraverserEngine {
virtual void smartSearch(arangodb::velocypack::Slice,
arangodb::velocypack::Builder&) = 0;
virtual void smartSearchBFS(arangodb::velocypack::Slice,
arangodb::velocypack::Builder&) = 0;
bool lockCollection(std::string const&);
std::shared_ptr<TransactionContext> context() const;
@ -103,7 +106,10 @@ class TraverserEngine : public BaseTraverserEngine {
~TraverserEngine();
void smartSearch(arangodb::velocypack::Slice,
arangodb::velocypack::Builder&);
arangodb::velocypack::Builder&) override;
void smartSearchBFS(arangodb::velocypack::Slice,
arangodb::velocypack::Builder&) override;
};
} // namespace traverser

2
arangod/InternalRestHandler/InternalRestTraverserHandler.cpp
View File

@ -224,6 +224,8 @@ void InternalRestTraverserHandler::queryEngine() {
}
} else if (option == "smartSearch") {
engine->smartSearch(body, result);
} else if (option == "smartSearchBFS") {
engine->smartSearchBFS(body, result);
} else {
// PATH Info wrong other error
generateError(

5
scripts/startLocalCluster.sh
View File

@ -1,9 +1,6 @@
#!/bin/bash
if [ -z "$XTERM" ] ; then
XTERM=x-terminal-emulator
fi
if [ -z "$XTERMOPTIONS" ] ; then
XTERMOPTIONS="--geometry=80x43"
XTERM=xterm
fi
if [ ! -d arangod ] || [ ! -d arangosh ] || [ ! -d UnitTests ] ; then

148
scripts/startStandAloneAgency.sh
View File

@ -1,89 +1,131 @@
#!/bin/bash
function help() {
echo "USAGE: scripts/startStandAloneAgency.sh [options]"
echo ""
echo "OPTIONS:"
echo " -a/--agency-size Agency size (odd integer default: 3))"
echo " -p/--pool-size Pool size (>= agency size default: [agency size])"
echo " -t/--transport Protocol (ssl|tcp default: tcp)"
echo " -l/--log-level Log level (INFO|DEBUG|TRACE default: INFO)"
echo ""
echo "EXAMPLES:"
echo " scripts/startStandaloneAgency.sh"
echo " scripts/startStandaloneAgency.sh -a 5 -p 10 -t ssl"
echo " scripts/startStandaloneAgency.sh --agency-size 3 --pool-size 5"
}
NRAGENTS=3
POOLSZ=""
TRANSPORT="tcp"
LOG_LEVEL="INFO"
while getopts ":a:p:t:l:" opt; do
case $opt in
a) NRAGENTS="$OPTARG"
;;
p) POOLSZ="$OPTARG"
;;
t) TRANSPORT="$OPTARG"
;;
l) LOG_LEVEL="$OPTARG"
;;
\?) echo "Invalid option -$OPTARG" >&2
;;
while [[ ${1} ]]; do
case "${1}" in
-a|--agency-size)
NRAGENTS=${2}
shift
;;
-p|--pool-size)
POOLSZ=${2}
shift
;;
-t|--transport)
TRANSPORT=${2}
shift
;;
-l|--log-level)
LOG_LEVEL=${2}
shift
;;
-h|--help)
help
exit 1
;;
*)
echo "Unknown parameter: ${1}" >&2
help
exit 1
;;
esac
if ! shift; then
echo 'Missing parameter argument.' >&2
return 1
fi
done
if [ "$POOLSZ" == "" ] ; then
POOLSZ=$NRAGENTS
POOLSZ=$NRAGENTS
fi
if [ "$TRANSPORT" == "ssl" ]; then
SSLKEYFILE="--ssl.keyfile UnitTests/server.pem"
SSLKEYFILE="--ssl.keyfile UnitTests/server.pem"
CURL="curl --insecure -ks https://"
else
SSLKEYFILE=""
SSLKEYFILE=""
CURL="curl -s http://"
fi
printf "agency-size: %s\n" "$NRAGENTS"
printf "pool-size: %s\n" "$POOLSZ"
printf "transport: %s\n" "$TRANSPORT"
printf "log-level: %s\n" "$LOG_LEVEL"
printf "Starting agency ... \n"
printf " agency-size: %s," "$NRAGENTS"
printf " pool-size: %s," "$POOLSZ"
printf " transport: %s," "$TRANSPORT"
printf " log-level: %s\n" "$LOG_LEVEL"
if [ ! -d arangod ] || [ ! -d arangosh ] || [ ! -d UnitTests ] ; then
echo Must be started in the main ArangoDB source directory.
exit 1
echo Must be started in the main ArangoDB source directory.
exit 1
fi
if [[ $(( $NRAGENTS % 2 )) == 0 ]]; then
echo Number of agents must be odd.
exit 1
echo Number of agents must be odd.
exit 1
fi
MINP=0.5
MAXP=2.0
SFRE=2.5
COMP=1000
BASE=5001
BASE=5000
rm -rf agency
mkdir -p agency
echo -n "Starting agency ... "
PIDS=""
for aid in `seq 0 $(( $POOLSZ - 1 ))`; do
port=$(( $BASE + $aid ))
build/bin/arangod \
-c none \
--agency.activate true \
--agency.election-timeout-min $MINP \
--agency.election-timeout-max $MAXP \
--agency.endpoint $TRANSPORT://localhost:$BASE \
--agency.my-address $TRANSPORT://localhost:$port \
--agency.compaction-step-size $COMP \
--agency.pool-size $POOLSZ \
--agency.size $NRAGENTS \
--agency.supervision true \
--agency.supervision-frequency $SFRE \
--agency.wait-for-sync false \
--database.directory agency/data$port \
--javascript.app-path ./js/apps \
--javascript.startup-directory ./js \
--javascript.v8-contexts 1 \
--log.file agency/$port.log \
--log.force-direct true \
--log.level agency=$LOG_LEVEL \
--server.authentication false \
--server.endpoint $TRANSPORT://localhost:$port \
--server.statistics false \
$SSLKEYFILE \
> agency/$port.stdout 2>&1 &
port=$(( $BASE + $aid ))
build/bin/arangod \
-c none \
--agency.activate true \
--agency.election-timeout-min $MINP \
--agency.election-timeout-max $MAXP \
--agency.endpoint $TRANSPORT://localhost:$BASE \
--agency.my-address $TRANSPORT://localhost:$port \
--agency.compaction-step-size $COMP \
--agency.pool-size $POOLSZ \
--agency.size $NRAGENTS \
--agency.supervision true \
--agency.supervision-frequency $SFRE \
--agency.wait-for-sync false \
--database.directory agency/data$port \
--javascript.app-path ./js/apps \
--javascript.startup-directory ./js \
--javascript.v8-contexts 1 \
--log.file agency/$port.log \
--log.force-direct true \
--log.level agency=$LOG_LEVEL \
--server.authentication false \
--server.endpoint $TRANSPORT://localhost:$port \
--server.statistics false \
$SSLKEYFILE \
> agency/$port.stdout 2>&1 &
PIDS+=$!
PIDS+=" "
done
echo "done."
echo "Your agents are ready at port $BASE onward"
echo " done. Your agents are ready at port $BASE onward."
#echo "Process ids: $PIDS"
echo "Try ${CURL}localhost:5000/_api/agency/config."

2
utils/jslint.sh
View File

@ -35,7 +35,7 @@ if [ "x$@" == "x" ] ; then
if [ -d ./enterprise ] ; then
echo Considering enterprise files...
JAVASCRIPT_JSLINT="$JAVASCRIPT_JSLINT \
`find ./enterprise/Enterprise/js -name "*.js"` \
`find ./enterprise/js -name "*.js"` \
"
fi
else