void remove_task_tracking ( int task ) { RPC . RPCCall rpc = _work . get ( task ) ; if ( rpc == null ) return ; // Already stopped tracking // Atomically attempt to remove the 'dt'. If we win, we are the sole // thread running the dt.onAckAck. Also helps GC: the 'dt' is done (sent // to client and we received the ACKACK), but the rpc might need to stick // around a long time - and the dt might be big. DTask dt = rpc . _dt ; // The existing DTask, if any if ( dt != null && RPC . RPCCall . CAS_DT . compareAndSet ( rpc , dt , null ) ) { assert rpc . _computed : "Still not done #" + task + " " + dt . getClass ( ) + " from " + rpc . _client ; AckAckTimeOutThread . PENDING . remove ( rpc ) ; dt . onAckAck ( ) ; // One-time call on stop-tracking } // Roll-up as many done RPCs as we can, into the _removed_task_ids list while ( true ) { int t = _removed_task_ids . get ( ) ; // Last already-removed ID RPC . RPCCall rpc2 = _work . get ( t + 1 ) ; // RPC of 1st not-removed ID if ( rpc2 == null || rpc2 . _dt != null || ! _removed_task_ids . compareAndSet ( t , t + 1 ) ) break ; // Stop when we hit in-progress tasks _work . remove ( t + 1 ) ; // Else we can remove the tracking now } }
Stop tracking a remote task because we got an ACKACK .