paleo/simulation.py
def simulate_scaling(layer_dependencies, layer_list, worker_sizes,
                     scaling_type, batch_size, device, network, use_pipeline,
                     use_only_gemm, ppp_comp, ppp_comm):
    """Run simulation for data parallel."""
    logger.debug(
        'Simulate scaling:\n  type: %s\n  device: %s\n  batch size: %s' %
        (scaling_type, device.name, batch_size))

    if scaling_type == 'weak':
        # Get times for different batch_sizes:
        forward_times, params_in_bytes = _profile_for_batch_size(
            layer_list, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm)
        backward_times, _ = _profile_for_batch_size(
            layer_list, 'backward', device, batch_size, use_only_gemm,
            ppp_comp, ppp_comm)

    all_times = []
    for num_workers in worker_sizes:
        if scaling_type == 'weak':
            num_iterations = 1
            batch_size_per_node = batch_size
            batch_size_per_iteration = batch_size
            effective_batch_size = batch_size * num_workers
        elif scaling_type == 'strong':
            batch_size_per_node = int(math.ceil(batch_size / num_workers))
            # Maximum batch size per worker is 128.
            # batch_size_per_iteration = min(batch_size_per_node, 128)
            batch_size_per_iteration = batch_size_per_node
            num_iterations = int(
                math.ceil(batch_size_per_node / batch_size_per_iteration))
            effective_batch_size = batch_size

            # Get times for different batch_sizes:
            forward_times, params_in_bytes = _profile_for_batch_size(
                layer_list, 'forward', device, batch_size_per_iteration,
                use_only_gemm, ppp_comp, ppp_comm)
            backward_times, _ = _profile_for_batch_size(
                layer_list, 'backward', device, batch_size_per_iteration,
                use_only_gemm, ppp_comp, ppp_comm)
        else:
            raise ValueError('Unknown scaling type: %s' % scaling_type)

        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

        # For each node size, we collect
        #     workers,effective_batch_size, num_iter, fwd_time, bwd_time,
        #     apply_time, comp_time, [sync_times...]
        times = [
            num_workers, effective_batch_size, num_iterations, time_fwd,
            time_bwd, time_apply, time_fwd + time_bwd + time_apply
        ]
        times.extend([
            c.all_reduce(params_in_bytes)
            for c in comm.get_all_comm_schemes(num_workers, network, ppp_comm)
        ])
        all_times.append(times)

    headers = [
        'workers', 'batch_size', 'iter', 'time_fwd', 'time_bwd', 'time_apply',
        'time_comp'
    ]
    headers.extend(
        [c.name for c in comm.get_all_comm_schemes(1, network, ppp_comm)])
    assert len(headers) == len(all_times[0])

    return headers, all_times
Run simulation for data parallel.
Run simulation for data parallel.
Run simulation for data parallel .
def simulate_scaling(layer_dependencies, layer_list, worker_sizes,
                     scaling_type, batch_size, device, network, use_pipeline,
                     use_only_gemm, ppp_comp, ppp_comm):
    logger.debug(
        'Simulate scaling:\n  type: %s\n  device: %s\n  batch size: %s' %
        (scaling_type, device.name, batch_size))
    if scaling_type == 'weak':
        forward_times, params_in_bytes = _profile_for_batch_size(
            layer_list, 'forward', device, batch_size, use_only_gemm, ppp_comp,
            ppp_comm)
        backward_times, _ = _profile_for_batch_size(
            layer_list, 'backward', device, batch_size, use_only_gemm,
            ppp_comp, ppp_comm)
    all_times = []
    for num_workers in worker_sizes:
        if scaling_type == 'weak':
            num_iterations = 1
            batch_size_per_node = batch_size
            batch_size_per_iteration = batch_size
            effective_batch_size = batch_size * num_workers
        elif scaling_type == 'strong':
            batch_size_per_node = int(math.ceil(batch_size / num_workers))
            batch_size_per_iteration = batch_size_per_node
            num_iterations = int(
                math.ceil(batch_size_per_node / batch_size_per_iteration))
            effective_batch_size = batch_size
            forward_times, params_in_bytes = _profile_for_batch_size(
                layer_list, 'forward', device, batch_size_per_iteration,
                use_only_gemm, ppp_comp, ppp_comm)
            backward_times, _ = _profile_for_batch_size(
                layer_list, 'backward', device, batch_size_per_iteration,
                use_only_gemm, ppp_comp, ppp_comm)
        else:
            raise ValueError('Unknown scaling type: %s' % scaling_type)
        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
        times = [
            num_workers, effective_batch_size, num_iterations, time_fwd,
            time_bwd, time_apply, time_fwd + time_bwd + time_apply
        ]
        times.extend([
            c.all_reduce(params_in_bytes)
            for c in comm.get_all_comm_schemes(num_workers, network, ppp_comm)
        ])
        all_times.append(times)
    headers = [
        'workers', 'batch_size', 'iter', 'time_fwd', 'time_bwd', 'time_apply',
        'time_comp'
    ]
    headers.extend(
        [c.name for c in comm.get_all_comm_schemes(1, network, ppp_comm)])
    assert len(headers) == len(all_times[0])
    return headers, all_times
def simulate_scaling ( layer_dependencies , layer_list , worker_sizes , scaling_type , batch_size , device , network , use_pipeline , use_only_gemm , ppp_comp , ppp_comm ) : logger . debug ( 'Simulate scaling:\n  type: %s\n  device: %s\n  batch size: %s' % ( scaling_type , device . name , batch_size ) ) if scaling_type == 'weak' : forward_times , params_in_bytes = _profile_for_batch_size ( layer_list , 'forward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm ) backward_times , _ = _profile_for_batch_size ( layer_list , 'backward' , device , batch_size , use_only_gemm , ppp_comp , ppp_comm ) all_times = [ ] for num_workers in worker_sizes : if scaling_type == 'weak' : num_iterations = 1 batch_size_per_node = batch_size batch_size_per_iteration = batch_size effective_batch_size = batch_size * num_workers elif scaling_type == 'strong' : batch_size_per_node = int ( math . ceil ( batch_size / num_workers ) ) batch_size_per_iteration = batch_size_per_node num_iterations = int ( math . ceil ( batch_size_per_node / batch_size_per_iteration ) ) effective_batch_size = batch_size forward_times , params_in_bytes = _profile_for_batch_size ( layer_list , 'forward' , device , batch_size_per_iteration , use_only_gemm , ppp_comp , ppp_comm ) backward_times , _ = _profile_for_batch_size ( layer_list , 'backward' , device , batch_size_per_iteration , use_only_gemm , ppp_comp , ppp_comm ) else : raise ValueError ( 'Unknown scaling type: %s' % scaling_type ) 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 times = [ num_workers , effective_batch_size , num_iterations , time_fwd , time_bwd , time_apply , time_fwd + time_bwd + time_apply ] times . extend ( [ c . all_reduce ( params_in_bytes ) for c in comm . get_all_comm_schemes ( num_workers , network , ppp_comm ) ] ) all_times . append ( times ) headers = [ 'workers' , 'batch_size' , 'iter' , 'time_fwd' , 'time_bwd' , 'time_apply' , 'time_comp' ] headers . extend ( [ c . name for c in comm . get_all_comm_schemes ( 1 , network , ppp_comm ) ] ) assert len ( headers ) == len ( all_times [ 0 ] ) return headers , all_times
Run simulation for data parallel.
Run simulation for data parallel .
"""Run simulation for data parallel.""" # Get times for different batch_sizes: # Maximum batch size per worker is 128. # batch_size_per_iteration = min(batch_size_per_node, 128) # Get times for different batch_sizes: # For each node size, we collect #     workers,effective_batch_size, num_iter, fwd_time, bwd_time, #     apply_time, comp_time, [sync_times...]
layer_dependencies
layer_list
worker_sizes
scaling_type
batch_size
device
network
use_pipeline
use_only_gemm
ppp_comp
ppp_comm