avformat/mxfdec: fix last packet timestamps
[ffmpeg.git] / libavformat / tee.c
1 /*
2  * Tee pseudo-muxer
3  * Copyright (c) 2012 Nicolas George
4  *
5  * This file is part of FFmpeg.
6  *
7  * FFmpeg is free software; you can redistribute it and/or
8  * modify it under the terms of the GNU Lesser General Public License
9  * as published by the Free Software Foundation; either
10  * version 2.1 of the License, or (at your option) any later version.
11  *
12  * FFmpeg is distributed in the hope that it will be useful,
13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  * GNU Lesser General Public License for more details.
16  *
17  * You should have received a copy of the GNU Lesser General Public License
18  * along with FFmpeg; if not, write to the Free Software * Foundation, Inc.,
19  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20  */
21
22
23 #include "libavutil/avutil.h"
24 #include "libavutil/avstring.h"
25 #include "libavutil/opt.h"
26 #include "internal.h"
27 #include "avformat.h"
28 #include "avio_internal.h"
29 #include "tee_common.h"
30
31 typedef enum {
32     ON_SLAVE_FAILURE_ABORT  = 1,
33     ON_SLAVE_FAILURE_IGNORE = 2
34 } SlaveFailurePolicy;
35
36 #define DEFAULT_SLAVE_FAILURE_POLICY ON_SLAVE_FAILURE_ABORT
37
38 typedef struct {
39     AVFormatContext *avf;
40     AVBSFContext **bsfs; ///< bitstream filters per stream
41
42     SlaveFailurePolicy on_fail;
43     int use_fifo;
44     AVDictionary *fifo_options;
45
46     /** map from input to output streams indexes,
47      * disabled output streams are set to -1 */
48     int *stream_map;
49     int header_written;
50 } TeeSlave;
51
52 typedef struct TeeContext {
53     const AVClass *class;
54     unsigned nb_slaves;
55     unsigned nb_alive;
56     TeeSlave *slaves;
57     int use_fifo;
58     AVDictionary *fifo_options;
59     char *fifo_options_str;
60 } TeeContext;
61
62 static const char *const slave_delim     = "|";
63 static const char *const slave_bsfs_spec_sep = "/";
64 static const char *const slave_select_sep = ",";
65
66 #define OFFSET(x) offsetof(TeeContext, x)
67 static const AVOption options[] = {
68         {"use_fifo", "Use fifo pseudo-muxer to separate actual muxers from encoder",
69          OFFSET(use_fifo), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
70         {"fifo_options", "fifo pseudo-muxer options", OFFSET(fifo_options_str),
71          AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
72         {NULL}
73 };
74
75 static const AVClass tee_muxer_class = {
76     .class_name = "Tee muxer",
77     .item_name  = av_default_item_name,
78     .option = options,
79     .version    = LIBAVUTIL_VERSION_INT,
80 };
81
82 static inline int parse_slave_failure_policy_option(const char *opt, TeeSlave *tee_slave)
83 {
84     if (!opt) {
85         tee_slave->on_fail = DEFAULT_SLAVE_FAILURE_POLICY;
86         return 0;
87     } else if (!av_strcasecmp("abort", opt)) {
88         tee_slave->on_fail = ON_SLAVE_FAILURE_ABORT;
89         return 0;
90     } else if (!av_strcasecmp("ignore", opt)) {
91         tee_slave->on_fail = ON_SLAVE_FAILURE_IGNORE;
92         return 0;
93     }
94     /* Set failure behaviour to abort, so invalid option error will not be ignored */
95     tee_slave->on_fail = ON_SLAVE_FAILURE_ABORT;
96     return AVERROR(EINVAL);
97 }
98
99 static int parse_slave_fifo_options(const char *use_fifo,
100                                     const char *fifo_options, TeeSlave *tee_slave)
101 {
102     int ret = 0;
103
104     if (use_fifo) {
105         /*TODO - change this to use proper function for parsing boolean
106          *       options when there is one */
107         if (av_match_name(use_fifo, "true,y,yes,enable,enabled,on,1")) {
108             tee_slave->use_fifo = 1;
109         } else if (av_match_name(use_fifo, "false,n,no,disable,disabled,off,0")) {
110             tee_slave->use_fifo = 0;
111         } else {
112             return AVERROR(EINVAL);
113         }
114     }
115
116     if (fifo_options)
117         ret = av_dict_parse_string(&tee_slave->fifo_options, fifo_options, "=", ":", 0);
118
119     return ret;
120 }
121
122 static int close_slave(TeeSlave *tee_slave)
123 {
124     AVFormatContext *avf;
125     unsigned i;
126     int ret = 0;
127
128     avf = tee_slave->avf;
129     if (!avf)
130         return 0;
131
132     if (tee_slave->header_written)
133         ret = av_write_trailer(avf);
134
135     if (tee_slave->bsfs) {
136         for (i = 0; i < avf->nb_streams; ++i)
137             av_bsf_free(&tee_slave->bsfs[i]);
138     }
139     av_freep(&tee_slave->stream_map);
140     av_freep(&tee_slave->bsfs);
141
142     ff_format_io_close(avf, &avf->pb);
143     avformat_free_context(avf);
144     tee_slave->avf = NULL;
145     return ret;
146 }
147
148 static void close_slaves(AVFormatContext *avf)
149 {
150     TeeContext *tee = avf->priv_data;
151     unsigned i;
152
153     for (i = 0; i < tee->nb_slaves; i++) {
154         close_slave(&tee->slaves[i]);
155     }
156     av_freep(&tee->slaves);
157 }
158
159 static int open_slave(AVFormatContext *avf, char *slave, TeeSlave *tee_slave)
160 {
161     int i, ret;
162     AVDictionary *options = NULL;
163     AVDictionaryEntry *entry;
164     char *filename;
165     char *format = NULL, *select = NULL, *on_fail = NULL;
166     char *use_fifo = NULL, *fifo_options_str = NULL;
167     AVFormatContext *avf2 = NULL;
168     AVStream *st, *st2;
169     int stream_count;
170     int fullret;
171     char *subselect = NULL, *next_subselect = NULL, *first_subselect = NULL, *tmp_select = NULL;
172
173     if ((ret = ff_tee_parse_slave_options(avf, slave, &options, &filename)) < 0)
174         return ret;
175
176 #define STEAL_OPTION(option, field) do {                                \
177         if ((entry = av_dict_get(options, option, NULL, 0))) {          \
178             field = entry->value;                                       \
179             entry->value = NULL; /* prevent it from being freed */      \
180             av_dict_set(&options, option, NULL, 0);                     \
181         }                                                               \
182     } while (0)
183
184     STEAL_OPTION("f", format);
185     STEAL_OPTION("select", select);
186     STEAL_OPTION("onfail", on_fail);
187     STEAL_OPTION("use_fifo", use_fifo);
188     STEAL_OPTION("fifo_options", fifo_options_str);
189
190     ret = parse_slave_failure_policy_option(on_fail, tee_slave);
191     if (ret < 0) {
192         av_log(avf, AV_LOG_ERROR,
193                "Invalid onfail option value, valid options are 'abort' and 'ignore'\n");
194         goto end;
195     }
196
197     ret = parse_slave_fifo_options(use_fifo, fifo_options_str, tee_slave);
198     if (ret < 0) {
199         av_log(avf, AV_LOG_ERROR, "Error parsing fifo options: %s\n", av_err2str(ret));
200         goto end;
201     }
202
203     if (tee_slave->use_fifo) {
204
205         if (options) {
206             char *format_options_str = NULL;
207             ret = av_dict_get_string(options, &format_options_str, '=', ':');
208             if (ret < 0)
209                 goto end;
210
211             ret = av_dict_set(&tee_slave->fifo_options, "format_opts", format_options_str,
212                               AV_DICT_DONT_STRDUP_VAL);
213             if (ret < 0)
214                 goto end;
215         }
216
217         if (format) {
218             ret = av_dict_set(&tee_slave->fifo_options, "fifo_format", format,
219                               AV_DICT_DONT_STRDUP_VAL);
220             format = NULL;
221             if (ret < 0)
222                 goto end;
223         }
224
225         av_dict_free(&options);
226         options = tee_slave->fifo_options;
227     }
228     ret = avformat_alloc_output_context2(&avf2, NULL,
229                                          tee_slave->use_fifo ? "fifo" :format, filename);
230     if (ret < 0)
231         goto end;
232     tee_slave->avf = avf2;
233     av_dict_copy(&avf2->metadata, avf->metadata, 0);
234     avf2->opaque   = avf->opaque;
235     avf2->io_open  = avf->io_open;
236     avf2->io_close = avf->io_close;
237     avf2->interrupt_callback = avf->interrupt_callback;
238     avf2->flags = avf->flags;
239
240     tee_slave->stream_map = av_calloc(avf->nb_streams, sizeof(*tee_slave->stream_map));
241     if (!tee_slave->stream_map) {
242         ret = AVERROR(ENOMEM);
243         goto end;
244     }
245
246     stream_count = 0;
247     for (i = 0; i < avf->nb_streams; i++) {
248         st = avf->streams[i];
249         if (select) {
250             tmp_select = av_strdup(select);  // av_strtok is destructive so we regenerate it in each loop
251             if (!tmp_select) {
252                 ret = AVERROR(ENOMEM);
253                 goto end;
254             }
255             fullret = 0;
256             first_subselect = tmp_select;
257             next_subselect = NULL;
258             while (subselect = av_strtok(first_subselect, slave_select_sep, &next_subselect)) {
259                 first_subselect = NULL;
260
261                 ret = avformat_match_stream_specifier(avf, avf->streams[i], subselect);
262                 if (ret < 0) {
263                     av_log(avf, AV_LOG_ERROR,
264                            "Invalid stream specifier '%s' for output '%s'\n",
265                            subselect, slave);
266                     goto end;
267                 }
268                 if (ret != 0) {
269                     fullret = 1; // match
270                     break;
271                 }
272             }
273             av_freep(&tmp_select);
274
275             if (fullret == 0) { /* no match */
276                 tee_slave->stream_map[i] = -1;
277                 continue;
278             }
279         }
280         tee_slave->stream_map[i] = stream_count++;
281
282         if (!(st2 = avformat_new_stream(avf2, NULL))) {
283             ret = AVERROR(ENOMEM);
284             goto end;
285         }
286
287         ret = ff_stream_encode_params_copy(st2, st);
288         if (ret < 0)
289             goto end;
290     }
291
292     ret = ff_format_output_open(avf2, filename, NULL);
293     if (ret < 0) {
294         av_log(avf, AV_LOG_ERROR, "Slave '%s': error opening: %s\n", slave,
295                av_err2str(ret));
296         goto end;
297     }
298
299     if ((ret = avformat_write_header(avf2, &options)) < 0) {
300         av_log(avf, AV_LOG_ERROR, "Slave '%s': error writing header: %s\n",
301                slave, av_err2str(ret));
302         goto end;
303     }
304     tee_slave->header_written = 1;
305
306     tee_slave->bsfs = av_calloc(avf2->nb_streams, sizeof(*tee_slave->bsfs));
307     if (!tee_slave->bsfs) {
308         ret = AVERROR(ENOMEM);
309         goto end;
310     }
311
312     entry = NULL;
313     while (entry = av_dict_get(options, "bsfs", NULL, AV_DICT_IGNORE_SUFFIX)) {
314         const char *spec = entry->key + strlen("bsfs");
315         if (*spec) {
316             if (strspn(spec, slave_bsfs_spec_sep) != 1) {
317                 av_log(avf, AV_LOG_ERROR,
318                        "Specifier separator in '%s' is '%c', but only characters '%s' "
319                        "are allowed\n", entry->key, *spec, slave_bsfs_spec_sep);
320                 ret = AVERROR(EINVAL);
321                 goto end;
322             }
323             spec++; /* consume separator */
324         }
325
326         for (i = 0; i < avf2->nb_streams; i++) {
327             ret = avformat_match_stream_specifier(avf2, avf2->streams[i], spec);
328             if (ret < 0) {
329                 av_log(avf, AV_LOG_ERROR,
330                        "Invalid stream specifier '%s' in bsfs option '%s' for slave "
331                        "output '%s'\n", spec, entry->key, filename);
332                 goto end;
333             }
334
335             if (ret > 0) {
336                 av_log(avf, AV_LOG_DEBUG, "spec:%s bsfs:%s matches stream %d of slave "
337                        "output '%s'\n", spec, entry->value, i, filename);
338                 if (tee_slave->bsfs[i]) {
339                     av_log(avf, AV_LOG_WARNING,
340                            "Duplicate bsfs specification associated to stream %d of slave "
341                            "output '%s', filters will be ignored\n", i, filename);
342                     continue;
343                 }
344                 ret = av_bsf_list_parse_str(entry->value, &tee_slave->bsfs[i]);
345                 if (ret < 0) {
346                     av_log(avf, AV_LOG_ERROR,
347                            "Error parsing bitstream filter sequence '%s' associated to "
348                            "stream %d of slave output '%s'\n", entry->value, i, filename);
349                     goto end;
350                 }
351             }
352         }
353
354         av_dict_set(&options, entry->key, NULL, 0);
355     }
356
357     for (i = 0; i < avf->nb_streams; i++){
358         int target_stream = tee_slave->stream_map[i];
359         if (target_stream < 0)
360             continue;
361
362         if (!tee_slave->bsfs[target_stream]) {
363             /* Add pass-through bitstream filter */
364             ret = av_bsf_get_null_filter(&tee_slave->bsfs[target_stream]);
365             if (ret < 0) {
366                 av_log(avf, AV_LOG_ERROR,
367                        "Failed to create pass-through bitstream filter: %s\n",
368                        av_err2str(ret));
369                 goto end;
370             }
371         }
372
373         tee_slave->bsfs[target_stream]->time_base_in = avf->streams[i]->time_base;
374         ret = avcodec_parameters_copy(tee_slave->bsfs[target_stream]->par_in,
375                                       avf->streams[i]->codecpar);
376         if (ret < 0)
377             goto end;
378
379         ret = av_bsf_init(tee_slave->bsfs[target_stream]);
380         if (ret < 0) {
381             av_log(avf, AV_LOG_ERROR,
382             "Failed to initialize bitstream filter(s): %s\n",
383             av_err2str(ret));
384             goto end;
385         }
386     }
387
388     if (options) {
389         entry = NULL;
390         while ((entry = av_dict_get(options, "", entry, AV_DICT_IGNORE_SUFFIX)))
391             av_log(avf2, AV_LOG_ERROR, "Unknown option '%s'\n", entry->key);
392         ret = AVERROR_OPTION_NOT_FOUND;
393         goto end;
394     }
395
396 end:
397     av_free(format);
398     av_free(select);
399     av_free(on_fail);
400     av_dict_free(&options);
401     av_freep(&tmp_select);
402     return ret;
403 }
404
405 static void log_slave(TeeSlave *slave, void *log_ctx, int log_level)
406 {
407     int i;
408     av_log(log_ctx, log_level, "filename:'%s' format:%s\n",
409            slave->avf->filename, slave->avf->oformat->name);
410     for (i = 0; i < slave->avf->nb_streams; i++) {
411         AVStream *st = slave->avf->streams[i];
412         AVBSFContext *bsf = slave->bsfs[i];
413         const char *bsf_name;
414
415         av_log(log_ctx, log_level, "    stream:%d codec:%s type:%s",
416                i, avcodec_get_name(st->codecpar->codec_id),
417                av_get_media_type_string(st->codecpar->codec_type));
418
419         bsf_name = bsf->filter->priv_class ?
420                    bsf->filter->priv_class->item_name(bsf) : bsf->filter->name;
421         av_log(log_ctx, log_level, " bsfs: %s\n", bsf_name);
422     }
423 }
424
425 static int tee_process_slave_failure(AVFormatContext *avf, unsigned slave_idx, int err_n)
426 {
427     TeeContext *tee = avf->priv_data;
428     TeeSlave *tee_slave = &tee->slaves[slave_idx];
429
430     tee->nb_alive--;
431
432     close_slave(tee_slave);
433
434     if (!tee->nb_alive) {
435         av_log(avf, AV_LOG_ERROR, "All tee outputs failed.\n");
436         return err_n;
437     } else if (tee_slave->on_fail == ON_SLAVE_FAILURE_ABORT) {
438         av_log(avf, AV_LOG_ERROR, "Slave muxer #%u failed, aborting.\n", slave_idx);
439         return err_n;
440     } else {
441         av_log(avf, AV_LOG_ERROR, "Slave muxer #%u failed: %s, continuing with %u/%u slaves.\n",
442                slave_idx, av_err2str(err_n), tee->nb_alive, tee->nb_slaves);
443         return 0;
444     }
445 }
446
447 static int tee_write_header(AVFormatContext *avf)
448 {
449     TeeContext *tee = avf->priv_data;
450     unsigned nb_slaves = 0, i;
451     const char *filename = avf->filename;
452     char **slaves = NULL;
453     int ret;
454
455     while (*filename) {
456         char *slave = av_get_token(&filename, slave_delim);
457         if (!slave) {
458             ret = AVERROR(ENOMEM);
459             goto fail;
460         }
461         ret = av_dynarray_add_nofree(&slaves, &nb_slaves, slave);
462         if (ret < 0) {
463             av_free(slave);
464             goto fail;
465         }
466         if (strspn(filename, slave_delim))
467             filename++;
468     }
469
470     if (tee->fifo_options_str) {
471         ret = av_dict_parse_string(&tee->fifo_options, tee->fifo_options_str, "=", ":", 0);
472         if (ret < 0)
473             goto fail;
474     }
475
476     if (!(tee->slaves = av_mallocz_array(nb_slaves, sizeof(*tee->slaves)))) {
477         ret = AVERROR(ENOMEM);
478         goto fail;
479     }
480     tee->nb_slaves = tee->nb_alive = nb_slaves;
481
482     for (i = 0; i < nb_slaves; i++) {
483
484         tee->slaves[i].use_fifo = tee->use_fifo;
485         ret = av_dict_copy(&tee->slaves[i].fifo_options, tee->fifo_options, 0);
486         if (ret < 0)
487             goto fail;
488
489         if ((ret = open_slave(avf, slaves[i], &tee->slaves[i])) < 0) {
490             ret = tee_process_slave_failure(avf, i, ret);
491             if (ret < 0)
492                 goto fail;
493         } else {
494             log_slave(&tee->slaves[i], avf, AV_LOG_VERBOSE);
495         }
496         av_freep(&slaves[i]);
497     }
498
499     for (i = 0; i < avf->nb_streams; i++) {
500         int j, mapped = 0;
501         for (j = 0; j < tee->nb_slaves; j++)
502             if (tee->slaves[j].avf)
503                 mapped += tee->slaves[j].stream_map[i] >= 0;
504         if (!mapped)
505             av_log(avf, AV_LOG_WARNING, "Input stream #%d is not mapped "
506                    "to any slave.\n", i);
507     }
508     av_free(slaves);
509     return 0;
510
511 fail:
512     for (i = 0; i < nb_slaves; i++)
513         av_freep(&slaves[i]);
514     close_slaves(avf);
515     av_free(slaves);
516     return ret;
517 }
518
519 static int tee_write_trailer(AVFormatContext *avf)
520 {
521     TeeContext *tee = avf->priv_data;
522     int ret_all = 0, ret;
523     unsigned i;
524
525     for (i = 0; i < tee->nb_slaves; i++) {
526         if ((ret = close_slave(&tee->slaves[i])) < 0) {
527             ret = tee_process_slave_failure(avf, i, ret);
528             if (!ret_all && ret < 0)
529                 ret_all = ret;
530         }
531     }
532     av_freep(&tee->slaves);
533     return ret_all;
534 }
535
536 static int tee_write_packet(AVFormatContext *avf, AVPacket *pkt)
537 {
538     TeeContext *tee = avf->priv_data;
539     AVFormatContext *avf2;
540     AVBSFContext *bsfs;
541     AVPacket pkt2;
542     int ret_all = 0, ret;
543     unsigned i, s;
544     int s2;
545
546     for (i = 0; i < tee->nb_slaves; i++) {
547         if (!(avf2 = tee->slaves[i].avf))
548             continue;
549
550         /* Flush slave if pkt is NULL*/
551         if (!pkt) {
552             ret = av_interleaved_write_frame(avf2, NULL);
553             if (ret < 0) {
554                 ret = tee_process_slave_failure(avf, i, ret);
555                 if (!ret_all && ret < 0)
556                     ret_all = ret;
557             }
558             continue;
559         }
560
561         s = pkt->stream_index;
562         s2 = tee->slaves[i].stream_map[s];
563         if (s2 < 0)
564             continue;
565
566         memset(&pkt2, 0, sizeof(AVPacket));
567         if ((ret = av_packet_ref(&pkt2, pkt)) < 0)
568             if (!ret_all) {
569                 ret_all = ret;
570                 continue;
571             }
572         bsfs = tee->slaves[i].bsfs[s2];
573         pkt2.stream_index = s2;
574
575         ret = av_bsf_send_packet(bsfs, &pkt2);
576         if (ret < 0) {
577             av_log(avf, AV_LOG_ERROR, "Error while sending packet to bitstream filter: %s\n",
578                    av_err2str(ret));
579             ret = tee_process_slave_failure(avf, i, ret);
580             if (!ret_all && ret < 0)
581                 ret_all = ret;
582         }
583
584         while(1) {
585             ret = av_bsf_receive_packet(bsfs, &pkt2);
586             if (ret == AVERROR(EAGAIN)) {
587                 ret = 0;
588                 break;
589             } else if (ret < 0) {
590                 break;
591             }
592
593             av_packet_rescale_ts(&pkt2, bsfs->time_base_out,
594                                  avf2->streams[s2]->time_base);
595             ret = av_interleaved_write_frame(avf2, &pkt2);
596             if (ret < 0)
597                 break;
598         };
599
600         if (ret < 0) {
601             ret = tee_process_slave_failure(avf, i, ret);
602             if (!ret_all && ret < 0)
603                 ret_all = ret;
604         }
605     }
606     return ret_all;
607 }
608
609 AVOutputFormat ff_tee_muxer = {
610     .name              = "tee",
611     .long_name         = NULL_IF_CONFIG_SMALL("Multiple muxer tee"),
612     .priv_data_size    = sizeof(TeeContext),
613     .write_header      = tee_write_header,
614     .write_trailer     = tee_write_trailer,
615     .write_packet      = tee_write_packet,
616     .priv_class        = &tee_muxer_class,
617     .flags             = AVFMT_NOFILE | AVFMT_ALLOW_FLUSH,
618 };