]> git.sesse.net Git - nageru/blob - httpd.cpp
Start a mux only on a keyframe, so that transcoding clients do not get too confused.
[nageru] / httpd.cpp
1 #include <assert.h>
2 #include <microhttpd.h>
3 #include <stdio.h>
4 #include <stdlib.h>
5 #include <string.h>
6
7 extern "C" {
8 #include <libavcodec/avcodec.h>
9 #include <libavutil/channel_layout.h>
10 #include <libavutil/mathematics.h>
11 #include <libavutil/mem.h>
12 #include <libavutil/pixfmt.h>
13 #include <libavutil/rational.h>
14 #include <libavutil/samplefmt.h>
15 }
16
17 #include <vector>
18
19 #include "httpd.h"
20
21 #include "defs.h"
22 #include "timebase.h"
23
24 struct MHD_Connection;
25 struct MHD_Response;
26
27 using namespace std;
28
29 HTTPD::HTTPD(int width, int height)
30         : width(width), height(height)
31 {
32 }
33
34 void HTTPD::start(int port)
35 {
36         MHD_start_daemon(MHD_USE_THREAD_PER_CONNECTION | MHD_USE_POLL_INTERNALLY | MHD_USE_DUAL_STACK,
37                          port,
38                          nullptr, nullptr,
39                          &answer_to_connection_thunk, this,
40                          MHD_OPTION_NOTIFY_COMPLETED, &request_completed_thunk, this, 
41                          MHD_OPTION_END);
42 }
43
44 void HTTPD::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
45 {
46         unique_lock<mutex> lock(streams_mutex);
47         for (Stream *stream : streams) {
48                 stream->add_packet(pkt, pts, dts);
49         }
50         if (file_mux) {
51                 file_mux->add_packet(pkt, pts, dts);
52         }
53 }
54
55 void HTTPD::open_output_file(const string &filename)
56 {
57         AVFormatContext *avctx = avformat_alloc_context();
58         avctx->oformat = av_guess_format(NULL, filename.c_str(), NULL);
59         assert(filename.size() < sizeof(avctx->filename) - 1);
60         strcpy(avctx->filename, filename.c_str());
61
62         string url = "file:" + filename;
63         int ret = avio_open2(&avctx->pb, url.c_str(), AVIO_FLAG_WRITE, &avctx->interrupt_callback, NULL);
64         if (ret < 0) {
65                 char tmp[AV_ERROR_MAX_STRING_SIZE];
66                 fprintf(stderr, "%s: avio_open2() failed: %s\n", filename.c_str(), av_make_error_string(tmp, sizeof(tmp), ret));
67                 exit(1);
68         }
69
70         file_mux.reset(new Mux(avctx, width, height));
71 }
72
73 void HTTPD::close_output_file()
74 {
75         file_mux.reset();
76 }
77
78 int HTTPD::answer_to_connection_thunk(void *cls, MHD_Connection *connection,
79                                       const char *url, const char *method,
80                                       const char *version, const char *upload_data,
81                                       size_t *upload_data_size, void **con_cls)
82 {
83         HTTPD *httpd = (HTTPD *)cls;
84         return httpd->answer_to_connection(connection, url, method, version, upload_data, upload_data_size, con_cls);
85 }
86
87 int HTTPD::answer_to_connection(MHD_Connection *connection,
88                                 const char *url, const char *method,
89                                 const char *version, const char *upload_data,
90                                 size_t *upload_data_size, void **con_cls)
91 {
92         AVOutputFormat *oformat = av_guess_format(STREAM_MUX_NAME, nullptr, nullptr);
93         assert(oformat != nullptr);
94         HTTPD::Stream *stream = new HTTPD::Stream(oformat, width, height);
95         {
96                 unique_lock<mutex> lock(streams_mutex);
97                 streams.insert(stream);
98         }
99         *con_cls = stream;
100
101         // Does not strictly have to be equal to MUX_BUFFER_SIZE.
102         MHD_Response *response = MHD_create_response_from_callback(
103                 (size_t)-1, MUX_BUFFER_SIZE, &HTTPD::Stream::reader_callback_thunk, stream, &HTTPD::free_stream);
104         int ret = MHD_queue_response(connection, MHD_HTTP_OK, response);
105         //MHD_destroy_response(response);
106
107         return ret;
108 }
109
110 void HTTPD::free_stream(void *cls)
111 {
112         // FIXME: When is this actually called, if ever?
113         // Also, shouldn't we remove it from streams?
114         HTTPD::Stream *stream = (HTTPD::Stream *)cls;
115         delete stream;
116 }
117
118 void HTTPD::request_completed_thunk(void *cls, struct MHD_Connection *connection, void **con_cls, enum MHD_RequestTerminationCode toe)
119 {
120         HTTPD *httpd = (HTTPD *)cls;
121         return httpd->request_completed(connection, con_cls, toe);
122 }
123
124 void HTTPD::request_completed(struct MHD_Connection *connection, void **con_cls, enum MHD_RequestTerminationCode toe)
125 {
126         if (con_cls == nullptr) {
127                 // Request was never set up.
128                 return;
129         }
130         HTTPD::Stream *stream = (HTTPD::Stream *)*con_cls;
131         {
132                 unique_lock<mutex> lock(streams_mutex);
133                 delete stream;
134                 streams.erase(stream);
135         }
136 }
137
138 HTTPD::Mux::Mux(AVFormatContext *avctx, int width, int height)
139         : avctx(avctx)
140 {
141         AVCodec *codec_video = avcodec_find_encoder(AV_CODEC_ID_H264);
142         avstream_video = avformat_new_stream(avctx, codec_video);
143         if (avstream_video == nullptr) {
144                 fprintf(stderr, "avformat_new_stream() failed\n");
145                 exit(1);
146         }
147         avstream_video->time_base = AVRational{1, TIMEBASE};
148         avstream_video->codec->codec_type = AVMEDIA_TYPE_VIDEO;
149         avstream_video->codec->codec_id = AV_CODEC_ID_H264;
150         avstream_video->codec->width = width;
151         avstream_video->codec->height = height;
152         avstream_video->codec->time_base = AVRational{1, TIMEBASE};
153         avstream_video->codec->ticks_per_frame = 1;  // or 2?
154
155         // Colorspace details. Closely correspond to settings in EffectChain_finalize,
156         // as noted in each comment.
157         // Note that the H.264 stream also contains this information and depending on the
158         // mux, this might simply get ignored. See sps_rbsp().
159         avstream_video->codec->color_primaries = AVCOL_PRI_BT709;  // RGB colorspace (inout_format.color_space).
160         avstream_video->codec->color_trc = AVCOL_TRC_UNSPECIFIED;  // Gamma curve (inout_format.gamma_curve).
161         avstream_video->codec->colorspace = AVCOL_SPC_SMPTE170M;  // YUV colorspace (output_ycbcr_format.luma_coefficients).
162         avstream_video->codec->color_range = AVCOL_RANGE_MPEG;  // Full vs. limited range (output_ycbcr_format.full_range).
163         avstream_video->codec->chroma_sample_location = AVCHROMA_LOC_LEFT;  // Chroma sample location. See chroma_offset_0[] in Mixer::subsample_chroma().
164         avstream_video->codec->field_order = AV_FIELD_PROGRESSIVE;
165
166         AVCodec *codec_audio = avcodec_find_encoder(AUDIO_OUTPUT_CODEC);
167         avstream_audio = avformat_new_stream(avctx, codec_audio);
168         if (avstream_audio == nullptr) {
169                 fprintf(stderr, "avformat_new_stream() failed\n");
170                 exit(1);
171         }
172         avstream_audio->time_base = AVRational{1, TIMEBASE};
173         avstream_audio->codec->bit_rate = AUDIO_OUTPUT_BIT_RATE;
174         avstream_audio->codec->sample_rate = OUTPUT_FREQUENCY;
175         avstream_audio->codec->sample_fmt = AUDIO_OUTPUT_SAMPLE_FMT;
176         avstream_audio->codec->channels = 2;
177         avstream_audio->codec->channel_layout = AV_CH_LAYOUT_STEREO;
178         avstream_audio->codec->time_base = AVRational{1, TIMEBASE};
179
180         AVDictionary *options = NULL;
181         vector<pair<string, string>> opts = MUX_OPTS;
182         for (pair<string, string> opt : opts) {
183                 av_dict_set(&options, opt.first.c_str(), opt.second.c_str(), 0);
184         }
185         if (avformat_write_header(avctx, &options) < 0) {
186                 fprintf(stderr, "avformat_write_header() failed\n");
187                 exit(1);
188         }
189 }
190
191 HTTPD::Mux::~Mux()
192 {
193         av_write_trailer(avctx);
194         avformat_free_context(avctx);
195 }
196
197 void HTTPD::Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
198 {
199         if (!seen_keyframe && !(pkt.stream_index == 0 && (pkt.flags & AV_PKT_FLAG_KEY))) {
200                 // Wait until we see the first (video) key frame.
201                 return;
202         }
203         seen_keyframe = true;
204
205         AVPacket pkt_copy;
206         av_copy_packet(&pkt_copy, &pkt);
207         if (pkt.stream_index == 0) {
208                 pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_video->time_base);
209                 pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_video->time_base);
210         } else if (pkt.stream_index == 1) {
211                 pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_audio->time_base);
212                 pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_audio->time_base);
213         } else {
214                 assert(false);
215         }
216
217         if (av_interleaved_write_frame(avctx, &pkt_copy) < 0) {
218                 fprintf(stderr, "av_interleaved_write_frame() failed\n");
219                 exit(1);
220         }
221 }
222
223 HTTPD::Stream::Stream(AVOutputFormat *oformat, int width, int height)
224 {
225         AVFormatContext *avctx = avformat_alloc_context();
226         avctx->oformat = oformat;
227         uint8_t *buf = (uint8_t *)av_malloc(MUX_BUFFER_SIZE);
228         avctx->pb = avio_alloc_context(buf, MUX_BUFFER_SIZE, 1, this, nullptr, &HTTPD::Stream::write_packet_thunk, nullptr);
229         avctx->flags = AVFMT_FLAG_CUSTOM_IO;
230
231         mux.reset(new Mux(avctx, width, height));
232 }
233
234 ssize_t HTTPD::Stream::reader_callback_thunk(void *cls, uint64_t pos, char *buf, size_t max)
235 {
236         HTTPD::Stream *stream = (HTTPD::Stream *)cls;
237         return stream->reader_callback(pos, buf, max);
238 }
239
240 ssize_t HTTPD::Stream::reader_callback(uint64_t pos, char *buf, size_t max)
241 {
242         unique_lock<mutex> lock(buffer_mutex);
243         has_buffered_data.wait(lock, [this]{ return !buffered_data.empty(); });
244
245         ssize_t ret = 0;
246         while (max > 0 && !buffered_data.empty()) {
247                 const string &s = buffered_data.front();
248                 assert(s.size() > used_of_buffered_data);
249                 size_t len = s.size() - used_of_buffered_data;
250                 if (max >= len) {
251                         // Consume the entire (rest of the) string.
252                         memcpy(buf, s.data() + used_of_buffered_data, len);
253                         ret += len;
254                         max -= len;
255                         buffered_data.pop_front();
256                         used_of_buffered_data = 0;
257                 } else {
258                         // We don't need the entire string; just use the first part of it.
259                         memcpy(buf, s.data() + used_of_buffered_data, max);
260                         used_of_buffered_data += max;
261                         ret += max;
262                         max = 0;
263                 }
264         }
265
266         return ret;
267 }
268
269 void HTTPD::Stream::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
270 {
271         mux->add_packet(pkt, pts, dts);
272 }
273
274 int HTTPD::Stream::write_packet_thunk(void *opaque, uint8_t *buf, int buf_size)
275 {
276         HTTPD::Stream *stream = (HTTPD::Stream *)opaque;
277         return stream->write_packet(buf, buf_size);
278 }
279
280 int HTTPD::Stream::write_packet(uint8_t *buf, int buf_size)
281 {
282         unique_lock<mutex> lock(buffer_mutex);
283         buffered_data.emplace_back((char *)buf, buf_size);
284         has_buffered_data.notify_all(); 
285         return buf_size;
286 }
287