]> git.sesse.net Git - nageru/blob - httpd.cpp
Fix yet another memory leak in the serving part.
[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         if (avctx->oformat->flags & AVFMT_GLOBALHEADER) {
166                 avstream_video->codec->flags = AV_CODEC_FLAG_GLOBAL_HEADER;
167         }
168
169         AVCodec *codec_audio = avcodec_find_encoder(AUDIO_OUTPUT_CODEC);
170         avstream_audio = avformat_new_stream(avctx, codec_audio);
171         if (avstream_audio == nullptr) {
172                 fprintf(stderr, "avformat_new_stream() failed\n");
173                 exit(1);
174         }
175         avstream_audio->time_base = AVRational{1, TIMEBASE};
176         avstream_audio->codec->bit_rate = AUDIO_OUTPUT_BIT_RATE;
177         avstream_audio->codec->sample_rate = OUTPUT_FREQUENCY;
178         avstream_audio->codec->sample_fmt = AUDIO_OUTPUT_SAMPLE_FMT;
179         avstream_audio->codec->channels = 2;
180         avstream_audio->codec->channel_layout = AV_CH_LAYOUT_STEREO;
181         avstream_audio->codec->time_base = AVRational{1, TIMEBASE};
182         if (avctx->oformat->flags & AVFMT_GLOBALHEADER) {
183                 avstream_audio->codec->flags = AV_CODEC_FLAG_GLOBAL_HEADER;
184         }
185
186         AVDictionary *options = NULL;
187         vector<pair<string, string>> opts = MUX_OPTS;
188         for (pair<string, string> opt : opts) {
189                 av_dict_set(&options, opt.first.c_str(), opt.second.c_str(), 0);
190         }
191         if (avformat_write_header(avctx, &options) < 0) {
192                 fprintf(stderr, "avformat_write_header() failed\n");
193                 exit(1);
194         }
195 }
196
197 HTTPD::Mux::~Mux()
198 {
199         av_write_trailer(avctx);
200         av_free(avctx->pb->buffer);
201         av_free(avctx->pb);
202         avformat_free_context(avctx);
203 }
204
205 void HTTPD::Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
206 {
207         if (!seen_keyframe && !(pkt.stream_index == 0 && (pkt.flags & AV_PKT_FLAG_KEY))) {
208                 // Wait until we see the first (video) key frame.
209                 return;
210         }
211         seen_keyframe = true;
212
213         AVPacket pkt_copy;
214         av_copy_packet(&pkt_copy, &pkt);
215         if (pkt.stream_index == 0) {
216                 pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_video->time_base);
217                 pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_video->time_base);
218         } else if (pkt.stream_index == 1) {
219                 pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_audio->time_base);
220                 pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_audio->time_base);
221         } else {
222                 assert(false);
223         }
224
225         if (av_interleaved_write_frame(avctx, &pkt_copy) < 0) {
226                 fprintf(stderr, "av_interleaved_write_frame() failed\n");
227                 exit(1);
228         }
229
230         av_packet_unref(&pkt_copy);
231 }
232
233 HTTPD::Stream::Stream(AVOutputFormat *oformat, int width, int height)
234 {
235         AVFormatContext *avctx = avformat_alloc_context();
236         avctx->oformat = oformat;
237         uint8_t *buf = (uint8_t *)av_malloc(MUX_BUFFER_SIZE);
238         avctx->pb = avio_alloc_context(buf, MUX_BUFFER_SIZE, 1, this, nullptr, &HTTPD::Stream::write_packet_thunk, nullptr);
239         avctx->flags = AVFMT_FLAG_CUSTOM_IO;
240
241         mux.reset(new Mux(avctx, width, height));
242 }
243
244 ssize_t HTTPD::Stream::reader_callback_thunk(void *cls, uint64_t pos, char *buf, size_t max)
245 {
246         HTTPD::Stream *stream = (HTTPD::Stream *)cls;
247         return stream->reader_callback(pos, buf, max);
248 }
249
250 ssize_t HTTPD::Stream::reader_callback(uint64_t pos, char *buf, size_t max)
251 {
252         unique_lock<mutex> lock(buffer_mutex);
253         has_buffered_data.wait(lock, [this]{ return !buffered_data.empty(); });
254
255         ssize_t ret = 0;
256         while (max > 0 && !buffered_data.empty()) {
257                 const string &s = buffered_data.front();
258                 assert(s.size() > used_of_buffered_data);
259                 size_t len = s.size() - used_of_buffered_data;
260                 if (max >= len) {
261                         // Consume the entire (rest of the) string.
262                         memcpy(buf, s.data() + used_of_buffered_data, len);
263                         ret += len;
264                         max -= len;
265                         buffered_data.pop_front();
266                         used_of_buffered_data = 0;
267                 } else {
268                         // We don't need the entire string; just use the first part of it.
269                         memcpy(buf, s.data() + used_of_buffered_data, max);
270                         used_of_buffered_data += max;
271                         ret += max;
272                         max = 0;
273                 }
274         }
275
276         return ret;
277 }
278
279 void HTTPD::Stream::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
280 {
281         mux->add_packet(pkt, pts, dts);
282 }
283
284 int HTTPD::Stream::write_packet_thunk(void *opaque, uint8_t *buf, int buf_size)
285 {
286         HTTPD::Stream *stream = (HTTPD::Stream *)opaque;
287         return stream->write_packet(buf, buf_size);
288 }
289
290 int HTTPD::Stream::write_packet(uint8_t *buf, int buf_size)
291 {
292         unique_lock<mutex> lock(buffer_mutex);
293         buffered_data.emplace_back((char *)buf, buf_size);
294         has_buffered_data.notify_all(); 
295         return buf_size;
296 }
297