HLT/BASE/AliHLTMessage.cxx

   1 // $Id$
   2
   3 /** @file   AliHLTMessage.cxx
   4     @author Matthias Richter (customization of Root TMessage )
   5     @date
   6     @brief  Serialization of Root objects in the ALICE HLT. */
   7
   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
  12
  13 /*************************************************************************
  14  * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers.               *
  15  * All rights reserved.                                                  *
  16  *                                                                       *
  17  * For the licensing terms see $ROOTSYS/LICENSE.                         *
  18  * For the list of contributors see $ROOTSYS/README/CREDITS.             *
  19  *************************************************************************/
  20
  21 //////////////////////////////////////////////////////////////////////////
  22 //                                                                      //
  23 // TMessage                                                             //
  24 //                                                                      //
  25 // Message buffer class used for serializing objects and sending them   //
  26 // over a network. This class inherits from TBuffer the basic I/O       //
  27 // serializer.                                                          //
  28 //                                                                      //
  29 //////////////////////////////////////////////////////////////////////////
  30
  31 #include "AliHLTMessage.h"
  32 #include "Bytes.h"
  33 #include "TFile.h"
  34
  35 extern "C" void R__zip (Int_t cxlevel, Int_t *nin, char *bufin, Int_t *lout, char *bufout, Int_t *nout);
  36 extern "C" void R__unzip(Int_t *nin, UChar_t *bufin, Int_t *lout, char *bufout, Int_t *nout);
  37 const Int_t kMAXBUF = 0xffffff;
  38
  39 ClassImp(AliHLTMessage)
  40
  41 //______________________________________________________________________________
  42 AliHLTMessage::AliHLTMessage(UInt_t what)
  43   :
  44 # ifdef ROOT_TBufferFile
  45   TBufferFile(kWrite),
  46 # else
  47   TBuffer(kWrite),
  48 # endif
  49   AliHLTLogging(),
  50   fWhat(what),
  51   fClass(0),
  52   fCompress(0),
  53   fBufComp(0),
  54   fBufCompCur(0),
  55   fCompPos(0)
  56 {
  57    // Create a AliHLTMessage object for storing objects. The "what" integer
  58    // describes the type of message. Predifined ROOT system message types
  59    // can be found in MessageTypes.h. Make sure your own message types are
  60    // unique from the ROOT defined message types (i.e. 0 - 10000 are
  61    // reserved by ROOT). In case you OR "what" with kMESS_ACK, the message
  62    // will wait for an acknowledgement from the remote side. This makes
  63    // the sending process synchronous. In case you OR "what" with kMESS_ZIP,
  64    // the message will be compressed in TSocket using the zip algorithm
  65    // (only if message is > 256 bytes).
  66
  67    // space at the beginning of the message reserved for the message length
  68    UInt_t   reserved = 0;
  69    *this << reserved;
  70
  71    *this << what;
  72
  73 }
  74
  75 const Int_t AliHLTMessage::fgkMinimumSize=30;
  76 UInt_t AliHLTMessage::fgkDefaultBuffer[2]={0,0};
  77
  78 //______________________________________________________________________________
  79 AliHLTMessage::AliHLTMessage(void *buf, Int_t bufsize)
  80   :
  81 # if defined(ROOT_TBufferFile)
  82   TBufferFile(kRead, bufsize>fgkMinimumSize?bufsize:sizeof(fgkDefaultBuffer), bufsize>fgkMinimumSize?buf:&fgkDefaultBuffer, 0),
  83 # else
  84   TBuffer(kRead, bufsize>fgkMinimumSize?bufsize:sizeof(fgkDefaultBuffer), bufsize>fgkMinimumSize?buf:&fgkDefaultBuffer, 0),
  85 # endif
  86   AliHLTLogging(),
  87   fWhat(0),
  88   fClass(0),
  89   fCompress(0),
  90   fBufComp(0),
  91   fBufCompCur(0),
  92   fCompPos(0)
  93 {
  94    // Create a AliHLTMessage object for reading objects. The objects will be
  95    // read from buf. Use the What() method to get the message type.
  96
  97    // skip space at the beginning of the message reserved for the message length
  98    fBufCur += sizeof(UInt_t);
  99
 100    *this >> fWhat;
 101
 102    if (fWhat & kMESS_ZIP) {
 103       // if buffer has kMESS_ZIP set, move it to fBufComp and uncompress
 104       fBufComp    = fBuffer;
 105       fBufCompCur = fBuffer + bufsize;
 106       fBuffer     = 0;
 107       Uncompress();
 108    }
 109
 110    if (fWhat == kMESS_OBJECT) {
 111       InitMap();
 112       fClass = ReadClass();     // get first the class stored in message
 113       SetBufferOffset(sizeof(UInt_t) + sizeof(fWhat));
 114       ResetMap();
 115    } else {
 116       fClass = 0;
 117    }
 118 }
 119
 120 //______________________________________________________________________________
 121 AliHLTMessage::~AliHLTMessage()
 122 {
 123    // Clean up compression buffer.
 124    delete [] fBufComp;
 125 }
 126
 127 //______________________________________________________________________________
 128 void AliHLTMessage::Forward()
 129 {
 130    // Change a buffer that was received into one that can be send, i.e.
 131    // forward a just received message.
 132
 133    if (IsReading()) {
 134       SetWriteMode();
 135       SetBufferOffset(fBufSize);
 136
 137       if (fBufComp) {
 138          fCompPos = fBufCur;
 139       }
 140    }
 141 }
 142
 143 //______________________________________________________________________________
 144 void AliHLTMessage::Reset()
 145 {
 146    // Reset the message buffer so we can use (i.e. fill) it again.
 147
 148    SetBufferOffset(sizeof(UInt_t) + sizeof(fWhat));
 149    ResetMap();
 150
 151    if (fBufComp) {
 152       delete [] fBufComp;
 153       fBufComp    = 0;
 154       fBufCompCur = 0;
 155       fCompPos    = 0;
 156    }
 157 }
 158
 159 //______________________________________________________________________________
 160 void AliHLTMessage::SetLength() const
 161 {
 162    // Set the message length at the beginning of the message buffer.
 163
 164    if (IsWriting()) {
 165       char *buf = Buffer();
 166       *((UInt_t*)buf) = (UInt_t)(Length() - sizeof(UInt_t));
 167
 168       if (fBufComp) {
 169          buf = fBufComp;
 170          *((UInt_t*)buf) = (UInt_t)(Length() - sizeof(UInt_t));
 171       }
 172    }
 173 }
 174
 175 //______________________________________________________________________________
 176 void AliHLTMessage::SetWhat(UInt_t what)
 177 {
 178    // Using this method one can change the message type a-posteriory.
 179    // In case you OR "what" with kMESS_ACK, the message will wait for
 180    // an acknowledgement from the remote side. This makes the sending
 181    // process synchronous.
 182
 183    fWhat = what;
 184
 185    char *buf = Buffer();
 186    buf += sizeof(UInt_t);   // skip reserved length space
 187    tobuf(buf, what);
 188
 189    if (fBufComp) {
 190       buf = fBufComp;
 191       buf += sizeof(UInt_t);   // skip reserved length space
 192       tobuf(buf, what | kMESS_ZIP);
 193    }
 194 }
 195
 196 //______________________________________________________________________________
 197 void AliHLTMessage::SetCompressionLevel(Int_t level)
 198 {
 199    // Set the message compression level. Can be between 0 and 9 with 0
 200    // being no compression and 9 maximum compression. In general the default
 201    // level of 1 is the best compromise between achieved compression and
 202    // cpu time. Compression will only happen when the message is > 256 bytes.
 203
 204    if (level < 0) level = 0;
 205    if (level > 9) level = 9;
 206
 207    if (level != fCompress && fBufComp) {
 208       delete [] fBufComp;
 209       fBufComp    = 0;
 210       fBufCompCur = 0;
 211       fCompPos    = 0;
 212    }
 213    fCompress = level;
 214 }
 215
 216 //______________________________________________________________________________
 217 Int_t AliHLTMessage::Compress()
 218 {
 219    // Compress the message. The message will only be compressed if the
 220    // compression level > 0 and the if the message is > 256 bytes.
 221    // Returns -1 in case of error (when compression fails or
 222    // when the message increases in size in some pathological cases),
 223    // otherwise returns 0.
 224
 225    if (fCompress == 0) {
 226       // no compression specified
 227       if (fBufComp) {
 228          delete [] fBufComp;
 229          fBufComp    = 0;
 230          fBufCompCur = 0;
 231          fCompPos    = 0;
 232       }
 233       return 0;
 234    }
 235
 236    if (fBufComp && fCompPos == fBufCur) {
 237       // the message was already compressed
 238       return 0;
 239    }
 240
 241    // remove any existing compressed buffer before compressing modified message
 242    if (fBufComp) {
 243       delete [] fBufComp;
 244       fBufComp    = 0;
 245       fBufCompCur = 0;
 246       fCompPos    = 0;
 247    }
 248
 249    if (Length() <= (Int_t)(256 + 2*sizeof(UInt_t))) {
 250       // this message is too small to be compressed
 251       return 0;
 252    }
 253
 254    Int_t hdrlen   = 2*sizeof(UInt_t);
 255    Int_t messlen  = Length() - hdrlen;
 256    Int_t nbuffers = messlen / kMAXBUF;
 257    Int_t chdrlen  = 3*sizeof(UInt_t);   // compressed buffer header length
 258    Int_t buflen   = TMath::Max(512, chdrlen + messlen + 9*nbuffers);
 259    fBufComp       = new char[buflen];
 260    char *messbuf  = Buffer() + hdrlen;
 261    char *bufcur   = fBufComp + chdrlen;
 262    Int_t noutot   = 0;
 263    Int_t nzip     = 0;
 264    Int_t nout, bufmax;
 265    for (Int_t i = 0; i <= nbuffers; i++) {
 266       if (i == nbuffers)
 267          bufmax = messlen - nzip;
 268       else
 269          bufmax = kMAXBUF;
 270       R__zip(fCompress, &bufmax, messbuf, &bufmax, bufcur, &nout);
 271       if (nout == 0 || nout >= messlen) {
 272          //this happens when the buffer cannot be compressed
 273          delete [] fBufComp;
 274          fBufComp    = 0;
 275          fBufCompCur = 0;
 276          fCompPos    = 0;
 277          return -1;
 278       }
 279       bufcur  += nout;
 280       noutot  += nout;
 281       messbuf += kMAXBUF;
 282       nzip    += kMAXBUF;
 283    }
 284    fBufCompCur = bufcur;
 285    fCompPos    = fBufCur;
 286
 287    bufcur = fBufComp;
 288    tobuf(bufcur, (UInt_t)(CompLength() - sizeof(UInt_t)));
 289    Int_t what = fWhat | kMESS_ZIP;
 290    tobuf(bufcur, what);
 291    tobuf(bufcur, Length());    // original uncompressed buffer length
 292
 293    return 0;
 294 }
 295
 296 //______________________________________________________________________________
 297 Int_t AliHLTMessage::Uncompress()
 298 {
 299    // Uncompress the message. The message will only be uncompressed when
 300    // kMESS_ZIP is set. Returns -1 in case of error, 0 otherwise.
 301
 302    if (!fBufComp || !(fWhat & kMESS_ZIP))
 303       return -1;
 304
 305    Int_t buflen;
 306    Int_t hdrlen = 2*sizeof(UInt_t);
 307    UChar_t *bufcur = (UChar_t*)fBufComp + hdrlen;
 308    frombuf((char *&)bufcur, &buflen);
 309    fBuffer  = new char[buflen];
 310    fBufSize = buflen;
 311    fBufCur  = fBuffer + sizeof(UInt_t) + sizeof(fWhat);
 312    fBufMax  = fBuffer + fBufSize;
 313    char *messbuf = fBuffer + hdrlen;
 314
 315    Int_t nin, nout, nbuf;
 316    Int_t noutot = 0;
 317    while (1) {
 318       nin  = 9 + ((Int_t)bufcur[3] | ((Int_t)bufcur[4] << 8) | ((Int_t)bufcur[5] << 16));
 319       nbuf = (Int_t)bufcur[6] | ((Int_t)bufcur[7] << 8) | ((Int_t)bufcur[8] << 16);
 320       R__unzip(&nin, bufcur, &nbuf, messbuf, &nout);
 321       if (!nout) break;
 322       noutot += nout;
 323       if (noutot >= buflen - hdrlen) break;
 324       bufcur  += nin;
 325       messbuf += nout;
 326    }
 327
 328    fWhat &= ~kMESS_ZIP;
 329    fCompress = 1;
 330
 331    return 0;
 332 }
 333