de/d76/engine_2daemon_8h_source.html

/*

-------------------------------------------------------------------------------

                            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_c13d19f0c3971de6c878039137bc456aee76c735c59394602b4d3ee39130825d

#define USGOV_c13d19f0c3971de6c878039137bc456aee76c735c59394602b4d3ee39130825d


#include <us/gov/auth.h>

#include "diff.h"

#include "peer_t.h"

#include "app.h"

#include "auth_app.h"

#include "policies.h"

#include <unordered_map>

#include <unordered_set>

#include <map>

#include <set>

#include <list>

#include <us/gov/dfs.h>

#include <vector>

#include <cassert>

#include <thread>

#include <chrono>

#include <ctime>

#include <condition_variable>

#include <mutex>

#include "shell.h"

#include "calendar.h"

#include "networking.h"

#include "syncd.h"

#include "evidence.h"

#include <inttypes.h>

#include <us/gov/dfs/fileattr_t.h>

#include <us/gov/config.h>

#include <us/gov/bgtask.h>

#include "evidence_processor.h"


namespace us { namespace gov {

namespace sys {

    struct app;

}

namespace cash {

    struct app;

}


namespace engine {

    using namespace std;

    using socket::datagram;

    using namespace std::chrono;


    struct daemon;


    struct mezzanine: bgtask {

        typedef bgtask b;

        mezzanine(daemon*);

        daemon* d;

    };


    struct daemon: syncd_t::actuator, mezzanine, evidence_processor {

        typedef mezzanine t;

        typedef crypto::ec::keys keys;


        struct apps final:map<int,app*> {

            virtual ~apps();

            void dump(const string& prefix, ostream& os) const;

            bool shell_command(int app_id, istream&, ostream&) const;

            void process(const evidence&);

        };


        struct neuralyzer_t final {

            neuralyzer_t(daemon* d);

            neuralyzer_t(const neuralyzer_t&)=delete;

            neuralyzer_t& operator = (const neuralyzer_t&)=delete;

            ~neuralyzer_t();

            void old_chain(const diff::hash_t& h);

            void delete_chain(const diff::hash_t& h);

            void get_blocks(set<diff::hash_t>& s, const diff::hash_t& h);


            int cleanup();

            daemon *d;

            diff::hash_t old;

        };


        struct votes_t final {

            #if CFG_COUNTERS==1

                unordered_map<pubkey_t::hash_t,const diff::hash_t> voters;

            #else

                unordered_set<pubkey_t::hash_t> voters;

            #endif

            unordered_map<diff::hash_t,uint64_t> votes;

            void clear();

            void dump(const string& prefix, ostream& os) const;

            bool add(const pubkey_t::hash_t&,const diff::hash_t& v);

            diff::hash_t select();

            void logline(ostream&os) const;

            #ifdef CFG_PERMISSIONED_NETWORK

                diff::hash_t master_vote{0};

            #endif

            mutable mutex mx;

        };


        struct sysops_t final:unordered_map<peer_t *,shell> {

            sysops_t(daemon&d): d(d) {}

            bool process_work(peer_t *p, datagram*d);

            mutex mx;

            daemon& d;

        };

        #if CFG_COUNTERS==1

           struct counters_t final {

                uint32_t cycles{0};

                uint32_t computed_right_delta{0};

                uint32_t diff_applied{0};

                uint32_t genesis_diff_applied{0};

                uint32_t diff_discarded_not_in_sequence{0};

                uint32_t clear_db{0};

                uint32_t patch_db{0};

                uint32_t problem_opening_block_file{0};

                uint32_t problem_reading_block_file{0};

                uint32_t problem_importing_diff_file{0};

                uint32_t problem_loading_base_file{0};

                uint32_t patch_db_ok{0};

                uint32_t load_db{0};

                uint32_t problem_opening_db{0};

                uint32_t problem_reading_db{0};

                uint32_t snapshot_create{0};

                uint32_t error_renaming_db_file{0};

                uint32_t relay_local_deltas{0};

                uint32_t created_empty_local_deltas{0};

                uint32_t created_nonempty_local_deltas{0};

                uint32_t sent_vote{0};

                uint32_t recv_vote{0};

                uint32_t recv_vote_wrong_signature{0};

                uint32_t relay_vote_success{0};

                uint32_t relay_vote_fail{0};

                uint32_t recv_dup_vote{0};

                uint32_t recv_local_deltas_accepted{0};

                uint32_t recv_local_deltas_unparseable{0};

                uint32_t recv_local_deltas_wrong_signature{0};

                uint32_t recv_local_deltas_duplicate{0};

                uint32_t lost_evidences{0};

                uint32_t processed_evidences{0};

                uint32_t processed_evidences_last_block{0};

                uint32_t processed_evidences_cur_block{0};

                void dump(ostream&os) const;

            };

            static counters_t counters;

        #endif


        struct pools_t final:unordered_map<uint64_t,diff*> {

            ~pools_t();

            bool add(local_deltas* ld, bool create_diff);

            diff* close(uint64_t id);

            mutable mutex mx;

        };


        struct reg final {

            void register_arrival(uint64_t r) {

                max=r;

                sum+=r;

                ++num;

            }

            void reset() {

                sum=num=max=0;

            }

            atomic<uint64_t> sum{0};

            atomic<uint32_t> num{0};

            atomic<uint32_t> max{0};

        };


        enum evt_status_t { //values coupled in fragment_w2w.java (android app)

            evt_unknown,

            evt_error,

            evt_wait_arrival,

            evt_calendar,

            evt_mempool,

            evt_consensus,

            evt_settled,

            evt_untracked,


            evt_num

        };


        static constexpr const char* evt_status_str[evt_num]={"unknown","error","wait_arrival","calendar","mempool","consensus","settled","untracked"};


        struct ev_track_t final: unordered_map<uint64_t,pair<evt_status_t,string>> {

            pair<evt_status_t,string> track(uint64_t);

            void set_error(uint64_t ts,const string& err);

            void set_error(uint64_t from, uint64_t to, const string&err);

            void set_status(uint64_t ts,evt_status_t st);

            void set_status(uint64_t from, uint64_t to, evt_status_t st);

            void dump(ostream&) const;

            uint64_t mints{0};

            mutex mx;

        };


        typedef us::gov::dfs::fileattr_t fileattr_t;

        static void fsinfo(const string& home, bool dot, ostream& os);

        daemon(const keys&);

        daemon(const keys&, const string& home, uint16_t port, uint16_t pport, uint8_t num_edges, uint8_t devices, int workers, const vector<pair<uint32_t,uint16_t>>& seed_nodes,const string&status_file);

        daemon(const daemon&)=delete;

        daemon(daemon&&)=delete;

        void constructor();

        ~daemon() override;

        daemon& operator =(const daemon&)=delete;

        void add(app*app);

        int clean_orphaned_blocks();

        void list_apps(const string& prefix, ostream& os) const;

        static bool file_exists(const string& f);

        void eat_diff(const diff::hash_t& voted_tip, diff*);

        bool process_work(peer_t *c, datagram*d);

        bool process_query(peer_t *c, datagram*d);

        bool process_evidence(evidence*);

        bool process_sysop_request(peer_t *c, datagram*d);

        bool snapshot(diff::hash_t&);

        void print_map(ostream& os) const;

        bool wait_for_secs_to(const seconds& s, uint64_t block_deadline);

        void fsinfo(bool dot, ostream& os) const;

        void dbg_output() const;

        void vote_tip(const diff& b);

        void dump(const string& prefix, ostream& os) const;

        bool import(const diff& b, const diff::hash_t&);

        bool import_(const diff& b, const diff::hash_t&);

        bool ready_for_consensus() const;

        diff::hash_t get_last_delta_imported() const;

        void set_last_delta_imported(const diff::hash_t&, uint64_t id);

        void set_last_delta_imported_(const diff::hash_t&, uint64_t id);

        string save_db(const string& dir) const;

        void save_db(ostream& os);

        void load_db(const string& file);

        bool load_db(istream&);

        void logline(ostream&os) const;

        static char block_type(istream&);

        void list_files(ostream& os) const;

        int clean_unlisted_files();

        bool get_prev(const diff::hash_t& h, diff::hash_t& prev) const override;

        bool patch_db(vector<diff::hash_t>& patches) override;

        void on_sync() override;

        void clear() override;

        int query_block(const diff::hash_t& hash) override;

        diff::hash_t save_db();

        bool load_db(const base::hash_t&);

        void run();

        void load_head();

        void wait(const seconds& t) const;

        local_deltas* create_local_deltas(uint64_t id);

        void update_peers_state();

        void clear_();

        bool shell_command(int app_id, istream&, ostream&) const;

        void save(const diff& bl) const;

        #if CFG_COUNTERS==1

            void print_performances(ostream&) const;

            void save_dashboard() const;

            void write_status(ostream& os) const;

        #endif

        void files_to_keep(vector<pair<dfs::hash_t,uint32_t>>&) const;

        void update_dfs_index();

        void update_dfs_index_delta();

        void on_block_closure(uint64_t block_closure) override;

        bool ready_for_block_opening(uint64_t) const override;

        #include <us/api/apitool_generated__c++__protocol_gov_engine_service_handler_headers>

        static string get_blocksdir(const string& govhome);

        static string get_fsrootdir(const string& govhome);

        string uptime() const;

        ko start() override;

        ko wait_ready(const chrono::system_clock::time_point& deadline) const override;

        void stop() override;

        void join() override;

        void onwakeup();

        void watch(ostream& os) const;

        void write_status(const string& line) const;

        void exec_evidence(const evidence& e) override;


        diff::hash_t last_vote{0};

        diff::hash_t last_consensus{0};

        condition_variable cv_cyc;

        string home;

        keys id;

        chrono::system_clock::time_point running_since;

        apps apps_;

        auth::app* auth_app;

        us::gov::sys::app* sys_app;

        us::gov::cash::app* cash_app;

        networking peerd;

        sysops_t sysops;

        bool sysop_allowed{false};

        bool encrypt_protocol{false};

        local_deltas* _local_deltas{0};

        votes_t votes;

        #if CFG_COUNTERS==1

            struct silent_t:set<diff::hash_t> {

                void dump(ostream&os) const;

            };

            silent_t silent;

        #endif

        pools_t pools;

        syncd_t syncd;

        neuralyzer_t neuralyzer;

        diff::hash_t last_delta_imported;

        uint64_t last_delta_imported_id;

        mutable mutex mx_import;

        uint16_t depth{0};

        string blocksdir;

        reg votes_timing;

        reg local_deltas_timing;

        #if CFG_LOGS==1

            string logdir;

        #endif

        string status_file;

        ev_track_t evt;

        mutable mutex mx_dashboard;

    };


}}

}


#endif