tasks queue

Author: tkvitko

CMakeLists.txt

@@ -24,7 +24,8 @@ if(Boost_FOUND)
     include_directories(${Boost_INCLUDE_DIRS})
     add_executable(cpp_search_qt_creator main.cpp Crowler.h Crowler.cpp root_certificates.hpp
         DbManager.h DbManager.cpp
-        Searcher.h Searcher.cpp)
+        Searcher.h Searcher.cpp
+        SafeQueue.h SafeQueue.cpp)
     add_subdirectory(./libpqxx-7.9.0 libpqxx-build)
     target_compile_features(cpp_search_qt_creator PRIVATE cxx_std_17)
     target_link_libraries(cpp_search_qt_creator ${Boost_LIBRARIES} OpenSSL::SSL OpenSSL::Crypto pqxx)# "${PQXX_LIBRARIES}")

Crowler.cpp

@@ -22,7 +22,7 @@ namespace ssl = net::ssl;       // from <boost/asio/ssl.hpp>
 using tcp = net::ip::tcp;           // from <boost/asio/ip/tcp.hpp>
 
 
-Crowler::Crowler() {}
+
 
 void Crowler::processUrl(std::string url, short depth)
 {
@@ -151,3 +151,42 @@ void Crowler::savePresencesToDb(std::vector<std::string> words, std::string url)
     }
 
 }
+
+Crowler::Crowler()
+{
+    const int cores_count = std::thread::hardware_concurrency();    // количество аппаратных ядер
+    for (size_t i = 0; i != cores_count; ++i) {
+        // наполнение вектора, хранящего потоки, задачами на обработку
+        std::thread t(&Crowler::work, this);
+        threadsPool_.push_back(std::move(t));
+    }
+}
+
+Crowler::~Crowler()
+{
+    for (auto& thread : threadsPool_) {
+        // ожидание окончания работы потоков
+        thread.join();
+    }
+    threadsPool_.clear();
+}
+
+void Crowler::addToCrowlingQueue(std::string url, unsigned short depth)
+{
+    UrlCrowlingTask task = {url, depth};
+    tasksQueue_.push(task);
+}
+
+void Crowler::work() {
+    while (true) {
+        if (!tasksQueue_.isEmpty()) {
+            // если в очереди задач есть задачи, вынимаем одну и выполняем
+            UrlCrowlingTask task;
+            tasksQueue_.pop(task);
+            processUrl(task.url, task.depth);
+        } else {
+            // иначе передаем управление другому потоку
+            std::this_thread::yield();
+        }
+    }
+}

Crowler.h

@@ -3,11 +3,22 @@
 
 #include <string>
 #include <regex>
+#include <queue>
+#include <mutex>
+#include <vector>
+#include <thread>
 #include "DbManager.h"
+#include "SafeQueue.h""
 
 class Crowler
 {
 private:
+
+    // вектор для хранения потоков обработки задач
+    std::vector<std::thread> threadsPool_;
+    // очередь задач на обработку
+    SafeQueue tasksQueue_;
+
     // скачивание html по url
     std::string download(std::string url);
     // получение данных из HTML
@@ -18,10 +29,14 @@ class Crowler
     std::vector<std::string> getSubUrls(std::string innerHtml);
     // вычисление частот слов и сохранение данных в базу
     void savePresencesToDb(std::vector<std::string> words, std::string url);
+    // обход ресурса
+    void processUrl(std::string url, short depth);
+    void work();
 
 public:
     Crowler();
-    void processUrl(std::string url, short depth);
+    ~Crowler();
+    void addToCrowlingQueue(std::string url, unsigned short depth);
 };
 
 #endif // CROWLER_H

SafeQueue.cpp

@@ -0,0 +1,33 @@
+#include "SafeQueue.h"
+
+
+
+SafeQueue::SafeQueue() {}
+SafeQueue::~SafeQueue() {}
+
+void SafeQueue::push(UrlCrowlingTask task)
+{
+    // добавить новую задачу в очередь
+
+    std::lock_guard lock_guard(mutex_);
+    queue_.push(task);
+    // уведомление, что очередь не пуста
+    data_cond.notify_all();
+}
+
+void SafeQueue::pop(UrlCrowlingTask &task)
+{
+    // вынуть задачу из очереди
+
+    std::unique_lock<std::mutex> lk(mutex_);
+    // ожидание, что очередь не пуста
+    data_cond.wait(lk, [this] {return !queue_.empty(); });
+    task = std::move(queue_.front());
+    queue_.pop();
+}
+
+bool SafeQueue::isEmpty()
+{
+    // проверить, пуста ли очередь
+    return queue_.empty();
+}

SafeQueue.h

@@ -0,0 +1,33 @@
+#ifndef SAFEQUEUE_H
+#define SAFEQUEUE_H
+
+#include <queue>
+#include <mutex>
+
+struct UrlCrowlingTask {
+    // Структура, описывающая задачу на обход ресурса
+    // std::function<void()> func;
+    std::string url;
+    unsigned short depth;
+};
+
+
+class SafeQueue
+{
+public:
+    SafeQueue();
+    ~SafeQueue();
+    void push(UrlCrowlingTask task);
+    void pop(UrlCrowlingTask& task);
+    bool isEmpty();
+
+private:
+    // внутренняя очередь для хранения задач
+    std::queue<UrlCrowlingTask> queue_;
+    // мьютекс для блокировки на момент добалвения новой задачи в очередь
+    std::mutex mutex_;
+    // переменная, используемая для ожидания непустой очереди и уведомления о непустоте
+    std::condition_variable data_cond;
+};
+
+#endif // SAFEQUEUE_H