SharedMemCatalog.h

Go to the documentation of this file.

/*  Copyright (C) 2003-2021 Free Electron Organization
    Any use of this software requires a license.  If a valid license
    was not distributed with this file, visit freeelectron.org. */

/** @file */

#ifndef __shm_SharedMem_h__
#define __shm_SharedMem_h__

#define FE_SMC_COMM_COUNT       8
#define FE_SMC_COMMAND_BYTES    1024

namespace fe
{
namespace ext
{

/**************************************************************************//**
    @brief ConnectedCatalog over ZeroMQ

    @ingroup shm
*//***************************************************************************/
class FE_DL_EXPORT SharedMemCatalog:
    public ConnectedCatalog,
    public Initialize<SharedMemCatalog>
{
    public:

                SharedMemCatalog(void);
virtual         ~SharedMemCatalog(void);

        void    initialize(void);

    protected:

virtual Result  disconnect(void) override;
virtual Result  dropConnection(void);

                using ConnectedCatalog::removeIdentity;
virtual Result  removeIdentity(I32 a_index) override;

virtual Result  connectAsServer(String a_address,U16 a_port) override;
virtual Result  connectAsClient(String a_address,U16 a_port) override;

virtual Result  catchUp(Identity& a_rIdentity);

    private:

virtual void    broadcastSelect(String a_name,String a_property,
                        String a_message,
                        I32 a_includeCount,const String* a_pIncludes,
                        I32 a_excludeCount,const String* a_pExcludes,
                        const U8* a_pRawBytes=NULL,I32 a_byteCount=0) override;

    struct Connect
    {
        public:
            sem_t       m_semStart;
            sem_t       m_semClient;
            sem_t       m_semIdentity;
            sem_t       m_semAck;
            Identity    m_identity;
    };

    struct Comm
    {
        public:
            sem_t   m_semWrite;         //* last reader should post
            sem_t   m_semRead;          //* one per intended recipient
            sem_t   m_semCount;         //* for m_readerCount and m_serial
            I32     m_readerCount;
            I32     m_serial;

            char    m_include[32];
            char    m_exclude[32];
            char    m_command[8];
            char    m_key[64];
            char    m_property[32];
            char    m_type[32];
            I32     m_byteCount;
            U8      m_pData[FE_SMC_COMMAND_BYTES];
    };
    struct Share
    {
        public:
            Connect m_connect;
            Comm    m_serverComm[FE_SMC_COMM_COUNT];
            Comm    m_clientComm[FE_SMC_COMM_COUNT];
    };
    class Channel
    {
        public:
                                Channel(void):
                                    m_pShare(NULL),
                                    m_serverIndex(0),
                                    m_clientIndex(0),
                                    m_sendSerial(0),
                                    m_receiveSerial(-1),
                                    m_discarded(FALSE)                      {}
    virtual                     ~Channel(void)
                                {
                                    if(m_pShare)
                                        munmap(m_pShare,sizeof(Share));
                                }
            void                reset(void)
                                {
                                    if(m_pShare)
                                        munmap(m_pShare,sizeof(Share));
                                    m_pShare=NULL;
                                    m_serverIndex=0;
                                    m_clientIndex=0;
                                    m_sendSerial=0;
                                    m_receiveSerial=-1;
                                    m_discarded=FALSE;
                                }

            Share*              m_pShare;
            std::deque<Comm>    m_commBacklog;
            I32                 m_serverIndex;
            I32                 m_clientIndex;
            I32                 m_sendSerial;
            I32                 m_receiveSerial;
            BWORD               m_discarded;
    };

    //* TODO additional clients need to know what serverIndex to start with

    class FE_DL_EXPORT ReceiverTask: public Thread::Functor
    {
        public:
                    ReceiverTask(void);
    virtual         ~ReceiverTask(void);

            void    reset(void);

    virtual void    operate(void);

            Result  connectAsServer(String a_transport,
                            String a_address,U16 a_port);
            Result  connectAsClient(String a_transport,
                            String a_address,U16 a_port);

            Result  startShare(String a_shareName,Share*& a_rpShare,
                            BWORD a_create);

            I32     writeBacklogs(void);
            I32     writeBacklog(Channel& a_rChannel);

            BWORD   receiveUpdate(const String& a_source,
                            SharedMemCatalog::Channel& a_rChannel,
                            SharedMemCatalog::Comm& a_rComm);

            void    decrementReaderCounts(void);
            BWORD   decrementReaderCount(SharedMemCatalog::Comm& a_rComm);
            void    dumpSemaphores(SharedMemCatalog::Channel& a_rChannel);

            SharedMemCatalog*               m_pSharedMemCatalog;
            String                          m_endpoint;
            std::atomic<bool>               m_terminate;
            std::atomic<I32>                m_readerDecrementCount;
            Identity                        m_identity;
            Channel                         m_publicChannel;
            AutoHashMap<String,Channel>     m_privateChannelMap;
            BWORD                           m_receivePublic;
            BWORD                           m_isServer;
    };

        BWORD               m_usePublicChannel;
        Thread*             m_pReceiverThread;
        ReceiverTask        m_receiverTask;
};

} /* namespace ext */
} /* namespace fe */

#endif /* __shm_SharedMem_h__ */