#include "accesslog.h"
#include "log.h"
#include "markpool.h"
-#include "metacube.h"
+#include "metacube2.h"
#include "mutexlock.h"
#include "parse.h"
#include "server.h"
#include "stream.h"
#include "util.h"
+#ifndef SO_MAX_PACING_RATE
+#define SO_MAX_PACING_RATE 47
+#endif
+
using namespace std;
extern AccessLogThread *access_log;
Server::Server()
{
pthread_mutex_init(&mutex, NULL);
- pthread_mutex_init(&queued_data_mutex, NULL);
+ pthread_mutex_init(&queued_clients_mutex, NULL);
epoll_fd = epoll_create(1024); // Size argument is ignored.
if (epoll_fd == -1) {
void Server::add_client_deferred(int sock)
{
- MutexLock lock(&queued_data_mutex);
+ MutexLock lock(&queued_clients_mutex);
queued_add_clients.push_back(sock);
}
exit(1);
}
- if (client_ptr->state == Client::SENDING_DATA &&
- client_ptr->stream_pos == client_ptr->stream->bytes_received) {
+ if (client_ptr->state == Client::WAITING_FOR_KEYFRAME ||
+ (client_ptr->state == Client::SENDING_DATA &&
+ client_ptr->stream_pos == client_ptr->stream->bytes_received)) {
client_ptr->stream->put_client_to_sleep(client_ptr);
} else {
process_client(client_ptr);
client_it != clients.end();
++client_it) {
Client *client = &client_it->second;
- if (client->state == Client::SENDING_DATA &&
- client->stream_pos == 0) {
+ if (client->state == Client::WAITING_FOR_KEYFRAME ||
+ (client->state == Client::SENDING_DATA &&
+ client->stream_pos == 0)) {
construct_header(client);
}
}
streams[stream_index]->mark_pool = mark_pool;
}
-void Server::add_data_deferred(int stream_index, const char *data, size_t bytes)
+void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate)
{
- MutexLock lock(&queued_data_mutex);
+ MutexLock lock(&mutex);
+ assert(clients.empty());
assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
- streams[stream_index]->add_data_deferred(data, bytes);
+ streams[stream_index]->pacing_rate = pacing_rate;
+}
+
+void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
+{
+ assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
+ streams[stream_index]->add_data_deferred(data, bytes, suitable_for_stream_start);
}
// See the .h file for postconditions after this function.
return;
}
- // Start sending from the end. In other words, we won't send any of the backlog,
- // but we'll start sending immediately as we get data.
+ // 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.
// This is postcondition #3.
- client->state = Client::SENDING_DATA;
if (client->stream_pos == size_t(-2)) {
client->stream_pos = std::min<size_t>(
client->stream->bytes_received - client->stream->backlog_size,
0);
+ client->state = Client::SENDING_DATA;
} else {
// client->stream_pos should be -1, but it might not be,
// if we have clients from an older version.
client->stream_pos = client->stream->bytes_received;
+ client->state = Client::WAITING_FOR_KEYFRAME;
}
client->stream->put_client_to_sleep(client);
return;
}
+ case Client::WAITING_FOR_KEYFRAME: {
+ Stream *stream = client->stream;
+ if (ssize_t(client->stream_pos) > stream->last_suitable_starting_point) {
+ // 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->last_suitable_starting_point;
+ client->state = Client::SENDING_DATA;
+ // Fall through.
+ }
case Client::SENDING_DATA: {
skip_lost_data(client);
Stream *stream = client->stream;
ssize_t ret;
do {
- loff_t offset = client->stream_pos % stream->backlog_size;
+ 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);
log_perror("setsockopt(SO_MARK)");
}
}
+ if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) {
+ if (client->stream->pacing_rate != ~0U) {
+ log_perror("setsockopt(SO_MAX_PACING_RATE)");
+ }
+ }
client->request.clear();
return 200; // OK!
"Content-encoding: metacube\r\n" +
"\r\n";
if (!stream->stream_header.empty()) {
- metacube_block_header hdr;
- memcpy(hdr.sync, METACUBE_SYNC, sizeof(hdr.sync));
+ metacube2_block_header hdr;
+ memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync));
hdr.size = htonl(stream->stream_header.size());
- hdr.flags = htonl(METACUBE_FLAGS_HEADER);
+ hdr.flags = htons(METACUBE_FLAGS_HEADER);
+ hdr.csum = htons(metacube2_compute_crc(&hdr));
client->header_or_error.append(
string(reinterpret_cast<char *>(&hdr), sizeof(hdr)));
}
void Server::process_queued_data()
{
- MutexLock lock(&queued_data_mutex);
+ {
+ MutexLock lock(&queued_clients_mutex);
- for (size_t i = 0; i < queued_add_clients.size(); ++i) {
- add_client(queued_add_clients[i]);
+ for (size_t i = 0; i < queued_add_clients.size(); ++i) {
+ add_client(queued_add_clients[i]);
+ }
+ queued_add_clients.clear();
}
- queued_add_clients.clear();
for (size_t i = 0; i < streams.size(); ++i) {
streams[i]->process_queued_data();