logic: reduced waiting time when interrupting indexer processes

* made multiprocess AST traversal interruptible again
* removed const variables that define sleep times for threads

src/lib/CMakeLists.txt

@@ -172,6 +172,7 @@ add_files(
 	data/indexer/IndexerCommandType.h
 	data/indexer/IndexerComposite.cpp
 	data/indexer/IndexerComposite.h
+	data/indexer/IndexerStateInfo.h
 	data/indexer/MemoryIndexerCommandProvider.cpp
 	data/indexer/MemoryIndexerCommandProvider.h
 	data/indexer/TaskBuildIndex.cpp
@@ -510,8 +511,6 @@ add_files(
 	utility/messaging/MessageBase.cpp
 	utility/messaging/MessageBase.h
 	utility/messaging/MessageFilter.h
-	utility/messaging/MessageInterruptTasksCounter.cpp
-	utility/messaging/MessageInterruptTasksCounter.h
 	utility/messaging/MessageListener.h
 	utility/messaging/MessageListenerBase.cpp
 	utility/messaging/MessageListenerBase.h

src/lib/data/TaskInjectStorage.cpp

@@ -35,8 +35,7 @@ Task::TaskState TaskInjectStorage::doUpdate(std::shared_ptr<Blackboard> blackboa
 	}
 	else
 	{
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 
 	return STATE_FAILURE;

src/lib/data/TaskMergeStorages.cpp

@@ -42,8 +42,7 @@ Task::TaskState TaskMergeStorages::doUpdate(std::shared_ptr<Blackboard> blackboa
 	}
 	else
 	{
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 
 	return STATE_FAILURE;

src/lib/data/indexer/Indexer.h

@@ -13,12 +13,13 @@ class Indexer
 {
 public:
 	IndexerCommandType getSupportedIndexerCommandType() const override;
-
 	std::shared_ptr<IntermediateStorage> index(std::shared_ptr<IndexerCommand> indexerCommand) override;
 
+private:
 	virtual std::shared_ptr<IntermediateStorage> doIndex(std::shared_ptr<T> indexerCommand) = 0;
 };
 
+
 template <typename T>
 IndexerCommandType Indexer<T>::getSupportedIndexerCommandType() const
 {

src/lib/data/indexer/IndexerBase.cpp

@@ -1,16 +1,5 @@
 #include "IndexerBase.h"
 
 IndexerBase::IndexerBase()
-	: m_interrupted(false)
 {
 }
-
-void IndexerBase::interrupt()
-{
-	m_interrupted = true;
-}
-
-bool IndexerBase::interrupted() const
-{
-	return m_interrupted;
-}

src/lib/data/indexer/IndexerBase.h

@@ -15,17 +15,9 @@ class IndexerBase
 public:
 	IndexerBase();
 	virtual ~IndexerBase() = default;
-
 	virtual IndexerCommandType getSupportedIndexerCommandType() const = 0;
-
 	virtual std::shared_ptr<IntermediateStorage> index(std::shared_ptr<IndexerCommand> indexerCommand) = 0;
-
-	virtual void interrupt();
-
-	bool interrupted() const;
-
-private:
-	bool m_interrupted;
+	virtual void interrupt() = 0;
 };
 
 #endif // INDEXER_BASE_H

src/lib/data/indexer/IndexerComposite.cpp

@@ -36,5 +36,4 @@ void IndexerComposite::interrupt()
 	{
 		it.second->interrupt();
 	}
-	IndexerBase::interrupt();
 }

src/lib/data/indexer/IndexerComposite.h

@@ -11,13 +11,13 @@ class IndexerComposite: public IndexerBase
 public:
 	virtual ~IndexerComposite();
 
-	virtual IndexerCommandType getSupportedIndexerCommandType() const;
+	IndexerCommandType getSupportedIndexerCommandType() const override;
 
 	void addIndexer(std::shared_ptr<IndexerBase> indexer);
 
-	virtual std::shared_ptr<IntermediateStorage> index(std::shared_ptr<IndexerCommand> indexerCommand);
+	std::shared_ptr<IntermediateStorage> index(std::shared_ptr<IndexerCommand> indexerCommand) override;
 
-	virtual void interrupt();
+	void interrupt() override;
 
 private:
 	std::map<IndexerCommandType, std::shared_ptr<IndexerBase>> m_indexers;

src/lib/data/indexer/IndexerStateInfo.h

@@ -0,0 +1,10 @@
+#ifndef INDEXER_STATE_INFO_H
+#define INDEXER_STATE_INFO_H
+
+struct IndexerStateInfo
+{
+public:
+	bool indexingInterrupted;
+};
+
+#endif // INDEXER_STATE_INFO_H

src/lib/data/indexer/TaskBuildIndex.cpp

@@ -41,6 +41,8 @@ TaskBuildIndex::TaskBuildIndex(
 
 void TaskBuildIndex::doEnter(std::shared_ptr<Blackboard> blackboard)
 {
+	m_interprocessIndexingStatusManager.setIndexingInterrupted(false);
+
 	m_indexingFileCount = 0;
 	updateIndexingDialog(blackboard, std::vector<FilePath>());
 
@@ -104,8 +106,7 @@ Task::TaskState TaskBuildIndex::doUpdate(std::shared_ptr<Blackboard> blackboard)
 		updateIndexingDialog(blackboard, std::vector<FilePath>());
 	}
 
-	const int SLEEP_TIME_MS = 50;
-	std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+	std::this_thread::sleep_for(std::chrono::milliseconds(50));
 
 	return STATE_RUNNING;
 }
@@ -154,6 +155,8 @@ void TaskBuildIndex::handleMessage(MessageInterruptTasks* message)
 {
 	if (!m_dialogView->dialogsHidden())
 	{
+		LOG_INFO("sending indexer interrupt command.");
+		m_interprocessIndexingStatusManager.setIndexingInterrupted(true);
 		m_interrupted = true;
 	}
 }
@@ -232,8 +235,7 @@ bool TaskBuildIndex::fetchIntermediateStorages(std::shared_ptr<Blackboard> black
 	{
 		LOG_INFO_STREAM(<< "waiting, too many storages queued: " << providerStorageCount);
 
-		const int SLEEP_TIME_MS = 100;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(100));
 
 		return true;
 	}

src/lib/data/indexer/TaskFillIndexerCommandQueue.cpp

@@ -82,8 +82,7 @@ Task::TaskState TaskFillIndexerCommandsQueue::doUpdate(std::shared_ptr<Blackboar
 		}
 	}
 
-	const int SLEEP_TIME_MS = 200;
-	std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+	std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
 	return STATE_RUNNING;
 }

src/lib/data/indexer/interprocess/InterprocessIndexer.cpp

@@ -5,6 +5,7 @@
 #include "FileRegister.h"
 #include "logging.h"
 #include "LanguagePackageManager.h"
+#include "ScopedFunctor.h"
 
 InterprocessIndexer::InterprocessIndexer(const std::string& uuid, Id processId)
 	: m_interprocessIndexerCommandManager(uuid, processId, false)
@@ -17,10 +18,42 @@ InterprocessIndexer::InterprocessIndexer(const std::string& uuid, Id processId)
 
 void InterprocessIndexer::work()
 {
+	bool updaterThreadRunning = false;
+	std::shared_ptr<std::thread> updaterThread;
+	std::shared_ptr<IndexerBase> indexer;
+
 	try
 	{
 		LOG_INFO(std::to_wstring(m_processId) + L" starting up indexer");
-		std::shared_ptr<IndexerBase> indexer = LanguagePackageManager::getInstance()->instantiateSupportedIndexers();
+		indexer = LanguagePackageManager::getInstance()->instantiateSupportedIndexers();
+
+		updaterThread = std::make_shared<std::thread>([&]()
+		{
+			updaterThreadRunning = true;
+			while (updaterThreadRunning)
+			{
+				std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+				if (m_interprocessIndexingStatusManager.getIndexingInterrupted())
+				{
+					LOG_INFO("received indexer interrupt command.");
+					if (indexer)
+					{
+						indexer->interrupt();
+					}
+					updaterThreadRunning = false;
+				}
+			}
+		});
+
+		ScopedFunctor threadStopper([&]()
+		{
+			updaterThreadRunning = false;
+			if (updaterThread)
+			{
+				updaterThread->join();
+				updaterThread.reset();
+			}
+		});
 
 		while (std::shared_ptr<IndexerCommand> indexerCommand = m_interprocessIndexerCommandManager.popIndexerCommand())
 		{
@@ -29,7 +62,7 @@ void InterprocessIndexer::work()
 
 			while (true)
 			{
-				size_t storageCount = m_interprocessIntermediateStorageManager.getIntermediateStorageCount();
+				const size_t storageCount = m_interprocessIntermediateStorageManager.getIntermediateStorageCount();
 				if (storageCount < 10)
 				{
 					break;
@@ -37,8 +70,7 @@ void InterprocessIndexer::work()
 
 				LOG_INFO_STREAM(<< m_processId << " waits, too many intermediate storages: " << storageCount);
 
-				const int SLEEP_TIME_MS = 200;
-				std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+				std::this_thread::sleep_for(std::chrono::milliseconds(200));
 			}
 
 			LOG_INFO_STREAM(<< m_processId << " updating indexer status with currently indexed filepath");
@@ -47,8 +79,11 @@ void InterprocessIndexer::work()
 			LOG_INFO_STREAM(<< m_processId << " starting to index current file");
 			std::shared_ptr<IntermediateStorage> result = indexer->index(indexerCommand);
 
-			LOG_INFO_STREAM(<< m_processId << " pushing index to shared memory");
-			m_interprocessIntermediateStorageManager.pushIntermediateStorage(result);
+			if (result)
+			{
+				LOG_INFO_STREAM(<< m_processId << " pushing index to shared memory");
+				m_interprocessIntermediateStorageManager.pushIntermediateStorage(result);
+			}
 
 			LOG_INFO_STREAM(<< m_processId << " finalizing indexer status for current file");
 			m_interprocessIndexingStatusManager.finishIndexingSourceFile();

src/lib/data/indexer/interprocess/InterprocessIndexingStatusManager.cpp

@@ -9,6 +9,7 @@ const char* InterprocessIndexingStatusManager::s_indexingFilesKeyName = "indexin
 const char* InterprocessIndexingStatusManager::s_currentFilesKeyName = "current_files";
 const char* InterprocessIndexingStatusManager::s_crashedFilesKeyName = "crashed_files";
 const char* InterprocessIndexingStatusManager::s_finishedProcessIdsKeyName = "finished_process_ids";
+const char* InterprocessIndexingStatusManager::s_indexingInterruptedKeyName = "indexing_interrupted_flag";
 
 InterprocessIndexingStatusManager::InterprocessIndexingStatusManager(const std::string& instanceUuid, Id processId, bool isOwner)
 	: BaseInterprocessDataManager(s_sharedMemoryNamePrefix + instanceUuid, 1048576 /* 1 MB */, instanceUuid, processId, isOwner)
@@ -97,6 +98,32 @@ void InterprocessIndexingStatusManager::finishIndexingSourceFile()
 	}
 }
 
+void InterprocessIndexingStatusManager::setIndexingInterrupted(bool interrupted)
+{
+	SharedMemory::ScopedAccess access(&m_sharedMemory);
+
+	bool* indexingInterruptedPtr =
+		access.accessValue<bool>(s_indexingInterruptedKeyName);
+	if (indexingInterruptedPtr)
+	{
+		*indexingInterruptedPtr = interrupted;
+	}
+}
+
+bool InterprocessIndexingStatusManager::getIndexingInterrupted()
+{
+	SharedMemory::ScopedAccess access(&m_sharedMemory);
+
+	bool* indexingInterruptedPtr =
+		access.accessValue<bool>(s_indexingInterruptedKeyName);
+	if (indexingInterruptedPtr)
+	{
+		return *indexingInterruptedPtr;
+	}
+
+	return false;
+}
+
 Id InterprocessIndexingStatusManager::getNextFinishedProcessId()
 {
 	SharedMemory::ScopedAccess access(&m_sharedMemory);

src/lib/data/indexer/interprocess/InterprocessIndexingStatusManager.h

@@ -16,6 +16,9 @@ public:
 	void startIndexingSourceFile(const FilePath& filePath);
 	void finishIndexingSourceFile();
 
+	void setIndexingInterrupted(bool interrupted);
+	bool getIndexingInterrupted();
+
 	Id getNextFinishedProcessId();
 
 	std::vector<FilePath> getCurrentlyIndexedSourceFilePaths();
@@ -28,6 +31,7 @@ private:
 	static const char* s_currentFilesKeyName;
 	static const char* s_crashedFilesKeyName;
 	static const char* s_finishedProcessIdsKeyName;
+	static const char* s_indexingInterruptedKeyName;
 };
 
 #endif // INTERPROCESS_INDEXING_STATUS_MANAGER_H

src/lib/utility/interprocess/SharedMemory.h

@@ -116,5 +116,4 @@ private:
 	AccessMode m_mode;
 };
 
-
 #endif // SHARED_MEMORY_H

src/lib/utility/messaging/MessageInterruptTasksCounter.cpp

@@ -1,40 +0,0 @@
-#include "MessageInterruptTasksCounter.h"
-
-#include "MessageInterruptTasks.h"
-#include "MessageListener.h"
-
-MessageInterruptTasksCounter::MessageInterruptTasksCounter()
-	: m_count(0)
-{
-	class InterruptListener: public MessageListener<MessageInterruptTasks>
-	{
-	public:
-		InterruptListener(size_t& counter)
-			: m_counter(counter)
-		{}
-
-	private:
-		virtual void handleMessage(MessageInterruptTasks* message)
-		{
-			m_counter++;
-		}
-
-		size_t& m_counter;
-	};
-
-	m_listener = std::make_shared<InterruptListener>(m_count);
-}
-
-MessageInterruptTasksCounter::~MessageInterruptTasksCounter()
-{
-}
-
-void MessageInterruptTasksCounter::reset()
-{
-	m_count = 0;
-}
-
-size_t MessageInterruptTasksCounter::getCount() const
-{
-	return m_count;
-}

src/lib/utility/messaging/MessageInterruptTasksCounter.h

@@ -1,23 +0,0 @@
-#ifndef MESSAGE_INTERRUPT_TASKS_COUNTER_H
-#define MESSAGE_INTERRUPT_TASKS_COUNTER_H
-
-#include <memory>
-
-class MessageListenerBase;
-
-class MessageInterruptTasksCounter
-{
-public:
-	MessageInterruptTasksCounter();
-	virtual ~MessageInterruptTasksCounter();
-
-	void reset();
-	size_t getCount() const;
-
-private:
-	std::shared_ptr<MessageListenerBase> m_listener;
-
-	size_t m_count;
-};
-
-#endif // MESSAGE_INTERRUPT_TASKS_COUNTER_H

src/lib/utility/messaging/MessageQueue.cpp

@@ -140,8 +140,7 @@ void MessageQueue::startMessageLoop()
 			}
 		}
 
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 
 	{
@@ -176,8 +175,7 @@ void MessageQueue::stopMessageLoop()
 			}
 		}
 
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 }
 

src/lib/utility/scheduling/TaskGroupParallel.cpp

@@ -35,8 +35,7 @@ void TaskGroupParallel::doEnter(std::shared_ptr<Blackboard> blackboard)
 
 Task::TaskState TaskGroupParallel::doUpdate(std::shared_ptr<Blackboard> blackboard)
 {
-	const int SLEEP_TIME_MS = 25;
-	std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+	std::this_thread::sleep_for(std::chrono::milliseconds(25));
 
 	if (m_tasks.size() != 0 && getActiveTaskCount() > 0)
 	{

src/lib/utility/scheduling/TaskScheduler.cpp

@@ -71,8 +71,7 @@ void TaskScheduler::startSchedulerLoop()
 			}
 		}
 
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 
 	{
@@ -107,8 +106,7 @@ void TaskScheduler::stopSchedulerLoop()
 			}
 		}
 
-		const int SLEEP_TIME_MS = 25;
-		std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP_TIME_MS));
+		std::this_thread::sleep_for(std::chrono::milliseconds(25));
 	}
 }