public void unregister ( ExecutionVertexID vertexId , Task task ) { final Environment environment = task . getEnvironment ( ) ; // destroy and remove OUTPUT channels from registered channels and cache for ( ChannelID id : environment . getOutputChannelIDs ( ) ) { Channel channel = this . channels . remove ( id ) ; if ( channel != null ) { channel . destroy ( ) ; } this . receiverCache . remove ( channel ) ; } // destroy and remove INPUT channels from registered channels and cache for ( ChannelID id : environment . getInputChannelIDs ( ) ) { Channel channel = this . channels . remove ( id ) ; if ( channel != null ) { channel . destroy ( ) ; } this . receiverCache . remove ( channel ) ; } // clear and remove INPUT side buffer pools for ( GateID id : environment . getInputGateIDs ( ) ) { LocalBufferPoolOwner bufferPool = this . localBuffersPools . remove ( id ) ; if ( bufferPool != null ) { bufferPool . clearLocalBufferPool ( ) ; } } // clear and remove OUTPUT side buffer pool LocalBufferPoolOwner bufferPool = this . localBuffersPools . remove ( vertexId ) ; if ( bufferPool != null ) { bufferPool . clearLocalBufferPool ( ) ; } // the number of channels per buffers has changed after unregistering the task // => redistribute the number of designated buffers of the registered local buffer pools redistributeBuffers ( ) ; }
Unregisters the given task from the channel manager .