+workers

Author: deadBranch

.idea/workspace.xml

@@ -5,8 +5,9 @@
 void GenTokenCommandHandler::Handle(char *bytes, size_t size, TcpSession &sessionContext) {
     char* data = bytes + 1;
     size_t dataLength = size - 1;
-    auto ptr = tokenMap.genToken(bytes, dataLength);
-    cout << ptr->token << endl;
+    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));
+    sessionContext.myWorker->pushRemoveTask(move(ptr));
 }

CommandHandlers/GenTokenCommandHandler.cpp

@@ -14,12 +14,9 @@ void GetTokenCommandHandler::Handle(char *bytes, size_t size, TcpSession &sessio
     //cout << "GetToken" << endl;
     char* token = bytes + 1;
     //cout << token << endl;
-
-    cout << "get_token:" << std::this_thread::get_id()<< endl;
-
     //Handle
     auto ptr = tokenMap.try_get(token);
-    if(ptr && ptr->isEnabled) {
+    if(ptr && ptr->timeToDie) {
 //        cout << "token exists" << endl;
         sessionContext.SendPacket<decltype(ptr)>(ptr->data, move(ptr));
     }

CommandHandlers/GetTokenCommandHandler.cpp

@@ -4,6 +4,7 @@
 #define NETWORK_ENDIANNESS __le
 #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 DATA_LENGTH 48 //the maximum length of the token content
 
 #endif //TOKENWIZARD_DEFINES_H

Definitions.h

@@ -15,9 +15,6 @@ TcpSession::~TcpSession() {
 
 void TcpSession::BeginCommunication() {
     auto self = shared_from_this();
-    myWorker->workerService->post([]() {
-        cout << "direct post: " << std::this_thread::get_id()<< endl;
-    });
     myWorker->workerService->post([this, self]()
                     {
                         read_loop();

TcpSession.cpp

@@ -13,7 +13,6 @@
 extern HandlerSelector handlerSelector;
 
 class TcpSession: public std::enable_shared_from_this<TcpSession> {
-    Worker* myWorker;
     tcp::socket mSocket;
     enum {
         maxLength = 65536
@@ -63,6 +62,7 @@ class TcpSession: public std::enable_shared_from_this<TcpSession> {
 
 
 public:
+    Worker* myWorker;
     template<class T>
     void SendPacket(Packet *packet, T &&smart_pointer) {
         auto self = shared_from_this();

TcpSession.h

@@ -4,14 +4,16 @@
 #include "BaseHeader.h"
 #include "Int32Encoder.h"
 #include "RandomFiller.h"
+#include "ServerResponses.h"
 
 template <size_t tokenLength>
 class Token {
 public:
     Packet* data;
     char token[tokenLength+7];
-    bool isEnabled = true;
+    time_t timeToDie;
     Token(uint32_t val, Packet* _dataPacket): data(_dataPacket) {
+        timeToDie = time(0) + TOKEN_TTL;
         token[tokenLength+6] = '\0';
         Int32Encoder::encode64Based(val, token);
         RandomFiller<tokenLength>::fillString(token+6);

Token.h

@@ -6,7 +6,6 @@ void TokenServer::acceptLoop() {
     mAcceptor.async_accept(mCurrSocket,
                            [this, pSelectedWorker](boost::system::error_code ec)
                            {
-                               cout << "acceptor:" << std::this_thread::get_id()<< endl;
                                if (!ec)
                                {
                                    std::make_shared<TcpSession>(std::move(mCurrSocket), pSelectedWorker)->BeginCommunication();

TokenServer.cpp

@@ -26,6 +26,7 @@ class UnorderedTokenMap {
     cg_shared_ptr<Token<tokenLength>> genToken(const char* data, size_t len) {
         Packet* dataPacket = new Packet((char)ServerResponse::tokenExists, len);
         memcpy(dataPacket->data()+5, data, len);
+        char* d  = dataPacket->data()+5;
         dataPacket->serialize();
 
         while(true) {

UnorderedTokenMap.h

@@ -2,16 +2,45 @@
 #define TOKENWIZARD_WORKER_H
 
 #include <thread>
+#include <queue>
 #include "BaseHeader.h"
+#include "Token.h"
+
 using namespace boost::asio;
 
 class Worker {
-    boost::asio::io_service::work* fakeWork;
+    typedef cg_shared_ptr<Token<TOKEN_LENGTH>> TokenPtr;
+    boost::asio::io_service::work* gcWork;
+
+    boost::posix_time::seconds interval; // 1 second
+    boost::asio::deadline_timer* timer;
+    time_t currTime;
+    void gcTick() {
+        currTime = time(0);
+        size_t currSize = removeQueue.size();
+        for(size_t i = 0; i < currSize; ++i) {
+            auto ptr = removeQueue.front();
+            if(ptr->timeToDie < currTime)
+                removeQueue.pop();
+            else
+                break;
+        }
+        //std::cout << "gcTick" << std::endl;
+        timer->expires_at(timer->expires_at() + interval);
+        timer->async_wait(std::bind(&Worker::gcTick, this));
+    }
 public:
+    void pushRemoveTask(TokenPtr&& tokenPtr) {
+        removeQueue.push(move(tokenPtr));
+    }
+    queue<TokenPtr> removeQueue;
     io_service* workerService;
-    Worker() {
+    Worker(): interval(1) {
+        currTime = time(0);
         workerService = new io_service();
-        fakeWork = new io_service::work(*workerService);
+        gcWork = new io_service::work(*workerService);
+        timer = new deadline_timer(*workerService, interval);
+        timer->async_wait(std::bind(&Worker::gcTick, this));
         thread([this]() {
             workerService->run();
         }).detach();