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