common/vsoc/lib/circqueue_impl.h

#pragma once

/*
 * Copyright (C) 2017 The Android Open Source Project
 *
 * Licensed 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 <cerrno>
#include <cstring>

#include <sys/uio.h>

#include "common/vsoc/lib/region_signaling_interface.h"
#include "common/vsoc/shm/circqueue.h"

// Increases the given index until it is naturally aligned for T.
template <typename T>
uintptr_t align(uintptr_t index) {
  return (index + sizeof(T) - 1) & ~(sizeof(T) - 1);
}

namespace vsoc {
class RegionSignalingInterface;
namespace layout {

template <uint32_t SizeLog2>
void CircularQueueBase<SizeLog2>::CopyInRange(const char* buffer_in,
                                              const Range& t) {
  size_t bytes = t.end_idx - t.start_idx;
  uint32_t index = t.start_idx & (BufferSize - 1);
  if (index + bytes <= BufferSize) {
    std::memcpy(buffer_ + index, buffer_in, bytes);
  } else {
    size_t part1_size = BufferSize - index;
    size_t part2_size = bytes - part1_size;
    std::memcpy(buffer_ + index, buffer_in, part1_size);
    std::memcpy(buffer_, buffer_in + part1_size, part2_size);
  }
}

template <uint32_t SizeLog2>
void CircularQueueBase<SizeLog2>::CopyOutRange(const Range& t,
                                               char* buffer_out) {
  uint32_t index = t.start_idx & (BufferSize - 1);
  size_t total_size = t.end_idx - t.start_idx;
  if (index + total_size <= BufferSize) {
    std::memcpy(buffer_out, buffer_ + index, total_size);
  } else {
    uint32_t part1_size = BufferSize - index;
    uint32_t part2_size = total_size - part1_size;
    std::memcpy(buffer_out, buffer_ + index, part1_size);
    std::memcpy(buffer_out + part1_size, buffer_, part2_size);
  }
}

template <uint32_t SizeLog2>
void CircularQueueBase<SizeLog2>::WaitForDataLocked(
    RegionSignalingInterface* r) {
  while (1) {
    uint32_t o_w_pub = w_pub_;
    // We don't have data. Wait until some appears and try again
    if (r_released_ != o_w_pub) {
      return;
    }
    lock_.Unlock();
    r->WaitForSignal(&w_pub_, o_w_pub);
    lock_.Lock();
  }
}

template <uint32_t SizeLog2>
intptr_t CircularQueueBase<SizeLog2>::WriteReserveLocked(
    RegionSignalingInterface* r, size_t bytes, Range* t, bool non_blocking) {
  // Can't write more than the buffer will hold
  if (bytes > BufferSize) {
    return -ENOSPC;
  }
  while (true) {
    uint32_t o_w_pub = w_pub_;
    uint32_t o_r_release = r_released_;
    uint32_t bytes_in_use = o_w_pub - o_r_release;
    size_t available = BufferSize - bytes_in_use;
    if (available >= bytes) {
      t->start_idx = o_w_pub;
      t->end_idx = o_w_pub + bytes;
      break;
    }
    if (non_blocking) {
      return -EWOULDBLOCK;
    }
    // If we can't write at the moment wait for a reader to release
    // some bytes.
    lock_.Unlock();
    r->WaitForSignal(&r_released_, o_r_release);
    lock_.Lock();
  }
  return t->end_idx - t->start_idx;
}

template <uint32_t SizeLog2>
intptr_t CircularByteQueue<SizeLog2>::Read(RegionSignalingInterface* r,
                                           char* buffer_out, size_t max_size) {
  this->lock_.Lock();
  this->WaitForDataLocked(r);
  Range t;
  t.start_idx = this->r_released_;
  t.end_idx = this->w_pub_;
  // The lock is still held here...
  // Trim the range if we got more than the reader wanted
  if ((t.end_idx - t.start_idx) > max_size) {
    t.end_idx = t.start_idx + max_size;
  }
  this->CopyOutRange(t, buffer_out);
  this->r_released_ = t.end_idx;
  this->lock_.Unlock();
  r->SendSignal(layout::Sides::Both, &this->r_released_);
  return t.end_idx - t.start_idx;
}

template <uint32_t SizeLog2>
intptr_t CircularByteQueue<SizeLog2>::Write(RegionSignalingInterface* r,
                                            const char* buffer_in, size_t bytes,
                                            bool non_blocking) {
  Range range;
  this->lock_.Lock();
  intptr_t rval = this->WriteReserveLocked(r, bytes, &range, non_blocking);
  if (rval < 0) {
    this->lock_.Unlock();
    return rval;
  }
  this->CopyInRange(buffer_in, range);
  // We can't publish until all of the previous write allocations where
  // published.
  this->w_pub_ = range.end_idx;
  this->lock_.Unlock();
  r->SendSignal(layout::Sides::Both, &this->w_pub_);
  return bytes;
}

template <uint32_t SizeLog2, uint32_t MaxPacketSize>
intptr_t CircularPacketQueue<SizeLog2, MaxPacketSize>::CalculateBufferedSize(
    size_t payload) {
  return align<uint32_t>(sizeof(uint32_t) + payload);
}

template <uint32_t SizeLog2, uint32_t MaxPacketSize>
intptr_t CircularPacketQueue<SizeLog2, MaxPacketSize>::Read(
    RegionSignalingInterface* r, char* buffer_out, size_t max_size) {
  this->lock_.Lock();
  this->WaitForDataLocked(r);
  uint32_t packet_size = *reinterpret_cast<uint32_t*>(
      this->buffer_ + (this->r_released_ & (this->BufferSize - 1)));
  if (packet_size > max_size) {
    this->lock_.Unlock();
    return -ENOSPC;
  }
  Range t;
  t.start_idx = this->r_released_ + sizeof(uint32_t);
  t.end_idx = t.start_idx + packet_size;
  this->CopyOutRange(t, buffer_out);
  this->r_released_ += this->CalculateBufferedSize(packet_size);
  this->lock_.Unlock();
  r->SendSignal(layout::Sides::Both, &this->r_released_);
  return packet_size;
}

template <uint32_t SizeLog2, uint32_t MaxPacketSize>
intptr_t CircularPacketQueue<SizeLog2, MaxPacketSize>::Write(
    RegionSignalingInterface* r, const char* buffer_in, uint32_t bytes,
    bool non_blocking) {
  iovec iov;
  iov.iov_base = const_cast<char *>(buffer_in);
  iov.iov_len = bytes;
  return Writev(r, &iov, 1 /* iov_count */, non_blocking);
}

template <uint32_t SizeLog2, uint32_t MaxPacketSize>
intptr_t CircularPacketQueue<SizeLog2, MaxPacketSize>::Writev(
      RegionSignalingInterface *r,
      const iovec *iov,
      size_t iov_count,
      bool non_blocking) {
  size_t bytes = 0;
  for (size_t i = 0; i < iov_count; ++i) {
    bytes += iov[i].iov_len;
  }

  if (bytes > MaxPacketSize) {
    return -ENOSPC;
  }

  Range range;
  size_t buffered_size = this->CalculateBufferedSize(bytes);
  this->lock_.Lock();
  intptr_t rval =
      this->WriteReserveLocked(r, buffered_size, &range, non_blocking);
  if (rval < 0) {
    this->lock_.Unlock();
    return rval;
  }
  Range header = range;
  header.end_idx = header.start_idx + sizeof(uint32_t);
  Range payload{
      static_cast<uint32_t>(range.start_idx + sizeof(uint32_t)),
      static_cast<uint32_t>(range.start_idx + sizeof(uint32_t) + bytes)};
  this->CopyInRange(reinterpret_cast<const char*>(&bytes), header);

  Range subRange = payload;
  for (size_t i = 0; i < iov_count; ++i) {
    subRange.end_idx = subRange.start_idx + iov[i].iov_len;
    this->CopyInRange(static_cast<const char *>(iov[i].iov_base), subRange);

    subRange.start_idx = subRange.end_idx;
  }

  this->w_pub_ = range.end_idx;
  this->lock_.Unlock();
  r->SendSignal(layout::Sides::Both, &this->w_pub_);
  return bytes;
}

}  // namespace layout
}  // namespace vsoc