Refactor pretranslation to use helper worker classes

Author: vslavik

src/pretranslate.cpp

@@ -26,45 +26,104 @@
 #include "pretranslate.h"
 
 #include "configuration.h"
+#include "errors.h"
 #include "progress.h"
 #include "str_helpers.h"
 #include "tm/transmem.h"
 
 #include <wx/stopwatch.h>
 
+#include <boost/thread/thread.hpp>
+
+#include <algorithm>
 #include <atomic>
+#include <chrono>
 #include <deque>
 #include <mutex>
+#include <thread>
+
+using namespace std::chrono_literals;
 
 
 namespace pretranslate
 {
 
-Stats PreTranslateCatalog(CatalogPtr catalog, const CatalogItemArray& range, PreTranslateOptions options, dispatch::cancellation_token_ptr cancellation_token)
+namespace
 {
-    wxStopWatch sw;
 
-    if (range.empty())
-        return {};
 
-    const bool use_local_tm = Config::UseTM();
-    if (!use_local_tm)
-        return {};
+struct JobMetadata
+{
+    Language srclang, lang;
+    unsigned nplurals;
+    PreTranslateOptions options;
+};
 
-    TranslationMemory& tm = TranslationMemory::Get();
-    auto srclang = catalog->GetSourceLanguage();
-    auto lang = catalog->GetLanguage();
-    const auto flags = options.flags;
 
-    Progress top_progress(1);
-    top_progress.message(_(L"Preparing strings…"));
+/**
+ Base class for a worked implementing pre-translation process.
+
+ Typically runs the work in some background threads, modifying catalog items passed to it.
+ */
+class Worker
+{
+public:
+    Worker(const JobMetadata& meta) : m_metadata(meta), m_completed(false) {}
+    virtual ~Worker() {}
+
+    /// Add another item for processing
+    void upload(CatalogItemPtr item)
+    {
+        std::lock_guard lock(m_mutex);
+        if (m_completed)
+            return;
+        m_queue.push_back(item);
+    }
 
-    // Function to apply fetched suggestions to a catalog item:
-    auto process_results = [=](CatalogItemPtr dt, unsigned index, const SuggestionsList& results) -> ResType
+    /**
+        Call to mark the job as done with adding items to the queue.
+        Can only be called from the primary thread, i.e. one that created the worker.
+     */
+    void upload_completed()
+    {
+        std::lock_guard lock(m_mutex);
+        m_completed = true;
+    }
+
+    /// Is processing of the entire queue finished?
+    bool is_finished() const
+    {
+        std::lock_guard lock(m_mutex);
+        return m_completed && m_queue.empty();
+    }
+
+    /**
+     Perform some amount of work on primary thread. May take some time, but shouldn't be _long_ time.
+
+     Returns true if there is more work to do, false otherwise.
+
+     Should accomodate cancellation by checking the cancellation token and returning false if cancellation
+     is requested, but may e.g. do it inside http request handling, not immediately.
+
+     Can only be called from the primary thread, i.e. one that created the worker.
+     */
+    virtual bool pump(dispatch::cancellation_token_ptr) = 0;
+
+    /// Assignable next worker to process items this worker couldn't handle
+    Worker *next_worker = nullptr;
+
+    /// Assignable stats collector for processed items
+    std::shared_ptr<Stats> stats;
+
+protected:
+    ResType process_results(CatalogItemPtr dt, unsigned index, const SuggestionsList& results)
     {
         if (results.empty())
             return ResType::None;
+
+        const auto flags = m_metadata.options.flags;
         auto& res = results.front();
+
         if ((flags & PreTranslate_OnlyExact) && !res.IsExactMatch())
             return ResType::Rejected;
 
@@ -89,33 +148,96 @@ Stats PreTranslateCatalog(CatalogPtr catalog, const CatalogItemArray& range, Pre
         dt->SetFuzzy(isFuzzy);
 
         return res.IsExactMatch() ? ResType::Exact : ResType::Fuzzy;
-    };
+    }
 
-    Stats stats;
+    ResType process_result(CatalogItemPtr dt, unsigned index, const Suggestion& result)
+    {
+        return process_results(dt, index, SuggestionsList{result});
+    }
 
-    std::vector<dispatch::future<ResType>> operations;
-    for (auto dt: range)
+    void clear_queue()
     {
-        if (dt->IsTranslated() && !dt->IsFuzzy())
-            continue;
+        std::lock_guard lock(m_mutex);
+        m_queue.clear();
+        m_completed = true;
+    }
+
+protected:
+    JobMetadata m_metadata;
 
-        stats.input_strings_count++;
+    mutable std::mutex m_mutex;
+    std::deque<CatalogItemPtr> m_queue;
+    std::atomic<bool> m_completed;
+};
 
-        operations.push_back(dispatch::async([=,&tm]() -> ResType
+
+class LocalDBWorker : public Worker
+{
+public:
+    LocalDBWorker(const JobMetadata& meta)
+        : Worker(meta), m_tm(TranslationMemory::Get())
+    {
+        const auto nthreads = std::clamp(std::thread::hardware_concurrency(), 4u, 16u);
+        for (unsigned i = 0; i < nthreads; ++i)
         {
-            if (cancellation_token->is_cancelled())
-                return {};
+            m_threads.create_thread([this]{ thread_worker(); });
+        }
+    }
 
-            auto results = tm.Search(srclang, lang, str::to_wstring(dt->GetString()));
+    bool pump(dispatch::cancellation_token_ptr cancellation_token) override
+    {
+        if (cancellation_token->is_cancelled())
+        {
+            clear_queue();
+            // fall through to wait for threads to finish and exit
+        }
+
+        if (is_finished())
+        {
+            m_threads.join_all();
+            return false;
+        }
+
+        return true;
+    }
+
+private:
+    void thread_worker()
+    {
+        CatalogItemPtr dt;
+
+        while (true)
+        {
+            // pop one item of work:
+            {
+                std::lock_guard lock(m_mutex);
+                if (m_queue.empty())
+                {
+                    if (m_completed)
+                    {
+                        break;  // no more work to do
+                    }
+                    else
+                    {
+                        std::this_thread::yield();
+                        continue; // wait for more work to be added
+                    }
+                }
+
+                dt = std::move(m_queue.front());
+                m_queue.pop_front();
+            }
+
+            auto results = m_tm.Search(m_metadata.srclang, m_metadata.lang, str::to_wstring(dt->GetString()));
             auto rt = process_results(dt, 0, results);
 
             if (translated(rt) && dt->HasPlural())
             {
-                switch (lang.nplurals())
+                switch (m_metadata.nplurals)
                 {
                     case 2:  // "simple" English-like plurals
                     {
-                        auto results_plural = tm.Search(srclang, lang, str::to_wstring(dt->GetPluralString()));
+                        auto results_plural = m_tm.Search(m_metadata.srclang, m_metadata.lang, str::to_wstring(dt->GetPluralString()));
                         process_results(dt, 1, results_plural);
                     }
                     case 1:  // nothing else to do
@@ -124,26 +246,129 @@ Stats PreTranslateCatalog(CatalogPtr catalog, const CatalogItemArray& range, Pre
                 }
             }
 
-            return rt;
-        }));
+            if (next_worker)
+            {
+                if (!translated(rt))
+                {
+                    // no usable translation, request elsewhere
+                    next_worker->upload(dt);
+                    continue;
+                }
+                else
+                {
+                    // usable local translation, but try to find better quality elsewhere if possible
+                    auto score = results.front().score;
+                    if (score < 0.95)
+                    {
+                        next_worker->upload(dt);
+                        continue;
+                    }
+                }
+            }
+
+            // if the item wasn't passed to next worker, count it
+            if (stats)
+            {
+                stats->inc_processed();
+                stats->add(rt);
+            }
+        }
     }
 
+private:
+    boost::thread_group m_threads;
+    TranslationMemory& m_tm;
+};
+
+
+} // anonymous namespace
+
+
+std::shared_ptr<Stats> PreTranslateCatalog(CatalogPtr catalog,
+                                           const CatalogItemArray& range,
+                                           PreTranslateOptions options,
+                                           dispatch::cancellation_token_ptr cancellation_token)
+{
+    wxStopWatch sw;
+
+    auto stats = std::make_shared<Stats>();
+
+    if (range.empty())
+        return stats;
+
+    Progress top_progress(1);
+    top_progress.message(_(L"Preparing strings…"));
+
+    const bool use_local_tm = Config::UseTM();
+    if (!use_local_tm)
+        return stats;
+
+    JobMetadata metadata;
+    metadata.srclang = catalog->GetSourceLanguage();
+    metadata.lang = catalog->GetLanguage();
+    metadata.nplurals = metadata.lang.nplurals();
+    metadata.options = options;
+
+    auto worker_local = use_local_tm ? std::make_unique<LocalDBWorker>(metadata) : nullptr;
+
+    if (worker_local)
+        worker_local->stats = stats;
+
+    Worker *worker_ingest = worker_local.get();
+
+    Progress top_progress(1);
+    top_progress.message(_(L"Preparing strings…"));
+
+    // Feed in the work to the worker:
+    for (auto dt: range)
     {
-        Progress progress((int)operations.size());
-        progress.message(_(L"Pre-translating from translation memory…"));
+        if (dt->IsTranslated() && !dt->IsFuzzy())
+            continue;
+
+        stats->input_strings_count++;
+        worker_ingest->upload(dt);
+    }
+    worker_ingest->upload_completed();
+
+    // Wait for completion:
+    Progress progress(stats->input_strings_count, top_progress, 1);
+    progress.message(_(L"Pre-translating…"));
 
-        for (auto& op: operations)
+    int last_matched = 0;
+    bool more_work = true;
+    while (more_work)
+    {
+        try
         {
-            if (cancellation_token->is_cancelled())
-                break;
+            std::this_thread::sleep_for(10ms);
+
+            // pump the workers:
+            more_work = false;
+            if (worker_local)
+            {
+                if (worker_local->pump(cancellation_token))
+                {
+                    more_work = true;
+                }
+                else
+                {
+                    worker_local.reset();
+                }
+            }
 
-            auto rt = op.get();
-            stats.add(rt);
-            if (translated(rt))
+            // update progress bar:
+            if (last_matched != stats->matched)
             {
-                progress.message(wxString::Format(wxPLURAL("Pre-translated %u string", "Pre-translated %u strings", stats.matched), stats.matched));
+                last_matched = stats->matched;
+                progress.message(wxString::Format(wxPLURAL("Pre-translated %u string", "Pre-translated %u strings", last_matched), last_matched));
             }
-            progress.increment();
+            progress.set(stats->input_strings_processed);
+        }
+        catch (...)
+        {
+            stats->errors++;
+            wxLogError("%s", DescribeCurrentException());
+            break;
         }
     }