DataOutput.hpp

Go to the documentation of this file.

#pragma once

#ifndef GEODE_DATAOUTPUT_H_
#define GEODE_DATAOUTPUT_H_

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "geode_globals.hpp"
#include "ExceptionTypes.hpp"
#include "Log.hpp"
#include "Serializable.hpp"
#include "CacheableString.hpp"

#include <cstring>
#include <string>
#include <cstdlib>

namespace apache {
namespace geode {
namespace client {

#define GF_ALLOC(v, t, s)                                                  \
  {                                                                        \
    v = (t*)malloc((s) * sizeof(t));                                       \
    if ((v) == NULL) {                                                     \
      throw OutOfMemoryException(                                          \
          "Out of Memory while allocating buffer for " #t " of size " #s); \
    }                                                                      \
  }

#define GF_RESIZE(v, t, s)                                \
  {                                                       \
    v = (t*)realloc(v, (s) * sizeof(t));                  \
    if ((v) == NULL) {                                    \
      throw OutOfMemoryException(                         \
          "Out of Memory while resizing buffer for " #t); \
    }                                                     \
  }

#define GF_FREE(v) free(v)

class CPPCACHE_EXPORT DataOutput {
 public:
  DataOutput();

  inline void write(uint8_t value) {
    ensureCapacity(1);
    writeNoCheck(value);
  }

  inline void write(int8_t value) { write(static_cast<uint8_t>(value)); }

  inline void writeBoolean(bool value) { write(static_cast<uint8_t>(value)); }

  inline void writeBytes(const uint8_t* bytes, int32_t len) {
    if (len >= 0) {
      ensureCapacity(len + 5);
      writeArrayLen(bytes == NULL ? 0 : len);  // length of bytes...
      if (len > 0 && bytes != NULL) {
        std::memcpy(m_buf, bytes, len);
        m_buf += len;
      }
    } else {
      write(static_cast<int8_t>(-1));
    }
  }

  inline void writeBytes(const int8_t* bytes, int32_t len) {
    writeBytes(reinterpret_cast<const uint8_t*>(bytes), len);
  }

  inline void writeBytesOnly(const uint8_t* bytes, uint32_t len) {
    ensureCapacity(len);
    std::memcpy(m_buf, bytes, len);
    m_buf += len;
  }

  inline void writeBytesOnly(const int8_t* bytes, uint32_t len) {
    writeBytesOnly(reinterpret_cast<const uint8_t*>(bytes), len);
  }

  inline void writeInt(uint16_t value) {
    ensureCapacity(2);
    *(m_buf++) = static_cast<uint8_t>(value >> 8);
    *(m_buf++) = static_cast<uint8_t>(value);
  }

  inline void writeChar(uint16_t value) {
    ensureCapacity(2);
    *(m_buf++) = static_cast<uint8_t>(value >> 8);
    *(m_buf++) = static_cast<uint8_t>(value);
  }

  inline void writeInt(uint32_t value) {
    ensureCapacity(4);
    *(m_buf++) = static_cast<uint8_t>(value >> 24);
    *(m_buf++) = static_cast<uint8_t>(value >> 16);
    *(m_buf++) = static_cast<uint8_t>(value >> 8);
    *(m_buf++) = static_cast<uint8_t>(value);
  }

  inline void writeInt(uint64_t value) {
    ensureCapacity(8);
    // the defines are not reliable and can be changed by compiler options.
    // Hence using sizeof() test instead.
    //#if defined(_LP64) || ( defined(__WORDSIZE) && __WORDSIZE == 64 ) ||
    //( defined(_INTEGRAL_MAX_BITS) && _INTEGRAL_MAX_BITS >= 64 )
    if (sizeof(long) == 8) {
      *(m_buf++) = static_cast<uint8_t>(value >> 56);
      *(m_buf++) = static_cast<uint8_t>(value >> 48);
      *(m_buf++) = static_cast<uint8_t>(value >> 40);
      *(m_buf++) = static_cast<uint8_t>(value >> 32);
      *(m_buf++) = static_cast<uint8_t>(value >> 24);
      *(m_buf++) = static_cast<uint8_t>(value >> 16);
      *(m_buf++) = static_cast<uint8_t>(value >> 8);
      *(m_buf++) = static_cast<uint8_t>(value);
    } else {
      uint32_t hword = static_cast<uint32_t>(value >> 32);
      *(m_buf++) = static_cast<uint8_t>(hword >> 24);
      *(m_buf++) = static_cast<uint8_t>(hword >> 16);
      *(m_buf++) = static_cast<uint8_t>(hword >> 8);
      *(m_buf++) = static_cast<uint8_t>(hword);

      hword = static_cast<uint32_t>(value);
      *(m_buf++) = static_cast<uint8_t>(hword >> 24);
      *(m_buf++) = static_cast<uint8_t>(hword >> 16);
      *(m_buf++) = static_cast<uint8_t>(hword >> 8);
      *(m_buf++) = static_cast<uint8_t>(hword);
    }
  }

  inline void writeInt(int16_t value) {
    writeInt(static_cast<uint16_t>(value));
  }

  inline void writeInt(int32_t value) {
    writeInt(static_cast<uint32_t>(value));
  }

  inline void writeInt(int64_t value) {
    writeInt(static_cast<uint64_t>(value));
  }

  inline void writeArrayLen(int32_t len) {
    if (len == -1) {
      write(static_cast<int8_t>(-1));
    } else if (len <= 252) {  // 252 is java's ((byte)-4 && 0xFF)
      write(static_cast<uint8_t>(len));
    } else if (len <= 0xFFFF) {
      write(static_cast<int8_t>(-2));
      writeInt(static_cast<uint16_t>(len));
    } else {
      write(static_cast<int8_t>(-3));
      writeInt(len);
    }
  }

  inline void writeFloat(float value) {
    union float_uint32_t {
      float f;
      uint32_t u;
    } v;
    v.f = value;
    writeInt(v.u);
  }

  inline void writeDouble(double value) {
    union double_uint64_t {
      double d;
      uint64_t ll;
    } v;
    v.d = value;
    writeInt(v.ll);
  }

  inline void writeASCII(const char* value, uint32_t length = 0) {
    if (value != NULL) {
      if (length == 0) {
        length = static_cast<uint32_t>(strlen(value));
      }
      uint16_t len = static_cast<uint16_t>(length > 0xFFFF ? 0xFFFF : length);
      writeInt(len);
      writeBytesOnly((int8_t*)value, len);  // K64
    } else {
      writeInt(static_cast<uint16_t>(0));
    }
  }

  inline void writeNativeString(const char* value) {
    // create cacheable string
    // write typeid id.
    // call todata
    CacheableStringPtr csPtr = CacheableString::create(value);
    write(csPtr->typeId());
    csPtr->toData(*this);
  }

  inline void writeASCIIHuge(const char* value, uint32_t length = 0) {
    if (value != NULL) {
      if (length == 0) {
        length = static_cast<uint32_t>(strlen(value));
      }
      writeInt(length);
      writeBytesOnly((int8_t*)value, length);
    } else {
      writeInt(static_cast<uint32_t>(0));
    }
  }

  inline void writeFullUTF(const char* value, uint32_t length = 0) {
    if (value != NULL) {
      int32_t encodedLen = getEncodedLength(value, length);
      writeInt(encodedLen);
      ensureCapacity(encodedLen);
      write(static_cast<int8_t>(0));  // isObject = 0 BYTE_CODE
      uint8_t* end = m_buf + encodedLen;
      while (m_buf < end) {
        encodeChar(*value++);
      }
      if (m_buf > end) m_buf = end;
    } else {
      writeInt(static_cast<uint16_t>(0));
    }
  }

  inline void writeUTF(const char* value, uint32_t length = 0) {
    if (value != NULL) {
      int32_t len = getEncodedLength(value, length);
      uint16_t encodedLen = static_cast<uint16_t>(len > 0xFFFF ? 0xFFFF : len);
      writeInt(encodedLen);
      ensureCapacity(encodedLen);
      uint8_t* end = m_buf + encodedLen;
      while (m_buf < end) {
        encodeChar(*value++);
      }
      if (m_buf > end) m_buf = end;
    } else {
      writeInt(static_cast<uint16_t>(0));
    }
  }

  inline void writeUTFHuge(const char* value, uint32_t length = 0) {
    if (value != NULL) {
      if (length == 0) {
        length = static_cast<uint32_t>(strlen(value));
      }
      writeInt(length);
      ensureCapacity(length * 2);
      for (uint32_t pos = 0; pos < length; pos++) {
        writeNoCheck(static_cast<int8_t>(0));
        writeNoCheck(static_cast<int8_t>(value[pos]));
      }
    } else {
      writeInt(static_cast<uint32_t>(0));
    }
  }

  inline void writeUTF(const wchar_t* value, uint32_t length = 0) {
    if (value != NULL) {
      int32_t len = getEncodedLength(value, length);
      uint16_t encodedLen = static_cast<uint16_t>(len > 0xFFFF ? 0xFFFF : len);
      writeInt(encodedLen);
      ensureCapacity(encodedLen);
      uint8_t* end = m_buf + encodedLen;
      while (m_buf < end) {
        encodeChar(*value++);
      }
      if (m_buf > end) m_buf = end;
    } else {
      writeInt(static_cast<uint16_t>(0));
    }
  }

  inline void writeUTFHuge(const wchar_t* value, uint32_t length = 0) {
    if (value != NULL) {
      if (length == 0) {
        length = static_cast<uint32_t>(wcslen(value));
      }
      writeInt(length);
      ensureCapacity(length * 2);
      for (uint32_t pos = 0; pos < length; pos++) {
        uint16_t item = static_cast<uint16_t>(value[pos]);
        writeNoCheck(static_cast<uint8_t>((item & 0xFF00) >> 8));
        writeNoCheck(static_cast<uint8_t>(item & 0xFF));
      }
    } else {
      writeInt(static_cast<uint32_t>(0));
    }
  }

  inline static int32_t getEncodedLength(const char* value, int32_t length = 0,
                                         uint32_t* valLength = NULL) {
    if (value == NULL) return 0;
    char currentChar;
    int32_t encodedLen = 0;
    const char* start = value;
    if (length == 0) {
      while ((currentChar = *value) != '\0') {
        getEncodedLength(currentChar, encodedLen);
        value++;
      }
    } else {
      const char* end = value + length;
      while (value < end) {
        currentChar = *value;
        getEncodedLength(currentChar, encodedLen);
        value++;
      }
    }
    if (valLength != NULL) {
      *valLength = static_cast<uint32_t>(value - start);
    }
    return encodedLen;
  }

  inline static int32_t getEncodedLength(const wchar_t* value,
                                         int32_t length = 0,
                                         uint32_t* valLength = NULL) {
    if (value == NULL) return 0;
    wchar_t currentChar;
    int32_t encodedLen = 0;
    const wchar_t* start = value;
    if (length == 0) {
      while ((currentChar = *value) != 0) {
        getEncodedLength(currentChar, encodedLen);
        value++;
      }
    } else {
      const wchar_t* end = value + length;
      while (value < end) {
        currentChar = *value;
        getEncodedLength(currentChar, encodedLen);
        value++;
      }
    }
    if (valLength != NULL) {
      *valLength = static_cast<uint32_t>(value - start);
    }
    return encodedLen;
  }

  template <class PTR>
  void writeObject(const SharedPtr<PTR>& objptr, bool isDelta = false) {
    writeObjectInternal(objptr.ptr(), isDelta);
  }

  void writeObject(const Serializable* objptr) { writeObjectInternal(objptr); }

  const uint8_t* getCursor() { return m_buf; }

  void advanceCursor(uint32_t offset) {
    ensureCapacity(offset);
    m_buf += offset;
  }

  void rewindCursor(uint32_t offset) { m_buf -= offset; }

  void updateValueAtPos(uint32_t offset, uint8_t value) {
    m_bytes[offset] = value;
  }

  uint8_t getValueAtPos(uint32_t offset) { return m_bytes[offset]; }
  ~DataOutput() {
    reset();
    DataOutput::checkinBuffer(m_bytes, m_size);
  }

  inline const uint8_t* getBuffer() const {
    // GF_R_ASSERT(!((uint32_t)(m_bytes) % 4));
    return m_bytes;
  }

  inline uint32_t getRemainingBufferLength() const {
    // GF_R_ASSERT(!((uint32_t)(m_bytes) % 4));
    return m_size - getBufferLength();
  }

  inline const uint8_t* getBuffer(uint32_t* rsize) const {
    *rsize = static_cast<uint32_t>(m_buf - m_bytes);
    // GF_R_ASSERT(!((uint32_t)(m_bytes) % 4));
    return m_bytes;
  }

  inline uint8_t* getBufferCopy() {
    uint32_t size = static_cast<uint32_t>(m_buf - m_bytes);
    uint8_t* result;
    GF_ALLOC(result, uint8_t, size);
    std::memcpy(result, m_bytes, size);
    return result;
  }

  inline uint32_t getBufferLength() const {
    return static_cast<uint32_t>(m_buf - m_bytes);
  }

  inline void reset() {
    if (m_haveBigBuffer) {
      // free existing buffer
      GF_FREE(m_bytes);
      // create smaller buffer
      GF_ALLOC(m_bytes, uint8_t, m_lowWaterMark);
      m_size = m_lowWaterMark;
      // reset the flag
      m_haveBigBuffer = false;
      // release the lock
      releaseLock();
    }
    m_buf = m_bytes;
  }

  // make sure there is room left for the requested size item.
  inline void ensureCapacity(uint32_t size) {
    uint32_t offset = static_cast<uint32_t>(m_buf - m_bytes);
    if ((m_size - offset) < size) {
      uint32_t newSize = m_size * 2 + (8192 * (size / 8192));
      if (newSize >= m_highWaterMark && !m_haveBigBuffer) {
        // acquire the lock
        acquireLock();
        // set flag
        m_haveBigBuffer = true;
      }
      m_size = newSize;
      GF_RESIZE(m_bytes, uint8_t, m_size);
      m_buf = m_bytes + offset;
    }
  }

  /*
  * This is for internal use
  */
  const char* getPoolName() { return m_poolName; }

  /*
   * This is for internal use
   */
  void setPoolName(const char* poolName) { m_poolName = poolName; }

  uint8_t* getBufferCopyFrom(const uint8_t* from, uint32_t length) {
    uint8_t* result;
    GF_NEW(result, uint8_t[length]);
    std::memcpy(result, from, length);

    return result;
  }

  static void safeDelete(uint8_t* src) { GF_SAFE_DELETE(src); }

  static DataOutput* getDataOutput() { return new DataOutput(); }
  static void releaseDataOutput(DataOutput* dataOutput) {
    GF_SAFE_DELETE(dataOutput);
  }

 private:
  void writeObjectInternal(const Serializable* ptr, bool isDelta = false);

  static void acquireLock();
  static void releaseLock();

  const char* m_poolName;
  // memory m_buffer to encode to.
  uint8_t* m_bytes;
  // cursor.
  uint8_t* m_buf;
  // size of m_bytes.
  uint32_t m_size;
  // high and low water marks for buffer size
  static uint32_t m_lowWaterMark;
  static uint32_t m_highWaterMark;
  // flag to indicate we have a big buffer
  volatile bool m_haveBigBuffer;

  inline static void getEncodedLength(const char val, int32_t& encodedLen) {
    if ((val == 0) || (val & 0x80)) {
      // two byte.
      encodedLen += 2;
    } else {
      // one byte.
      encodedLen++;
    }
  }

  inline static void getEncodedLength(const wchar_t val, int32_t& encodedLen) {
    if (val == 0) {
      encodedLen += 2;
    } else if (val < 0x80)  // ASCII character
    {
      encodedLen++;
    } else if (val < 0x800) {
      encodedLen += 2;
    } else {
      encodedLen += 3;
    }
  }

  inline void encodeChar(const char value) {
    uint8_t tmp = static_cast<uint8_t>(value);
    if ((tmp == 0) || (tmp & 0x80)) {
      // two byte.
      *(m_buf++) = static_cast<uint8_t>(0xc0 | ((tmp & 0xc0) >> 6));
      *(m_buf++) = static_cast<uint8_t>(0x80 | (tmp & 0x3f));
    } else {
      // one byte.
      *(m_buf++) = tmp;
    }
  }

  // this will lose the character set encoding.
  inline void encodeChar(const wchar_t value) {
    uint16_t c = static_cast<uint16_t>(value);
    if (c == 0) {
      *(m_buf++) = 0xc0;
      *(m_buf++) = 0x80;
    } else if (c < 0x80) {  // ASCII character
      *(m_buf++) = static_cast<uint8_t>(c);
    } else if (c < 0x800) {
      *(m_buf++) = static_cast<uint8_t>(0xC0 | c >> 6);
      *(m_buf++) = static_cast<uint8_t>(0x80 | (c & 0x3F));
    } else {
      *(m_buf++) = static_cast<uint8_t>(0xE0 | c >> 12);
      *(m_buf++) = static_cast<uint8_t>(0x80 | ((c >> 6) & 0x3F));
      *(m_buf++) = static_cast<uint8_t>(0x80 | (c & 0x3F));
    }
  }

  inline void writeNoCheck(uint8_t value) { *(m_buf++) = value; }

  inline void writeNoCheck(int8_t value) {
    writeNoCheck(static_cast<uint8_t>(value));
  }

  static uint8_t* checkoutBuffer(uint32_t* size);
  static void checkinBuffer(uint8_t* buffer, uint32_t size);

  // disable copy constructor and assignment
  DataOutput(const DataOutput&);
  DataOutput& operator=(const DataOutput&);
};
}  // namespace client
}  // namespace geode
}  // namespace apache

#endif  // GEODE_DATAOUTPUT_H_