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