[mod] change files 755 to 644, use threadpool, modify msg_que.

Author: tsymiar

CMakeLists.txt

@@ -1,12 +1,12 @@
 #pragma once
-#include <algorithm>
-#include <chrono>
-#include <cmath>
 #include <csignal>
 #include <cstdlib>
 #include <cstring>
-#include <deque>
 #include <fcntl.h>
+#include <chrono>
+#include <algorithm>
+#include <cmath>
+#include <deque>
 #include <fstream>
 #include <functional>
 #include <iostream>

CMakeSettings.json

@@ -177,7 +177,7 @@ int Broker::setup(unsigned short port)
         return -3;
     }
 
-    queue_init(&g_msgQue);
+    mq_init(&g_msgQue);
     m_socket = sock;
     m_active = true;
 
@@ -305,13 +305,13 @@ int Broker::ProxyTask(Networks& works, const Network& work)
         }
     } while (left > 0);
     msg->head = work.head;
-    queue_push(&g_msgQue, msg);
+    mq_push(&g_msgQue, msg);
     setOffline(works, work.socket);
     std::vector<Network>& vec = works[SUBSCRIBER];// only SUBSCRIBER should be sent
     if (vec.empty()) {
         LOGW("No subscriber to publish!");
     }
-    void* message = queue_front(&g_msgQue);
+    void* message = mq_front(&g_msgQue);
     if (message != nullptr) {
         Message val = *reinterpret_cast<Message*>(message);
         if (val.head.flag != PUBLISHER) {
@@ -346,7 +346,7 @@ int Broker::ProxyTask(Networks& works, const Network& work)
                 }
             }
             Delete(val.payload.content)
-                queue_pop(&g_msgQue);
+                mq_pop(&g_msgQue);
         } else {
             LOGW("Message size(%u) invalid!", val.head.size);
         }
@@ -481,5 +481,5 @@ void Broker::exit()
         Close(m_socket);
         m_socket = -1;
     }
-    queue_deinit(&g_msgQue);
+    mq_deinit(&g_msgQue);
 }

src/CMakeLists.txt

@@ -52,6 +52,9 @@ int Publisher::publish(uint32_t topic, const std::string& payload, ...)
     msg.head.size = static_cast<unsigned int>(msgLen);
     msg.head.topic = topic;
     msg.head.flag = PUBLISHER;
+    msg.payload.status[0] = 'O';
+    msg.payload.status[1] = 'K';
+    msg.payload.status[2] = '\0';
     memcpy(message, &msg, sizeof(Message));
     memcpy(message + HEAD_SIZE + sizeof(Message::Payload::status), payload.c_str(), size);
 

src/common/Scadup.h

@@ -2,16 +2,19 @@
 
 #define LOG_TAG "Subscriber"
 #include "../utils/logging.h"
+#include "../utils/threadpool.hpp"
+#include "../utils/TaskBase.h"
 
 using namespace Scadup;
 extern const char* GET_FLAG(G_ScaFlag x);
 
 bool Subscriber::m_exit = false;
+threadpool g_threadpool{};
 
 void Subscriber::setup(const char* ip, unsigned short port)
 {
     m_socket = socket2Broker(ip, port, m_ssid, 60);
-    std::thread task([&](SOCKET sock, bool& exit) -> void {
+    std::function<void(SOCKET, bool&)> func = [&](SOCKET sock, bool& exit) -> void {
         try {
             LOGI("start keep-alive task");
             keepAlive(sock, exit);
@@ -20,9 +23,9 @@ void Subscriber::setup(const char* ip, unsigned short port)
         } catch (...) {
             LOGE("Unknown exception in keep-alive task");
         }
-        }, m_socket, std::ref(m_exit));
-    if (task.joinable())
-        task.detach();
+        };
+    g_threadpool.enqueue(func, m_socket, std::ref(m_exit));
+    g_threadpool.start(3);
 }
 
 ssize_t Subscriber::subscribe(uint32_t topic, RECV_CALLBACK callback)
@@ -85,9 +88,6 @@ ssize_t Subscriber::subscribe(uint32_t topic, RECV_CALLBACK callback)
                 Delete(body);
                 return -5;
             } else {
-                msg.payload.status[0] = 'O';
-                msg.payload.status[1] = 'K';
-                msg.payload.status[2] = '\0';
                 auto* message = reinterpret_cast<Message*>(new char[size + len]);
                 if (message != nullptr) {
                     memcpy(message, &msg.head, HEAD_SIZE);
@@ -97,7 +97,8 @@ ssize_t Subscriber::subscribe(uint32_t topic, RECV_CALLBACK callback)
                         message->payload.content[len - 1] = '\0';
                     }
                     if (callback != nullptr) {
-                        callback(*message);
+                        // std::function<RECV_CALLBACK> func = callback;
+                        g_threadpool.enqueue(callback, *message);
                     }
                     LOGI("message payload = [%s]-[%s]", message->payload.status, message->payload.content);
                     Delete(message);
@@ -128,6 +129,7 @@ void Subscriber::keepAlive(SOCKET socket, bool& exit)
 
 void Subscriber::quit()
 {
+    g_threadpool.stop();
     Header head{};
     head.cmd = 0xff;
     ::send(m_socket, reinterpret_cast<char*>(&head), HEAD_SIZE, 0);

src/scadup/Broker.cpp

@@ -13,13 +13,13 @@
 
 static pthread_mutex_t g_mutex;
 
-void queue_init(struct MsgQue* q)
+void mq_init(struct MsgQue* q)
 {
     pthread_mutex_init(&g_mutex, NULL);
     q->head = q->tail = NULL;
 }
 
-void queue_deinit(struct MsgQue* q)
+void mq_deinit(struct MsgQue* q)
 {
     if (q == NULL)
         return;
@@ -33,7 +33,7 @@ void queue_deinit(struct MsgQue* q)
     pthread_mutex_destroy(&g_mutex);
 }
 
-int queue_push(struct MsgQue* q, void* x)
+int mq_push(struct MsgQue* q, void* x)
 {
     pthread_mutex_lock(&g_mutex);
     if (q == NULL) {
@@ -58,7 +58,7 @@ int queue_push(struct MsgQue* q, void* x)
     return 0;
 }
 
-void queue_pop(struct MsgQue* q)
+void mq_pop(struct MsgQue* q)
 {
     pthread_mutex_lock(&g_mutex);
     if (q != NULL && q->head != NULL) {
@@ -74,7 +74,7 @@ void queue_pop(struct MsgQue* q)
     pthread_mutex_unlock(&g_mutex);
 }
 
-void* queue_front(struct MsgQue* q)
+void* mq_front(struct MsgQue* q)
 {
     pthread_mutex_lock(&g_mutex);
     if (q == NULL || q->head == NULL) {
@@ -86,7 +86,7 @@ void* queue_front(struct MsgQue* q)
     return p;
 }
 
-int queue_size(struct MsgQue* q)
+int mq_size(struct MsgQue* q)
 {
     pthread_mutex_lock(&g_mutex);
     if (q != NULL) {

src/scadup/Publisher.cpp

@@ -11,11 +11,11 @@ typedef struct MsgQue {
     INode* tail;
 } MsgQue;
 
-void queue_init(struct MsgQue* q);
-int queue_push(struct MsgQue* q, void* x);
-void* queue_front(struct MsgQue* q);
-void queue_pop(struct MsgQue* q);
-int queue_size(struct MsgQue* q);
-void queue_deinit(struct MsgQue* q);
+void mq_init(struct MsgQue* q);
+int mq_push(struct MsgQue* q, void* x);
+void* mq_front(struct MsgQue* q);
+void mq_pop(struct MsgQue* q);
+int mq_size(struct MsgQue* q);
+void mq_deinit(struct MsgQue* q);
 
 #endif

src/scadup/Subscriber.cpp

@@ -1,4 +1,38 @@
-#include "threadpool.h"
+
+#ifndef THREADPOOL_HPP
+#define THREADPOOL_HPP
+
+#include <vector>
+#include <queue>
+#include <thread>
+#include <functional>
+#include <mutex>
+#include <condition_variable>
+#include <atomic>
+#include <future>
+#include <stdexcept>
+
+class threadpool {
+public:
+    threadpool();
+    ~threadpool();
+
+    template<class F>
+    void enqueue(F&& f);
+
+    template<class F, class... Args>
+    std::future<typename std::result_of<F(Args...)>::type> enqueue(F&& f, Args&&... args);
+
+    void start(size_t threads);
+    void stop();
+
+private:
+    std::vector<std::thread> m_workers = {};
+    std::queue<std::function<void()>> m_tasks = {};
+    std::mutex m_queueMutex{};
+    std::condition_variable m_condition{};
+    std::atomic<bool> m_stopPool;
+};
 
 threadpool::threadpool() : m_stopPool(false) { }
 
@@ -68,3 +102,5 @@ void threadpool::stop()
         worker.join();
     }
 }
+
+#endif // THREADPOOL_HPP