DataStoreStreamer.h

/**************************************************************************
 * basf2 (Belle II Analysis Software Framework)                           *
 * Author: The Belle II Collaboration                                     *
 *                                                                        *
 * See git log for contributors and copyright holders.                    *
 * This file is licensed under LGPL-3.0, see LICENSE.md.                  *
 **************************************************************************/
 
#pragma once
 
#include <framework/pcore/EvtMessage.h>
 
#include <Rtypes.h> //for BIT()
 
#include <pthread.h>
 
#include <vector>
#include <string>
 
class TList;
 
namespace Belle2 {
  class MsgHandler;
 
  class DataStoreStreamer {
  public:
    static const unsigned int c_maxThreads = 16;
    static const unsigned int c_maxQueueDepth = 64;
 
    explicit DataStoreStreamer(int complevel = 0, bool handleMergeable = true, int maxthread = 0);
    DataStoreStreamer(const DataStoreStreamer&) = delete;
    DataStoreStreamer& operator=(const DataStoreStreamer&) = delete;
 
    ~DataStoreStreamer();
 
    // DataStore->EvtMessage
    EvtMessage* streamDataStore(bool addPersistentDurability, bool streamTransientObjects = false);
 
 
    // EvtMessage->DataStore
    int restoreDataStore(EvtMessage* msg);
 
    void setStreamingObjects(const std::vector<std::string>& list);
 
    // Pipelined destreaming of EvtMessage using thread
 
    int queueEvtMessage(char* msg);
 
    void* decodeEvtMessage(int id);
 
    int restoreDataStoreAsync();
 
    void setMaxThreads(int);
    int  getMaxThreads();
    void setDecoderStatus(int);
    int  getDecoderStatus();
 
    static bool isMergeable(const TObject* object);
 
    static void clearMergeable(TObject* object);
 
    static void mergeIntoExisting(TObject* existing, const TObject* received);
 
    static void removeSideEffects();
 
  private:
 
    int restoreStreamerInfos(const TList* list);
 
    enum ETObjectBits {
      c_IsTransient = BIT(19), 
      c_IsNull = BIT(20), 
      c_PersistentDurability = BIT(21) 
    };
 
    MsgHandler* m_msghandler;
 
    int m_compressionLevel;
 
    bool m_handleMergeable; 
    int m_initStatus;
 
    std::vector<std::string> m_streamobjnames;
 
    int m_maxthread;
 
    pthread_t m_pt[c_maxThreads];
    int m_id[c_maxThreads];
    int m_threadin;
    //int m_threadout;
    int m_decoderStatus[c_maxThreads] {0};
    //MsgHandler* m_pmsghandler[c_maxThreads];
    //    char* m_evtbuf[c_maxThreads];
    //std::queue<char*> m_evtbuf[c_maxThreads];
 
  };
 
  // Function to hook DataStoreStreamer::decodeEvtMessage to pthread
 
  //  void* RunDecodeEvtMessage ( void* );
 
 
} // namespace Belle2