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 "ringbuffer.h"
19 : #include "logger.h"
20 : #include "client/jami_signal.h"
21 : #include "media_buffer.h"
22 : #include "libav_deps.h"
23 :
24 : #include <chrono>
25 : #include <cstdlib>
26 : #include <cstring>
27 : #include <algorithm>
28 :
29 : namespace jami {
30 :
31 : static constexpr const int RMS_SIGNAL_INTERVAL = 5;
32 :
33 270 : RingBuffer::RingBuffer(const std::string& rbuf_id, AudioFormat format)
34 270 : : id(rbuf_id)
35 270 : , endPos_(0)
36 270 : , format_(format)
37 270 : , lock_()
38 270 : , not_empty_()
39 270 : , readoffsets_()
40 270 : , resizer_(format_, static_cast<int>(format_.sample_rate) / 50, [this](std::shared_ptr<AudioFrame>&& frame) {
41 0 : putToBuffer(std::move(frame));
42 1620 : })
43 : {
44 270 : JAMI_LOG("Create new RingBuffer {}", id);
45 270 : }
46 :
47 540 : RingBuffer::~RingBuffer()
48 : {
49 270 : JAMI_LOG("Destroy RingBuffer {}", id);
50 270 : }
51 :
52 : void
53 158 : RingBuffer::flush(const std::string& ringbufferId)
54 : {
55 158 : storeReadOffset(endPos_, ringbufferId);
56 158 : }
57 :
58 : void
59 837 : RingBuffer::flushAll()
60 : {
61 1594 : for (auto& offset : readoffsets_)
62 757 : offset.second.offset = endPos_;
63 837 : }
64 :
65 : std::vector<std::string>
66 19 : RingBuffer::getSubscribers()
67 : {
68 19 : std::vector<std::string> subscribers;
69 34 : for (const auto& offset : readoffsets_) {
70 15 : subscribers.push_back(offset.first);
71 : }
72 19 : return subscribers;
73 0 : }
74 :
75 : size_t
76 0 : RingBuffer::putLength() const
77 : {
78 0 : const size_t buffer_size = buffer_.size();
79 0 : if (buffer_size == 0)
80 0 : return 0;
81 0 : const size_t startPos = getSmallestReadOffset();
82 0 : return (endPos_ + buffer_size - startPos) % buffer_size;
83 : }
84 :
85 : size_t
86 5473 : RingBuffer::getLength(const std::string& ringbufferId) const
87 : {
88 5473 : const size_t buffer_size = buffer_.size();
89 5473 : if (buffer_size == 0)
90 0 : return 0;
91 5473 : return (endPos_ + buffer_size - getReadOffset(ringbufferId)) % buffer_size;
92 : }
93 :
94 : void
95 0 : RingBuffer::debug()
96 : {
97 0 : JAMI_LOG("Start={}; End={}; BufferSize={}", getSmallestReadOffset(), endPos_, buffer_.size());
98 0 : }
99 :
100 : size_t
101 5473 : RingBuffer::getReadOffset(const std::string& ringbufferId) const
102 : {
103 5473 : auto iter = readoffsets_.find(ringbufferId);
104 5473 : return (iter != readoffsets_.end()) ? iter->second.offset : 0;
105 : }
106 :
107 : size_t
108 0 : RingBuffer::getSmallestReadOffset() const
109 : {
110 0 : if (hasNoReadOffsets())
111 0 : return 0;
112 0 : size_t smallest = buffer_.size();
113 0 : for (auto const& iter : readoffsets_)
114 0 : smallest = std::min(smallest, iter.second.offset);
115 0 : return smallest;
116 : }
117 :
118 : void
119 158 : RingBuffer::storeReadOffset(size_t offset, const std::string& ringbufferId)
120 : {
121 158 : ReadOffsetMap::iterator iter = readoffsets_.find(ringbufferId);
122 :
123 158 : if (iter != readoffsets_.end())
124 158 : iter->second.offset = offset;
125 : else
126 0 : JAMI_ERROR("RingBuffer::storeReadOffset() failed: unknown ringbuffer '{}'", ringbufferId);
127 158 : }
128 :
129 : void
130 372 : RingBuffer::createReadOffset(const std::string& ringbufferId)
131 : {
132 372 : std::lock_guard l(lock_);
133 372 : if (!hasThisReadOffset(ringbufferId))
134 289 : readoffsets_.emplace(ringbufferId, ReadOffset {endPos_, {}});
135 372 : }
136 :
137 : void
138 342 : RingBuffer::removeReadOffset(const std::string& ringbufferId)
139 : {
140 342 : std::lock_guard l(lock_);
141 342 : auto iter = readoffsets_.find(ringbufferId);
142 342 : if (iter != readoffsets_.end())
143 252 : readoffsets_.erase(iter);
144 342 : }
145 :
146 : bool
147 372 : RingBuffer::hasThisReadOffset(const std::string& ringbufferId) const
148 : {
149 372 : return readoffsets_.find(ringbufferId) != readoffsets_.end();
150 : }
151 :
152 : bool
153 0 : RingBuffer::hasNoReadOffsets() const
154 : {
155 0 : return readoffsets_.empty();
156 : }
157 :
158 : //
159 : // For the writer only:
160 : //
161 :
162 : void
163 0 : RingBuffer::put(std::shared_ptr<AudioFrame>&& data)
164 : {
165 0 : std::lock_guard l(writeLock_);
166 0 : resizer_.enqueue(resampler_.resample(std::move(data), format_));
167 0 : }
168 :
169 : // This one puts some data inside the ring buffer.
170 : void
171 0 : RingBuffer::putToBuffer(std::shared_ptr<AudioFrame>&& data)
172 : {
173 0 : std::lock_guard l(lock_);
174 0 : const size_t buffer_size = buffer_.size();
175 0 : if (buffer_size == 0)
176 0 : return;
177 :
178 0 : size_t len = buffer_size - putLength();
179 0 : if (len == 0)
180 0 : discard(1);
181 :
182 0 : size_t pos = endPos_;
183 :
184 0 : buffer_[pos] = std::move(data);
185 0 : const auto& newBuf = buffer_[pos];
186 0 : pos = (pos + 1) % buffer_size;
187 :
188 0 : endPos_ = pos;
189 :
190 0 : if (rmsSignal_) {
191 0 : ++rmsFrameCount_;
192 0 : rmsLevel_ += newBuf->calcRMS();
193 0 : if (rmsFrameCount_ == RMS_SIGNAL_INTERVAL) {
194 0 : emitSignal<libjami::AudioSignal::AudioMeter>(id, rmsLevel_ / RMS_SIGNAL_INTERVAL);
195 0 : rmsLevel_ = 0;
196 0 : rmsFrameCount_ = 0;
197 : }
198 : }
199 :
200 0 : for (auto& offset : readoffsets_) {
201 0 : if (offset.second.callback)
202 0 : offset.second.callback(newBuf);
203 : }
204 :
205 0 : not_empty_.notify_all();
206 0 : }
207 :
208 : //
209 : // For the reader only:
210 : //
211 :
212 : size_t
213 0 : RingBuffer::availableForGet(const std::string& ringbufferId) const
214 : {
215 : // Used space
216 0 : return getLength(ringbufferId);
217 : }
218 :
219 : std::shared_ptr<AudioFrame>
220 22720 : RingBuffer::get(const std::string& ringbufferId)
221 : {
222 22720 : std::lock_guard l(lock_);
223 :
224 22720 : auto offset = readoffsets_.find(ringbufferId);
225 22720 : if (offset == readoffsets_.end())
226 0 : return {};
227 :
228 22720 : const size_t buffer_size = buffer_.size();
229 22720 : if (buffer_size == 0)
230 0 : return {};
231 :
232 22720 : size_t startPos = offset->second.offset;
233 22720 : size_t len = (endPos_ + buffer_size - startPos) % buffer_size;
234 22720 : if (len == 0)
235 22720 : return {};
236 :
237 0 : auto ret = buffer_[startPos];
238 0 : offset->second.offset = (startPos + 1) % buffer_size;
239 0 : return ret;
240 0 : }
241 :
242 : size_t
243 22749 : RingBuffer::waitForDataAvailable(const std::string& ringbufferId, const time_point& deadline) const
244 : {
245 22749 : std::unique_lock l(lock_);
246 :
247 22749 : if (buffer_.empty())
248 0 : return 0;
249 22749 : if (readoffsets_.find(ringbufferId) == readoffsets_.end())
250 0 : return 0;
251 :
252 22749 : size_t getl = 0;
253 45498 : auto check = [=, &getl] {
254 : // Re-find read_ptr: it may be destroyed during the wait
255 45498 : const size_t buffer_size = buffer_.size();
256 45498 : const auto read_ptr = readoffsets_.find(ringbufferId);
257 45498 : if (buffer_size == 0 || read_ptr == readoffsets_.end())
258 30 : return true;
259 45468 : getl = (endPos_ + buffer_size - read_ptr->second.offset) % buffer_size;
260 45468 : return getl != 0;
261 22749 : };
262 :
263 22749 : if (deadline == time_point::max()) {
264 : // no timeout provided, wait as long as necessary
265 0 : not_empty_.wait(l, check);
266 : } else {
267 22749 : not_empty_.wait_until(l, deadline, check);
268 : }
269 :
270 22749 : return getl;
271 22749 : }
272 :
273 : size_t
274 0 : RingBuffer::discard(size_t toDiscard, const std::string& ringbufferId)
275 : {
276 0 : std::lock_guard l(lock_);
277 :
278 0 : const size_t buffer_size = buffer_.size();
279 0 : if (buffer_size == 0)
280 0 : return 0;
281 :
282 0 : auto offset = readoffsets_.find(ringbufferId);
283 0 : if (offset == readoffsets_.end())
284 0 : return 0;
285 :
286 0 : size_t len = (endPos_ + buffer_size - offset->second.offset) % buffer_size;
287 0 : toDiscard = std::min(toDiscard, len);
288 :
289 0 : offset->second.offset = (offset->second.offset + toDiscard) % buffer_size;
290 0 : return toDiscard;
291 0 : }
292 :
293 : size_t
294 0 : RingBuffer::discard(size_t toDiscard)
295 : {
296 0 : const size_t buffer_size = buffer_.size();
297 0 : if (buffer_size == 0)
298 0 : return 0;
299 :
300 0 : for (auto& r : readoffsets_) {
301 0 : size_t dst = (r.second.offset + buffer_size - endPos_) % buffer_size;
302 0 : if (dst < toDiscard)
303 0 : r.second.offset = (r.second.offset + toDiscard - dst) % buffer_size;
304 : }
305 0 : return toDiscard;
306 : }
307 :
308 : } // namespace jami
|
