paleo/simulation.py
def simulate_hybrid_parallel(nested_list, layer_list, batch_size, device,
                             network, use_pipeline, use_only_gemm, ppp_comp,
                             ppp_comm, hybrid_workers):
    """Run simulation for hybrid parallel described as in OneWeird Trick."""

    # Pretty hacky implementation. Need organize and refactor later.

    def _simulate_data_parallel_layers(layers):
        # Forward and backward time for data parallel layers.
        forward_times, params_in_bytes = _profile_for_batch_size(
            layers, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)
        backward_times, _ = _profile_for_batch_size(
            layers, 'backward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)

        # Weight sync between data parallel layers.
        comm_scheme = comm.TreeAllReduce(hybrid_workers, network, ppp_comm)
        time_sync_weights = comm_scheme.all_reduce(params_in_bytes)
        time_apply_updates = _profile_for_apply_updates(params_in_bytes,
                                                        device)

        time_fwd, time_bwd = sum(forward_times), sum(backward_times)
        time_fwd, time_bwd = time_fwd.total_time, time_bwd.total_time
        time_apply = time_apply_updates.total_time
        if use_pipeline:
            time_fwd = sum([t.lowerbound for t in forward_times])
            time_bwd = sum([t.lowerbound for t in backward_times])
            time_apply = time_apply_updates.lowerbound
        return (time_fwd, time_bwd, time_apply, time_sync_weights)

    # Get times for different batch_sizes:
    def _simulate_model_parallel_layers(layers):
        forward_times, params_in_bytes = _profile_for_batch_size(
            layers, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)
        backward_times, _ = _profile_for_batch_size(
            layers, 'backward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)  # Use GB/s in profiler.
        time_apply_updates = _profile_for_apply_updates(params_in_bytes,
                                                        device)
        fwd_lower, fwd_upper = _sum_with_parallel(nested_list, layers,
                                                  forward_times)
        bwd_lower, bwd_upper = _sum_with_parallel(nested_list, layers,
                                                  backward_times)
        return fwd_lower, bwd_lower, time_apply_updates.total_time

    effective_batch_size = batch_size * hybrid_workers

    # Separate data model parallel layers.
    data_parallel_layers = filter(lambda l: '@' not in l.name, layer_list)
    model_parallel_layers = filter(lambda l: '@' in l.name, layer_list)
    logger.debug('Data parallel layers: %s' % data_parallel_layers)
    logger.debug('Model parallel layers: %s' % model_parallel_layers)

    (d_time_fwd, d_time_bwd, d_time_apply,
     d_time_sync) = _simulate_data_parallel_layers(data_parallel_layers)
    (m_time_fwd, m_time_bwd,
     m_time_apply) = _simulate_model_parallel_layers(model_parallel_layers)

    # Estimate data fetch time for the last data parallel layer.
    time_fetch = 0
    if len(model_parallel_layers) > 0:
        parent_layer = model_parallel_layers[0].parents[0]
        bytes_to_transfer = np.prod(parent_layer.layer_op.outputs)
        time_fetch = bytes_to_transfer / 2 ** 30 / (network.bandwidth /
                                                    8) * 10 ** 3
        time_fetch *= (hybrid_workers - 1)

        # Multiple model parallel stages.
        # Only one fetch_time because of pipeline.
        m_time_fwd = (m_time_fwd - time_fetch) * hybrid_workers + time_fetch
        m_time_bwd = (m_time_bwd - time_fetch) * hybrid_workers + time_fetch
        logger.info('Saved fetch time from %s by pipelining: %f' %
                    (parent_layer, time_fetch * (hybrid_workers - 1)))

    headers = [
        'workers', 'batch_size', 'fwd_time', 'bwd_time', 'apply_time',
        'sync_time(tree)'
    ]
    results = []
    results.append([
        hybrid_workers, effective_batch_size, d_time_fwd, d_time_bwd,
        d_time_apply, d_time_sync
    ])
    results.append([
        hybrid_workers, effective_batch_size, m_time_fwd, m_time_bwd,
        m_time_apply, 0
    ])
    return headers, results
Run simulation for hybrid parallel described as in OneWeird Trick.
Run simulation for hybrid parallel described as in OneWeird Trick.
Run simulation for hybrid parallel described as in OneWeird Trick .
def simulate_hybrid_parallel(nested_list, layer_list, batch_size, device,
                             network, use_pipeline, use_only_gemm, ppp_comp,
                             ppp_comm, hybrid_workers):
    def _simulate_data_parallel_layers(layers):
        forward_times, params_in_bytes = _profile_for_batch_size(
            layers, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)
        backward_times, _ = _profile_for_batch_size(
            layers, 'backward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)
        comm_scheme = comm.TreeAllReduce(hybrid_workers, network, ppp_comm)
        time_sync_weights = comm_scheme.all_reduce(params_in_bytes)
        time_apply_updates = _profile_for_apply_updates(params_in_bytes,
                                                        device)
        time_fwd, time_bwd = sum(forward_times), sum(backward_times)
        time_fwd, time_bwd = time_fwd.total_time, time_bwd.total_time
        time_apply = time_apply_updates.total_time
        if use_pipeline:
            time_fwd = sum([t.lowerbound for t in forward_times])
            time_bwd = sum([t.lowerbound for t in backward_times])
            time_apply = time_apply_updates.lowerbound
        return (time_fwd, time_bwd, time_apply, time_sync_weights)
    def _simulate_model_parallel_layers(layers):
        forward_times, params_in_bytes = _profile_for_batch_size(
            layers, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)
        backward_times, _ = _profile_for_batch_size(
            layers, 'backward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm, network.bandwidth / 8)  
        time_apply_updates = _profile_for_apply_updates(params_in_bytes,
                                                        device)
        fwd_lower, fwd_upper = _sum_with_parallel(nested_list, layers,
                                                  forward_times)
        bwd_lower, bwd_upper = _sum_with_parallel(nested_list, layers,
                                                  backward_times)
        return fwd_lower, bwd_lower, time_apply_updates.total_time
    effective_batch_size = batch_size * hybrid_workers
    data_parallel_layers = filter(lambda l: '@' not in l.name, layer_list)
    model_parallel_layers = filter(lambda l: '@' in l.name, layer_list)
    logger.debug('Data parallel layers: %s' % data_parallel_layers)
    logger.debug('Model parallel layers: %s' % model_parallel_layers)
    (d_time_fwd, d_time_bwd, d_time_apply,
     d_time_sync) = _simulate_data_parallel_layers(data_parallel_layers)
    (m_time_fwd, m_time_bwd,
     m_time_apply) = _simulate_model_parallel_layers(model_parallel_layers)
    time_fetch = 0
    if len(model_parallel_layers) > 0:
        parent_layer = model_parallel_layers[0].parents[0]
        bytes_to_transfer = np.prod(parent_layer.layer_op.outputs)
        time_fetch = bytes_to_transfer / 2 ** 30 / (network.bandwidth /
                                                    8) * 10 ** 3
        time_fetch *= (hybrid_workers - 1)
        m_time_fwd = (m_time_fwd - time_fetch) * hybrid_workers + time_fetch
        m_time_bwd = (m_time_bwd - time_fetch) * hybrid_workers + time_fetch
        logger.info('Saved fetch time from %s by pipelining: %f' %
                    (parent_layer, time_fetch * (hybrid_workers - 1)))
    headers = [
        'workers', 'batch_size', 'fwd_time', 'bwd_time', 'apply_time',
        'sync_time(tree)'
    ]
    results = []
    results.append([
        hybrid_workers, effective_batch_size, d_time_fwd, d_time_bwd,
        d_time_apply, d_time_sync
    ])
    results.append([
        hybrid_workers, effective_batch_size, m_time_fwd, m_time_bwd,
        m_time_apply, 0
    ])
    return headers, results
def simulate_hybrid_parallel ( nested_list , layer_list , batch_size , device , network , use_pipeline , use_only_gemm , ppp_comp , ppp_comm , hybrid_workers ) : def _simulate_data_parallel_layers ( layers ) : forward_times , params_in_bytes = _profile_for_batch_size ( layers , 'forward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm , network . bandwidth / 8 ) backward_times , _ = _profile_for_batch_size ( layers , 'backward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm , network . bandwidth / 8 ) comm_scheme = comm . TreeAllReduce ( hybrid_workers , network , ppp_comm ) time_sync_weights = comm_scheme . all_reduce ( params_in_bytes ) time_apply_updates = _profile_for_apply_updates ( params_in_bytes , device ) time_fwd , time_bwd = sum ( forward_times ) , sum ( backward_times ) time_fwd , time_bwd = time_fwd . total_time , time_bwd . total_time time_apply = time_apply_updates . total_time if use_pipeline : time_fwd = sum ( [ t . lowerbound for t in forward_times ] ) time_bwd = sum ( [ t . lowerbound for t in backward_times ] ) time_apply = time_apply_updates . lowerbound return ( time_fwd , time_bwd , time_apply , time_sync_weights ) def _simulate_model_parallel_layers ( layers ) : forward_times , params_in_bytes = _profile_for_batch_size ( layers , 'forward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm , network . bandwidth / 8 ) backward_times , _ = _profile_for_batch_size ( layers , 'backward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm , network . bandwidth / 8 ) time_apply_updates = _profile_for_apply_updates ( params_in_bytes , device ) fwd_lower , fwd_upper = _sum_with_parallel ( nested_list , layers , forward_times ) bwd_lower , bwd_upper = _sum_with_parallel ( nested_list , layers , backward_times ) return fwd_lower , bwd_lower , time_apply_updates . total_time effective_batch_size = batch_size * hybrid_workers data_parallel_layers = filter ( lambda l : '@' not in l . name , layer_list ) model_parallel_layers = filter ( lambda l : '@' in l . name , layer_list ) logger . debug ( 'Data parallel layers: %s' % data_parallel_layers ) logger . debug ( 'Model parallel layers: %s' % model_parallel_layers ) ( d_time_fwd , d_time_bwd , d_time_apply , d_time_sync ) = _simulate_data_parallel_layers ( data_parallel_layers ) ( m_time_fwd , m_time_bwd , m_time_apply ) = _simulate_model_parallel_layers ( model_parallel_layers ) time_fetch = 0 if len ( model_parallel_layers ) > 0 : parent_layer = model_parallel_layers [ 0 ] . parents [ 0 ] bytes_to_transfer = np . prod ( parent_layer . layer_op . outputs ) time_fetch = bytes_to_transfer / 2 ** 30 / ( network . bandwidth / 8 ) * 10 ** 3 time_fetch *= ( hybrid_workers - 1 ) m_time_fwd = ( m_time_fwd - time_fetch ) * hybrid_workers + time_fetch m_time_bwd = ( m_time_bwd - time_fetch ) * hybrid_workers + time_fetch logger . info ( 'Saved fetch time from %s by pipelining: %f' % ( parent_layer , time_fetch * ( hybrid_workers - 1 ) ) ) headers = [ 'workers' , 'batch_size' , 'fwd_time' , 'bwd_time' , 'apply_time' , 'sync_time(tree)' ] results = [ ] results . append ( [ hybrid_workers , effective_batch_size , d_time_fwd , d_time_bwd , d_time_apply , d_time_sync ] ) results . append ( [ hybrid_workers , effective_batch_size , m_time_fwd , m_time_bwd , m_time_apply , 0 ] ) return headers , results
Run simulation for hybrid parallel described as in OneWeird Trick.
Run simulation for hybrid parallel described as in OneWeird Trick .
"""Run simulation for hybrid parallel described as in OneWeird Trick.""" # Pretty hacky implementation. Need organize and refactor later. # Forward and backward time for data parallel layers. # Weight sync between data parallel layers. # Get times for different batch_sizes: # Use GB/s in profiler. # Separate data model parallel layers. # Estimate data fetch time for the last data parallel layer. # Multiple model parallel stages. # Only one fetch_time because of pipeline.
nested_list
layer_list
batch_size
device
network
use_pipeline
use_only_gemm
ppp_comp
ppp_comm
hybrid_workers