?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Fix SA1 open bus
[lsnes.git] / src / core / inthread.cpp
blob711820430d634d79b0fc52713896f20581f58848
1 #include <lsnes.hpp>
2 #include <cstdint>
3 #include "library/filesystem.hpp"
4 #include "library/minmax.hpp"
5 #include "library/workthread.hpp"
6 #include "library/serialization.hpp"
7 #include "library/string.hpp"
8 #include "library/ogg.hpp"
9 #include "library/opus-ogg.hpp"
10 #include "library/opus.hpp"
11 #include "core/audioapi.hpp"
12 #include "core/command.hpp"
13 #include "core/dispatch.hpp"
14 #include "core/framerate.hpp"
15 #include "core/inthread.hpp"
16 #include "core/keymapper.hpp"
17 #include "core/settings.hpp"
18 #include "core/misc.hpp"
19 #include <cmath>
20 #include <list>
21 #include <iostream>
22 #include <fstream>
23 #include <cstring>
24 #include <unistd.h>
25 #include <sys/time.h>
26 #include <zlib.h>
27
28 //Farther than this, packets can be fastskipped.
29 #define OPUS_CONVERGE_MAX 5760
30 //Maximum size of PCM output for one packet.
31 #define OPUS_MAX_OUT 5760
32 //Output block size.
33 #define OUTPUT_BLOCK 1440
34 //Main sampling rate.
35 #define OPUS_SAMPLERATE 48000
36 //Opus block size
37 #define OPUS_BLOCK_SIZE 960
38 //Threshold for decoding additional block
39 #define BLOCK_THRESHOLD 1200
40 //Maximum output block size.
41 #define OUTPUT_SIZE (BLOCK_THRESHOLD + OUTPUT_BLOCK)
42 //Amount of microseconds per interation.
43 #define ITERATION_TIME 15000
44 //Opus bitrate to use.
45 #define OPUS_BITRATE 48000
46 //Opus min bitrate to use.
47 #define OPUS_MIN_BITRATE 8000
48 //Opus max bitrate to use.
49 #define OPUS_MAX_BITRATE 255000
50 //Ogg Opus granule rate.
51 #define OGGOPUS_GRANULERATE 48000
52 //Record buffer size threshold divider.
53 #define REC_THRESHOLD_DIV 40
54 //Playback buffer size threshold divider.
55 #define PLAY_THRESHOLD_DIV 30
56 //Special granule position: None.
57 #define GRANULEPOS_NONE 0xFFFFFFFFFFFFFFFFULL
58
59 namespace
60 {
61 class opus_playback_stream;
62 class opus_stream;
63 class stream_collection;
64
65 settingvar::variable<settingvar::model_int<OPUS_MIN_BITRATE,OPUS_MAX_BITRATE>> opus_bitrate(lsnes_vset,
66 "opus-bitrate", "commentary‣Bitrate", OPUS_BITRATE);
67 settingvar::variable<settingvar::model_int<OPUS_MIN_BITRATE,OPUS_MAX_BITRATE>> opus_max_bitrate(lsnes_vset,
68 "opus-max-bitrate", "commentary‣Max bitrate", OPUS_MAX_BITRATE);
69
70 //Recording active flag.
71 volatile bool active_flag = false;
72 //Last seen frame number.
73 uint64_t last_frame_number = 0;
74 //Last seen rate.
75 double last_rate = 0;
76 //Mutex protecting current_time and time_jump.
77 threads::lock time_mutex;
78 //The current time.
79 uint64_t current_time;
80 //Time jump flag. Set if time jump is detected.
81 //If time jump is detected, all current playing streams are stopped, stream locks are cleared and
82 //apropriate streams are restarted. If time jump is false, all unlocked streams coming into range
83 //are started.
84 bool time_jump;
85 //Lock protecting active_playback_streams.
86 threads::lock active_playback_streams_lock;
87 //List of streams currently playing.
88 std::list<opus_playback_stream*> active_playback_streams;
89 //The collection of streams.
90 stream_collection* current_collection;
91 //Lock protecting current collection.
92 threads::lock current_collection_lock;
93
94 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
95 //Bitrate tracker.
96 struct bitrate_tracker
97 {
98 bitrate_tracker() throw();
99 void reset() throw();
100 double get_min() throw();
101 double get_avg() throw();
102 double get_max() throw();
103 double get_length() throw();
104 uint64_t get_bytes() throw();
105 uint64_t get_blocks() throw();
106 void submit(uint32_t bytes, uint32_t samples) throw();
107 private:
108 uint64_t blocks;
109 uint64_t samples;
110 uint64_t bytes;
111 uint32_t minrate;
112 uint32_t maxrate;
113 };
114
115 bitrate_tracker::bitrate_tracker() throw()
116 {
117 reset();
118 }
119
120 void bitrate_tracker::reset() throw()
121 {
122 blocks = 0;
123 samples = 0;
124 bytes = 0;
125 minrate = std::numeric_limits<uint32_t>::max();
126 maxrate = 0;
127 }
128
129 double bitrate_tracker::get_min() throw()
130 {
131 return blocks ? minrate / 1000.0 : 0.0;
132 }
133
134 double bitrate_tracker::get_avg() throw()
135 {
136 return samples ? bytes / (125.0 * samples / OPUS_SAMPLERATE) : 0.0;
137 }
138
139 double bitrate_tracker::get_max() throw()
140 {
141 return blocks ? maxrate / 1000.0 : 0.0;
142 }
143
144 double bitrate_tracker::get_length() throw()
145 {
146 return 1.0 * samples / OPUS_SAMPLERATE;
147 }
148
149 uint64_t bitrate_tracker::get_bytes() throw()
150 {
151 return bytes;
152 }
153
154 uint64_t bitrate_tracker::get_blocks() throw()
155 {
156 return blocks;
157 }
158
159 void bitrate_tracker::submit(uint32_t _bytes, uint32_t _samples) throw()
160 {
161 blocks++;
162 samples += _samples;
163 bytes += _bytes;
164 uint32_t irate = _bytes * 8 * OPUS_SAMPLERATE / OPUS_BLOCK_SIZE;
165 minrate = min(minrate, irate);
166 maxrate = max(maxrate, irate);
167 }
168
169 std::ostream& operator<<(std::ostream& s, bitrate_tracker& t)
170 {
171 s << t.get_bytes() << " bytes for " << t.get_length() << "s (" << t.get_blocks() << " blocks)"
172 << std::endl << "Bitrate (kbps): min: " << t.get_min() << " avg: " << t.get_avg() << " max:"
173 << t.get_max() << std::endl;
174 return s;
175 }
176
177 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
178 //Information about individual opus packet in stream.
179 struct opus_packetinfo
180 {
181 //Length is in units of 1/400th of a second.
182 opus_packetinfo(uint16_t datasize, uint8_t length, uint64_t offset)
183 {
184 descriptor = (offset & 0xFFFFFFFFFFULL) | (static_cast<uint64_t>(length) << 40) |
185 (static_cast<uint64_t>(datasize) << 48);
186 }
187 //Get the data size of the packet.
188 uint16_t size() { return descriptor >> 48; }
189 //Calculate the length of packet in samples.
190 uint16_t length() { return 120 * ((descriptor >> 40) & 0xFF); }
191 //Calculate the true offset.
192 uint64_t offset() { return descriptor & 0xFFFFFFFFFFULL; }
193 //Read the packet.
194 //Can throw.
195 std::vector<unsigned char> packet(filesystem::ref from_sys);
196 private:
197 uint64_t descriptor;
198 };
199
200 std::vector<unsigned char> opus_packetinfo::packet(filesystem::ref from_sys)
201 {
202 std::vector<unsigned char> ret;
203 uint64_t off = offset();
204 uint32_t sz = size();
205 uint32_t cluster = off / CLUSTER_SIZE;
206 uint32_t coff = off % CLUSTER_SIZE;
207 ret.resize(sz);
208 size_t r = from_sys.read_data(cluster, coff, &ret[0], sz);
209 if(r != sz)
210 throw std::runtime_error("Incomplete read");
211 return ret;
212 }
213
214 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
215 //Information about opus stream.
216 struct opus_stream
217 {
218 //Create new empty stream with specified base time.
219 opus_stream(uint64_t base, filesystem::ref filesys);
220 //Read stream with specified base time and specified start clusters.
221 //Can throw.
222 opus_stream(uint64_t base, filesystem::ref filesys, uint32_t ctrl_cluster, uint32_t data_cluster);
223 //Import a stream with specified base time.
224 //Can throw.
225 opus_stream(uint64_t base, filesystem::ref filesys, std::ifstream& data,
226 external_stream_format extfmt);
227 //Delete this stream (also puts a ref)
228 void delete_stream() { deleting = true; put_ref(); }
229 //Export a stream.
230 //Can throw.
231 void export_stream(std::ofstream& data, external_stream_format extfmt);
232 //Get length of specified packet in samples.
233 uint16_t packet_length(uint32_t seqno)
234 {
235 return (seqno < packets.size()) ? packets[seqno].length() : 0;
236 }
237 //Get data of specified packet.
238 //Can throw.
239 std::vector<unsigned char> packet(uint32_t seqno)
240 {
241 return (seqno < packets.size()) ? packets[seqno].packet(fs) : std::vector<unsigned char>();
242 }
243 //Get base time in samples for stream.
244 uint64_t timebase() { return s_timebase; }
245 //Set base time in samples for stream.
246 void timebase(uint64_t ts) { s_timebase = ts; }
247 //Get length of stream in samples.
248 uint64_t length()
249 {
250 if(pregap_length + postgap_length > total_len)
251 return 0;
252 else
253 return total_len - pregap_length - postgap_length;
254 }
255 //Set the pregap length.
256 void set_pregap(uint32_t p) { pregap_length = p; }
257 //Get the pregap length.
258 uint32_t get_pregap() { return pregap_length; }
259 //Set the postgap length.
260 void set_potsgap(uint32_t p) { postgap_length = p; }
261 //Get the postgap length.
262 uint32_t get_postgap() { return postgap_length; }
263 //Set gain.
264 void set_gain(int16_t g) { gain = g; }
265 //Get gain.
266 int16_t get_gain() { return gain; }
267 //Get linear gain.
268 float get_gain_linear() { return pow(10, gain / 20); }
269 //Get number of packets in stream.
270 uint32_t blocks() { return packets.size(); }
271 //Is this stream locked?
272 bool islocked() { return locked; }
273 //Lock a stream.
274 void lock() { locked = true; }
275 //Unlock a stream.
276 void unlock() { locked = false; }
277 //Increment reference count.
278 void get_ref() { threads::alock m(reflock); refcount++; }
279 //Decrement reference count, destroying object if it hits zero.
280 void put_ref() { threads::alock m(reflock); refcount--; if(!refcount) destroy(); }
281 //Add new packet into stream.
282 //Not safe to call simultaneously with packet_length() or packet().
283 //Can throw.
284 void write(uint8_t len, const unsigned char* payload, size_t payload_len);
285 //Write stream trailer.
286 void write_trailier();
287 //Get clusters.
288 std::pair<uint32_t, uint32_t> get_clusters() { return std::make_pair(ctrl_cluster, data_cluster); }
289 private:
290 void export_stream_sox(std::ofstream& data);
291 void export_stream_oggopus(std::ofstream& data);
292 void import_stream_sox(std::ifstream& data);
293 void import_stream_oggopus(std::ifstream& data);
294
295 opus_stream(const opus_stream&);
296 opus_stream& operator=(const opus_stream&);
297 void destroy();
298 filesystem::ref fs;
299 std::vector<opus_packetinfo> packets;
300 uint64_t total_len;
301 uint64_t s_timebase;
302 uint32_t next_cluster;
303 uint32_t next_offset;
304 uint32_t next_mcluster;
305 uint32_t next_moffset;
306 uint32_t ctrl_cluster;
307 uint32_t data_cluster;
308 uint32_t pregap_length;
309 uint32_t postgap_length;
310 int16_t gain;
311 bool locked;
312 threads::lock reflock;
313 unsigned refcount;
314 bool deleting;
315 };
316
317 opus_stream::opus_stream(uint64_t base, filesystem::ref filesys)
318 : fs(filesys)
319 {
320 refcount = 1;
321 deleting = false;
322 total_len = 0;
323 s_timebase = base;
324 locked = false;
325 next_cluster = 0;
326 next_mcluster = 0;
327 next_offset = 0;
328 next_moffset = 0;
329 ctrl_cluster = 0;
330 data_cluster = 0;
331 pregap_length = 0;
332 postgap_length = 0;
333 gain = 0;
334 }
335
336 opus_stream::opus_stream(uint64_t base, filesystem::ref filesys, uint32_t _ctrl_cluster,
337 uint32_t _data_cluster)
338 : fs(filesys)
339 {
340 refcount = 1;
341 deleting = false;
342 total_len = 0;
343 s_timebase = base;
344 locked = false;
345 next_cluster = data_cluster = _data_cluster;
346 next_mcluster = ctrl_cluster = _ctrl_cluster;
347 next_offset = 0;
348 next_moffset = 0;
349 pregap_length = 0;
350 postgap_length = 0;
351 gain = 0;
352 //Read the data buffers.
353 char buf[CLUSTER_SIZE];
354 uint32_t last_cluster_seen = next_mcluster;
355 uint64_t total_size = 0;
356 uint64_t total_frames = 0;
357 bool trailers = false;
358 bool saved_pointer_valid = false;
359 uint32_t saved_next_mcluster = 0;
360 uint32_t saved_next_moffset = 0;
361 while(true) {
362 last_cluster_seen = next_mcluster;
363 size_t r = fs.read_data(next_mcluster, next_moffset, buf, CLUSTER_SIZE);
364 if(!r) {
365 //The stream ends here.
366 break;
367 }
368 //Find the first unused entry if any.
369 for(unsigned i = 0; i < CLUSTER_SIZE; i += 4)
370 if(!buf[i + 3] || trailers) {
371 //This entry is unused. If the next entry is also unused, that is the end.
372 //Otherwise, there might be stream trailers.
373 if(trailers && !buf[i + 3]) {
374 goto out_parsing; //Ends for real.
375 }
376 if(!trailers) {
377 //Set the trailer flag and continue parsing.
378 //The saved offset must be placed here.
379 saved_next_mcluster = last_cluster_seen;
380 saved_next_moffset = i;
381 saved_pointer_valid = true;
382 trailers = true;
383 continue;
384 }
385 //This is a trailer entry.
386 if(buf[i + 3] == 2) {
387 //Pregap.
388 pregap_length = serialization::u32b(buf + i) >> 8;
389 } else if(buf[i + 3] == 3) {
390 //Postgap.
391 postgap_length = serialization::u32b(buf + i) >> 8;
392 } else if(buf[i + 3] == 4) {
393 //Gain.
394 gain = serialization::s16b(buf + i);
395 }
396 } else {
397 uint16_t psize = serialization::u16b(buf + i);
398 uint8_t plen = serialization::u8b(buf + i + 2);
399 total_size += psize;
400 total_len += 120 * plen;
401 opus_packetinfo p(psize, plen, 1ULL * next_cluster * CLUSTER_SIZE +
402 next_offset);
403 size_t r2 = fs.skip_data(next_cluster, next_offset, psize);
404 if(r2 < psize)
405 throw std::runtime_error("Incomplete data stream");
406 packets.push_back(p);
407 total_frames++;
408 }
409 }
410 out_parsing:
411 //If saved pointer is valid, restore to that.
412 if(saved_pointer_valid) {
413 next_mcluster = saved_next_mcluster;
414 next_moffset = saved_next_moffset;
415 }
416 }
417
418 opus_stream::opus_stream(uint64_t base, filesystem::ref filesys, std::ifstream& data,
419 external_stream_format extfmt)
420 : fs(filesys)
421 {
422 refcount = 1;
423 deleting = false;
424 total_len = 0;
425 s_timebase = base;
426 locked = false;
427 next_cluster = 0;
428 next_mcluster = 0;
429 next_offset = 0;
430 next_moffset = 0;
431 ctrl_cluster = 0;
432 data_cluster = 0;
433 pregap_length = 0;
434 postgap_length = 0;
435 gain = 0;
436 if(extfmt == EXTFMT_OGGOPUS)
437 import_stream_oggopus(data);
438 else if(extfmt == EXTFMT_SOX)
439 import_stream_sox(data);
440 }
441
442 void opus_stream::import_stream_oggopus(std::ifstream& data)
443 {
444 ogg::stream_reader_iostreams reader(data);
445 reader.set_errors_to(messages);
446 struct opus::ogg_header h;
447 struct opus::ogg_tags t;
448 ogg::page page;
449 ogg::demuxer d(messages);
450 int state = 0;
451 postgap_length = 0;
452 uint64_t datalen = 0;
453 uint64_t last_datalen = 0;
454 uint64_t last_granulepos = 0;
455 try {
456 while(true) {
457 ogg::packet p;
458 if(!d.wants_packet_out()) {
459 if(!reader.get_page(page))
460 break;
461 d.page_in(page);
462 continue;
463 } else
464 d.packet_out(p);
465 switch(state) {
466 case 0: //Not locked.
467 h.parse(p);
468 if(h.streams != 1)
469 throw std::runtime_error("Multistream OggOpus streams are not "
470 "supported");
471 state = 1; //Expecting comment.
472 pregap_length = h.preskip;
473 gain = h.gain;
474 break;
475 case 1: //Expecting comment.
476 t.parse(p);
477 state = 2; //Data page.
478 if(page.get_eos())
479 throw std::runtime_error("Empty OggOpus stream");
480 break;
481 case 2: //Data page.
482 case 3: //Data page.
483 const std::vector<uint8_t>& pkt = p.get_vector();
484 uint8_t tcnt = opus::packet_tick_count(&pkt[0], pkt.size());
485 if(tcnt) {
486 write(tcnt, &pkt[0], pkt.size());
487 datalen += tcnt * 120;
488 }
489 if(p.get_last_page()) {
490 uint64_t samples = p.get_granulepos() - last_granulepos;
491 if(samples > p.get_granulepos())
492 samples = 0;
493 uint64_t rsamples = datalen - last_datalen;
494 if((samples > rsamples && state == 3) || (samples <
495 rsamples && !p.get_on_eos_page()))
496 messages << "Warning: Granulepos says there are "
497 << samples << " samples, found " << rsamples
498 << std::endl;
499 last_datalen = datalen;
500 last_granulepos = p.get_granulepos();
501 if(p.get_on_eos_page()) {
502 if(samples < rsamples)
503 postgap_length = rsamples - samples;
504 state = 4;
505 goto out;
506 }
507 }
508 state = 3;
509 break;
510 }
511 }
512 out:
513 if(state == 0)
514 throw std::runtime_error("No OggOpus stream found");
515 if(state == 1)
516 throw std::runtime_error("Oggopus stream missing required tags pages");
517 if(state == 2 || state == 3)
518 messages << "Warning: Incomplete Oggopus stream." << std::endl;
519 if(datalen <= pregap_length)
520 throw std::runtime_error("Stream too short (entiere pregap not present)");
521 write_trailier();
522 } catch(...) {
523 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
524 if(data_cluster) fs.free_cluster_chain(data_cluster);
525 throw;
526 }
527 }
528
529 void opus_stream::import_stream_sox(std::ifstream& data)
530 {
531 bitrate_tracker brtrack;
532 unsigned char tmpi[65536];
533 float tmp[OPUS_MAX_OUT];
534 char header[260];
535 data.read(header, 32);
536 if(!data)
537 throw std::runtime_error("Can't read .sox header");
538 if(serialization::u32l(header + 0) != 0x586F532EULL)
539 throw std::runtime_error("Bad .sox header magic");
540 if(serialization::u8b(header + 4) > 28)
541 data.read(header + 32, serialization::u8b(header + 4) - 28);
542 if(!data)
543 throw std::runtime_error("Can't read .sox header");
544 if(serialization::u64l(header + 16) != 4676829883349860352ULL)
545 throw std::runtime_error("Bad .sox sampling rate");
546 if(serialization::u32l(header + 24) != 1)
547 throw std::runtime_error("Only mono streams are supported");
548 uint64_t samples = serialization::u64l(header + 8);
549 opus::encoder enc(opus::samplerate::r48k, false, opus::application::voice);
550 enc.ctl(opus::bitrate(opus_bitrate.get()));
551 int32_t pregap = enc.ctl(opus::lookahead);
552 pregap_length = pregap;
553 for(uint64_t i = 0; i < samples + pregap; i += OPUS_BLOCK_SIZE) {
554 size_t bs = OPUS_BLOCK_SIZE;
555 if(i + bs > samples + pregap)
556 bs = samples + pregap - i;
557 //We have to read zero bytes after the end of stream.
558 size_t readable = bs;
559 if(readable + i > samples)
560 readable = max(samples, i) - i;
561 if(readable > 0)
562 data.read(reinterpret_cast<char*>(tmpi), 4 * readable);
563 if(readable < bs)
564 memset(tmpi + 4 * readable, 0, 4 * (bs - readable));
565 if(!data) {
566 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
567 if(data_cluster) fs.free_cluster_chain(data_cluster);
568 throw std::runtime_error("Can't read .sox data");
569 }
570 for(size_t j = 0; j < bs; j++)
571 tmp[j] = static_cast<float>(serialization::s32l(tmpi + 4 * j)) / 268435456;
572 if(bs < OPUS_BLOCK_SIZE)
573 postgap_length = OPUS_BLOCK_SIZE - bs;
574 for(size_t j = bs; j < OPUS_BLOCK_SIZE; j++)
575 tmp[j] = 0;
576 try {
577 const size_t opus_out_max2 = opus_max_bitrate.get() * OPUS_BLOCK_SIZE / 384000;
578 size_t r = enc.encode(tmp, OPUS_BLOCK_SIZE, tmpi, opus_out_max2);
579 write(OPUS_BLOCK_SIZE / 120, tmpi, r);
580 brtrack.submit(r, bs);
581 } catch(std::exception& e) {
582 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
583 if(data_cluster) fs.free_cluster_chain(data_cluster);
584 (stringfmt() << "Error encoding opus packet: " << e.what()).throwex();
585 }
586 }
587 messages << "Imported stream: " << brtrack;
588 try {
589 write_trailier();
590 } catch(...) {
591 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
592 if(data_cluster) fs.free_cluster_chain(data_cluster);
593 throw;
594 }
595 }
596
597 void opus_stream::destroy()
598 {
599 if(deleting) {
600 //We catch the errors and print em, because otherwise put_ref could throw, which would
601 //be too much.
602 try {
603 fs.free_cluster_chain(ctrl_cluster);
604 } catch(std::exception& e) {
605 messages << "Failed to delete stream control file: " << e.what();
606 }
607 try {
608 fs.free_cluster_chain(data_cluster);
609 } catch(std::exception& e) {
610 messages << "Failed to delete stream data file: " << e.what();
611 }
612 }
613 delete this;
614 }
615
616 void opus_stream::export_stream_oggopus(std::ofstream& data)
617 {
618 if(!packets.size())
619 throw std::runtime_error("Empty oggopus stream is not valid");
620 opus::ogg_header header;
621 opus::ogg_tags tags;
622 ogg::stream_writer_iostreams writer(data);
623 unsigned stream_id = 1;
624 uint64_t true_granule = 0;
625 uint32_t seq = 2;
626 //Headers / Tags.
627 header.version = 1;
628 header.channels = 1;
629 header.preskip = pregap_length;
630 header.rate = OPUS_SAMPLERATE;
631 header.gain = 0;
632 header.map_family = 0;
633 header.streams = 1;
634 header.coupled = 0;
635 header.chanmap[0] = 0;
636 memset(header.chanmap + 1, 255, 254);
637 tags.vendor = "unknown";
638 tags.comments.push_back((stringfmt() << "ENCODER=lsnes rr" + lsnes_version).str());
639 tags.comments.push_back((stringfmt() << "LSNES_STREAM_TS=" << s_timebase).str());
640
641 struct ogg::page hpage = header.serialize();
642 hpage.set_stream(stream_id);
643 writer.put_page(hpage);
644 seq = tags.serialize([&writer](const ogg::page& p) { writer.put_page(p); }, stream_id);
645
646 struct ogg::page ppage;
647 ogg::muxer mux(stream_id, seq);
648 for(size_t i = 0; i < packets.size(); i++) {
649 std::vector<unsigned char> p;
650 try {
651 p = packet(i);
652 } catch(std::exception& e) {
653 (stringfmt() << "Error reading opus packet: " << e.what()).throwex();
654 }
655 if(!p.size())
656 (stringfmt() << "Empty Opus packet is not valid").throwex();
657 uint32_t samples = static_cast<uint32_t>(opus::packet_tick_count(&p[0], p.size())) * 120;
658 if(i + 1 < packets.size())
659 true_granule += samples;
660 else
661 true_granule = max(true_granule, true_granule + samples - postgap_length);
662 if(!mux.wants_packet_in() || !mux.packet_fits(p.size()))
663 while(mux.has_page_out()) {
664 mux.page_out(ppage);
665 writer.put_page(ppage);
666 }
667 mux.packet_in(p, true_granule);
668 }
669 mux.signal_eos();
670 while(mux.has_page_out()) {
671 mux.page_out(ppage);
672 writer.put_page(ppage);
673 }
674 }
675
676 void opus_stream::export_stream_sox(std::ofstream& data)
677 {
678 opus::decoder dec(opus::samplerate::r48k, false);
679 std::vector<unsigned char> p;
680 float tmp[OPUS_MAX_OUT];
681 char header[32];
682 serialization::u64l(header, 0x1C586F532EULL); //Magic and header size.
683 serialization::u64l(header + 16, 4676829883349860352ULL); //Sampling rate.
684 serialization::u32l(header + 24, 1);
685 uint64_t tlen = 0;
686 uint32_t lookahead_thrown = 0;
687 data.write(header, 32);
688 if(!data)
689 throw std::runtime_error("Error writing PCM data.");
690 float lgain = get_gain_linear();
691 for(size_t i = 0; i < packets.size(); i++) {
692 char blank[4] = {0, 0, 0, 0};
693 try {
694 uint32_t pregap_throw = 0;
695 uint32_t postgap_throw = 0;
696 std::vector<unsigned char> p = packet(i);
697 uint32_t len = packet_length(i);
698 dec.decode(&p[0], p.size(), tmp, OPUS_MAX_OUT);
699 bool is_last = (i == packets.size() - 1);
700 if(lookahead_thrown < pregap_length) {
701 //We haven't yet thrown the full pregap. Throw some.
702 uint32_t maxthrow = pregap_length - lookahead_thrown;
703 pregap_throw = min(len, maxthrow);
704 lookahead_thrown += pregap_length;
705 }
706 if(is_last)
707 postgap_throw = min(len - pregap_throw, postgap_length);
708 tlen += (len - pregap_throw - postgap_throw);
709 for(uint32_t j = pregap_throw; j < len - postgap_throw; j++) {
710 int32_t s = (int32_t)(tmp[j] * lgain * 268435456.0);
711 serialization::s32l(blank, s);
712 data.write(blank, 4);
713 if(!data)
714 throw std::runtime_error("Error writing PCM data.");
715 }
716 } catch(std::exception& e) {
717 (stringfmt() << "Error decoding opus packet: " << e.what()).throwex();
718 }
719 }
720 data.seekp(0, std::ios_base::beg);
721 serialization::u64l(header + 8, tlen);
722 data.write(header, 32);
723 if(!data) {
724 throw std::runtime_error("Error writing PCM data.");
725 }
726 }
727
728 void opus_stream::export_stream(std::ofstream& data, external_stream_format extfmt)
729 {
730 if(extfmt == EXTFMT_OGGOPUS)
731 export_stream_oggopus(data);
732 else if(extfmt == EXTFMT_SOX)
733 export_stream_sox(data);
734 }
735
736 void opus_stream::write(uint8_t len, const unsigned char* payload, size_t payload_len)
737 {
738 try {
739 char descriptor[4];
740 uint32_t used_cluster, used_offset;
741 uint32_t used_mcluster, used_moffset;
742 if(!next_cluster)
743 next_cluster = data_cluster = fs.allocate_cluster();
744 if(!next_mcluster)
745 next_mcluster = ctrl_cluster = fs.allocate_cluster();
746 serialization::u16b(descriptor, payload_len);
747 serialization::u8b(descriptor + 2, len);
748 serialization::u8b(descriptor + 3, 1);
749 fs.write_data(next_cluster, next_offset, payload, payload_len, used_cluster, used_offset);
750 fs.write_data(next_mcluster, next_moffset, descriptor, 4, used_mcluster, used_moffset);
751 uint64_t off = static_cast<uint64_t>(used_cluster) * CLUSTER_SIZE + used_offset;
752 opus_packetinfo p(payload_len, len, off);
753 total_len += p.length();
754 packets.push_back(p);
755 } catch(std::exception& e) {
756 (stringfmt() << "Can't write opus packet: " << e.what()).throwex();
757 }
758 }
759
760 void opus_stream::write_trailier()
761 {
762 try {
763 char descriptor[16];
764 uint32_t used_mcluster, used_moffset;
765 //The allocation must be done for real.
766 if(!next_mcluster)
767 next_mcluster = ctrl_cluster = fs.allocate_cluster();
768 //But the write must not update the pointers..
769 uint32_t tmp_mcluster = next_mcluster;
770 uint32_t tmp_moffset = next_moffset;
771 serialization::u32b(descriptor, 0);
772 serialization::u32b(descriptor + 4, (pregap_length << 8) | 0x02);
773 serialization::u32b(descriptor + 8, (postgap_length << 8) | 0x03);
774 serialization::s16b(descriptor + 12, gain);
775 serialization::u16b(descriptor + 14, 0x0004);
776 fs.write_data(tmp_mcluster, tmp_moffset, descriptor, 16, used_mcluster, used_moffset);
777 } catch(std::exception& e) {
778 (stringfmt() << "Can't write stream trailer: " << e.what()).throwex();
779 }
780 }
781
782
783 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
784 //Playing opus stream.
785 struct opus_playback_stream
786 {
787 //Create a new playing stream from given opus stream.
788 opus_playback_stream(opus_stream& data);
789 //Destroy playing opus stream.
790 ~opus_playback_stream();
791 //Read samples from stream.
792 //Can throw.
793 void read(float* data, size_t samples);
794 //Skip samples from stream.
795 //Can throw.
796 void skip(uint64_t samples);
797 //Has the stream already ended?
798 bool eof();
799 private:
800 opus_playback_stream(const opus_playback_stream&);
801 opus_playback_stream& operator=(const opus_playback_stream&);
802 //Can throw.
803 void decode_block();
804 float output[OPUS_MAX_OUT];
805 unsigned output_left;
806 uint32_t pregap_thrown;
807 bool postgap_thrown;
808 opus::decoder* decoder;
809 opus_stream& stream;
810 uint32_t next_block;
811 uint32_t blocks;
812 };
813
814 opus_playback_stream::opus_playback_stream(opus_stream& data)
815 : stream(data)
816 {
817 stream.get_ref();
818 stream.lock();
819 next_block = 0;
820 output_left = 0;
821 pregap_thrown = 0;
822 postgap_thrown = false;
823 blocks = stream.blocks();
824 decoder = new opus::decoder(opus::samplerate::r48k, false);
825 if(!decoder)
826 throw std::bad_alloc();
827 }
828
829 opus_playback_stream::~opus_playback_stream()
830 {
831 //No, we don't unlock the stream.
832 stream.put_ref();
833 delete decoder;
834 }
835
836 bool opus_playback_stream::eof()
837 {
838 return (next_block >= blocks && !output_left);
839 }
840
841 void opus_playback_stream::decode_block()
842 {
843 if(next_block >= blocks)
844 return;
845 if(output_left >= OPUS_MAX_OUT)
846 return;
847 unsigned plen = stream.packet_length(next_block);
848 if(plen + output_left > OPUS_MAX_OUT)
849 return;
850 std::vector<unsigned char> pdata = stream.packet(next_block);
851 try {
852 size_t c = decoder->decode(&pdata[0], pdata.size(), output + output_left,
853 OPUS_MAX_OUT - output_left);
854 output_left = min(output_left + c, static_cast<size_t>(OPUS_MAX_OUT));
855 } catch(...) {
856 //Bad packet, insert silence.
857 for(unsigned i = 0; i < plen; i++)
858 output[output_left++] = 0;
859 }
860 //Throw the pregap away if needed.
861 if(pregap_thrown < stream.get_pregap()) {
862 uint32_t throw_amt = min(stream.get_pregap() - pregap_thrown, (uint32_t)output_left);
863 if(throw_amt && throw_amt < output_left)
864 memmove(output, output + throw_amt, (output_left - throw_amt) * sizeof(float));
865 output_left -= throw_amt;
866 pregap_thrown += throw_amt;
867 }
868 next_block++;
869 }
870
871 void opus_playback_stream::read(float* data, size_t samples)
872 {
873 float lgain = stream.get_gain_linear();
874 while(samples > 0) {
875 decode_block();
876 if(next_block >= blocks && !postgap_thrown) {
877 //This is the final packet. Throw away postgap samples at the end.
878 uint32_t thrown = min(stream.get_postgap(), (uint32_t)output_left);
879 output_left -= thrown;
880 postgap_thrown = true;
881 }
882 if(next_block >= blocks && !output_left) {
883 //Zerofill remainder.
884 for(size_t i = 0; i < samples; i++)
885 data[i] = 0;
886 return;
887 }
888 unsigned maxcopy = min(static_cast<unsigned>(samples), output_left);
889 if(maxcopy) {
890 memcpy(data, output, maxcopy * sizeof(float));
891 for(size_t i = 0; i < maxcopy; i++)
892 data[i] *= lgain;
893 }
894 if(maxcopy < output_left && maxcopy)
895 memmove(output, output + maxcopy, (output_left - maxcopy) * sizeof(float));
896 output_left -= maxcopy;
897 samples -= maxcopy;
898 data += maxcopy;
899 }
900 }
901
902 void opus_playback_stream::skip(uint64_t samples)
903 {
904 //Adjust for preskip and declare all preskip already thrown away.
905 pregap_thrown = stream.get_pregap();
906 samples += pregap_thrown;
907 postgap_thrown = false;
908 //First, skip inside decoded samples.
909 if(samples < output_left) {
910 //Skipping less than amount in output buffer. Just discard from output buffer and try
911 //to decode a new block.
912 memmove(output, output + samples, (output_left - samples) * sizeof(float));
913 output_left -= samples;
914 decode_block();
915 return;
916 } else {
917 //Skipping at least the amount of samples in output buffer. First, blank the output buffer
918 //and count those towards samples discarded.
919 samples -= output_left;
920 output_left = 0;
921 }
922 //While number of samples is so great that adequate convergence period can be ensured without
923 //decoding this packet, just skip the samples from the packet.
924 while(samples > OPUS_CONVERGE_MAX) {
925 samples -= stream.packet_length(next_block++);
926 //Did we hit EOF?
927 if(next_block >= blocks)
928 return;
929 }
930 //Okay, we are near the point. Start decoding packets.
931 while(samples > 0) {
932 decode_block();
933 //Did we hit EOF?
934 if(next_block >= blocks && !output_left)
935 return;
936 //Skip as many samples as possible.
937 unsigned maxskip = min(static_cast<unsigned>(samples), output_left);
938 if(maxskip < output_left)
939 memmove(output, output + maxskip, (output_left - maxskip) * sizeof(float));
940 output_left -= maxskip;
941 samples -= maxskip;
942 }
943 //Just to be nice, decode a extra block.
944 decode_block();
945 }
946
947
948 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
949 //Collection of streams.
950 struct stream_collection
951 {
952 public:
953 //Create a new collection.
954 //Can throw.
955 stream_collection(filesystem::ref filesys);
956 //Destroy a collection. All streams are destroyed but not deleted.
957 ~stream_collection();
958 //Get list of streams active at given point.
959 std::list<uint64_t> streams_at(uint64_t point);
960 //Add a stream into collection.
961 //Can throw.
962 uint64_t add_stream(opus_stream& stream);
963 //Get the filesystem this collection is for.
964 filesystem::ref get_filesystem() { return fs; }
965 //Unlock all streams in collection.
966 void unlock_all();
967 //Get stream with given index (NULL if not found).
968 opus_stream* get_stream(uint64_t index)
969 {
970 threads::alock m(mlock);
971 if(streams.count(index)) {
972 streams[index]->get_ref();
973 return streams[index];
974 }
975 return NULL;
976 }
977 //Delete a stream.
978 //Can throw.
979 void delete_stream(uint64_t index);
980 //Alter stream timebase.
981 //Can throw.
982 void alter_stream_timebase(uint64_t index, uint64_t newts);
983 //Alter stream gain.
984 void alter_stream_gain(uint64_t index, uint16_t newgain);
985 //Enumerate all valid stream indices, in time order.
986 std::list<uint64_t> all_streams();
987 //Export the entiere superstream.
988 //Can throw.
989 void export_superstream(std::ofstream& out);
990 private:
991 filesystem::ref fs;
992 uint64_t next_index;
993 unsigned next_stream;
994 threads::lock mlock;
995 std::set<uint64_t> free_indices;
996 std::map<uint64_t, uint64_t> entries;
997 std::multimap<uint64_t, uint64_t> streams_by_time;
998 //FIXME: Something more efficient.
999 std::map<uint64_t, opus_stream*> streams;
1000 };
1001
1002 stream_collection::stream_collection(filesystem::ref filesys)
1003 : fs(filesys)
1004 {
1005 next_stream = 0;
1006 next_index = 0;
1007 //The stream index table is in cluster 2.
1008 uint32_t next_cluster = 2;
1009 uint32_t next_offset = 0;
1010 uint32_t i = 0;
1011 try {
1012 while(true) {
1013 char buffer[16];
1014 size_t r = fs.read_data(next_cluster, next_offset, buffer, 16);
1015 if(r < 16)
1016 break;
1017 uint64_t timebase = serialization::u64b(buffer);
1018 uint32_t ctrl_cluster = serialization::u32b(buffer + 8);
1019 uint32_t data_cluster = serialization::u32b(buffer + 12);
1020 if(ctrl_cluster) {
1021 opus_stream* x = new opus_stream(timebase, fs, ctrl_cluster, data_cluster);
1022 entries[next_index] = i;
1023 streams_by_time.insert(std::make_pair(timebase, next_index));
1024 streams[next_index++] = x;
1025 } else
1026 free_indices.insert(i);
1027 next_stream = ++i;
1028 }
1029 } catch(std::exception& e) {
1030 for(auto i : streams)
1031 i.second->put_ref();
1032 (stringfmt() << "Failed to parse LSVS: " << e.what()).throwex();
1033 }
1034 }
1035
1036 stream_collection::~stream_collection()
1037 {
1038 threads::alock m(mlock);
1039 for(auto i : streams)
1040 i.second->put_ref();
1041 streams.clear();
1042 }
1043
1044 std::list<uint64_t> stream_collection::streams_at(uint64_t point)
1045 {
1046 threads::alock m(mlock);
1047 std::list<uint64_t> s;
1048 for(auto i : streams) {
1049 uint64_t start = i.second->timebase();
1050 uint64_t end = start + i.second->length();
1051 if(point >= start && point < end) {
1052 i.second->get_ref();
1053 s.push_back(i.first);
1054 }
1055 }
1056 return s;
1057 }
1058
1059 uint64_t stream_collection::add_stream(opus_stream& stream)
1060 {
1061 uint64_t idx;
1062 try {
1063 threads::alock m(mlock);
1064 //Lock the added stream so it doesn't start playing back immediately.
1065 stream.lock();
1066 idx = next_index++;
1067 streams[idx] = &stream;
1068 char buffer[16];
1069 serialization::u64b(buffer, stream.timebase());
1070 auto r = stream.get_clusters();
1071 serialization::u32b(buffer + 8, r.first);
1072 serialization::u32b(buffer + 12, r.second);
1073 uint64_t entry_number = 0;
1074 if(free_indices.empty())
1075 entry_number = next_stream++;
1076 else {
1077 entry_number = *free_indices.begin();
1078 free_indices.erase(entry_number);
1079 }
1080 uint32_t write_cluster = 2;
1081 uint32_t write_offset = 0;
1082 uint32_t dummy1, dummy2;
1083 fs.skip_data(write_cluster, write_offset, 16 * entry_number);
1084 fs.write_data(write_cluster, write_offset, buffer, 16, dummy1, dummy2);
1085 streams_by_time.insert(std::make_pair(stream.timebase(), idx));
1086 entries[idx] = entry_number;
1087 return idx;
1088 } catch(std::exception& e) {
1089 (stringfmt() << "Failed to add stream: " << e.what()).throwex();
1090 }
1091 return idx;
1092 }
1093
1094 void stream_collection::unlock_all()
1095 {
1096 threads::alock m(mlock);
1097 for(auto i : streams)
1098 i.second->unlock();
1099 }
1100
1101 void stream_collection::delete_stream(uint64_t index)
1102 {
1103 threads::alock m(mlock);
1104 if(!entries.count(index))
1105 return;
1106 uint64_t entry_number = entries[index];
1107 uint32_t write_cluster = 2;
1108 uint32_t write_offset = 0;
1109 uint32_t dummy1, dummy2;
1110 char buffer[16] = {0};
1111 fs.skip_data(write_cluster, write_offset, 16 * entry_number);
1112 fs.write_data(write_cluster, write_offset, buffer, 16, dummy1, dummy2);
1113 auto itr = streams_by_time.lower_bound(streams[index]->timebase());
1114 auto itr2 = streams_by_time.upper_bound(streams[index]->timebase());
1115 for(auto x = itr; x != itr2; x++)
1116 if(x->second == index) {
1117 streams_by_time.erase(x);
1118 break;
1119 }
1120 streams[index]->delete_stream();
1121 streams.erase(index);
1122 }
1123
1124 void stream_collection::alter_stream_timebase(uint64_t index, uint64_t newts)
1125 {
1126 try {
1127 threads::alock m(mlock);
1128 if(!streams.count(index))
1129 return;
1130 if(entries.count(index)) {
1131 char buffer[8];
1132 uint32_t write_cluster = 2;
1133 uint32_t write_offset = 0;
1134 uint32_t dummy1, dummy2;
1135 serialization::u64b(buffer, newts);
1136 fs.skip_data(write_cluster, write_offset, 16 * entries[index]);
1137 fs.write_data(write_cluster, write_offset, buffer, 8, dummy1, dummy2);
1138 }
1139 auto itr = streams_by_time.lower_bound(streams[index]->timebase());
1140 auto itr2 = streams_by_time.upper_bound(streams[index]->timebase());
1141 for(auto x = itr; x != itr2; x++)
1142 if(x->second == index) {
1143 streams_by_time.erase(x);
1144 break;
1145 }
1146 streams[index]->timebase(newts);
1147 streams_by_time.insert(std::make_pair(newts, index));
1148 } catch(std::exception& e) {
1149 (stringfmt() << "Failed to alter stream timebase: " << e.what()).throwex();
1150 }
1151 }
1152
1153 void stream_collection::alter_stream_gain(uint64_t index, uint16_t newgain)
1154 {
1155 try {
1156 threads::alock m(mlock);
1157 if(!streams.count(index))
1158 return;
1159 streams[index]->set_gain(newgain);
1160 streams[index]->write_trailier();
1161 } catch(std::exception& e) {
1162 (stringfmt() << "Failed to alter stream gain: " << e.what()).throwex();
1163 }
1164 }
1165
1166 std::list<uint64_t> stream_collection::all_streams()
1167 {
1168 threads::alock m(mlock);
1169 std::list<uint64_t> s;
1170 for(auto i : streams_by_time)
1171 s.push_back(i.second);
1172 return s;
1173 }
1174
1175 void stream_collection::export_superstream(std::ofstream& out)
1176 {
1177 std::list<uint64_t> slist = all_streams();
1178 //Find the total length of superstream.
1179 uint64_t len = 0;
1180 for(auto i : slist) {
1181 opus_stream* s = get_stream(i);
1182 if(s) {
1183 len = max(len, s->timebase() + s->length());
1184 s->put_ref();
1185 }
1186 }
1187 char header[32];
1188 serialization::u64l(header, 0x1C586F532EULL); //Magic and header size.
1189 serialization::u64l(header + 8, len);
1190 serialization::u64l(header + 16, 4676829883349860352ULL); //Sampling rate.
1191 serialization::u64l(header + 24, 1);
1192 out.write(header, 32);
1193 if(!out)
1194 throw std::runtime_error("Error writing PCM output");
1195
1196 //Find the first valid stream.
1197 auto next_i = slist.begin();
1198 opus_stream* next_stream = NULL;
1199 while(next_i != slist.end()) {
1200 next_stream = get_stream(*next_i);
1201 next_i++;
1202 if(next_stream)
1203 break;
1204 }
1205 uint64_t next_ts;
1206 next_ts = next_stream ? next_stream->timebase() : len;
1207
1208 std::list<opus_playback_stream*> active;
1209 try {
1210 for(uint64_t s = 0; s < len;) {
1211 if(s == next_ts) {
1212 active.push_back(new opus_playback_stream(*next_stream));
1213 next_stream->put_ref();
1214 next_stream = NULL;
1215 while(next_i != slist.end()) {
1216 next_stream = get_stream(*next_i);
1217 next_i++;
1218 if(!next_stream)
1219 continue;
1220 uint64_t next_ts = next_stream->timebase();
1221 if(next_ts > s)
1222 break;
1223 //Okay, this starts too...
1224 active.push_back(new opus_playback_stream(*next_stream));
1225 next_stream->put_ref();
1226 next_stream = NULL;
1227 };
1228 next_ts = next_stream ? next_stream->timebase() : len;
1229 }
1230 uint64_t maxsamples = min(next_ts - s, static_cast<uint64_t>(OUTPUT_BLOCK));
1231 maxsamples = min(maxsamples, len - s);
1232 char outbuf[4 * OUTPUT_BLOCK];
1233 float buf1[OUTPUT_BLOCK];
1234 float buf2[OUTPUT_BLOCK];
1235 for(size_t t = 0; t < maxsamples; t++)
1236 buf1[t] = 0;
1237 for(auto t : active) {
1238 t->read(buf2, maxsamples);
1239 for(size_t u = 0; u < maxsamples; u++)
1240 buf1[u] += buf2[u];
1241 }
1242 for(auto t = active.begin(); t != active.end();) {
1243 if((*t)->eof()) {
1244 auto todel = t;
1245 t++;
1246 delete *todel;
1247 active.erase(todel);
1248 } else
1249 t++;
1250 }
1251 for(size_t t = 0; t < maxsamples; t++)
1252 serialization::s32l(outbuf + 4 * t, buf1[t] * 268435456);
1253 out.write(outbuf, 4 * maxsamples);
1254 if(!out)
1255 throw std::runtime_error("Failed to write PCM");
1256 s += maxsamples;
1257 }
1258 } catch(std::exception& e) {
1259 (stringfmt() << "Failed to export PCM: " << e.what()).throwex();
1260 }
1261 for(auto t = active.begin(); t != active.end();) {
1262 if((*t)->eof()) {
1263 auto todelete = t;
1264 t++;
1265 delete *todelete;
1266 active.erase(todelete);
1267 } else
1268 t++;
1269 }
1270 }
1271
1272 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
1273 void start_management_stream(opus_stream& s)
1274 {
1275 opus_playback_stream* p = new opus_playback_stream(s);
1276 threads::alock m(active_playback_streams_lock);
1277 active_playback_streams.push_back(p);
1278 }
1279
1280 void advance_time(uint64_t newtime)
1281 {
1282 threads::alock m2(current_collection_lock);
1283 if(!current_collection) {
1284 //Clear all.
1285 threads::alock m(active_playback_streams_lock);
1286 for(auto i : active_playback_streams)
1287 delete i;
1288 active_playback_streams.clear();
1289 return;
1290 }
1291 std::list<uint64_t> sactive = current_collection->streams_at(newtime);
1292 for(auto j : sactive) {
1293 opus_stream* i = current_collection->get_stream(j);
1294 if(!i)
1295 continue;
1296 //Don't play locked streams in order to avoid double playing.
1297 threads::alock m(active_playback_streams_lock);
1298 try {
1299 if(!i->islocked())
1300 active_playback_streams.push_back(new opus_playback_stream(*i));
1301 } catch(std::exception& e) {
1302 messages << "Can't start stream: " << e.what() << std::endl;
1303 }
1304 i->put_ref();
1305 }
1306 }
1307
1308 void jump_time(uint64_t newtime)
1309 {
1310 threads::alock m2(current_collection_lock);
1311 if(!current_collection) {
1312 //Clear all.
1313 threads::alock m(active_playback_streams_lock);
1314 for(auto i : active_playback_streams)
1315 delete i;
1316 active_playback_streams.clear();
1317 return;
1318 }
1319 //Close all currently playing streams.
1320 {
1321 threads::alock m(active_playback_streams_lock);
1322 for(auto i : active_playback_streams)
1323 delete i;
1324 active_playback_streams.clear();
1325 }
1326 //Unlock all streams, so they will play.
1327 current_collection->unlock_all();
1328 //Reopen all streams that should be open (with seeking)
1329 std::list<uint64_t> sactive = current_collection->streams_at(newtime);
1330 for(auto j : sactive) {
1331 opus_stream* i = current_collection->get_stream(j);
1332 if(!i)
1333 continue;
1334 //No need to check for locks, because we just busted all of those.
1335 uint64_t p = newtime - i->timebase();
1336 opus_playback_stream* s;
1337 try {
1338 s = new opus_playback_stream(*i);
1339 } catch(std::exception& e) {
1340 messages << "Can't start stream: " << e.what() << std::endl;
1341 return;
1342 }
1343 i->put_ref();
1344 if(!s)
1345 continue;
1346 s->skip(p);
1347 threads::alock m(active_playback_streams_lock);
1348 active_playback_streams.push_back(s);
1349 }
1350 }
1351
1352 //Resample.
1353 void do_resample(audioapi_resampler& r, float* srcbuf, size_t& srcuse, float* dstbuf, size_t& dstuse,
1354 size_t dstmax, double ratio)
1355 {
1356 if(srcuse == 0 || dstuse >= dstmax)
1357 return;
1358 float* in = srcbuf;
1359 size_t in_u = srcuse;
1360 float* out = dstbuf + dstuse;
1361 size_t out_u = dstmax - dstuse;
1362 r.resample(in, in_u, out, out_u, ratio, false);
1363 size_t offset = in - srcbuf;
1364 if(offset < srcuse)
1365 memmove(srcbuf, srcbuf + offset, sizeof(float) * (srcuse - offset));
1366 srcuse -= offset;
1367 dstuse = dstmax - out_u;
1368 }
1369
1370 //Drain the input buffer.
1371 void drain_input()
1372 {
1373 while(audioapi_voice_r_status() > 0) {
1374 float buf[256];
1375 unsigned size = min(audioapi_voice_r_status(), 256u);
1376 audioapi_record_voice(buf, size);
1377 }
1378 }
1379
1380 //Read the input buffer.
1381 void read_input(float* buf, size_t& use, size_t maxuse)
1382 {
1383 size_t rleft = audioapi_voice_r_status();
1384 unsigned toread = min(rleft, max(maxuse, use) - use);
1385 if(toread > 0) {
1386 audioapi_record_voice(buf + use, toread);
1387 use += toread;
1388 }
1389 }
1390
1391 //Compress Opus block.
1392 void compress_opus_block(opus::encoder& e, float* buf, size_t& use, opus_stream& active_stream,
1393 bitrate_tracker& brtrack)
1394 {
1395 const size_t opus_out_max = 1276;
1396 unsigned char opus_output[opus_out_max];
1397 size_t cblock = 0;
1398 if(use >= 960)
1399 cblock = 960;
1400 else if(use >= 480)
1401 cblock = 480;
1402 else if(use >= 240)
1403 cblock = 240;
1404 else if(use >= 120)
1405 cblock = 120;
1406 else
1407 return; //No valid data to compress.
1408 const size_t opus_out_max2 = opus_max_bitrate.get() * cblock / 384000;
1409 try {
1410 size_t c = e.encode(buf, cblock, opus_output, opus_out_max2);
1411 //Successfully compressed a block.
1412 size_t opus_output_len = c;
1413 brtrack.submit(c, cblock);
1414 try {
1415 active_stream.write(cblock / 120, opus_output, opus_output_len);
1416 } catch(std::exception& e) {
1417 messages << "Error writing data: " << e.what() << std::endl;
1418 }
1419 } catch(std::exception& e) {
1420 messages << "Opus encoder error: " << e.what() << std::endl;
1421 }
1422 use -= cblock;
1423 }
1424
1425 void update_time()
1426 {
1427 uint64_t sampletime;
1428 bool jumping;
1429 {
1430 threads::alock m(time_mutex);
1431 sampletime = current_time;
1432 jumping = time_jump;
1433 time_jump = false;
1434 }
1435 if(jumping)
1436 jump_time(sampletime);
1437 else
1438 advance_time(sampletime);
1439 }
1440
1441 void decompress_active_streams(float* out, size_t& use)
1442 {
1443 size_t base = use;
1444 use += OUTPUT_BLOCK;
1445 for(unsigned i = 0; i < OUTPUT_BLOCK; i++)
1446 out[i + base] = 0;
1447 //Do it this way to minimize the amount of time playback streams lock
1448 //is held.
1449 std::list<opus_playback_stream*> stmp;
1450 {
1451 threads::alock m(active_playback_streams_lock);
1452 stmp = active_playback_streams;
1453 }
1454 std::set<opus_playback_stream*> toerase;
1455 for(auto i : stmp) {
1456 float tmp[OUTPUT_BLOCK];
1457 try {
1458 i->read(tmp, OUTPUT_BLOCK);
1459 } catch(std::exception& e) {
1460 messages << "Failed to decompress: " << e.what() << std::endl;
1461 for(unsigned j = 0; j < OUTPUT_BLOCK; j++)
1462 tmp[j] = 0;
1463 }
1464 for(unsigned j = 0; j < OUTPUT_BLOCK; j++)
1465 out[j + base] += tmp[j];
1466 if(i->eof())
1467 toerase.insert(i);
1468 }
1469 {
1470 threads::alock m(active_playback_streams_lock);
1471 for(auto i = active_playback_streams.begin(); i != active_playback_streams.end();) {
1472 if(toerase.count(*i)) {
1473 auto toerase = i;
1474 i++;
1475 delete *toerase;
1476 active_playback_streams.erase(toerase);
1477 } else
1478 i++;
1479 }
1480 }
1481 }
1482
1483 void handle_tangent_positive_edge(opus::encoder& e, opus_stream*& active_stream, bitrate_tracker& brtrack)
1484 {
1485 threads::alock m2(current_collection_lock);
1486 if(!current_collection)
1487 return;
1488 try {
1489 e.ctl(opus::reset);
1490 e.ctl(opus::bitrate(opus_bitrate.get()));
1491 brtrack.reset();
1492 uint64_t ctime;
1493 {
1494 threads::alock m(time_mutex);
1495 ctime = current_time;
1496 }
1497 active_stream = NULL;
1498 active_stream = new opus_stream(ctime, current_collection->get_filesystem());
1499 int32_t pregap = e.ctl(opus::lookahead);
1500 active_stream->set_pregap(pregap);
1501 } catch(std::exception& e) {
1502 messages << "Can't start stream: " << e.what() << std::endl;
1503 return;
1504 }
1505 messages << "Tangent enaged." << std::endl;
1506 }
1507
1508 void handle_tangent_negative_edge(opus_stream*& active_stream, bitrate_tracker& brtrack)
1509 {
1510 threads::alock m2(current_collection_lock);
1511 messages << "Tangent disenaged: " << brtrack;
1512 try {
1513 active_stream->write_trailier();
1514 } catch(std::exception& e) {
1515 messages << e.what() << std::endl;
1516 }
1517 if(current_collection) {
1518 try {
1519 current_collection->add_stream(*active_stream);
1520 } catch(std::exception& e) {
1521 messages << "Can't add stream: " << e.what() << std::endl;
1522 active_stream->put_ref();
1523 }
1524 notify_voice_stream_change();
1525 } else
1526 active_stream->put_ref();
1527 active_stream = NULL;
1528 }
1529
1530 class inthread_th : public workthread::worker
1531 {
1532 public:
1533 inthread_th()
1534 {
1535 quit = false;
1536 quit_ack = false;
1537 rptr = 0;
1538 fire();
1539 }
1540 void kill()
1541 {
1542 quit = true;
1543 {
1544 threads::alock h(lmut);
1545 lcond.notify_all();
1546 }
1547 while(!quit_ack)
1548 usleep(100000);
1549 usleep(100000);
1550 }
1551 protected:
1552 void entry()
1553 {
1554 try {
1555 entry2();
1556 } catch(std::bad_alloc& e) {
1557 OOM_panic();
1558 } catch(std::exception& e) {
1559 messages << "AIEEE... Fatal exception in voice thread: " << e.what() << std::endl;
1560 }
1561 quit_ack = true;
1562 }
1563 void entry2()
1564 {
1565 //Wait for libopus to load...
1566 size_t cbh = opus::add_callback([this]() {
1567 threads::alock h(this->lmut);
1568 this->lcond.notify_all();
1569 });
1570 while(true) {
1571 threads::alock h(lmut);
1572 if(opus::libopus_loaded() || quit)
1573 break;
1574 lcond.wait(h);
1575 }
1576 opus::cancel_callback(cbh);
1577 if(quit)
1578 return;
1579
1580 opus::encoder oenc(opus::samplerate::r48k, false, opus::application::voice);
1581 oenc.ctl(opus::bitrate(opus_bitrate.get()));
1582 audioapi_resampler rin;
1583 audioapi_resampler rout;
1584 const unsigned buf_max = 6144; //These buffers better be large.
1585 size_t buf_in_use = 0;
1586 size_t buf_inr_use = 0;
1587 size_t buf_outr_use = 0;
1588 size_t buf_out_use = 0;
1589 float buf_in[buf_max];
1590 float buf_inr[OPUS_BLOCK_SIZE];
1591 float buf_outr[OUTPUT_SIZE];
1592 float buf_out[buf_max];
1593 bitrate_tracker brtrack;
1594 opus_stream* active_stream = NULL;
1595
1596 drain_input();
1597 while(1) {
1598 if(clear_workflag(workthread::quit_request) & workthread::quit_request) {
1599 if(!active_flag && active_stream)
1600 handle_tangent_negative_edge(active_stream, brtrack);
1601 break;
1602 }
1603 uint64_t ticks = get_utime();
1604 //Handle tangent edgets.
1605 if(active_flag && !active_stream) {
1606 drain_input();
1607 buf_in_use = 0;
1608 buf_inr_use = 0;
1609 handle_tangent_positive_edge(oenc, active_stream, brtrack);
1610 }
1611 else if((!active_flag || quit) && active_stream)
1612 handle_tangent_negative_edge(active_stream, brtrack);
1613 if(quit)
1614 break;
1615
1616 //Read input, up to 25ms.
1617 unsigned rate_in = audioapi_voice_rate().first;
1618 unsigned rate_out = audioapi_voice_rate().second;
1619 size_t dbuf_max = min(buf_max, rate_in / REC_THRESHOLD_DIV);
1620 read_input(buf_in, buf_in_use, dbuf_max);
1621
1622 //Resample up to full opus block.
1623 do_resample(rin, buf_in, buf_in_use, buf_inr, buf_inr_use, OPUS_BLOCK_SIZE,
1624 1.0 * OPUS_SAMPLERATE / rate_in);
1625
1626 //If we have full opus block and recording is enabled, compress it.
1627 if(buf_inr_use >= OPUS_BLOCK_SIZE && active_stream)
1628 compress_opus_block(oenc, buf_inr, buf_inr_use, *active_stream, brtrack);
1629
1630 //Update time, starting/ending streams.
1631 update_time();
1632
1633 //Decompress active streams.
1634 if(buf_outr_use < BLOCK_THRESHOLD)
1635 decompress_active_streams(buf_outr, buf_outr_use);
1636
1637 //Resample to output rate.
1638 do_resample(rout, buf_outr, buf_outr_use, buf_out, buf_out_use, buf_max,
1639 1.0 * rate_out / OPUS_SAMPLERATE);
1640
1641 //Output stuff.
1642 if(buf_out_use > 0 && audioapi_voice_p_status2() < rate_out / PLAY_THRESHOLD_DIV) {
1643 audioapi_play_voice(buf_out, buf_out_use);
1644 buf_out_use = 0;
1645 }
1646
1647 //Sleep a bit to save CPU use.
1648 uint64_t ticks_spent = get_utime() - ticks;
1649 if(ticks_spent < ITERATION_TIME)
1650 usleep(ITERATION_TIME - ticks_spent);
1651 }
1652 threads::alock h(current_collection_lock);
1653 delete current_collection;
1654 current_collection = NULL;
1655 }
1656 private:
1657 size_t rptr;
1658 double position;
1659 volatile bool quit;
1660 volatile bool quit_ack;
1661 threads::lock lmut;
1662 threads::cv lcond;
1663 };
1664
1665 //The tangent function.
1666 command::fnptr<> ptangent(lsnes_cmd, "+tangent", "Voice tangent",
1667 "Syntax: +tangent\nVoice tangent.\n",
1668 []() throw(std::bad_alloc, std::runtime_error) {
1669 active_flag = true;
1670 });
1671 command::fnptr<> ntangent(lsnes_cmd, "-tangent", "Voice tangent",
1672 "Syntax: -tangent\nVoice tangent.\n",
1673 []() throw(std::bad_alloc, std::runtime_error) {
1674 active_flag = false;
1675 });
1676 keyboard::invbind itangent(lsnes_mapper, "+tangent", "Movie‣Voice tangent");
1677 inthread_th* int_task;
1678 }
1679
1680 //Rate is not sampling rate!
1681 void voice_frame_number(uint64_t newframe, double rate)
1682 {
1683 if(rate == last_rate && last_frame_number == newframe)
1684 return;
1685 threads::alock m(time_mutex);
1686 current_time = newframe / rate * OPUS_SAMPLERATE;
1687 if(fabs(rate - last_rate) > 1e-6 || last_frame_number + 1 != newframe)
1688 time_jump = true;
1689 last_frame_number = newframe;
1690 last_rate = rate;
1691 }
1692
1693 void voicethread_task()
1694 {
1695 int_task = new inthread_th;
1696 }
1697
1698 void voicethread_kill()
1699 {
1700 int_task->kill();
1701 delete int_task;
1702 int_task = NULL;
1703 }
1704
1705 uint64_t voicesub_parse_timebase(const std::string& n)
1706 {
1707 std::string x = n;
1708 if(x.length() > 0 && x[x.length() - 1] == 's') {
1709 x = x.substr(0, x.length() - 1);
1710 return 48000 * parse_value<double>(x);
1711 } else
1712 return parse_value<uint64_t>(x);
1713 }
1714
1715 bool voicesub_collection_loaded()
1716 {
1717 threads::alock m2(current_collection_lock);
1718 return (current_collection != NULL);
1719 }
1720
1721 std::list<playback_stream_info> voicesub_get_stream_info()
1722 {
1723 threads::alock m2(current_collection_lock);
1724 std::list<playback_stream_info> in;
1725 if(!current_collection)
1726 return in;
1727 for(auto i : current_collection->all_streams()) {
1728 opus_stream* s = current_collection->get_stream(i);
1729 playback_stream_info pi;
1730 if(!s)
1731 continue;
1732 pi.id = i;
1733 pi.base = s->timebase();
1734 pi.length = s->length();
1735 try {
1736 in.push_back(pi);
1737 } catch(...) {
1738 }
1739 s->put_ref();
1740 }
1741 return in;
1742 }
1743
1744 void voicesub_play_stream(uint64_t id)
1745 {
1746 threads::alock m2(current_collection_lock);
1747 if(!current_collection)
1748 throw std::runtime_error("No collection loaded");
1749 opus_stream* s = current_collection->get_stream(id);
1750 if(!s)
1751 return;
1752 try {
1753 start_management_stream(*s);
1754 } catch(...) {
1755 s->put_ref();
1756 throw;
1757 }
1758 s->put_ref();
1759 }
1760
1761 void voicesub_export_stream(uint64_t id, const std::string& filename, external_stream_format fmt)
1762 {
1763 threads::alock m2(current_collection_lock);
1764 if(!current_collection)
1765 throw std::runtime_error("No collection loaded");
1766 opus_stream* st = current_collection->get_stream(id);
1767 if(!st)
1768 return;
1769 std::ofstream s(filename, std::ios_base::out | std::ios_base::binary);
1770 if(!s) {
1771 st->put_ref();
1772 throw std::runtime_error("Can't open output file");
1773 }
1774 try {
1775 st->export_stream(s, fmt);
1776 } catch(std::exception& e) {
1777 st->put_ref();
1778 (stringfmt() << "Export failed: " << e.what()).throwex();
1779 }
1780 st->put_ref();
1781 }
1782
1783 uint64_t voicesub_import_stream(uint64_t ts, const std::string& filename, external_stream_format fmt)
1784 {
1785 threads::alock m2(current_collection_lock);
1786 if(!current_collection)
1787 throw std::runtime_error("No collection loaded");
1788
1789 std::ifstream s(filename, std::ios_base::in | std::ios_base::binary);
1790 if(!s)
1791 throw std::runtime_error("Can't open input file");
1792 opus_stream* st = new opus_stream(ts, current_collection->get_filesystem(), s, fmt);
1793 uint64_t id;
1794 try {
1795 id = current_collection->add_stream(*st);
1796 } catch(...) {
1797 st->delete_stream();
1798 throw;
1799 }
1800 st->unlock(); //Not locked.
1801 notify_voice_stream_change();
1802 return id;
1803 }
1804
1805 void voicesub_delete_stream(uint64_t id)
1806 {
1807 threads::alock m2(current_collection_lock);
1808 if(!current_collection)
1809 throw std::runtime_error("No collection loaded");
1810 current_collection->delete_stream(id);
1811 notify_voice_stream_change();
1812 }
1813
1814 void voicesub_export_superstream(const std::string& filename)
1815 {
1816 threads::alock m2(current_collection_lock);
1817 if(!current_collection)
1818 throw std::runtime_error("No collection loaded");
1819 std::ofstream s(filename, std::ios_base::out | std::ios_base::binary);
1820 if(!s)
1821 throw std::runtime_error("Can't open output file");
1822 current_collection->export_superstream(s);
1823 }
1824
1825 void voicesub_load_collection(const std::string& filename)
1826 {
1827 threads::alock m2(current_collection_lock);
1828 filesystem::ref newfs;
1829 stream_collection* newc;
1830 newfs = filesystem::ref(filename);
1831 newc = new stream_collection(newfs);
1832 if(current_collection)
1833 delete current_collection;
1834 current_collection = newc;
1835 notify_voice_stream_change();
1836 }
1837
1838 void voicesub_unload_collection()
1839 {
1840 threads::alock m2(current_collection_lock);
1841 if(current_collection)
1842 delete current_collection;
1843 current_collection = NULL;
1844 notify_voice_stream_change();
1845 }
1846
1847 void voicesub_alter_timebase(uint64_t id, uint64_t ts)
1848 {
1849 threads::alock m2(current_collection_lock);
1850 if(!current_collection)
1851 throw std::runtime_error("No collection loaded");
1852 current_collection->alter_stream_timebase(id, ts);
1853 notify_voice_stream_change();
1854 }
1855
1856 float voicesub_get_gain(uint64_t id)
1857 {
1858 threads::alock m2(current_collection_lock);
1859 if(!current_collection)
1860 throw std::runtime_error("No collection loaded");
1861 return current_collection->get_stream(id)->get_gain() / 256.0;
1862 }
1863
1864 void voicesub_set_gain(uint64_t id, float gain)
1865 {
1866 threads::alock m2(current_collection_lock);
1867 if(!current_collection)
1868 throw std::runtime_error("No collection loaded");
1869 int64_t _gain = gain * 256;
1870 if(_gain < -32768 || _gain > 32767)
1871 throw std::runtime_error("Gain out of range (+-128dB)");
1872 current_collection->alter_stream_gain(id, _gain);
1873 notify_voice_stream_change();
1874 }
1875
1876 double voicesub_ts_seconds(uint64_t ts)
1877 {
1878 return ts / 48000.0;
1879 }
Rerecording SNES/GB(C) emulator based on bsnes / gambatte core