X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=88819129d45a4d1a85fdd86f156bc4617b61f716;hp=e91ba395ed6e2efa3c106fabf98ffa9756e159ef;hb=3fdf2e48bca3edcb0de00e0dbd0d0aae81ba9aa9;hpb=9b565a9e6e66f076abb7266b2c2f015f585fa9cb diff --git a/server.cpp b/server.cpp index e91ba39..8881912 100644 --- a/server.cpp +++ b/server.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #include @@ -8,12 +9,15 @@ #include #include #include +#include +#include #include #include #include #include #include +#include "markpool.h" #include "metacube.h" #include "server.h" #include "mutexlock.h" @@ -24,16 +28,37 @@ using namespace std; Client::Client(int sock) : sock(sock), + fwmark(0), + connect_time(time(NULL)), state(Client::READING_REQUEST), stream(NULL), header_or_error_bytes_sent(0), bytes_sent(0) { request.reserve(1024); + + // Find the remote address, and convert it to ASCII. + sockaddr_in6 addr; + socklen_t addr_len = sizeof(addr); + + if (getpeername(sock, reinterpret_cast(&addr), &addr_len) == -1) { + perror("getpeername"); + remote_addr = ""; + } else { + char buf[INET6_ADDRSTRLEN]; + if (inet_ntop(addr.sin6_family, &addr.sin6_addr, buf, sizeof(buf)) == NULL) { + perror("inet_ntop"); + remote_addr = ""; + } else { + remote_addr = buf; + } + } } Client::Client(const ClientProto &serialized, Stream *stream) : sock(serialized.sock()), + remote_addr(serialized.remote_addr()), + connect_time(serialized.connect_time()), state(State(serialized.state())), request(serialized.request()), stream_id(serialized.stream_id()), @@ -42,12 +67,24 @@ Client::Client(const ClientProto &serialized, Stream *stream) header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()), bytes_sent(serialized.bytes_sent()) { + if (stream->mark_pool != NULL) { + fwmark = stream->mark_pool->get_mark(); + } else { + fwmark = 0; // No mark. + } + if (setsockopt(sock, SOL_SOCKET, SO_MARK, &fwmark, sizeof(fwmark)) == -1) { + if (fwmark != 0) { + perror("setsockopt(SO_MARK)"); + } + } } ClientProto Client::serialize() const { ClientProto serialized; serialized.set_sock(sock); + serialized.set_remote_addr(remote_addr); + serialized.set_connect_time(connect_time); serialized.set_state(state); serialized.set_request(request); serialized.set_stream_id(stream_id); @@ -56,11 +93,22 @@ ClientProto Client::serialize() const serialized.set_bytes_sent(bytes_sent); return serialized; } + +ClientStats Client::get_stats() const +{ + ClientStats stats; + stats.stream_id = stream_id; + stats.remote_addr = remote_addr; + stats.connect_time = connect_time; + stats.bytes_sent = bytes_sent; + return stats; +} Stream::Stream(const string &stream_id) : stream_id(stream_id), data(new char[BACKLOG_SIZE]), - data_size(0) + data_size(0), + mark_pool(NULL) { memset(data, 0, BACKLOG_SIZE); } @@ -74,7 +122,8 @@ Stream::Stream(const StreamProto &serialized) : stream_id(serialized.stream_id()), header(serialized.header()), data(new char[BACKLOG_SIZE]), - data_size(serialized.data_size()) + data_size(serialized.data_size()), + mark_pool(NULL) { assert(serialized.data().size() == BACKLOG_SIZE); memcpy(data, serialized.data().data(), BACKLOG_SIZE); @@ -129,35 +178,17 @@ Server::~Server() } } -void Server::run() -{ - should_stop = false; - - // Joinable is already the default, but it's good to be certain. - pthread_attr_t attr; - pthread_attr_init(&attr); - pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE); - pthread_create(&worker_thread, &attr, Server::do_work_thunk, this); -} - -void Server::stop() +vector Server::get_client_stats() const { - { - MutexLock lock(&mutex); - should_stop = true; - } + vector ret; - if (pthread_join(worker_thread, NULL) == -1) { - perror("pthread_join"); - exit(1); + MutexLock lock(&mutex); + for (map::const_iterator client_it = clients.begin(); + client_it != clients.end(); + ++client_it) { + ret.push_back(client_it->second.get_stats()); } -} - -void *Server::do_work_thunk(void *arg) -{ - Server *server = static_cast(arg); - server->do_work(); - return NULL; + return ret; } void Server::do_work() @@ -165,6 +196,9 @@ void Server::do_work() for ( ;; ) { int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS); if (nfds == -1 && errno == EINTR) { + if (should_stop) { + return; + } continue; } if (nfds == -1) { @@ -174,10 +208,6 @@ void Server::do_work() MutexLock lock(&mutex); // We release the mutex between iterations. - if (should_stop) { - return; - } - process_queued_data(); for (int i = 0; i < nfds; ++i) { @@ -202,6 +232,10 @@ void Server::do_work() process_client(to_process[i]); } } + + if (should_stop) { + return; + } } } @@ -307,6 +341,13 @@ void Server::set_header(const string &stream_id, const string &header) } } } + +void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool) +{ + MutexLock lock(&mutex); + assert(clients.empty()); + find_stream(stream_id)->mark_pool = mark_pool; +} void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes) { @@ -531,6 +572,16 @@ int Server::parse_request(Client *client) client->stream_id = request_tokens[1]; client->stream = find_stream(client->stream_id); + if (client->stream->mark_pool != NULL) { + client->fwmark = client->stream->mark_pool->get_mark(); + } else { + client->fwmark = 0; // No mark. + } + if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) { + if (client->fwmark != 0) { + perror("setsockopt(SO_MARK)"); + } + } client->request.clear(); return 200; // OK! @@ -593,8 +644,12 @@ void Server::close_client(Client *client) if (client->stream != NULL) { delete_from(&client->stream->sleeping_clients, client); delete_from(&client->stream->to_process, client); + if (client->stream->mark_pool != NULL) { + int fwmark = client->fwmark; + client->stream->mark_pool->release_mark(fwmark); + } } - + // Bye-bye! int ret; do {