Stream/restore DataStore objects to/from EvtMessage. More...

#include <DataStoreStreamer.h>

Collaboration diagram for DataStoreStreamer:

[legend]

Public Member Functions
	DataStoreStreamer (int complevel=0, bool handleMergeable=true, int maxthread=0)
	Constructor.

	DataStoreStreamer (const DataStoreStreamer &)=delete
	No copying.

DataStoreStreamer &	operator= (const DataStoreStreamer &)=delete
	No assignment.

	~DataStoreStreamer ()
	destructor

EvtMessage *	streamDataStore (bool addPersistentDurability, bool streamTransientObjects=false)
	Store DataStore objects in EvtMessage.

int	restoreDataStore (EvtMessage *msg)
	Restore DataStore objects from EvtMessage.

void	setStreamingObjects (const std::vector< std::string > &list)
	Set names of objects to be streamed/destreamed.

int	queueEvtMessage (char *msg)
	Queue EvtMessage for destreaming.

int	restoreDataStoreAsync ()
	Restore objects in DataStore from temporary buffer.

void	setMaxThreads (int)
	maximum number of threads.

int	getMaxThreads ()
	maximum number of threads.

void	setDecoderStatus (int)
	Ask Itoh-san about this.

int	getDecoderStatus ()
	Ask Itoh-san about this.

Static Public Member Functions
static void *	decodeEvtMessage (int id)
	Decode EvtMessage and store objects in temporary buffer.

static bool	isMergeable (const TObject *object)
	Is the given object of a type that can be merged?

static void	clearMergeable (TObject *object)
	assuming object is mergeable, clear its contents.

static void	mergeIntoExisting (TObject existing, const TObject received)
	Assuming both objects are mergeable, merge 'received' into 'existing'.

static void	removeSideEffects ()
	call clear() and removeSideEffects() for all Mergeable objects in datastore (for c_Persistent durability).

Static Public Attributes
static const unsigned int	c_maxThreads = 16
	global maximum number of threads (cannot set higher number).

static const unsigned int	c_maxQueueDepth = 64
	Ask Itoh-san.

Private Types
enum	ETObjectBits { c_IsTransient = BIT(19) , c_IsNull = BIT(20) , c_PersistentDurability = BIT(21) }
	bits to store in TObject. More...

Static Private Member Functions
static int	restoreStreamerInfos (const TList *list)
	restore StreamerInfo from data in a file

Private Attributes
MsgHandler *	m_msghandler
	MsgHandler.

int	m_compressionLevel
	Compression level in streaming.

bool	m_handleMergeable
	Whether to handle Mergeable objects.

int	m_initStatus
	first event flag.

std::vector< std::string >	m_streamobjnames
	names of object to be streamed

int	m_maxthread
	Max.

pthread_t	m_pt [c_maxThreads]
	thread pointer

int	m_id [c_maxThreads]
	thread index.

int	m_threadin
	current thread?

int	m_decoderStatus [c_maxThreads] {0}
	thread decoder status.

Detailed Description

Stream/restore DataStore objects to/from EvtMessage.

Main interface is provided by streamDataStore() and restoreDataStore(). Other functions provide more obscure features.

Definition at line 34 of file DataStoreStreamer.h.

Member Enumeration Documentation

◆ ETObjectBits

enum ETObjectBits

private

bits to store in TObject.

Bits 14-23 are available for use in derived classes, and are reused here to transmit additional information. This is really quite ugly and should be replaced with some more sane way of transmitting object-level data. All bits are checked before using them, so if they are used by other code we know what happens.

Enumerator
c_IsTransient	The corresponding StoreEntry has flag c_DontWriteOut.
c_IsNull	object is not valid for current event, set StoreEntry::ptr to NULL.
c_PersistentDurability	Object is of persistent durability.

Definition at line 124 of file DataStoreStreamer.h.

                      {
      c_IsTransient = BIT(19), 
      c_IsNull = BIT(20), 
      c_PersistentDurability = BIT(21) 
    };

Constructor & Destructor Documentation

◆ DataStoreStreamer()

DataStoreStreamer	(	int	complevel = 0,
		bool	handleMergeable = true,
		int	maxthread = 0 )

explicit

Constructor.

Parameters

complevel	Compression level of streaming, 0 to disable
handleMergeable	perform special handling for Mergeable objects?
maxthread	maximal number of threads, 0 to disable

Definition at line 60 of file DataStoreStreamer.cc.

                                                                                      :
  m_compressionLevel(complevel),
  m_handleMergeable(handleMergeable),
  m_initStatus(0),
  m_maxthread(maxthread),
  m_threadin(0)
  //, m_threadout(0)
{
  if ((unsigned int)m_maxthread > c_maxThreads) {
    B2FATAL("DataStoreStreamer : Too many threads " << m_maxthread);
    m_maxthread = c_maxThreads;
  }
  m_msghandler = new MsgHandler(m_compressionLevel);
 
  if (m_maxthread > 0) {
    // Run decoder threads as sustainable detached threads
    pthread_attr_t thread_attr;
    pthread_attr_init(&thread_attr);
    //    pthread_attr_setschedpolicy(&thread_attr , SCHED_FIFO);
    //    pthread_attr_setdetachstate(&thread_attr , PTHREAD_CREATE_DETACHED);
    for (int i = 0; i < m_maxthread; i++) {
      my_decstat[i] = 0;
      m_pt[i] = (pthread_t)0;
      m_id[i] = i;
      //      m_pmsghandler[i] = new MsgHandler ( m_compressionLevel );
      mutex_thread[i] = PTHREAD_MUTEX_INITIALIZER;
      msg_compLevel[i] = m_compressionLevel;
    }
    for (int i = 0; i < m_maxthread; i++) {
      //      args.evtbuf = m_evtbuf[i];
      pthread_create(&m_pt[i], nullptr, RunDecodeEvtMessage, static_cast<void*>(&m_id[i]));
    }
    pthread_attr_destroy(&thread_attr);
  }
  s_streamer = this;
}

◆ ~DataStoreStreamer()

~DataStoreStreamer ( )

destructor

Definition at line 98 of file DataStoreStreamer.cc.

{
  delete m_msghandler;
}

Member Function Documentation

◆ clearMergeable()

void clearMergeable ( TObject * object )

static

assuming object is mergeable, clear its contents.

Use this after sending it to prevent sending the same data again in the next event.

Definition at line 113 of file DataStoreStreamer.cc.

{
  static_cast<Mergeable*>(object)->clear();
}

◆ decodeEvtMessage()

void * decodeEvtMessage ( int id )

static

Decode EvtMessage and store objects in temporary buffer.

Parameters

id Thread id

Definition at line 328 of file DataStoreStreamer.cc.

{
  printf("decodeEvtMessge : started. Thread ID = %d\n", id);
  // Clear Message Handler
  //  m_msghandler->clear();
  //  MsgHandler* msghandler = new MsgHandler(m_compressionLevel);
  //  MsgHandler* msghandler = m_pmsghandler[id];
 
  pthread_mutex_lock(&mutex);     // Lock thread
  MsgHandler msghandler(msg_compLevel[id]);
  pthread_mutex_unlock(&mutex);    // Unlock thread
 
  for (;;) {
    // Clear message handler event by event
    //    MsgHandler msghandler(m_compressionLevel);
    //    m_pmsghandler[id]->clear();
    // Wait for event in queue becomes ready
    msghandler.clear();
 
    while (my_evtbuf[id].empty()) usleep(10);
 
    // Pick up event buffer
    pthread_mutex_lock(&mutex_thread[id]);
    int nqueue = my_evtbuf[id].size();
    if (nqueue <= 0) printf("!!!!! Nqueue = %d\n", nqueue);
    char* evtbuf = my_evtbuf[id].front(); my_evtbuf[id].pop();
    pthread_mutex_unlock(&mutex_thread[id]);
 
    // In case of EOF
    if (evtbuf == nullptr) {
      printf("decodeEvtMessage: NULL evtbuf detected, nq = %d\n", nqueue);
      my_nobjs.push(-1);
      return nullptr;
    }
 
    // Construct EvtMessage
    auto* msg = new EvtMessage(evtbuf);
 
    // Decode EvtMessage into Objects
    std::vector<TObject*> objlist;
    std::vector<std::string> namelist;
 
    //    pthread_mutex_lock(&mutex);     // Lock test
    msghandler.decode_msg(msg, objlist, namelist);
    //    pthread_mutex_unlock(&mutex);    // Unlock test
 
 
    // Queue them for the registration in DataStore
    while (my_nobjs.size() >= c_maxQueueDepth) usleep(10);
    pthread_mutex_lock(&mutex);     // Lock queueing
    my_objlist.push(objlist);
    my_namelist.push(namelist);
    my_nobjs.push((msg->header())->nObjects);
    my_narrays.push((msg->header())->nArrays);
    pthread_mutex_unlock(&mutex);    // Unlock queueing
 
    // Release EvtMessage
    delete msg;
    delete[] evtbuf;
 
    // Preparation for next event
    my_decstat[id]  = 0; // Ready to read next event
 
  }
 
  return nullptr;
}

◆ getDecoderStatus()

int getDecoderStatus ( )

Ask Itoh-san about this.

Definition at line 468 of file DataStoreStreamer.cc.

{
  //  printf ( "Decode thread %d = %d\n", m_threadin, m_done_decode[m_threadin] );
  return (my_decstat[m_threadin]);
}

◆ getMaxThreads()

int getMaxThreads ( )

maximum number of threads.

Definition at line 463 of file DataStoreStreamer.cc.

{
  return m_maxthread;
}

◆ isMergeable()

bool isMergeable ( const TObject * object )

static

Is the given object of a type that can be merged?

Definition at line 108 of file DataStoreStreamer.cc.

{
  return object->InheritsFrom(Mergeable::Class());
}

◆ mergeIntoExisting()

void mergeIntoExisting	(	TObject *	existing,
		const TObject *	received )

static

Assuming both objects are mergeable, merge 'received' into 'existing'.

Definition at line 117 of file DataStoreStreamer.cc.

{
  auto* existingObject = static_cast<Mergeable*>(existing);
  existingObject->merge(static_cast<const Mergeable*>(received));
}

◆ queueEvtMessage()

int queueEvtMessage ( char * msg )

Queue EvtMessage for destreaming.

Parameters

msg	Event buffer to be restored.

Definition at line 298 of file DataStoreStreamer.cc.

{
  // EOF case
  if (evtbuf == nullptr) {
    printf("queueEvtMessage : NULL evtbuf detected. \n");
    for (int i = 0; i < m_maxthread; i++) {
      while (my_evtbuf[i].size() >= c_maxQueueDepth) usleep(10);
      my_evtbuf[m_threadin].push(evtbuf);
    }
    return 0;
  }
 
  // Put the event buffer in the queue of current thread
  for (;;) {
    if (my_evtbuf[m_threadin].size() < c_maxQueueDepth) {
      pthread_mutex_lock(&mutex_thread[m_threadin]);
      my_evtbuf[m_threadin].push(evtbuf);
      pthread_mutex_unlock(&mutex_thread[m_threadin]);
      break;
    }
    usleep(20);
  }
 
  // Switch to next thread
  my_decstat[m_threadin] = 1; // Event queued for decoding
  m_threadin++;
  if (m_threadin >= m_maxthread) m_threadin = 0;
  return 1;
}

◆ removeSideEffects()

void removeSideEffects ( )

static

call clear() and removeSideEffects() for all Mergeable objects in datastore (for c_Persistent durability).

Definition at line 122 of file DataStoreStreamer.cc.

{
  DataStore::StoreEntryMap& map = DataStore::Instance().getStoreEntryMap(DataStore::c_Persistent);
  for (auto& entryPair : map) {
    const DataStore::StoreEntry& entry = entryPair.second;
    if (isMergeable(entry.object)) {
      static_cast<Mergeable*>(entry.object)->removeSideEffects();
      static_cast<Mergeable*>(entry.object)->clear();
    }
  }
}

◆ restoreDataStore()

int restoreDataStore ( EvtMessage * msg )

Restore DataStore objects from EvtMessage.

Parameters

msg	EvtMessage to be restored.

Definition at line 209 of file DataStoreStreamer.cc.

{
  if (msg->type() == MSG_TERMINATE) {
    B2INFO("Got termination message. Exiting...");
    //msg doesn't really contain data, set EventMetaData to something equivalent
    StoreObjPtr<EventMetaData> eventMetaData;
    if (m_initStatus == 0 && DataStore::Instance().getInitializeActive())
      eventMetaData.registerInDataStore();
    eventMetaData.create();
    eventMetaData->setEndOfData();
  } else {
    // Clear Message Handler
    m_msghandler->clear();
 
    // List of objects to be restored
    std::vector<TObject*> objlist;
    std::vector<std::string> namelist;
 
    // Decode EvtMessage
    m_msghandler->decode_msg(msg, objlist, namelist);
    int nobjs = (msg->header())->nObjects;
    int narrays = (msg->header())->nArrays;
    if (unsigned(nobjs + narrays) != objlist.size())
      B2WARNING("restoreDataStore(): inconsistent #objects/#arrays in header");
 
    // Restore objects in DataStore
    for (int i = 0; i < nobjs + narrays; i++) {
      TObject* obj = objlist.at(i);
      bool array = (dynamic_cast<TClonesArray*>(obj) != nullptr);
      if (obj != nullptr) {
 
        // Read and Build StreamerInfo
        if (msg->type() == MSG_STREAMERINFO) {
          restoreStreamerInfos(static_cast<TList*>(obj));
          return 0;
        }
 
        bool isPersistent = obj->TestBit(c_PersistentDurability);
        DataStore::EDurability durability = isPersistent ? (DataStore::c_Persistent) : (DataStore::c_Event);
        TClass* cl = obj->IsA();
        if (array)
          cl = static_cast<TClonesArray*>(obj)->GetClass();
        if (m_initStatus == 0 && DataStore::Instance().getInitializeActive()) { //are we called by the module's initialize() function?
          auto flags = obj->TestBit(c_IsTransient) ? DataStore::c_DontWriteOut : DataStore::c_WriteOut;
          DataStore::Instance().registerEntry(namelist.at(i), durability, cl, array, flags);
        }
        DataStore::StoreEntry* entry = DataStore::Instance().getEntry(StoreAccessorBase(namelist.at(i), durability, cl, array));
        B2ASSERT("Can not find a data store entry with the name " << namelist.at(i) << ". Did you maybe forget to register it?", entry);
        //only restore object if it is valid for current event
        bool ptrIsNULL = obj->TestBit(c_IsNull);
        if (!ptrIsNULL) {
          bool merge = m_handleMergeable and !array and entry->ptr != nullptr and isMergeable(obj);
          if (merge) {
            B2DEBUG(100, "Will now merge " << namelist.at(i));
 
            mergeIntoExisting(entry->ptr, obj);
            delete obj;
          } else {
            //note: replace=true
            DataStore::Instance().createObject(obj, true,
                                               StoreAccessorBase(namelist.at(i), durability, cl, array));
 
            //reset bits of object in DataStore (are checked to be false when streaming the object)
            obj->SetBit(c_IsTransient, false);
            obj->SetBit(c_IsNull, false);
            obj->SetBit(c_PersistentDurability, false);
          }
          //   B2DEBUG(100, "restoreDS: " << (array ? "Array" : "Object") << ": " << namelist.at(i) << " stored");
        } else {
          //usually entry should already be invalidated, but e.g. for CrashHandler, it might not be.
          if (entry->ptr)
            entry->invalidate();
          //not stored, clean up
          delete obj;
        }
      } else {
        //DataStore always has non-NULL content (whether they're available is a different matter)
        B2ERROR("restoreDS: " << (array ? "Array" : "Object") << ": " << namelist.at(i) << " is NULL!");
      }
    }
 
  }
  // Return with normal exit status
  if (m_initStatus == 0) m_initStatus = 1;
  return 0;
}

◆ restoreDataStoreAsync()

int restoreDataStoreAsync ( )

Restore objects in DataStore from temporary buffer.

Definition at line 396 of file DataStoreStreamer.cc.

{
// Wait for the queue to become ready
  while (my_nobjs.empty()) usleep(10);
 
  // Register decoded objects in DataStore
 
  // Pick up event on the top and remove it from the queue
  pthread_mutex_lock(&mutex);
  int nobjs = my_nobjs.front(); my_nobjs.pop();
  if (nobjs == -1) {
    printf("restoreDataStore: EOF detected. exiting with status 0\n");
    pthread_mutex_unlock(&mutex);
    return 0;
  }
  int narrays = my_narrays.front(); my_narrays.pop();
  std::vector<TObject*> objlist = my_objlist.front(); my_objlist.pop();
  std::vector<std::string> namelist = my_namelist.front(); my_namelist.pop();
  pthread_mutex_unlock(&mutex);
 
  // Restore objects in DataStore
  //TODO refactor, just copied & pasted right now
  for (int i = 0; i < nobjs + narrays; i++) {
    //    printf ( "restoring object %d = %s\n", i, namelist.at(i).c_str() );
    bool array = (dynamic_cast<TClonesArray*>(objlist.at(i)) != nullptr);
    if (objlist.at(i) != nullptr) {
      TObject* obj = objlist.at(i);
      bool isPersistent = obj->TestBit(c_PersistentDurability);
      DataStore::EDurability durability = isPersistent ? (DataStore::c_Persistent) : (DataStore::c_Event);
      TClass* cl = obj->IsA();
      if (array)
        cl = static_cast<TClonesArray*>(obj)->GetClass();
      if (m_initStatus == 0 && DataStore::Instance().getInitializeActive()) { //are we called by the module's initialize() function?
        auto flags = obj->TestBit(c_IsTransient) ? DataStore::c_DontWriteOut : DataStore::c_WriteOut;
        DataStore::Instance().registerEntry(namelist.at(i), durability, cl, array, flags);
      }
      //only restore object if it is valid for current event
      bool ptrIsNULL = obj->TestBit(c_IsNull);
      if (!ptrIsNULL) {
        DataStore::Instance().createObject(obj, true,
                                           StoreAccessorBase(namelist.at(i), durability, cl, array));
        B2DEBUG(100, "restoreDS: " << (array ? "Array" : "Object") << ": " << namelist.at(i) << " stored");
 
        //reset bits of object in DataStore (are checked to be false when streaming the object)
        obj->SetBit(c_IsTransient, false);
        obj->SetBit(c_IsNull, false);
        obj->SetBit(c_PersistentDurability, false);
      } else {
        //not stored, clean up
        delete obj;
      }
    } else {
      //DataStore always has non-NULL content (whether they're available is a different matter)
      B2ERROR("restoreDS: " << (array ? "Array" : "Object") << ": " << namelist.at(i) << " is NULL!");
    }
  }
 
  //  printf ( "Objects restored in DataStore\n" );
 
  return 1;
}

◆ restoreStreamerInfos()

int restoreStreamerInfos ( const TList * list )

staticprivate

restore StreamerInfo from data in a file

Definition at line 481 of file DataStoreStreamer.cc.

{
  //
  // Copy from TSocket::RecvStreamerInfos()
  //
 
  //      TList *list = (TList*)mess->ReadObject(TList::Class());
  TStreamerInfo* info;
  TObjLink* lnk = list->FirstLink();
 
  std::vector<std::string> class_name;
 
  // First call BuildCheck for regular class
  while (lnk) {
    info = static_cast<TStreamerInfo*>(lnk->GetObject());
 
    int ovlap = 0;
    for (const auto& itr : class_name) {
      if (strcmp(itr.c_str(), info->GetName()) == 0) {
        ovlap = 1;
        B2DEBUG(100, "Regular Class Loop : The class " << info->GetName() << " has already appeared. Skipping...");
        break;
      }
    }
 
    // If the same class is in the object, ignore it. ( Otherwise it causes error. )
    if (ovlap == 0) {
      std::string temp_classname = info->GetName();
      class_name.push_back(temp_classname);
 
      TObject* element = info->GetElements()->UncheckedAt(0);
      Bool_t isstl = element && strcmp("This", element->GetName()) == 0;
      if (!isstl) {
        info->BuildCheck();
        //              if (gDebug > 0)
        B2INFO("importing TStreamerInfo: " << info->GetName() <<
               " version = " << info->GetClassVersion());
      }
    }
    lnk = lnk->Next();
  }
 
 
  class_name.clear();
  // Then call BuildCheck for stl class
  lnk = list->FirstLink();
  while (lnk) {
    info = static_cast<TStreamerInfo*>(lnk->GetObject());
 
    int ovlap = 0;
    for (const auto& itr : class_name) {
      if (strcmp(itr.c_str(), info->GetName()) == 0) {
        ovlap = 1;
        B2DEBUG(100, "STL Class Loop : The class " << info->GetName() << " has already appeared. Skipping...");
        break;
      }
    }
 
    // If the same class is in the object, ignore it. ( Otherwise it causes error. )
    if (ovlap == 0) {
      std::string temp_classname = info->GetName();
      class_name.push_back(temp_classname);
 
      TObject* element = info->GetElements()->UncheckedAt(0);
      Bool_t isstl = element && strcmp("This", element->GetName()) == 0;
      if (isstl) {
        info->BuildCheck();
        //              if (gDebug > 0)
        B2INFO("STL importing TStreamerInfo: " << info->GetName() <<
               " version = " << info->GetClassVersion());
      }
    }
    lnk = lnk->Next();
  }
 
  return 0;
}

◆ setDecoderStatus()

void setDecoderStatus ( int val )

Ask Itoh-san about this.

Definition at line 474 of file DataStoreStreamer.cc.

{
  //  printf ( "Decode thread %d = %d\n", m_threadin, m_done_decode[m_threadin] );
  m_decoderStatus[m_threadin] = val;
}

◆ setMaxThreads()

void setMaxThreads ( int maxthread )

maximum number of threads.

Definition at line 458 of file DataStoreStreamer.cc.

{
  m_maxthread = maxthread;
}

◆ setStreamingObjects()

void setStreamingObjects ( const std::vector< std::string > & list )

Set names of objects to be streamed/destreamed.

Definition at line 103 of file DataStoreStreamer.cc.

{
  m_streamobjnames = objlist;
}

◆ streamDataStore()

EvtMessage * streamDataStore	(	bool	addPersistentDurability,
		bool	streamTransientObjects = false )

Store DataStore objects in EvtMessage.

Parameters

addPersistentDurability	By default, only c_Event data is streamed. Setting this to true will add c_Persistent data to the EvtMessage.
streamTransientObjects	Should objects/arrays registered as transient be streamed?

Returns: pointer to EvtMessage, caller is responsible for deletion

Definition at line 136 of file DataStoreStreamer.cc.

{
  // Clear Message Handler
  m_msghandler->clear();
 
  // Stream objects (for all included durabilities)
  int narrays = 0;
  int nobjs = 0;
  DataStore::EDurability durability = DataStore::c_Event;
  while (true) {
    auto& map = DataStore::Instance().getStoreEntryMap(durability);
    for (auto &[name, entry] : map) {
      //skip transient objects/arrays?
      if (!streamTransientObjects and entry.dontWriteOut)
        continue;
 
      //skip objects not in the list
      if (!m_streamobjnames.empty()) {
        auto pos = std::find(m_streamobjnames.begin(), m_streamobjnames.end(), name);
        if (pos == m_streamobjnames.end())  continue;
      }
 
      //verify that bits are unused
      if (entry.object->TestBit(c_IsTransient)) {
        B2FATAL("DataStoreStreamer::c_IsTransient bit is set for " << name << "!");
      }
      if (entry.object->TestBit(c_IsNull)) {
        B2FATAL("DataStoreStreamer::c_IsNull bit is set for " << name << "!");
      }
      if (entry.object->TestBit(c_PersistentDurability)) {
        B2FATAL("DataStoreStreamer::c_PersistentDurability bit is set for " << name << "!");
      }
      //verify TObject bits are serialised
      if (entry.object->IsA()->CanIgnoreTObjectStreamer()) {
        B2FATAL("TObject streamers disabled for " << name << "!");
      }
      //store some information in TObject bits to ensure consistent state even if entry.ptr is NULL
      entry.object->SetBit(c_IsTransient, entry.dontWriteOut);
      entry.object->SetBit(c_IsNull, (entry.ptr == nullptr));
      entry.object->SetBit(c_PersistentDurability, (durability == DataStore::c_Persistent));
      m_msghandler->add(entry.object, name);
      B2DEBUG(100, "adding item " << name);
 
      if (entry.isArray)
        narrays++;
      else
        nobjs++;
 
      //reset bits (are checked to be false when streaming the object)
      entry.object->SetBit(c_IsTransient, false);
      entry.object->SetBit(c_IsNull, false);
      entry.object->SetBit(c_PersistentDurability, false);
 
      bool merge = m_handleMergeable and entry.ptr != nullptr and isMergeable(entry.object);
      if (merge)
        clearMergeable(entry.object);
    }
 
    if (addPersistentDurability and durability == DataStore::c_Event)
      durability = DataStore::c_Persistent;
    else
      break;
  }
 
  // Encode EvtMessage
  EvtMessage* msg = m_msghandler->encode_msg(MSG_EVENT);
  (msg->header())->nObjects = nobjs;
  (msg->header())->nArrays = narrays;
 
  return msg;
}

Member Data Documentation

◆ c_maxQueueDepth

const unsigned int c_maxQueueDepth = 64

static

Ask Itoh-san.

Definition at line 39 of file DataStoreStreamer.h.

◆ c_maxThreads

const unsigned int c_maxThreads = 16

static

global maximum number of threads (cannot set higher number).

Definition at line 37 of file DataStoreStreamer.h.

◆ m_compressionLevel

int m_compressionLevel

private

Compression level in streaming.

Definition at line 137 of file DataStoreStreamer.h.

◆ m_decoderStatus

int m_decoderStatus[c_maxThreads] {0}

private

thread decoder status.

Definition at line 168 of file DataStoreStreamer.h.

168{0};

◆ m_handleMergeable

bool m_handleMergeable

private

Whether to handle Mergeable objects.

Definition at line 139 of file DataStoreStreamer.h.

◆ m_id

int m_id[c_maxThreads]

private

thread index.

Definition at line 163 of file DataStoreStreamer.h.

◆ m_initStatus

int m_initStatus

private

first event flag.

0 during first event, 1 otherwise.

Definition at line 145 of file DataStoreStreamer.h.

◆ m_maxthread

int m_maxthread

private

Max.

number of threads for asynchronous processing

4 for default

Definition at line 157 of file DataStoreStreamer.h.

◆ m_msghandler

MsgHandler* m_msghandler

private

MsgHandler.

MsgHandler object used to form/decode EvtMessage

Definition at line 134 of file DataStoreStreamer.h.

◆ m_pt

pthread_t m_pt[c_maxThreads]

private

thread pointer

Definition at line 161 of file DataStoreStreamer.h.

◆ m_streamobjnames

std::vector<std::string> m_streamobjnames

private

names of object to be streamed

If size=0, all objects to be streamed

Definition at line 151 of file DataStoreStreamer.h.

◆ m_threadin

int m_threadin

private

current thread?

Definition at line 165 of file DataStoreStreamer.h.

The documentation for this class was generated from the following files:

framework/pcore/include/DataStoreStreamer.h
framework/pcore/src/DataStoreStreamer.cc

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Types

Static Private Member Functions

Private Attributes

Detailed Description

Member Enumeration Documentation

◆ ETObjectBits

Constructor & Destructor Documentation

◆ DataStoreStreamer()

◆ ~DataStoreStreamer()

Member Function Documentation

◆ clearMergeable()

◆ decodeEvtMessage()

◆ getDecoderStatus()

◆ getMaxThreads()

◆ isMergeable()

◆ mergeIntoExisting()

◆ queueEvtMessage()

◆ removeSideEffects()

◆ restoreDataStore()

◆ restoreDataStoreAsync()

◆ restoreStreamerInfos()

◆ setDecoderStatus()

◆ setMaxThreads()

◆ setStreamingObjects()

◆ streamDataStore()

Member Data Documentation

◆ c_maxQueueDepth

◆ c_maxThreads

◆ m_compressionLevel

◆ m_decoderStatus

◆ m_handleMergeable

◆ m_id

◆ m_initStatus

◆ m_maxthread

◆ m_msghandler

◆ m_pt

◆ m_streamobjnames

◆ m_threadin