Protocol changes to support server-side delayed echoing

This commit is contained in:
Keith Winstein
2012-02-03 16:30:33 -05:00
parent da66244eab
commit 5f31fd3f7a
10 changed files with 127 additions and 92 deletions
+5
View File
@@ -15,6 +15,7 @@ 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 ),
sent_state_late_acked( 0 ),
fragments(), fragments(),
verbose( false ) verbose( false )
{ {
@@ -28,6 +29,7 @@ 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 ),
sent_state_late_acked( 0 ),
fragments(), fragments(),
verbose( false ) verbose( false )
{ {
@@ -48,6 +50,9 @@ void Transport<MyState, RemoteState>::recv( void )
} }
sender.process_acknowledgment_through( inst.ack_num() ); sender.process_acknowledgment_through( inst.ack_num() );
if ( inst.late_ack_num() > sent_state_late_acked ) {
sent_state_late_acked = inst.late_ack_num();
}
/* first, make sure we don't already have the new state */ /* first, make sure we don't already have the new state */
for ( typename list< TimestampedState<RemoteState> >::iterator i = received_states.begin(); for ( typename list< TimestampedState<RemoteState> >::iterator i = received_states.begin();
+15 -11
View File
@@ -30,6 +30,7 @@ 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 */
uint64_t sent_state_late_acked;
FragmentAssembly fragments; FragmentAssembly fragments;
bool verbose; bool verbose;
@@ -53,32 +54,35 @@ namespace Network {
/* Shut down other side of connection. */ /* Shut down other side of connection. */
/* Illegal to change current_state after this. */ /* Illegal to change current_state after this. */
void start_shutdown( void ) { sender.start_shutdown(); } void start_shutdown( void ) { sender.start_shutdown(); }
bool shutdown_in_progress( void ) { return sender.get_shutdown_in_progress(); } bool shutdown_in_progress( void ) const { return sender.get_shutdown_in_progress(); }
bool shutdown_acknowledged( void ) { return sender.get_shutdown_acknowledged(); } bool shutdown_acknowledged( void ) const { return sender.get_shutdown_acknowledged(); }
bool shutdown_ack_timed_out( void ) { return sender.shutdown_ack_timed_out(); } bool shutdown_ack_timed_out( void ) const { return sender.shutdown_ack_timed_out(); }
bool attached( void ) { return connection.get_attached(); } bool attached( void ) const { return connection.get_attached(); }
/* Other side has requested shutdown and we have sent one ACK */ /* Other side has requested shutdown and we have sent one ACK */
bool counterparty_shutdown_ack_sent( void ) { return sender.get_counterparty_shutdown_acknowledged(); } bool counterparty_shutdown_ack_sent( void ) const { return sender.get_counterparty_shutdown_acknowledged(); }
int port( void ) { return connection.port(); } int port( void ) const { return connection.port(); }
string get_key( void ) { return connection.get_key(); } string get_key( void ) const { 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 ); }
uint64_t get_remote_state_num( void ) { return received_states.back().num; } uint64_t get_remote_state_num( void ) const { return received_states.back().num; }
const TimestampedState<RemoteState> & get_latest_remote_state( void ) const { return received_states.back(); } const TimestampedState<RemoteState> & get_latest_remote_state( void ) const { return received_states.back(); }
int fd( void ) { return connection.fd(); } int fd( void ) const { return connection.fd(); }
void set_verbose( void ) { sender.set_verbose(); verbose = true; } void set_verbose( void ) { sender.set_verbose(); verbose = true; }
void set_send_delay( int new_delay ) { sender.set_send_delay( new_delay ); } void set_send_delay( int new_delay ) { sender.set_send_delay( new_delay ); }
uint64_t get_sent_state_acked( void ) { return sender.get_sent_state_acked(); } uint64_t get_sent_state_acked( void ) const { return sender.get_sent_state_acked(); }
uint64_t get_sent_state_last( void ) { return sender.get_sent_state_last(); } uint64_t get_sent_state_last( void ) const { return sender.get_sent_state_last(); }
uint64_t get_sent_state_late_acked( void ) const { return sent_state_late_acked; }
unsigned int send_interval( void ) const { return sender.send_interval(); }
}; };
} }
+3
View File
@@ -155,9 +155,12 @@ bool STMClient::process_network_input( void )
{ {
network->recv(); network->recv();
/* Now give hints to the overlays */
overlays.get_notification_engine().server_heard( network->get_latest_remote_state().timestamp ); overlays.get_notification_engine().server_heard( network->get_latest_remote_state().timestamp );
overlays.get_prediction_engine().set_local_frame_acked( network->get_sent_state_acked() ); overlays.get_prediction_engine().set_local_frame_acked( network->get_sent_state_acked() );
overlays.get_prediction_engine().set_send_interval( network->send_interval() );
overlays.get_prediction_engine().set_local_frame_late_acked( network->get_sent_state_late_acked() );
return true; return true;
} }
+54 -61
View File
@@ -8,10 +8,10 @@
using namespace Overlay; using namespace Overlay;
bool ConditionalOverlay::start_clock( uint64_t local_frame_acked, uint64_t now ) bool ConditionalOverlay::start_clock( uint64_t local_frame_acked, uint64_t now, unsigned int send_interval )
{ {
if ( (local_frame_acked >= expiration_frame) && (expiration_time == uint64_t(-1)) ) { if ( (local_frame_acked >= expiration_frame) && (expiration_time == uint64_t(-1)) ) {
expiration_time = now + 50; expiration_time = now + 25 + 125 * send_interval / 100;
return true; return true;
} }
return false; return false;
@@ -53,7 +53,9 @@ void ConditionalOverlayCell::apply( Framebuffer &fb, uint64_t confirmed_epoch, i
} }
} }
Validity ConditionalOverlayCell::get_validity( const Framebuffer &fb, int row, uint64_t current_frame, uint64_t now ) const Validity ConditionalOverlayCell::get_validity( const Framebuffer &fb, int row,
uint64_t sent_frame, uint64_t early_ack, uint64_t late_ack,
uint64_t now ) const
{ {
if ( !active ) { if ( !active ) {
return Inactive; return Inactive;
@@ -67,7 +69,7 @@ Validity ConditionalOverlayCell::get_validity( const Framebuffer &fb, int row, u
const Cell &current = *( fb.get_cell( row, col ) ); const Cell &current = *( fb.get_cell( row, col ) );
/* see if it hasn't been updated yet */ /* see if it hasn't been updated yet */
if ( current_frame < expiration_frame ) { if ( early_ack < expiration_frame ) {
return Pending; return Pending;
} }
@@ -92,14 +94,15 @@ Validity ConditionalOverlayCell::get_validity( const Framebuffer &fb, int row, u
assert( expiration_time != uint64_t(-1) ); assert( expiration_time != uint64_t(-1) );
if ( now < expiration_time ) { if ( (late_ack >= expiration_frame)
return Pending; || ( (sent_frame <= early_ack) && (expiration_time >= now) ) ) {
}
return IncorrectOrExpired; return IncorrectOrExpired;
} }
Validity ConditionalCursorMove::get_validity( const Framebuffer &fb, uint64_t current_frame, uint64_t now ) const return Pending;
}
Validity ConditionalCursorMove::get_validity( const Framebuffer &fb, uint64_t sent_frame, uint64_t early_ack, uint64_t late_ack, uint64_t now ) const
{ {
if ( !active ) { if ( !active ) {
return Inactive; return Inactive;
@@ -112,7 +115,7 @@ Validity ConditionalCursorMove::get_validity( const Framebuffer &fb, uint64_t cu
return IncorrectOrExpired; return IncorrectOrExpired;
} }
if ( current_frame < expiration_frame ) { if ( early_ack < expiration_frame ) {
return Pending; return Pending;
} }
@@ -123,15 +126,12 @@ Validity ConditionalCursorMove::get_validity( const Framebuffer &fb, uint64_t cu
assert( expiration_time != uint64_t(-1) ); assert( expiration_time != uint64_t(-1) );
if ( now < expiration_time ) { if ( (late_ack >= expiration_frame)
return Pending; || ( (sent_frame <= early_ack) && (expiration_time >= now) ) ) {
return IncorrectOrExpired;
} }
/* return Pending;
fprintf( stderr, "Bad cursor in %d (I thought (%d,%d) vs actual (%d,%d)).\n", (int)current_frame,
row, col, fb.ds.get_cursor_row(), fb.ds.get_cursor_col() );
*/
return IncorrectOrExpired;
} }
void ConditionalCursorMove::apply( Framebuffer &fb, uint64_t confirmed_epoch ) const void ConditionalCursorMove::apply( Framebuffer &fb, uint64_t confirmed_epoch ) const
@@ -349,6 +349,14 @@ void PredictionEngine::init_cursor( const Framebuffer &fb )
fb.ds.get_cursor_row(), fb.ds.get_cursor_row(),
fb.ds.get_cursor_col(), fb.ds.get_cursor_col(),
prediction_epoch ) ); prediction_epoch ) );
cursor().active = true;
} else if ( cursor().tentative_until_epoch != prediction_epoch ) {
cursors.push_back( ConditionalCursorMove( local_frame_sent + 1,
cursor().row,
cursor().col,
prediction_epoch ) );
cursor().active = true; cursor().active = true;
} }
} }
@@ -369,22 +377,13 @@ void PredictionEngine::cull( const Framebuffer &fb )
continue; continue;
} }
/* smash through "shell prompt" barrier if host has let us */
if ( fb.ds.get_cursor_row() == i->row_num ) {
if ( i->first_col != INT_MAX ) {
if ( fb.ds.get_cursor_col() < i->first_col ) {
i->first_col = 0;
}
} else {
i->first_col = fb.ds.get_cursor_col();
}
}
for ( auto j = i->overlay_cells.begin(); j != i->overlay_cells.end(); j++ ) { for ( auto j = i->overlay_cells.begin(); j != i->overlay_cells.end(); j++ ) {
if ( j->start_clock( local_frame_acked, now ) ) { if ( j->start_clock( local_frame_acked, now, send_interval ) ) {
last_scheduled_timeout = max( last_scheduled_timeout, j->expiration_time ); last_scheduled_timeout = max( last_scheduled_timeout, j->expiration_time );
} }
switch ( j->get_validity( fb, i->row_num, local_frame_acked, now ) ) { switch ( j->get_validity( fb, i->row_num,
local_frame_sent, local_frame_acked, local_frame_late_acked,
now ) ) {
case IncorrectOrExpired: case IncorrectOrExpired:
if ( j->tentative( confirmed_epoch ) ) { if ( j->tentative( confirmed_epoch ) ) {
@@ -440,29 +439,23 @@ void PredictionEngine::cull( const Framebuffer &fb )
if ( j->display_time != uint64_t(-1) ) { if ( j->display_time != uint64_t(-1) ) {
if ( now - j->display_time < 75 ) { if ( now - j->display_time < 75 ) {
if ( flagging > 0 ) { if ( flag_score > -10 ) {
flagging--; flag_score--;
} }
} }
} }
/* no break */ /* no break */
case CorrectNoCredit: case CorrectNoCredit:
if ( i->first_col != INT_MAX ) {
if ( j->col < i->first_col ) {
i->first_col = 0;
}
} else {
i->first_col = j->col;
}
j->reset(); j->reset();
break; break;
case Pending: case Pending:
if ( j->display_time != uint64_t(-1) ) { if ( j->display_time != uint64_t(-1) ) {
if ( now - j->display_time >= 150 ) { if ( now - j->display_time > 150 ) {
flagging = 10; if ( flag_score < 10 ) {
flag_score++;
}
} }
} }
break; break;
@@ -474,15 +467,24 @@ void PredictionEngine::cull( const Framebuffer &fb )
i = inext; i = inext;
} }
/* control flagging with hysteresis */
if ( flag_score > 5 ) {
flagging = true;
} else if ( flag_score < -5 ) {
flagging = false;
}
/* go through cursor predictions */ /* go through cursor predictions */
for ( auto it = cursors.begin(); it != cursors.end(); it++ ) { for ( auto it = cursors.begin(); it != cursors.end(); it++ ) {
if ( it->start_clock( local_frame_acked, now ) ) { if ( it->start_clock( local_frame_acked, now, send_interval ) ) {
last_scheduled_timeout = max( last_scheduled_timeout, it->expiration_time ); last_scheduled_timeout = max( last_scheduled_timeout, it->expiration_time );
} }
} }
if ( !cursors.empty() ) { if ( !cursors.empty() ) {
if ( cursor().get_validity( fb, local_frame_acked, now ) == IncorrectOrExpired ) { if ( cursor().get_validity( fb,
local_frame_sent, local_frame_acked, local_frame_late_acked,
now ) == IncorrectOrExpired ) {
/* /*
fprintf( stderr, "Sadly, we're predicting (%d,%d) vs. (%d,%d) [tau: %ld, expiration_time=%ld, now=%ld]\n", fprintf( stderr, "Sadly, we're predicting (%d,%d) vs. (%d,%d) [tau: %ld, expiration_time=%ld, now=%ld]\n",
cursor().row, cursor().col, cursor().row, cursor().col,
@@ -497,7 +499,10 @@ void PredictionEngine::cull( const Framebuffer &fb )
} }
} }
cursors.remove_if( [&]( ConditionalCursorMove &x ) { return (x.get_validity( fb, local_frame_acked, now ) != Pending); } ); cursors.remove_if( [&]( const ConditionalCursorMove &x ) {
return (x.get_validity( fb,
local_frame_sent, local_frame_acked, local_frame_late_acked,
now ) != Pending); } );
} }
ConditionalOverlayRow & PredictionEngine::get_or_make_row( int row_num, int num_cols ) ConditionalOverlayRow & PredictionEngine::get_or_make_row( int row_num, int num_cols )
@@ -554,9 +559,6 @@ void PredictionEngine::new_user_byte( char the_byte, const Framebuffer &fb )
if ( ch == 0x7f ) { /* backspace */ if ( ch == 0x7f ) { /* backspace */
// fprintf( stderr, "Backspace.\n" ); // fprintf( stderr, "Backspace.\n" );
ConditionalOverlayRow &the_row = get_or_make_row( cursor().row, fb.ds.get_width() ); ConditionalOverlayRow &the_row = get_or_make_row( cursor().row, fb.ds.get_width() );
if ( cursor().col <= the_row.first_col ) {
become_tentative();
}
if ( cursor().col > 0 ) { if ( cursor().col > 0 ) {
cursor().col--; cursor().col--;
@@ -654,10 +656,12 @@ void PredictionEngine::new_user_byte( char the_byte, const Framebuffer &fb )
*/ */
cursor().expire( local_frame_sent + 1 ); cursor().expire( local_frame_sent + 1 );
cursor().col++;
/* do we need to wrap? */ /* do we need to wrap? */
if ( cursor().col >= fb.ds.get_width() ) { if ( cursor().col < fb.ds.get_width() - 1 ) {
cursor().col++;
} else {
become_tentative();
newline_carriage_return( fb ); newline_carriage_return( fb );
} }
} }
@@ -681,10 +685,6 @@ void PredictionEngine::new_user_byte( char the_byte, const Framebuffer &fb )
} }
} else if ( act->char_present && (act->ch == L'D') ) { /* left arrow */ } else if ( act->char_present && (act->ch == L'D') ) { /* left arrow */
init_cursor( fb ); init_cursor( fb );
ConditionalOverlayRow &the_row = get_or_make_row( cursor().row, fb.ds.get_width() );
if ( cursor().col <= the_row.first_col ) {
become_tentative();
}
if ( cursor().col > 0 ) { if ( cursor().col > 0 ) {
cursor().col--; cursor().col--;
@@ -733,13 +733,6 @@ void PredictionEngine::become_tentative( void )
{ {
prediction_epoch++; prediction_epoch++;
if ( !cursors.empty() ) {
cursors.push_back( ConditionalCursorMove( local_frame_sent + 1,
cursor().row,
cursor().col,
prediction_epoch ) );
cursor().active = true;
}
/* /*
fprintf( stderr, "Now tentative in epoch %lu (confirmed=%lu)\n", fprintf( stderr, "Now tentative in epoch %lu (confirmed=%lu)\n",
prediction_epoch, confirmed_epoch ); prediction_epoch, confirmed_epoch );
+15 -8
View File
@@ -38,7 +38,7 @@ namespace Overlay {
bool tentative( uint64_t confirmed_epoch ) const { return tentative_until_epoch > confirmed_epoch; } bool tentative( uint64_t confirmed_epoch ) const { return tentative_until_epoch > confirmed_epoch; }
void reset( void ) { expiration_frame = expiration_time = tentative_until_epoch = -1; active = false; } void reset( void ) { expiration_frame = expiration_time = tentative_until_epoch = -1; active = false; }
bool start_clock( uint64_t local_frame_acked, uint64_t now ); bool start_clock( uint64_t local_frame_acked, uint64_t now, unsigned int send_interval );
void expire( uint64_t s_exp ) { expiration_frame = s_exp; expiration_time = uint64_t(-1); } void expire( uint64_t s_exp ) { expiration_frame = s_exp; expiration_time = uint64_t(-1); }
}; };
@@ -48,7 +48,7 @@ namespace Overlay {
void apply( Framebuffer &fb, uint64_t confirmed_epoch ) const; void apply( Framebuffer &fb, uint64_t confirmed_epoch ) const;
Validity get_validity( const Framebuffer &fb, uint64_t current_frame, uint64_t now ) const; Validity get_validity( const Framebuffer &fb, uint64_t sent_frame, uint64_t early_ack, uint64_t late_ack, uint64_t now ) const;
ConditionalCursorMove( uint64_t s_exp, int s_row, int s_col, uint64_t s_tentative ) ConditionalCursorMove( uint64_t s_exp, int s_row, int s_col, uint64_t s_tentative )
: ConditionalOverlay( s_exp, s_col, s_tentative ), row( s_row ) : ConditionalOverlay( s_exp, s_col, s_tentative ), row( s_row )
@@ -66,7 +66,7 @@ namespace Overlay {
that match the original contents */ that match the original contents */
void apply( Framebuffer &fb, uint64_t confirmed_epoch, int row, bool flag ) const; void apply( Framebuffer &fb, uint64_t confirmed_epoch, int row, bool flag ) const;
Validity get_validity( const Framebuffer &fb, int row, uint64_t current_frame, uint64_t now ) const; Validity get_validity( const Framebuffer &fb, int row, uint64_t sent_frame, uint64_t early_ack, uint64_t late_ack, uint64_t now ) const;
ConditionalOverlayCell( uint64_t s_exp, int s_col, uint64_t s_tentative ) ConditionalOverlayCell( uint64_t s_exp, int s_col, uint64_t s_tentative )
: ConditionalOverlay( s_exp, s_col, s_tentative ), : ConditionalOverlay( s_exp, s_col, s_tentative ),
@@ -93,13 +93,12 @@ namespace Overlay {
class ConditionalOverlayRow { class ConditionalOverlayRow {
public: public:
int row_num; int row_num;
int first_col;
vector<ConditionalOverlayCell> overlay_cells; vector<ConditionalOverlayCell> overlay_cells;
void apply( Framebuffer &fb, uint64_t confirmed_epoch, bool flag ) const; void apply( Framebuffer &fb, uint64_t confirmed_epoch, bool flag ) const;
ConditionalOverlayRow( int s_row_num ) : row_num( s_row_num ), first_col( INT_MAX ), overlay_cells() {} ConditionalOverlayRow( int s_row_num ) : row_num( s_row_num ), overlay_cells() {}
}; };
/* the various overlays */ /* the various overlays */
@@ -130,7 +129,7 @@ namespace Overlay {
list<ConditionalCursorMove> cursors; list<ConditionalCursorMove> cursors;
uint64_t local_frame_sent, local_frame_acked; uint64_t local_frame_sent, local_frame_acked, local_frame_late_acked;
ConditionalOverlayRow & get_or_make_row( int row_num, int num_cols ); ConditionalOverlayRow & get_or_make_row( int row_num, int num_cols );
@@ -141,7 +140,8 @@ namespace Overlay {
void newline_carriage_return( const Framebuffer &fb ); void newline_carriage_return( const Framebuffer &fb );
int flagging; int flag_score;
bool flagging;
ConditionalCursorMove & cursor( void ) { assert( !cursors.empty() ); return cursors.back(); } ConditionalCursorMove & cursor( void ) { assert( !cursors.empty() ); return cursors.back(); }
@@ -151,6 +151,8 @@ namespace Overlay {
uint64_t last_scheduled_timeout; uint64_t last_scheduled_timeout;
unsigned int send_interval;
public: public:
void apply( Framebuffer &fb ) const; void apply( Framebuffer &fb ) const;
void new_user_byte( char the_byte, const Framebuffer &fb ); void new_user_byte( char the_byte, const Framebuffer &fb );
@@ -162,11 +164,16 @@ namespace Overlay {
void set_local_frame_sent( uint64_t x ) { local_frame_sent = x; } void set_local_frame_sent( uint64_t x ) { local_frame_sent = x; }
void set_local_frame_acked( uint64_t x ) { local_frame_acked = x; } void set_local_frame_acked( uint64_t x ) { local_frame_acked = x; }
void set_local_frame_late_acked( uint64_t x ) { local_frame_late_acked = x; }
void set_send_interval( unsigned int x ) { send_interval = x; }
PredictionEngine( void ) : last_byte( 0 ), parser(), overlays(), cursors(), PredictionEngine( void ) : last_byte( 0 ), parser(), overlays(), cursors(),
local_frame_sent( 0 ), local_frame_acked( 0 ), local_frame_sent( 0 ), local_frame_acked( 0 ),
local_frame_late_acked( 0 ),
prediction_epoch( 1 ), confirmed_epoch( 0 ), prediction_epoch( 1 ), confirmed_epoch( 0 ),
flagging( 0 ), last_scheduled_timeout( 0 ) flag_score( 0 ), flagging( false ), last_scheduled_timeout( 0 ),
send_interval( 250 )
{ {
} }
}; };
+2
View File
@@ -124,6 +124,8 @@ vector<Fragment> Fragmenter::make_fragments( Instruction &inst, int MTU )
|| (inst.new_num() != last_instruction.new_num()) || (inst.new_num() != last_instruction.new_num())
|| (inst.ack_num() != last_instruction.ack_num()) || (inst.ack_num() != last_instruction.ack_num())
|| (inst.throwaway_num() != last_instruction.throwaway_num()) || (inst.throwaway_num() != last_instruction.throwaway_num())
|| (inst.late_ack_num() != last_instruction.late_ack_num())
|| (inst.protocol_version() != last_instruction.protocol_version())
|| (last_MTU != MTU) ) { || (last_MTU != MTU) ) {
next_instruction_id++; next_instruction_id++;
} }
+1 -1
View File
@@ -70,7 +70,7 @@ namespace Network {
last_instruction.set_new_num( -1 ); last_instruction.set_new_num( -1 );
} }
vector<Fragment> make_fragments( Instruction &inst, int MTU ); vector<Fragment> make_fragments( Instruction &inst, int MTU );
uint64_t last_ack_sent( void ) { return last_instruction.ack_num(); } uint64_t last_ack_sent( void ) const { return last_instruction.ack_num(); }
}; };
} }
+1
View File
@@ -9,6 +9,7 @@ message Instruction {
optional uint64 new_num = 3; optional uint64 new_num = 3;
optional uint64 ack_num = 4; optional uint64 ack_num = 4;
optional uint64 throwaway_num = 5; optional uint64 throwaway_num = 5;
optional uint64 late_ack_num = 7;
optional bytes diff = 6; optional bytes diff = 6;
} }
+19 -4
View File
@@ -21,13 +21,14 @@ TransportSender<MyState>::TransportSender( Connection *s_connection, MyState &in
ack_num( 0 ), ack_num( 0 ),
pending_data_ack( false ), pending_data_ack( false ),
ack_timestamp( 0 ), ack_timestamp( 0 ),
ack_history(),
SEND_MINDELAY( 15 ) SEND_MINDELAY( 15 )
{ {
} }
/* Try to send roughly two frames per RTT, bounded by limits on frame rate */ /* Try to send roughly two frames per RTT, bounded by limits on frame rate */
template <class MyState> template <class MyState>
unsigned int TransportSender<MyState>::send_interval( void ) unsigned int TransportSender<MyState>::send_interval( void ) const
{ {
int SEND_INTERVAL = lrint( ceil( connection->get_SRTT() / 2.0 ) ); int SEND_INTERVAL = lrint( ceil( connection->get_SRTT() / 2.0 ) );
if ( SEND_INTERVAL < SEND_INTERVAL_MIN ) { if ( SEND_INTERVAL < SEND_INTERVAL_MIN ) {
@@ -227,6 +228,7 @@ void TransportSender<MyState>::send_in_fragments( string diff, uint64_t new_num
inst.set_new_num( new_num ); inst.set_new_num( new_num );
inst.set_ack_num( ack_num ); inst.set_ack_num( ack_num );
inst.set_throwaway_num( sent_states.front().num ); inst.set_throwaway_num( sent_states.front().num );
inst.set_late_ack_num( get_late_ack( now ) );
inst.set_diff( diff ); inst.set_diff( diff );
if ( new_num == uint64_t(-1) ) { if ( new_num == uint64_t(-1) ) {
@@ -258,8 +260,8 @@ void TransportSender<MyState>::process_acknowledgment_through( uint64_t ack_num
/* Ignore ack if we have culled the state it's acknowledging */ /* Ignore ack if we have culled the state it's acknowledging */
if ( sent_states.end() != find_if( sent_states.begin(), sent_states.end(), if ( sent_states.end() != find_if( sent_states.begin(), sent_states.end(),
[&]( TimestampedState<MyState> &x ) { return x.num == ack_num; } ) ) { [&]( const TimestampedState<MyState> &x ) { return x.num == ack_num; } ) ) {
sent_states.remove_if( [&]( TimestampedState<MyState> &x ) { return x.num < ack_num; } ); sent_states.remove_if( [&]( const TimestampedState<MyState> &x ) { return x.num < ack_num; } );
} }
assert( !sent_states.empty() ); assert( !sent_states.empty() );
@@ -267,7 +269,7 @@ void TransportSender<MyState>::process_acknowledgment_through( uint64_t ack_num
/* give up on getting acknowledgement for shutdown */ /* give up on getting acknowledgement for shutdown */
template <class MyState> template <class MyState>
bool TransportSender<MyState>::shutdown_ack_timed_out( void ) bool TransportSender<MyState>::shutdown_ack_timed_out( void ) const
{ {
return shutdown_tries >= SHUTDOWN_RETRIES; return shutdown_tries >= SHUTDOWN_RETRIES;
} }
@@ -278,4 +280,17 @@ void TransportSender<MyState>::set_ack_num( uint64_t s_ack_num )
{ {
ack_num = s_ack_num; ack_num = s_ack_num;
ack_timestamp = timestamp(); ack_timestamp = timestamp();
ack_history.push_back( make_pair( ack_num, ack_timestamp ) );
}
/* The "late" ack is for the input state that has had enough time on the host to have been echoed */
template <class MyState>
uint64_t TransportSender<MyState>::get_late_ack( uint64_t now )
{
ack_history.remove_if( [&]( const pair<uint64_t, uint64_t> &x ) { return x.second < now - ECHO_TIMEOUT; } );
if ( !ack_history.empty() ) {
return ack_history.front().first - 1;
} else { /* every ack has gone past the echo timeout */
return ack_num;
}
} }
+12 -7
View File
@@ -24,9 +24,9 @@ namespace Network {
static const int ACK_INTERVAL = 3000; /* ms between empty acks */ static const int ACK_INTERVAL = 3000; /* ms between empty acks */
static const int ACK_DELAY = 100; /* ms before delayed ack */ static const int ACK_DELAY = 100; /* ms before delayed ack */
static const int SHUTDOWN_RETRIES = 3; /* number of shutdown packets to send before giving up */ static const int SHUTDOWN_RETRIES = 3; /* number of shutdown packets to send before giving up */
static const int ECHO_TIMEOUT = 50; /* for late ack */
/* helper methods for tick() */ /* helper methods for tick() */
unsigned int send_interval( void );
void update_assumed_receiver_state( void ); void update_assumed_receiver_state( void );
void rationalize_states( void ); void rationalize_states( void );
void send_to_receiver( string diff ); void send_to_receiver( string diff );
@@ -62,6 +62,9 @@ namespace Network {
bool pending_data_ack; bool pending_data_ack;
uint64_t ack_timestamp; uint64_t ack_timestamp;
list< pair<uint64_t, uint64_t> > ack_history;
uint64_t get_late_ack( uint64_t now ); /* calculate delayed "echo" acknowledgment */
unsigned int SEND_MINDELAY; /* ms to collect all input */ unsigned int SEND_MINDELAY; /* ms to collect all input */
public: public:
@@ -92,16 +95,18 @@ namespace Network {
void set_current_state( const MyState &x ) { assert( !shutdown_in_progress ); current_state = x; } void set_current_state( const MyState &x ) { assert( !shutdown_in_progress ); current_state = x; }
void set_verbose( void ) { verbose = true; } void set_verbose( void ) { verbose = true; }
bool get_shutdown_in_progress( void ) { return shutdown_in_progress; } bool get_shutdown_in_progress( void ) const { return shutdown_in_progress; }
bool get_shutdown_acknowledged( void ) { return sent_states.front().num == uint64_t(-1); } bool get_shutdown_acknowledged( void ) const { return sent_states.front().num == uint64_t(-1); }
bool get_counterparty_shutdown_acknowledged( void ) { return fragmenter.last_ack_sent() == uint64_t(-1); } bool get_counterparty_shutdown_acknowledged( void ) const { return fragmenter.last_ack_sent() == uint64_t(-1); }
uint64_t get_sent_state_acked( void ) { return sent_states.front().num; } uint64_t get_sent_state_acked( void ) const { return sent_states.front().num; }
uint64_t get_sent_state_last( void ) { return sent_states.back().num; } uint64_t get_sent_state_last( void ) const { return sent_states.back().num; }
bool shutdown_ack_timed_out( void ); bool shutdown_ack_timed_out( void ) const;
void set_send_delay( int new_delay ) { SEND_MINDELAY = new_delay; } void set_send_delay( int new_delay ) { SEND_MINDELAY = new_delay; }
unsigned int send_interval( void ) const;
/* 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 );