1 /*
2 * Copyright (c) 2023 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "audio_process_in_client.h"
17
18 #include <atomic>
19 #include <cinttypes>
20 #include <condition_variable>
21 #include <sstream>
22 #include <string>
23 #include <mutex>
24 #include <thread>
25
26 #include "iservice_registry.h"
27 #include "system_ability_definition.h"
28
29 #include "audio_errors.h"
30 #include "audio_log.h"
31 #include "audio_system_manager.h"
32 #include "audio_utils.h"
33 #include "securec.h"
34
35 #include "audio_manager_base.h"
36 #include "audio_process_cb_stub.h"
37 #include "audio_server_death_recipient.h"
38 #include "i_audio_process.h"
39 #include "linear_pos_time_model.h"
40
41 namespace OHOS {
42 namespace AudioStandard {
43 class AudioProcessInClientInner : public AudioProcessInClient, public ProcessCbStub {
44 public:
45 explicit AudioProcessInClientInner(const sptr<IAudioProcess> &ipcProxy);
46 ~AudioProcessInClientInner();
47
48 // ProcessCbStub
49 int32_t OnEndpointChange(int32_t status) override;
50
51 int32_t SaveDataCallback(const std::shared_ptr<AudioDataCallback> &dataCallback) override;
52
53 int32_t SaveUnderrunCallback(const std::shared_ptr<ClientUnderrunCallBack> &underrunCallback) override;
54
55 int32_t GetBufferDesc(BufferDesc &bufDesc) const override;
56
57 int32_t Enqueue(const BufferDesc &bufDesc) const override;
58
59 int32_t SetVolume(int32_t vol) override;
60
61 int32_t Start() override;
62
63 int32_t Pause(bool isFlush) override;
64
65 int32_t Resume() override;
66
67 int32_t Stop() override;
68
69 int32_t Release() override;
70
71 // methods for support IAudioStream
72 int32_t GetSessionID(uint32_t &sessionID) override;
73
74 bool GetAudioTime(uint32_t &framePos, int64_t &sec, int64_t &nanoSec) override;
75
76 int32_t GetBufferSize(size_t &bufferSize) override;
77
78 int32_t GetFrameCount(uint32_t &frameCount) override;
79
80 int32_t GetLatency(uint64_t &latency) override;
81
82 int32_t SetVolume(float vol) override;
83
84 float GetVolume() override;
85
86 uint32_t GetUnderflowCount() override;
87
88 int64_t GetFramesWritten() override;
89
90 int64_t GetFramesRead() override;
91
92 void SetApplicationCachePath(const std::string &cachePath) override;
93
94 bool Init(const AudioProcessConfig &config);
95
96 static const sptr<IStandardAudioService> GetAudioServerProxy();
97 static void AudioServerDied(pid_t pid);
98 static constexpr AudioStreamInfo g_targetStreamInfo = {SAMPLE_RATE_48000, ENCODING_PCM, SAMPLE_S16LE, STEREO};
99
100 private:
101 // move it to a common folder
102 static bool ChannelFormatConvert(const AudioStreamData &srcData, const AudioStreamData &dstData);
103
104 bool InitAudioBuffer();
105
106 bool PrepareCurrent(uint64_t curWritePos);
107 void CallClientHandleCurrent();
108 bool FinishHandleCurrent(uint64_t &curWritePos, int64_t &clientWriteCost);
109 int32_t ReadFromProcessClient() const;
110 int32_t RecordReSyncServicePos();
111 int32_t RecordPrepareCurrent(uint64_t curReadPos);
112 int32_t RecordFinishHandleCurrent(uint64_t &curReadPos, int64_t &clientReadCost);
113
114 void UpdateHandleInfo();
115 int64_t GetPredictNextHandleTime(uint64_t posInFrame);
116 bool PrepareNext(uint64_t curHandPos, int64_t &wakeUpTime);
117
118 std::string GetStatusInfo(StreamStatus status);
119 bool KeepLoopRunning();
120 void ProcessCallbackFuc();
121 void RecordProcessCallbackFuc();
122
123 private:
124 static constexpr int64_t ONE_MILLISECOND_DURATION = 1000000; // 1ms
125 static constexpr int64_t MAX_WRITE_COST_DURATION_NANO = 5000000; // 5ms
126 static constexpr int64_t MAX_READ_COST_DURATION_NANO = 5000000; // 5ms
127 static constexpr int64_t WRITE_BEFORE_DURATION_NANO = 2000000; // 2ms
128 static constexpr int64_t RECORD_RESYNC_SLEEP_NANO = 2000000; // 2ms
129 static constexpr int64_t RECORD_HANDLE_DELAY_NANO = 3000000; // 3ms
130 enum ThreadStatus : uint32_t {
131 WAITTING = 0,
132 SLEEPING,
133 INRUNNING,
134 INVALID
135 };
136 AudioProcessConfig processConfig_;
137 bool needConvert_ = false;
138 size_t clientByteSizePerFrame_ = 0;
139 size_t clientSpanSizeInByte_ = 0;
140 sptr<IAudioProcess> processProxy_ = nullptr;
141 std::shared_ptr<OHAudioBuffer> audioBuffer_ = nullptr;
142
143 uint32_t totalSizeInFrame_ = 0;
144 uint32_t spanSizeInFrame_ = 0;
145 uint32_t byteSizePerFrame_ = 0;
146 size_t spanSizeInByte_ = 0;
147 std::weak_ptr<AudioDataCallback> audioDataCallback_;
148 std::weak_ptr<ClientUnderrunCallBack> underrunCallback_;
149
150 std::unique_ptr<uint8_t[]> callbackBuffer_ = nullptr;
151
152 std::mutex statusSwitchLock_;
153 std::atomic<StreamStatus> *streamStatus_ = nullptr;
154 bool isInited_ = false;
155 bool needReSyncPosition_ = true;
156
157 float volumeInFloat_ = 1.0f;
158 int32_t processVolume_ = PROCESS_VOLUME_MAX; // 0 ~ 65536
159 LinearPosTimeModel handleTimeModel_;
160
161 std::thread callbackLoop_; // thread for callback to client and write.
162 bool isCallbackLoopEnd_ = false;
163 std::atomic<ThreadStatus> threadStatus_ = INVALID;
164 std::mutex loopThreadLock_;
165 std::condition_variable threadStatusCV_;
166
167 std::atomic<uint32_t> underflowCount_ = 0;
168 std::string cachePath_;
169 #ifdef DUMP_CLIENT
170 FILE *dcp_ = nullptr;
171 #endif
172 };
173
174 std::mutex g_audioServerProxyMutex;
175 sptr<IStandardAudioService> gAudioServerProxy = nullptr;
176
AudioProcessInClientInner(const sptr<IAudioProcess> & ipcProxy)177 AudioProcessInClientInner::AudioProcessInClientInner(const sptr<IAudioProcess> &ipcProxy) : processProxy_(ipcProxy)
178 {
179 AUDIO_INFO_LOG("AudioProcessInClient construct.");
180 }
181
GetAudioServerProxy()182 const sptr<IStandardAudioService> AudioProcessInClientInner::GetAudioServerProxy()
183 {
184 std::lock_guard<std::mutex> lock(g_audioServerProxyMutex);
185 if (gAudioServerProxy == nullptr) {
186 auto samgr = SystemAbilityManagerClient::GetInstance().GetSystemAbilityManager();
187 if (samgr == nullptr) {
188 AUDIO_ERR_LOG("GetAudioServerProxy: get sa manager failed");
189 return nullptr;
190 }
191 sptr<IRemoteObject> object = samgr->GetSystemAbility(AUDIO_DISTRIBUTED_SERVICE_ID);
192 if (object == nullptr) {
193 AUDIO_ERR_LOG("GetAudioServerProxy: get audio service remote object failed");
194 return nullptr;
195 }
196 gAudioServerProxy = iface_cast<IStandardAudioService>(object);
197 if (gAudioServerProxy == nullptr) {
198 AUDIO_ERR_LOG("GetAudioServerProxy: get audio service proxy failed");
199 return nullptr;
200 }
201
202 // register death recipent to restore proxy
203 sptr<AudioServerDeathRecipient> asDeathRecipient = new(std::nothrow) AudioServerDeathRecipient(getpid());
204 if (asDeathRecipient != nullptr) {
205 asDeathRecipient->SetNotifyCb(std::bind(&AudioProcessInClientInner::AudioServerDied,
206 std::placeholders::_1));
207 bool result = object->AddDeathRecipient(asDeathRecipient);
208 if (!result) {
209 AUDIO_ERR_LOG("GetAudioServerProxy: failed to add deathRecipient");
210 }
211 }
212 }
213 sptr<IStandardAudioService> gasp = gAudioServerProxy;
214 return gasp;
215 }
216
217 /**
218 * When AudioServer died, all stream in client should be notified. As they were proxy stream ,the stub stream
219 * has been destoried in server.
220 */
AudioServerDied(pid_t pid)221 void AudioProcessInClientInner::AudioServerDied(pid_t pid)
222 {
223 AUDIO_INFO_LOG("audio server died, will restore proxy in next call");
224 std::lock_guard<std::mutex> lock(g_audioServerProxyMutex);
225 gAudioServerProxy = nullptr;
226 }
227
Create(const AudioProcessConfig & config)228 std::shared_ptr<AudioProcessInClient> AudioProcessInClient::Create(const AudioProcessConfig &config)
229 {
230 AUDIO_INFO_LOG("Create with config: render flag %{public}d, capturer flag %{public}d, streamType %{public}d.",
231 config.rendererInfo.rendererFlags, config.capturerInfo.capturerFlags, config.streamType);
232 if (config.audioMode == AUDIO_MODE_PLAYBACK && !AudioProcessInClient::CheckIfSupport(config)) {
233 AUDIO_ERR_LOG("CheckIfSupport failed!");
234 return nullptr;
235 }
236 sptr<IStandardAudioService> gasp = AudioProcessInClientInner::GetAudioServerProxy();
237 CHECK_AND_RETURN_RET_LOG(gasp != nullptr, nullptr, "Create failed, can not get service.");
238 AudioProcessConfig resetConfig = config;
239 resetConfig.streamInfo = AudioProcessInClientInner::g_targetStreamInfo;
240 sptr<IRemoteObject> ipcProxy = gasp->CreateAudioProcess(resetConfig);
241 CHECK_AND_RETURN_RET_LOG(ipcProxy != nullptr, nullptr, "Create failed with null ipcProxy.");
242 sptr<IAudioProcess> iProcessProxy = iface_cast<IAudioProcess>(ipcProxy);
243 CHECK_AND_RETURN_RET_LOG(iProcessProxy != nullptr, nullptr, "Create failed when iface_cast.");
244 std::shared_ptr<AudioProcessInClientInner> process = std::make_shared<AudioProcessInClientInner>(iProcessProxy);
245 if (!process->Init(config)) {
246 AUDIO_ERR_LOG("Init failed!");
247 process = nullptr;
248 }
249
250 return process;
251 }
252
~AudioProcessInClientInner()253 AudioProcessInClientInner::~AudioProcessInClientInner()
254 {
255 AUDIO_INFO_LOG("AudioProcessInClient deconstruct.");
256 if (callbackLoop_.joinable()) {
257 if (threadStatus_ == WAITTING) {
258 threadStatusCV_.notify_all();
259 }
260 isCallbackLoopEnd_ = true;
261 callbackLoop_.join();
262 }
263 if (isInited_) {
264 AudioProcessInClientInner::Release();
265 }
266 #ifdef DUMP_CLIENT
267 if (dcp_) {
268 fclose(dcp_);
269 dcp_ = nullptr;
270 }
271 #endif
272 }
273
OnEndpointChange(int32_t status)274 int32_t AudioProcessInClientInner::OnEndpointChange(int32_t status)
275 {
276 AUDIO_INFO_LOG("OnEndpointChange:%{public}d", status);
277 return SUCCESS;
278 }
279
GetSessionID(uint32_t & sessionID)280 int32_t AudioProcessInClientInner::GetSessionID(uint32_t &sessionID)
281 {
282 // note: Get the session id from server.
283 int32_t pid = processConfig_.appInfo.appPid;
284 if (pid < 0) {
285 AUDIO_ERR_LOG("GetSessionID failed:%{public}d", pid);
286 return ERR_OPERATION_FAILED;
287 }
288 sessionID = static_cast<uint32_t>(pid); // using pid as sessionID temporarily
289 return SUCCESS;
290 }
291
GetAudioTime(uint32_t & framePos,int64_t & sec,int64_t & nanoSec)292 bool AudioProcessInClientInner::GetAudioTime(uint32_t &framePos, int64_t &sec, int64_t &nanoSec)
293 {
294 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, false, "buffer is null, maybe not inited.");
295 uint64_t pos = 0;
296 if (processConfig_.audioMode == AUDIO_MODE_PLAYBACK) {
297 pos = audioBuffer_->GetCurWriteFrame();
298 } else {
299 pos = audioBuffer_->GetCurReadFrame();
300 }
301
302 if (pos > UINT32_MAX) {
303 framePos = pos % UINT32_MAX;
304 } else {
305 framePos = static_cast<uint32_t>(pos);
306 }
307 int64_t time = handleTimeModel_.GetTimeOfPos(pos);
308 int64_t deltaTime = 20000000; // note: 20ms
309 time += deltaTime;
310
311 sec = time / AUDIO_NS_PER_SECOND;
312 nanoSec = time % AUDIO_NS_PER_SECOND;
313 return true;
314 }
315
GetBufferSize(size_t & bufferSize)316 int32_t AudioProcessInClientInner::GetBufferSize(size_t &bufferSize)
317 {
318 bufferSize = clientSpanSizeInByte_;
319 return SUCCESS;
320 }
321
GetFrameCount(uint32_t & frameCount)322 int32_t AudioProcessInClientInner::GetFrameCount(uint32_t &frameCount)
323 {
324 frameCount = spanSizeInFrame_;
325 return SUCCESS;
326 }
327
GetLatency(uint64_t & latency)328 int32_t AudioProcessInClientInner::GetLatency(uint64_t &latency)
329 {
330 latency = 20; // 20ms for debug
331 return SUCCESS;
332 }
333
SetVolume(float vol)334 int32_t AudioProcessInClientInner::SetVolume(float vol)
335 {
336 float minVol = 0.0f;
337 float maxVol = 1.0f;
338 if (vol < minVol || vol > maxVol) {
339 AUDIO_ERR_LOG("SetVolume failed to with invalid volume:%{public}f", vol);
340 return ERR_INVALID_PARAM;
341 }
342 int32_t volumeInt = static_cast<int32_t>(vol * PROCESS_VOLUME_MAX);
343 int32_t ret = SetVolume(volumeInt);
344 if (ret == SUCCESS) {
345 volumeInFloat_ = vol;
346 }
347 return ret;
348 }
349
GetVolume()350 float AudioProcessInClientInner::GetVolume()
351 {
352 return volumeInFloat_;
353 }
354
GetUnderflowCount()355 uint32_t AudioProcessInClientInner::GetUnderflowCount()
356 {
357 return underflowCount_.load();
358 }
359
GetFramesWritten()360 int64_t AudioProcessInClientInner::GetFramesWritten()
361 {
362 if (processConfig_.audioMode != AUDIO_MODE_PLAYBACK) {
363 AUDIO_ERR_LOG("Playback not support.");
364 return -1;
365 }
366 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, -1, "buffer is null, maybe not inited.");
367 return audioBuffer_->GetCurWriteFrame();
368 }
369
GetFramesRead()370 int64_t AudioProcessInClientInner::GetFramesRead()
371 {
372 if (processConfig_.audioMode != AUDIO_MODE_RECORD) {
373 AUDIO_ERR_LOG("Record not support.");
374 return -1;
375 }
376 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, -1, "buffer is null, maybe not inited.");
377 return audioBuffer_->GetCurReadFrame();
378 }
379
SetApplicationCachePath(const std::string & cachePath)380 void AudioProcessInClientInner::SetApplicationCachePath(const std::string &cachePath)
381 {
382 AUDIO_INFO_LOG("Using cachePath:%{public}s", cachePath.c_str());
383 cachePath_ = cachePath;
384 }
385
386
InitAudioBuffer()387 bool AudioProcessInClientInner::InitAudioBuffer()
388 {
389 CHECK_AND_RETURN_RET_LOG(processProxy_ != nullptr, false, "Init failed with null ipcProxy.");
390
391 CHECK_AND_RETURN_RET_LOG(processProxy_->RegisterProcessCb(this->AsObject()) == SUCCESS, false,
392 "RegisterProcessCb failed.");
393 int32_t ret = processProxy_->ResolveBuffer(audioBuffer_);
394 if (ret != SUCCESS || audioBuffer_ == nullptr) {
395 AUDIO_ERR_LOG("Init failed to call ResolveBuffer");
396 return false;
397 }
398 streamStatus_ = audioBuffer_->GetStreamStatus();
399 CHECK_AND_RETURN_RET_LOG(streamStatus_ != nullptr, false, "Init failed, access buffer failed.");
400
401 audioBuffer_->GetSizeParameter(totalSizeInFrame_, spanSizeInFrame_, byteSizePerFrame_);
402 spanSizeInByte_ = spanSizeInFrame_ * byteSizePerFrame_;
403
404 if (processConfig_.audioMode == AUDIO_MODE_PLAYBACK && clientByteSizePerFrame_ != 0) {
405 clientSpanSizeInByte_ = spanSizeInFrame_ * clientByteSizePerFrame_;
406 } else {
407 clientSpanSizeInByte_ = spanSizeInByte_;
408 }
409
410 AUDIO_INFO_LOG("Using totalSizeInFrame_ %{public}d spanSizeInFrame_ %{public}d byteSizePerFrame_ %{public}d "
411 "spanSizeInByte_ %{public}zu", totalSizeInFrame_, spanSizeInFrame_, byteSizePerFrame_, spanSizeInByte_);
412
413 callbackBuffer_ = std::make_unique<uint8_t[]>(clientSpanSizeInByte_);
414 CHECK_AND_RETURN_RET_LOG(callbackBuffer_ != nullptr, false, "Init callbackBuffer_ failed.");
415 memset_s(callbackBuffer_.get(), clientSpanSizeInByte_, 0, clientSpanSizeInByte_);
416
417 return true;
418 }
419
GetFormatSize(const AudioStreamInfo & info)420 inline size_t GetFormatSize(const AudioStreamInfo &info)
421 {
422 size_t result = 0;
423 size_t bitWidthSize = 0;
424 switch (info.format) {
425 case SAMPLE_U8:
426 bitWidthSize = 1; // size is 1
427 break;
428 case SAMPLE_S16LE:
429 bitWidthSize = 2; // size is 2
430 break;
431 case SAMPLE_S24LE:
432 bitWidthSize = 3; // size is 3
433 break;
434 case SAMPLE_S32LE:
435 bitWidthSize = 4; // size is 4
436 break;
437 default:
438 bitWidthSize = 2; // size is 2
439 break;
440 }
441
442 size_t channelSize = 0;
443 switch (info.channels) {
444 case MONO:
445 channelSize = 1; // size is 1
446 break;
447 case STEREO:
448 channelSize = 2; // size is 2
449 break;
450 default:
451 channelSize = 2; // size is 2
452 break;
453 }
454 result = bitWidthSize * channelSize;
455 return result;
456 }
457
Init(const AudioProcessConfig & config)458 bool AudioProcessInClientInner::Init(const AudioProcessConfig &config)
459 {
460 AUDIO_INFO_LOG("Call Init.");
461 if (config.streamInfo.format != g_targetStreamInfo.format ||
462 config.streamInfo.channels != g_targetStreamInfo.channels) {
463 needConvert_ = true;
464 }
465 if (config.audioMode == AUDIO_MODE_PLAYBACK) {
466 clientByteSizePerFrame_ = GetFormatSize(config.streamInfo);
467 }
468 AUDIO_DEBUG_LOG("Using clientByteSizePerFrame_:%{public}zu", clientByteSizePerFrame_);
469 bool isBufferInited = InitAudioBuffer();
470 CHECK_AND_RETURN_RET_LOG(isBufferInited, isBufferInited, "%{public}s init audio buffer fail.", __func__);
471 processConfig_ = config;
472
473 bool ret = handleTimeModel_.ConfigSampleRate(processConfig_.streamInfo.samplingRate);
474 CHECK_AND_RETURN_RET_LOG(ret != false, false, "Init LinearPosTimeModel failed.");
475 uint64_t handlePos = 0;
476 int64_t handleTime = 0;
477 audioBuffer_->GetHandleInfo(handlePos, handleTime);
478 handleTimeModel_.ResetFrameStamp(handlePos, handleTime);
479
480 streamStatus_->store(StreamStatus::STREAM_IDEL);
481 if (config.audioMode == AUDIO_MODE_RECORD) {
482 callbackLoop_ = std::thread(&AudioProcessInClientInner::RecordProcessCallbackFuc, this);
483 pthread_setname_np(callbackLoop_.native_handle(), "AudioProcessRecordCb");
484 } else {
485 callbackLoop_ = std::thread(&AudioProcessInClientInner::ProcessCallbackFuc, this);
486 pthread_setname_np(callbackLoop_.native_handle(), "AudioProcessCb");
487 }
488
489 int waitThreadStartTime = 5; // wait for thread start.
490 while (threadStatus_.load() == INVALID) {
491 AUDIO_DEBUG_LOG("%{public}s wait %{public}d ms for %{public}s started...", __func__, waitThreadStartTime,
492 config.audioMode == AUDIO_MODE_RECORD ? "RecordProcessCallbackFuc" : "ProcessCallbackFuc");
493 ClockTime::RelativeSleep(ONE_MILLISECOND_DURATION * waitThreadStartTime);
494 }
495
496 #ifdef DUMP_CLIENT
497 std::stringstream strStream;
498 std::string dumpPatch;
499 strStream << "/data/local/tmp/client-" << processConfig_.appInfo.appUid << ".pcm";
500 strStream >> dumpPatch;
501 AUDIO_INFO_LOG("Client dump using path: %{public}s with uid:%{public}d", dumpPatch.c_str(),
502 processConfig_.appInfo.appUid);
503
504 dcp_ = fopen(dumpPatch.c_str(), "a+");
505 CHECK_AND_BREAK_LOG(dcp_ != nullptr, "Error opening pcm test file!");
506 #endif
507 isInited_ = true;
508 return true;
509 }
510
SaveDataCallback(const std::shared_ptr<AudioDataCallback> & dataCallback)511 int32_t AudioProcessInClientInner::SaveDataCallback(const std::shared_ptr<AudioDataCallback> &dataCallback)
512 {
513 AUDIO_INFO_LOG("%{public}s enter.", __func__);
514 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
515
516 if (dataCallback == nullptr) {
517 AUDIO_ERR_LOG("%{public}s data callback is null.", __func__);
518 return ERR_INVALID_PARAM;
519 }
520 audioDataCallback_ = dataCallback;
521 return SUCCESS;
522 }
523
SaveUnderrunCallback(const std::shared_ptr<ClientUnderrunCallBack> & underrunCallback)524 int32_t AudioProcessInClientInner::SaveUnderrunCallback(const std::shared_ptr<ClientUnderrunCallBack> &underrunCallback)
525 {
526 AUDIO_INFO_LOG("%{public}s enter.", __func__);
527 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
528
529 if (underrunCallback == nullptr) {
530 AUDIO_ERR_LOG("%{public}s underrun callback is null.", __func__);
531 return ERR_INVALID_PARAM;
532 }
533 underrunCallback_ = underrunCallback;
534 return SUCCESS;
535 }
536
ReadFromProcessClient() const537 int32_t AudioProcessInClientInner::ReadFromProcessClient() const
538 {
539 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, ERR_INVALID_HANDLE,
540 "%{public}s audio buffer is null.", __func__);
541 uint64_t curReadPos = audioBuffer_->GetCurReadFrame();
542 Trace trace("AudioProcessInClient::ReadProcessData-<" + std::to_string(curReadPos));
543 BufferDesc readbufDesc = {nullptr, 0, 0};
544 int32_t ret = audioBuffer_->GetReadbuffer(curReadPos, readbufDesc);
545 if (ret != SUCCESS || readbufDesc.buffer == nullptr || readbufDesc.bufLength != spanSizeInByte_ ||
546 readbufDesc.dataLength != spanSizeInByte_) {
547 AUDIO_ERR_LOG("%{public}s get client mmap read buffer failed, ret %{public}d.", __func__, ret);
548 return ERR_OPERATION_FAILED;
549 }
550 ret = memcpy_s(static_cast<void *>(callbackBuffer_.get()), spanSizeInByte_,
551 static_cast<void *>(readbufDesc.buffer), spanSizeInByte_);
552 CHECK_AND_RETURN_RET_LOG(ret == EOK, ERR_OPERATION_FAILED, "%{public}s memcpy fail, ret %{public}d,"
553 " spanSizeInByte %{public}zu.", __func__, ret, spanSizeInByte_);
554 #ifdef DUMP_CLIENT
555 if (dcp_ != nullptr) {
556 fwrite(static_cast<void *>(readbufDesc.buffer), 1, spanSizeInByte_, dcp_);
557 }
558 #endif
559
560 ret = memset_s(readbufDesc.buffer, readbufDesc.bufLength, 0, readbufDesc.bufLength);
561 CHECK_AND_BREAK_LOG(ret == EOK, "%{public}s reset buffer fail, ret %{public}d.", __func__, ret);
562 return SUCCESS;
563 }
564
565 // the buffer will be used by client
GetBufferDesc(BufferDesc & bufDesc) const566 int32_t AudioProcessInClientInner::GetBufferDesc(BufferDesc &bufDesc) const
567 {
568 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "%{public}s not inited!", __func__);
569 Trace trace("AudioProcessInClient::GetBufferDesc");
570
571 if (processConfig_.audioMode == AUDIO_MODE_RECORD) {
572 ReadFromProcessClient();
573 }
574
575 bufDesc.buffer = callbackBuffer_.get();
576 bufDesc.dataLength = clientSpanSizeInByte_;
577 bufDesc.bufLength = clientSpanSizeInByte_;
578 return SUCCESS;
579 }
580
CheckIfSupport(const AudioProcessConfig & config)581 bool AudioProcessInClient::CheckIfSupport(const AudioProcessConfig &config)
582 {
583 if (config.streamInfo.encoding != ENCODING_PCM || config.streamInfo.samplingRate != SAMPLE_RATE_48000) {
584 return false;
585 }
586
587 if (config.streamInfo.format != SAMPLE_S16LE && config.streamInfo.format != SAMPLE_S32LE) {
588 return false;
589 }
590
591 if (config.streamInfo.channels != MONO && config.streamInfo.channels != STEREO) {
592 return false;
593 }
594 return true;
595 }
596
S16MonoToS16Stereo(const BufferDesc & srcDesc,const BufferDesc & dstDesc)597 inline bool S16MonoToS16Stereo(const BufferDesc &srcDesc, const BufferDesc &dstDesc)
598 {
599 size_t half = 2;
600 if (srcDesc.bufLength != dstDesc.bufLength / half || srcDesc.buffer == nullptr || dstDesc.buffer == nullptr) {
601 return false;
602 }
603 int16_t *stcPtr = reinterpret_cast<int16_t *>(srcDesc.buffer);
604 int16_t *dstPtr = reinterpret_cast<int16_t *>(dstDesc.buffer);
605 size_t count = srcDesc.bufLength / half;
606 for (size_t idx = 0; idx < count; idx++) {
607 *(dstPtr++) = *stcPtr;
608 *(dstPtr++) = *stcPtr++;
609 }
610 return true;
611 }
612
S32MonoToS16Stereo(const BufferDesc & srcDesc,const BufferDesc & dstDesc)613 inline bool S32MonoToS16Stereo(const BufferDesc &srcDesc, const BufferDesc &dstDesc)
614 {
615 size_t quarter = 4;
616 if (srcDesc.bufLength != dstDesc.bufLength || srcDesc.buffer == nullptr || dstDesc.buffer == nullptr ||
617 srcDesc.bufLength % quarter != 0) {
618 return false;
619 }
620 int32_t *stcPtr = reinterpret_cast<int32_t *>(srcDesc.buffer);
621 int16_t *dstPtr = reinterpret_cast<int16_t *>(dstDesc.buffer);
622 size_t count = srcDesc.bufLength / quarter;
623
624 double maxInt32 = INT32_MAX;
625 double maxInt16 = INT16_MAX;
626 for (size_t idx = 0; idx < count; idx++) {
627 int16_t temp = static_cast<int16_t>((static_cast<double>(*stcPtr) / maxInt32) * maxInt16);
628 stcPtr++;
629 *(dstPtr++) = temp;
630 *(dstPtr++) = temp;
631 }
632 return true;
633 }
634
S32StereoS16Stereo(const BufferDesc & srcDesc,const BufferDesc & dstDesc)635 inline bool S32StereoS16Stereo(const BufferDesc &srcDesc, const BufferDesc &dstDesc)
636 {
637 size_t half = 2;
638 if (srcDesc.bufLength / half != dstDesc.bufLength || srcDesc.buffer == nullptr || dstDesc.buffer == nullptr ||
639 dstDesc.bufLength % half != 0) {
640 return false;
641 }
642 int32_t *stcPtr = reinterpret_cast<int32_t *>(srcDesc.buffer);
643 int16_t *dstPtr = reinterpret_cast<int16_t *>(dstDesc.buffer);
644 size_t count = srcDesc.bufLength / half / half;
645 double maxInt32 = INT32_MAX;
646 double maxInt16 = INT16_MAX;
647 for (size_t idx = 0; idx < count; idx++) {
648 int16_t temp = static_cast<int16_t>((static_cast<double>(*stcPtr) / maxInt32) * maxInt16);
649 stcPtr++;
650 *(dstPtr++) = temp;
651 }
652 return true;
653 }
654
655 // only support MONO to STEREO and SAMPLE_S32LE to SAMPLE_S16LE
ChannelFormatConvert(const AudioStreamData & srcData,const AudioStreamData & dstData)656 bool AudioProcessInClientInner::ChannelFormatConvert(const AudioStreamData &srcData, const AudioStreamData &dstData)
657 {
658 if (srcData.streamInfo.samplingRate != dstData.streamInfo.samplingRate ||
659 srcData.streamInfo.encoding != dstData.streamInfo.encoding) {
660 return false;
661 }
662 if (srcData.streamInfo.format == SAMPLE_S16LE && srcData.streamInfo.channels == STEREO) {
663 return true; // no need convert
664 }
665 if (srcData.streamInfo.format == SAMPLE_S16LE && srcData.streamInfo.channels == MONO) {
666 return S16MonoToS16Stereo(srcData.bufferDesc, dstData.bufferDesc);
667 }
668 if (srcData.streamInfo.format == SAMPLE_S32LE && srcData.streamInfo.channels == MONO) {
669 return S32MonoToS16Stereo(srcData.bufferDesc, dstData.bufferDesc);
670 }
671 if (srcData.streamInfo.format == SAMPLE_S32LE && srcData.streamInfo.channels == STEREO) {
672 return S32StereoS16Stereo(srcData.bufferDesc, dstData.bufferDesc);
673 }
674
675 return false;
676 }
677
Enqueue(const BufferDesc & bufDesc) const678 int32_t AudioProcessInClientInner::Enqueue(const BufferDesc &bufDesc) const
679 {
680 Trace trace("AudioProcessInClient::Enqueue");
681 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "%{public}s not inited!", __func__);
682
683 if (bufDesc.buffer == nullptr || bufDesc.bufLength != clientSpanSizeInByte_ ||
684 bufDesc.dataLength != clientSpanSizeInByte_) {
685 AUDIO_ERR_LOG("%{public}s bufDesc error, bufLen %{public}zu, dataLen %{public}zu, spanSize %{public}zu.",
686 __func__, bufDesc.bufLength, bufDesc.dataLength, clientSpanSizeInByte_);
687 return ERR_INVALID_PARAM;
688 }
689 // check if this buffer is form us.
690 CHECK_AND_BREAK_LOG(bufDesc.buffer == callbackBuffer_.get(),
691 "%{public}s the buffer is not created by client.", __func__);
692
693 if (processConfig_.audioMode == AUDIO_MODE_PLAYBACK) {
694 BufferDesc curWriteBuffer = {nullptr, 0, 0};
695 uint64_t curWritePos = audioBuffer_->GetCurWriteFrame();
696 Trace writeProcessDataTrace("AudioProcessInClient::WriteProcessData->" + std::to_string(curWritePos));
697 int32_t ret = audioBuffer_->GetWriteBuffer(curWritePos, curWriteBuffer);
698 if (ret != SUCCESS || curWriteBuffer.buffer == nullptr || curWriteBuffer.bufLength != spanSizeInByte_ ||
699 curWriteBuffer.dataLength != spanSizeInByte_) {
700 AUDIO_ERR_LOG("%{public}s get write buffer fail, ret:%{public}d", __func__, ret);
701 return ERR_OPERATION_FAILED;
702 }
703
704 if (!needConvert_) {
705 ret = memcpy_s(static_cast<void *>(curWriteBuffer.buffer), spanSizeInByte_,
706 static_cast<void *>(bufDesc.buffer), spanSizeInByte_);
707 CHECK_AND_RETURN_RET_LOG(ret == SUCCESS, ERR_OPERATION_FAILED, "Copy data failed!");
708 } else {
709 Trace traceConvert("AudioProcessInClient::ChannelFormatConvert");
710 AudioStreamData srcData = {processConfig_.streamInfo, bufDesc, 0, 0};
711 AudioStreamData dstData = {g_targetStreamInfo, curWriteBuffer, 0, 0};
712 bool succ = ChannelFormatConvert(srcData, dstData);
713 CHECK_AND_RETURN_RET_LOG(succ == true, ERR_OPERATION_FAILED, "Convert data failed!");
714 }
715
716 #ifdef DUMP_CLIENT
717 if (dcp_ != nullptr) {
718 fwrite(static_cast<void *>(bufDesc.buffer), 1, clientSpanSizeInByte_, dcp_);
719 }
720 #endif
721 }
722
723 CHECK_AND_BREAK_LOG(memset_s(callbackBuffer_.get(), clientSpanSizeInByte_, 0, clientSpanSizeInByte_) == EOK,
724 "%{public}s reset callback buffer fail.", __func__);
725
726 return SUCCESS;
727 }
728
SetVolume(int32_t vol)729 int32_t AudioProcessInClientInner::SetVolume(int32_t vol)
730 {
731 AUDIO_INFO_LOG("SetVolume proc client mode %{public}d to %{public}d.", processConfig_.audioMode, vol);
732 Trace trace("AudioProcessInClient::SetVolume " + std::to_string(vol));
733 if (vol < 0 || vol > PROCESS_VOLUME_MAX) {
734 AUDIO_ERR_LOG("SetVolume failed, invalid volume:%{public}d", vol);
735 return ERR_INVALID_PARAM;
736 }
737 processVolume_ = vol;
738 return SUCCESS;
739 }
740
Start()741 int32_t AudioProcessInClientInner::Start()
742 {
743 Trace traceWithLog("AudioProcessInClient::Start", true);
744 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
745
746 std::lock_guard<std::mutex> lock(statusSwitchLock_);
747 if (streamStatus_->load() == StreamStatus::STREAM_RUNNING) {
748 AUDIO_INFO_LOG("Start find already started");
749 return SUCCESS;
750 }
751
752 StreamStatus targetStatus = StreamStatus::STREAM_IDEL;
753 if (!streamStatus_->compare_exchange_strong(targetStatus, StreamStatus::STREAM_STARTING)) {
754 AUDIO_ERR_LOG("Start failed, invalid status: %{public}s", GetStatusInfo(targetStatus).c_str());
755 return ERR_ILLEGAL_STATE;
756 }
757
758 if (processProxy_->Start() != SUCCESS) {
759 streamStatus_->store(StreamStatus::STREAM_IDEL);
760 AUDIO_ERR_LOG("Start failed to call process proxy, reset status to IDEL.");
761 threadStatusCV_.notify_all();
762 return ERR_OPERATION_FAILED;
763 }
764 UpdateHandleInfo();
765 streamStatus_->store(StreamStatus::STREAM_RUNNING);
766 threadStatusCV_.notify_all();
767 return SUCCESS;
768 }
769
Pause(bool isFlush)770 int32_t AudioProcessInClientInner::Pause(bool isFlush)
771 {
772 Trace traceWithLog("AudioProcessInClient::Pause", true);
773 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
774
775 std::lock_guard<std::mutex> lock(statusSwitchLock_);
776 if (streamStatus_->load() == StreamStatus::STREAM_PAUSED) {
777 AUDIO_INFO_LOG("Pause find already paused");
778 return SUCCESS;
779 }
780 StreamStatus targetStatus = StreamStatus::STREAM_RUNNING;
781 if (!streamStatus_->compare_exchange_strong(targetStatus, StreamStatus::STREAM_PAUSING)) {
782 AUDIO_ERR_LOG("Pause failed, invalid status : %{public}s", GetStatusInfo(targetStatus).c_str());
783 return ERR_ILLEGAL_STATE;
784 }
785
786 if (processProxy_->Pause(isFlush) != SUCCESS) {
787 streamStatus_->store(StreamStatus::STREAM_RUNNING);
788 AUDIO_ERR_LOG("Pause failed to call process proxy, reset status to RUNNING");
789 threadStatusCV_.notify_all(); // avoid thread blocking with status PAUSING
790 return ERR_OPERATION_FAILED;
791 }
792 streamStatus_->store(StreamStatus::STREAM_PAUSED);
793
794 return SUCCESS;
795 }
796
Resume()797 int32_t AudioProcessInClientInner::Resume()
798 {
799 Trace traceWithLog("AudioProcessInClient::Resume", true);
800 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
801 std::lock_guard<std::mutex> lock(statusSwitchLock_);
802
803 if (streamStatus_->load() == StreamStatus::STREAM_RUNNING) {
804 AUDIO_INFO_LOG("Resume find already running");
805 return SUCCESS;
806 }
807
808 StreamStatus targetStatus = StreamStatus::STREAM_PAUSED;
809 if (!streamStatus_->compare_exchange_strong(targetStatus, StreamStatus::STREAM_STARTING)) {
810 AUDIO_ERR_LOG("Resume failed, invalid status : %{public}s", GetStatusInfo(targetStatus).c_str());
811 return ERR_ILLEGAL_STATE;
812 }
813
814 if (processProxy_->Resume() != SUCCESS) {
815 streamStatus_->store(StreamStatus::STREAM_PAUSED);
816 AUDIO_ERR_LOG("Resume failed to call process proxy, reset status to PAUSED.");
817 threadStatusCV_.notify_all();
818 return ERR_OPERATION_FAILED;
819 }
820 UpdateHandleInfo();
821 streamStatus_->store(StreamStatus::STREAM_RUNNING);
822 threadStatusCV_.notify_all();
823
824 return SUCCESS;
825 }
826
Stop()827 int32_t AudioProcessInClientInner::Stop()
828 {
829 Trace traceWithLog("AudioProcessInClient::Stop", true);
830 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
831 std::lock_guard<std::mutex> lock(statusSwitchLock_);
832 if (streamStatus_->load() == StreamStatus::STREAM_STOPPED) {
833 AUDIO_INFO_LOG("Stop find already stopped");
834 return SUCCESS;
835 }
836
837 StreamStatus oldStatus = streamStatus_->load();
838 if (oldStatus == STREAM_IDEL || oldStatus == STREAM_RELEASED || oldStatus == STREAM_INVALID) {
839 AUDIO_ERR_LOG("Stop failed, invalid status : %{public}s", GetStatusInfo(oldStatus).c_str());
840 return ERR_ILLEGAL_STATE;
841 }
842
843 streamStatus_->store(StreamStatus::STREAM_STOPPING);
844 if (processProxy_->Stop() != SUCCESS) {
845 streamStatus_->store(oldStatus);
846 AUDIO_ERR_LOG("Stop failed in server, reset status to %{public}s", GetStatusInfo(oldStatus).c_str());
847 threadStatusCV_.notify_all(); // avoid thread blocking with status RUNNING
848 return ERR_OPERATION_FAILED;
849 }
850 isCallbackLoopEnd_ = true;
851 threadStatusCV_.notify_all();
852
853 streamStatus_->store(StreamStatus::STREAM_STOPPED);
854 AUDIO_INFO_LOG("Success stop proc client mode %{public}d form %{public}s.",
855 processConfig_.audioMode, GetStatusInfo(oldStatus).c_str());
856 return SUCCESS;
857 }
858
Release()859 int32_t AudioProcessInClientInner::Release()
860 {
861 Trace traceWithLog("AudioProcessInClient::Release", true);
862 CHECK_AND_RETURN_RET_LOG(isInited_, ERR_ILLEGAL_STATE, "not inited!");
863
864 // not lock as status is already released
865 if (streamStatus_->load() == StreamStatus::STREAM_RELEASED) {
866 return SUCCESS;
867 }
868 Stop();
869 std::lock_guard<std::mutex> lock(statusSwitchLock_);
870 if (streamStatus_->load() != StreamStatus::STREAM_STOPPED ||
871 streamStatus_->load() != StreamStatus::STREAM_IDEL) {
872 AUDIO_ERR_LOG("Process status error:%{public}s", GetStatusInfo(streamStatus_->load()).c_str());
873 return ERR_ILLEGAL_STATE;
874 }
875
876 if (processProxy_->Release() != SUCCESS) {
877 AUDIO_ERR_LOG("Release may failed in server");
878 threadStatusCV_.notify_all(); // avoid thread blocking with status RUNNING
879 return ERR_OPERATION_FAILED;
880 }
881
882 streamStatus_->store(StreamStatus::STREAM_RELEASED);
883 AUDIO_INFO_LOG("Success release proc client mode %{public}d.", processConfig_.audioMode);
884 isInited_ = false;
885 return SUCCESS;
886 }
887
888 // client should call GetBufferDesc and Enqueue in OnHandleData
CallClientHandleCurrent()889 void AudioProcessInClientInner::CallClientHandleCurrent()
890 {
891 Trace trace("AudioProcessInClient::CallClientHandleCurrent");
892 std::shared_ptr<AudioDataCallback> cb = audioDataCallback_.lock();
893 if (cb == nullptr) {
894 AUDIO_ERR_LOG("%{public}s audio data callback is null.", __func__);
895 return;
896 }
897
898 cb->OnHandleData(spanSizeInByte_);
899 }
900
UpdateHandleInfo()901 void AudioProcessInClientInner::UpdateHandleInfo()
902 {
903 Trace traceSync("AudioProcessInClient::UpdateHandleInfo");
904 uint64_t serverHandlePos = 0;
905 int64_t serverHandleTime = 0;
906 int32_t ret = processProxy_->RequestHandleInfo();
907 CHECK_AND_RETURN_LOG(ret == SUCCESS, "RequestHandleInfo failed ret:%{public}d", ret);
908 audioBuffer_->GetHandleInfo(serverHandlePos, serverHandleTime);
909
910 bool isSuccess = handleTimeModel_.UpdataFrameStamp(serverHandlePos, serverHandleTime);
911 if (!isSuccess) {
912 handleTimeModel_.ResetFrameStamp(serverHandlePos, serverHandleTime);
913 }
914 }
915
GetPredictNextHandleTime(uint64_t posInFrame)916 int64_t AudioProcessInClientInner::GetPredictNextHandleTime(uint64_t posInFrame)
917 {
918 Trace trace("AudioProcessInClient::GetPredictNextRead");
919 uint64_t handleSpanCnt = posInFrame / spanSizeInFrame_;
920 uint32_t startPeriodCnt = 20; // sync each time when start
921 uint32_t oneBigPeriodCnt = 40; // 200ms
922 if (handleSpanCnt < startPeriodCnt || handleSpanCnt % oneBigPeriodCnt == 0) {
923 UpdateHandleInfo();
924 }
925
926 int64_t nextHandleTime = handleTimeModel_.GetTimeOfPos(posInFrame);
927
928 return nextHandleTime;
929 }
930
PrepareNext(uint64_t curHandPos,int64_t & wakeUpTime)931 bool AudioProcessInClientInner::PrepareNext(uint64_t curHandPos, int64_t &wakeUpTime)
932 {
933 Trace trace("AudioProcessInClient::PrepareNext " + std::to_string(curHandPos));
934 int64_t handleModifyTime = 0;
935 if (processConfig_.audioMode == AUDIO_MODE_RECORD) {
936 handleModifyTime = RECORD_HANDLE_DELAY_NANO;
937 } else {
938 handleModifyTime = -WRITE_BEFORE_DURATION_NANO;
939 }
940
941 int64_t nextServerHandleTime = GetPredictNextHandleTime(curHandPos) + handleModifyTime;
942 if (nextServerHandleTime < ClockTime::GetCurNano()) {
943 wakeUpTime = ClockTime::GetCurNano() + ONE_MILLISECOND_DURATION; // make sure less than duration
944 } else {
945 wakeUpTime = nextServerHandleTime;
946 }
947 AUDIO_DEBUG_LOG("%{public}s end, audioMode %{public}d, curReadPos %{public}" PRIu64", nextServerHandleTime "
948 "%{public}" PRId64" wakeUpTime %{public}" PRId64".", __func__, processConfig_.audioMode, curHandPos,
949 nextServerHandleTime, wakeUpTime);
950 return true;
951 }
952
GetStatusInfo(StreamStatus status)953 std::string AudioProcessInClientInner::GetStatusInfo(StreamStatus status)
954 {
955 switch (status) {
956 case STREAM_IDEL:
957 return "STREAM_IDEL";
958 case STREAM_STARTING:
959 return "STREAM_STARTING";
960 case STREAM_RUNNING:
961 return "STREAM_RUNNING";
962 case STREAM_PAUSING:
963 return "STREAM_PAUSING";
964 case STREAM_PAUSED:
965 return "STREAM_PAUSED";
966 case STREAM_STOPPING:
967 return "STREAM_STOPPING";
968 case STREAM_STOPPED:
969 return "STREAM_STOPPED";
970 case STREAM_RELEASED:
971 return "STREAM_RELEASED";
972 case STREAM_INVALID:
973 return "STREAM_INVALID";
974 default:
975 break;
976 }
977 return "NO_SUCH_STATUS";
978 }
979
KeepLoopRunning()980 bool AudioProcessInClientInner::KeepLoopRunning()
981 {
982 StreamStatus targetStatus = STREAM_INVALID;
983
984 switch (streamStatus_->load()) {
985 case STREAM_RUNNING:
986 return true;
987 case STREAM_STARTING:
988 targetStatus = STREAM_RUNNING;
989 break;
990 case STREAM_IDEL:
991 targetStatus = STREAM_STARTING;
992 break;
993 case STREAM_PAUSING:
994 targetStatus = STREAM_PAUSED;
995 break;
996 case STREAM_PAUSED:
997 targetStatus = STREAM_STARTING;
998 break;
999 case STREAM_STOPPING:
1000 targetStatus = STREAM_STOPPED;
1001 break;
1002 case STREAM_STOPPED:
1003 targetStatus = STREAM_RELEASED;
1004 break;
1005 default:
1006 break;
1007 }
1008
1009 Trace trace("AudioProcessInClient::InWaitStatus");
1010 std::unique_lock<std::mutex> lock(loopThreadLock_);
1011 AUDIO_DEBUG_LOG("Process status is %{public}s now, wait for %{public}s...",
1012 GetStatusInfo(streamStatus_->load()).c_str(), GetStatusInfo(targetStatus).c_str());
1013 threadStatus_ = WAITTING;
1014 threadStatusCV_.wait(lock);
1015 AUDIO_DEBUG_LOG("Process wait end. Cur is %{public}s now, target is %{public}s...",
1016 GetStatusInfo(streamStatus_->load()).c_str(), GetStatusInfo(targetStatus).c_str());
1017
1018 return false;
1019 }
1020
RecordProcessCallbackFuc()1021 void AudioProcessInClientInner::RecordProcessCallbackFuc()
1022 {
1023 AUDIO_INFO_LOG("%{public}s enter.", __func__);
1024 AudioSystemManager::GetInstance()->RequestThreadPriority(gettid());
1025 uint64_t curReadPos = 0;
1026 int64_t wakeUpTime = ClockTime::GetCurNano();
1027 int64_t clientReadCost = 0;
1028
1029 while (!isCallbackLoopEnd_ && audioBuffer_ != nullptr) {
1030 if (!KeepLoopRunning()) {
1031 continue;
1032 }
1033 threadStatus_ = INRUNNING;
1034 Trace traceLoop("AudioProcessInClient Record InRunning");
1035 if (needReSyncPosition_ && RecordReSyncServicePos() == SUCCESS) {
1036 wakeUpTime = ClockTime::GetCurNano();
1037 needReSyncPosition_ = false;
1038 continue;
1039 }
1040 int64_t curTime = ClockTime::GetCurNano();
1041 if (curTime - wakeUpTime > ONE_MILLISECOND_DURATION) {
1042 AUDIO_WARNING_LOG("%{public}s wake up too late.", __func__);
1043 wakeUpTime = curTime;
1044 }
1045
1046 curReadPos = audioBuffer_->GetCurReadFrame();
1047 if (RecordPrepareCurrent(curReadPos) != SUCCESS) {
1048 AUDIO_ERR_LOG("%{public}s prepare current fail.", __func__);
1049 continue;
1050 }
1051 CallClientHandleCurrent();
1052 if (RecordFinishHandleCurrent(curReadPos, clientReadCost) != SUCCESS) {
1053 AUDIO_ERR_LOG("%{public}s finish handle current fail.", __func__);
1054 continue;
1055 }
1056
1057 if (!PrepareNext(curReadPos, wakeUpTime)) {
1058 AUDIO_ERR_LOG("%{public}s prepare next loop in process fail.", __func__);
1059 break;
1060 }
1061
1062 threadStatus_ = SLEEPING;
1063 curTime = ClockTime::GetCurNano();
1064 if (wakeUpTime > curTime && wakeUpTime - curTime < MAX_READ_COST_DURATION_NANO + clientReadCost) {
1065 ClockTime::AbsoluteSleep(wakeUpTime);
1066 } else {
1067 Trace trace("RecordBigWakeUpTime");
1068 AUDIO_WARNING_LOG("%{public}s wakeUpTime is too late...", __func__);
1069 ClockTime::RelativeSleep(MAX_READ_COST_DURATION_NANO);
1070 }
1071 }
1072 }
1073
RecordReSyncServicePos()1074 int32_t AudioProcessInClientInner::RecordReSyncServicePos()
1075 {
1076 CHECK_AND_RETURN_RET_LOG(processProxy_ != nullptr && audioBuffer_ != nullptr, ERR_INVALID_HANDLE,
1077 "%{public}s process proxy or audio buffer is null.", __func__);
1078 uint64_t serverHandlePos = 0;
1079 int64_t serverHandleTime = 0;
1080 int32_t tryTimes = 3;
1081 int32_t ret = 0;
1082 while (tryTimes > 0) {
1083 ret = processProxy_->RequestHandleInfo();
1084 CHECK_AND_RETURN_RET_LOG(ret == SUCCESS, ret, "%{public}s request handle info fail, ret %{public}d.",
1085 __func__, ret);
1086
1087 CHECK_AND_RETURN_RET_LOG(audioBuffer_->GetHandleInfo(serverHandlePos, serverHandleTime), ERR_OPERATION_FAILED,
1088 "%{public}s get handle info fail.", __func__);
1089 if (serverHandlePos > 0) {
1090 break;
1091 }
1092 ClockTime::RelativeSleep(MAX_READ_COST_DURATION_NANO);
1093 tryTimes--;
1094 }
1095 AUDIO_INFO_LOG("%{public}s get handle info OK, tryTimes %{public}d, serverHandlePos %{public}" PRIu64", "
1096 "serverHandleTime %{public}" PRId64".", __func__, tryTimes, serverHandlePos, serverHandleTime);
1097 ClockTime::AbsoluteSleep(serverHandleTime + RECORD_HANDLE_DELAY_NANO);
1098
1099 ret = audioBuffer_->SetCurReadFrame(serverHandlePos);
1100 CHECK_AND_RETURN_RET_LOG(ret == SUCCESS, ret, "%{public}s set curReadPos fail, ret %{public}d.", __func__, ret);
1101 return SUCCESS;
1102 }
1103
RecordPrepareCurrent(uint64_t curReadPos)1104 int32_t AudioProcessInClientInner::RecordPrepareCurrent(uint64_t curReadPos)
1105 {
1106 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, ERR_INVALID_HANDLE,
1107 "%{public}s audio buffer is null.", __func__);
1108 SpanInfo *curReadSpan = audioBuffer_->GetSpanInfo(curReadPos);
1109 CHECK_AND_RETURN_RET_LOG(curReadSpan != nullptr, ERR_INVALID_HANDLE,
1110 "%{public}s get read span info of process client fail.", __func__);
1111
1112 int tryCount = 10;
1113 SpanStatus targetStatus = SpanStatus::SPAN_WRITE_DONE;
1114 while (!curReadSpan->spanStatus.compare_exchange_strong(targetStatus, SpanStatus::SPAN_READING)
1115 && tryCount > 0) {
1116 AUDIO_WARNING_LOG("%{public}s unready, curReadSpan %{public}" PRIu64", curSpanStatus %{public}d, wait 2ms.",
1117 __func__, curReadPos, curReadSpan->spanStatus.load());
1118 targetStatus = SpanStatus::SPAN_WRITE_DONE;
1119 tryCount--;
1120 ClockTime::RelativeSleep(RECORD_RESYNC_SLEEP_NANO);
1121 }
1122 CHECK_AND_RETURN_RET_LOG(tryCount > 0, ERR_INVALID_READ,
1123 "%{public}s wait too long, curReadSpan %{public}" PRIu64".", __func__, curReadPos);
1124
1125 curReadSpan->readStartTime = ClockTime::GetCurNano();
1126 return SUCCESS;
1127 }
1128
RecordFinishHandleCurrent(uint64_t & curReadPos,int64_t & clientReadCost)1129 int32_t AudioProcessInClientInner::RecordFinishHandleCurrent(uint64_t &curReadPos, int64_t &clientReadCost)
1130 {
1131 CHECK_AND_RETURN_RET_LOG(audioBuffer_ != nullptr, ERR_INVALID_HANDLE,
1132 "%{public}s audio buffer is null.", __func__);
1133 SpanInfo *curReadSpan = audioBuffer_->GetSpanInfo(curReadPos);
1134 CHECK_AND_RETURN_RET_LOG(curReadSpan != nullptr, ERR_INVALID_HANDLE,
1135 "%{public}s get read span info of process client fail.", __func__);
1136
1137 SpanStatus targetStatus = SpanStatus::SPAN_READING;
1138 if (!curReadSpan->spanStatus.compare_exchange_strong(targetStatus, SpanStatus::SPAN_READ_DONE)) {
1139 AUDIO_ERR_LOG("%{public}s status error, curReadSpan %{public}" PRIu64", curSpanStatus %{public}d.",
1140 __func__, curReadPos, curReadSpan->spanStatus.load());
1141 return ERR_INVALID_OPERATION;
1142 }
1143 curReadSpan->readDoneTime = ClockTime::GetCurNano();
1144
1145 clientReadCost = curReadSpan->readDoneTime - curReadSpan->readStartTime;
1146 if (clientReadCost > MAX_READ_COST_DURATION_NANO) {
1147 AUDIO_WARNING_LOG("Client write cost too long...");
1148 }
1149
1150 uint64_t nextWritePos = curReadPos + spanSizeInFrame_;
1151 int32_t ret = audioBuffer_->SetCurReadFrame(nextWritePos);
1152 CHECK_AND_RETURN_RET_LOG(ret == SUCCESS, ret, "%{public}s set next hand frame %{public}" PRIu64" fail, "
1153 "ret %{public}d.", __func__, nextWritePos, ret);
1154 curReadPos = nextWritePos;
1155
1156 return SUCCESS;
1157 }
1158
PrepareCurrent(uint64_t curWritePos)1159 bool AudioProcessInClientInner::PrepareCurrent(uint64_t curWritePos)
1160 {
1161 Trace trace("AudioProcessInClient::PrepareCurrent " + std::to_string(curWritePos));
1162 SpanInfo *tempSpan = audioBuffer_->GetSpanInfo(curWritePos);
1163 if (tempSpan == nullptr) {
1164 AUDIO_ERR_LOG("GetSpanInfo failed!");
1165 return false;
1166 }
1167
1168 int tryCount = 10; // try 10 * 2 = 20ms
1169 SpanStatus targetStatus = SpanStatus::SPAN_READ_DONE;
1170 while (!tempSpan->spanStatus.compare_exchange_strong(targetStatus, SpanStatus::SPAN_WRITTING) && tryCount-- > 0) {
1171 AUDIO_WARNING_LOG("current span %{public}" PRIu64" is not ready, status is %{public}d, wait 2ms.", curWritePos,
1172 targetStatus);
1173 targetStatus = SpanStatus::SPAN_READ_DONE;
1174 ClockTime::RelativeSleep(ONE_MILLISECOND_DURATION + ONE_MILLISECOND_DURATION);
1175 }
1176 // If the last attempt is successful, tryCount will be equal to zero.
1177 if (tryCount < 0) {
1178 AUDIO_ERR_LOG("wait on current span %{public}" PRIu64" too long...", curWritePos);
1179 return false;
1180 }
1181 tempSpan->writeStartTime = ClockTime::GetCurNano();
1182 return true;
1183 }
1184
FinishHandleCurrent(uint64_t & curWritePos,int64_t & clientWriteCost)1185 bool AudioProcessInClientInner::FinishHandleCurrent(uint64_t &curWritePos, int64_t &clientWriteCost)
1186 {
1187 Trace trace("AudioProcessInClient::FinishHandleCurrent " + std::to_string(curWritePos));
1188 SpanInfo *tempSpan = audioBuffer_->GetSpanInfo(curWritePos);
1189 if (tempSpan == nullptr) {
1190 AUDIO_ERR_LOG("GetSpanInfo failed!");
1191 return false;
1192 }
1193
1194 int32_t ret = ERROR;
1195 // mark status write-done and then server can read
1196 SpanStatus targetStatus = SpanStatus::SPAN_WRITTING;
1197 if (tempSpan->spanStatus.load() == targetStatus) {
1198 uint64_t nextWritePos = curWritePos + spanSizeInFrame_;
1199 ret = audioBuffer_->SetCurWriteFrame(nextWritePos); // move ahead before writedone
1200 curWritePos = nextWritePos;
1201 tempSpan->spanStatus.store(SpanStatus::SPAN_WRITE_DONE);
1202 }
1203 if (ret != SUCCESS) {
1204 AUDIO_ERR_LOG("SetCurWriteFrame %{public}" PRIu64" failed, ret:%{public}d", curWritePos, ret);
1205 return false;
1206 }
1207 tempSpan->writeDoneTime = ClockTime::GetCurNano();
1208 tempSpan->volumeStart = processVolume_;
1209 tempSpan->volumeEnd = processVolume_;
1210 clientWriteCost = tempSpan->writeDoneTime - tempSpan->writeStartTime;
1211 if (clientWriteCost > MAX_WRITE_COST_DURATION_NANO) {
1212 AUDIO_WARNING_LOG("Client write cost too long...");
1213 underflowCount_++;
1214 // todo
1215 // handle write time out: send underrun msg to client, reset time model with latest server handle time.
1216 }
1217
1218 return true;
1219 }
1220
ProcessCallbackFuc()1221 void AudioProcessInClientInner::ProcessCallbackFuc()
1222 {
1223 AUDIO_INFO_LOG("Callback loop start.");
1224 AudioSystemManager::GetInstance()->RequestThreadPriority(gettid());
1225
1226 uint64_t curWritePos = 0;
1227 int64_t curTime = 0;
1228 int64_t wakeUpTime = ClockTime::GetCurNano();
1229 int64_t clientWriteCost = 0;
1230
1231 while (!isCallbackLoopEnd_) {
1232 if (!KeepLoopRunning()) {
1233 continue;
1234 }
1235 threadStatus_ = INRUNNING;
1236 Trace traceLoop("AudioProcessInClient::InRunning");
1237 curTime = ClockTime::GetCurNano();
1238 if (curTime - wakeUpTime > ONE_MILLISECOND_DURATION) {
1239 AUDIO_WARNING_LOG("Wake up too late...");
1240 wakeUpTime = curTime;
1241 }
1242 curWritePos = audioBuffer_->GetCurWriteFrame();
1243 if (!PrepareCurrent(curWritePos)) {
1244 AUDIO_ERR_LOG("PrepareCurrent failed!");
1245 continue;
1246 }
1247 // call client write
1248 CallClientHandleCurrent();
1249
1250 // client write done, check if time out
1251 if (!FinishHandleCurrent(curWritePos, clientWriteCost)) {
1252 AUDIO_ERR_LOG("FinishHandleCurrent failed!");
1253 continue;
1254 }
1255
1256 // prepare next sleep
1257 if (!PrepareNext(curWritePos, wakeUpTime)) {
1258 AUDIO_ERR_LOG("PrepareNextLoop in process failed!");
1259 break;
1260 }
1261
1262 traceLoop.End();
1263 // start safe sleep
1264 threadStatus_ = SLEEPING;
1265 curTime = ClockTime::GetCurNano();
1266 if (wakeUpTime - curTime > MAX_WRITE_COST_DURATION_NANO + clientWriteCost) {
1267 Trace trace("BigWakeUpTime curTime[" + std::to_string(curTime) + "] target[" + std::to_string(wakeUpTime) +
1268 "] delay " + std::to_string(wakeUpTime - curTime) + "ns");
1269 AUDIO_WARNING_LOG("wakeUpTime is too late...");
1270 }
1271 ClockTime::AbsoluteSleep(wakeUpTime);
1272 }
1273 }
1274 } // namespace AudioStandard
1275 } // namespace OHOS
1276