fixed issue #2012

CHANGELOG

@@ -1,6 +1,8 @@
 devel
 -----
 
+* fixed issue #2012
+
 * added a memory expection in case V8 memory gets too low
 
 * fixed epoch computation in hybrid logical clock

arangod/Wal/CollectorThread.cpp

@@ -278,6 +278,9 @@ int CollectorThread::waitForResult(uint64_t timeout) {
 void CollectorThread::beginShutdown() {
   Thread::beginShutdown();
 
+  // deactivate write-throttling on shutdown
+  _logfileManager->throttleWhenPending(0); 
+
   CONDITION_LOCKER(guard, _condition);
   guard.signal();
 }
@@ -656,6 +659,7 @@ int CollectorThread::processCollectionOperations(CollectorCache* cache) {
               true);  // already locked by guard above
   trx.addHint(TRI_TRANSACTION_HINT_NO_COMPACTION_LOCK,
               true);  // already locked above
+  trx.addHint(TRI_TRANSACTION_HINT_NO_THROTTLING, true);
   trx.addHint(TRI_TRANSACTION_HINT_NO_BEGIN_MARKER, true);
   trx.addHint(TRI_TRANSACTION_HINT_NO_ABORT_MARKER, true);
   trx.addHint(TRI_TRANSACTION_HINT_TRY_LOCK, true);
@@ -1004,12 +1008,15 @@ int CollectorThread::queueOperations(arangodb::wal::Logfile* logfile,
 
   if (maxNumPendingOperations > 0 &&
       _numPendingOperations < maxNumPendingOperations &&
-      (_numPendingOperations + numOperations) >= maxNumPendingOperations) {
+      (_numPendingOperations + numOperations) >= maxNumPendingOperations &&
+      !isStopping()) {
     // activate write-throttling!
     _logfileManager->activateWriteThrottling();
     LOG_TOPIC(WARN, Logger::COLLECTOR)
         << "queued more than " << maxNumPendingOperations
-        << " pending WAL collector operations. now activating write-throttling";
+        << " pending WAL collector operations." 
+        << " current queue size: " << (_numPendingOperations + numOperations) 
+        << ". now activating write-throttling";
   }
 
   _numPendingOperations += numOperations;

arangod/Wal/LogfileManager.cpp

@@ -107,7 +107,7 @@ LogfileManager::LogfileManager(ApplicationServer* server)
       _droppedCollections(),
       _droppedDatabases(),
       _idLock(),
-      _writeThrottled(0),
+      _writeThrottled(false),
       _shutdown(0) {
   LOG(TRACE) << "creating WAL logfile manager";
   TRI_ASSERT(!_allowWrites);
@@ -454,7 +454,15 @@ bool LogfileManager::open() {
   return true;
 }
 
+void LogfileManager::beginShutdown() {
+  throttleWhenPending(0); // deactivate write-throttling on shutdown
+}
+
 void LogfileManager::unprepare() {
+  // deactivate write-throttling (again) on shutdown in case it was set again
+  // after beginShutdown
+  throttleWhenPending(0); 
+  
   _shutdown = 1;
 
   LOG(TRACE) << "shutting down WAL";

arangod/Wal/LogfileManager.h

@@ -116,6 +116,7 @@ class LogfileManager final : public application_features::ApplicationFeature {
   void validateOptions(std::shared_ptr<options::ProgramOptions>) override final;
   void prepare() override final;
   void start() override final;
+  void beginShutdown() override final;
   void unprepare() override final;
 
  public:
@@ -186,13 +187,13 @@ class LogfileManager final : public application_features::ApplicationFeature {
   inline void maxThrottleWait(uint64_t value) { _maxThrottleWait = value; }
 
   // whether or not write-throttling is currently enabled
-  inline bool isThrottled() { return (_writeThrottled != 0); }
+  inline bool isThrottled() { return _writeThrottled; }
 
   // activate write-throttling
-  void activateWriteThrottling() { _writeThrottled = 1; }
+  void activateWriteThrottling() { _writeThrottled = true; }
 
   // deactivate write-throttling
-  void deactivateWriteThrottling() { _writeThrottled = 0; }
+  void deactivateWriteThrottling() { _writeThrottled = false; }
 
   // allow or disallow writes to the WAL
   inline void allowWrites(bool value) { _allowWrites = value; }
@@ -537,7 +538,7 @@ class LogfileManager final : public application_features::ApplicationFeature {
   Mutex _idLock;
 
   // whether or not write-throttling is currently enabled
-  int _writeThrottled;
+  bool _writeThrottled;
 
   // whether or not we have been shut down already
   volatile sig_atomic_t _shutdown;