Graceful shutdown on quit
This commit is contained in:
@@ -4,7 +4,7 @@
|
|||||||
|
|
||||||
[DONE] Figure out packet double-sending issue [if delay > 1/50 sec?]
|
[DONE] Figure out packet double-sending issue [if delay > 1/50 sec?]
|
||||||
|
|
||||||
Graceful exit / server shutdown
|
[DONE] Graceful exit / server shutdown
|
||||||
|
|
||||||
Make roaming much smoother (with error message placement)
|
Make roaming much smoother (with error message placement)
|
||||||
|
|
||||||
|
|||||||
@@ -12,7 +12,8 @@ Transport<MyState, RemoteState>::Transport( MyState &initial_state, RemoteState
|
|||||||
sender( &connection, initial_state ),
|
sender( &connection, initial_state ),
|
||||||
received_states( 1, TimestampedState<RemoteState>( timestamp(), 0, initial_remote ) ),
|
received_states( 1, TimestampedState<RemoteState>( timestamp(), 0, initial_remote ) ),
|
||||||
last_receiver_state( initial_remote ),
|
last_receiver_state( initial_remote ),
|
||||||
fragments()
|
fragments(),
|
||||||
|
verbose( false )
|
||||||
{
|
{
|
||||||
/* server */
|
/* server */
|
||||||
}
|
}
|
||||||
@@ -24,7 +25,8 @@ Transport<MyState, RemoteState>::Transport( MyState &initial_state, RemoteState
|
|||||||
sender( &connection, initial_state ),
|
sender( &connection, initial_state ),
|
||||||
received_states( 1, TimestampedState<RemoteState>( timestamp(), 0, initial_remote ) ),
|
received_states( 1, TimestampedState<RemoteState>( timestamp(), 0, initial_remote ) ),
|
||||||
last_receiver_state( initial_remote ),
|
last_receiver_state( initial_remote ),
|
||||||
fragments()
|
fragments(),
|
||||||
|
verbose( false )
|
||||||
{
|
{
|
||||||
/* client */
|
/* client */
|
||||||
}
|
}
|
||||||
@@ -82,11 +84,9 @@ void Transport<MyState, RemoteState>::recv( void )
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
/*
|
|
||||||
if ( verbose )
|
if ( verbose )
|
||||||
fprintf( stderr, "[%d] Received state %d [ack %d]\n",
|
fprintf( stderr, "[%d] Received state %d [ack %d]\n",
|
||||||
(int)timestamp() % 100000, (int)new_state.num, (int)inst.ack_num );
|
(int)timestamp() % 100000, (int)new_state.num, (int)inst.ack_num() );
|
||||||
*/
|
|
||||||
received_states.push_back( new_state );
|
received_states.push_back( new_state );
|
||||||
sender.set_ack_num( received_states.back().num );
|
sender.set_ack_num( received_states.back().num );
|
||||||
}
|
}
|
||||||
@@ -128,4 +128,3 @@ string Transport<MyState, RemoteState>::get_remote_diff( void )
|
|||||||
|
|
||||||
return ret;
|
return ret;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+14
-4
@@ -30,8 +30,8 @@ namespace Network {
|
|||||||
/* simple receiver */
|
/* simple receiver */
|
||||||
list< TimestampedState<RemoteState> > received_states;
|
list< TimestampedState<RemoteState> > received_states;
|
||||||
RemoteState last_receiver_state; /* the state we were in when user last queried state */
|
RemoteState last_receiver_state; /* the state we were in when user last queried state */
|
||||||
|
|
||||||
FragmentAssembly fragments;
|
FragmentAssembly fragments;
|
||||||
|
bool verbose;
|
||||||
|
|
||||||
public:
|
public:
|
||||||
Transport( MyState &initial_state, RemoteState &initial_remote );
|
Transport( MyState &initial_state, RemoteState &initial_remote );
|
||||||
@@ -47,14 +47,24 @@ namespace Network {
|
|||||||
/* Blocks waiting for a packet. */
|
/* Blocks waiting for a packet. */
|
||||||
void recv( void );
|
void recv( void );
|
||||||
|
|
||||||
|
/* Find diff between last receiver state and current remote state, then rationalize states. */
|
||||||
|
string get_remote_diff( void );
|
||||||
|
|
||||||
|
/* Shut down other side of connection. */
|
||||||
|
/* Illegal to change current_state after this. */
|
||||||
|
void start_shutdown( void ) { sender.start_shutdown(); }
|
||||||
|
bool shutdown_in_progress( void ) { return sender.get_shutdown_in_progress(); }
|
||||||
|
bool shutdown_acknowledged( void ) { return sender.get_shutdown_acknowledged(); }
|
||||||
|
|
||||||
|
/* Other side has requested shutdown and we have sent one ACK */
|
||||||
|
bool counterparty_shutdown_ack_sent( void ) { return sender.get_counterparty_shutdown_acknowledged(); }
|
||||||
|
|
||||||
int port( void ) { return connection.port(); }
|
int port( void ) { return connection.port(); }
|
||||||
string get_key( void ) { return connection.get_key(); }
|
string get_key( void ) { return connection.get_key(); }
|
||||||
|
|
||||||
MyState &get_current_state( void ) { return sender.get_current_state(); }
|
MyState &get_current_state( void ) { return sender.get_current_state(); }
|
||||||
void set_current_state( const MyState &x ) { sender.set_current_state( x ); }
|
void set_current_state( const MyState &x ) { sender.set_current_state( x ); }
|
||||||
|
|
||||||
string get_remote_diff( void );
|
|
||||||
|
|
||||||
typename list< TimestampedState<RemoteState > >::iterator begin( void ) { return received_states.begin(); }
|
typename list< TimestampedState<RemoteState > >::iterator begin( void ) { return received_states.begin(); }
|
||||||
typename list< TimestampedState<RemoteState > >::iterator end( void ) { return received_states.end(); }
|
typename list< TimestampedState<RemoteState > >::iterator end( void ) { return received_states.end(); }
|
||||||
|
|
||||||
@@ -62,7 +72,7 @@ namespace Network {
|
|||||||
|
|
||||||
int fd( void ) { return connection.fd(); }
|
int fd( void ) { return connection.fd(); }
|
||||||
|
|
||||||
void set_verbose( void ) { sender.set_verbose(); }
|
void set_verbose( void ) { sender.set_verbose(); verbose = true; }
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+25
-4
@@ -168,7 +168,9 @@ void serve( int host_fd )
|
|||||||
}
|
}
|
||||||
|
|
||||||
/* update client with new state of terminal */
|
/* update client with new state of terminal */
|
||||||
network.set_current_state( terminal );
|
if ( !network.shutdown_in_progress() ) {
|
||||||
|
network.set_current_state( terminal );
|
||||||
|
}
|
||||||
|
|
||||||
/* write any writeback octets back to the host */
|
/* write any writeback octets back to the host */
|
||||||
if ( swrite( host_fd, terminal_to_host.c_str(), terminal_to_host.length() ) < 0 ) {
|
if ( swrite( host_fd, terminal_to_host.c_str(), terminal_to_host.length() ) < 0 ) {
|
||||||
@@ -194,16 +196,35 @@ void serve( int host_fd )
|
|||||||
string terminal_to_host = terminal.act( string( buf, bytes_read ) );
|
string terminal_to_host = terminal.act( string( buf, bytes_read ) );
|
||||||
|
|
||||||
/* update client with new state of terminal */
|
/* update client with new state of terminal */
|
||||||
network.set_current_state( terminal );
|
if ( !network.shutdown_in_progress() ) {
|
||||||
|
network.set_current_state( terminal );
|
||||||
|
}
|
||||||
|
|
||||||
/* write any writeback octets back to the host */
|
/* write any writeback octets back to the host */
|
||||||
if ( swrite( host_fd, terminal_to_host.c_str(), terminal_to_host.length() ) < 0 ) {
|
if ( swrite( host_fd, terminal_to_host.c_str(), terminal_to_host.length() ) < 0 ) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if ( (pollfds[ 0 ].revents | pollfds[ 1 ].revents)
|
if ( (pollfds[ 0 ].revents)
|
||||||
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
||||||
|
/* network problem */
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
if ( (pollfds[ 1 ].revents)
|
||||||
|
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
||||||
|
/* host problem */
|
||||||
|
network.start_shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
/* quit if our shutdown has been acknowledged */
|
||||||
|
if ( network.shutdown_in_progress() && network.shutdown_acknowledged() ) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* quit if we received and acknowledged a shutdown request */
|
||||||
|
if ( network.counterparty_shutdown_ack_sent() ) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -174,8 +174,10 @@ void client( const char *ip, int port, const char *key )
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
for ( int i = 0; i < bytes_read; i++ ) {
|
if ( !network.shutdown_in_progress() ) {
|
||||||
network.get_current_state().push_back( Parser::UserByte( buf[ i ] ) );
|
for ( int i = 0; i < bytes_read; i++ ) {
|
||||||
|
network.get_current_state().push_back( Parser::UserByte( buf[ i ] ) );
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -193,8 +195,10 @@ void client( const char *ip, int port, const char *key )
|
|||||||
|
|
||||||
/* tell remote emulator */
|
/* tell remote emulator */
|
||||||
Parser::Resize res( window_size.ws_col, window_size.ws_row );
|
Parser::Resize res( window_size.ws_col, window_size.ws_row );
|
||||||
|
|
||||||
network.get_current_state().push_back( res );
|
if ( !network.shutdown_in_progress() ) {
|
||||||
|
network.get_current_state().push_back( res );
|
||||||
|
}
|
||||||
|
|
||||||
/* tell local emulator -- there is probably a safer way to do this */
|
/* tell local emulator -- there is probably a safer way to do this */
|
||||||
for ( list< Network::TimestampedState<Terminal::Complete> >::iterator i = network.begin();
|
for ( list< Network::TimestampedState<Terminal::Complete> >::iterator i = network.begin();
|
||||||
@@ -204,8 +208,25 @@ void client( const char *ip, int port, const char *key )
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if ( (pollfds[ 0 ].revents | pollfds[ 1 ].revents)
|
if ( (pollfds[ 0 ].revents)
|
||||||
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
||||||
|
/* network problem */
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
if ( (pollfds[ 1 ].revents)
|
||||||
|
& (POLLERR | POLLHUP | POLLNVAL) ) {
|
||||||
|
/* user problem */
|
||||||
|
network.start_shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
/* quit if our shutdown has been acknowledged */
|
||||||
|
if ( network.shutdown_in_progress() && network.shutdown_acknowledged() ) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* quit if we received and acknowledged a shutdown request */
|
||||||
|
if ( network.counterparty_shutdown_ack_sent() ) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+20
-4
@@ -14,6 +14,7 @@ TransportSender<MyState>::TransportSender( Connection *s_connection, MyState &in
|
|||||||
next_ack_time( timestamp() ),
|
next_ack_time( timestamp() ),
|
||||||
next_send_time( timestamp() ),
|
next_send_time( timestamp() ),
|
||||||
verbose( false ),
|
verbose( false ),
|
||||||
|
shutdown_in_progress( false ),
|
||||||
ack_num( 0 )
|
ack_num( 0 )
|
||||||
{
|
{
|
||||||
}
|
}
|
||||||
@@ -50,10 +51,15 @@ int TransportSender<MyState>::wait_time( void )
|
|||||||
if ( next_send_time < sent_states.back().timestamp + send_interval() ) {
|
if ( next_send_time < sent_states.back().timestamp + send_interval() ) {
|
||||||
next_send_time = sent_states.back().timestamp + send_interval();
|
next_send_time = sent_states.back().timestamp + send_interval();
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if ( next_send_time < next_wakeup ) {
|
/* speed up shutdown sequence */
|
||||||
next_wakeup = next_send_time;
|
if ( shutdown_in_progress || (ack_num == uint64_t(-1)) ) {
|
||||||
}
|
next_ack_time = sent_states.back().timestamp + send_interval();
|
||||||
|
}
|
||||||
|
|
||||||
|
if ( next_send_time < next_wakeup ) {
|
||||||
|
next_wakeup = next_send_time;
|
||||||
}
|
}
|
||||||
|
|
||||||
if ( !connection->get_attached() ) {
|
if ( !connection->get_attached() ) {
|
||||||
@@ -110,6 +116,11 @@ void TransportSender<MyState>::send_empty_ack( void )
|
|||||||
|
|
||||||
uint64_t new_num = sent_states.back().num + 1;
|
uint64_t new_num = sent_states.back().num + 1;
|
||||||
|
|
||||||
|
/* special case for shutdown sequence */
|
||||||
|
if ( shutdown_in_progress ) {
|
||||||
|
new_num = uint64_t( -1 );
|
||||||
|
}
|
||||||
|
|
||||||
send_in_fragments( "", new_num, false );
|
send_in_fragments( "", new_num, false );
|
||||||
sent_states.push_back( TimestampedState<MyState>( sent_states.back().timestamp, new_num, current_state ) );
|
sent_states.push_back( TimestampedState<MyState>( sent_states.back().timestamp, new_num, current_state ) );
|
||||||
next_ack_time = timestamp() + ACK_INTERVAL;
|
next_ack_time = timestamp() + ACK_INTERVAL;
|
||||||
@@ -125,6 +136,11 @@ void TransportSender<MyState>::send_to_receiver( string diff )
|
|||||||
new_num = sent_states.back().num + 1;
|
new_num = sent_states.back().num + 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* special case for shutdown sequence */
|
||||||
|
if ( shutdown_in_progress ) {
|
||||||
|
new_num = uint64_t( -1 );
|
||||||
|
}
|
||||||
|
|
||||||
bool done = false;
|
bool done = false;
|
||||||
int MTU_tries = 0;
|
int MTU_tries = 0;
|
||||||
while ( !done ) {
|
while ( !done ) {
|
||||||
@@ -157,7 +173,7 @@ void TransportSender<MyState>::send_to_receiver( string diff )
|
|||||||
assumed_receiver_state = sent_states.end();
|
assumed_receiver_state = sent_states.end();
|
||||||
assumed_receiver_state--;
|
assumed_receiver_state--;
|
||||||
next_ack_time = timestamp() + ACK_INTERVAL;
|
next_ack_time = timestamp() + ACK_INTERVAL;
|
||||||
next_send_time = -1;
|
next_send_time = uint64_t(-1);
|
||||||
}
|
}
|
||||||
|
|
||||||
template <class MyState>
|
template <class MyState>
|
||||||
|
|||||||
+15
-5
@@ -53,6 +53,7 @@ namespace Network {
|
|||||||
uint64_t next_send_time;
|
uint64_t next_send_time;
|
||||||
|
|
||||||
bool verbose;
|
bool verbose;
|
||||||
|
bool shutdown_in_progress;
|
||||||
|
|
||||||
/* information about receiver state */
|
/* information about receiver state */
|
||||||
uint64_t ack_num;
|
uint64_t ack_num;
|
||||||
@@ -67,15 +68,24 @@ namespace Network {
|
|||||||
/* Returns the number of ms to wait until next possible event. */
|
/* Returns the number of ms to wait until next possible event. */
|
||||||
int wait_time( void );
|
int wait_time( void );
|
||||||
|
|
||||||
/* executed upon receipt of ack */
|
/* Executed upon receipt of ack */
|
||||||
void process_acknowledgment_through( uint64_t ack_num );
|
void process_acknowledgment_through( uint64_t ack_num );
|
||||||
|
|
||||||
/* getters and setters */
|
/* Executed upon entry to new receiver state */
|
||||||
MyState &get_current_state( void ) { return current_state; }
|
|
||||||
void set_current_state( const MyState &x ) { current_state = x; }
|
|
||||||
void set_verbose( void ) { verbose = true; }
|
|
||||||
void set_ack_num( uint64_t s_ack_num ) { ack_num = s_ack_num; }
|
void set_ack_num( uint64_t s_ack_num ) { ack_num = s_ack_num; }
|
||||||
|
|
||||||
|
/* Starts shutdown sequence */
|
||||||
|
void start_shutdown( void ) { shutdown_in_progress = true; }
|
||||||
|
bool get_shutdown_in_progress( void ) { return shutdown_in_progress; }
|
||||||
|
bool get_shutdown_acknowledged( void ) { return sent_states.front().num == uint64_t(-1); }
|
||||||
|
bool get_counterparty_shutdown_acknowledged( void ) { return last_instruction_sent.ack_num() == uint64_t(-1); }
|
||||||
|
|
||||||
|
/* Misc. getters and setters */
|
||||||
|
/* Cannot modify current_state while shutdown in progress */
|
||||||
|
MyState &get_current_state( void ) { assert( !shutdown_in_progress ); return current_state; }
|
||||||
|
void set_current_state( const MyState &x ) { assert( !shutdown_in_progress ); current_state = x; }
|
||||||
|
void set_verbose( void ) { verbose = true; }
|
||||||
|
|
||||||
/* nonexistent methods to satisfy -Weffc++ */
|
/* nonexistent methods to satisfy -Weffc++ */
|
||||||
TransportSender( const TransportSender &x );
|
TransportSender( const TransportSender &x );
|
||||||
TransportSender & operator=( const TransportSender &x );
|
TransportSender & operator=( const TransportSender &x );
|
||||||
|
|||||||
Reference in New Issue
Block a user