producer_channel.cpp revision f0a7bd033941e26e380232a0515e903cf8e678e5 
1#include "producer_channel.h"
2
3#include <log/log.h>
4#include <sync/sync.h>
5#include <sys/poll.h>
6#include <utils/Trace.h>
7
8#include <algorithm>
9#include <atomic>
10#include <thread>
11
12#include <private/dvr/bufferhub_rpc.h>
13#include "consumer_channel.h"
14
15using android::pdx::BorrowedHandle;
16using android::pdx::Message;
17using android::pdx::RemoteChannelHandle;
18using android::pdx::rpc::BufferWrapper;
19using android::pdx::rpc::DispatchRemoteMethod;
20using android::pdx::rpc::WrapBuffer;
21
22namespace android {
23namespace dvr {
24
25ProducerChannel::ProducerChannel(BufferHubService* service, int channel_id,
26                                 int width, int height, int format, int usage,
27                                 size_t meta_size_bytes, size_t slice_count,
28                                 int* error)
29    : BufferHubChannel(service, channel_id, channel_id, kProducerType),
30      pending_consumers_(0),
31      slices_(std::max(static_cast<size_t>(1), slice_count)),
32      producer_owns_(true),
33      meta_size_bytes_(meta_size_bytes),
34      meta_(meta_size_bytes ? new uint8_t[meta_size_bytes] : nullptr) {
35  for (auto& ion_buffer : slices_) {
36    const int ret = ion_buffer.Alloc(width, height, format, usage);
37    if (ret < 0) {
38      ALOGE("ProducerChannel::ProducerChannel: Failed to allocate buffer: %s",
39            strerror(-ret));
40      *error = ret;
41      return;
42    }
43  }
44
45  // Success.
46  *error = 0;
47}
48
49std::shared_ptr<ProducerChannel> ProducerChannel::Create(
50    BufferHubService* service, int channel_id, int width, int height,
51    int format, int usage, size_t meta_size_bytes, size_t slice_count,
52    int* error) {
53  std::shared_ptr<ProducerChannel> producer(
54      new ProducerChannel(service, channel_id, width, height, format, usage,
55                          meta_size_bytes, slice_count, error));
56  if (*error < 0)
57    return nullptr;
58  else
59    return producer;
60}
61
62ProducerChannel::~ProducerChannel() {
63  ALOGD_IF(TRACE,
64           "ProducerChannel::~ProducerChannel: channel_id=%d buffer_id=%d",
65           channel_id(), buffer_id());
66  for (auto consumer : consumer_channels_)
67    consumer->OnProducerClosed();
68}
69
70BufferHubChannel::BufferInfo ProducerChannel::GetBufferInfo() const {
71  return BufferInfo(buffer_id(), consumer_channels_.size(), slices_[0].width(),
72                    slices_[0].height(), slices_[0].format(),
73                    slices_[0].usage(), slices_.size(), name_);
74}
75
76void ProducerChannel::HandleImpulse(Message& message) {
77  ATRACE_NAME("ProducerChannel::HandleImpulse");
78  switch (message.GetOp()) {
79    case BufferHubRPC::ProducerGain::Opcode:
80      OnProducerGain(message);
81      break;
82  }
83}
84
85bool ProducerChannel::HandleMessage(Message& message) {
86  ATRACE_NAME("ProducerChannel::HandleMessage");
87  switch (message.GetOp()) {
88    case BufferHubRPC::GetBuffer::Opcode:
89      DispatchRemoteMethod<BufferHubRPC::GetBuffer>(
90          *this, &ProducerChannel::OnGetBuffer, message);
91      return true;
92
93    case BufferHubRPC::GetBuffers::Opcode:
94      DispatchRemoteMethod<BufferHubRPC::GetBuffers>(
95          *this, &ProducerChannel::OnGetBuffers, message);
96      return true;
97
98    case BufferHubRPC::NewConsumer::Opcode:
99      DispatchRemoteMethod<BufferHubRPC::NewConsumer>(
100          *this, &ProducerChannel::OnNewConsumer, message);
101      return true;
102
103    case BufferHubRPC::ProducerPost::Opcode:
104      DispatchRemoteMethod<BufferHubRPC::ProducerPost>(
105          *this, &ProducerChannel::OnProducerPost, message);
106      return true;
107
108    case BufferHubRPC::ProducerGain::Opcode:
109      DispatchRemoteMethod<BufferHubRPC::ProducerGain>(
110          *this, &ProducerChannel::OnProducerGain, message);
111      return true;
112
113    case BufferHubRPC::ProducerMakePersistent::Opcode:
114      DispatchRemoteMethod<BufferHubRPC::ProducerMakePersistent>(
115          *this, &ProducerChannel::OnProducerMakePersistent, message);
116      return true;
117
118    case BufferHubRPC::ProducerRemovePersistence::Opcode:
119      DispatchRemoteMethod<BufferHubRPC::ProducerRemovePersistence>(
120          *this, &ProducerChannel::OnRemovePersistence, message);
121      return true;
122
123    default:
124      return false;
125  }
126}
127
128NativeBufferHandle<BorrowedHandle> ProducerChannel::OnGetBuffer(
129    Message& message, unsigned index) {
130  ATRACE_NAME("ProducerChannel::OnGetBuffer");
131  ALOGD_IF(TRACE, "ProducerChannel::OnGetBuffer: buffer=%d", buffer_id());
132  if (index < slices_.size()) {
133    return NativeBufferHandle<BorrowedHandle>(slices_[index], buffer_id());
134  } else {
135    REPLY_ERROR_RETURN(message, EINVAL, NativeBufferHandle<BorrowedHandle>());
136  }
137}
138
139std::vector<NativeBufferHandle<BorrowedHandle>> ProducerChannel::OnGetBuffers(
140    Message&) {
141  ATRACE_NAME("ProducerChannel::OnGetBuffers");
142  ALOGD_IF(TRACE, "ProducerChannel::OnGetBuffers: buffer_id=%d", buffer_id());
143  std::vector<NativeBufferHandle<BorrowedHandle>> buffer_handles;
144  for (const auto& buffer : slices_)
145    buffer_handles.emplace_back(buffer, buffer_id());
146  return buffer_handles;
147}
148
149RemoteChannelHandle ProducerChannel::CreateConsumer(Message& message) {
150  ATRACE_NAME("ProducerChannel::CreateConsumer");
151  ALOGD_IF(TRACE, "ProducerChannel::CreateConsumer: buffer_id=%d", buffer_id());
152
153  int channel_id;
154  auto status = message.PushChannel(0, nullptr, &channel_id);
155  if (!status) {
156    ALOGE(
157        "ProducerChannel::CreateConsumer: failed to push consumer channel: %s",
158        status.GetErrorMessage().c_str());
159    return RemoteChannelHandle();
160  }
161
162  auto consumer = std::make_shared<ConsumerChannel>(
163      service(), buffer_id(), channel_id, shared_from_this());
164  const auto channel_status = service()->SetChannel(channel_id, consumer);
165  if (!channel_status) {
166    ALOGE(
167        "ProducerChannel::CreateConsumer: failed to set new consumer channel: "
168        "%s",
169        channel_status.GetErrorMessage().c_str());
170    return RemoteChannelHandle();
171  }
172
173  if (!producer_owns_) {
174    // Signal the new consumer when adding it to a posted producer.
175    if (consumer->OnProducerPosted())
176      pending_consumers_++;
177  }
178
179  return status.take();
180}
181
182RemoteChannelHandle ProducerChannel::OnNewConsumer(Message& message) {
183  ATRACE_NAME("ProducerChannel::OnNewConsumer");
184  ALOGD_IF(TRACE, "ProducerChannel::OnNewConsumer: buffer_id=%d", buffer_id());
185
186  RemoteChannelHandle consumer_handle(CreateConsumer(message));
187
188  if (consumer_handle.valid())
189    return consumer_handle;
190  else
191    REPLY_ERROR_RETURN(message, ENOMEM, RemoteChannelHandle());
192}
193
194int ProducerChannel::OnProducerPost(
195    Message&, LocalFence acquire_fence,
196    BufferWrapper<std::vector<std::uint8_t>> metadata) {
197  ATRACE_NAME("ProducerChannel::OnProducerPost");
198  ALOGD_IF(TRACE, "ProducerChannel::OnProducerPost: buffer_id=%d", buffer_id());
199  if (!producer_owns_) {
200    ALOGE("ProducerChannel::OnProducerPost: Not in gained state!");
201    return -EBUSY;
202  }
203
204  if (meta_size_bytes_ != metadata.size())
205    return -EINVAL;
206  std::copy(metadata.begin(), metadata.end(), meta_.get());
207
208  post_fence_ = std::move(acquire_fence);
209  producer_owns_ = false;
210
211  // Signal any interested consumers. If there are none, automatically release
212  // the buffer.
213  pending_consumers_ = 0;
214  for (auto consumer : consumer_channels_) {
215    if (consumer->OnProducerPosted())
216      pending_consumers_++;
217  }
218  if (pending_consumers_ == 0)
219    SignalAvailable();
220  ALOGD_IF(TRACE, "ProducerChannel::OnProducerPost: %d pending consumers",
221           pending_consumers_);
222
223  return 0;
224}
225
226LocalFence ProducerChannel::OnProducerGain(Message& message) {
227  ATRACE_NAME("ProducerChannel::OnGain");
228  ALOGD_IF(TRACE, "ProducerChannel::OnGain: buffer_id=%d", buffer_id());
229  if (producer_owns_) {
230    ALOGE("ProducerChanneL::OnGain: Already in gained state: channel=%d",
231          channel_id());
232    REPLY_ERROR_RETURN(message, EALREADY, {});
233  }
234
235  // There are still pending consumers, return busy.
236  if (pending_consumers_ > 0)
237    REPLY_ERROR_RETURN(message, EBUSY, {});
238
239  ClearAvailable();
240  producer_owns_ = true;
241  post_fence_.close();
242  return std::move(returned_fence_);
243}
244
245std::pair<BorrowedFence, BufferWrapper<std::uint8_t*>>
246ProducerChannel::OnConsumerAcquire(Message& message,
247                                   std::size_t metadata_size) {
248  ATRACE_NAME("ProducerChannel::OnConsumerAcquire");
249  ALOGD_IF(TRACE, "ProducerChannel::OnConsumerAcquire: buffer_id=%d",
250           buffer_id());
251  if (producer_owns_) {
252    ALOGE("ProducerChannel::OnConsumerAcquire: Not in posted state!");
253    REPLY_ERROR_RETURN(message, EBUSY, {});
254  }
255
256  // Return a borrowed fd to avoid unnecessary duplication of the underlying fd.
257  // Serialization just needs to read the handle.
258  if (metadata_size == 0)
259    return std::make_pair(post_fence_.borrow(),
260                          WrapBuffer<std::uint8_t>(nullptr, 0));
261  else
262    return std::make_pair(post_fence_.borrow(),
263                          WrapBuffer(meta_.get(), meta_size_bytes_));
264}
265
266int ProducerChannel::OnConsumerRelease(Message&, LocalFence release_fence) {
267  ATRACE_NAME("ProducerChannel::OnConsumerRelease");
268  ALOGD_IF(TRACE, "ProducerChannel::OnConsumerRelease: buffer_id=%d",
269           buffer_id());
270  if (producer_owns_) {
271    ALOGE("ProducerChannel::OnConsumerRelease: Not in acquired state!");
272    return -EBUSY;
273  }
274
275  // Attempt to merge the fences if necessary.
276  if (release_fence) {
277    if (returned_fence_) {
278      LocalFence merged_fence(sync_merge("bufferhub_merged",
279                                         returned_fence_.get_fd(),
280                                         release_fence.get_fd()));
281      const int error = errno;
282      if (!merged_fence) {
283        ALOGE("ProducerChannel::OnConsumerRelease: Failed to merge fences: %s",
284              strerror(error));
285        return -error;
286      }
287      returned_fence_ = std::move(merged_fence);
288    } else {
289      returned_fence_ = std::move(release_fence);
290    }
291  }
292
293  OnConsumerIgnored();
294  return 0;
295}
296
297void ProducerChannel::OnConsumerIgnored() {
298  if (!--pending_consumers_)
299    SignalAvailable();
300  ALOGD_IF(TRACE,
301           "ProducerChannel::OnConsumerIgnored: buffer_id=%d %d consumers left",
302           buffer_id(), pending_consumers_);
303}
304
305int ProducerChannel::OnProducerMakePersistent(Message& message,
306                                              const std::string& name,
307                                              int user_id, int group_id) {
308  ATRACE_NAME("ProducerChannel::OnProducerMakePersistent");
309  ALOGD_IF(TRACE,
310           "ProducerChannel::OnProducerMakePersistent: buffer_id=%d name=%s "
311           "user_id=%d group_id=%d",
312           buffer_id(), name.c_str(), user_id, group_id);
313
314  if (name.empty() || (user_id < 0 && user_id != kNoCheckId) ||
315      (group_id < 0 && group_id != kNoCheckId)) {
316    return -EINVAL;
317  }
318
319  // Try to add this buffer with the requested name.
320  if (service()->AddNamedBuffer(name, std::static_pointer_cast<ProducerChannel>(
321                                          shared_from_this()))) {
322    // If successful, set the requested permissions.
323
324    // A value of zero indicates that the ids from the sending process should be
325    // used.
326    if (user_id == kUseCallerId)
327      user_id = message.GetEffectiveUserId();
328    if (group_id == kUseCallerId)
329      group_id = message.GetEffectiveGroupId();
330
331    owner_user_id_ = user_id;
332    owner_group_id_ = group_id;
333    name_ = name;
334    return 0;
335  } else {
336    // Otherwise a buffer with that name already exists.
337    return -EALREADY;
338  }
339}
340
341int ProducerChannel::OnRemovePersistence(Message&) {
342  if (service()->RemoveNamedBuffer(*this))
343    return 0;
344  else
345    return -ENOENT;
346}
347
348void ProducerChannel::AddConsumer(ConsumerChannel* channel) {
349  consumer_channels_.push_back(channel);
350}
351
352void ProducerChannel::RemoveConsumer(ConsumerChannel* channel) {
353  consumer_channels_.erase(
354      std::find(consumer_channels_.begin(), consumer_channels_.end(), channel));
355}
356
357// Returns true if either the user or group ids match the owning ids or both
358// owning ids are not set, in which case access control does not apply.
359bool ProducerChannel::CheckAccess(int euid, int egid) {
360  const bool no_check =
361      owner_user_id_ == kNoCheckId && owner_group_id_ == kNoCheckId;
362  const bool euid_check = euid == owner_user_id_ || euid == kRootId;
363  const bool egid_check = egid == owner_group_id_ || egid == kRootId;
364  return no_check || euid_check || egid_check;
365}
366
367// Returns true if the given parameters match the underlying buffer parameters.
368bool ProducerChannel::CheckParameters(int width, int height, int format,
369                                      int usage, size_t meta_size_bytes,
370                                      size_t slice_count) {
371  return slices_.size() == slice_count && meta_size_bytes == meta_size_bytes_ &&
372         slices_[0].width() == width && slices_[0].height() == height &&
373         slices_[0].format() == format && slices_[0].usage() == usage;
374}
375
376}  // namespace dvr
377}  // namespace android
378