libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp

#include "include/private/dvr/buffer_hub_queue_client.h"

//#define LOG_NDEBUG 0

#include <inttypes.h>
#include <log/log.h>
#include <sys/epoll.h>

#include <array>

#include <pdx/default_transport/client_channel.h>
#include <pdx/default_transport/client_channel_factory.h>
#include <pdx/file_handle.h>
#include <private/dvr/bufferhub_rpc.h>

namespace android {
namespace dvr {

BufferHubQueue::BufferHubQueue(LocalChannelHandle channel_handle,
                               size_t meta_size)
    : Client{pdx::default_transport::ClientChannel::Create(
          std::move(channel_handle))},
      meta_size_(meta_size),
      meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
      buffers_(BufferHubQueue::kMaxQueueCapacity),
      epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
      available_buffers_(BufferHubQueue::kMaxQueueCapacity),
      fences_(BufferHubQueue::kMaxQueueCapacity),
      capacity_(0) {
  Initialize();
}

BufferHubQueue::BufferHubQueue(const std::string& endpoint_path,
                               size_t meta_size)
    : Client{pdx::default_transport::ClientChannelFactory::Create(
          endpoint_path)},
      meta_size_(meta_size),
      meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
      buffers_(BufferHubQueue::kMaxQueueCapacity),
      epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
      available_buffers_(BufferHubQueue::kMaxQueueCapacity),
      fences_(BufferHubQueue::kMaxQueueCapacity),
      capacity_(0) {
  Initialize();
}

void BufferHubQueue::Initialize() {
  int ret = epoll_fd_.Create();
  if (ret < 0) {
    ALOGE("BufferHubQueue::BufferHubQueue: Failed to create epoll fd: %s",
          strerror(-ret));
    return;
  }

  epoll_event event = {.events = EPOLLIN | EPOLLET,
                       .data = {.u64 = static_cast<uint64_t>(
                                    BufferHubQueue::kEpollQueueEventIndex)}};
  ret = epoll_fd_.Control(EPOLL_CTL_ADD, event_fd(), &event);
  if (ret < 0) {
    ALOGE("Failed to register ConsumerQueue into epoll event: %s",
          strerror(-ret));
  }
}

std::unique_ptr<ConsumerQueue> BufferHubQueue::CreateConsumerQueue() {
  Status<std::pair<LocalChannelHandle, size_t>> status =
      InvokeRemoteMethod<BufferHubRPC::CreateConsumerQueue>();

  if (!status) {
    ALOGE("Cannot create ConsumerQueue: %s", status.GetErrorMessage().c_str());
    return nullptr;
  }

  auto return_value = status.take();

  ALOGD("CreateConsumerQueue: meta_size_bytes=%zu", return_value.second);
  return ConsumerQueue::Create(std::move(return_value.first),
                               return_value.second);
}

bool BufferHubQueue::WaitForBuffers(int timeout) {
  std::array<epoll_event, kMaxEvents> events;

  while (count() == 0) {
    int ret = epoll_fd_.Wait(events.data(), events.size(), timeout);

    if (ret == 0) {
      ALOGD("Wait on epoll returns nothing before timeout.");
      return false;
    }

    if (ret < 0 && ret != -EINTR) {
      ALOGE("Failed to wait for buffers: %s", strerror(-ret));
      return false;
    }

    const int num_events = ret;

    // A BufferQueue's epoll fd tracks N+1 events, where there are N events,
    // one for each buffer, in the queue and one extra event for the queue
    // client itself.
    for (int i = 0; i < num_events; i++) {
      int64_t index = static_cast<int64_t>(events[i].data.u64);

      ALOGD("New BufferHubQueue event %d: index=%" PRId64, i, index);

      if (is_buffer_event_index(index)) {
        HandleBufferEvent(static_cast<size_t>(index), events[i]);
      } else if (is_queue_event_index(index)) {
        HandleQueueEvent(events[i]);
      } else {
        ALOGW("Unknown event index: %" PRId64, index);
      }
    }
  }

  return true;
}

void BufferHubQueue::HandleBufferEvent(size_t slot, const epoll_event& event) {
  auto buffer = buffers_[slot];
  if (!buffer) {
    ALOGW("BufferHubQueue::HandleBufferEvent: Invalid buffer slot: %zu", slot);
    return;
  }

  auto status = buffer->GetEventMask(event.events);
  if (!status) {
    ALOGW("BufferHubQueue::HandleBufferEvent: Failed to get event mask: %s",
          status.GetErrorMessage().c_str());
    return;
  }

  int events = status.get();
  if (events & EPOLLIN) {
    int ret = OnBufferReady(buffer, &fences_[slot]);
    if (ret < 0) {
      ALOGE("Failed to set buffer ready: %s", strerror(-ret));
      return;
    }
    Enqueue(buffer, slot);
  } else if (events & EPOLLHUP) {
    // This might be caused by producer replacing an existing buffer slot, or
    // when BufferHubQueue is shutting down. For the first case, currently the
    // epoll FD is cleaned up when the replacement consumer client is imported,
    // we shouldn't detach again if |epollhub_pending_[slot]| is set.
    ALOGW(
        "Receives EPOLLHUP at slot: %zu, buffer event fd: %d, EPOLLHUP "
        "pending: %d",
        slot, buffer->event_fd(), epollhup_pending_[slot]);
    if (epollhup_pending_[slot]) {
      epollhup_pending_[slot] = false;
    } else {
      DetachBuffer(slot);
    }
  } else {
    ALOGW("Unknown event, slot=%zu, epoll events=%d", slot, events);
  }
}

void BufferHubQueue::HandleQueueEvent(const epoll_event& event) {
  auto status = GetEventMask(event.events);
  if (!status) {
    ALOGW("BufferHubQueue::HandleQueueEvent: Failed to get event mask: %s",
          status.GetErrorMessage().c_str());
    return;
  }

  int events = status.get();
  if (events & EPOLLIN) {
    // Note that after buffer imports, if |count()| still returns 0, epoll
    // wait will be tried again to acquire the newly imported buffer.
    int ret = OnBufferAllocated();
    if (ret < 0) {
      ALOGE("Failed to import buffer: %s", strerror(-ret));
    }
  } else {
    ALOGW("Unknown epoll events=%d", events);
  }
}

int BufferHubQueue::AddBuffer(const std::shared_ptr<BufferHubBuffer>& buf,
                              size_t slot) {
  if (is_full()) {
    // TODO(jwcai) Move the check into Producer's AllocateBuffer and consumer's
    // import buffer.
    ALOGE("BufferHubQueue::AddBuffer queue is at maximum capacity: %zu",
          capacity_);
    return -E2BIG;
  }

  if (buffers_[slot] != nullptr) {
    // Replace the buffer if the slot is preoccupied. This could happen when the
    // producer side replaced the slot with a newly allocated buffer. Detach the
    // buffer before setting up with the new one.
    DetachBuffer(slot);
    epollhup_pending_[slot] = true;
  }

  epoll_event event = {.events = EPOLLIN | EPOLLET, .data = {.u64 = slot}};
  const int ret = epoll_fd_.Control(EPOLL_CTL_ADD, buf->event_fd(), &event);
  if (ret < 0) {
    ALOGE("BufferHubQueue::AddBuffer: Failed to add buffer to epoll set: %s",
          strerror(-ret));
    return ret;
  }

  buffers_[slot] = buf;
  capacity_++;
  return 0;
}

int BufferHubQueue::DetachBuffer(size_t slot) {
  auto& buf = buffers_[slot];
  if (buf == nullptr) {
    ALOGE("BufferHubQueue::DetachBuffer: Invalid slot: %zu", slot);
    return -EINVAL;
  }

  const int ret = epoll_fd_.Control(EPOLL_CTL_DEL, buf->event_fd(), nullptr);
  if (ret < 0) {
    ALOGE(
        "BufferHubQueue::DetachBuffer: Failed to detach buffer from epoll set: "
        "%s",
        strerror(-ret));
    return ret;
  }

  buffers_[slot] = nullptr;
  capacity_--;
  return 0;
}

void BufferHubQueue::Enqueue(std::shared_ptr<BufferHubBuffer> buf,
                             size_t slot) {
  if (count() == capacity_) {
    ALOGE("Buffer queue is full!");
    return;
  }

  // Set slot buffer back to vector.
  // TODO(jwcai) Here have to dynamically allocate BufferInfo::metadata due to
  // the limitation of the RingBuffer we are using. Would be better to refactor
  // that.
  BufferInfo buffer_info(slot, meta_size_);
  // Swap buffer into vector.
  std::swap(buffer_info.buffer, buf);
  // Swap metadata loaded during onBufferReady into vector.
  std::swap(buffer_info.metadata, meta_buffer_tmp_);

  available_buffers_.Append(std::move(buffer_info));
}

std::shared_ptr<BufferHubBuffer> BufferHubQueue::Dequeue(int timeout,
                                                         size_t* slot,
                                                         void* meta,
                                                         LocalHandle* fence) {
  ALOGD("Dequeue: count=%zu, timeout=%d", count(), timeout);

  if (count() == 0 && !WaitForBuffers(timeout))
    return nullptr;

  std::shared_ptr<BufferHubBuffer> buf;
  BufferInfo& buffer_info = available_buffers_.Front();

  *fence = std::move(fences_[buffer_info.slot]);

  // Report current pos as the output slot.
  std::swap(buffer_info.slot, *slot);
  // Swap buffer from vector to be returned later.
  std::swap(buffer_info.buffer, buf);
  // Swap metadata from vector into tmp so that we can write out to |meta|.
  std::swap(buffer_info.metadata, meta_buffer_tmp_);

  available_buffers_.PopFront();

  if (!buf) {
    ALOGE("Dequeue: Buffer to be dequeued is nullptr");
    return nullptr;
  }

  if (meta) {
    std::copy(meta_buffer_tmp_.get(), meta_buffer_tmp_.get() + meta_size_,
              reinterpret_cast<uint8_t*>(meta));
  }

  return buf;
}

ProducerQueue::ProducerQueue(size_t meta_size)
    : ProducerQueue(meta_size, 0, 0, 0, 0) {}

ProducerQueue::ProducerQueue(LocalChannelHandle handle, size_t meta_size)
    : BASE(std::move(handle), meta_size) {}

ProducerQueue::ProducerQueue(size_t meta_size, int usage_set_mask,
                             int usage_clear_mask, int usage_deny_set_mask,
                             int usage_deny_clear_mask)
    : BASE(BufferHubRPC::kClientPath, meta_size) {
  auto status = InvokeRemoteMethod<BufferHubRPC::CreateProducerQueue>(
      meta_size_, usage_set_mask, usage_clear_mask, usage_deny_set_mask,
      usage_deny_clear_mask);
  if (!status) {
    ALOGE("ProducerQueue::ProducerQueue: Failed to create producer queue: %s",
          status.GetErrorMessage().c_str());
    Close(-status.error());
    return;
  }
}

int ProducerQueue::AllocateBuffer(int width, int height, int format, int usage,
                                  size_t slice_count, size_t* out_slot) {
  if (out_slot == nullptr) {
    ALOGE("Parameter out_slot cannot be null.");
    return -EINVAL;
  }

  if (is_full()) {
    ALOGE("ProducerQueue::AllocateBuffer queue is at maximum capacity: %zu",
          capacity());
    return -E2BIG;
  }

  const size_t kBufferCount = 1U;

  Status<std::vector<std::pair<LocalChannelHandle, size_t>>> status =
      InvokeRemoteMethod<BufferHubRPC::ProducerQueueAllocateBuffers>(
          width, height, format, usage, slice_count, kBufferCount);
  if (!status) {
    ALOGE(
        "ProducerQueue::AllocateBuffer failed to create producer buffer "
        "through BufferHub.");
    return -status.error();
  }

  auto buffer_handle_slots = status.take();
  LOG_ALWAYS_FATAL_IF(buffer_handle_slots.size() != kBufferCount,
                      "BufferHubRPC::ProducerQueueAllocateBuffers should "
                      "return one and only one buffer handle.");

  // We only allocate one buffer at a time.
  auto& buffer_handle = buffer_handle_slots[0].first;
  size_t buffer_slot = buffer_handle_slots[0].second;
  ALOGD("ProducerQueue::AllocateBuffer, new buffer, channel_handle: %d",
        buffer_handle.value());

  *out_slot = buffer_slot;
  return AddBuffer(BufferProducer::Import(std::move(buffer_handle)),
                   buffer_slot);
}

int ProducerQueue::AddBuffer(const std::shared_ptr<BufferProducer>& buf,
                             size_t slot) {
  // For producer buffer, we need to enqueue the newly added buffer
  // immediately. Producer queue starts with all buffers in available state.
  const int ret = BufferHubQueue::AddBuffer(buf, slot);
  if (ret < 0)
    return ret;

  Enqueue(buf, slot);
  return 0;
}

int ProducerQueue::DetachBuffer(size_t slot) {
  Status<int> status =
      InvokeRemoteMethod<BufferHubRPC::ProducerQueueDetachBuffer>(slot);
  if (!status) {
    ALOGE(
        "ProducerQueue::DetachBuffer failed to detach producer buffer through "
        "BufferHub, error: %s",
        status.GetErrorMessage().c_str());
    return -status.error();
  }

  return BufferHubQueue::DetachBuffer(slot);
}

std::shared_ptr<BufferProducer> ProducerQueue::Dequeue(
    int timeout, size_t* slot, LocalHandle* release_fence) {
  if (slot == nullptr || release_fence == nullptr) {
    ALOGE("invalid parameter, slot=%p, release_fence=%p", slot, release_fence);
    return nullptr;
  }

  auto buf = BufferHubQueue::Dequeue(timeout, slot, nullptr, release_fence);
  return std::static_pointer_cast<BufferProducer>(buf);
}

int ProducerQueue::OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
                                 LocalHandle* release_fence) {
  auto buffer = std::static_pointer_cast<BufferProducer>(buf);
  return buffer->Gain(release_fence);
}

ConsumerQueue::ConsumerQueue(LocalChannelHandle handle, size_t meta_size)
    : BASE(std::move(handle), meta_size) {
  // TODO(b/34387835) Import consumer queue in case the ProducerQueue we are
  // based on was not empty.
}

int ConsumerQueue::ImportBuffers() {
  Status<std::vector<std::pair<LocalChannelHandle, size_t>>> status =
      InvokeRemoteMethod<BufferHubRPC::ConsumerQueueImportBuffers>();
  if (!status) {
    ALOGE(
        "ConsumerQueue::ImportBuffers failed to import consumer buffer through "
        "BufferBub, error: %s",
        status.GetErrorMessage().c_str());
    return -status.error();
  }

  int last_error = 0;
  int imported_buffers = 0;

  auto buffer_handle_slots = status.take();
  for (auto& buffer_handle_slot : buffer_handle_slots) {
    ALOGD("ConsumerQueue::ImportBuffers, new buffer, buffer_handle: %d",
          buffer_handle_slot.first.value());

    std::unique_ptr<BufferConsumer> buffer_consumer =
        BufferConsumer::Import(std::move(buffer_handle_slot.first));
    int ret = AddBuffer(std::move(buffer_consumer), buffer_handle_slot.second);
    if (ret < 0) {
      ALOGE("ConsumerQueue::ImportBuffers failed to add buffer, ret: %s",
            strerror(-ret));
      last_error = ret;
      continue;
    } else {
      imported_buffers++;
    }
  }

  return imported_buffers > 0 ? imported_buffers : last_error;
}

int ConsumerQueue::AddBuffer(const std::shared_ptr<BufferConsumer>& buf,
                             size_t slot) {
  // Consumer queue starts with all buffers in unavailable state.
  return BufferHubQueue::AddBuffer(buf, slot);
}

std::shared_ptr<BufferConsumer> ConsumerQueue::Dequeue(
    int timeout, size_t* slot, void* meta, size_t meta_size,
    LocalHandle* acquire_fence) {
  if (meta_size != meta_size_) {
    ALOGE(
        "metadata size (%zu) for the dequeuing buffer does not match metadata "
        "size (%zu) for the queue.",
        meta_size, meta_size_);
    return nullptr;
  }

  if (slot == nullptr || meta == nullptr || acquire_fence == nullptr) {
    ALOGE("invalid parameter, slot=%p, meta=%p, acquire_fence=%p", slot, meta,
          acquire_fence);
    return nullptr;
  }

  auto buf = BufferHubQueue::Dequeue(timeout, slot, meta, acquire_fence);
  return std::static_pointer_cast<BufferConsumer>(buf);
}

int ConsumerQueue::OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
                                 LocalHandle* acquire_fence) {
  auto buffer = std::static_pointer_cast<BufferConsumer>(buf);
  return buffer->Acquire(acquire_fence, meta_buffer_tmp_.get(), meta_size_);
}

int ConsumerQueue::OnBufferAllocated() {
  const int ret = ImportBuffers();
  if (ret == 0) {
    ALOGW("No new buffer can be imported on buffer allocated event.");
  } else if (ret < 0) {
    ALOGE("Failed to import buffers on buffer allocated event.");
  }
  ALOGD("Imported %d consumer buffers.", ret);
  return ret;
}

}  // namespace dvr
}  // namespace android