#include "server.h"
#include "state.pb.h"
#include "stream.h"
+#include "util.h"
using namespace std;
delete stream_it->second;
}
- int ret;
- do {
- ret = close(epoll_fd);
- } while (ret == -1 && errno == EINTR);
-
- if (ret == -1) {
- log_perror("close(epoll_fd)");
- }
+ safe_close(epoll_fd);
}
vector<ClientStats> Server::get_client_stats() const
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) {
+ while (!should_stop()) {
+ // Wait until there's activity on at least one of the fds,
+ // or 20 ms (about one frame at 50 fps) has elapsed.
+ //
+ // We could in theory wait forever and rely on wakeup()
+ // from add_client_deferred() and add_data_deferred(),
+ // but wakeup is a pretty expensive operation, and the
+ // two threads might end up fighting over a lock, so it's
+ // seemingly (much) more efficient to just have a timeout here.
+ int nfds = epoll_pwait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS, &sigset_without_usr1_block);
+ if (nfds == -1 && errno != EINTR) {
log_perror("epoll_wait");
exit(1);
}
process_client(to_process[i]);
}
}
-
- if (should_stop) {
- return;
- }
}
}
void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes)
{
MutexLock lock(&queued_data_mutex);
- queued_data[stream_id].append(string(data, data + bytes));
+ find_stream(stream_id)->add_data_deferred(data, bytes);
}
// See the .h file for postconditions after this function.
access_log->write(client->get_stats());
// Bye-bye!
- int ret;
- do {
- ret = close(client->sock);
- } while (ret == -1 && errno == EINTR);
-
- if (ret == -1) {
- log_perror("close");
- }
+ safe_close(client->sock);
clients.erase(client->sock);
}
}
queued_add_clients.clear();
- for (map<string, string>::iterator queued_it = queued_data.begin();
- queued_it != queued_data.end();
- ++queued_it) {
- Stream *stream = find_stream(queued_it->first);
- stream->add_data(queued_it->second.data(), queued_it->second.size());
- stream->wake_up_all_clients();
+ for (map<string, Stream *>::iterator stream_it = streams.begin();
+ stream_it != streams.end();
+ ++stream_it) {
+ stream_it->second->process_queued_data();
}
- queued_data.clear();
}