packet buffering realized

Author: deadBranch

.idea/tokenwizard.iml

@@ -26,11 +26,15 @@ using std::endl;
 #include <boost/shared_ptr.hpp>
 #include <boost/make_shared.hpp>
 #include <boost/asio.hpp>
+#include <alien.pm/tokenwizard/serialization/PacketBuffer.h>
 
 #include "concurrent-guard/cg_shared_ptr.h"
+#include "Definitions.h"
 
 using boost::make_shared;
 using boost::asio::ip::tcp;
 
+typedef PacketBuffer<PACKET_BUFFER_SIZE> PBuff;
+
 
 #endif //TOKENWIZARD_BASEHEADER_H

.idea/workspace.xml

@@ -18,5 +18,5 @@ file(GLOB SOURCE_FILES
         "*/*.cpp"
         )
 
-add_executable(tokenwizard ${SOURCE_FILES} ${Boost_LIBRARIES} HandlerSelector.h CommandHandler.h CommandHandlers/GetTokenCommandHandler.cpp CommandHandlers/GetTokenCommandHandler.h CommandHandlers/GenTokenCommandHandler.cpp CommandHandlers/GenTokenCommandHandler.h Worker.h)
+add_executable(tokenwizard ${SOURCE_FILES} ${Boost_LIBRARIES} HandlerSelector.h CommandHandler.h CommandHandlers/GetTokenCommandHandler.cpp CommandHandlers/GetTokenCommandHandler.h CommandHandlers/GenTokenCommandHandler.cpp CommandHandlers/GenTokenCommandHandler.h Worker.h serialization/PacketBuffer.h serialization/StaticPacket.h)
 TARGET_LINK_LIBRARIES (tokenwizard ${Boost_LIBRARIES})

BaseHeader.h

@@ -1,3 +1,4 @@
+#include <alien.pm/tokenwizard/ClientCommands.h>
 #include "GenTokenCommandHandler.h"
 #include "../UnorderedTokenMap.h"
 #include "../TcpSession.h"
@@ -7,7 +8,8 @@ void GenTokenCommandHandler::Handle(char *bytes, size_t size, TcpSession &sessio
     size_t dataLength = size - 1;
     auto ptr = tokenMap.genToken(data, dataLength);
     cout << "Generated: " << ptr->token << endl;
-    shared_ptr<Packet> infoPtr = ptr->tokenInfo();
-    sessionContext.SendPacket<shared_ptr<Packet>>(infoPtr.get(), move(infoPtr));
+    ptr->writeTokenInfo(sessionContext.getPacketBuffer(TOKEN_LENGTH+8+4));
+    //shared_ptr<StaticPacket> infoPtr = ptr->tokenInfo();
+    //sessionContext.writeStaticPacket(*infoPtr.get());
     sessionContext.myWorker->pushRemoveTask(move(ptr));
 }

CMakeLists.txt

@@ -7,7 +7,7 @@
 #include "../UnorderedTokenMap.h"
 #include "../TcpSession.h"
 
-extern Packet tokenDoesNotExistPacket;
+extern StaticPacket tokenDoesNotExistPacket;
 
 void GetTokenCommandHandler::Handle(char *bytes, size_t size, TcpSession &sessionContext) {
     //Deserialize
@@ -18,11 +18,9 @@ void GetTokenCommandHandler::Handle(char *bytes, size_t size, TcpSession &sessio
     auto ptr = tokenMap.try_get(token);
     if(ptr && ptr->timeToDie) {
 //        cout << "token exists" << endl;
-        sessionContext.SendPacket<decltype(ptr)>(ptr->data, move(ptr));
+        sessionContext.writeStaticPacket(*ptr->dataPacket);
     }
     else {
-        cout << token << endl;
-        cout << "Unknown token" << endl;
-        sessionContext.SendPacket(&tokenDoesNotExistPacket);
+        sessionContext.writeStaticPacket(tokenDoesNotExistPacket);
     }
 }

CommandHandlers/GenTokenCommandHandler.cpp

@@ -5,6 +5,8 @@
 #define MAP_SIZE_EXPONENT 26 //the number of the elements = 2^mapSizeExponent; range: 0 < MAP_SIZE_EXPONENT < 32
 #define TOKEN_LENGTH 32 //the length of the entire token
 #define TOKEN_TTL 20 //time to live of a token
+#define SEND_INTERVAL 1000
 #define DATA_LENGTH 48 //the maximum length of the token content
+#define PACKET_BUFFER_SIZE 32768
 
 #endif //TOKENWIZARD_DEFINES_H

CommandHandlers/GetTokenCommandHandler.cpp

@@ -2,7 +2,6 @@
 #define TOKENWIZARD_INT32ENCODER_H
 
 #include "BaseHeader.h"
-#include "serialization/Packet.h"
 
 class Int32Encoder {
     static char encodeMap[];

Definitions.h

@@ -3,7 +3,6 @@
 
 #include "BaseHeader.h"
 #include "Definitions.h"
-#include "serialization/Packet.h"
 
 extern const char alphabet[63];
 

Int32Encoder.h

@@ -2,8 +2,11 @@
 #include "TcpSession.h"
 
 TcpSession::TcpSession(tcp::socket socket, Worker* myWorker):
-        mSocket(std::move(socket)), myWorker(myWorker)
+        mSocket(std::move(socket)), myWorker(myWorker), mSendInterval(SEND_INTERVAL)
 {
+    currentBuffer = new PBuff();
+    timer = new deadline_timer(*myWorker->workerService, mSendInterval);
+    timer->async_wait(std::bind(&TcpSession::sendDaemonTick, this));
     writeOffset = mData;
     readOffset = mData;
     cout << "Session has been created!" << endl;
@@ -21,3 +24,82 @@ void TcpSession::BeginCommunication() {
                     }
     );
 }
+
+void TcpSession::handlePacket(char *bytes, size_t size) {
+    auto handlerPtr = handlerSelector.getHandler(*bytes);
+    assert(handlerPtr);
+    if (handlerPtr)
+        handlerPtr->Handle(bytes, size, *this);
+}
+
+void TcpSession::readHandler(size_t written) {
+    currSize += written;
+    writeOffset += written;
+    while (true) {
+        uint32_t packetSize = fixEndianness(*reinterpret_cast<uint32_t *>(readOffset));
+        if ((packetSize + 4) <= currSize) {
+            handlePacket(readOffset + 4, packetSize);
+            readOffset += 4 + packetSize;
+            currSize -= 4 + packetSize;
+        } else
+            break;
+    }
+    if (currSize) {
+        memmove(mData, readOffset, currSize);
+    }
+    readOffset = mData;
+    writeOffset = readOffset + currSize;
+}
+
+void TcpSession::read_loop() {
+    auto self(shared_from_this());
+    mSocket.async_read_some(boost::asio::buffer(writeOffset, maxLength - currSize),
+                            [this, self](boost::system::error_code ec, std::size_t length) {
+                                if (!ec) {
+                                    readHandler(length);
+                                    read_loop();
+                                }
+                            });
+}
+
+void TcpSession::sendCurrPBuff() {
+    auto self = shared_from_this();
+    PBuff* pbuff = currentBuffer;
+    boost::asio::async_write(mSocket, boost::asio::buffer(pbuff->getBuff(), pbuff->currSize),
+                             [self, pbuff](boost::system::error_code ec, std::size_t l)
+                             {
+                                 if (!ec) {
+                                     cout << "Sent! " << pbuff->currSize << endl;
+                                 }
+                                 delete pbuff;
+                             });
+}
+
+Packet TcpSession::beginPacket(size_t size) {
+    if(currentBuffer->bytesLeft() > size)
+        replacePBuff();
+    return currentBuffer->beginPacket();
+}
+
+void TcpSession::endPacket(Packet &p) {
+    currentBuffer->endPacket(p);
+    tryStartDaemon();
+}
+
+void TcpSession::replacePBuff() {
+    sendCurrPBuff();
+    currentBuffer = new PBuff();
+}
+
+void TcpSession::writeStaticPacket(StaticPacket &p) {
+    if(currentBuffer->bytesLeft() < p.size())
+        replacePBuff();
+    currentBuffer->writeStaticPacket(p);
+    tryStartDaemon();
+}
+
+PBuff* TcpSession::getPacketBuffer(size_t maxSize) {
+    if(currentBuffer->bytesLeft() < maxSize)
+        replacePBuff();
+    return currentBuffer;
+}

RandomFiller.h

@@ -1,15 +1,11 @@
-//
-// Created by origin on 29.10.16.
-//
-
 #ifndef TOKENWIZARD_SESSION_H
 #define TOKENWIZARD_SESSION_H
 
 #include "BaseHeader.h"
-#include "serialization/Packet.h"
 #include "HandlerSelector.h"
 #include "Worker.h"
 
+
 extern HandlerSelector handlerSelector;
 
 class TcpSession: public std::enable_shared_from_this<TcpSession> {
@@ -18,72 +14,40 @@ class TcpSession: public std::enable_shared_from_this<TcpSession> {
         maxLength = 65536
     };
     char mData[maxLength];
-
-    void handlePacket(char *bytes, size_t size) {
-        auto handlerPtr = handlerSelector.getHandler(*bytes);
-        assert(handlerPtr);
-        if (handlerPtr)
-            handlerPtr->Handle(bytes, size, *this);
-    }
-
+    void handlePacket(char *bytes, size_t size);
     char *writeOffset;
     char *readOffset;
     size_t currSize = 0;
-
-    void readHandler(size_t written) {
-        currSize += written;
-        writeOffset += written;
-        while (true) {
-            uint32_t packetSize = fixEndianness(*reinterpret_cast<uint32_t *>(readOffset));
-            if ((packetSize + 4) <= currSize) {
-                handlePacket(readOffset + 4, packetSize);
-                readOffset += 4 + packetSize;
-                currSize -= 4 + packetSize;
-            } else
-                break;
-        }
-        if (currSize) {
-            memmove(mData, readOffset, currSize);
-        }
-        readOffset = mData;
-        writeOffset = readOffset + currSize;
+    bool sendDaemonStarted = false;
+    boost::posix_time::milliseconds mSendInterval;
+    boost::asio::deadline_timer* timer;
+
+    void sendDaemonTick() {
+        sendDaemonStarted = false;
+        if(currentBuffer->currSize)
+            replacePBuff();
+        std::cout << "sd tick" << std::endl;
     }
 
-    void read_loop() {
-        auto self(shared_from_this());
-        mSocket.async_read_some(boost::asio::buffer(writeOffset, maxLength - currSize),
-                                [this, self](boost::system::error_code ec, std::size_t length) {
-                                    if (!ec) {
-                                        readHandler(length);
-                                        read_loop();
-                                    }
-                                });
+    void tryStartDaemon() {
+        if(!sendDaemonStarted) {
+            sendDaemonStarted = true;
+            timer->expires_from_now(boost::posix_time::seconds(5));
+            //timer->async_wait(std::bind(&TcpSession::sendDaemonTick, this));
+        }
     }
 
-
+    void readHandler(size_t written);
+    void read_loop();
 public:
     Worker* myWorker;
-    template<class T>
-    void SendPacket(Packet *packet, T &&smart_pointer) {
-        auto self = shared_from_this();
-        boost::asio::async_write(mSocket, boost::asio::buffer(packet -> data(), packet->size()),
-                                 [self, packet, smart_pointer](boost::system::error_code ec, std::size_t l)
-                                 {
-                                     if (!ec) {
-                                         //cout << "Sent!" << endl;
-                                     }
-                                 });
-    }
-    void SendPacket(Packet *packet) {
-        auto self = shared_from_this();
-        boost::asio::async_write(mSocket, boost::asio::buffer(packet -> data(), packet->size()),
-                                 [self, packet](boost::system::error_code ec, std::size_t l)
-                                 {
-                                     if (!ec) {
-                                         cout << "Sent!" << endl;
-                                     }
-                                 });
-    }
+    PBuff* currentBuffer;
+    void sendCurrPBuff();
+    void replacePBuff();
+    Packet beginPacket(size_t size);
+    void endPacket(Packet& p);
+    void writeStaticPacket(StaticPacket& p);
+    PBuff* getPacketBuffer(size_t maxSize);
     TcpSession(tcp::socket socket, Worker* myWorker);
     void BeginCommunication();
     ~TcpSession();

TcpSession.cpp

@@ -1,31 +1,47 @@
 #ifndef TOKENWIZARD_TOKENRECORD_H
 #define TOKENWIZARD_TOKENRECORD_H
 
-#include "BaseHeader.h"
+#include <boost/thread/pthread/once_atomic.hpp>
+#include "serialization/PacketBuffer.h"
+#include "serialization/StaticPacket.h"
 #include "Int32Encoder.h"
 #include "RandomFiller.h"
 #include "ServerResponses.h"
+#include "BaseHeader.h"
 
 template <size_t tokenLength>
 class Token {
+    boost::once_flag decreaseOnce = BOOST_ONCE_INIT;
 public:
-    Packet* data;
+    void deleteOnce
+    StaticPacket* dataPacket;
     char token[tokenLength+7];
     time_t timeToDie;
-    Token(uint32_t val, Packet* _dataPacket): data(_dataPacket) {
+
+    Token(uint32_t val, StaticPacket* _dataPacket): dataPacket(_dataPacket) {
         timeToDie = time(0) + TOKEN_TTL;
         token[tokenLength+6] = '\0';
         Int32Encoder::encode64Based(val, token);
         RandomFiller<tokenLength>::fillString(token+6);
     }
-    shared_ptr<Packet> tokenInfo() {
-        shared_ptr<Packet> packetPtr = std::make_shared<Packet>((char)ServerResponse::tokenCreated, tokenLength+7);
+
+    shared_ptr<StaticPacket> tokenInfo() {
+        shared_ptr<StaticPacket> packetPtr = std::make_shared<StaticPacket>((char)ServerResponse::tokenCreated, tokenLength+7);
         memcpy(packetPtr->data()+5, token, tokenLength + 7);
         packetPtr->serialize();
         return packetPtr;
     }
+
+    void writeTokenInfo(PBuff* buff) {
+        char* dataPtr = buff->getPtr();
+        writeCount(tokenLength+8, dataPtr);
+        dataPtr[4] = (char)ServerResponse::tokenCreated;
+        memcpy(dataPtr+5, token, tokenLength + 7);
+        buff->currSize += tokenLength+8+4;
+    }
+
     ~Token() {
-        delete data;
+        delete dataPacket;
     }
 };