dlib C++ Library - server_kernel.cpp
// Copyright (C) 2003 Davis E. King (davis@dlib.net)
// License: Boost Software License See LICENSE.txt for the full license.
#ifndef DLIB_SERVER_KERNEL_CPp_
#define DLIB_SERVER_KERNEL_CPp_
#include "server_kernel.h"
#include "../string.h"
namespace dlib
{
// ----------------------------------------------------------------------------------------
server::
server (
) :
listening_port(0),
running(false),
shutting_down(false),
running_signaler(running_mutex),
thread_count(0),
thread_count_signaler(thread_count_mutex),
max_connections(1000),
thread_count_zero(thread_count_mutex),
graceful_close_timeout(500)
{
}
// ----------------------------------------------------------------------------------------
server::
~server (
)
{
clear();
}
// ----------------------------------------------------------------------------------------
unsigned long server::
get_graceful_close_timeout (
) const
{
auto_mutex lock(max_connections_mutex);
return graceful_close_timeout;
}
// ----------------------------------------------------------------------------------------
void server::
set_graceful_close_timeout (
unsigned long timeout
)
{
auto_mutex lock(max_connections_mutex);
graceful_close_timeout = timeout;
}
// ----------------------------------------------------------------------------------------
int server::
get_max_connections (
) const
{
max_connections_mutex.lock();
int temp = max_connections;
max_connections_mutex.unlock();
return temp;
}
// ----------------------------------------------------------------------------------------
void server::
set_max_connections (
int max
)
{
// make sure requires clause is not broken
DLIB_CASSERT(
max >= 0 ,
"\tvoid server::set_max_connections"
<< "\n\tmax == " << max
<< "\n\tthis: " << this
);
max_connections_mutex.lock();
max_connections = max;
max_connections_mutex.unlock();
}
// ----------------------------------------------------------------------------------------
void server::
clear (
)
{
// signal that we are shutting down
shutting_down_mutex.lock();
shutting_down = true;
shutting_down_mutex.unlock();
max_connections_mutex.lock();
listening_port_mutex.lock();
listening_ip_mutex.lock();
listening_ip = "";
listening_port = 0;
max_connections = 1000;
graceful_close_timeout = 500;
listening_port_mutex.unlock();
listening_ip_mutex.unlock();
max_connections_mutex.unlock();
// tell all the connections to shut down
cons_mutex.lock();
connection* temp;
while (cons.size() > 0)
{
cons.remove_any(temp);
temp->shutdown();
}
cons_mutex.unlock();
// wait for all the connections to shut down
thread_count_mutex.lock();
while (thread_count > 0)
{
thread_count_zero.wait();
}
thread_count_mutex.unlock();
// wait for the listener to close
running_mutex.lock();
while (running == true)
{
running_signaler.wait();
}
running_mutex.unlock();
// signal that the shutdown is complete
shutting_down_mutex.lock();
shutting_down = false;
shutting_down_mutex.unlock();
}
// ----------------------------------------------------------------------------------------
void server::
start_async_helper (
)
{
try
{
start_accepting_connections();
}
catch (std::exception& e)
{
sdlog << LERROR << e.what();
}
}
// ----------------------------------------------------------------------------------------
void server::
start_async (
)
{
auto_mutex lock(running_mutex);
if (running)
return;
// Any exceptions likely to be thrown by the server are going to be
// thrown when trying to bind the port. So calling this here rather
// than in the thread we are about to make will cause start_async()
// to report errors back to the user in a very straight forward way.
open_listening_socket();
async_start_thread.reset(new thread_function(make_mfp(*this,&server::start_async_helper)));
}
// ----------------------------------------------------------------------------------------
void server::
open_listening_socket (
)
{
if (!sock)
{
int status = create_listener(sock,listening_port,listening_ip);
const int port_used = listening_port;
// if there was an error then clear this object
if (status < 0)
{
max_connections_mutex.lock();
listening_port_mutex.lock();
listening_ip_mutex.lock();
listening_ip = "";
listening_port = 0;
max_connections = 1000;
graceful_close_timeout = 500;
listening_port_mutex.unlock();
listening_ip_mutex.unlock();
max_connections_mutex.unlock();
}
// throw an exception for the error
if (status == PORTINUSE)
{
throw dlib::socket_error(
EPORT_IN_USE,
"error occurred in server::start()\nport " + cast_to_string(port_used) + " already in use"
);
}
else if (status == OTHER_ERROR)
{
throw dlib::socket_error(
"error occurred in server::start()\nunable to create listener"
);
}
}
running_mutex.lock();
running = true;
running_mutex.unlock();
}
// ----------------------------------------------------------------------------------------
void server::
start (
)
{
// make sure requires clause is not broken
DLIB_CASSERT(
this->is_running() == false,
"\tvoid server::start"
<< "\n\tis_running() == " << this->is_running()
<< "\n\tthis: " << this
);
start_accepting_connections();
}
// ----------------------------------------------------------------------------------------
void server::
start_accepting_connections (
)
{
open_listening_socket();
// determine the listening port
bool port_assigned = false;
listening_port_mutex.lock();
if (listening_port == 0)
{
port_assigned = true;
listening_port = sock->get_listening_port();
}
listening_port_mutex.unlock();
if (port_assigned)
on_listening_port_assigned();
int status = 0;
connection* client;
bool exit = false;
while ( true )
{
// accept the next connection
status = sock->accept(client,1000);
// if there was an error then quit the loop
if (status == OTHER_ERROR)
{
break;
}
shutting_down_mutex.lock();
// if we are shutting down then signal that we should quit the loop
exit = shutting_down;
shutting_down_mutex.unlock();
// if we should be shutting down
if (exit)
{
// if a connection was opened then close it
if (status == 0)
delete client;
break;
}
// if the accept timed out
if (status == TIMEOUT)
{
continue;
}
// add this new connection to cons
cons_mutex.lock();
connection* client_temp = client;
try{cons.add(client_temp);}
catch(...)
{
sock.reset();
delete client;
cons_mutex.unlock();
// signal that we are not running start() anymore
running_mutex.lock();
running = false;
running_signaler.broadcast();
running_mutex.unlock();
clear();
throw;
}
cons_mutex.unlock();
// make a param structure
param* temp = 0;
try{
temp = new param (
*this,
*client,
get_graceful_close_timeout()
);
} catch (...)
{
sock.reset();
delete client;
running_mutex.lock();
running = false;
running_signaler.broadcast();
running_mutex.unlock();
clear();
throw;
}
// if create_new_thread failed
if (!create_new_thread(service_connection,temp))
{
delete temp;
// close the listening socket
sock.reset();
// close the new connection and remove it from cons
cons_mutex.lock();
connection* ctemp;
if (cons.is_member(client))
{
cons.remove(client,ctemp);
}
delete client;
cons_mutex.unlock();
// signal that the listener has closed
running_mutex.lock();
running = false;
running_signaler.broadcast();
running_mutex.unlock();
// make sure the object is cleared
clear();
// throw the exception
throw dlib::thread_error(
ECREATE_THREAD,
"error occurred in server::start()\nunable to start thread"
);
}
// if we made the new thread then update thread_count
else
{
// increment the thread count
thread_count_mutex.lock();
++thread_count;
if (thread_count == 0)
thread_count_zero.broadcast();
thread_count_mutex.unlock();
}
// check if we have hit the maximum allowed number of connections
max_connections_mutex.lock();
// if max_connections is zero or the loop is ending then skip this
if (max_connections != 0)
{
// wait for thread_count to be less than max_connections
thread_count_mutex.lock();
while (thread_count >= max_connections)
{
max_connections_mutex.unlock();
thread_count_signaler.wait();
max_connections_mutex.lock();
// if we are shutting down the quit the loop
shutting_down_mutex.lock();
exit = shutting_down;
shutting_down_mutex.unlock();
if (exit)
break;
}
thread_count_mutex.unlock();
}
max_connections_mutex.unlock();
if (exit)
{
break;
}
} //while ( true )
// close the socket
sock.reset();
// signal that the listener has closed
running_mutex.lock();
running = false;
running_signaler.broadcast();
running_mutex.unlock();
// if there was an error with accept then throw an exception
if (status == OTHER_ERROR)
{
// make sure the object is cleared
clear();
// throw the exception
throw dlib::socket_error(
"error occurred in server::start()\nlistening socket returned error"
);
}
}
// ----------------------------------------------------------------------------------------
bool server::
is_running (
) const
{
running_mutex.lock();
bool temp = running;
running_mutex.unlock();
return temp;
}
// ----------------------------------------------------------------------------------------
const std::string server::
get_listening_ip (
) const
{
listening_ip_mutex.lock();
std::string ip(listening_ip);
listening_ip_mutex.unlock();
return ip;
}
// ----------------------------------------------------------------------------------------
int server::
get_listening_port (
) const
{
listening_port_mutex.lock();
int port = listening_port;
listening_port_mutex.unlock();
return port;
}
// ----------------------------------------------------------------------------------------
void server::
set_listening_port (
int port
)
{
// make sure requires clause is not broken
DLIB_CASSERT(
( port >= 0 &&
this->is_running() == false ),
"\tvoid server::set_listening_port"
<< "\n\tport == " << port
<< "\n\tis_running() == " << this->is_running()
<< "\n\tthis: " << this
);
listening_port_mutex.lock();
listening_port = port;
listening_port_mutex.unlock();
}
// ----------------------------------------------------------------------------------------
void server::
set_listening_ip (
const std::string& ip
)
{
// make sure requires clause is not broken
DLIB_CASSERT(
( ( is_ip_address(ip) || ip == "" ) &&
this->is_running() == false ),
"\tvoid server::set_listening_ip"
<< "\n\tip == " << ip
<< "\n\tis_running() == " << this->is_running()
<< "\n\tthis: " << this
);
listening_ip_mutex.lock();
listening_ip = ip;
listening_ip_mutex.unlock();
}
// ----------------------------------------------------------------------------------------
// ----------------------------------------------------------------------------------------
// static member function definitions
// ----------------------------------------------------------------------------------------
// ----------------------------------------------------------------------------------------
const logger server::sdlog("dlib.server");
void server::
service_connection(
void* item
)
{
param& p = *static_cast<param*>(item);
p.the_server.on_connect(p.new_connection);
// remove this connection from cons and close it
p.the_server.cons_mutex.lock();
connection* temp;
if (p.the_server.cons.is_member(&p.new_connection))
p.the_server.cons.remove(&p.new_connection,temp);
p.the_server.cons_mutex.unlock();
try{ close_gracefully(&p.new_connection, p.graceful_close_timeout); }
catch (...) { sdlog << LERROR << "close_gracefully() threw"; }
// decrement the thread count and signal if it is now zero
p.the_server.thread_count_mutex.lock();
--p.the_server.thread_count;
p.the_server.thread_count_signaler.broadcast();
if (p.the_server.thread_count == 0)
p.the_server.thread_count_zero.broadcast();
p.the_server.thread_count_mutex.unlock();
delete &p;
}
// ----------------------------------------------------------------------------------------
}
#endif // DLIB_SERVER_KERNEL_CPp_