logic: Improvements in indexing to avoid OOM (issue #471)

* indexer processes wait if already too many storages queued
* storage fetcher waits if already too many storages queued
* increased size estimation for shared memory of indexer commands
* Added MessageFilterNewErrors to combine messages for new errors

bug id = 471

setup/Linux/Sourcetrail.sh

@@ -1,5 +1,3 @@
-#!/bin/sh
-
 # get the current directory
 SOURCE="${0}"
 while [ -h "$SOURCE" ]; do # resolve $SOURCE until the file is no longer a symlink

src/lib/Application.cpp

@@ -4,6 +4,7 @@
 #include "utility/logging/logging.h"
 #include "utility/logging/LogManager.h"
 #include "utility/messaging/filter_types/MessageFilterFocusInOut.h"
+#include "utility/messaging/filter_types/MessageFilterNewErrors.h"
 #include "utility/messaging/filter_types/MessageFilterSearchAutocomplete.h"
 #include "utility/messaging/MessageQueue.h"
 #include "utility/messaging/type/MessageQuitApplication.h"
@@ -338,6 +339,7 @@ void Application::startMessagingAndScheduling()
 
 	MessageQueue* queue = MessageQueue::getInstance().get();
 	queue->addMessageFilter(std::make_shared<MessageFilterFocusInOut>());
+	queue->addMessageFilter(std::make_shared<MessageFilterNewErrors>());
 	queue->addMessageFilter(std::make_shared<MessageFilterSearchAutocomplete>());
 
 	queue->setSendMessagesAsTasks(true);

src/lib/CMakeLists.txt

@@ -363,6 +363,7 @@ add_files(
 	utility/math/VectorBase.h
 
 	utility/messaging/filter_types/MessageFilterFocusInOut.h
+	utility/messaging/filter_types/MessageFilterNewErrors.h
 	utility/messaging/filter_types/MessageFilterSearchAutocomplete.h
 
 	utility/messaging/type/MessageActivateAll.h

src/lib/data/indexer/TaskBuildIndex.cpp

@@ -228,6 +228,17 @@ bool TaskBuildIndex::fetchIntermediateStorages(std::shared_ptr<Blackboard> black
 {
 	int poppedStorageCount = 0;
 
+	int providerStorageCount = m_storageProvider->getStorageCount();
+	if (providerStorageCount > 10)
+	{
+		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));
+
+		return true;
+	}
+
 	TimeStamp t = TimeStamp::now();
 	do
 	{

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

@@ -33,6 +33,20 @@ void InterprocessIndexer::work()
 			LOG_INFO_STREAM(<< m_processId << " fetched indexer command for \"" << indexerCommand->getSourceFilePath().str() << "\"");
 			LOG_INFO_STREAM(<< m_processId << " indexer commands left: " << (m_interprocessIndexerCommandManager.indexerCommandCount() + 1));
 
+			while (true)
+			{
+				size_t storageCount = m_interprocessIntermediateStorageManager.getIntermediateStorageCount();
+				if (storageCount < 10)
+				{
+					break;
+				}
+
+				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));
+			}
+
 			LOG_INFO_STREAM(<< m_processId << " updating indexer status with currently indexed filepath");
 			m_interprocessIndexingStatusManager.startIndexingSourceFile(indexerCommand->getSourceFilePath());
 

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

@@ -19,7 +19,7 @@ InterprocessIndexerCommandManager::~InterprocessIndexerCommandManager()
 void InterprocessIndexerCommandManager::setIndexerCommands(
 	const std::vector<std::shared_ptr<IndexerCommand>>& indexerCommands)
 {
-	const unsigned int overestimationMultiplier = 2;
+	const unsigned int overestimationMultiplier = 3;
 
 	size_t estimatedSize = 1048576; /* 1 MB */
 	for (auto& command : indexerCommands)

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

@@ -39,7 +39,7 @@ void InterprocessIndexingStatusManager::startIndexingSourceFile(const FilePath&
 		SharedMemory::Map<Id, SharedMemory::String>::iterator it = currentFilesPtr->find(getProcessId());
 		if (it != currentFilesPtr->end())
 		{
-			const int overestimationMultiplier = 2;
+			const int overestimationMultiplier = 3;
 			const std::string crashedFilePath = it->second.c_str();
 
 			size_t estimatedSize = 262144 + sizeof(std::string) + crashedFilePath.size();
@@ -186,7 +186,7 @@ std::set<FilePath> InterprocessIndexingStatusManager::getIndexedFiles()
 
 void InterprocessIndexingStatusManager::addIndexedFiles(std::set<FilePath> filePaths)
 {
-	const unsigned int overestimationMultiplier = 2;
+	const unsigned int overestimationMultiplier = 3;
 
 	SharedMemory::ScopedAccess access(&m_sharedMemory);
 

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

@@ -28,7 +28,7 @@ void InterprocessIntermediateStorageManager::pushIntermediateStorage(
 	const std::shared_ptr<IntermediateStorage>& intermediateStorage)
 {
     const unsigned int overestimationMultiplier = 3;
-    size_t size = (intermediateStorage->getByteSize() + sizeof(SharedIntermediateStorage))* overestimationMultiplier + 1048576/* 1 MB */;
+    size_t size = (intermediateStorage->getByteSize() + sizeof(SharedIntermediateStorage)) * overestimationMultiplier + 1048576/* 1 MB */;
 
 	SharedMemory::ScopedAccess access(&m_sharedMemory);
 

src/lib/utility/messaging/filter_types/MessageFilterNewErrors.h

@@ -0,0 +1,41 @@
+#ifndef MESSAGE_FILTER_NEW_ERRORS_H
+#define MESSAGE_FILTER_NEW_ERRORS_H
+
+#include "utility/messaging/MessageFilter.h"
+#include "utility/messaging/type/MessageNewErrors.h"
+
+class MessageFilterNewErrors
+	: public MessageFilter
+{
+	void filter(MessageQueue::MessageBufferType* messageBuffer) override
+	{
+		if (messageBuffer->size() < 2)
+		{
+			return;
+		}
+
+		MessageBase* message = messageBuffer->front().get();
+		if (message->getType() == MessageNewErrors::getStaticType())
+		{
+			for (auto it = messageBuffer->begin() + 1; it != messageBuffer->end(); it++)
+			{
+				if ((*it)->getType() == MessageNewErrors::getStaticType())
+				{
+					MessageNewErrors* frontErrorsMessage = dynamic_cast<MessageNewErrors*>(message);
+					MessageNewErrors* backErrorsMessage = dynamic_cast<MessageNewErrors*>(it->get());
+
+					backErrorsMessage->errors.insert(
+						backErrorsMessage->errors.begin(),
+						frontErrorsMessage->errors.begin(),
+						frontErrorsMessage->errors.end()
+					);
+
+					messageBuffer->pop_front();
+					return;
+				}
+			}
+		}
+	}
+};
+
+#endif // MESSAGE_FILTER_NEW_ERRORS_H

src/lib/utility/messaging/type/MessageNewErrors.h

@@ -27,7 +27,7 @@ public:
 		os << errors.size() << " errors";
 	}
 
-	const std::vector<ErrorInfo> errors;
+	std::vector<ErrorInfo> errors;
 	const ErrorCountInfo errorCount;
 };