<jupyter_start><jupyter_text>Searching optimal parallelism configurationsThe objective of this notebook is to find the best possible configuration of the downloader. In general we will be interested in optimizing two KPIs:- the overall download time of a given set of byte ranges- the time to get the first chunks so that we can get started with processingWe have observed in the `cold_latencies` notebook that download times between consecutive invocations of the same container do not change significantly, so we ignore that dimension in this analysis.<jupyter_code>import os

region_name="us-east-2"
binary_name="lambda"
aws_profile=os.environ["AWS_PROFILE"] # Specify the profile you want to use from your .aws/credentials file with the AWS_PROFILE env variable

MEGA = 1024*1024<jupyter_output><empty_output><jupyter_text>Get the name of the lambda function deployed with the `infra` notebook<jupyter_code>lambda_name = !docker run \
    --rm \
    -v $HOME/.aws/credentials:/creds:ro \
    -v cloud-reader-tf:/mnt/state_vol \
    cloudfuse/cloud-reader-terraform output lambda_arn
lambda_name = lambda_name[0][1:len(lambda_name[0])-1]
print('lambda_name:', lambda_name)<jupyter_output>lambda_name: arn:aws:lambda:us-east-2:615900053518:function:cloud-reader-benchmark<jupyter_text>Define the function invokation routines and plottingWe launch multiple invokations in parallel (~100) to get good averages statistics in different locations of the datacenters.<jupyter_code>from joblib import Parallel, delayed
import boto3
import json
import base64
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('once')

def q90(x):
    return x.quantile(0.9)
def q99(x):
    return x.quantile(0.99)

# call the AWS Lambda functions with the reader binary
def invoke_function(initial_permits, release_rate, max_parallel, ranges, show_logs = False):
    session = boto3.Session(profile_name=aws_profile)
    client = session.client('lambda', region_name = region_name)
    inputParams = {
        "region": region_name,
        "bucket": "cloudfuse-taxi-data",
        "key": "synthetic/pattern-3gb",
        "size": 1024*1024*16*200,
        "ranges": ranges,
        "max_parallel": max_parallel,
        "initial_permits": initial_permits,
        "release_rate": release_rate,
        
    }
    response = client.invoke(
        FunctionName = lambda_name,
        InvocationType = 'RequestResponse',
        Payload = json.dumps(inputParams),
        LogType='Tail' if show_logs else 'None'
    )
    if show_logs:
        print(base64.b64decode(response['LogResult']).decode("utf-8") )
    return json.load(response['Payload'])

# run `nb_execution` instances in parallel, each trying out all download parallelisms between 1 and `max_max_parallel`
def run_different_parallelism(chunk_megabytes, nb_execution, max_max_parallel):
    def invoke_batch():
        results = []
        for max_parallel in range(1, max_max_parallel+1):
            results.append({
                "response": invoke_function(max_parallel, 1, max_parallel, [{"start": i*chunk_megabytes*MEGA, "length": chunk_megabytes*MEGA} for i in range(0,20)]),
                "max_parallel": max_parallel,
                "initial_permits": max_parallel,
                "release_rate": 1,
            })
        return results

    return  [item for sublist in Parallel(n_jobs=nb_execution)(delayed(invoke_batch)() for i in range(nb_execution)) for item in sublist]

# run `nb_execution` instances in parallel, with a configuration that increases the number of parallel downloads from 1 to 16 gradually
def run_progressive_release(chunk_megabytes, nb_execution):
    def invoke_one():
        return [{
                "response": invoke_function(1, 2, 16, [{"start": i*chunk_megabytes*MEGA, "length": chunk_megabytes*MEGA} for i in range(0,20)]),
                "max_parallel": 16,
                "initial_permits": 1,
                "release_rate": 2,
            }]

    return  [item for sublist in Parallel(n_jobs=nb_execution)(delayed(invoke_one)() for i in range(nb_execution)) for item in sublist]

# flatten the results to a pandas and plot them
def plot_bench(results):
    downloads = []
    for res in results:
        for dl in res['response']['cache_stats']:
            downloads.append({
                "dl_duration": dl['dl_duration'],
                "first_read": res['response']['range_durations'][0],
                "last_read": res['response']['range_durations'][-1],
                "initial_permits": res['initial_permits'],
                "release_rate": res['release_rate'],
                "max_parallel": res['max_parallel'],
            })

    df = pd.DataFrame(downloads)

    df_grouped = df.groupby(['initial_permits', 'release_rate']).agg({
        'dl_duration': 'mean', 
        'first_read': ['mean', q90, q99], 
        'last_read': ['mean', q90, q99]
    }).reset_index(level='release_rate')
    plt.figure();
    
    different_parallelism_df = df_grouped[df_grouped['release_rate']==1].drop(columns=['release_rate'])
    ax = different_parallelism_df.plot(figsize=[20,10], color=['blue', 'lightgreen', 'green', 'darkgreen', 'pink', 'red', 'brown'], grid=True)
    
    progressive_release_df = df_grouped[df_grouped['release_rate']==2].drop(columns=['release_rate'])
    progressive_release_df.plot(figsize=[20,10], color=['blue', 'lightgreen', 'green', 'darkgreen', 'pink', 'red', 'brown'], grid=True, ax=ax, marker='o', legend=False, use_index=False)<jupyter_output><empty_output><jupyter_text>10MB chunks**Note:** We ran the same experience at different points in time to check if the results where consistent.<jupyter_code>!date
res = run_different_parallelism(chunk_megabytes = 10, nb_execution = 100, max_max_parallel = 16)
res.extend(run_progressive_release(chunk_megabytes = 10, nb_execution = 100))
plot_bench(res)
!date
res = run_different_parallelism(chunk_megabytes = 10, nb_execution = 100, max_max_parallel = 16)
res.extend(run_progressive_release(chunk_megabytes = 10, nb_execution = 100))
plot_bench(res)<jupyter_output>Mon Aug  9 15:35:16 CEST 2021<jupyter_text>100MB chunksWe also check that the results are similar with larger chunks<jupyter_code>!date
res = run_different_parallelism(chunk_megabytes = 100, nb_execution = 50, max_max_parallel = 16)
res.extend(run_progressive_release(chunk_megabytes = 100, nb_execution = 50))
plot_bench(res)<jupyter_output>Mon Aug  9 10:46:17 CEST 2021
{'fraction_non_alphanumeric': 0.11052631578947368, 'fraction_numerical': 0.025696594427244583, 'mean_word_length': 4.227346278317152, 'pattern_counts': {'":': 26, '<': 18, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '504667', 'license': 'MIT', 'n_tokens_mistral': 2151, 'n_tokens_neox': 1968, 'n_words': 512, 'path': 'notebooks/optimal_parallelism.ipynb', 'repo_name': 'cloudfuse-io/cloud-readers-rs', 'size': '6460'}