1 /*
2 * Copyright (C) 2022 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #define LOG_TAG "AidlBufferPoolStatus"
18 //#define LOG_NDEBUG 0
19
20 #include <thread>
21 #include <time.h>
22 #include <aidl/android/hardware/media/bufferpool2/BufferStatus.h>
23 #include "BufferStatus.h"
24
25 namespace aidl::android::hardware::media::bufferpool2::implementation {
26
27 using aidl::android::hardware::media::bufferpool2::BufferStatus;
28
wrappedMinus(uint32_t a,uint32_t b)29 uint32_t wrappedMinus(uint32_t a, uint32_t b) {
30 if (a >= b) {
31 return a - b;
32 } else {
33 return ~(b - a) + 1;
34 }
35 }
36
isMessageLater(uint32_t curMsgId,uint32_t prevMsgId)37 bool isMessageLater(uint32_t curMsgId, uint32_t prevMsgId) {
38 return curMsgId != prevMsgId &&
39 wrappedMinus(curMsgId, prevMsgId) < wrappedMinus(prevMsgId, curMsgId);
40 }
41
isBufferInRange(BufferId from,BufferId to,BufferId bufferId)42 bool isBufferInRange(BufferId from, BufferId to, BufferId bufferId) {
43 if (from < to) {
44 return from <= bufferId && bufferId < to;
45 } else { // wrap happens
46 return from <= bufferId || bufferId < to;
47 }
48 }
49
50 static constexpr int kNumElementsInQueue = 1024*16;
51 static constexpr int kMinElementsToSyncInQueue = 128;
52
open(ConnectionId id,StatusDescriptor * fmqDescPtr)53 BufferPoolStatus BufferStatusObserver::open(
54 ConnectionId id, StatusDescriptor* fmqDescPtr) {
55 if (mBufferStatusQueues.find(id) != mBufferStatusQueues.end()) {
56 ALOGE("connection id collision %lld", (unsigned long long)id);
57 return ResultStatus::CRITICAL_ERROR;
58 }
59 auto queue = std::make_unique<BufferStatusQueue>(kNumElementsInQueue);
60 if (!queue || queue->isValid() == false) {
61 return ResultStatus::NO_MEMORY;
62 }
63 *fmqDescPtr = queue->dupeDesc();
64 auto result = mBufferStatusQueues.insert(
65 std::make_pair(id, std::move(queue)));
66 if (!result.second) {
67 return ResultStatus::NO_MEMORY;
68 }
69 return ResultStatus::OK;
70 }
71
close(ConnectionId id)72 BufferPoolStatus BufferStatusObserver::close(ConnectionId id) {
73 if (mBufferStatusQueues.find(id) == mBufferStatusQueues.end()) {
74 return ResultStatus::CRITICAL_ERROR;
75 }
76 mBufferStatusQueues.erase(id);
77 return ResultStatus::OK;
78 }
79
getBufferStatusChanges(std::vector<BufferStatusMessage> & messages)80 void BufferStatusObserver::getBufferStatusChanges(std::vector<BufferStatusMessage> &messages) {
81 for (auto it = mBufferStatusQueues.begin(); it != mBufferStatusQueues.end(); ++it) {
82 BufferStatusMessage message;
83 size_t avail = it->second->availableToRead();
84 while (avail > 0) {
85 if (!it->second->read(&message, 1)) {
86 // Since available # of reads are already confirmed,
87 // this should not happen.
88 // TODO: error handling (spurious client?)
89 ALOGW("FMQ message cannot be read from %lld", (long long)it->first);
90 return;
91 }
92 message.connectionId = it->first;
93 messages.push_back(message);
94 --avail;
95 }
96 }
97 }
98
BufferStatusChannel(const StatusDescriptor & fmqDesc)99 BufferStatusChannel::BufferStatusChannel(
100 const StatusDescriptor &fmqDesc) {
101 auto queue = std::make_unique<BufferStatusQueue>(fmqDesc);
102 if (!queue || queue->isValid() == false) {
103 mValid = false;
104 return;
105 }
106 mValid = true;
107 mBufferStatusQueue = std::move(queue);
108 }
109
isValid()110 bool BufferStatusChannel::isValid() {
111 return mValid;
112 }
113
needsSync()114 bool BufferStatusChannel::needsSync() {
115 if (mValid) {
116 size_t avail = mBufferStatusQueue->availableToWrite();
117 return avail + kMinElementsToSyncInQueue < kNumElementsInQueue;
118 }
119 return false;
120 }
121
postBufferRelease(ConnectionId connectionId,std::list<BufferId> & pending,std::list<BufferId> & posted)122 void BufferStatusChannel::postBufferRelease(
123 ConnectionId connectionId,
124 std::list<BufferId> &pending, std::list<BufferId> &posted) {
125 if (mValid && pending.size() > 0) {
126 size_t avail = mBufferStatusQueue->availableToWrite();
127 avail = std::min(avail, pending.size());
128 BufferStatusMessage message;
129 for (size_t i = 0 ; i < avail; ++i) {
130 BufferId id = pending.front();
131 message.status = BufferStatus::NOT_USED;
132 message.bufferId = id;
133 message.connectionId = connectionId;
134 if (!mBufferStatusQueue->write(&message, 1)) {
135 // Since available # of writes are already confirmed,
136 // this should not happen.
137 // TODO: error handing?
138 ALOGW("FMQ message cannot be sent from %lld", (long long)connectionId);
139 return;
140 }
141 pending.pop_front();
142 posted.push_back(id);
143 }
144 }
145 }
146
postBufferInvalidateAck(ConnectionId connectionId,uint32_t invalidateId,bool * invalidated)147 void BufferStatusChannel::postBufferInvalidateAck(
148 ConnectionId connectionId,
149 uint32_t invalidateId,
150 bool *invalidated) {
151 if (mValid && !*invalidated) {
152 size_t avail = mBufferStatusQueue->availableToWrite();
153 if (avail > 0) {
154 BufferStatusMessage message;
155 message.status = BufferStatus::INVALIDATION_ACK;
156 message.bufferId = invalidateId;
157 message.connectionId = connectionId;
158 if (!mBufferStatusQueue->write(&message, 1)) {
159 // Since available # of writes are already confirmed,
160 // this should not happen.
161 // TODO: error handing?
162 ALOGW("FMQ message cannot be sent from %lld", (long long)connectionId);
163 return;
164 }
165 *invalidated = true;
166 }
167 }
168 }
169
postBufferStatusMessage(TransactionId transactionId,BufferId bufferId,BufferStatus status,ConnectionId connectionId,ConnectionId targetId,std::list<BufferId> & pending,std::list<BufferId> & posted)170 bool BufferStatusChannel::postBufferStatusMessage(
171 TransactionId transactionId, BufferId bufferId,
172 BufferStatus status, ConnectionId connectionId, ConnectionId targetId,
173 std::list<BufferId> &pending, std::list<BufferId> &posted) {
174 if (mValid) {
175 size_t avail = mBufferStatusQueue->availableToWrite();
176 size_t numPending = pending.size();
177 if (avail >= numPending + 1) {
178 BufferStatusMessage release, message;
179 for (size_t i = 0; i < numPending; ++i) {
180 BufferId id = pending.front();
181 release.status = BufferStatus::NOT_USED;
182 release.bufferId = id;
183 release.connectionId = connectionId;
184 if (!mBufferStatusQueue->write(&release, 1)) {
185 // Since available # of writes are already confirmed,
186 // this should not happen.
187 // TODO: error handling?
188 ALOGW("FMQ message cannot be sent from %lld", (long long)connectionId);
189 return false;
190 }
191 pending.pop_front();
192 posted.push_back(id);
193 }
194 message.transactionId = transactionId;
195 message.bufferId = bufferId;
196 message.status = status;
197 message.connectionId = connectionId;
198 message.targetConnectionId = targetId;
199 // TODO : timesatamp
200 message.timestampUs = 0;
201 if (!mBufferStatusQueue->write(&message, 1)) {
202 // Since available # of writes are already confirmed,
203 // this should not happen.
204 ALOGW("FMQ message cannot be sent from %lld", (long long)connectionId);
205 return false;
206 }
207 return true;
208 }
209 }
210 return false;
211 }
212
BufferInvalidationListener(const InvalidationDescriptor & fmqDesc)213 BufferInvalidationListener::BufferInvalidationListener(
214 const InvalidationDescriptor &fmqDesc) {
215 std::unique_ptr<BufferInvalidationQueue> queue =
216 std::make_unique<BufferInvalidationQueue>(fmqDesc);
217 if (!queue || queue->isValid() == false) {
218 mValid = false;
219 return;
220 }
221 mValid = true;
222 mBufferInvalidationQueue = std::move(queue);
223 // drain previous messages
224 size_t avail = std::min(
225 mBufferInvalidationQueue->availableToRead(), (size_t) kNumElementsInQueue);
226 std::vector<BufferInvalidationMessage> temp(avail);
227 if (avail > 0) {
228 mBufferInvalidationQueue->read(temp.data(), avail);
229 }
230 }
231
getInvalidations(std::vector<BufferInvalidationMessage> & messages)232 void BufferInvalidationListener::getInvalidations(
233 std::vector<BufferInvalidationMessage> &messages) {
234 // Try twice in case of overflow.
235 // TODO: handling overflow though it may not happen.
236 for (int i = 0; i < 2; ++i) {
237 size_t avail = std::min(
238 mBufferInvalidationQueue->availableToRead(), (size_t) kNumElementsInQueue);
239 if (avail > 0) {
240 std::vector<BufferInvalidationMessage> temp(avail);
241 if (mBufferInvalidationQueue->read(temp.data(), avail)) {
242 messages.reserve(messages.size() + avail);
243 for (auto it = temp.begin(); it != temp.end(); ++it) {
244 messages.push_back(*it);
245 }
246 break;
247 }
248 } else {
249 return;
250 }
251 }
252 }
253
isValid()254 bool BufferInvalidationListener::isValid() {
255 return mValid;
256 }
257
BufferInvalidationChannel()258 BufferInvalidationChannel::BufferInvalidationChannel()
259 : mValid(true),
260 mBufferInvalidationQueue(
261 std::make_unique<BufferInvalidationQueue>(kNumElementsInQueue, true)) {
262 if (!mBufferInvalidationQueue || mBufferInvalidationQueue->isValid() == false) {
263 mValid = false;
264 }
265 }
266
isValid()267 bool BufferInvalidationChannel::isValid() {
268 return mValid;
269 }
270
getDesc(InvalidationDescriptor * fmqDescPtr)271 void BufferInvalidationChannel::getDesc(InvalidationDescriptor *fmqDescPtr) {
272 if (mValid) {
273 *fmqDescPtr = mBufferInvalidationQueue->dupeDesc();
274 }
275 // TODO: writing invalid descriptor?
276 }
277
postInvalidation(uint32_t msgId,BufferId fromId,BufferId toId)278 void BufferInvalidationChannel::postInvalidation(
279 uint32_t msgId, BufferId fromId, BufferId toId) {
280 BufferInvalidationMessage message;
281
282 message.messageId = msgId;
283 message.fromBufferId = fromId;
284 message.toBufferId = toId;
285 // TODO: handle failure (it does not happen normally.)
286 mBufferInvalidationQueue->write(&message);
287 }
288
289 } // namespace ::aidl::android::hardware::media::bufferpool2::implementation
290
291