3 /** @file AliHLTMessage.cxx
4 @author Matthias Richter (customization of Root TMessage )
6 @brief Serialization of Root objects in the ALICE HLT. */
8 // This is the original Root TMessage implementation with a few minor
9 // modifications, original revision:
10 // root/net: v5-14-00 $: TMessage.cxx,v 1.6 2004/05/07 09:51:58 brun
11 // Author: Fons Rademakers 19/12/96
13 /*************************************************************************
14 * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers. *
15 * All rights reserved. *
17 * For the licensing terms see $ROOTSYS/LICENSE. *
18 * For the list of contributors see $ROOTSYS/README/CREDITS. *
19 *************************************************************************/
21 //////////////////////////////////////////////////////////////////////////
25 // Message buffer class used for serializing objects and sending them //
26 // over a network. This class inherits from TBuffer the basic I/O //
29 //////////////////////////////////////////////////////////////////////////
31 #include "AliHLTMessage.h"
32 #include "TVirtualStreamerInfo.h"
35 #include "TProcessID.h"
38 extern "C" void R__zip (Int_t cxlevel, Int_t *nin, char *bufin, Int_t *lout, char *bufout, Int_t *nout);
39 extern "C" void R__unzip(Int_t *nin, UChar_t *bufin, Int_t *lout, char *bufout, Int_t *nout);
40 const Int_t kMAXBUF = 0xffffff;
42 Bool_t AliHLTMessage::fgEvolution = kFALSE;
44 ClassImp(AliHLTMessage)
46 //______________________________________________________________________________
47 AliHLTMessage::AliHLTMessage(UInt_t what)
49 # ifdef ROOT_TBufferFile
66 // Create a AliHLTMessage object for storing objects. The "what" integer
67 // describes the type of message. Predifined ROOT system message types
68 // can be found in MessageTypes.h. Make sure your own message types are
69 // unique from the ROOT defined message types (i.e. 0 - 10000 are
70 // reserved by ROOT). In case you OR "what" with kMESS_ACK, the message
71 // will wait for an acknowledgement from the remote side. This makes
72 // the sending process synchronous. In case you OR "what" with kMESS_ZIP,
73 // the message will be compressed in TSocket using the zip algorithm
74 // (only if message is > 256 bytes).
76 // space at the beginning of the message reserved for the message length
82 SetBit(kCannotHandleMemberWiseStreaming);
85 const Int_t AliHLTMessage::fgkMinimumSize=30;
86 UInt_t AliHLTMessage::fgkDefaultBuffer[2]={0,0};
88 //______________________________________________________________________________
89 AliHLTMessage::AliHLTMessage(void *buf, Int_t bufsize)
91 # if defined(ROOT_TBufferFile)
92 TBufferFile(kRead, bufsize>fgkMinimumSize?bufsize:sizeof(fgkDefaultBuffer), bufsize>fgkMinimumSize?buf:&fgkDefaultBuffer, 0),
94 TBuffer(kRead, bufsize>fgkMinimumSize?bufsize:sizeof(fgkDefaultBuffer), bufsize>fgkMinimumSize?buf:&fgkDefaultBuffer, 0),
103 , fBufUncompressed(0)
108 // Create a AliHLTMessage object for reading objects. The objects will be
109 // read from buf. Use the What() method to get the message type.
111 // skip space at the beginning of the message reserved for the message length
112 fBufCur += sizeof(UInt_t);
116 if (fWhat & kMESS_ZIP) {
117 // if buffer has kMESS_ZIP set, move it to fBufComp and uncompress
119 fBufCompCur = fBuffer + bufsize;
123 // NOTE: this is not done in TMessage and will lead to the deletion
124 // of the buffer. This is not allowed in case of HLT where the
125 // buffer is handled by the framework. In general, I think this
126 // is a very bad idea to do it like that in TMessage
131 if (fWhat == kMESS_OBJECT) {
133 fClass = ReadClass(); // get first the class stored in message
134 SetBufferOffset(sizeof(UInt_t) + sizeof(fWhat));
141 //______________________________________________________________________________
142 AliHLTMessage::~AliHLTMessage()
144 // Clean up compression buffer.
148 //______________________________________________________________________________
149 void AliHLTMessage::EnableSchemaEvolutionForAll(Bool_t enable)
151 // Static function enabling or disabling the automatic schema evolution.
152 // By default schema evolution support is off.
154 fgEvolution = enable;
157 //______________________________________________________________________________
158 Bool_t AliHLTMessage::UsesSchemaEvolutionForAll()
160 // Static function returning status of global schema evolution.
165 //______________________________________________________________________________
166 void AliHLTMessage::ForceWriteInfo(TVirtualStreamerInfo *info, Bool_t /* force */)
168 // Force writing the TStreamerInfo to the message.
170 if (fgEvolution || fEvolution) {
171 if (!fInfos) fInfos = new TList();
172 if (fInfos->FindObject(info->GetName())==NULL) {
178 //______________________________________________________________________________
179 void AliHLTMessage::Forward()
181 // Change a buffer that was received into one that can be send, i.e.
182 // forward a just received message.
186 SetBufferOffset(fBufSize);
187 SetBit(kCannotHandleMemberWiseStreaming);
195 //______________________________________________________________________________
196 void AliHLTMessage::IncrementLevel(TVirtualStreamerInfo *info)
200 TBufferFile::IncrementLevel(info);
203 if (fgEvolution || fEvolution) {
204 if (!fInfos) fInfos = new TList();
206 // add the streamer info, but only once
207 // this assumes that there is only one version
208 if (fInfos->FindObject(info->GetName())==NULL) {
214 //______________________________________________________________________________
215 void AliHLTMessage::Reset()
217 // Reset the message buffer so we can use (i.e. fill) it again.
219 SetBufferOffset(sizeof(UInt_t) + sizeof(fWhat));
228 if (fBufUncompressed) {
229 delete [] fBufUncompressed;
230 fBufUncompressed=NULL;
234 //______________________________________________________________________________
235 void AliHLTMessage::SetLength() const
237 // Set the message length at the beginning of the message buffer.
240 char *buf = Buffer();
241 *((UInt_t*)buf) = (UInt_t)(Length() - sizeof(UInt_t));
245 *((UInt_t*)buf) = (UInt_t)(CompLength() - sizeof(UInt_t));
250 //______________________________________________________________________________
251 void AliHLTMessage::SetWhat(UInt_t what)
253 // Using this method one can change the message type a-posteriory.
254 // In case you OR "what" with kMESS_ACK, the message will wait for
255 // an acknowledgement from the remote side. This makes the sending
256 // process synchronous.
260 char *buf = Buffer();
261 buf += sizeof(UInt_t); // skip reserved length space
266 buf += sizeof(UInt_t); // skip reserved length space
267 tobuf(buf, what | kMESS_ZIP);
271 //______________________________________________________________________________
272 void AliHLTMessage::SetCompressionLevel(Int_t level)
274 // Set the message compression level. Can be between 0 and 9 with 0
275 // being no compression and 9 maximum compression. In general the default
276 // level of 1 is the best compromise between achieved compression and
277 // cpu time. Compression will only happen when the message is > 256 bytes.
279 if (level < 0) level = 0;
280 if (level > 9) level = 9;
282 if (level != fCompress && fBufComp) {
291 //______________________________________________________________________________
292 Int_t AliHLTMessage::Compress()
294 // Compress the message. The message will only be compressed if the
295 // compression level > 0 and the if the message is > 256 bytes.
296 // Returns -1 in case of error (when compression fails or
297 // when the message increases in size in some pathological cases),
298 // otherwise returns 0.
300 if (fCompress == 0) {
301 // no compression specified
311 if (fBufComp && fCompPos == fBufCur) {
312 // the message was already compressed
316 // remove any existing compressed buffer before compressing modified message
324 if (Length() <= (Int_t)(256 + 2*sizeof(UInt_t))) {
325 // this message is too small to be compressed
329 Int_t hdrlen = 2*sizeof(UInt_t);
330 Int_t messlen = Length() - hdrlen;
331 Int_t nbuffers = messlen / kMAXBUF;
332 Int_t chdrlen = 3*sizeof(UInt_t); // compressed buffer header length
333 Int_t buflen = TMath::Max(512, chdrlen + messlen + 9*nbuffers);
334 fBufComp = new char[buflen];
335 char *messbuf = Buffer() + hdrlen;
336 char *bufcur = fBufComp + chdrlen;
340 for (Int_t i = 0; i <= nbuffers; i++) {
342 bufmax = messlen - nzip;
345 R__zip(fCompress, &bufmax, messbuf, &bufmax, bufcur, &nout);
346 if (nout == 0 || nout >= messlen) {
347 //this happens when the buffer cannot be compressed
359 fBufCompCur = bufcur;
363 tobuf(bufcur, (UInt_t)(CompLength() - sizeof(UInt_t)));
364 Int_t what = fWhat | kMESS_ZIP;
366 tobuf(bufcur, Length()); // original uncompressed buffer length
371 //______________________________________________________________________________
372 Int_t AliHLTMessage::Uncompress()
374 // Uncompress the message. The message will only be uncompressed when
375 // kMESS_ZIP is set. Returns -1 in case of error, 0 otherwise.
377 if (!fBufComp || !(fWhat & kMESS_ZIP))
381 Int_t hdrlen = 2*sizeof(UInt_t);
382 char *bufcur1 = fBufComp + hdrlen;
383 frombuf(bufcur1, &buflen);
384 UChar_t *bufcur = (UChar_t*)bufcur1;
385 fBuffer = new char[buflen];
386 fBufUncompressed = fBuffer;
388 fBufCur = fBuffer + sizeof(UInt_t) + sizeof(fWhat);
389 fBufMax = fBuffer + fBufSize;
390 char *messbuf = fBuffer + hdrlen;
392 Int_t nin, nout, nbuf;
395 nin = 9 + ((Int_t)bufcur[3] | ((Int_t)bufcur[4] << 8) | ((Int_t)bufcur[5] << 16));
396 nbuf = (Int_t)bufcur[6] | ((Int_t)bufcur[7] << 8) | ((Int_t)bufcur[8] << 16);
397 R__unzip(&nin, bufcur, &nbuf, messbuf, &nout);
400 if (noutot >= buflen - hdrlen) break;
411 //______________________________________________________________________________
412 void AliHLTMessage::WriteObject(const TObject *obj)
414 // Write object to message buffer.
415 // When support for schema evolution is enabled the list of TStreamerInfo
416 // used to stream this object is kept in fInfos. This information is used
417 // by TSocket::Send that sends this list through the socket. This list is in
418 // turn used by TSocket::Recv to store the TStreamerInfo objects in the
419 // relevant TClass in case the TClass does not know yet about a particular
420 // class version. This feature is implemented to support clients and servers
421 // with either different ROOT versions or different user classes versions.
423 if (fgEvolution || fEvolution) {
427 fInfos = new TList();
430 fBitsPIDs.ResetAllBits();
431 WriteObjectAny(obj, TObject::Class());
434 //______________________________________________________________________________
435 UShort_t AliHLTMessage::WriteProcessID(TProcessID *pid)
437 // Check if the ProcessID pid is already in the message.
439 // - mark bit 0 of fBitsPIDs to indicate that a ProcessID has been found
440 // - mark bit uid+1 where uid id the uid of the ProcessID
442 if (fBitsPIDs.TestBitNumber(0)) return 0;
444 pid = TProcessID::GetPID();
446 fBitsPIDs.SetBitNumber(0);
447 UInt_t uid = pid->GetUniqueID();
448 fBitsPIDs.SetBitNumber(uid+1);
452 AliHLTMessage* AliHLTMessage::Stream(TObject* pSrc, Int_t compression, unsigned verbosity)
454 /// Helper function to stream an object into an AliHLTMessage
455 /// The returned instance must be cleaned by the caller
457 /// Get the data and data size from the message:
459 /// pMsg->CompLength();
460 /// pMsg->CompBuffer();
465 /// Note: accessing scheme will be change din the future to just have the two
467 if (!pSrc) return NULL;
470 AliHLTMessage* pMsg=new AliHLTMessage(kMESS_OBJECT);
472 log.LoggingVarargs(kHLTLogError, "AliHLTMessage", "Stream" , __FILE__ , __LINE__ , "memory allocation failed");
476 pMsg->SetCompressionLevel(compression);
477 pMsg->WriteObject(pSrc);
478 if (pMsg->Length()>0) {
480 // NOTE: AliHLTMessage does implement it's own SetLength method
481 // which is not architecture independent. The original SetLength
482 // stores the size always in network byte order.
483 // I'm trying to remember the rational for that, might be that
484 // it was just some lack of knowledge. Want to change this, but
485 // has to be done carefully to be backward compatible.
486 pMsg->SetLength(); // sets the length to the first (reserved) word
488 // does nothing if the level is 0
491 if (pMsg->CompBuffer()) {
492 pMsg->SetLength(); // set once more to have the byte order
493 if (verbosity>0) log.LoggingVarargs(kHLTLogInfo, "AliHLTMessage", "Stream" , __FILE__ , __LINE__ , "object %p type %s streamed: size %d", pSrc, pSrc->GetName(), pMsg->CompLength());
495 if (verbosity>0) log.LoggingVarargs(kHLTLogInfo, "AliHLTMessage", "Stream" , __FILE__ , __LINE__ , "object %p type %s streamed: size %d", pSrc, pSrc->GetName(), pMsg->Length());
501 TObject* AliHLTMessage::Extract(const void* pBuffer, unsigned bufferSize, unsigned verbosity)
503 /// Helper function to extract an object from a buffer.
504 /// The returned object must be cleaned by the caller
506 if (!pBuffer || bufferSize<sizeof(AliHLTUInt32_t)) {
507 if (verbosity>0) log.LoggingVarargs(kHLTLogWarning, "AliHLTMessage", "Extract" , __FILE__ , __LINE__ , "invalid input buffer %p %d", pBuffer, bufferSize);
511 AliHLTUInt32_t firstWord=*((AliHLTUInt32_t*)pBuffer);
512 if (firstWord==bufferSize-sizeof(AliHLTUInt32_t) &&
513 firstWord>=34 /*thats the minimum size of a streamed TObject*/) {
514 AliHLTMessage msg((AliHLTUInt8_t*)pBuffer, bufferSize);
515 TClass* objclass=msg.GetClass();
516 TObject* pObject=msg.ReadObject(objclass);
517 if (pObject && objclass) {
518 if (verbosity>0) log.LoggingVarargs(kHLTLogInfo, "AliHLTMessage", "Extract" , __FILE__ , __LINE__ , "object %p type %s created", pObject, objclass->GetName());
521 if (verbosity>0) log.LoggingVarargs(kHLTLogWarning, "AliHLTMessage", "Extract" , __FILE__ , __LINE__ , "failed to create object from buffer of size %d", bufferSize);
524 if (verbosity>0) log.LoggingVarargs(kHLTLogWarning, "AliHLTMessage", "Extract" , __FILE__ , __LINE__ , "not a streamed TObject: block size %d, indicated %d", bufferSize, firstWord+sizeof(AliHLTUInt32_t));
529 TObject* AliHLTMessage::Extract(const char* filename, unsigned verbosity)
531 /// Helper function to extract an object from a file containing the streamed object.
532 /// The returned object must be cleaned by the caller
533 if (!filename) return NULL;
536 TString input=filename;
537 input+="?filetype=raw";
538 TFile* pFile=new TFile(input);
539 if (!pFile) return NULL;
540 TObject* pObject=NULL;
541 if (!pFile->IsZombie()) {
544 buffer.Set(pFile->GetSize());
545 if (pFile->ReadBuffer(buffer.GetArray(), buffer.GetSize())==0) {
546 pObject=Extract(buffer.GetArray(), buffer.GetSize(), verbosity);
548 log.LoggingVarargs(kHLTLogError, "AliHLTMessage", "Extract" , __FILE__ , __LINE__ , "failed reading %d byte(s) from file %s", pFile->GetSize(), filename);