Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "ringbufferpool.h"
19 : #include "ringbuffer.h"
20 : #include "logger.h"
21 :
22 : #include <cstring>
23 :
24 : namespace jami {
25 :
26 : const char* const RingBufferPool::DEFAULT_ID = "audiolayer_id";
27 :
28 39 : RingBufferPool::RingBufferPool()
29 78 : : defaultRingBuffer_(createRingBuffer(DEFAULT_ID))
30 39 : {}
31 :
32 39 : RingBufferPool::~RingBufferPool()
33 : {
34 39 : readBindingsMap_.clear();
35 39 : defaultRingBuffer_.reset();
36 :
37 : // Verify ringbuffer not removed yet
38 : // XXXX: With a good design this should never happen! :-P
39 159 : for (const auto& item : ringBufferMap_) {
40 120 : const auto& weak = item.second;
41 120 : if (not weak.expired())
42 0 : JAMI_WARNING("Leaking RingBuffer '{}'", item.first);
43 : }
44 39 : }
45 :
46 : void
47 0 : RingBufferPool::setInternalSamplingRate(unsigned sr)
48 : {
49 0 : std::lock_guard lk(stateLock_);
50 :
51 0 : if (sr != internalAudioFormat_.sample_rate) {
52 0 : flushAllBuffersLocked();
53 0 : internalAudioFormat_.sample_rate = sr;
54 : }
55 0 : }
56 :
57 : void
58 0 : RingBufferPool::setInternalAudioFormat(AudioFormat format)
59 : {
60 0 : std::lock_guard lk(stateLock_);
61 :
62 0 : if (format != internalAudioFormat_) {
63 0 : flushAllBuffersLocked();
64 0 : internalAudioFormat_ = format;
65 0 : for (auto& wrb : ringBufferMap_)
66 0 : if (auto rb = wrb.second.lock())
67 0 : rb->setFormat(internalAudioFormat_);
68 : }
69 0 : }
70 :
71 : std::shared_ptr<RingBuffer>
72 2951 : RingBufferPool::getRingBufferLocked(const std::string& id)
73 : {
74 2951 : const auto& it = ringBufferMap_.find(id);
75 2951 : if (it != ringBufferMap_.cend()) {
76 2453 : if (const auto& sptr = it->second.lock())
77 2453 : return sptr;
78 14 : ringBufferMap_.erase(it);
79 : }
80 :
81 512 : return nullptr;
82 : }
83 :
84 : std::shared_ptr<RingBuffer>
85 0 : RingBufferPool::getRingBufferLocked(const std::string& id) const
86 : {
87 0 : const auto& it = ringBufferMap_.find(id);
88 0 : if (it != ringBufferMap_.cend())
89 0 : return it->second.lock();
90 :
91 0 : return nullptr;
92 : }
93 :
94 : std::shared_ptr<RingBuffer>
95 36 : RingBufferPool::getRingBuffer(const std::string& id)
96 : {
97 36 : std::lock_guard lk(stateLock_);
98 72 : return getRingBufferLocked(id);
99 36 : }
100 :
101 : std::shared_ptr<RingBuffer>
102 0 : RingBufferPool::getRingBuffer(const std::string& id) const
103 : {
104 0 : std::lock_guard lk(stateLock_);
105 0 : return getRingBufferLocked(id);
106 0 : }
107 :
108 : std::shared_ptr<RingBuffer>
109 907 : RingBufferPool::createRingBuffer(const std::string& id)
110 : {
111 907 : std::lock_guard lk(stateLock_);
112 :
113 907 : auto rbuf = getRingBufferLocked(id);
114 907 : if (rbuf) {
115 1580 : JAMI_DEBUG("Ringbuffer already exists for id '{}'", id);
116 395 : return rbuf;
117 : }
118 :
119 512 : rbuf.reset(new RingBuffer(id, internalAudioFormat_));
120 512 : ringBufferMap_.emplace(id, std::weak_ptr<RingBuffer>(rbuf));
121 512 : return rbuf;
122 907 : }
123 :
124 : const RingBufferPool::ReadBindings*
125 51170 : RingBufferPool::getReadBindings(const std::string& ringbufferId) const
126 : {
127 51170 : const auto& iter = readBindingsMap_.find(ringbufferId);
128 51170 : return iter != readBindingsMap_.cend() ? &iter->second : nullptr;
129 : }
130 :
131 : RingBufferPool::ReadBindings*
132 53507 : RingBufferPool::getReadBindings(const std::string& ringbufferId)
133 : {
134 53507 : const auto& iter = readBindingsMap_.find(ringbufferId);
135 53507 : return iter != readBindingsMap_.cend() ? &iter->second : nullptr;
136 : }
137 :
138 : void
139 360 : RingBufferPool::removeReadBindings(const std::string& ringbufferId)
140 : {
141 360 : if (not readBindingsMap_.erase(ringbufferId))
142 0 : JAMI_ERROR("Ringbuffer {} does not exist!", ringbufferId);
143 360 : }
144 :
145 : void
146 1004 : RingBufferPool::addReaderToRingBuffer(const std::shared_ptr<RingBuffer>& sourceBuffer, const std::string& readerBufferId)
147 : {
148 1004 : if (readerBufferId != DEFAULT_ID and sourceBuffer->getId() == readerBufferId)
149 40 : JAMI_WARNING("RingBuffer has a readoffset on itself");
150 :
151 1004 : sourceBuffer->createReadOffset(readerBufferId);
152 1004 : readBindingsMap_[readerBufferId].insert(sourceBuffer);
153 1004 : }
154 :
155 : void
156 1085 : RingBufferPool::removeReaderFromRingBuffer(const std::shared_ptr<RingBuffer>& sourceBuffer,
157 : const std::string& readerBufferId)
158 : {
159 1085 : if (auto* bindings = getReadBindings(readerBufferId)) {
160 983 : bindings->erase(sourceBuffer);
161 983 : if (bindings->empty())
162 360 : removeReadBindings(readerBufferId);
163 : }
164 :
165 1085 : sourceBuffer->removeReadOffset(readerBufferId);
166 1085 : }
167 :
168 : void
169 388 : RingBufferPool::bindRingBuffers(const std::string& ringbufferId1, const std::string& ringbufferId2)
170 : {
171 1552 : JAMI_LOG("Bind ringbuffer {} to ringbuffer {}", ringbufferId1, ringbufferId2);
172 :
173 388 : std::lock_guard lk(stateLock_);
174 :
175 388 : const auto& rb1 = getRingBufferLocked(ringbufferId1);
176 388 : if (not rb1) {
177 0 : JAMI_ERROR("No ringbuffer associated with id '{}'", ringbufferId1);
178 0 : return;
179 : }
180 :
181 388 : const auto& rb2 = getRingBufferLocked(ringbufferId2);
182 388 : if (not rb2) {
183 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId2);
184 0 : return;
185 : }
186 :
187 388 : addReaderToRingBuffer(rb1, ringbufferId2);
188 388 : addReaderToRingBuffer(rb2, ringbufferId1);
189 388 : }
190 :
191 : void
192 228 : RingBufferPool::bindHalfDuplexOut(const std::string& readerBufferId, const std::string& sourceBufferId)
193 : {
194 : /* This method is used only for active ringbuffers, if this ringbuffer does not exist,
195 : * do nothing */
196 228 : std::lock_guard lk(stateLock_);
197 :
198 228 : if (const auto& rb = getRingBufferLocked(sourceBufferId)) {
199 : // p1 est le binding de p2 (p2 lit le stream de p1)
200 228 : addReaderToRingBuffer(rb, readerBufferId);
201 228 : }
202 228 : }
203 :
204 : void
205 71 : RingBufferPool::unbindRingBuffers(const std::string& ringbufferId1, const std::string& ringbufferId2)
206 : {
207 284 : JAMI_LOG("Unbind ringbuffers {} and {}", ringbufferId1, ringbufferId2);
208 :
209 71 : std::lock_guard lk(stateLock_);
210 :
211 71 : const auto& rb1 = getRingBufferLocked(ringbufferId1);
212 71 : if (not rb1) {
213 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId1);
214 0 : return;
215 : }
216 :
217 71 : const auto& rb2 = getRingBufferLocked(ringbufferId2);
218 71 : if (not rb2) {
219 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId2);
220 0 : return;
221 : }
222 :
223 71 : removeReaderFromRingBuffer(rb1, ringbufferId2);
224 71 : removeReaderFromRingBuffer(rb2, ringbufferId1);
225 71 : }
226 :
227 : void
228 239 : RingBufferPool::unBindHalfDuplexOut(const std::string& readerBufferId, const std::string& sourceBufferId)
229 : {
230 239 : std::lock_guard lk(stateLock_);
231 :
232 239 : if (const auto& rb = getRingBufferLocked(sourceBufferId))
233 239 : removeReaderFromRingBuffer(rb, readerBufferId);
234 239 : }
235 :
236 : void
237 69 : RingBufferPool::unBindAllHalfDuplexOut(const std::string& ringbufferId)
238 : {
239 69 : std::lock_guard lk(stateLock_);
240 :
241 69 : const auto& rb = getRingBufferLocked(ringbufferId);
242 69 : if (not rb) {
243 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId);
244 0 : return;
245 : }
246 :
247 69 : auto* bindings = getReadBindings(ringbufferId);
248 69 : if (not bindings)
249 54 : return;
250 15 : const auto bindings_copy = *bindings; // temporary copy
251 48 : for (const auto& rbuf : bindings_copy) {
252 33 : removeReaderFromRingBuffer(rb, rbuf->getId());
253 : }
254 123 : }
255 :
256 : void
257 96 : RingBufferPool::unBindAllHalfDuplexIn(const std::string& sourceBufferId)
258 : {
259 96 : std::lock_guard lk(stateLock_);
260 :
261 96 : auto ringBuffer = getRingBufferLocked(sourceBufferId);
262 96 : if (not ringBuffer) {
263 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", sourceBufferId);
264 0 : return;
265 : }
266 :
267 96 : const std::vector<std::string>& subscribers = ringBuffer->getSubscribers();
268 259 : for (const auto& subscriber : subscribers) {
269 163 : removeReaderFromRingBuffer(ringBuffer, subscriber);
270 : }
271 96 : }
272 :
273 : void
274 458 : RingBufferPool::unBindAll(const std::string& ringbufferId)
275 : {
276 1832 : JAMI_LOG("Unbind ringbuffer {} from all bound ringbuffers", ringbufferId);
277 :
278 458 : std::lock_guard lk(stateLock_);
279 :
280 458 : const auto& rb = getRingBufferLocked(ringbufferId);
281 458 : if (not rb) {
282 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId);
283 0 : return;
284 : }
285 :
286 458 : auto* bindings = getReadBindings(ringbufferId);
287 458 : if (not bindings)
288 227 : return;
289 :
290 231 : const auto bindings_copy = *bindings; // temporary copy
291 485 : for (const auto& rbuf : bindings_copy) {
292 254 : removeReaderFromRingBuffer(rbuf, ringbufferId);
293 254 : removeReaderFromRingBuffer(rb, rbuf->getId());
294 : }
295 685 : }
296 :
297 : std::shared_ptr<AudioFrame>
298 51170 : RingBufferPool::getData(const std::string& ringbufferId)
299 : {
300 51170 : std::lock_guard lk(stateLock_);
301 :
302 51170 : auto* const bindings = getReadBindings(ringbufferId);
303 51170 : if (not bindings)
304 6253 : return {};
305 :
306 : // No mixing
307 44917 : if (bindings->size() == 1)
308 35166 : return (*bindings->cbegin())->get(ringbufferId);
309 :
310 9751 : auto mixBuffer = std::make_shared<AudioFrame>(internalAudioFormat_);
311 9751 : auto mixed = false;
312 33178 : for (const auto& rbuf : *bindings) {
313 23427 : if (auto b = rbuf->get(ringbufferId)) {
314 0 : mixed = true;
315 0 : mixBuffer->mix(*b);
316 :
317 : // voice is true if any of mixed frames has voice
318 0 : mixBuffer->has_voice |= b->has_voice;
319 23427 : }
320 : }
321 :
322 9751 : return mixed ? mixBuffer : nullptr;
323 51170 : }
324 :
325 : bool
326 0 : RingBufferPool::waitForDataAvailable(const std::string& ringbufferId, const duration& max_wait) const
327 : {
328 0 : return waitForDataAvailable(ringbufferId, clock::now() + max_wait);
329 : }
330 :
331 : bool
332 51170 : RingBufferPool::waitForDataAvailable(const std::string& ringbufferId, const time_point& deadline) const
333 : {
334 51170 : std::unique_lock lk(stateLock_);
335 51170 : const auto* bindings = getReadBindings(ringbufferId);
336 51170 : if (not bindings)
337 6127 : return false;
338 45043 : const auto bindings_copy = *bindings; // temporary copy
339 :
340 45043 : lk.unlock();
341 45043 : for (const auto& rbuf : bindings_copy) {
342 45043 : if (rbuf->waitForDataAvailable(ringbufferId, deadline) == 0)
343 45043 : return false;
344 : }
345 0 : return true;
346 51170 : }
347 :
348 : std::shared_ptr<AudioFrame>
349 0 : RingBufferPool::getAvailableData(const std::string& ringbufferId)
350 : {
351 0 : std::lock_guard lk(stateLock_);
352 :
353 0 : auto* bindings = getReadBindings(ringbufferId);
354 0 : if (not bindings)
355 0 : return {};
356 :
357 : // No mixing
358 0 : if (bindings->size() == 1) {
359 0 : return (*bindings->cbegin())->get(ringbufferId);
360 : }
361 :
362 0 : size_t availableFrames = 0;
363 :
364 0 : for (const auto& rbuf : *bindings)
365 0 : availableFrames = std::min(availableFrames, rbuf->availableForGet(ringbufferId));
366 :
367 0 : if (availableFrames == 0)
368 0 : return {};
369 :
370 0 : auto buf = std::make_shared<AudioFrame>(internalAudioFormat_);
371 0 : for (const auto& rbuf : *bindings) {
372 0 : if (auto b = rbuf->get(ringbufferId)) {
373 0 : buf->mix(*b);
374 :
375 : // voice is true if any of mixed frames has voice
376 0 : buf->has_voice |= b->has_voice;
377 0 : }
378 : }
379 :
380 0 : return buf;
381 0 : }
382 :
383 : size_t
384 0 : RingBufferPool::availableForGet(const std::string& ringbufferId) const
385 : {
386 0 : std::lock_guard lk(stateLock_);
387 :
388 0 : const auto* const bindings = getReadBindings(ringbufferId);
389 0 : if (not bindings)
390 0 : return 0;
391 :
392 : // No mixing
393 0 : if (bindings->size() == 1) {
394 0 : return (*bindings->begin())->availableForGet(ringbufferId);
395 : }
396 :
397 0 : size_t availableSamples = std::numeric_limits<size_t>::max();
398 :
399 0 : for (const auto& rbuf : *bindings) {
400 0 : const size_t nbSamples = rbuf->availableForGet(ringbufferId);
401 0 : if (nbSamples != 0)
402 0 : availableSamples = std::min(availableSamples, nbSamples);
403 : }
404 :
405 0 : return availableSamples != std::numeric_limits<size_t>::max() ? availableSamples : 0;
406 0 : }
407 :
408 : size_t
409 0 : RingBufferPool::discard(size_t toDiscard, const std::string& ringbufferId)
410 : {
411 0 : std::lock_guard lk(stateLock_);
412 :
413 0 : auto* const bindings = getReadBindings(ringbufferId);
414 0 : if (not bindings)
415 0 : return 0;
416 :
417 0 : for (const auto& rbuf : *bindings)
418 0 : rbuf->discard(toDiscard, ringbufferId);
419 :
420 0 : return toDiscard;
421 0 : }
422 :
423 : void
424 725 : RingBufferPool::flush(const std::string& ringbufferId)
425 : {
426 725 : std::lock_guard lk(stateLock_);
427 :
428 725 : auto* const bindings = getReadBindings(ringbufferId);
429 725 : if (not bindings)
430 220 : return;
431 :
432 2168 : for (const auto& rbuf : *bindings)
433 1663 : rbuf->flush(ringbufferId);
434 725 : }
435 :
436 : void
437 305 : RingBufferPool::flushAllBuffersLocked()
438 : {
439 2986 : for (auto item = ringBufferMap_.begin(); item != ringBufferMap_.end();) {
440 2681 : if (const auto rb = item->second.lock()) {
441 2303 : rb->flushAll();
442 2303 : ++item;
443 : } else {
444 : // Use this version of erase to avoid using invalidated iterator
445 378 : item = ringBufferMap_.erase(item);
446 2681 : }
447 : }
448 305 : }
449 :
450 : void
451 305 : RingBufferPool::flushAllBuffers()
452 : {
453 305 : std::lock_guard lk(stateLock_);
454 305 : flushAllBuffersLocked();
455 305 : }
456 :
457 : bool
458 0 : RingBufferPool::isAudioMeterActive(const std::string& id)
459 : {
460 0 : std::lock_guard lk(stateLock_);
461 0 : if (!id.empty()) {
462 0 : if (auto rb = getRingBufferLocked(id)) {
463 0 : return rb->isAudioMeterActive();
464 0 : }
465 : } else {
466 0 : for (auto item = ringBufferMap_.begin(); item != ringBufferMap_.end(); ++item) {
467 0 : if (const auto rb = item->second.lock()) {
468 0 : if (rb->isAudioMeterActive()) {
469 0 : return true;
470 : }
471 0 : }
472 : }
473 : }
474 0 : return false;
475 0 : }
476 :
477 : void
478 0 : RingBufferPool::setAudioMeterState(const std::string& id, bool state)
479 : {
480 0 : std::lock_guard lk(stateLock_);
481 0 : if (!id.empty()) {
482 0 : if (auto rb = getRingBufferLocked(id)) {
483 0 : rb->setAudioMeterState(state);
484 0 : }
485 : } else {
486 0 : for (auto item = ringBufferMap_.begin(); item != ringBufferMap_.end(); ++item) {
487 0 : if (const auto rb = item->second.lock()) {
488 0 : rb->setAudioMeterState(state);
489 0 : }
490 : }
491 : }
492 0 : }
493 :
494 : } // namespace jami
|
