+ switch (status) {
+ case RP_OUT_OF_SPACE:
+ log(WARNING, "[%s] Client sent overlong request!", client->remote_addr.c_str());
+ close_client(client);
+ return;
+ case RP_NOT_FINISHED_YET:
+ // OK, we don't have the entire header yet. Fine; we'll get it later.
+ // See if there's more data for us.
+ goto read_request_again;
+ case RP_EXTRA_DATA:
+ log(WARNING, "[%s] Junk data after request!", client->remote_addr.c_str());
+ close_client(client);
+ return;
+ case RP_FINISHED:
+ break;
+ }
+
+ assert(status == RP_FINISHED);
+
+ if (client->tls_context && !client->in_ktls_mode && tls_established(client->tls_context)) {
+ // We're ready to enter kTLS mode, unless we still have some
+ // handshake data to send (which then must be sent as non-kTLS).
+ if (send_pending_tls_data(client)) {
+ // send_pending_tls_data() hit postconditions #1 or #4.
+ return;
+ }
+ ret = tls_make_ktls(client->tls_context, client->sock);
+ if (ret < 0) {
+ log_tls_error("tls_make_ktls", ret);
+ close_client(client);
+ return;
+ }
+ client->in_ktls_mode = true;
+ }
+
+ int error_code = parse_request(client);
+ if (error_code == 200) {
+ construct_header(client);
+ } else if (error_code == 204) {
+ construct_204(client);
+ } else {
+ construct_error(client, error_code);
+ }
+
+ // We've changed states, so fall through.
+ assert(client->state == Client::SENDING_SHORT_RESPONSE ||
+ client->state == Client::SENDING_HEADER);
+ }
+ case Client::SENDING_SHORT_RESPONSE:
+ case Client::SENDING_HEADER: {
+sending_header_or_short_response_again:
+ int ret;
+ do {
+ ret = write(client->sock,
+ client->header_or_short_response.data() + client->header_or_short_response_bytes_sent,
+ client->header_or_short_response.size() - client->header_or_short_response_bytes_sent);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret == -1 && errno == EAGAIN) {
+ // We're out of socket space, so now we're at the “low edge” of epoll's
+ // edge triggering. epoll will tell us when there is more room, so for now,
+ // just return.
+ // This is postcondition #4.
+ return;
+ }
+
+ if (ret == -1) {
+ // Error! Postcondition #1.
+ log_perror("write");
+ close_client(client);
+ return;
+ }
+
+ client->header_or_short_response_bytes_sent += ret;
+ assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response.size());
+
+ if (client->header_or_short_response_bytes_sent < client->header_or_short_response.size()) {
+ // We haven't sent all yet. Fine; go another round.
+ goto sending_header_or_short_response_again;
+ }
+
+ // We're done sending the header or error! Clear it to release some memory.
+ client->header_or_short_response.clear();
+
+ if (client->state == Client::SENDING_SHORT_RESPONSE) {
+ // We're done sending the error, so now close.
+ // This is postcondition #1.
+ close_client(client);
+ return;
+ }
+
+ Stream *stream = client->stream;
+ if (client->stream_pos == size_t(-2)) {
+ // Start sending from the beginning of the backlog.
+ client->stream_pos = min<size_t>(
+ stream->bytes_received - stream->backlog_size,
+ 0);
+ client->state = Client::SENDING_DATA;
+ goto sending_data;
+ } else if (stream->prebuffering_bytes == 0) {
+ // Start sending from the first keyframe we get. In other
+ // words, we won't send any of the backlog, but we'll start
+ // sending immediately as we get the next keyframe block.
+ // Note that this is functionally identical to the next if branch,
+ // except that we save a binary search.
+ client->stream_pos = stream->bytes_received;
+ client->state = Client::WAITING_FOR_KEYFRAME;
+ } else {
+ // We're not going to send anything to the client before we have
+ // N bytes. However, this wait might be boring; we can just as well
+ // use it to send older data if we have it. We use lower_bound()
+ // so that we are conservative and never add extra latency over just
+ // waiting (assuming CBR or nearly so); otherwise, we could want e.g.
+ // 100 kB prebuffer but end up sending a 10 MB GOP.
+ deque<size_t>::const_iterator starting_point_it =
+ lower_bound(stream->suitable_starting_points.begin(),
+ stream->suitable_starting_points.end(),
+ stream->bytes_received - stream->prebuffering_bytes);
+ if (starting_point_it == stream->suitable_starting_points.end()) {
+ // None found. Just put us at the end, and then wait for the
+ // first keyframe to appear.
+ client->stream_pos = stream->bytes_received;
+ client->state = Client::WAITING_FOR_KEYFRAME;
+ } else {
+ client->stream_pos = *starting_point_it;
+ client->state = Client::PREBUFFERING;
+ goto prebuffering;
+ }
+ }
+ // Fall through.
+ }
+ case Client::WAITING_FOR_KEYFRAME: {
+ Stream *stream = client->stream;
+ if (stream->suitable_starting_points.empty() ||
+ client->stream_pos > stream->suitable_starting_points.back()) {
+ // We haven't received a keyframe since this stream started waiting,
+ // so keep on waiting for one.
+ // This is postcondition #3.
+ stream->put_client_to_sleep(client);
+ return;
+ }
+ client->stream_pos = stream->suitable_starting_points.back();
+ client->state = Client::PREBUFFERING;
+ // Fall through.
+ }
+ case Client::PREBUFFERING: {
+prebuffering:
+ Stream *stream = client->stream;
+ size_t bytes_to_send = stream->bytes_received - client->stream_pos;
+ assert(bytes_to_send <= stream->backlog_size);
+ if (bytes_to_send < stream->prebuffering_bytes) {
+ // We don't have enough bytes buffered to start this client yet.
+ // This is postcondition #3.
+ stream->put_client_to_sleep(client);
+ return;
+ }
+ client->state = Client::SENDING_DATA;
+ // Fall through.
+ }
+ case Client::SENDING_DATA: {
+sending_data:
+ skip_lost_data(client);
+ Stream *stream = client->stream;
+
+sending_data_again:
+ size_t bytes_to_send = stream->bytes_received - client->stream_pos;
+ assert(bytes_to_send <= stream->backlog_size);
+ if (bytes_to_send == 0) {
+ return;
+ }
+
+ // See if we need to split across the circular buffer.
+ bool more_data = false;
+ if ((client->stream_pos % stream->backlog_size) + bytes_to_send > stream->backlog_size) {
+ bytes_to_send = stream->backlog_size - (client->stream_pos % stream->backlog_size);
+ more_data = true;
+ }
+
+ ssize_t ret;
+ do {
+ off_t offset = client->stream_pos % stream->backlog_size;
+ ret = sendfile(client->sock, stream->data_fd, &offset, bytes_to_send);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret == -1 && errno == EAGAIN) {
+ // We're out of socket space, so return; epoll will wake us up
+ // when there is more room.
+ // This is postcondition #4.
+ return;
+ }
+ if (ret == -1) {
+ // Error, close; postcondition #1.
+ log_perror("sendfile");
+ close_client(client);
+ return;
+ }
+ client->stream_pos += ret;
+ client->bytes_sent += ret;
+
+ if (client->stream_pos == stream->bytes_received) {
+ // We don't have any more data for this client, so put it to sleep.
+ // This is postcondition #3.
+ stream->put_client_to_sleep(client);
+ } else if (more_data && size_t(ret) == bytes_to_send) {
+ goto sending_data_again;
+ }
+ break;
+ }
+ default:
+ assert(false);
+ }
+}
+
+bool Server::send_pending_tls_data(Client *client)
+{
+ // See if there's data from the TLS library to write.
+ if (client->tls_data_to_send == NULL) {
+ client->tls_data_to_send = tls_get_write_buffer(client->tls_context, &client->tls_data_left_to_send);
+ if (client->tls_data_to_send == NULL) {
+ // Really no data to send.
+ return false;
+ }
+ }
+
+send_data_again:
+ int ret;
+ do {
+ ret = write(client->sock, client->tls_data_to_send, client->tls_data_left_to_send);
+ } while (ret == -1 && errno == EINTR);
+ assert(ret < 0 || size_t(ret) <= client->tls_data_left_to_send);
+
+ if (ret == -1 && errno == EAGAIN) {
+ // We're out of socket space, so now we're at the “low edge” of epoll's
+ // edge triggering. epoll will tell us when there is more room, so for now,
+ // just return.
+ // This is postcondition #4.
+ return true;
+ }
+ if (ret == -1) {
+ // Error! Postcondition #1.
+ log_perror("write");
+ close_client(client);
+ return true;
+ }
+ if (ret > 0 && size_t(ret) == client->tls_data_left_to_send) {
+ // All data has been sent, so we don't need to go to sleep.
+ tls_buffer_clear(client->tls_context);
+ client->tls_data_to_send = NULL;
+ return false;
+ }
+
+ // More data to send, so try again.
+ client->tls_data_to_send += ret;
+ client->tls_data_left_to_send -= ret;
+ goto send_data_again;
+}
+
+int Server::read_nontls_data(Client *client, char *buf, size_t max_size)
+{
+ int ret;
+ do {
+ ret = read(client->sock, buf, max_size);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret == -1 && errno == EAGAIN) {
+ // No more data right now. Nothing to do.
+ // This is postcondition #2.
+ return -1;
+ }
+ if (ret == -1) {
+ log_perror("read");
+ close_client(client);
+ return -1;
+ }
+ if (ret == 0) {
+ // OK, the socket is closed.
+ close_client(client);
+ return -1;
+ }
+
+ return ret;
+}
+
+int Server::read_tls_data(Client *client, char *buf, size_t max_size)
+{
+read_again:
+ int ret;
+ do {
+ ret = read(client->sock, buf, max_size);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret == -1 && errno == EAGAIN) {
+ // No more data right now. Nothing to do.
+ // This is postcondition #2.
+ return -1;
+ }
+ if (ret == -1) {
+ log_perror("read");
+ close_client(client);
+ return -1;
+ }
+ if (ret == 0) {
+ // OK, the socket is closed.
+ close_client(client);
+ return -1;
+ }
+
+ // Give it to the TLS library.
+ int err = tls_consume_stream(client->tls_context, reinterpret_cast<const unsigned char *>(buf), ret, nullptr);
+ if (err < 0) {
+ log_tls_error("tls_consume_stream", err);
+ close_client(client);
+ return -1;
+ }
+ if (err == 0) {
+ // Not consumed any data. See if we can read more.
+ goto read_again;
+ }
+
+ // Read any decrypted data available for us. (We can reuse buf, since it's free now.)
+ ret = tls_read(client->tls_context, reinterpret_cast<unsigned char *>(buf), max_size);
+ if (ret == 0) {
+ // No decrypted data for us yet, but there might be some more handshaking
+ // to send. Do that if needed, then look for more data.
+ if (send_pending_tls_data(client)) {
+ // send_pending_tls_data() hit postconditions #1 or #4.
+ return -1;
+ }
+ goto read_again;
+ }
+ if (ret < 0) {
+ log_tls_error("tls_read", ret);
+ close_client(client);
+ return -1;
+ }
+
+ assert(ret > 0);
+ return ret;
+}
+
+// See if there's some data we've lost. Ideally, we should drop to a block boundary,
+// but resync will be the mux's problem.
+void Server::skip_lost_data(Client *client)