support for multithreaded processing

Author: benapetr

gp.cpp

@@ -13,23 +13,35 @@
 #include <QTcpSocket>
 #include <QSslSocket>
 #include <QDataStream>
+#include "thread.h"
 #include "gp_exception.h"
 #include "gp.h"
 
 using namespace libgp;
 
-GP::GP(QTcpSocket *tcp_socket)
+GP::GP(QTcpSocket *tcp_socket, bool mt)
 {
     this->socket = tcp_socket;
     this->sentBytes = 0;
     this->recvBytes = 0;
     // We don't want to receive single packet bigger than 800kb
     this->MaxIncomingCacheSize = 800 * 1024;
     this->incomingPacketSize = 0;
+    if (!mt)
+    {
+        this->thread = NULL;
+    }
+    else
+    {
+        this->thread = new Thread(this);
+        this->thread->start();
+    }
+    this->isMultithreaded = mt;
 }
 
 GP::~GP()
 {
+    delete this->thread;
     delete this->socket;
 }
 
@@ -75,6 +87,23 @@ void GP::OnIncomingCommand(QString text, QHash<QString, QVariant> parameters)
     emit this->Event_IncomingCommand(text, parameters);
 }
 
+void GP::processPacket()
+{
+    if (this->isMultithreaded)
+    {
+        // Store this byte array into fifo for later processing by processor thread
+        this->mtLock.lock();
+        this->mtBuffer.append(this->incomingCache);
+        this->mtLock.unlock();
+        this->incomingPacketSize = 0;
+        this->incomingCache.clear();
+        return;
+    }
+
+    QHash<QString, QVariant> pack = this->packetFromIncomingCache();
+    this->processPacket(pack);
+}
+
 void GP::processPacket(QHash<QString, QVariant> pack)
 {
     if (!pack.contains("type"))
@@ -112,7 +141,7 @@ void GP::processIncoming(QByteArray data)
             // this is extremely rare
             // the packet we received is exactly the remaining part of a block of data we are receiving
             this->incomingCache.append(data);
-            this->processPacket(this->packetFromIncomingCache());
+            this->processPacket();
         } else if (data.size() < remaining_packet_data)
         {
             // just append and skip
@@ -123,7 +152,7 @@ void GP::processIncoming(QByteArray data)
             // packet, so we need to cut the remaining part and process
             QByteArray remaining_part = data.mid(0, remaining_packet_data);
             this->incomingCache.append(remaining_part);
-            this->processPacket(this->packetFromIncomingCache());
+            this->processPacket();
             data = data.mid(remaining_packet_data);
             this->processIncoming(data);
         }
@@ -185,6 +214,15 @@ QHash<QString, QVariant> GP::packetFromIncomingCache()
     return data;
 }
 
+QHash<QString, QVariant> GP::packetFromRawBytes(QByteArray packet)
+{
+    QDataStream stream(&packet, QIODevice::ReadWrite);
+    GP_INIT_DS(stream);
+    QHash<QString, QVariant> data;
+    stream >> data;
+    return data;
+}
+
 void GP::processHeader(QByteArray data)
 {
     qint64 header;
@@ -197,6 +235,19 @@ void GP::processHeader(QByteArray data)
     this->incomingPacketSize = header;
 }
 
+QByteArray GP::mtPop()
+{
+    QByteArray result;
+    this->mtLock.lock();
+    if (this->mtBuffer.size() > 0)
+    {
+        result = this->mtBuffer.at(0);
+        this->mtBuffer.removeAt(0);
+    }
+    this->mtLock.unlock();
+    return result;
+}
+
 bool GP::SendPacket(QHash<QString, QVariant> packet)
 {
     if (!this->socket)

gp.h

@@ -68,6 +68,8 @@ class QTcpSocket;
 
 namespace libgp
 {
+    class Thread;
+
     //! Grumpy protocol
 
     //! This class is able to handle server or client connection using
@@ -76,7 +78,7 @@ namespace libgp
     {
             Q_OBJECT
         public:
-            GP(QTcpSocket *tcp_socket = 0);
+            GP(QTcpSocket *tcp_socket = 0, bool mt = false);
             virtual ~GP();
             virtual bool IsConnected() const;
             virtual bool SendPacket(QHash<QString, QVariant> packet);
@@ -90,6 +92,7 @@ namespace libgp
             unsigned long long GetBytesRcvd();
             virtual int GetVersion();
             unsigned long MaxIncomingCacheSize;
+            friend class libgp::Thread;
 
         signals:
             void Event_Connected();
@@ -108,10 +111,15 @@ namespace libgp
 
         protected:
             virtual void OnIncomingCommand(QString text, QHash<QString, QVariant> parameters);
-            void processPacket(QHash<QString, QVariant> pack);
-            void processIncoming(QByteArray data);
+            virtual void processPacket();
+            virtual void processPacket(QHash<QString, QVariant> pack);
+            virtual void processIncoming(QByteArray data);
             QHash<QString, QVariant> packetFromIncomingCache();
+            QHash<QString, QVariant> packetFromRawBytes(QByteArray packet);
             void processHeader(QByteArray data);
+            QByteArray mtPop();
+            QMutex mtLock;
+            QList<QByteArray> mtBuffer;
             QMutex mutex;
             qint64 incomingPacketSize;
             QByteArray incomingCache;
@@ -125,6 +133,8 @@ namespace libgp
 #endif
             unsigned long long sentBytes;
             unsigned long long recvBytes;
+            Thread *thread;
+            bool isMultithreaded;
     };
 }
 

gp.pro

@@ -14,11 +14,13 @@ TEMPLATE = lib
 DEFINES += GP_LIBRARY
 
 SOURCES += gp.cpp \
-    gp_exception.cpp
+    gp_exception.cpp \
+    thread.cpp
 
 HEADERS += gp.h\
         gp_global.h \
-    gp_exception.h
+    gp_exception.h \
+    thread.h
 
 unix {
     target.path = /usr/lib

thread.cpp

@@ -0,0 +1,43 @@
+//This program is free software: you can redistribute it and/or modify
+//it under the terms of the GNU Lesser General Public License as published by
+//the Free Software Foundation, either version 3 of the License, or
+//(at your option) any later version.
+
+//This program is distributed in the hope that it will be useful,
+//but WITHOUT ANY WARRANTY; without even the implied warranty of
+//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+//GNU Lesser General Public License for more details.
+
+// Copyright (c) Petr Bena 2015
+
+#include "thread.h"
+#include "gp.h"
+
+using namespace libgp;
+
+Thread::Thread(GP *gp)
+{
+    this->owner = gp;
+}
+
+Thread::~Thread()
+{
+
+}
+
+void Thread::run()
+{
+    while (this->isRunning())
+    {
+        QByteArray incoming = this->owner->mtPop();
+        if (incoming.isEmpty())
+        {
+            this->msleep(100);
+            continue;
+        }
+        // These 2 calls are probably CPU intensive
+        QHash<QString, QVariant> packet = this->owner->packetFromRawBytes(incoming);
+        this->owner->processPacket(packet);
+    }
+}
+

thread.h

@@ -0,0 +1,34 @@
+//This program is free software: you can redistribute it and/or modify
+//it under the terms of the GNU Lesser General Public License as published by
+//the Free Software Foundation, either version 3 of the License, or
+//(at your option) any later version.
+
+//This program is distributed in the hope that it will be useful,
+//but WITHOUT ANY WARRANTY; without even the implied warranty of
+//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+//GNU Lesser General Public License for more details.
+
+// Copyright (c) Petr Bena 2015
+
+#ifndef THREAD_H
+#define THREAD_H
+
+#include <QThread>
+
+namespace libgp
+{
+    class GP;
+
+    class Thread : public QThread
+    {
+        public:
+            Thread(GP *gp);
+            ~Thread();
+
+        private:
+            GP *owner;
+            void run();
+    };
+}
+
+#endif // THREAD_H