X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;ds=sidebyside;f=server.cpp;h=a000d9c6d08f0f3d799c07f3bfec1902f68fc141;hb=5cd46e39b4063d94f6dc559ae350beeb8406a8f9;hp=3d915c22c99ad2c94e5aac6878e92f76b233afb2;hpb=6bf4dfc2c08d51de1d08e80078dd06f20ca0149c;p=cubemap

diff --git a/server.cpp b/server.cpp
index 3d915c2..a000d9c 100644
--- a/server.cpp
+++ b/server.cpp
@@ -1,8 +1,11 @@
 #include <assert.h>
 #include <errno.h>
+#include <netinet/in.h>
 #include <pthread.h>
+#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <string.h>
 #include <sys/epoll.h>
 #include <sys/sendfile.h>
 #include <sys/socket.h>
@@ -14,8 +17,10 @@
 #include <utility>
 #include <vector>
 
+#include "accesslog.h"
 #include "log.h"
 #include "markpool.h"
+#include "metacube.h"
 #include "mutexlock.h"
 #include "parse.h"
 #include "server.h"
@@ -24,6 +29,8 @@
 
 using namespace std;
 
+extern AccessLogThread *access_log;
+
 Server::Server()
 {
 	pthread_mutex_init(&mutex, NULL);
@@ -38,6 +45,12 @@ Server::Server()
 
 Server::~Server()
 {
+	for (map<string, Stream *>::iterator stream_it = streams.begin();
+	     stream_it != streams.end();
+	     ++stream_it) {
+		delete stream_it->second;
+	}
+
 	int ret;
 	do {
 		ret = close(epoll_fd);
@@ -185,29 +198,37 @@ void Server::add_client_from_serialized(const ClientProto &client)
 	}
 }
 
-void Server::add_stream(const string &stream_id, size_t backlog_size)
+void Server::add_stream(const string &stream_id, size_t backlog_size, Stream::Encoding encoding)
 {
 	MutexLock lock(&mutex);
-	streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size)));
+	streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size, encoding)));
 }
 
-void Server::add_stream_from_serialized(const StreamProto &stream)
+void Server::add_stream_from_serialized(const StreamProto &stream, int data_fd)
 {
 	MutexLock lock(&mutex);
-	streams.insert(make_pair(stream.stream_id(), new Stream(stream)));
+	streams.insert(make_pair(stream.stream_id(), new Stream(stream, data_fd)));
 }
 	
-void Server::set_backlog_size(const std::string &stream_id, size_t new_size)
+void Server::set_backlog_size(const string &stream_id, size_t new_size)
 {
 	MutexLock lock(&mutex);
 	assert(streams.count(stream_id) != 0);
 	streams[stream_id]->set_backlog_size(new_size);
 }
 	
-void Server::set_header(const string &stream_id, const string &header)
+void Server::set_encoding(const string &stream_id, Stream::Encoding encoding)
+{
+	MutexLock lock(&mutex);
+	assert(streams.count(stream_id) != 0);
+	streams[stream_id]->encoding = encoding;
+}
+	
+void Server::set_header(const string &stream_id, const string &http_header, const string &stream_header)
 {
 	MutexLock lock(&mutex);
-	find_stream(stream_id)->header = header;
+	find_stream(stream_id)->http_header = http_header;
+	find_stream(stream_id)->stream_header = stream_header;
 
 	// If there are clients we haven't sent anything to yet, we should give
 	// them the header, so push back into the SENDING_HEADER state.
@@ -222,7 +243,7 @@ void Server::set_header(const string &stream_id, const string &header)
 	}
 }
 	
-void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool)
+void Server::set_mark_pool(const string &stream_id, MarkPool *mark_pool)
 {
 	MutexLock lock(&mutex);
 	assert(clients.empty());
@@ -357,13 +378,17 @@ sending_data_again:
 			return;
 		}
 		if (bytes_to_send > stream->backlog_size) {
-			log(WARNING, "[%s] Client lost %lld bytes, maybe too slow connection",
-			        client->remote_addr.c_str(),
-			        (long long int)(bytes_to_send - stream->backlog_size));
+			size_t bytes_lost = bytes_to_send - stream->backlog_size;
 			client->stream_pos = stream->bytes_received - stream->backlog_size;
-			client->bytes_lost += bytes_to_send - stream->backlog_size;
+			client->bytes_lost += bytes_lost;
 			++client->num_loss_events;
 			bytes_to_send = stream->backlog_size;
+
+			double loss_fraction = double(client->bytes_lost) / double(client->bytes_lost + client->bytes_sent);
+			log(WARNING, "[%s] Client lost %lld bytes (total loss: %.2f%%), maybe too slow connection",
+			        client->remote_addr.c_str(),
+			        (long long int)(bytes_lost),
+				100.0 * loss_fraction);
 		}
 
 		// See if we need to split across the circular buffer.
@@ -445,7 +470,27 @@ int Server::parse_request(Client *client)
 
 void Server::construct_header(Client *client)
 {
-	client->header_or_error = find_stream(client->stream_id)->header;
+	Stream *stream = find_stream(client->stream_id);
+	if (stream->encoding == Stream::STREAM_ENCODING_RAW) {
+		client->header_or_error = stream->http_header +
+			"\r\n" +
+			stream->stream_header;
+	} else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) {
+		client->header_or_error = stream->http_header +
+			"Content-encoding: metacube\r\n" +
+			"\r\n";
+		if (!stream->stream_header.empty()) {
+			metacube_block_header hdr;
+			memcpy(hdr.sync, METACUBE_SYNC, sizeof(hdr.sync));
+			hdr.size = htonl(stream->stream_header.size());
+			hdr.flags = htonl(METACUBE_FLAGS_HEADER);
+			client->header_or_error.append(
+				string(reinterpret_cast<char *>(&hdr), sizeof(hdr)));
+		}
+		client->header_or_error.append(stream->stream_header);
+	} else {
+		assert(false);
+	}
 
 	// Switch states.
 	client->state = Client::SENDING_HEADER;
@@ -504,6 +549,9 @@ void Server::close_client(Client *client)
 		}
 	}
 
+	// Log to access_log.
+	access_log->write(client->get_stats());
+
 	// Bye-bye!
 	int ret;
 	do {