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