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