Message.h

/**************************************************************************/
/*                                                                        */
/* Copyright (c) 2001, 2011 NoMachine (http://www.nomachine.com)          */
/* Copyright (c) 2008-2014 Oleksandr Shneyder <o.shneyder@phoca-gmbh.de>  */
/* Copyright (c) 2014-2016 Ulrich Sibiller <uli42@gmx.de>                 */
/* Copyright (c) 2014-2016 Mihai Moldovan <ionic@ionic.de>                */
/* Copyright (c) 2011-2016 Mike Gabriel <mike.gabriel@das-netzwerkteam.de>*/
/* Copyright (c) 2015-2016 Qindel Group (http://www.qindel.com)           */
/*                                                                        */
/* NXCOMP, NX protocol compression and NX extensions to this software     */
/* are copyright of the aforementioned persons and companies.             */
/*                                                                        */
/* Redistribution and use of the present software is allowed according    */
/* to terms specified in the file LICENSE.nxcomp which comes in the       */
/* source distribution.                                                   */
/*                                                                        */
/* All rights reserved.                                                   */
/*                                                                        */
/* NOTE: This software has received contributions from various other      */
/* contributors, only the core maintainers and supporters are listed as   */
/* copyright holders. Please contact us, if you feel you should be listed */
/* as copyright holder, as well.                                          */
/*                                                                        */
/**************************************************************************/

#ifndef Message_H
#define Message_H

#include <X11/Xproto.h>

#include "NXproto.h"

#include "Misc.h"
#include "Control.h"

#include "Types.h"
#include "Timestamp.h"

#include "ActionCache.h"

#include "StaticCompressor.h"

//
// Forward class declarations.
//

class ChannelCache;

class EncodeBuffer;
class DecodeBuffer;

class WriteBuffer;

//
// Set the verbosity level.
//

#define PANIC
#define WARNING
#undef  TEST
#undef  DEBUG

//
// Define this to know how many messages
// are allocated and deallocated.
//

#undef  REFERENCES

//
// Set default values.  We limit the maximum
// size of a request to 262144 but we need to
// consider the replies, whose size may be up
// to 4MB.
//

#define MESSAGE_ENABLE_CACHE                        0
#define MESSAGE_ENABLE_DATA                         0
#define MESSAGE_ENABLE_SPLIT                        0
#define MESSAGE_ENABLE_COMPRESS                     0

#define MESSAGE_DATA_LIMIT                          4194304 - 4
#define MESSAGE_DATA_OFFSET                         4

#define MESSAGE_CACHE_SLOTS                         6000
#define MESSAGE_CACHE_THRESHOLD                     50
#define MESSAGE_CACHE_LOWER_THRESHOLD               5

//
// Base message class.
//

class Message
{
  friend class MessageStore;
  friend class RenderExtensionStore;

  public:

  Message()
  {
    hits_  = 0;
    last_  = 0;
    locks_ = 0;

    size_   = 0;
    c_size_ = 0;

    md5_digest_ = NULL;

    #ifdef REFERENCES

    references_++;

    *logofs << "Message: Created new message at "
            << this << " out of " << references_
            << " allocated messages.\n"
            << logofs_flush;

    #endif
  }

  Message(const Message &message)
  {
    size_   = message.size_;
    c_size_ = message.c_size_;
    i_size_ = message.i_size_;

    hits_  = message.hits_;
    last_  = message.last_;
    locks_ = message.locks_;

    data_ = message.data_;

    #ifdef REFERENCES

    references_++;

    *logofs << "Message: Creating new copied message at "
            << this << " out of " << references_
            << " allocated messages.\n"
            << logofs_flush;
    #endif

    if (message.md5_digest_ != NULL)
    {
      md5_digest_ = new md5_byte_t[MD5_LENGTH];

      memcpy(md5_digest_, message.md5_digest_, MD5_LENGTH);

      #ifdef DEBUG
      *logofs << "Message: Created MD5 digest for object at "
              << this << ".\n" << logofs_flush;
      #endif
    }
    else
    {
      md5_digest_ = NULL;
    }
  }

  ~Message()
  {
    #ifdef DEBUG
    if (md5_digest_ != NULL)
    {
      *logofs << "Message: Deleted MD5 digest for object at "
              << this << ".\n" << logofs_flush;
    }
    #endif

    delete [] md5_digest_;

    #ifdef REFERENCES

    references_--;

    *logofs << "Message: Deleted message at "
            << this << " out of " << references_
            << " allocated messages.\n"
            << logofs_flush;
    #endif
  }

  //
  // This is the original message size 
  // including the data part regardless
  // data is still stored in the object.
  //

  int size_;

  //
  // This is the size of the identity.
  //

  int i_size_;

  //
  // This is the size, including identity,
  // after message has been 'updated' to 
  // reflect storage of data in compressed
  // format.
  //

  int c_size_;

  protected:

  //
  // This is the data part.
  //

  T_data data_;

  //
  // Time of last hit.
  //

  time_t last_;

  //
  // This is the number of cache hits 
  // registered for the object.
  //

  short int hits_;

  //
  // This is used to mark messages
  // that have been split. 
  //

  short int locks_;

  //
  // This is the MD5 checksum.
  //

  md5_byte_t *md5_digest_;

  //
  // Keep a reference counter
  // of allocated objects.
  //

  #ifdef REFERENCES

  static int references_;

  #endif
};

//
// Repository of messages.
//

class MessageStore
{
  public:

  //
  // Enable or disable cache of messages in store.
  //

  int enableCache;

  //
  // Does message have a distinct data part.
  //

  int enableData;

  //
  // Enable or disable split of data part.
  //

  int enableSplit;

  //
  // Enable or disable compression of data part.
  //

  int enableCompress;

  //
  // Set starting point of data part in the message.
  //

  int dataOffset;

  //
  // Set maximum size for the data part of each message.
  //

  int dataLimit;

  //
  // Set maximum elements in cache.
  //

  int cacheSlots;

  //
  // Set the percentage of total cache memory which
  // a given type of message is allowed to occupy.
  // When threshold is exceeded store is cleaned to
  // make room for a new message of the same type.
  //

  int cacheThreshold;

  //
  // Don't clean the store if percentage of cache
  // memory occupied by messages of this type is
  // below the threshold.
  //

  int cacheLowerThreshold;

  //
  // Last operation performed on cache.
  //

  T_store_action lastAction;

  //
  // Position of last element stored in cache.
  //

  short int lastAdded;

  //
  // Positions of last element found in cache.
  //

  short int lastHit;

  //
  // Position of last element erased.
  //

  short int lastRemoved;

  //
  // Used to encode the the action to
  // perform on the store and the slot
  // involved.
  //

  ActionCache lastActionCache;

  //
  // Position in cache where next insertion
  // is going to take place.
  //

  short int lastRated;

  //
  // Constructors and destructors.
  //

  public:

  MessageStore(StaticCompressor *compressor = NULL);

  virtual ~MessageStore();

  virtual const char *name() const = 0;

  virtual unsigned char opcode() const = 0;

  virtual unsigned int storage() const = 0;

  //
  // These are members that must be specialized.
  //

  public:

  virtual Message *create() const = 0;

  virtual Message *create(const Message &message) const = 0;

  virtual void destroy(Message *message) const = 0;

  void validateSize(int size)
  {
    if (size < control -> MinimumMessageSize ||
            size > control -> MaximumMessageSize)
    {
      *logofs << name() << ": PANIC! Invalid size " << size
              << " for message.\n" << logofs_flush;

      cerr << "Error" << ": Invalid size " << size
           << " for message opcode " << opcode() << ".\n";

      HandleAbort();
    }
  }

  void validateSize(int dataSize, int compressedDataSize)
  {
    if (dataSize < 0 || dataSize > control ->
            MaximumMessageSize - 4 || compressedDataSize < 0 ||
                compressedDataSize >= dataSize)
    {
      *logofs << name() << ": PANIC! Invalid data size "
              << dataSize << " and compressed data size "
              << compressedDataSize << " for message.\n"
              << logofs_flush;

      cerr << "Error" << ": Invalid data size "
           << dataSize << " and compressed data size "
           << compressedDataSize << " for message "
           << "opcode " << (unsigned) opcode() << ".\n";

      HandleAbort();
    }
  }

  //
  // Determine if the message can be stored
  // in the cache.
  //

  virtual int validateMessage(const unsigned char *buffer, int size)
  {
    return (size >= control -> MinimumMessageSize &&
                size <= control -> MaximumMessageSize);
  }

  //
  // Get data offset based on major and minor
  // opcode of the message.
  //

  virtual int identitySize(const unsigned char *buffer, unsigned int size)
  {
    return dataOffset;
  }

  //
  // Encode identity and data using the
  // specific message encoding.
  //
  // Some messages do not implement these
  // methods because the encoding is done
  // directly in the channel loop. Should
  // move the encoding methods in in the
  // message classes.
  //

  virtual int encodeIdentity(EncodeBuffer &encodeBuffer, const unsigned char *buffer,
                                 unsigned int size, int bigEndian,
                                     ChannelCache *channelCache) const
  {
    return 1;
  }

  virtual int decodeIdentity(DecodeBuffer &decodeBuffer, unsigned char *&buffer,
                                 unsigned int &size, int bigEndian, WriteBuffer *writeBuffer,
                                      ChannelCache *channelCache) const
  {
    return 1;
  }

  //
  // Encode differences between message
  // in cache and the one to be encoded.
  //

  virtual void updateIdentity(EncodeBuffer &encodeBuffer, const Message *message,
                                  const Message *cachedMessage, ChannelCache *channelCache) const
  {
  }

  //
  // Decode differences and update the
  // cached version of the same message.
  //

  virtual void updateIdentity(DecodeBuffer &decodeBuffer, const Message *message,
                                  ChannelCache *channelCache) const
  {
  }

  //
  // Post process the message information
  // contained in the store by either up-
  // dating the size record or the actual
  // data part once the message has been
  // completely sent to our peer.
  //

  void updateData(const int position, unsigned int dataSize,
                      unsigned int compressedDataSize);

  void updateData(const T_checksum checksum, unsigned int compressedDataSize);

  void updateData(const int position, const unsigned char *newData,
                      unsigned int dataSize, unsigned int compressedDataSize);

  //
  // These members, used internally
  // in the message store class, are
  // mandatory.
  //

  protected:

  virtual int parseIdentity(Message *message, const unsigned char *buffer,
                                unsigned int size, int bigEndian) const = 0;

  virtual int unparseIdentity(const Message *message, unsigned char *buffer,
                                  unsigned int size, int bigEndian) const = 0;

  virtual void identityChecksum(const Message *message, const unsigned char *buffer,
                                    unsigned int size, int bigEndian) const = 0;

  virtual void dumpIdentity(const Message *message) const = 0;

  //
  // Design should preserve these from being
  // virtual.
  //

  int parseData(Message *message, int split, const unsigned char *buffer,
                    unsigned int size, T_checksum_action checksumAction,
                        T_data_action dataAction, int bigEndian);

  int parseData(Message *message, const unsigned char *buffer,
                    unsigned int size, const unsigned char *compressedData,
                        const unsigned int compressedDataSize, T_checksum_action checksumAction,
                            T_data_action dataAction, int bigEndian);

  int unparseData(const Message *message, unsigned char *buffer,
                      unsigned int size, int bigEndian);

  //
  // Manage efficient allocation of messages
  // in the heap.
  //

  void recycle(Message *message)
  {
    #ifdef TEST

    if (message == NULL)
    {
      *logofs << name() << ": PANIC! Cannot recycle a null message.\n"
              << logofs_flush;

      cerr << "Error" << ": Cannot recycle a null message.\n";

      HandleAbort();
    }

    #endif

    if (temporary_ == NULL)
    {
      //
      // Be careful when reusing the message as
      // it can contain valid data that must be
      // explicitly deallocated if not needed.
      // Note also that you cannot count on the
      // initialization made in costructor.
      //

      temporary_ = message;
    }
    else
    {
      destroy(message);
    }
  }

  void beginChecksum(Message *message)
  {
    if (message -> md5_digest_ == NULL)
    {
      message -> md5_digest_ = new md5_byte_t[MD5_LENGTH];

      #ifdef DEBUG
      *logofs << name() << ": Created MD5 digest structure "
              << "for object at " << message << ".\n"
              << logofs_flush;
      #endif
    }
    #ifdef DEBUG
    else
    {
      *logofs << name() << ": Using existing MD5 digest structure "
              << "for object at " << message << ".\n"
              << logofs_flush;
    }
    #endif

    #ifdef DEBUG
    *logofs << name() << ": Prepared MD5 digest for object at "
            << message << ".\n" << logofs_flush;
    #endif

    md5_init(md5_state_);
  }

  void endChecksum(Message *message)
  {
    md5_finish(md5_state_, message -> md5_digest_);

    #ifdef DEBUG
    *logofs << name() << ": Calculated checksum for object at "
            << message << ".\n" << logofs_flush;
    #endif
  }

  void dataChecksum(Message *message, const unsigned char *buffer,
                        unsigned int size, int bigEndian)
  {
    //
    // Messages that have a data part starting
    // at an offset possibly beyond the end of
    // the message, must include the size in
    // the identity checksum.
    //

    if ((int) size > message -> i_size_)
    {
      md5_append(md5_state_, buffer + message -> i_size_,
                     size - message -> i_size_);
    }
  }

  //
  // Repository handling methods.
  //

  public:

  //
  // Extract identity and data from buffer.
  // The size field will be updated at the
  // time of data parsing.
  //

  int parse(Message *message, int split, const unsigned char *buffer, unsigned int size,
                T_checksum_action checksumAction, T_data_action dataAction, int bigEndian);

  int parse(Message *message, const unsigned char *buffer, unsigned int size,
                const unsigned char *compressedData, const unsigned int compressedDataSize,
                     T_checksum_action checksumAction, T_data_action dataAction, int bigEndian);

  //
  // From identity and data write the
  // final message to the raw buffer.
  //

  int unparse(const Message *message, unsigned char *buffer,
                  unsigned int size, int bigEndian)
  {
    return (unparseData(message, buffer, size, bigEndian) &&
                unparseIdentity(message, buffer, size, bigEndian));
  }

  void dump(const Message *message) const
  {
    dumpIdentity(message);

    dumpData(message);
  }

  void dumpData(const Message *message) const;

  //
  // This returns the original message size as it
  // was received on the link. It takes in account
  // the data part, regardless data is still stored
  // in the message object. This information will
  // be used at the time message is unparsed.
  //

  int plainSize(const int position) const
  {
    return (*messages_)[position] -> size_;
  }

  //
  // This returns either the size of identity plus
  // the compressed data part or 0 if message is
  // stored in uncompressed format.
  //

  int compressedSize(const int position) const
  {
    return (*messages_)[position] -> c_size_;
  }

  //
  // Returns a pointer to message
  // given its position in cache.
  //

  Message *get(const int position) const
  {
    if (position < 0 || position >= cacheSlots)
    {
      #ifdef PANIC
      *logofs << name() << ": PANIC! Requested position "
              << position << " is not inside the "
              << "container.\n" << logofs_flush;
      #endif

      cerr << "Error" << ": Requested position "
           << position << " is not inside the"
           << "container.\n";

      HandleAbort();
    }
    else if ((*messages_)[position] == NULL)
    {
      #ifdef PANIC
      *logofs << name() << ": PANIC! Message at position "
              << position << " is NULL.\n"
              << logofs_flush;
      #endif

      cerr << "Error" << ": Message at position "
           << position << " is NULL.\n";

      HandleAbort();
    }

    return (*messages_)[position];
  }

  //
  // This is the method called at encoding
  // side to add a message to cache.
  //

  int findOrAdd(Message *message, T_checksum_action checksumAction,
                    T_data_action dataAction, int &added, int &locked);

  //
  // Utility interfaces to message insertion
  // and deletion.
  //

  int add(Message *message, const int position,
              T_checksum_action checksumAction, T_data_action dataAction);

  int remove(const int position, T_checksum_action checksumAction,
                 T_data_action dataAction);

  //
  // Make space in the repository by remove
  // the first suitable message object.
  //

  int clean(T_checksum_action checksumAction);

  //
  // Increase or decrease the "rating" of
  // the message object.
  //

  int touch(Message *message) const;
  int untouch(Message *message) const;

  int getTouches(const int position) const
  {
    Message *message = (*messages_)[position];

    if (message == NULL)
    {
      return 0;
    }

    return message -> hits_;
  }

  //
  // Gives a 'weight' to the cached message. A zero
  // value means object can be safely removed. A value
  // greater than zero means it is advisable to retain
  // the object. A negative result means it is mandato-
  // ry to keep object in cache.
  //

  int getRating(Message *message, T_rating type) const;

  //
  // Increase or decrease locks of message at given
  // position. A locked message will not be removed
  // from the message store until the lock counter
  // is zero.
  //

  int lock(const int position) const;
  int unlock(const int position) const;

  int getLocks(const int position) const
  {
    Message *message = (*messages_)[position];

    if (message == NULL)
    {
      return 0;
    }

    return message -> locks_;
  }

  T_checksum const getChecksum(const int position) const
  {
    return getChecksum(get(position));
  }

  T_checksum const getChecksum(const Message *message) const
  {
    if (message -> md5_digest_ == NULL)
    {
      #ifdef PANIC
      *logofs << name() << ": PANIC! Checksum not initialized "
              << "for object at " << message << ".\n"
              << logofs_flush;
      #endif

      cerr << "Error" << ": Checksum not initialized "
           << "for object at " << message << ".\n";

      HandleAbort();
    }

    #ifdef DEBUG
    *logofs << name() << ": Got checksum for object at " 
            << message << ".\n" << logofs_flush;
    #endif

    return message -> md5_digest_;
  }

  //
  // Calculate the checksum on the fly based the
  // opcode in the buffer. Useful in the case a
  // message was not processed or was not stored
  // in the cache. The returned checksum must be
  // explicitly deallocated by the caller, after
  // use.
  //

  T_checksum getChecksum(const unsigned char *buffer,
                             unsigned int size, int bigEndian);

  const unsigned char *getData(const Message *message) const
  {
    return message -> data_.begin();
  }

  int plainSize(const Message *message) const
  {
    return message -> size_;
  }

  int identitySize(Message *message)
  {
    return message -> i_size_;
  }

  int compressedSize(const Message *message) const
  {
    return message -> c_size_;
  }

  Message *getTemporary()
  {
    if (temporary_ == NULL)
    {
      temporary_ = create();
    }

    return temporary_;
  }

  void resetTemporary()
  {
    temporary_ = NULL;
  }

  //
  // On side where we don't have checksums, we
  // count how many messages are in the array.
  // This is obviously expensive and should be
  // only performed when reporting statistics.
  //

  int getSize() const
  {
    int size = checksums_ -> size();

    if (size == 0)
    {
      for (int i = 0; i < cacheSlots; i++)
      {
        if ((*messages_)[i] != NULL)
        {
          size++;
        }
      }
    }

    return size;
  }

  int getLocalStorageSize() const
  {
    return localStorageSize_;
  }

  int getRemoteStorageSize() const
  {
    return remoteStorageSize_;
  }

  int getLocalTotalStorageSize() const
  {
    return totalLocalStorageSize_;
  }

  int getRemoteTotalStorageSize() const
  {
    return totalRemoteStorageSize_;
  }

  static int getCumulativeTotalStorageSize()
  {
    return (totalLocalStorageSize_ > totalRemoteStorageSize_ ?
                totalLocalStorageSize_ : totalRemoteStorageSize_);
  }

  int saveStore(ostream *cachefs, md5_state_t *md5_state_stream,
                    md5_state_t *md5_state_client, T_checksum_action checksumAction,
                        T_data_action dataAction, int bigEndian);

  int loadStore(istream *cachefs, md5_state_t *md5_state_stream,
                    T_checksum_action checksumAction, T_data_action dataAction,
                        int bigEndian);

  protected:

  //
  // Estimate the memory requirements of given
  // instance of message. Size includes memory
  // allocated from heap to store checksum and
  // data.
  //

  void storageSize(const Message *message, unsigned int &local,
                       unsigned int &remote) const;

  //
  // Just used for debug.
  //

  void printStorageSize();

  //
  // Repositories where to save cached messages.
  // First is a vector of pointers, the second
  // is a hash table used for fast lookups.
  //

  T_messages  *messages_;
  T_checksums *checksums_;

  //
  // A message object to be used as a temporary.
  // Reuse the temporary object if possible, if
  // not, create a new instance.
  //

  Message *temporary_;

  //
  // Used to calculate message's checksum.
  //

  md5_state_t *md5_state_;

  private:

  //
  // Used to compress data payload.
  //

  StaticCompressor *compressor_;

  //
  // Keep track of how many bytes 
  // are taken by cache.
  //

  int localStorageSize_;
  int remoteStorageSize_;

  static int totalLocalStorageSize_;
  static int totalRemoteStorageSize_;

  //
  // Used to track object allocation and deallocation.
  //

  #ifdef REFERENCES

  static int references_;

  #endif
};

//
// This is an ancillary class of the message
// store, used to encode extensions based on
// the minor opcode.
//

class MinorMessageStore
{
  public:

  virtual ~MinorMessageStore()
  {
  }

  virtual const char *name() const = 0;

  virtual int identitySize(const unsigned char *buffer, unsigned int size) = 0;

  virtual int encodeMessage(EncodeBuffer &encodeBuffer, const unsigned char *buffer,
                                const unsigned int size, int bigEndian,
                                    ChannelCache *channelCache) const
  {
    return 1;
  }

  virtual int decodeMessage(DecodeBuffer &decodeBuffer, unsigned char *&buffer,
                                unsigned int &size, unsigned char type, int bigEndian,
                                    WriteBuffer *writeBuffer, ChannelCache *channelCache) const
  {
    return 1;
  }

  virtual void encodeData(EncodeBuffer &encodeBuffer, const unsigned char *buffer,
                              unsigned int size, int bigEndian,
                                  ChannelCache *channelCache) const
  {
  }

  virtual void decodeData(DecodeBuffer &decodeBuffer, unsigned char *buffer,
                              unsigned int size, int bigEndian,
                                  ChannelCache *channelCache) const
  {
  }

  virtual int parseIdentity(Message *message, const unsigned char *buffer,
                                unsigned int size, int bigEndian) const = 0;

  virtual int unparseIdentity(const Message *message, unsigned char *buffer,
                                  unsigned int size, int bigEndian) const = 0;

  virtual void updateIdentity(EncodeBuffer &encodeBuffer, const Message *message,
                                  const Message *cachedMessage,
                                      ChannelCache *channelCache) const
  {
  }

  virtual void updateIdentity(DecodeBuffer &decodeBuffer, const Message *message,
                                  ChannelCache *channelCache) const
  {
  }

  virtual void identityChecksum(const Message *message, const unsigned char *buffer,
                                    unsigned int size, md5_state_t *md5_state,
                                        int bigEndian) const = 0;
};

#endif /* Message_H */