diff --git a/recording-daemon/decoder.c b/recording-daemon/decoder.c index 9e123a3d9..da7a1ce2f 100644 --- a/recording-daemon/decoder.c +++ b/recording-daemon/decoder.c @@ -1,6 +1,8 @@ #include "decoder.h" #include #include +#include +#include #include #include #include "types.h" @@ -28,6 +30,10 @@ struct output_s { AVFormatContext *fmtctx; AVStream *avst; AVPacket avpkt; + AVAudioFifo *fifo; + int64_t fifo_pts; // pts of first data in fifo + int64_t mux_dts; // last dts passed to muxer + AVFrame *frame; }; @@ -77,6 +83,11 @@ typedef struct decoder_def_s decoder_def_t; +static int output_codec_id; +static const char *output_file_format; + + + static void output_shutdown(output_t *output); static int output_config(output_t *output, unsigned int clockrate, unsigned int channels); @@ -151,38 +162,80 @@ err: } -static int output_add(output_t *output, AVFrame *frame) { - if (!output) - return -1; +static int output_flush(output_t *output) { + while (av_audio_fifo_size(output->fifo) >= output->frame->nb_samples) { -#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 0, 0) - int ret = avcodec_send_frame(output->avcctx, frame); - dbg("send frame ret %i", ret); - if (ret) - return -1; + if (av_audio_fifo_read(output->fifo, (void **) output->frame->data, + output->frame->nb_samples) <= 0) + abort(); - ret = avcodec_receive_packet(output->avcctx, &output->avpkt); - dbg("receive packet ret %i", ret); - if (ret) - return -1; + dbg("%p output fifo pts %lu", output, (unsigned long) output->fifo_pts); + output->frame->pts = output->fifo_pts; + +#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 0, 0) + int ret = avcodec_send_frame(output->avcctx, output->frame); + dbg("%p send frame ret %i", output, ret); + if (ret) + return -1; + + ret = avcodec_receive_packet(output->avcctx, &output->avpkt); + dbg("%p receive packet ret %i", output, ret); + if (ret) + return -1; #else - int got_packet = 0; - int ret = avcodec_encode_audio2(output->avcctx, &output->avpkt, frame, &got_packet); - dbg("encode frame ret %i, got packet %i", ret, got_packet); - if (!got_packet) - return 0; + int got_packet = 0; + int ret = avcodec_encode_audio2(output->avcctx, &output->avpkt, output->frame, &got_packet); + dbg("%p encode frame ret %i, got packet %i", output, ret, got_packet); + if (!got_packet) + return 0; #endif - av_write_frame(output->fmtctx, &output->avpkt); + dbg("%p output avpkt size is %i", output, (int) output->avpkt.size); + dbg("%p output pkt pts/dts is %li/%li", output, (long) output->avpkt.pts, + (long) output->avpkt.dts); + dbg("%p output dts %li", output, (long) output->mux_dts); + + // the encoder may return frames with the same dts multiple consecutive times. + // the muxer may not like this, so ensure monotonically increasing dts. + if (output->mux_dts > output->avpkt.dts) + output->avpkt.dts = output->mux_dts; + if (output->avpkt.pts < output->avpkt.dts) + output->avpkt.pts = output->avpkt.dts; + + av_write_frame(output->fmtctx, &output->avpkt); + + output->fifo_pts += output->frame->nb_samples; + output->mux_dts = output->avpkt.dts + 1; // min next expected dts + } return 0; } +static int output_add(output_t *output, AVFrame *frame) { + if (!output) + return -1; + + dbg("%p output fifo size %u fifo_pts %lu", output, (unsigned int) av_audio_fifo_size(output->fifo), + (unsigned long) output->fifo_pts); + // fix up output pts + if (av_audio_fifo_size(output->fifo) == 0) + output->fifo_pts = frame->pts; + + if (av_audio_fifo_write(output->fifo, (void **) frame->extended_data, frame->nb_samples) < 0) + return -1; + + return output_flush(output); +} + + int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *output) { if (G_UNLIKELY(!dec)) return -1; + dbg("%p dec pts %lu rtp_ts %lu incoming ts %lu", dec, (unsigned long) dec->pts, + (unsigned long) dec->rtp_ts, (unsigned long) ts); + if (G_UNLIKELY(dec->rtp_ts == (unsigned long) -1L)) { // initialize pts dec->pts = 0; @@ -217,9 +270,12 @@ int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *o #endif dec->frame->pts = dec->frame->pkt_pts; + dbg("%p dec frame pts %lu pkt_pts %lu", dec, (unsigned long) dec->frame->pts, + (unsigned long) dec->frame->pkt_dts); output_config(output, dec->avcctx->sample_rate, dec->avcctx->channels); - output_add(output, dec->frame); + if (output_add(output, dec->frame)) + return -1; return 0; } @@ -227,9 +283,11 @@ int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *o output_t *output_new(const char *filename) { output_t *ret = g_slice_alloc0(sizeof(*ret)); - ret->filename = strdup(filename); + if (asprintf(&ret->filename, "%s.%s", filename, output_file_format) <= 0) + abort(); ret->clockrate = -1; ret->channels = -1; + ret->frame = av_frame_alloc(); return ret; } @@ -255,11 +313,11 @@ format_mismatch: output->fmtctx = avformat_alloc_context(); if (!output->fmtctx) goto err; - output->fmtctx->oformat = av_guess_format("wav", NULL, NULL); // XXX better way? + output->fmtctx->oformat = av_guess_format(output_file_format, NULL, NULL); if (!output->fmtctx->oformat) goto err; - AVCodec *codec = avcodec_find_encoder(AV_CODEC_ID_PCM_S16LE); + AVCodec *codec = avcodec_find_encoder(output_codec_id); // XXX error handling output->avst = avformat_new_stream(output->fmtctx, codec); if (!output->avst) @@ -273,6 +331,7 @@ format_mismatch: #endif output->avcctx->channels = output->channels; + output->avcctx->channel_layout = av_get_default_channel_layout(output->channels); output->avcctx->sample_rate = output->clockrate; output->avcctx->sample_fmt = AV_SAMPLE_FMT_S16; output->avcctx->time_base = (AVRational){output->clockrate,1}; @@ -294,6 +353,19 @@ format_mismatch: av_init_packet(&output->avpkt); + // output frame and fifo + output->frame->nb_samples = output->avcctx->frame_size ? : 256; + output->frame->format = output->avcctx->sample_fmt; + output->frame->sample_rate = output->avcctx->sample_rate; + output->frame->channel_layout = output->avcctx->channel_layout; + if (!output->frame->channel_layout) + output->frame->channel_layout = av_get_default_channel_layout(output->avcctx->channels); + if (av_frame_get_buffer(output->frame, 0) < 0) + abort(); + + output->fifo = av_audio_fifo_alloc(output->avcctx->sample_fmt, output->avcctx->channels, + output->frame->nb_samples); + return 0; err: @@ -318,10 +390,13 @@ static void output_shutdown(output_t *output) { avcodec_close(output->avcctx); avio_closep(&output->fmtctx->pb); avformat_free_context(output->fmtctx); + av_audio_fifo_free(output->fifo); + av_frame_free(&output->frame); output->avcctx = NULL; output->fmtctx = NULL; output->avst = NULL; + output->fifo = NULL; } @@ -332,3 +407,17 @@ void output_close(output_t *output) { free(output->filename); g_slice_free1(sizeof(*output), output); } + + +void output_init(const char *format) { + if (!strcmp(format, "wav")) { + output_codec_id = AV_CODEC_ID_PCM_S16LE; + output_file_format = "wav"; + } + else if (!strcmp(format, "mp3")) { + output_codec_id = AV_CODEC_ID_MP3; + output_file_format = "mp3"; + } + else + die("Unknown output format '%s'", format); +} diff --git a/recording-daemon/decoder.h b/recording-daemon/decoder.h index 5c3d516a6..126b6a42b 100644 --- a/recording-daemon/decoder.h +++ b/recording-daemon/decoder.h @@ -5,6 +5,8 @@ #include "str.h" +void output_init(const char *format); + decoder_t *decoder_new(const char *payload_str); int decoder_input(decoder_t *, const str *, unsigned long ts, output_t *); void decoder_close(decoder_t *); diff --git a/recording-daemon/epoll.c b/recording-daemon/epoll.c index b03933ba6..5386fc3bc 100644 --- a/recording-daemon/epoll.c +++ b/recording-daemon/epoll.c @@ -42,8 +42,11 @@ void *poller_thread(void *ptr) { int ret = epoll_wait(epoll_fd, &epev, 1, 10000); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); - if (ret == -1) + if (ret == -1) { + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) + continue; die_errno("epoll_wait failed"); + } if (ret > 0) { dbg("thread %u handling event", me_num); diff --git a/recording-daemon/main.c b/recording-daemon/main.c index e1757033e..0b0adae02 100644 --- a/recording-daemon/main.c +++ b/recording-daemon/main.c @@ -17,6 +17,7 @@ #include "garbage.h" #include "loglib.h" #include "auxlib.h" +#include "decoder.h" @@ -24,6 +25,7 @@ int ktable = 0; int num_threads = 8; const char *spool_dir = "/var/spool/rtpengine"; const char *output_dir = "/var/lib/rtpengine-recording"; +static const char *output_format = "wav"; static GQueue threads = G_QUEUE_INIT; // only accessed from main thread @@ -44,11 +46,19 @@ static void signals(void) { static void avlog_ilog(void *ptr, int loglevel, const char *fmt, va_list ap) { - __vpilog(loglevel, NULL, fmt, ap); + char *msg; + if (vasprintf(&msg, fmt, ap) <= 0) + ilog(LOG_ERR, "av_log message dropped"); + else { + ilog(MAX(LOG_ERR, loglevel), "av_log: %s", msg); + free(msg); + } } static void setup(void) { + openlog("rtpengine-recording", LOG_PID | LOG_NDELAY, LOG_DAEMON); + log_init(); av_register_all(); avcodec_register_all(); @@ -58,7 +68,7 @@ static void setup(void) { epoll_setup(); inotify_setup(); av_log_set_callback(avlog_ilog); - openlog("rtpengine-recording", LOG_PID | LOG_NDELAY, LOG_DAEMON); + output_init(output_format); if (!g_file_test(output_dir, G_FILE_TEST_IS_DIR)) { ilog(LOG_INFO, "Creating output dir '%s'", output_dir); @@ -123,6 +133,7 @@ static void options(int *argc, char ***argv) { { "table", 't', 0, G_OPTION_ARG_INT, &ktable, "Kernel table rtpengine uses", "INT" }, { "spool-dir", 0, 0, G_OPTION_ARG_STRING, &spool_dir, "Directory containing rtpengine metadata files", "PATH" }, { "output-dir", 0, 0, G_OPTION_ARG_STRING, &output_dir, "Where to write media files to", "PATH" }, + { "output-format", 0, 0, G_OPTION_ARG_STRING, &output_format, "Write audio files of this type", "wav|mp3" }, { "num-threads", 0, 0, G_OPTION_ARG_INT, &num_threads, "Number of worker threads", "INT" }, { NULL, } }; diff --git a/recording-daemon/packet.c b/recording-daemon/packet.c index c0d476648..8f98595f7 100644 --- a/recording-daemon/packet.c +++ b/recording-daemon/packet.c @@ -56,7 +56,7 @@ static ssrc_t *ssrc_get(metafile_t *mf, unsigned long ssrc) { ret->seq = -1; char buf[256]; - snprintf(buf, sizeof(buf), "%s/%s-%08lx.wav", output_dir, mf->parent, ssrc); + snprintf(buf, sizeof(buf), "%s/%s-%08lx", output_dir, mf->parent, ssrc); ret->output = output_new(buf); g_hash_table_insert(mf->ssrc_hash, GUINT_TO_POINTER(ssrc), ret); @@ -161,8 +161,9 @@ have_packet:; } } - decoder_input(ssrc->decoders[payload_type], &packet->payload, ntohl(packet->rtp->timestamp), - ssrc->output); + if (decoder_input(ssrc->decoders[payload_type], &packet->payload, ntohl(packet->rtp->timestamp), + ssrc->output)) + ilog(LOG_ERR, "Failed to decode media packet"); next_packet: ssrc->seq = (packet->seq + 1) & 0xffff;