blob: be5887c0e48e18b273a7848d681c6770a51538d5 [file] [log] [blame]
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +00001/*
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 "AHAL_Stream"
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000018#include <android-base/logging.h>
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000019#include <utils/SystemClock.h>
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000020
Mikhail Naganovef6bc742022-10-06 00:14:19 +000021#include <Utils.h>
22
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000023#include "core-impl/Module.h"
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000024#include "core-impl/Stream.h"
25
26using aidl::android::hardware::audio::common::SinkMetadata;
27using aidl::android::hardware::audio::common::SourceMetadata;
Mikhail Naganovef6bc742022-10-06 00:14:19 +000028using aidl::android::media::audio::common::AudioDevice;
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000029using aidl::android::media::audio::common::AudioOffloadInfo;
Mikhail Naganovef6bc742022-10-06 00:14:19 +000030using android::hardware::audio::common::getChannelCount;
31using android::hardware::audio::common::getFrameSizeInBytes;
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000032
33namespace aidl::android::hardware::audio::core {
34
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000035void StreamContext::fillDescriptor(StreamDescriptor* desc) {
36 if (mCommandMQ) {
37 desc->command = mCommandMQ->dupeDesc();
38 }
39 if (mReplyMQ) {
40 desc->reply = mReplyMQ->dupeDesc();
41 }
42 if (mDataMQ) {
Mikhail Naganovef6bc742022-10-06 00:14:19 +000043 const size_t frameSize = getFrameSize();
44 desc->frameSizeBytes = frameSize;
45 desc->bufferSizeFrames = mDataMQ->getQuantumCount() * mDataMQ->getQuantumSize() / frameSize;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000046 desc->audio.set<StreamDescriptor::AudioBuffer::Tag::fmq>(mDataMQ->dupeDesc());
47 }
Mikhail Naganov6a4872d2022-06-15 21:39:04 +000048}
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +000049
Mikhail Naganovef6bc742022-10-06 00:14:19 +000050size_t StreamContext::getFrameSize() const {
51 return getFrameSizeInBytes(mFormat, mChannelLayout);
52}
53
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000054bool StreamContext::isValid() const {
55 if (mCommandMQ && !mCommandMQ->isValid()) {
56 LOG(ERROR) << "command FMQ is invalid";
57 return false;
58 }
59 if (mReplyMQ && !mReplyMQ->isValid()) {
60 LOG(ERROR) << "reply FMQ is invalid";
61 return false;
62 }
Mikhail Naganovef6bc742022-10-06 00:14:19 +000063 if (getFrameSize() == 0) {
64 LOG(ERROR) << "frame size is invalid";
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +000065 return false;
66 }
67 if (mDataMQ && !mDataMQ->isValid()) {
68 LOG(ERROR) << "data FMQ is invalid";
69 return false;
70 }
71 return true;
72}
73
74void StreamContext::reset() {
75 mCommandMQ.reset();
76 mReplyMQ.reset();
77 mDataMQ.reset();
78}
79
80std::string StreamWorkerCommonLogic::init() {
81 if (mCommandMQ == nullptr) return "Command MQ is null";
82 if (mReplyMQ == nullptr) return "Reply MQ is null";
83 if (mDataMQ == nullptr) return "Data MQ is null";
84 if (sizeof(decltype(mDataBuffer)::element_type) != mDataMQ->getQuantumSize()) {
85 return "Unexpected Data MQ quantum size: " + std::to_string(mDataMQ->getQuantumSize());
86 }
87 mDataBufferSize = mDataMQ->getQuantumCount() * mDataMQ->getQuantumSize();
88 mDataBuffer.reset(new (std::nothrow) int8_t[mDataBufferSize]);
89 if (mDataBuffer == nullptr) {
90 return "Failed to allocate data buffer for element count " +
91 std::to_string(mDataMQ->getQuantumCount()) +
92 ", size in bytes: " + std::to_string(mDataBufferSize);
93 }
94 return "";
95}
96
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +000097void StreamWorkerCommonLogic::populateReply(StreamDescriptor::Reply* reply,
98 bool isConnected) const {
Mikhail Naganov549a8222022-11-23 18:30:07 +000099 reply->status = STATUS_OK;
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000100 if (isConnected) {
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000101 reply->observable.frames = mFrameCount;
102 reply->observable.timeNs = ::android::elapsedRealtimeNano();
103 } else {
Mikhail Naganov549a8222022-11-23 18:30:07 +0000104 reply->observable.frames = StreamDescriptor::Position::UNKNOWN;
105 reply->observable.timeNs = StreamDescriptor::Position::UNKNOWN;
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000106 }
107}
108
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000109void StreamWorkerCommonLogic::populateReplyWrongState(
110 StreamDescriptor::Reply* reply, const StreamDescriptor::Command& command) const {
111 LOG(WARNING) << "command '" << toString(command.getTag())
112 << "' can not be handled in the state " << toString(mState);
113 reply->status = STATUS_INVALID_OPERATION;
114}
115
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000116const std::string StreamInWorkerLogic::kThreadName = "reader";
117
118StreamInWorkerLogic::Status StreamInWorkerLogic::cycle() {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000119 // Note: for input streams, draining is driven by the client, thus
120 // "empty buffer" condition can only happen while handling the 'burst'
121 // command. Thus, unlike for output streams, it does not make sense to
122 // delay the 'DRAINING' state here by 'mTransientStateDelayMs'.
123 // TODO: Add a delay for transitions of async operations when/if they added.
124
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000125 StreamDescriptor::Command command{};
126 if (!mCommandMQ->readBlocking(&command, 1)) {
127 LOG(ERROR) << __func__ << ": reading of command from MQ failed";
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000128 mState = StreamDescriptor::State::ERROR;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000129 return Status::ABORT;
130 }
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000131 LOG(DEBUG) << __func__ << ": received command " << command.toString() << " in " << kThreadName;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000132 StreamDescriptor::Reply reply{};
Mikhail Naganov98334432022-11-09 02:44:32 +0000133 reply.status = STATUS_BAD_VALUE;
134 using Tag = StreamDescriptor::Command::Tag;
135 switch (command.getTag()) {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000136 case Tag::halReservedExit:
137 if (const int32_t cookie = command.get<Tag::halReservedExit>();
Mikhail Naganov98334432022-11-09 02:44:32 +0000138 cookie == mInternalCommandCookie) {
Mikhail Naganov98334432022-11-09 02:44:32 +0000139 setClosed();
140 // This is an internal command, no need to reply.
141 return Status::EXIT;
142 } else {
143 LOG(WARNING) << __func__ << ": EXIT command has a bad cookie: " << cookie;
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000144 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000145 break;
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000146 case Tag::getStatus:
147 populateReply(&reply, mIsConnected);
148 break;
Mikhail Naganov98334432022-11-09 02:44:32 +0000149 case Tag::start:
Mikhail Naganov98334432022-11-09 02:44:32 +0000150 if (mState == StreamDescriptor::State::STANDBY ||
151 mState == StreamDescriptor::State::DRAINING) {
152 populateReply(&reply, mIsConnected);
153 mState = mState == StreamDescriptor::State::STANDBY
154 ? StreamDescriptor::State::IDLE
155 : StreamDescriptor::State::ACTIVE;
156 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000157 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000158 }
159 break;
160 case Tag::burst:
161 if (const int32_t fmqByteCount = command.get<Tag::burst>(); fmqByteCount >= 0) {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000162 LOG(DEBUG) << __func__ << ": '" << toString(command.getTag()) << "' command for "
163 << fmqByteCount << " bytes";
Mikhail Naganov98334432022-11-09 02:44:32 +0000164 if (mState == StreamDescriptor::State::IDLE ||
165 mState == StreamDescriptor::State::ACTIVE ||
166 mState == StreamDescriptor::State::PAUSED ||
167 mState == StreamDescriptor::State::DRAINING) {
168 if (!read(fmqByteCount, &reply)) {
169 mState = StreamDescriptor::State::ERROR;
170 }
171 if (mState == StreamDescriptor::State::IDLE ||
172 mState == StreamDescriptor::State::PAUSED) {
173 mState = StreamDescriptor::State::ACTIVE;
174 } else if (mState == StreamDescriptor::State::DRAINING) {
175 // To simplify the reference code, we assume that the read operation
176 // has consumed all the data remaining in the hardware buffer.
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000177 // In a real implementation, here we would either remain in
178 // the 'DRAINING' state, or transfer to 'STANDBY' depending on the
179 // buffer state.
Mikhail Naganov98334432022-11-09 02:44:32 +0000180 mState = StreamDescriptor::State::STANDBY;
181 }
182 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000183 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000184 }
185 } else {
186 LOG(WARNING) << __func__ << ": invalid burst byte count: " << fmqByteCount;
187 }
188 break;
189 case Tag::drain:
Mikhail Naganov30301a42022-09-13 01:20:45 +0000190 if (command.get<Tag::drain>() == StreamDescriptor::DrainMode::DRAIN_UNSPECIFIED) {
191 if (mState == StreamDescriptor::State::ACTIVE) {
192 usleep(1000); // Simulate a blocking call into the driver.
193 populateReply(&reply, mIsConnected);
194 // Can switch the state to ERROR if a driver error occurs.
195 mState = StreamDescriptor::State::DRAINING;
196 } else {
197 populateReplyWrongState(&reply, command);
198 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000199 } else {
Mikhail Naganov30301a42022-09-13 01:20:45 +0000200 LOG(WARNING) << __func__
201 << ": invalid drain mode: " << toString(command.get<Tag::drain>());
Mikhail Naganov98334432022-11-09 02:44:32 +0000202 }
203 break;
204 case Tag::standby:
Mikhail Naganov98334432022-11-09 02:44:32 +0000205 if (mState == StreamDescriptor::State::IDLE) {
206 usleep(1000); // Simulate a blocking call into the driver.
207 populateReply(&reply, mIsConnected);
208 // Can switch the state to ERROR if a driver error occurs.
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000209 mState = StreamDescriptor::State::STANDBY;
Mikhail Naganov98334432022-11-09 02:44:32 +0000210 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000211 populateReplyWrongState(&reply, command);
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000212 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000213 break;
214 case Tag::pause:
Mikhail Naganov98334432022-11-09 02:44:32 +0000215 if (mState == StreamDescriptor::State::ACTIVE) {
216 usleep(1000); // Simulate a blocking call into the driver.
217 populateReply(&reply, mIsConnected);
218 // Can switch the state to ERROR if a driver error occurs.
219 mState = StreamDescriptor::State::PAUSED;
220 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000221 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000222 }
223 break;
224 case Tag::flush:
Mikhail Naganov98334432022-11-09 02:44:32 +0000225 if (mState == StreamDescriptor::State::PAUSED) {
226 usleep(1000); // Simulate a blocking call into the driver.
227 populateReply(&reply, mIsConnected);
228 // Can switch the state to ERROR if a driver error occurs.
229 mState = StreamDescriptor::State::STANDBY;
230 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000231 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000232 }
233 break;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000234 }
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000235 reply.state = mState;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000236 LOG(DEBUG) << __func__ << ": writing reply " << reply.toString();
237 if (!mReplyMQ->writeBlocking(&reply, 1)) {
238 LOG(ERROR) << __func__ << ": writing of reply " << reply.toString() << " to MQ failed";
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000239 mState = StreamDescriptor::State::ERROR;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000240 return Status::ABORT;
241 }
242 return Status::CONTINUE;
243}
244
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000245bool StreamInWorkerLogic::read(size_t clientSize, StreamDescriptor::Reply* reply) {
246 // Can switch the state to ERROR if a driver error occurs.
247 const size_t byteCount = std::min({clientSize, mDataMQ->availableToWrite(), mDataBufferSize});
248 const bool isConnected = mIsConnected;
249 bool fatal = false;
250 // Simulate reading of data, or provide zeroes if the stream is not connected.
251 for (size_t i = 0; i < byteCount; ++i) {
252 using buffer_type = decltype(mDataBuffer)::element_type;
253 constexpr int kBufferValueRange = std::numeric_limits<buffer_type>::max() -
254 std::numeric_limits<buffer_type>::min() + 1;
255 mDataBuffer[i] = isConnected ? (std::rand() % kBufferValueRange) +
256 std::numeric_limits<buffer_type>::min()
257 : 0;
258 }
259 usleep(3000); // Simulate a blocking call into the driver.
260 // Set 'fatal = true' if a driver error occurs.
261 if (bool success = byteCount > 0 ? mDataMQ->write(&mDataBuffer[0], byteCount) : true; success) {
262 LOG(DEBUG) << __func__ << ": writing of " << byteCount << " bytes into data MQ"
263 << " succeeded; connected? " << isConnected;
264 // Frames are provided and counted regardless of connection status.
265 reply->fmqByteCount += byteCount;
266 mFrameCount += byteCount / mFrameSize;
267 populateReply(reply, isConnected);
268 } else {
269 LOG(WARNING) << __func__ << ": writing of " << byteCount << " bytes of data to MQ failed";
270 reply->status = STATUS_NOT_ENOUGH_DATA;
271 }
272 reply->latencyMs = Module::kLatencyMs;
273 return !fatal;
274}
275
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000276const std::string StreamOutWorkerLogic::kThreadName = "writer";
277
278StreamOutWorkerLogic::Status StreamOutWorkerLogic::cycle() {
Mikhail Naganov30301a42022-09-13 01:20:45 +0000279 if (mState == StreamDescriptor::State::DRAINING ||
280 mState == StreamDescriptor::State::TRANSFERRING) {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000281 if (auto stateDurationMs = std::chrono::duration_cast<std::chrono::milliseconds>(
282 std::chrono::steady_clock::now() - mTransientStateStart);
283 stateDurationMs >= mTransientStateDelayMs) {
Mikhail Naganov30301a42022-09-13 01:20:45 +0000284 if (mAsyncCallback == nullptr) {
285 // In blocking mode, mState can only be DRAINING.
286 mState = StreamDescriptor::State::IDLE;
287 } else {
288 // In a real implementation, the driver should notify the HAL about
289 // drain or transfer completion. In the stub, we switch unconditionally.
290 if (mState == StreamDescriptor::State::DRAINING) {
291 mState = StreamDescriptor::State::IDLE;
292 ndk::ScopedAStatus status = mAsyncCallback->onDrainReady();
293 if (!status.isOk()) {
294 LOG(ERROR) << __func__ << ": error from onDrainReady: " << status;
295 }
296 } else {
297 mState = StreamDescriptor::State::ACTIVE;
298 ndk::ScopedAStatus status = mAsyncCallback->onTransferReady();
299 if (!status.isOk()) {
300 LOG(ERROR) << __func__ << ": error from onTransferReady: " << status;
301 }
302 }
303 }
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000304 if (mTransientStateDelayMs.count() != 0) {
305 LOG(DEBUG) << __func__ << ": switched to state " << toString(mState)
306 << " after a timeout";
307 }
308 }
309 }
310
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000311 StreamDescriptor::Command command{};
312 if (!mCommandMQ->readBlocking(&command, 1)) {
313 LOG(ERROR) << __func__ << ": reading of command from MQ failed";
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000314 mState = StreamDescriptor::State::ERROR;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000315 return Status::ABORT;
316 }
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000317 LOG(DEBUG) << __func__ << ": received command " << command.toString() << " in " << kThreadName;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000318 StreamDescriptor::Reply reply{};
Mikhail Naganov98334432022-11-09 02:44:32 +0000319 reply.status = STATUS_BAD_VALUE;
320 using Tag = StreamDescriptor::Command::Tag;
321 switch (command.getTag()) {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000322 case Tag::halReservedExit:
323 if (const int32_t cookie = command.get<Tag::halReservedExit>();
Mikhail Naganov98334432022-11-09 02:44:32 +0000324 cookie == mInternalCommandCookie) {
Mikhail Naganov98334432022-11-09 02:44:32 +0000325 setClosed();
326 // This is an internal command, no need to reply.
327 return Status::EXIT;
328 } else {
329 LOG(WARNING) << __func__ << ": EXIT command has a bad cookie: " << cookie;
330 }
331 break;
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000332 case Tag::getStatus:
333 populateReply(&reply, mIsConnected);
334 break;
Mikhail Naganov30301a42022-09-13 01:20:45 +0000335 case Tag::start: {
336 bool commandAccepted = true;
Mikhail Naganov98334432022-11-09 02:44:32 +0000337 switch (mState) {
338 case StreamDescriptor::State::STANDBY:
339 mState = StreamDescriptor::State::IDLE;
340 break;
341 case StreamDescriptor::State::PAUSED:
342 mState = StreamDescriptor::State::ACTIVE;
343 break;
344 case StreamDescriptor::State::DRAIN_PAUSED:
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000345 switchToTransientState(StreamDescriptor::State::DRAINING);
Mikhail Naganov30301a42022-09-13 01:20:45 +0000346 break;
347 case StreamDescriptor::State::TRANSFER_PAUSED:
348 switchToTransientState(StreamDescriptor::State::TRANSFERRING);
Mikhail Naganov98334432022-11-09 02:44:32 +0000349 break;
350 default:
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000351 populateReplyWrongState(&reply, command);
Mikhail Naganov30301a42022-09-13 01:20:45 +0000352 commandAccepted = false;
Mikhail Naganov98334432022-11-09 02:44:32 +0000353 }
Mikhail Naganov30301a42022-09-13 01:20:45 +0000354 if (commandAccepted) {
355 populateReply(&reply, mIsConnected);
356 }
357 } break;
Mikhail Naganov98334432022-11-09 02:44:32 +0000358 case Tag::burst:
359 if (const int32_t fmqByteCount = command.get<Tag::burst>(); fmqByteCount >= 0) {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000360 LOG(DEBUG) << __func__ << ": '" << toString(command.getTag()) << "' command for "
361 << fmqByteCount << " bytes";
Mikhail Naganov30301a42022-09-13 01:20:45 +0000362 if (mState != StreamDescriptor::State::ERROR &&
363 mState != StreamDescriptor::State::TRANSFERRING &&
364 mState != StreamDescriptor::State::TRANSFER_PAUSED) {
Mikhail Naganov98334432022-11-09 02:44:32 +0000365 if (!write(fmqByteCount, &reply)) {
366 mState = StreamDescriptor::State::ERROR;
367 }
368 if (mState == StreamDescriptor::State::STANDBY ||
Mikhail Naganov30301a42022-09-13 01:20:45 +0000369 mState == StreamDescriptor::State::DRAIN_PAUSED ||
370 mState == StreamDescriptor::State::PAUSED) {
371 if (mAsyncCallback == nullptr ||
372 mState != StreamDescriptor::State::DRAIN_PAUSED) {
373 mState = StreamDescriptor::State::PAUSED;
374 } else {
375 mState = StreamDescriptor::State::TRANSFER_PAUSED;
376 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000377 } else if (mState == StreamDescriptor::State::IDLE ||
Mikhail Naganov30301a42022-09-13 01:20:45 +0000378 mState == StreamDescriptor::State::DRAINING ||
379 mState == StreamDescriptor::State::ACTIVE) {
380 if (mAsyncCallback == nullptr || reply.fmqByteCount == fmqByteCount) {
381 mState = StreamDescriptor::State::ACTIVE;
382 } else {
383 switchToTransientState(StreamDescriptor::State::TRANSFERRING);
384 }
385 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000386 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000387 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000388 }
389 } else {
390 LOG(WARNING) << __func__ << ": invalid burst byte count: " << fmqByteCount;
391 }
392 break;
393 case Tag::drain:
Mikhail Naganov30301a42022-09-13 01:20:45 +0000394 if (command.get<Tag::drain>() == StreamDescriptor::DrainMode::DRAIN_ALL ||
395 command.get<Tag::drain>() == StreamDescriptor::DrainMode::DRAIN_EARLY_NOTIFY) {
396 if (mState == StreamDescriptor::State::ACTIVE ||
397 mState == StreamDescriptor::State::TRANSFERRING) {
398 usleep(1000); // Simulate a blocking call into the driver.
399 populateReply(&reply, mIsConnected);
400 // Can switch the state to ERROR if a driver error occurs.
401 switchToTransientState(StreamDescriptor::State::DRAINING);
402 } else if (mState == StreamDescriptor::State::TRANSFER_PAUSED) {
403 mState = StreamDescriptor::State::DRAIN_PAUSED;
404 populateReply(&reply, mIsConnected);
405 } else {
406 populateReplyWrongState(&reply, command);
407 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000408 } else {
Mikhail Naganov30301a42022-09-13 01:20:45 +0000409 LOG(WARNING) << __func__
410 << ": invalid drain mode: " << toString(command.get<Tag::drain>());
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000411 }
Mikhail Naganov98334432022-11-09 02:44:32 +0000412 break;
413 case Tag::standby:
Mikhail Naganov98334432022-11-09 02:44:32 +0000414 if (mState == StreamDescriptor::State::IDLE) {
415 usleep(1000); // Simulate a blocking call into the driver.
416 populateReply(&reply, mIsConnected);
417 // Can switch the state to ERROR if a driver error occurs.
418 mState = StreamDescriptor::State::STANDBY;
419 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000420 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000421 }
422 break;
Mikhail Naganov30301a42022-09-13 01:20:45 +0000423 case Tag::pause: {
424 bool commandAccepted = true;
425 switch (mState) {
426 case StreamDescriptor::State::ACTIVE:
427 mState = StreamDescriptor::State::PAUSED;
428 break;
429 case StreamDescriptor::State::DRAINING:
430 mState = StreamDescriptor::State::DRAIN_PAUSED;
431 break;
432 case StreamDescriptor::State::TRANSFERRING:
433 mState = StreamDescriptor::State::TRANSFER_PAUSED;
434 break;
435 default:
436 populateReplyWrongState(&reply, command);
437 commandAccepted = false;
Mikhail Naganov98334432022-11-09 02:44:32 +0000438 }
Mikhail Naganov30301a42022-09-13 01:20:45 +0000439 if (commandAccepted) {
440 populateReply(&reply, mIsConnected);
441 }
442 } break;
Mikhail Naganov98334432022-11-09 02:44:32 +0000443 case Tag::flush:
Mikhail Naganov98334432022-11-09 02:44:32 +0000444 if (mState == StreamDescriptor::State::PAUSED ||
Mikhail Naganov30301a42022-09-13 01:20:45 +0000445 mState == StreamDescriptor::State::DRAIN_PAUSED ||
446 mState == StreamDescriptor::State::TRANSFER_PAUSED) {
Mikhail Naganov98334432022-11-09 02:44:32 +0000447 populateReply(&reply, mIsConnected);
448 mState = StreamDescriptor::State::IDLE;
449 } else {
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000450 populateReplyWrongState(&reply, command);
Mikhail Naganov98334432022-11-09 02:44:32 +0000451 }
452 break;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000453 }
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000454 reply.state = mState;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000455 LOG(DEBUG) << __func__ << ": writing reply " << reply.toString();
456 if (!mReplyMQ->writeBlocking(&reply, 1)) {
457 LOG(ERROR) << __func__ << ": writing of reply " << reply.toString() << " to MQ failed";
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000458 mState = StreamDescriptor::State::ERROR;
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000459 return Status::ABORT;
460 }
461 return Status::CONTINUE;
462}
463
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000464bool StreamOutWorkerLogic::write(size_t clientSize, StreamDescriptor::Reply* reply) {
465 const size_t readByteCount = mDataMQ->availableToRead();
466 // Amount of data that the HAL module is going to actually use.
467 const size_t byteCount = std::min({clientSize, readByteCount, mDataBufferSize});
468 bool fatal = false;
469 if (bool success = readByteCount > 0 ? mDataMQ->read(&mDataBuffer[0], readByteCount) : true) {
470 const bool isConnected = mIsConnected;
471 LOG(DEBUG) << __func__ << ": reading of " << readByteCount << " bytes from data MQ"
472 << " succeeded; connected? " << isConnected;
473 // Frames are consumed and counted regardless of connection status.
474 reply->fmqByteCount += byteCount;
475 mFrameCount += byteCount / mFrameSize;
476 populateReply(reply, isConnected);
477 usleep(3000); // Simulate a blocking call into the driver.
478 // Set 'fatal = true' if a driver error occurs.
479 } else {
480 LOG(WARNING) << __func__ << ": reading of " << readByteCount
481 << " bytes of data from MQ failed";
482 reply->status = STATUS_NOT_ENOUGH_DATA;
483 }
484 reply->latencyMs = Module::kLatencyMs;
485 return !fatal;
486}
487
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000488template <class Metadata, class StreamWorker>
489StreamCommon<Metadata, StreamWorker>::~StreamCommon() {
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000490 if (!isClosed()) {
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000491 LOG(ERROR) << __func__ << ": stream was not closed prior to destruction, resource leak";
492 stopWorker();
493 // The worker and the context should clean up by themselves via destructors.
494 }
495}
496
497template <class Metadata, class StreamWorker>
498ndk::ScopedAStatus StreamCommon<Metadata, StreamWorker>::close() {
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000499 LOG(DEBUG) << __func__;
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000500 if (!isClosed()) {
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000501 stopWorker();
502 LOG(DEBUG) << __func__ << ": joining the worker thread...";
503 mWorker.stop();
504 LOG(DEBUG) << __func__ << ": worker thread joined";
505 mContext.reset();
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000506 mWorker.setClosed();
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000507 return ndk::ScopedAStatus::ok();
508 } else {
509 LOG(ERROR) << __func__ << ": stream was already closed";
510 return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);
511 }
512}
513
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000514template <class Metadata, class StreamWorker>
515void StreamCommon<Metadata, StreamWorker>::stopWorker() {
516 if (auto commandMQ = mContext.getCommandMQ(); commandMQ != nullptr) {
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000517 LOG(DEBUG) << __func__ << ": asking the worker to exit...";
Mikhail Naganovbd483c02022-11-17 20:33:39 +0000518 auto cmd = StreamDescriptor::Command::make<StreamDescriptor::Command::Tag::halReservedExit>(
519 mContext.getInternalCommandCookie());
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000520 // Note: never call 'pause' and 'resume' methods of StreamWorker
521 // in the HAL implementation. These methods are to be used by
522 // the client side only. Preventing the worker loop from running
523 // on the HAL side can cause a deadlock.
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000524 if (!commandMQ->writeBlocking(&cmd, 1)) {
525 LOG(ERROR) << __func__ << ": failed to write exit command to the MQ";
526 }
527 LOG(DEBUG) << __func__ << ": done";
528 }
529}
530
531template <class Metadata, class StreamWorker>
532ndk::ScopedAStatus StreamCommon<Metadata, StreamWorker>::updateMetadata(const Metadata& metadata) {
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000533 LOG(DEBUG) << __func__;
Mikhail Naganovcce8e5f2022-09-13 01:20:45 +0000534 if (!isClosed()) {
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000535 mMetadata = metadata;
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000536 return ndk::ScopedAStatus::ok();
537 }
538 LOG(ERROR) << __func__ << ": stream was closed";
539 return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);
540}
541
Mikhail Naganovef6bc742022-10-06 00:14:19 +0000542namespace {
543static std::map<AudioDevice, std::string> transformMicrophones(
544 const std::vector<MicrophoneInfo>& microphones) {
545 std::map<AudioDevice, std::string> result;
546 std::transform(microphones.begin(), microphones.end(), std::inserter(result, result.begin()),
547 [](const auto& mic) { return std::make_pair(mic.device, mic.id); });
548 return result;
549}
550} // namespace
551
552StreamIn::StreamIn(const SinkMetadata& sinkMetadata, StreamContext context,
553 const std::vector<MicrophoneInfo>& microphones)
554 : StreamCommon<SinkMetadata, StreamInWorker>(sinkMetadata, std::move(context)),
555 mMicrophones(transformMicrophones(microphones)) {
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000556 LOG(DEBUG) << __func__;
557}
558
Mikhail Naganovef6bc742022-10-06 00:14:19 +0000559ndk::ScopedAStatus StreamIn::getActiveMicrophones(
560 std::vector<MicrophoneDynamicInfo>* _aidl_return) {
561 std::vector<MicrophoneDynamicInfo> result;
562 std::vector<MicrophoneDynamicInfo::ChannelMapping> channelMapping{
563 getChannelCount(mContext.getChannelLayout()),
564 MicrophoneDynamicInfo::ChannelMapping::DIRECT};
565 for (auto it = mConnectedDevices.begin(); it != mConnectedDevices.end(); ++it) {
566 if (auto micIt = mMicrophones.find(*it); micIt != mMicrophones.end()) {
567 MicrophoneDynamicInfo dynMic;
568 dynMic.id = micIt->second;
569 dynMic.channelMapping = channelMapping;
570 result.push_back(std::move(dynMic));
571 }
572 }
573 *_aidl_return = std::move(result);
574 LOG(DEBUG) << __func__ << ": returning " << ::android::internal::ToString(*_aidl_return);
575 return ndk::ScopedAStatus::ok();
576}
577
578ndk::ScopedAStatus StreamIn::getMicrophoneDirection(MicrophoneDirection* _aidl_return) {
579 LOG(DEBUG) << __func__;
580 (void)_aidl_return;
581 return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
582}
583
584ndk::ScopedAStatus StreamIn::setMicrophoneDirection(MicrophoneDirection in_direction) {
585 LOG(DEBUG) << __func__ << ": direction " << toString(in_direction);
586 return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
587}
588
589ndk::ScopedAStatus StreamIn::getMicrophoneFieldDimension(float* _aidl_return) {
590 LOG(DEBUG) << __func__;
591 (void)_aidl_return;
592 return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
593}
594
595ndk::ScopedAStatus StreamIn::setMicrophoneFieldDimension(float in_zoom) {
596 LOG(DEBUG) << __func__ << ": zoom " << in_zoom;
597 return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
598}
599
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000600StreamOut::StreamOut(const SourceMetadata& sourceMetadata, StreamContext context,
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000601 const std::optional<AudioOffloadInfo>& offloadInfo)
Mikhail Naganov4f5d3f12022-07-22 23:23:25 +0000602 : StreamCommon<SourceMetadata, StreamOutWorker>(sourceMetadata, std::move(context)),
603 mOffloadInfo(offloadInfo) {
Mikhail Naganov6a4872d2022-06-15 21:39:04 +0000604 LOG(DEBUG) << __func__;
605}
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000606
Mikhail Naganovdf5adfd2021-11-11 22:09:22 +0000607} // namespace aidl::android::hardware::audio::core