Added a correct implementation of multi-threaded indexing using a thread

pool and a semaphore.

Fixed a couple more hotkey issues.

README.md

@@ -1,6 +1,6 @@
 # musikcube
 
-a cross-platform audio engine written in C++.
+a cross-platform audio engine and metadata indexer written in C++.
 
 # musikbox
 

src/core/library/Indexer.cpp

@@ -51,7 +51,7 @@
 #include <boost/thread/xtime.hpp>
 #include <boost/bind.hpp>
 
-#define MULTI_THREADED_INDEXER 0
+#define MULTI_THREADED_INDEXER 1
 
 static const std::string TAG = "Indexer";
 static const int MAX_THREADS = 10;
@@ -78,17 +78,16 @@ static std::string normalizePath(const std::string& path) {
 }
 
 Indexer::Indexer(const std::string& libraryPath, const std::string& dbFilename)
-: thread(NULL)
+: thread(nullptr)
 , status(0)
 , restart(false)
 , filesIndexed(0)
 , filesSaved(0)
-, maxReadThreads(MAX_THREADS) {
+, prefs(Preferences::ForComponent(INDEXER_PREFS_COMPONENT))
+, readSemaphore(prefs->GetInt(INDEXER_PREFS_MAX_TAG_READ_THREADS, MAX_THREADS)) {
     this->dbFilename = dbFilename;
     this->libraryPath = libraryPath;
-    this->prefs = Preferences::ForComponent(INDEXER_PREFS_COMPONENT);
     this->thread = new boost::thread(boost::bind(&Indexer::ThreadLoop, this));
-    this->maxReadThreads = this->prefs->GetInt(INDEXER_PREFS_MAX_TAG_READ_THREADS, MAX_THREADS);
 }
 
 Indexer::~Indexer() {
@@ -96,7 +95,7 @@ Indexer::~Indexer() {
         this->Exit();
         this->thread->join();
         delete this->thread;
-        this->thread = NULL;
+        this->thread = nullptr;
     }
 }
 
@@ -137,7 +136,7 @@ void Indexer::RemovePath(const std::string& path) {
     this->Synchronize(true);
 }
 
-void Indexer::SynchronizeInternal() {
+void Indexer::SynchronizeInternal(boost::asio::io_service* io) {
     /* load all of the metadata (tag) reader plugins */
     typedef PluginFactory::DestroyDeleter<IMetadataReader> MetadataDeleter;
     typedef PluginFactory::DestroyDeleter<IDecoderFactory> DecoderDeleter;
@@ -190,7 +189,7 @@ void Indexer::SynchronizeInternal() {
     for(std::size_t i = 0; i < paths.size(); ++i) {
         this->trackTransaction.reset(new db::ScopedTransaction(this->dbConnection));
         std::string path = paths[i];
-        this->SyncDirectory(path, path, pathIds[i]);
+        this->SyncDirectory(io, path, path, pathIds[i]);
     }
 
     if (this->trackTransaction) {
@@ -257,10 +256,7 @@ void Indexer::ReadMetadataFromFile(
     musik::core::IndexerTrack track(0);
 
     /* get cached filesize, parts, size, etc */
-    if (!track.NeedsToBeIndexed(file, this->dbConnection)) {
+    if (track.NeedsToBeIndexed(file, this->dbConnection)) {
-        return;
-    }
-
         bool saveToDb = false;
 
         /* read the tag from the plugin */
@@ -299,31 +295,13 @@ void Indexer::ReadMetadataFromFile(
         }
     }
 
-#if MULTI_THREADED_INDEXER
+#ifdef MULTI_THREADED_INDEXER
-static inline void joinAndNotify(
+    this->readSemaphore.post();
-    std::vector<Thread>& threads,
-    std::shared_ptr<musik::core::db::ScopedTransaction> transaction,
-    sigslot::signal0<>& event, 
-    std::atomic<size_t>& saved)
-{
-    for (size_t i = 0; i < threads.size(); i++) {
-        threads.at(i)->join(); 
-    }
-
-    threads.clear();
-
-    if (saved.load() > 200) {
-        if (transaction) {
-            transaction->CommitAndRestart();
-        }
-
-        event();
-        saved.store(0);
-    }
-}
 #endif
+}
 
 void Indexer::SyncDirectory(
+    boost::asio::io_service* io,
     const std::string &syncRoot,
     const std::string &currentPath,
     DBID pathId)
@@ -349,44 +327,7 @@ void Indexer::SyncDirectory(
         std::string pathIdStr = boost::lexical_cast<std::string>(pathId);
         std::vector<Thread> threads;
 
-#if MULTI_THREADED_INDEXER
-        #define WAIT_FOR_ACTIVE() \
-            joinAndNotify( \
-                threads, \
-                this->trackTransaction, \
-                this->TrackRefreshed, \
-                this->filesSaved);
-#endif
-
         for( ; file != end && !this->Exited() && !this->Restarted(); file++) {
-#if MULTI_THREADED_INDEXER
-            /* we do things in batches of 5. wait for this batch to
-            finish, then we'll spin up some more... */
-            if (threads.size() >= this->maxReadThreads) {
-                WAIT_FOR_ACTIVE();
-            }
-#endif
-            if (is_directory(file->status())) {
-                /* recursion here */
-                musik::debug::info(TAG, "scanning " + file->path().string());
-#if MULTI_THREADED_INDEXER
-                WAIT_FOR_ACTIVE();
-#endif
-                this->SyncDirectory(syncRoot, file->path().string(), pathId);
-            }
-            else {
-                ++this->filesIndexed;
-
-#if MULTI_THREADED_INDEXER
-                threads.push_back(Thread(new boost::thread(
-                    boost::bind(
-                        &Indexer::ReadMetadataFromFile, 
-                        this, 
-                        file->path(), 
-                        pathIdStr))));
-#else
-                this->ReadMetadataFromFile(file->path(), pathIdStr);
-
             if (this->filesSaved.load() > 200) {
                 if (this->trackTransaction) {
                     this->trackTransaction->CommitAndRestart();
@@ -395,14 +336,28 @@ void Indexer::SyncDirectory(
                 this->TrackRefreshed();
                 this->filesSaved.store(0);
             }
-#endif  
+            if (is_directory(file->status())) {
-            }
+                /* recursion here */
+                musik::debug::info(TAG, "scanning " + file->path().string());
+                this->SyncDirectory(io, syncRoot, file->path().string(), pathId);
             }
+            else {
+                ++this->filesIndexed;
 
-#if MULTI_THREADED_INDEXER
+                if (io) {
-        /* there may be a few left... */
+                    this->readSemaphore.wait();
-        WAIT_FOR_ACTIVE();
+
-#endif
+                    io->post(boost::bind(
+                        &Indexer::ReadMetadataFromFile,
+                        this,
+                        file->path(),
+                        pathIdStr));
+                }
+                else {
+                    this->ReadMetadataFromFile(file->path(), pathIdStr);
+                }
+            }
+        }
     }
     catch(...) {
     }
@@ -427,7 +382,27 @@ void Indexer::ThreadLoop() {
 
             this->dbConnection.Open(this->dbFilename.c_str(), 0); /* ensure the db is open */
 
-            this->SynchronizeInternal();
+#ifdef MULTI_THREADED_INDEXER
+            boost::asio::io_service io;
+            boost::thread_group threadPool;
+            boost::asio::io_service::work work(io);
+
+            /* initialize the thread pool -- we'll use this to index tracks in parallel. */
+            int threadCount = prefs->GetInt(INDEXER_PREFS_MAX_TAG_READ_THREADS, MAX_THREADS);
+            for (int i = 0; i < threadCount; i++) {
+                threadPool.create_thread(boost::bind(&boost::asio::io_service::run, &io));
+            }
+
+            this->SynchronizeInternal(&io);
+
+            /* done with sync, remove all the threads in the pool to free resources. they'll
+            be re-created later if we index again. */
+            io.stop();
+            threadPool.join_all();
+#else
+            this->SynchronizeInternal(nullptr);
+#endif
+
             this->RunAnalyzers();
 
             {
@@ -442,9 +417,10 @@ void Indexer::ThreadLoop() {
 
         firstTime = false;
 
-        int waitTime = prefs->GetInt(INDEXER_PREFS_SYNC_TIMEOUT, 3600); /* sleep before we try again... */
+        /* sleep before we try again; disabled by default */
+        int waitTime = prefs->GetInt(INDEXER_PREFS_AUTO_SYNC_MILLIS, 0);
 
-        if (waitTime) {
+        if (waitTime > 0) {
             boost::xtime waitTimeout;
             boost::xtime_get(&waitTimeout, boost::TIME_UTC_);
             waitTimeout.sec += waitTime;

src/core/library/Indexer.h

@@ -46,6 +46,7 @@
 #include <boost/thread/thread.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/asio/io_service.hpp>
+#include <boost/interprocess/sync/interprocess_semaphore.hpp>
 
 #include <deque>
 #include <vector>
@@ -53,7 +54,7 @@
 
 #define INDEXER_PREFS_COMPONENT "indexer"
 #define INDEXER_PREFS_SYNC_ON_STARTUP "SyncOnStartup"
-#define INDEXER_PREFS_SYNC_TIMEOUT "SyncTimeout"
+#define INDEXER_PREFS_AUTO_SYNC_MILLIS "AutoSyncIntervalMillis"
 #define INDEXER_PREFS_REMOVE_MISSING_FILES "RemoveMissingFiles"
 #define INDEXER_PREFS_MAX_TAG_READ_THREADS "MaxTagReadThreads"
 
@@ -83,9 +84,10 @@ namespace musik { namespace core {
             void SyncOptimize();
             void RunAnalyzers();
 
-            void SynchronizeInternal();
+            void SynchronizeInternal(boost::asio::io_service* io);
 
             void SyncDirectory(
+                boost::asio::io_service* io,
                 const std::string& syncRoot,
                 const std::string& currentPath,
                 DBID pathId);
@@ -125,7 +127,7 @@ namespace musik { namespace core {
             DecoderList audioDecoders;
             std::shared_ptr<musik::core::Preferences> prefs;
             std::shared_ptr<musik::core::db::ScopedTransaction> trackTransaction;
-            size_t maxReadThreads;
+            boost::interprocess::interprocess_semaphore readSemaphore;
     };
 
     typedef std::shared_ptr<Indexer> IndexerPtr;

src/musikbox/cursespp/ListWindow.cpp

@@ -83,9 +83,11 @@ void ListWindow::ScrollToBottom() {
 }
 
 void ListWindow::ScrollUp(int delta) {
-    ScrollPos spos = this->GetScrollPosition();
     IScrollAdapter& adapter = this->GetScrollAdapter();
 
+    if (adapter.GetEntryCount() > 0) {
+        ScrollPos spos = this->GetScrollPosition();
+
         size_t first = spos.firstVisibleEntryIndex;
         size_t last = first + spos.visibleEntryCount;
         int drawIndex = first;
@@ -103,15 +105,18 @@ void ListWindow::ScrollUp(int delta) {
         this->SetSelectedIndex(newIndex);
         this->ScrollTo(drawIndex);
     }
+}
 
 void ListWindow::OnInvalidated() {
     this->Invalidated(this, this->GetSelectedIndex());
 }
 
 void ListWindow::ScrollDown(int delta) {
-    ScrollPos spos = this->GetScrollPosition();
     IScrollAdapter& adapter = this->GetScrollAdapter();
 
+    if (adapter.GetEntryCount() > 0) {
+        ScrollPos spos = this->GetScrollPosition();
+
         size_t first = spos.firstVisibleEntryIndex;
         size_t last = first + spos.visibleEntryCount;
         size_t drawIndex = first;
@@ -127,6 +132,7 @@ void ListWindow::ScrollDown(int delta) {
         this->SetSelectedIndex(newIndex);
         this->ScrollTo(drawIndex);
     }
+}
 
 void ListWindow::PageUp() {
     IScrollAdapter &adapter = this->GetScrollAdapter();

src/musikbox/cursespp/ScrollableWindow.cpp

@@ -43,6 +43,8 @@
 
 using namespace cursespp;
 
+static const size_t INVALID_INDEX = (size_t) -1;
+
 typedef IScrollAdapter::ScrollPosition ScrollPos;
 
 #define REDRAW_VISIBLE_PAGE() \

src/musikbox/cursespp/Window.cpp

@@ -395,8 +395,6 @@ void Window::Hide() {
 }
 
 void Window::Destroy() {
-    this->Hide();
-
     if (this->frame) {
         del_panel(this->framePanel);
         delwin(this->frame);