send_queue.h

Go to the documentation of this file.

/*
-------------------------------------------------------------------------------
                            PLEBBLE
-------------------------------------------------------------------------------
 Copyright (C) 2019-2020 KATLAS Technology. (http://katlastechnology.com)
 Copyright (C) 2017-2020 Marcos Mayorga. (mm@mm-studios.com)
 
 This file is part of our Plebble(R) Platform.
 
 The code below cannot be copied, used for any purpose other than the one
 agreed and/or distributed without the express permission of
 KATLAS Technology.
-------------------------------------------------------------------------------
 
 
*/
#ifndef USGOV_3cf53a0bc929f5209a42abf8163c17b908f6d7dfda0b34da09ce1d75d036a947
#define USGOV_3cf53a0bc929f5209a42abf8163c17b908f6d7dfda0b34da09ce1d75d036a947
 
#include <queue>
#include <vector>
#include <unordered_map>
#include <us/gov/bgtask.h>
#include <us/gov/config.h>
#include "busyled.h"
 
namespace us{ namespace gov {
namespace socket {
using namespace std;
 
    struct datagram;
    struct client;
 
    struct qi:pair<datagram*,client*> {
        typedef pair<datagram*,client*> b;
        qi(datagram* d, client*c);
        qi(const qi&)=delete;
        ~qi();
        short skip{0};
    };
 
    struct priority_service {
        bool operator () (qi*& lhs, qi*& rhs) const;
    };
 
    typedef priority_queue<qi*,vector<qi*>,priority_service> pri_q_t;
 
    struct queue_t: pri_q_t, bgtask {  //datagrams with lower service code have more priority
        typedef pri_q_t b;
        typedef bgtask t;
 
        struct ongoing_t: unordered_map<client*,qi*>  {
            qi* reset(qi*);
            void set(qi*);
        };
 
        static constexpr size_t wmh{CFG_SENDQ_WMH}; //watermark high
        static constexpr size_t schunk{CFG_SENDQ_SCHUNK}; 
 
        queue_t();
        ~queue_t() override;
 
        ko send(datagram* d, client* cli, uint16_t pri);
        
        qi* next();
        qi* next_();
        qi* replace(qi*);
        size_t get_size() const;
        void wait();
        void dump(const string& prefix, ostream&) const;
        void clear();
//        virtual void flood(const datagram&, const client&) const;
        void purge(client*);
        void run();
        void onwakeup();
        void set_busy_handler(busyled_t::handler_t* hsend);
 
        condition_variable cv;
        mutable mutex mx;
        #if CFG_COUNTERS==1
            uint64_t sent{0};
            uint64_t dropped{0};
            uint64_t bytes_sent{0};
        #endif
        ongoing_t ongoing;
        //pri_q_t skipped;
        busyled_t busyled;
        #if CFG_LOGS==1
            string logdir;
        #endif
    };
 
    typedef queue_t send_queue_t;
 
}
}}
 
#endif