rock/rock.h

/*--------------------------------------------------------------------------

  This is part of the Distributed system of the Karoo project.

  (C) 2006,2007,2008,2009 Brian Modra <brian@zwartberg.com>

  This library is free software; you can redistribute them and/or modify
  it under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation; either version 2.1 of the License,
  or (at your option) any later version.

  This library 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.

  You should have received a copy of the GNU Lesser General Public License
  along with these libraries; if not, write to the Free Software Foundation,
  Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

--------------------------------------------------------------------------*/



#ifndef rock_h
#define rock_h

#ifndef _REENTRANT
#define _REENTRANT
#endif

#include "config.h"
#include "datagram.h"
#include "base.h"
#include "data.h"
#include "util.h"
#include "protocol.h"
#include <map>
#include <vector>
#include <list>
#include <glib.h>
#include <stdint.h>
#include <stdexcept>
#include <sys/types.h>
#include <unistd.h>

using std::vector;
using std::list;
using std::map;
using std::multimap;
using std::exception;

namespace karoo {
    class rock;
    class rock_datagram_message;

    class rock_datagram_reply : public pebble
    {
    private:
        void setAckMessage(rock_datagram_message* ack_msg);

    protected:
        friend class rock;
        rock_datagram_message* msg;
        rock_datagram_message* ack_msg;

    public:
        rock_datagram_reply(rock_datagram_message* msg);
        virtual ~rock_datagram_reply();

        virtual void run() = 0;
    };

    class rock_datagram_message : public datagram_message, public rock_data
    {
    protected:
        size_t pos;
        rock* parent;

        rock_datagram_reply* reply;

        text recipient_name;

        size_t readFrom(void* data, size_t size, size_t p) const;
    public:
        rock_datagram_message(size_t alloc_size, rock* parent);
        rock_datagram_message(size_t alloc_size, const text& ip, int port, rock* parent);
        virtual ~rock_datagram_message();

        datagram_message* clone(size_t alloc_size = 0) const;
        datagram_message* clone(int port, size_t alloc_size = 0) const;

        size_t getReadPos() const { return pos; }

        void resetReadPos(size_t pos = 0) { this->pos = pos; }
        
        size_t available() { return getSize() - pos; }

        void write(const void* data, size_t size);
        void read(void* data, size_t size);

        void accomodate(size_t bytes);


        uint32_t getType();
        uint64_t getSequence();
        int32_t getSourcePort();
        text getName();

        void setReply(rock_datagram_reply* reply);

        rock_datagram_reply* getReply();

        text getHost() const; 
        int32_t getPort() const;

        void run();


        void setRecipientName(const text& recipient_name) { this->recipient_name = recipient_name; }

        text getRecipientName() const { return recipient_name; }
    };

    class rock_datagram_factory : public datagram_message_factory
    {
    private:
    protected:
        rock* parent;
    public:
        rock_datagram_factory(rock* parent) { this->parent = parent; }
        virtual ~rock_datagram_factory() {}
        
        datagram_message* create_datagram_message(size_t alloc_size) {
            return new rock_datagram_message(alloc_size, parent);
        }
    };

    class rock_datagram_message_handler : public referable
    {
    protected:
        rock* parent;
    public:
        rock_datagram_message_handler(rock* parent) { this->parent = parent; }
        virtual ~rock_datagram_message_handler() {}

        rock_datagram_message* createAckMessage(const rock_datagram_message* msg, uint64_t seq, int32_t port);
        void sendAck(rock_datagram_message* reply);
        void createAndSendAck(const rock_datagram_message* msg, uint64_t seq, int32_t port);
        rock_datagram_message* createReplyMessage(const rock_datagram_message* msg, int32_t port, const text& name, uint32_t type);
        void sendMessage(rock_datagram_message* msg);
        virtual bool handle(uint32_t message_type, uint64_t seq, int32_t port, const text& name, rock_datagram_message* msg) = 0;
    };

    class default_rock_datagram_message_handler : public rock_datagram_message_handler
    {
    private:
        synchronised<uint64_t> bye_seq;
        friend class rock;
    public:
        default_rock_datagram_message_handler(rock* parent);
        virtual ~default_rock_datagram_message_handler();
        virtual bool handle(uint32_t message_type, uint64_t seq, int32_t port, const text& name, rock_datagram_message* msg);
    };

    class message_handler_runner : public pebble
    {
    protected:
        rock* parent;
        rock_datagram_message* msg;
        uint32_t msg_type;
        uint64_t seq;
        int32_t port;
        text name;
    public:
        message_handler_runner(rock* parent, rock_datagram_message* msg, uint32_t msg_type, uint64_t seq, int32_t port, const text& name);
        virtual ~message_handler_runner();
        void run();
    };

    class rock_long_message : public referable
    {
    protected:
        rock* parent;
        rock_datagram_message* hdr;
        uint64_t seq;
        size_t size;
        size_t num_packets;
        unsigned char** packets;
        size_t* packet_size;
    public:
        rock_long_message(rock* parent, rock_datagram_message* hdr, uint64_t seq, size_t size, size_t num_packets);
        virtual ~rock_long_message();

        void addPacket(size_t pkt_num, size_t this_size, unsigned char* data);
        bool gotAllPackets() const;
        void dispatch() const;
    };

    class cached_sequence {
    protected:
        uint64_t seq;
        time_t ts;
    public:
        cached_sequence();
        cached_sequence(uint64_t seq);
        cached_sequence(const cached_sequence& ref);
        virtual ~cached_sequence();

        uint64_t getSeq() const { return seq; }
        time_t getTime() const { return ts; }
        void touch();

        const cached_sequence& operator=(const cached_sequence& ref);
    };

    class remote_rock : public referable {
    protected:
        mutable karoo_mutex sem;

        rock* parent;
        text name;
        text type;
        text host;
        int32_t port;
        pid_t pid;
        map<text,text> args;
        text path;
        text file;
        time_t dying;
        int8_t load_factor;
        map<uint64_t,cached_sequence> recent_seq_array;
        time_t validated;
        time_t executed;
        time_t startup_time_allowance;
        time_t shutdown_time_allowance;
        time_t pinged;
        time_t exe_file_mtime;
        map<uint64_t,rock_long_message*> long_array;
        karoo_mutex waiting_sem;
        list<rock_datagram_message*> waiting;

        class ping_reply : public rock_datagram_reply
        {
        private:
            remote_rock* parent;
        public:
            ping_reply(remote_rock* parent, rock_datagram_message* msg)
                : rock_datagram_reply(msg) { this->parent = parent; }
            virtual ~ping_reply() {}
            void run();
        };

    public:
        remote_rock(rock* parent, const text& name);
        remote_rock(rock* parent, const remote_rock& ref);
        remote_rock(rock* parent, const text& name, const text& type, const text& host, int32_t port, int8_t load_factor, time_t startup_time_allowance = (time_t)15, time_t shutdown_time_allowance = (time_t)15);
        virtual ~remote_rock();

        bool operator==(const remote_rock& ref) const;
        
        void setArgs(const map<text,text>& args);
        const map<text,text>& getArgs() const { return args; }

        void setPath(const text& path) { this->path = path; }
        void setFile(const text& file) { this->file = file; }
        text getPath() const { return path; }
        text getFile() const { return file; }

        text getName() const { return name; }
        text getType() const { return type; }
        text getHost() const { return host; }
        int32_t getPort() const { return port; }
        uint8_t getLoadFactor() const { return load_factor; }

        void hello(const text& type, const text& host, int32_t port);
        void validate();
        void validate(const text& host, int32_t port);
        void validate(const text& type, const text& host, int32_t port, uint8_t load_factor);
        void invalidate();
        time_t getValidated() const;
        time_t getExecuted() const;
        time_t getStartupTimeAllowance() const;
        time_t getShutdownTimeAllowance() const;
        bool getExeFileChanged();
        time_t getPinged() const;

        bool containsSeq(uint64_t seq) const;
        void addSeq(uint64_t seq);

        void addWaiting(rock_datagram_message* to_send);
        void sendWaiting();
        void nackWaiting();

        rock_long_message* addLongMessagePacket(uint64_t hdr_seq, size_t pkt_num, size_t this_size, unsigned char* data);
        void addLongMessageHeader(rock* parent, rock_datagram_message* hdr_msg, uint64_t seq, size_t long_size, size_t num_packets);

        void execute();

        void kill();

        void ping();

        time_t inDeathThrows() const;

        pid_t getPID() const;
    };

    class rock_datagram_message_ack : public pebble
    {
    private:
        rock* parent;
        uint64_t seq;
        rock_datagram_message* msg;
        int retry_count;
        int retry_delay;

    public:
        rock_datagram_message_ack(rock* parent, uint64_t seq, rock_datagram_message* msg, int retry_delay);
        virtual ~rock_datagram_message_ack();

        void run();
    };


    typedef struct {
        text key;
        text value;
    } argtuple;

    class rock
    {
    private:
        argtuple* args;
        int args_size;

        vector<text> arguments;

        text name;

        text host;
        int port;
        int self_port;
        int ack_retry;
        int max_retry;
        size_t message_size;
        size_t long_message_burst_size;

        synchronised<uint64_t> sequence;

        karoo_mutex ack_sem;
        map<uint64_t,rock_datagram_message*> ack_array;
        friend void rock_datagram_message_ack::run();
        map<uint64_t,rock_datagram_message*> long_ack_array;

        exepool* general_pool;

        datagram_server* server;
        datagram_client* client;
        friend void rock_datagram_message_handler::sendAck(rock_datagram_message* reply);

        exeque* output_queue;
        exeque* run_queue;
        rock_datagram_factory* factory;
        default_rock_datagram_message_handler* handler;
        vector<rock_datagram_message_handler*> handlers;
        friend void message_handler_runner::run();

        virtual rock_datagram_message* createRockByeMessage();

        mutable karoo_mutex bye_sem;
        bool stopping;

        class rock_bye: public pebble
        {
        private:
            rock_datagram_message* bye;
            uint64_t bye_seq;
            rock* parent;
            log logger;
        public:
            rock_bye(rock* parent, const log& logger);
            virtual ~rock_bye();
            
            uint64_t getSequence() { return bye_seq; }
            
            void run();
        };

        class long_runner : public pebble
        {
        private:
            rock* parent;
            karoo_mutex sem;
            map<uint64_t,rock_datagram_message*> waiting;
            vector<rock_datagram_message*> packets;
            size_t long_message_burst_size;
            size_t min_size;
            rock_datagram_message* msg;
            bool nacked;
            
        public:
            long_runner(rock* parent, size_t long_message_burst_size, rock_datagram_message* msg);
            virtual ~long_runner();
            void ack(uint64_t seq);
            void nack();
            void add(rock_datagram_message* pkt);
            void run();
        };

        class long_hdr_reply : public rock_datagram_reply
        {
        private:
            rock* parent;
            long_runner* runner;
            
        public:
            long_hdr_reply(rock* parent, long_runner* runner, rock_datagram_message* msg);
            virtual ~long_hdr_reply();
            void run();
        };

        class long_runner;

        class long_pkt_reply : public rock_datagram_reply
        {
        private:
            rock* parent;
            long_runner* runner;

        public:
            long_pkt_reply(rock* parent, long_runner* runner, rock_datagram_message* msg);
            virtual ~long_pkt_reply();
            void run();
        };

        class rock_keep_connected: public pebble
        {
        private:
            time_t poll_secs;
            rock* parent;
            log logger;

            class rock_keep_connected_reply : public rock_datagram_reply
            {
            private:
                rock_keep_connected* parent;
            public:
                rock_keep_connected_reply(rock_keep_connected* parent, rock_datagram_message* msg)
                    : rock_datagram_reply(msg) { this->parent = parent; parent->reference(); }
                virtual ~rock_keep_connected_reply() { parent->dereference(); }
                void run() { parent->acked(); }
            };

        public:
            rock_keep_connected(rock* parent, const log& logger, time_t poll_secs);
            virtual ~rock_keep_connected() {}

            void acked();
            void run();
        };


        mutable karoo_mutex children_sem;
        map<text,remote_rock*> children;
        multimap<text,remote_rock*> children_by_type;
    protected:
        void getArguments(const vector<text>& opts, int argc, char *argv[]);
        virtual datagram_status create_client();
        virtual datagram_status create_server();
        virtual datagram_status  init();

        log logger;

        remote_rock* findRockByType(const text& type);

        bool syncChildren(map<text,remote_rock*>& rocks, vector<remote_rock*>& rocks_arr);

    public:
        rock();
        
        rock(const vector<text>& opts, int argc, char *argv[]);

        virtual ~rock();


        datagram_status run();

        void stop(bool block = true);

        void refuse() { output_queue->refuse(); }

        void bye();

        bool isStopping() const;

        void add(pebble* p);
        void add(const struct timespec& ts, pebble* p);
        void add(int after_seconds, pebble* p);

        int getImminentCount(unsigned usecs = 0) const;

        int getQueueCount() const;

        exepool* getGeneralPool() { return general_pool; }

        const text& getArgument(const text& opt) const;
        const text& getArgument(const text& opt, const text& def) const;
        bool getBoolArgument(const text& opt, bool def) const;

        virtual void handle(rock_datagram_message* msg);

        void addHandler(rock_datagram_message_handler* h);


        virtual text getType() const { return "rock"; }

        /*
         * return the load factor for this rock. The default implementation always returns 0 (no load).
         */
        virtual uint8_t getLoadFactor() const { return 0; }

        const text& getHost() const { return host; }
        int getPort() const { return port; }
        int getSelfPort() const { return self_port; }
        text getName() const { return name; }
        int getAckRetry() const { return ack_retry; }
        int getMaxRetry() const { return max_retry; }
        size_t getMessageSize() const { return message_size; }
        log getLogger() const { return logger; }
        enum log_level getLoggerLevel() const { return (enum log_level)logger; }

        virtual rock_datagram_message* createMessage(uint32_t msg_type, text recipient_host = null_string, int recipient_port = 0, size_t message_size = 0);
        virtual rock_datagram_message* createRockHelloMessage();
        virtual rock_datagram_message* createRockPingMessage();

        void startKeepConnectedHello(time_t poll_secs = 5, time_t initial_delay = 0);

        void sendMessageByAddress(rock_datagram_message* msg);

        void sendMessage(rock_datagram_message* msg);

        bool sendMessageByType(rock_datagram_message* msg, const text& type);

        void doHello(const text& name, const text& type, const text& host, int port);
        void doPing(const text& name, const text& host, int port, bool create = false);
        void doPing(const text& name, const text& type, const text& host, int port, uint8_t load_factor);
        void doBye(const text& name);
        void doLongHeader(const text& name, const text& host, int port, uint32_t long_size, uint32_t num_packets, rock_datagram_message* msg, uint64_t seq);
        bool doLongPacket(const text& name, const text& host, int port, uint64_t hdr_seq, uint32_t pkt_num, uint32_t this_size, unsigned char* data);

        void notifyAcked(uint64_t seq, rock_datagram_message* msg);

        uint64_t getSequence() { return ++sequence; }

        bool getRemoteRockType(const text& name, text& type) const;
    };
}

#endif

---