TBufferTransports.h

Go to the documentation of this file.

/*
 * 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.
 */

#ifndef _SANDESH_TRANSPORT_TBUFFERTRANSPORTS_H_
#define _SANDESH_TRANSPORT_TBUFFERTRANSPORTS_H_ 1

#include <cstring>
#include "boost/scoped_array.hpp"

#include <sandesh/transport/TTransport.h>
#include <sandesh/transport/TVirtualTransport.h>

#ifdef __GNUC__
#define TDB_LIKELY(val) (__builtin_expect((val), 1))
#define TDB_UNLIKELY(val) (__builtin_expect((val), 0))
#else
#define TDB_LIKELY(val) (val)
#define TDB_UNLIKELY(val) (val)
#endif

namespace contrail { namespace sandesh { namespace transport {


class TBufferBase : public TVirtualTransport<TBufferBase> {

 public:

  int32_t read(uint8_t* buf, uint32_t len) {
    uint8_t* new_rBase = rBase_ + len;
    if (TDB_LIKELY(new_rBase <= rBound_)) {
      std::memcpy(buf, rBase_, len);
      rBase_ = new_rBase;
      return len;
    }
    return readSlow(buf, len);
  }

  int32_t readAll(uint8_t* buf, uint32_t len) {
    uint8_t* new_rBase = rBase_ + len;
    if (TDB_LIKELY(new_rBase <= rBound_)) {
      std::memcpy(buf, rBase_, len);
      rBase_ = new_rBase;
      return len;
    }
    return contrail::sandesh::transport::readAll(*this, buf, len);
  }

  int write(const uint8_t* buf, uint32_t len) {
    uint8_t* new_wBase = wBase_ + len;
    if (TDB_LIKELY(new_wBase <= wBound_)) {
      std::memcpy(wBase_, buf, len);
      wBase_ = new_wBase;
      return 0;
    }
    return writeSlow(buf, len);
  }

  const uint8_t* borrow(uint8_t* buf, uint32_t* len) {
    if (TDB_LIKELY(static_cast<ptrdiff_t>(*len) <= rBound_ - rBase_)) {
      // With strict aliasing, writing to len shouldn't force us to
      // refetch rBase_ from memory.  TODO(dreiss): Verify this.
      *len = static_cast<uint32_t>(rBound_ - rBase_);
      return rBase_;
    }
    return borrowSlow(buf, len);
  }

  void consume(uint32_t len) {
    if (TDB_LIKELY(static_cast<ptrdiff_t>(len) <= rBound_ - rBase_)) {
      rBase_ += len;
    } else {
      LOG(ERROR, __func__ << ": Consume did not follow a borrow.");
      assert(false);
    }
  }

  void setReadBuffer(uint8_t* buf, uint32_t len) {
    rBase_ = buf;
    rBound_ = buf+len;
  }

  void setWriteBuffer(uint8_t* buf, uint32_t len) {
    wBase_ = buf;
    wBound_ = buf+len;
  }

 protected:

  virtual uint32_t readSlow(uint8_t* buf, uint32_t len) = 0;

  virtual int writeSlow(const uint8_t* buf, uint32_t len) = 0;

  virtual const uint8_t* borrowSlow(uint8_t* buf, uint32_t* len) = 0;

  TBufferBase()
    : rBase_(NULL)
    , rBound_(NULL)
    , wBase_(NULL)
    , wBound_(NULL)
  {}

  virtual ~TBufferBase() {}

  uint8_t* rBase_;
  uint8_t* rBound_;

  uint8_t* wBase_;
  uint8_t* wBound_;
};

class TMemoryBuffer : public TVirtualTransport<TMemoryBuffer, TBufferBase> {
 private:

  // Common initialization done by all constructors.
  void initCommon(uint8_t* buf, uint32_t size, bool owner, uint32_t wPos) {
    if (buf == NULL && size != 0) {
      assert(owner);
      buf = (uint8_t*)std::malloc(size);
      if (buf == NULL) {
        LOG(ERROR, __func__ << ": Allocation of " << size << " bytes FAILED");
        assert(false);
      }
    }

    buffer_ = buf;
    bufferSize_ = size;

    rBase_ = buffer_;
    rBound_ = buffer_ + wPos;
    // TODO(dreiss): Investigate NULL-ing this if !owner.
    wBase_ = buffer_ + wPos;
    wBound_ = buffer_ + bufferSize_;

    owner_ = owner;

    // rBound_ is really an artifact.  In principle, it should always be
    // equal to wBase_.  We update it in a few places (computeRead, etc.).
  }

 public:
  static const uint32_t defaultSize = 1024;

  enum MemoryPolicy
  { OBSERVE = 1
  , COPY = 2
  , TAKE_OWNERSHIP = 3
  };

  TMemoryBuffer() {
    initCommon(NULL, defaultSize, true, 0);
  }

  TMemoryBuffer(uint32_t sz) {
    initCommon(NULL, sz, true, 0);
  }

  TMemoryBuffer(uint8_t* buf, uint32_t sz, MemoryPolicy policy = OBSERVE) {
    if (buf == NULL && sz != 0) {
      LOG(ERROR, __func__ << ": TMemoryBuffer given null buffer with "
          "non-zero size.");
      assert(false);
    }
    switch (policy) {
      case OBSERVE:
      case TAKE_OWNERSHIP:
        initCommon(buf, sz, policy == TAKE_OWNERSHIP, sz);
        break;
      case COPY:
        initCommon(NULL, sz, true, 0);
        this->write(buf, sz);
        break;
      default:
        LOG(ERROR, __func__ << ": Invalid MemoryPolicy for TMemoryBuffer");
        assert(false);
        break;
    }
  }

  ~TMemoryBuffer() {
    if (owner_) {
      std::free(buffer_);
    }
  }

  bool isOpen() {
    return true;
  }

  bool peek() {
    return (rBase_ < wBase_);
  }

  void open() {}

  void close() {}

  // TODO(dreiss): Make bufPtr const.
  void getBuffer(uint8_t** bufPtr, uint32_t* sz) {
    *bufPtr = rBase_;
    *sz = static_cast<uint32_t>(wBase_ - rBase_);
  }

  std::string getBufferAsString() {
    if (buffer_ == NULL) {
      return "";
    }
    uint8_t* buf;
    uint32_t sz;
    getBuffer(&buf, &sz);
    return std::string((char*)buf, (std::string::size_type)sz);
  }

  void appendBufferToString(std::string& str) {
    if (buffer_ == NULL) {
      return;
    }
    uint8_t* buf;
    uint32_t sz;
    getBuffer(&buf, &sz);
    str.append((char*)buf, sz);
  }

  void resetBuffer() {
    rBase_ = buffer_;
    rBound_ = buffer_;
    wBase_ = buffer_;
    // It isn't safe to write into a buffer we don't own.
    if (!owner_) {
      wBound_ = wBase_;
      bufferSize_ = 0;
    }
  }

  void resetBuffer(uint8_t* buf, uint32_t sz, MemoryPolicy policy = OBSERVE) {
    // Use a variant of the copy-and-swap trick for assignment operators.
    // This is sub-optimal in terms of performance for two reasons:
    //   1/ The constructing and swapping of the (small) values
    //      in the temporary object takes some time, and is not necessary.
    //   2/ If policy == COPY, we allocate the new buffer before
    //      freeing the old one, precluding the possibility of
    //      reusing that memory.
    // I doubt that either of these problems could be optimized away,
    // but the second is probably no a common case, and the first is minor.
    // I don't expect resetBuffer to be a common operation, so I'm willing to
    // bite the performance bullet to make the method this simple.

    // Construct the new buffer.
    TMemoryBuffer new_buffer(buf, sz, policy);
    // Move it into ourself.
    this->swap(new_buffer);
    // Our old self gets destroyed.
  }

  void resetBuffer(uint32_t sz) {
    // Construct the new buffer.
    TMemoryBuffer new_buffer(sz);
    // Move it into ourself.
    this->swap(new_buffer);
    // Our old self gets destroyed.
  }

  std::string readAsString(uint32_t len) {
    std::string str;
    (void)readAppendToString(str, len);
    return str;
  }

  uint32_t readAppendToString(std::string& str, uint32_t len);

  // return number of bytes read
  uint32_t readEnd() {
    uint32_t bytes = static_cast<uint32_t>(rBase_ - buffer_);
    if (rBase_ == wBase_) {
      resetBuffer();
    }
    return bytes;
  }

  // Return number of bytes written
  uint32_t writeEnd() {
    return static_cast<uint32_t>(wBase_ - buffer_);
  }

  uint32_t available_read() const {
    // Remember, wBase_ is the real rBound_.
    return static_cast<uint32_t>(wBase_ - rBase_);
  }

  uint32_t available_write() const {
    return static_cast<uint32_t>(wBound_ - wBase_);
  }

  // Returns a pointer to where the client can write data to append to
  // the TMemoryBuffer, and ensures the buffer is big enough to accomodate a
  // write of the provided length.  The returned pointer is very convenient for
  // passing to read(), recv(), or similar. You must call wroteBytes() as soon
  // as data is written or the buffer will not be aware that data has changed.
  uint8_t* getWritePtr(uint32_t len) {
    ensureCanWrite(len);
    return wBase_;
  }

  // Informs the buffer that the client has written 'len' bytes into storage
  // that had been provided by getWritePtr().
  void wroteBytes(uint32_t len);

  /*
   * TVirtualTransport provides a default implementation of readAll().
   * We want to use the TBufferBase version instead.
   */
  uint32_t readAll(uint8_t* buf, uint32_t len) {
    return TBufferBase::readAll(buf,len);
  }

 protected:
  void swap(TMemoryBuffer& that) {
    using std::swap;
    swap(buffer_,     that.buffer_);
    swap(bufferSize_, that.bufferSize_);

    swap(rBase_,      that.rBase_);
    swap(rBound_,     that.rBound_);
    swap(wBase_,      that.wBase_);
    swap(wBound_,     that.wBound_);

    swap(owner_,      that.owner_);
  }

  // Make sure there's at least 'len' bytes available for writing.
  int ensureCanWrite(uint32_t len);

  // Compute the position and available data for reading.
  void computeRead(uint32_t len, uint8_t** out_start, uint32_t* out_give);

  uint32_t readSlow(uint8_t* buf, uint32_t len);

  int writeSlow(const uint8_t* buf, uint32_t len);

  const uint8_t* borrowSlow(uint8_t* buf, uint32_t* len);

  // Data buffer
  uint8_t* buffer_;

  // Allocated buffer size
  uint32_t bufferSize_;

  // Is this object the owner of the buffer?
  bool owner_;

  // Don't forget to update constrctors, initCommon, and swap if
  // you add new members.
};

}}} // contrail::sandesh::transport

#endif // #ifndef _SANDESH_TRANSPORT_TBUFFERTRANSPORTS_H_