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 144 : for (const auto& item : ringBufferMap_) {
40 105 : const auto& weak = item.second;
41 105 : 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 1142 : RingBufferPool::getRingBufferLocked(const std::string& id)
73 : {
74 1142 : const auto& it = ringBufferMap_.find(id);
75 1142 : if (it != ringBufferMap_.cend()) {
76 922 : if (const auto& sptr = it->second.lock())
77 922 : return sptr;
78 14 : ringBufferMap_.erase(it);
79 : }
80 :
81 234 : 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 418 : RingBufferPool::createRingBuffer(const std::string& id)
110 : {
111 418 : std::lock_guard lk(stateLock_);
112 :
113 418 : auto rbuf = getRingBufferLocked(id);
114 418 : if (rbuf) {
115 184 : JAMI_DEBUG("Ringbuffer already exists for id '{}'", id);
116 184 : return rbuf;
117 : }
118 :
119 234 : rbuf.reset(new RingBuffer(id, internalAudioFormat_));
120 234 : ringBufferMap_.emplace(id, std::weak_ptr<RingBuffer>(rbuf));
121 234 : return rbuf;
122 418 : }
123 :
124 : const RingBufferPool::ReadBindings*
125 23753 : RingBufferPool::getReadBindings(const std::string& ringbufferId) const
126 : {
127 23753 : const auto& iter = readBindingsMap_.find(ringbufferId);
128 23753 : return iter != readBindingsMap_.cend() ? &iter->second : nullptr;
129 : }
130 :
131 : RingBufferPool::ReadBindings*
132 24447 : RingBufferPool::getReadBindings(const std::string& ringbufferId)
133 : {
134 24447 : const auto& iter = readBindingsMap_.find(ringbufferId);
135 24447 : return iter != readBindingsMap_.cend() ? &iter->second : nullptr;
136 : }
137 :
138 : void
139 144 : RingBufferPool::removeReadBindings(const std::string& ringbufferId)
140 : {
141 144 : if (not readBindingsMap_.erase(ringbufferId))
142 0 : JAMI_ERROR("Ringbuffer {} does not exist!", ringbufferId);
143 144 : }
144 :
145 : void
146 336 : RingBufferPool::addReaderToRingBuffer(const std::shared_ptr<RingBuffer>& sourceBuffer, const std::string& readerBufferId)
147 : {
148 336 : if (readerBufferId != DEFAULT_ID and sourceBuffer->getId() == readerBufferId)
149 10 : JAMI_WARNING("RingBuffer has a readoffset on itself");
150 :
151 336 : sourceBuffer->createReadOffset(readerBufferId);
152 336 : readBindingsMap_[readerBufferId].insert(sourceBuffer);
153 336 : }
154 :
155 : void
156 342 : RingBufferPool::removeReaderFromRingBuffer(const std::shared_ptr<RingBuffer>& sourceBuffer,
157 : const std::string& readerBufferId)
158 : {
159 342 : if (auto* bindings = getReadBindings(readerBufferId)) {
160 298 : bindings->erase(sourceBuffer);
161 298 : if (bindings->empty())
162 144 : removeReadBindings(readerBufferId);
163 : }
164 :
165 342 : sourceBuffer->removeReadOffset(readerBufferId);
166 342 : }
167 :
168 : void
169 109 : RingBufferPool::bindRingBuffers(const std::string& ringbufferId1, const std::string& ringbufferId2)
170 : {
171 109 : JAMI_LOG("Bind ringbuffer {} to ringbuffer {}", ringbufferId1, ringbufferId2);
172 :
173 109 : std::lock_guard lk(stateLock_);
174 :
175 109 : const auto& rb1 = getRingBufferLocked(ringbufferId1);
176 109 : if (not rb1) {
177 0 : JAMI_ERROR("No ringbuffer associated with id '{}'", ringbufferId1);
178 0 : return;
179 : }
180 :
181 109 : const auto& rb2 = getRingBufferLocked(ringbufferId2);
182 109 : if (not rb2) {
183 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId2);
184 0 : return;
185 : }
186 :
187 109 : addReaderToRingBuffer(rb1, ringbufferId2);
188 109 : addReaderToRingBuffer(rb2, ringbufferId1);
189 109 : }
190 :
191 : void
192 118 : 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 118 : std::lock_guard lk(stateLock_);
197 :
198 118 : if (const auto& rb = getRingBufferLocked(sourceBufferId)) {
199 : // p1 est le binding de p2 (p2 lit le stream de p1)
200 118 : addReaderToRingBuffer(rb, readerBufferId);
201 118 : }
202 118 : }
203 :
204 : void
205 11 : RingBufferPool::unbindRingBuffers(const std::string& ringbufferId1, const std::string& ringbufferId2)
206 : {
207 11 : JAMI_LOG("Unbind ringbuffers {} and {}", ringbufferId1, ringbufferId2);
208 :
209 11 : std::lock_guard lk(stateLock_);
210 :
211 11 : const auto& rb1 = getRingBufferLocked(ringbufferId1);
212 11 : if (not rb1) {
213 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId1);
214 0 : return;
215 : }
216 :
217 11 : const auto& rb2 = getRingBufferLocked(ringbufferId2);
218 11 : if (not rb2) {
219 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId2);
220 0 : return;
221 : }
222 :
223 11 : removeReaderFromRingBuffer(rb1, ringbufferId2);
224 11 : removeReaderFromRingBuffer(rb2, ringbufferId1);
225 11 : }
226 :
227 : void
228 133 : RingBufferPool::unBindHalfDuplexOut(const std::string& readerBufferId, const std::string& sourceBufferId)
229 : {
230 133 : std::lock_guard lk(stateLock_);
231 :
232 134 : if (const auto& rb = getRingBufferLocked(sourceBufferId))
233 134 : removeReaderFromRingBuffer(rb, readerBufferId);
234 134 : }
235 :
236 : void
237 11 : RingBufferPool::unBindAllHalfDuplexOut(const std::string& ringbufferId)
238 : {
239 11 : std::lock_guard lk(stateLock_);
240 :
241 11 : const auto& rb = getRingBufferLocked(ringbufferId);
242 11 : if (not rb) {
243 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId);
244 0 : return;
245 : }
246 :
247 11 : auto* bindings = getReadBindings(ringbufferId);
248 11 : if (not bindings)
249 9 : return;
250 2 : const auto bindings_copy = *bindings; // temporary copy
251 5 : for (const auto& rbuf : bindings_copy) {
252 3 : removeReaderFromRingBuffer(rb, rbuf->getId());
253 : }
254 20 : }
255 :
256 : void
257 19 : RingBufferPool::unBindAllHalfDuplexIn(const std::string& sourceBufferId)
258 : {
259 19 : std::lock_guard lk(stateLock_);
260 :
261 19 : auto ringBuffer = getRingBufferLocked(sourceBufferId);
262 19 : if (not ringBuffer) {
263 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", sourceBufferId);
264 0 : return;
265 : }
266 :
267 19 : const std::vector<std::string>& subscribers = ringBuffer->getSubscribers();
268 34 : for (const auto& subscriber : subscribers) {
269 15 : removeReaderFromRingBuffer(ringBuffer, subscriber);
270 : }
271 19 : }
272 :
273 : void
274 166 : RingBufferPool::unBindAll(const std::string& ringbufferId)
275 : {
276 166 : JAMI_LOG("Unbind ringbuffer {} from all bound ringbuffers", ringbufferId);
277 :
278 166 : std::lock_guard lk(stateLock_);
279 :
280 166 : const auto& rb = getRingBufferLocked(ringbufferId);
281 166 : if (not rb) {
282 0 : JAMI_ERROR("No ringbuffer associated to id '{}'", ringbufferId);
283 0 : return;
284 : }
285 :
286 166 : auto* bindings = getReadBindings(ringbufferId);
287 166 : if (not bindings)
288 83 : return;
289 :
290 83 : const auto bindings_copy = *bindings; // temporary copy
291 167 : for (const auto& rbuf : bindings_copy) {
292 84 : removeReaderFromRingBuffer(rbuf, ringbufferId);
293 84 : removeReaderFromRingBuffer(rb, rbuf->getId());
294 : }
295 249 : }
296 :
297 : std::shared_ptr<AudioFrame>
298 23753 : RingBufferPool::getData(const std::string& ringbufferId)
299 : {
300 23753 : std::lock_guard lk(stateLock_);
301 :
302 23753 : auto* const bindings = getReadBindings(ringbufferId);
303 23753 : if (not bindings)
304 1033 : return {};
305 :
306 : // No mixing
307 22720 : if (bindings->size() == 1)
308 22720 : return (*bindings->cbegin())->get(ringbufferId);
309 :
310 0 : auto mixBuffer = std::make_shared<AudioFrame>(internalAudioFormat_);
311 0 : auto mixed = false;
312 0 : for (const auto& rbuf : *bindings) {
313 0 : 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 0 : }
320 : }
321 :
322 0 : return mixed ? mixBuffer : nullptr;
323 0 : }
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 23753 : RingBufferPool::waitForDataAvailable(const std::string& ringbufferId, const time_point& deadline) const
333 : {
334 23753 : std::unique_lock lk(stateLock_);
335 23753 : const auto* bindings = getReadBindings(ringbufferId);
336 23753 : if (not bindings)
337 1004 : return false;
338 22749 : const auto bindings_copy = *bindings; // temporary copy
339 :
340 22749 : lk.unlock();
341 22749 : for (const auto& rbuf : bindings_copy) {
342 22749 : if (rbuf->waitForDataAvailable(ringbufferId, deadline) == 0)
343 22749 : return false;
344 : }
345 0 : return true;
346 23753 : }
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 175 : RingBufferPool::flush(const std::string& ringbufferId)
425 : {
426 175 : std::lock_guard lk(stateLock_);
427 :
428 175 : auto* const bindings = getReadBindings(ringbufferId);
429 175 : if (not bindings)
430 67 : return;
431 :
432 266 : for (const auto& rbuf : *bindings)
433 158 : rbuf->flush(ringbufferId);
434 175 : }
435 :
436 : void
437 125 : RingBufferPool::flushAllBuffersLocked()
438 : {
439 855 : for (auto item = ringBufferMap_.begin(); item != ringBufferMap_.end();) {
440 730 : if (const auto rb = item->second.lock()) {
441 615 : rb->flushAll();
442 615 : ++item;
443 : } else {
444 : // Use this version of erase to avoid using invalidated iterator
445 115 : item = ringBufferMap_.erase(item);
446 730 : }
447 : }
448 125 : }
449 :
450 : void
451 125 : RingBufferPool::flushAllBuffers()
452 : {
453 125 : std::lock_guard lk(stateLock_);
454 125 : flushAllBuffersLocked();
455 125 : }
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
|
