]> git.sesse.net Git - nageru/blob - httpd.cpp
Fix leaking of streams in HTTPD.
[nageru] / httpd.cpp
1 #include <arpa/inet.h>
2 #include <assert.h>
3 #include <microhttpd.h>
4 #include <stdio.h>
5 #include <stdlib.h>
6 #include <string.h>
7
8 #include <vector>
9
10 #include "httpd.h"
11
12 #include "defs.h"
13 #include "flags.h"
14 #include "metacube2.h"
15 #include "timebase.h"
16
17 struct MHD_Connection;
18 struct MHD_Response;
19
20 using namespace std;
21
22 HTTPD::HTTPD()
23 {
24 }
25
26 HTTPD::~HTTPD()
27 {
28         MHD_quiesce_daemon(mhd);
29         for (Stream *stream : streams) {
30                 stream->stop();
31         }
32         MHD_stop_daemon(mhd);
33 }
34
35 void HTTPD::start(int port)
36 {
37         mhd = MHD_start_daemon(MHD_USE_THREAD_PER_CONNECTION | MHD_USE_POLL_INTERNALLY | MHD_USE_DUAL_STACK,
38                                port,
39                                nullptr, nullptr,
40                                &answer_to_connection_thunk, this,
41                                MHD_OPTION_NOTIFY_COMPLETED, nullptr, this,
42                                MHD_OPTION_END);
43 }
44
45 void HTTPD::add_data(const char *buf, size_t size, bool keyframe)
46 {
47         unique_lock<mutex> lock(streams_mutex);
48         for (Stream *stream : streams) {
49                 stream->add_data(buf, size, keyframe ? Stream::DATA_TYPE_KEYFRAME : Stream::DATA_TYPE_OTHER);
50         }
51 }
52
53 int HTTPD::answer_to_connection_thunk(void *cls, MHD_Connection *connection,
54                                       const char *url, const char *method,
55                                       const char *version, const char *upload_data,
56                                       size_t *upload_data_size, void **con_cls)
57 {
58         HTTPD *httpd = (HTTPD *)cls;
59         return httpd->answer_to_connection(connection, url, method, version, upload_data, upload_data_size, con_cls);
60 }
61
62 int HTTPD::answer_to_connection(MHD_Connection *connection,
63                                 const char *url, const char *method,
64                                 const char *version, const char *upload_data,
65                                 size_t *upload_data_size, void **con_cls)
66 {
67         // See if the URL ends in “.metacube”.
68         HTTPD::Stream::Framing framing;
69         if (strstr(url, ".metacube") == url + strlen(url) - strlen(".metacube")) {
70                 framing = HTTPD::Stream::FRAMING_METACUBE;
71         } else {
72                 framing = HTTPD::Stream::FRAMING_RAW;
73         }
74
75         HTTPD::Stream *stream = new HTTPD::Stream(this, framing);
76         stream->add_data(header.data(), header.size(), Stream::DATA_TYPE_HEADER);
77         {
78                 unique_lock<mutex> lock(streams_mutex);
79                 streams.insert(stream);
80         }
81         *con_cls = stream;
82
83         // Does not strictly have to be equal to MUX_BUFFER_SIZE.
84         MHD_Response *response = MHD_create_response_from_callback(
85                 (size_t)-1, MUX_BUFFER_SIZE, &HTTPD::Stream::reader_callback_thunk, stream, &HTTPD::free_stream);
86
87         // TODO: Content-type?
88         if (framing == HTTPD::Stream::FRAMING_METACUBE) {
89                 MHD_add_response_header(response, "Content-encoding", "metacube");
90         }
91         int ret = MHD_queue_response(connection, MHD_HTTP_OK, response);
92         MHD_destroy_response(response);  // Only decreases the refcount; actual free is after the request is done.
93
94         return ret;
95 }
96
97 void HTTPD::free_stream(void *cls)
98 {
99         HTTPD::Stream *stream = (HTTPD::Stream *)cls;
100         HTTPD *httpd = stream->get_parent();
101         {
102                 unique_lock<mutex> lock(httpd->streams_mutex);
103                 delete stream;
104                 httpd->streams.erase(stream);
105         }
106 }
107
108 ssize_t HTTPD::Stream::reader_callback_thunk(void *cls, uint64_t pos, char *buf, size_t max)
109 {
110         HTTPD::Stream *stream = (HTTPD::Stream *)cls;
111         return stream->reader_callback(pos, buf, max);
112 }
113
114 ssize_t HTTPD::Stream::reader_callback(uint64_t pos, char *buf, size_t max)
115 {
116         unique_lock<mutex> lock(buffer_mutex);
117         has_buffered_data.wait(lock, [this]{ return should_quit || !buffered_data.empty(); });
118         if (should_quit) {
119                 return 0;
120         }
121
122         ssize_t ret = 0;
123         while (max > 0 && !buffered_data.empty()) {
124                 const string &s = buffered_data.front();
125                 assert(s.size() > used_of_buffered_data);
126                 size_t len = s.size() - used_of_buffered_data;
127                 if (max >= len) {
128                         // Consume the entire (rest of the) string.
129                         memcpy(buf, s.data() + used_of_buffered_data, len);
130                         buf += len;
131                         ret += len;
132                         max -= len;
133                         buffered_data.pop_front();
134                         used_of_buffered_data = 0;
135                 } else {
136                         // We don't need the entire string; just use the first part of it.
137                         memcpy(buf, s.data() + used_of_buffered_data, max);
138                         buf += max;
139                         used_of_buffered_data += max;
140                         ret += max;
141                         max = 0;
142                 }
143         }
144
145         return ret;
146 }
147
148 void HTTPD::Stream::add_data(const char *buf, size_t buf_size, HTTPD::Stream::DataType data_type)
149 {
150         if (buf_size == 0) {
151                 return;
152         }
153         if (data_type == DATA_TYPE_KEYFRAME) {
154                 seen_keyframe = true;
155         } else if (data_type == DATA_TYPE_OTHER && !seen_keyframe) {
156                 // Start sending only once we see a keyframe.
157                 return;
158         }
159
160         unique_lock<mutex> lock(buffer_mutex);
161
162         if (framing == FRAMING_METACUBE) {
163                 metacube2_block_header hdr;
164                 memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync));
165                 hdr.size = htonl(buf_size);
166                 int flags = 0;
167                 if (data_type == DATA_TYPE_HEADER) {
168                         flags |= METACUBE_FLAGS_HEADER;
169                 } else if (data_type == DATA_TYPE_OTHER) {
170                         flags |= METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START;
171                 }
172                 hdr.flags = htons(flags);
173                 hdr.csum = htons(metacube2_compute_crc(&hdr));
174                 buffered_data.emplace_back((char *)&hdr, sizeof(hdr));
175         }
176         buffered_data.emplace_back(buf, buf_size);
177         has_buffered_data.notify_all(); 
178 }
179
180 void HTTPD::Stream::stop()
181 {
182         unique_lock<mutex> lock(buffer_mutex);
183         should_quit = true;
184         has_buffered_data.notify_all();
185 }