sharedmem.h

Go to the documentation of this file.

#ifndef sharedmem_HEADER_GUARD
#define sharedmem_HEADER_GUARD
/*
 * sharedmem.h : Posix shared memory segment server/client management, shared memory ring buffer synchronized using posix semaphores.
 * 
 * (c) Copyright 2017-2024 Sampsa Riikonen
 * 
 * Authors: Sampsa Riikonen <sampsa.riikonen@iki.fi>
 * 
 * This file is part of the Valkka library.
 * 
 * Valkka is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 * 
 * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>
 *
 */
 
#include "common.h"
#include <sys/shm.h>
#include <sys/mman.h>
#include <sys/stat.h>        /* For mode constants */
#include <fcntl.h>           /* For O_* constants */
#include <semaphore.h> // semaphores
#include "framefilter.h"
#include "Python.h"
 
 
// #define USE_SHMEM_CACHE 1 // Don't expose shmem segments directly as numpy arrays, but an intermediate copy instead
// #define SAFE_TEXT
 
#ifdef SAFE_TEST
static std::mutex test_mutex;
#endif
 
class EventFd {     // <pyapi>
 
public:             // <pyapi>
    EventFd();      // <pyapi>
    ~EventFd();     // <pyapi>
 
private:
    int fd;
 
public:             // <pyapi>
    int getFd();    // <pyapi>
    void set();     // <pyapi>
    void clear();   // <pyapi>
};                  // <pyapi>
 
 
// https://stackoverflow.com/questions/115703/storing-c-template-function-definitions-in-a-cpp-file
class SharedMemSegment {
 
public:
  SharedMemSegment(const char* name, std::size_t n_bytes, bool is_server=false);
  virtual ~SharedMemSegment();
  
protected:
    virtual void serverInit() = 0;      
    virtual bool clientInit() = 0;      
    virtual void serverClose() = 0;     
    virtual void clientClose() = 0;
  
public:
    virtual std::size_t getSize() = 0;        
    
    virtual void put(std::vector<uint8_t> &inp_payload, void *meta_) = 0; 
    virtual void put(uint8_t* buf_, void* meta_) = 0;
 
    virtual void copyMetaFrom(void *meta_) = 0; 
    virtual void copyMetaTo(void *meta_)   = 0; 
  
protected:
    std::string   name;         
    std::string   payload_name; 
    std::string   meta_name;    
    bool          is_server;    
    void          *ptr,*ptr_;   
    bool          client_state; 
    
public:
    uint8_t     *payload;   
    // void        *meta;      ///< Metadata (define in child classes);
    std::size_t n_bytes;    
    
public: // client
    void init();                           
    void close_();                         
    bool getClientState();                 
};
 
 
class SimpleSharedMemSegment : public SharedMemSegment {
   
public:
    SimpleSharedMemSegment(const char* name, std::size_t n_bytes, bool is_server=false);
    virtual ~SimpleSharedMemSegment();
 
protected:
    virtual void serverInit();       
    virtual bool clientInit();       
    virtual void serverClose();      
    virtual void clientClose();
    
public:
    virtual std::size_t  getSize();               
 
    /* virtual method, since metadata depends on the shared mem segment type */
    virtual void put(std::vector<uint8_t> &inp_payload, void* meta_); 
    virtual void put(uint8_t* buf_, void* meta_);
    virtual void putAVRGBFrame(AVRGBFrame *f);                        
 
    virtual void copyMetaFrom(void *meta_);       
    virtual void copyMetaTo(void *meta_);         
  
public:
    std::size_t  *meta;
    
public:
    void         put(std::vector<uint8_t> &inp_payload);              
    
};
 
 
struct RGB24Meta {                                  // <pyapi>
    std::size_t size; 
    int width;                                      // <pyapi>
    int height;                                     // <pyapi>
    SlotNumber slot;                                // <pyapi>
    long int mstimestamp;                           // <pyapi>
};                                                  // <pyapi>
 
 
class RGB24SharedMemSegment : public SharedMemSegment {
    
public:
    RGB24SharedMemSegment(const char* name, int width, int height, bool is_server=false);
    virtual ~RGB24SharedMemSegment();
    
protected:
    virtual void serverInit();       
    virtual bool clientInit();       
    virtual void serverClose();      
    virtual void clientClose();
    
public:
    virtual std::size_t getSize();                                    
  
    virtual void put(std::vector<uint8_t> &inp_payload, void* meta_); 
    virtual void put(uint8_t* buf_, void* meta_);
    virtual void putAVRGBFrame(AVRGBFrame *f);                        
 
    virtual void copyMetaFrom(void *meta_);       
    virtual void copyMetaTo(void *meta_);         
  
public:
    RGB24Meta    *meta;
    
};
 
 
class SharedMemRingBufferBase { // <pyapi>
 
public: // <pyapi>
    SharedMemRingBufferBase(const char* name, int n_cells, std::size_t n_bytes, int mstimeout=0, bool is_server=false); // <pyapi>
    virtual ~SharedMemRingBufferBase(); // <pyapi>
 
protected: // at constructor init list
    std::string  name;
    int          n_cells, n_bytes; 
    int          mstimeout;        
    bool         is_server;        
    
protected: // posix semaphores and mmap'd files
    sem_t         *sema, *flagsema;
    std::string   sema_name;       
    std::string   flagsema_name;   
    int  index;                    
    struct timespec ts;            
    int fd;                        
 
public:
    std::vector<SharedMemSegment*> shmems; 
    uint8_t** cache; 
 
protected: // internal methods - not for the api user
    void  setFlag();       
    bool  flagIsSet();     
    void  clearFlag();     
    int   getFlagValue();  
    void  zero();          
    void  setEventFd();    
    void  clearEventFd();  
 
 
public:
  int   getValue();       
  bool  getClientState(); 
  
public: // server side routines - call these only from the server  // <pyapi>         
    void serverPush(std::vector<uint8_t> &inp_payload, void* meta);
    void serverUseFd(EventFd &event_fd);                            // <pyapi>
 
 
public: // client side routines - call only from the client side // <pyapi>
    bool clientPull(int &index_out, void* meta);                // <pyapi>
    bool clientPullThread(int &index_out, void* meta);          // <pyapi>
    PyObject *getBufferListPy();                                // <pyapi>
    void clientUseFd(EventFd &event_fd);                        // <pyapi>
}; // <pyapi>
 
 
 
class SharedMemRingBuffer : public SharedMemRingBufferBase { // <pyapi>
 
public: // <pyapi>
    SharedMemRingBuffer(const char* name, int n_cells, std::size_t n_bytes, int mstimeout=0, bool is_server=false); // <pyapi>
    virtual ~SharedMemRingBuffer(); // <pyapi>
    
public:                                                     // <pyapi>
    void serverPush(std::vector<uint8_t> &inp_payload);     // <pyapi>
    bool serverPushPy(PyObject *po);                        // <pyapi>
    bool clientPull(int &index_out, int &size_out);         // <pyapi>
};                                                          // <pyapi>
 
 
 
class SharedMemRingBufferRGB : public SharedMemRingBufferBase { // <pyapi>
 
public:                                                     // <pyapi>
    SharedMemRingBufferRGB(const char* name, int n_cells, int width, int height, int mstimeout=0, bool is_server=false); // <pyapi>
    virtual ~SharedMemRingBufferRGB(); // <pyapi>
 
protected:
    int width, height;
 
public:
    void serverPushAVRGBFrame(AVRGBFrame *f);
  
public:                              // <pyapi>
    PyObject* clientPullPy();                                     // <pyapi>
    
    bool clientPull(int &index_out, int &size_out);               // <pyapi>
    
    bool clientPullFrame(int &index_out, RGB24Meta &meta);        // <pyapi>
    bool clientPullFrameThread(int &index_out, RGB24Meta &meta);  // <pyapi>
    // ..not sure about passing RGB24Meta from python
    // by reference.  Not sure if swig handles the refcounting
    // correctly.
 
    bool serverPushPyRGB(PyObject *po, SlotNumber slot, long int mstimestamp); // <pyapi>
    
    // in these versions, modify the python object in-place: TODO
    //bool clientPullFramePy(int &index_out, PyObject *po)          // <pyapi>
    //bool clientPullFrameThreadPy(int &index_out, PyObject *po)    // <pyapi>
};                                                                // <pyapi>
 
 
 
class ShmemFrameFilter : public FrameFilter {                                               // <pyapi>
  
public:                                                                                     // <pyapi>
    ShmemFrameFilter(const char* name, int n_cells, std::size_t n_bytes, int mstimeout=0);  // <pyapi>
    //~ShmemFrameFilter(); // <pyapi>
 
protected: // initialized at constructor                                                  
    //int                    n_cells;
    //std::size_t            n_bytes;
    //int                    mstimeout;
    SharedMemRingBuffer shmembuf;
    
protected:
    virtual void go(Frame* frame);
    
public:                                                                                   // <pyapi>
    void useFd(EventFd &event_fd);                                                        // <pyapi>
};                                                                                        // <pyapi>
  
 
class RGBShmemFrameFilter : public FrameFilter { // <pyapi> 
  
public: // <pyapi>
    RGBShmemFrameFilter(const char* name, int n_cells, int width, int height, int mstimeout=0); // <pyapi>
    //~RGBShmemFrameFilter(); // <pyapi>
  
protected: // initialized at constructor                                                  
    //int                    n_cells;
    //int                    width;
    //int                    height;
    //int                    mstimeout;
    SharedMemRingBufferRGB shmembuf;
  
protected:
    virtual void go(Frame* frame);
 
public:                                                                                   // <pyapi>
    void useFd(EventFd &event_fd);                                                        // <pyapi>
}; // <pyapi>
  
 
// macros
 
#define server_init(CLASSNAME, TYPENAME)\
void CLASSNAME::serverInit() {\
  int fd, fd_, r, r_;\
  shm_unlink(payload_name.c_str());\
  shm_unlink(meta_name.c_str());\
  fd = shm_open(payload_name.c_str(),O_CREAT | O_EXCL | O_TRUNC | O_RDWR, 0600);\
  fd_= shm_open(meta_name.c_str(),   O_CREAT | O_EXCL | O_TRUNC | O_RDWR, 0600);\
  if (fd == -1 or fd_==-1) {\
    perror("valkka_core: sharedmem.cpp: SharedMemSegment::serverInit: shm_open failed");\
    exit(2);\
  }\
  /* std::cout << "got shmem" << std::endl; */\
  r = ftruncate(fd, n_bytes);\
  /* r_= ftruncate(fd_,n_bytes); // BUG!*/\
  r_= ftruncate(fd_, sizeof(TYPENAME));\
  if (r != 0 or r_ !=0) {\
    perror("valkka_core: sharedmem.cpp: SharedMemSegment::serverInit: ftruncate failed");\
    exit(2);\
  }\
  ptr = mmap(0, n_bytes,          PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);\
  ptr_= mmap(0, sizeof(TYPENAME), PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);\
  if (ptr == MAP_FAILED or ptr_ == MAP_FAILED) {\
    perror("valkka_core: sharedmem.cpp: SharedMemSegment::serverInit: mmap failed");\
    exit(2);\
  }\
  payload = (uint8_t*) ptr;\
  meta    = (TYPENAME*) ptr_;\
  /*std::cout << "server meta reserved" << std::endl;*/\
  close(fd);\
  close(fd_);\
  /*std::cout << "Server reserved " << sizeof(TYPENAME) << "bytes" << std::endl;*/\
};\
 
 
#define client_init(CLASSNAME, TYPENAME) \
bool CLASSNAME::clientInit() { \
  int fd, fd_;\
  fd = shm_open(payload_name.c_str(), O_RDONLY, 0400);\
  fd_= shm_open(meta_name.c_str(), O_RDONLY, 0400);\
  if (fd == -1 or fd_==-1) {\
    /* perror("valkka_core: sharedmem.cpp: SharedMemSegment::clientInit: shm_open failed"); */\
    /* exit(2); */\
    return false;\
  }\
  /* std::cout << "got shmem" << std::endl; */\
  ptr  =mmap(0, n_bytes, PROT_READ, MAP_SHARED, fd,  0);\
  ptr_ =mmap(0, sizeof(TYPENAME), PROT_READ, MAP_SHARED, fd_, 0);\
  if (ptr == MAP_FAILED or ptr_ == MAP_FAILED) {\
    std::cout << "valkka_core: sharedmem.cpp: SharedMemSegment::clientInit: mmap failed" << std::endl;\
    /* perror("valkka_core: sharedmem.cpp: SharedMemSegment::clientInit: mmap failed"); */\
    /* exit(2);*/\
    return false;\
  }\
  payload = (uint8_t*) ptr;\
  meta    = (TYPENAME*) ptr_;\
  /*std::cout << "client meta reserved" << std::endl;*/\
  close(fd);\
  close(fd_);\
  /*std::cout << "Client reserved " << sizeof(TYPENAME) << "bytes" << std::endl;*/\
  return true;\
};\
 
 
#define server_close(CLASSNAME, TYPENAME) \
void CLASSNAME::serverClose() { \
 if (munmap(ptr, n_bytes)!=0 or munmap(ptr_, sizeof(TYPENAME)!=0)) { \
    perror("valkka_core: sharedmem.cpp: CLASSNAME::serverClose: munmap failed"); \
    exit(2); \
  } \
  if (shm_unlink(payload_name.c_str())!=0 or shm_unlink(meta_name.c_str())!=0) { \
    perror("valkka_core: sharedmem.cpp: CLASSNAME::serverClose: shm_unlink failed"); \
    /*exit(2);*/ \
  } \
} \
 
#define client_close(CLASSNAME, TYPENAME) \
void CLASSNAME::clientClose() { \
if (munmap(ptr, n_bytes)!=0 or munmap(ptr_, sizeof(TYPENAME)!=0)) { \
    perror("valkka_core: sharedmem.cpp: CLASSNAME::clientClose: munmap failed"); \
    exit(2); \
  } \
} \
 
 
#define copy_meta_from(CLASSNAME, TYPENAME) \
void CLASSNAME::copyMetaFrom(void *meta_) { \
    *meta = *((TYPENAME*)(meta_)); \
}\
 
#define copy_meta_to(CLASSNAME, TYPENAME) \
void CLASSNAME::copyMetaTo(void *meta_) { \
    *((TYPENAME*)(meta_)) = *meta; \
}\
 
#endif