@ Override public MigratablePipeline findPipelineToMigrate ( LoadImbalance imbalance ) { Set < ? extends MigratablePipeline > candidates = imbalance . getPipelinesOwnedBy ( imbalance . srcOwner ) ; long migrationThreshold = ( long ) ( ( imbalance . maximumLoad - imbalance . minimumLoad ) * MAXIMUM_NO_OF_EVENTS_AFTER_MIGRATION_COEFFICIENT ) ; MigratablePipeline candidate = null ; long loadInSelectedPipeline = 0 ; for ( MigratablePipeline pipeline : candidates ) { long load = imbalance . getLoad ( pipeline ) ; if ( load > loadInSelectedPipeline ) { if ( load < migrationThreshold ) { loadInSelectedPipeline = load ; candidate = pipeline ; } } } return candidate ; }
Attempt to find a pipeline to migrate to a new NioThread .