services/vr/bufferhubd/producer_queue_channel.cpp

   1 #include "producer_queue_channel.h"
   2
   3 #include <inttypes.h>
   4
   5 #include "consumer_queue_channel.h"
   6 #include "producer_channel.h"
   7
   8 using android::pdx::ErrorStatus;
   9 using android::pdx::Message;
  10 using android::pdx::RemoteChannelHandle;
  11 using android::pdx::Status;
  12 using android::pdx::rpc::DispatchRemoteMethod;
  13
  14 namespace android {
  15 namespace dvr {
  16
  17 ProducerQueueChannel::ProducerQueueChannel(BufferHubService* service,
  18                                            int channel_id,
  19                                            const ProducerQueueConfig& config,
  20                                            const UsagePolicy& usage_policy,
  21                                            int* error)
  22     : BufferHubChannel(service, channel_id, channel_id, kProducerQueueType),
  23       config_(config),
  24       usage_policy_(usage_policy),
  25       capacity_(0) {
  26   *error = 0;
  27 }
  28
  29 ProducerQueueChannel::~ProducerQueueChannel() {
  30   ALOGD_IF(TRACE, "ProducerQueueChannel::~ProducerQueueChannel: queue_id=%d",
  31            buffer_id());
  32   for (auto* consumer : consumer_channels_)
  33     consumer->OnProducerClosed();
  34 }
  35
  36 /* static */
  37 Status<std::shared_ptr<ProducerQueueChannel>> ProducerQueueChannel::Create(
  38     BufferHubService* service, int channel_id,
  39     const ProducerQueueConfig& config, const UsagePolicy& usage_policy) {
  40   // Configuration between |usage_deny_set_mask| and |usage_deny_clear_mask|
  41   // should be mutually exclusive.
  42   if ((usage_policy.usage_deny_set_mask & usage_policy.usage_deny_clear_mask)) {
  43     ALOGE(
  44         "BufferHubService::OnCreateProducerQueue: illegal usage mask "
  45         "configuration: usage_deny_set_mask=%" PRIx64
  46         " usage_deny_clear_mask=%" PRIx64,
  47         usage_policy.usage_deny_set_mask, usage_policy.usage_deny_clear_mask);
  48     return ErrorStatus(EINVAL);
  49   }
  50
  51   int error = 0;
  52   std::shared_ptr<ProducerQueueChannel> producer(new ProducerQueueChannel(
  53       service, channel_id, config, usage_policy, &error));
  54   if (error < 0)
  55     return ErrorStatus(-error);
  56   else
  57     return {std::move(producer)};
  58 }
  59
  60 bool ProducerQueueChannel::HandleMessage(Message& message) {
  61   ATRACE_NAME("ProducerQueueChannel::HandleMessage");
  62   switch (message.GetOp()) {
  63     case BufferHubRPC::CreateConsumerQueue::Opcode:
  64       DispatchRemoteMethod<BufferHubRPC::CreateConsumerQueue>(
  65           *this, &ProducerQueueChannel::OnCreateConsumerQueue, message);
  66       return true;
  67
  68     case BufferHubRPC::GetQueueInfo::Opcode:
  69       DispatchRemoteMethod<BufferHubRPC::GetQueueInfo>(
  70           *this, &ProducerQueueChannel::OnGetQueueInfo, message);
  71       return true;
  72
  73     case BufferHubRPC::ProducerQueueAllocateBuffers::Opcode:
  74       DispatchRemoteMethod<BufferHubRPC::ProducerQueueAllocateBuffers>(
  75           *this, &ProducerQueueChannel::OnProducerQueueAllocateBuffers,
  76           message);
  77       return true;
  78
  79     case BufferHubRPC::ProducerQueueRemoveBuffer::Opcode:
  80       DispatchRemoteMethod<BufferHubRPC::ProducerQueueRemoveBuffer>(
  81           *this, &ProducerQueueChannel::OnProducerQueueRemoveBuffer, message);
  82       return true;
  83
  84     default:
  85       return false;
  86   }
  87 }
  88
  89 void ProducerQueueChannel::HandleImpulse(Message& /* message */) {
  90   ATRACE_NAME("ProducerQueueChannel::HandleImpulse");
  91 }
  92
  93 BufferHubChannel::BufferInfo ProducerQueueChannel::GetBufferInfo() const {
  94   return BufferInfo(channel_id(), consumer_channels_.size(), capacity_,
  95                     usage_policy_);
  96 }
  97
  98 Status<RemoteChannelHandle> ProducerQueueChannel::OnCreateConsumerQueue(
  99     Message& message, bool silent) {
 100   ATRACE_NAME("ProducerQueueChannel::OnCreateConsumerQueue");
 101   ALOGD_IF(
 102       TRACE,
 103       "ProducerQueueChannel::OnCreateConsumerQueue: channel_id=%d slient=%d",
 104       channel_id(), silent);
 105
 106   int channel_id;
 107   auto status = message.PushChannel(0, nullptr, &channel_id);
 108   if (!status) {
 109     ALOGE(
 110         "ProducerQueueChannel::OnCreateConsumerQueue: failed to push consumer "
 111         "channel: %s",
 112         status.GetErrorMessage().c_str());
 113     return ErrorStatus(ENOMEM);
 114   }
 115
 116   auto consumer_queue_channel = std::make_shared<ConsumerQueueChannel>(
 117       service(), buffer_id(), channel_id, shared_from_this(), silent);
 118
 119   // Register the existing buffers with the new consumer queue.
 120   for (size_t slot = 0; slot < BufferHubRPC::kMaxQueueCapacity; slot++) {
 121     if (auto buffer = buffers_[slot].lock())
 122       consumer_queue_channel->RegisterNewBuffer(buffer, slot);
 123   }
 124
 125   const auto channel_status =
 126       service()->SetChannel(channel_id, consumer_queue_channel);
 127   if (!channel_status) {
 128     ALOGE(
 129         "ProducerQueueChannel::OnCreateConsumerQueue: Failed to set channel: "
 130         "%s",
 131         channel_status.GetErrorMessage().c_str());
 132     return ErrorStatus(ENOMEM);
 133   }
 134
 135   return {status.take()};
 136 }
 137
 138 Status<QueueInfo> ProducerQueueChannel::OnGetQueueInfo(Message&) {
 139   return {{config_, buffer_id()}};
 140 }
 141
 142 Status<std::vector<std::pair<RemoteChannelHandle, size_t>>>
 143 ProducerQueueChannel::OnProducerQueueAllocateBuffers(
 144     Message& message, uint32_t width, uint32_t height, uint32_t layer_count,
 145     uint32_t format, uint64_t usage, size_t buffer_count) {
 146   ATRACE_NAME("ProducerQueueChannel::OnProducerQueueAllocateBuffers");
 147   ALOGD_IF(TRACE,
 148            "ProducerQueueChannel::OnProducerQueueAllocateBuffers: "
 149            "producer_channel_id=%d",
 150            channel_id());
 151
 152   std::vector<std::pair<RemoteChannelHandle, size_t>> buffer_handles;
 153
 154   // Deny buffer allocation violating preset rules.
 155   if (usage & usage_policy_.usage_deny_set_mask) {
 156     ALOGE(
 157         "ProducerQueueChannel::OnProducerQueueAllocateBuffers: usage: %" PRIx64
 158         " is not permitted. Violating usage_deny_set_mask, the following  bits "
 159         "shall not be set: %" PRIx64 ".",
 160         usage, usage_policy_.usage_deny_set_mask);
 161     return ErrorStatus(EINVAL);
 162   }
 163
 164   if (~usage & usage_policy_.usage_deny_clear_mask) {
 165     ALOGE(
 166         "ProducerQueueChannel::OnProducerQueueAllocateBuffers: usage: %" PRIx64
 167         " is not permitted. Violating usage_deny_clear_mask, the following "
 168         "bits must be set: %" PRIx64 ".",
 169         usage, usage_policy_.usage_deny_clear_mask);
 170     return ErrorStatus(EINVAL);
 171   }
 172
 173   // Force set mask and clear mask. Note that |usage_policy_.usage_set_mask_|
 174   // takes precedence and will overwrite |usage_policy_.usage_clear_mask|.
 175   uint64_t effective_usage =
 176       (usage & ~usage_policy_.usage_clear_mask) | usage_policy_.usage_set_mask;
 177
 178   for (size_t i = 0; i < buffer_count; i++) {
 179     auto status = AllocateBuffer(message, width, height, layer_count, format,
 180                                  effective_usage);
 181     if (!status) {
 182       ALOGE(
 183           "ProducerQueueChannel::OnProducerQueueAllocateBuffers: Failed to "
 184           "allocate new buffer.");
 185       return ErrorStatus(status.error());
 186     }
 187     buffer_handles.push_back(status.take());
 188   }
 189
 190   return {std::move(buffer_handles)};
 191 }
 192
 193 Status<std::pair<RemoteChannelHandle, size_t>>
 194 ProducerQueueChannel::AllocateBuffer(Message& message, uint32_t width,
 195                                      uint32_t height, uint32_t layer_count,
 196                                      uint32_t format, uint64_t usage) {
 197   ATRACE_NAME("ProducerQueueChannel::AllocateBuffer");
 198   ALOGD_IF(TRACE,
 199            "ProducerQueueChannel::AllocateBuffer: producer_channel_id=%d",
 200            channel_id());
 201
 202   if (capacity_ >= BufferHubRPC::kMaxQueueCapacity) {
 203     ALOGE("ProducerQueueChannel::AllocateBuffer: reaches kMaxQueueCapacity.");
 204     return ErrorStatus(E2BIG);
 205   }
 206
 207   // Here we are creating a new BufferHubBuffer, initialize the producer
 208   // channel, and returning its file handle back to the client.
 209   // buffer_id is the id of the producer channel of BufferHubBuffer.
 210   int buffer_id;
 211   auto status = message.PushChannel(0, nullptr, &buffer_id);
 212
 213   if (!status) {
 214     ALOGE("ProducerQueueChannel::AllocateBuffer: failed to push channel: %s",
 215           status.GetErrorMessage().c_str());
 216     return ErrorStatus(status.error());
 217   }
 218
 219   ALOGD_IF(TRACE,
 220            "ProducerQueueChannel::AllocateBuffer: buffer_id=%d width=%u "
 221            "height=%u layer_count=%u format=%u usage=%" PRIx64,
 222            buffer_id, width, height, layer_count, format, usage);
 223   auto buffer_handle = status.take();
 224
 225   auto producer_channel_status =
 226       ProducerChannel::Create(service(), buffer_id, width, height, layer_count,
 227                               format, usage, config_.user_metadata_size);
 228   if (!producer_channel_status) {
 229     ALOGE(
 230         "ProducerQueueChannel::AllocateBuffer: Failed to create producer "
 231         "buffer: %s",
 232         producer_channel_status.GetErrorMessage().c_str());
 233     return ErrorStatus(ENOMEM);
 234   }
 235   auto producer_channel = producer_channel_status.take();
 236
 237   ALOGD_IF(
 238       TRACE,
 239       "ProducerQueueChannel::AllocateBuffer: buffer_id=%d, buffer_handle=%d",
 240       buffer_id, buffer_handle.value());
 241
 242   const auto channel_status =
 243       service()->SetChannel(buffer_id, producer_channel);
 244   if (!channel_status) {
 245     ALOGE(
 246         "ProducerQueueChannel::AllocateBuffer: failed to set producer channel "
 247         "for new BufferHubBuffer: %s",
 248         channel_status.GetErrorMessage().c_str());
 249     return ErrorStatus(ENOMEM);
 250   }
 251
 252   // Register the newly allocated buffer's channel_id into the first empty
 253   // buffer slot.
 254   size_t slot = 0;
 255   for (; slot < BufferHubRPC::kMaxQueueCapacity; slot++) {
 256     if (buffers_[slot].expired())
 257       break;
 258   }
 259   if (slot == BufferHubRPC::kMaxQueueCapacity) {
 260     ALOGE(
 261         "ProducerQueueChannel::AllocateBuffer: Cannot find empty slot for new "
 262         "buffer allocation.");
 263     return ErrorStatus(E2BIG);
 264   }
 265
 266   buffers_[slot] = producer_channel;
 267   capacity_++;
 268
 269   // Notify each consumer channel about the new buffer.
 270   for (auto* consumer_channel : consumer_channels_) {
 271     ALOGD(
 272         "ProducerQueueChannel::AllocateBuffer: Notified consumer with new "
 273         "buffer, buffer_id=%d",
 274         buffer_id);
 275     consumer_channel->RegisterNewBuffer(producer_channel, slot);
 276   }
 277
 278   return {{std::move(buffer_handle), slot}};
 279 }
 280
 281 Status<void> ProducerQueueChannel::OnProducerQueueRemoveBuffer(
 282     Message& /*message*/, size_t slot) {
 283   if (buffers_[slot].expired()) {
 284     ALOGE(
 285         "ProducerQueueChannel::OnProducerQueueRemoveBuffer: trying to remove "
 286         "an invalid buffer producer at slot %zu",
 287         slot);
 288     return ErrorStatus(EINVAL);
 289   }
 290
 291   if (capacity_ == 0) {
 292     ALOGE(
 293         "ProducerQueueChannel::OnProducerQueueRemoveBuffer: trying to remove a "
 294         "buffer producer while the queue's capacity is already zero.");
 295     return ErrorStatus(EINVAL);
 296   }
 297
 298   buffers_[slot].reset();
 299   capacity_--;
 300   return {};
 301 }
 302
 303 void ProducerQueueChannel::AddConsumer(ConsumerQueueChannel* channel) {
 304   consumer_channels_.push_back(channel);
 305 }
 306
 307 void ProducerQueueChannel::RemoveConsumer(ConsumerQueueChannel* channel) {
 308   consumer_channels_.erase(
 309       std::find(consumer_channels_.begin(), consumer_channels_.end(), channel));
 310 }
 311
 312 }  // namespace dvr
 313 }  // namespace android