added x-arango-start-thread header feature

arangod/Cluster/HeartbeatThread.cpp

@@ -657,7 +657,7 @@ bool HeartbeatThread::syncDBServerStatusQuo() {
           << "could not schedule dbserver sync - dispatcher gone.";
       return false;
     }
-    if (dispatcher->addJob(job) == TRI_ERROR_NO_ERROR) {
+    if (dispatcher->addJob(job, false) == TRI_ERROR_NO_ERROR) {
       LOG_TOPIC(TRACE, Logger::HEARTBEAT) << "scheduled dbserver sync";
       return true;
     }

arangod/Dispatcher/Dispatcher.cpp

@@ -110,7 +110,7 @@ int Dispatcher::addExtraQueue(size_t identifier, size_t nrThreads,
 /// @brief adds a new job
 ////////////////////////////////////////////////////////////////////////////////
 
-int Dispatcher::addJob(std::unique_ptr<Job>& job) {
+int Dispatcher::addJob(std::unique_ptr<Job>& job, bool startThread) {
   job->requestStatisticsAgentSetQueueStart();
 
   // do not start new jobs if we are already shutting down
@@ -133,7 +133,7 @@ int Dispatcher::addJob(std::unique_ptr<Job>& job) {
   LOG(TRACE) << "added job " << (void*)(job.get()) << " to queue '" << qnr << "'";
 
   // add the job to the list of ready jobs
-  return queue->addJob(job);
+  return queue->addJob(job, startThread);
 }
 
 ////////////////////////////////////////////////////////////////////////////////

arangod/Dispatcher/Dispatcher.h

@@ -102,7 +102,7 @@ class Dispatcher {
   /// the response over the network to the caller.
   //////////////////////////////////////////////////////////////////////////////
 
-  int addJob(std::unique_ptr<Job>&);
+  int addJob(std::unique_ptr<Job>&, bool startThread);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief tries to cancel a job

arangod/Dispatcher/DispatcherQueue.cpp

@@ -82,7 +82,7 @@ DispatcherQueue::~DispatcherQueue() {
 /// @brief adds a job
 ////////////////////////////////////////////////////////////////////////////////
 
-int DispatcherQueue::addJob(std::unique_ptr<Job>& job) {
+int DispatcherQueue::addJob(std::unique_ptr<Job>& job, bool startThread) {
   TRI_ASSERT(job.get() != nullptr);
 
   // get next free slot, return false is queue is full
@@ -125,8 +125,8 @@ int DispatcherQueue::addJob(std::unique_ptr<Job>& job) {
   }
 
   // if all threads are blocked, start a new one - we ignore race conditions
-  else if (notEnoughThreads()) {
+  else if (startThread || notEnoughThreads()) {
-    startQueueThread();
+    startQueueThread(startThread);
   }
 
   return TRI_ERROR_NO_ERROR;
@@ -351,7 +351,7 @@ void DispatcherQueue::shutdown() {
 /// @brief starts a new queue thread
 ////////////////////////////////////////////////////////////////////////////////
 
-void DispatcherQueue::startQueueThread() {
+void DispatcherQueue::startQueueThread(bool force) {
   DispatcherThread* thread = (*createDispatcherThread)(this);
 
   if (!_affinityCores.empty()) {
@@ -371,12 +371,17 @@ void DispatcherQueue::startQueueThread() {
   {
     MUTEX_LOCKER(mutexLocker, _threadsLock);
 
-    if (!notEnoughThreads()) {
+    if (!force && !notEnoughThreads()) {
       delete thread;
       return;
     }
 
+    try {
       _startedThreads.insert(thread);
+    } catch (...) {
+      delete thread;
+      return;
+    }
 
     ++_nrRunning;
   }

arangod/Dispatcher/DispatcherQueue.h

@@ -65,7 +65,7 @@ class DispatcherQueue {
   /// @brief adds a job
   //////////////////////////////////////////////////////////////////////////////
 
-  int addJob(std::unique_ptr<Job>&);
+  int addJob(std::unique_ptr<Job>&, bool startThread);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief removes a job
@@ -110,7 +110,7 @@ class DispatcherQueue {
   /// @brief starts a new queue thread
   //////////////////////////////////////////////////////////////////////////////
 
-  void startQueueThread();
+  void startQueueThread(bool force);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief called when a thread has stopped

arangod/HttpServer/HttpCommTask.cpp

@@ -68,6 +68,7 @@ HttpCommTask::HttpCommTask(HttpServer* server, TRI_socket_t socket,
       _acceptDeflate(false),
       _newRequest(true),
       _isChunked(false),
+      _startThread(false),
       _request(nullptr),
       _httpVersion(GeneralRequest::ProtocolVersion::UNKNOWN),
       _requestType(GeneralRequest::RequestType::ILLEGAL),
@@ -115,6 +116,7 @@ HttpCommTask::~HttpCommTask() {
 void HttpCommTask::handleResponse(HttpResponse* response) {
   _requestPending = false;
   _isChunked = false;
+  _startThread = false;
 
   addResponse(response);
 }
@@ -576,6 +578,7 @@ void HttpCommTask::finishedChunked() {
   _writeBuffersStats.push_back(nullptr);
 
   _isChunked = false;
+  _startThread = false;
   _requestPending = false;
 
   fillWriteBuffer();
@@ -848,6 +851,13 @@ void HttpCommTask::processRequest() {
   
   handler->setTaskId(_taskId, _loop);
   
+  std::string const& startThread = _request->header(StaticStrings::StartThread, found);
+
+  if (found) {
+    _startThread = StringUtils::boolean(startThread);
+  }
+
+
   // clear request object
   _request = nullptr;
 
@@ -949,6 +959,7 @@ void HttpCommTask::resetState(bool close) {
 
   _newRequest = true;
   _readRequestBody = false;
+  _startThread = false;
 }
 
 ////////////////////////////////////////////////////////////////////////////////

arangod/HttpServer/HttpCommTask.h

@@ -71,6 +71,12 @@ class HttpCommTask : public SocketTask, public RequestStatisticsAgent {
   ~HttpCommTask();
 
  public:
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief return whether or not the task desires to start a dispatcher thread
+  //////////////////////////////////////////////////////////////////////////////
+  
+  bool startThread() const { return _startThread; }
+
   //////////////////////////////////////////////////////////////////////////////
   /// @brief handles response
   //////////////////////////////////////////////////////////////////////////////
@@ -258,6 +264,12 @@ class HttpCommTask : public SocketTask, public RequestStatisticsAgent {
 
   bool _isChunked;
   
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief start a separate thread if the task is added to the dispatcher?
+  //////////////////////////////////////////////////////////////////////////////
+
+  bool _startThread;
+
   //////////////////////////////////////////////////////////////////////////////
   /// @brief the request with possible incomplete body
   //////////////////////////////////////////////////////////////////////////////

arangod/HttpServer/HttpServer.cpp

@@ -220,7 +220,7 @@ bool HttpServer::handleRequestAsync(HttpCommTask* task,
   }
 
   // execute the handler using the dispatcher
-  int res = _dispatcher->addJob(job);
+  int res = _dispatcher->addJob(job, task->startThread());
 
   // could not add job to job queue
   if (res != TRI_ERROR_NO_ERROR) {
@@ -260,7 +260,8 @@ bool HttpServer::handleRequest(HttpCommTask* task,
              << (void*)job.get();
 
   // add the job to the dispatcher
-  int res = _dispatcher->addJob(job);
+
+  int res = _dispatcher->addJob(job, task->startThread());
 
   // job is in queue now
   return res == TRI_ERROR_NO_ERROR;

arangod/V8Server/V8PeriodicTask.cpp

@@ -89,7 +89,7 @@ bool V8PeriodicTask::handlePeriod() {
     return false;
   }
   
-  DispatcherFeature::DISPATCHER->addJob(job);
+  DispatcherFeature::DISPATCHER->addJob(job, false);
 
   return true;
 }

arangod/V8Server/V8TimerTask.cpp

@@ -90,7 +90,7 @@ bool V8TimerTask::handleTimeout() {
     return false;
   }
   
-  int res = DispatcherFeature::DISPATCHER->addJob(job);
+  int res = DispatcherFeature::DISPATCHER->addJob(job, false);
 
   if (res != TRI_ERROR_NO_ERROR) {
     LOG(WARN) << "could not add task " << _command << " to queue";

lib/Basics/StaticStrings.cpp

@@ -30,6 +30,7 @@ std::string const StaticStrings::Binary("binary");
 std::string const StaticStrings::Empty("");
 std::string const StaticStrings::N1800("1800");
 
+
 // system attribute names
 std::string const StaticStrings::IdString("_id");
 std::string const StaticStrings::KeyString("_key");
@@ -74,6 +75,7 @@ std::string const StaticStrings::OmitWwwAuthenticate("x-omit-www-authenticate");
 std::string const StaticStrings::Origin("origin");
 std::string const StaticStrings::Queue("x-arango-queue");
 std::string const StaticStrings::Server("server");
+std::string const StaticStrings::StartThread("x-arango-start-thread");
 std::string const StaticStrings::WwwAuthenticate("www-authenticate");
     
 

lib/Basics/StaticStrings.h

@@ -80,6 +80,7 @@ class StaticStrings {
   static std::string const Origin;
   static std::string const Queue;
   static std::string const Server;
+  static std::string const StartThread;
   static std::string const WwwAuthenticate;
 
   // mime types