Add scheduler overhead testing

[krlberry]

Adds the following overhead tests for #36 :

• flow_scaling_test
• task_scaling_test

The flow_scaling_test is run conditionally if an existing deployment is detected.

• Finalize csv format
• Validate output in artifacts and results saved to disk (e.g., n_processes in flow case)
• Set final scaling of flow tests (no need to go up to 80k concurrent sub-flows?)
• Add DaskTaskRunner version of task_scaling_test
• Plotting script

[Jan-Willem]

@krlberry can you please split out the csv writing functionality to a separate module so that we can reuse it for the Dask and later airflow tests?

@amcnicho @krlberry @taktsutsumi Do you want to add the DaskTaskRunner to this request or another one? If I understand correctly, the DaskTaskRunner requires setting up a Dask cluster.

[amcnicho]

Creating a DaskTaskRunner that allows deployment to both local process and k3d deployment is fairly straightforward so I think it makes sense to include it in this PR. We can probably reuse/adapt prefect_workflow/resource_management.py

[Jan-Willem]

When I try running the latest version of the code I get the following error:

RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

        To fix this issue, refer to the "Safe importing of main module"
        section in https://docs.python.org/3/library/multiprocessing.html

18:20:32.034 | ERROR   | distributed.nanny - Failed to start process

Have any of you encountered this?

[Jan-Willem]

Part of the problem is that tr = connect_to_scheduler() should not be outside if name == "main":

https://prefecthq.github.io/prefect-dask/task_runners/

[Jan-Willem]

I was also getting an
could not serialize object of type HighLevelGraph', '<ToPickle: HighLevelGraph with 1 layers.
error.

To fix it I had to create the dask cluster instead of using resource_management import connect_to_scheduler.

Code:

@flow(log_prints=True, task_runner=DaskTaskRunner)
def task_scaling_test_dask(number_of_tasks=1000, min_time=0.001, max_time=0.01):

    print(f"Working on {number_of_tasks} tasks in this flow invocation")

    start = time.time()
    results = []
    for num_tasks in range(0, number_of_tasks):
        duration = task_test.submit(0.001, 0.01)
        results.append(duration.result())
    wait(results)
    end = time.time()

    T_sum_task_times = sum(results)

    T_workflow = end - start

    pc = get_run_context()

    timing_results = [
        {
            "date_and_time": datetime.now().isoformat(),
            "developer": os.getlogin(),
            "system_name": platform.node(),
            "workflow": inspect.stack()[0][3],
            "n_tasks": number_of_tasks,
            "min_sleep": min_time,
            "max_sleep": max_time,
            "Wall clock time": T_workflow,
            "Sum of sleep times": T_sum_task_times,
            "n_threads": os.cpu_count(),
            "n_processes": 1,  # think this should always be 1 for the task case
            "n_parallelism": os.cpu_count(),
            "runner": str(type(pc.task_runner)),
            "workflow_orchestration_framework": "prefect",
            "backend_database": "postgres",
            "workflow_type": "task",  # populate via argument?
            "total_memory": psutil.virtual_memory().total / (1024**3),
        }
    ]
    create_table_artifact(table=timing_results)
    return timing_results

if __name__ == "__main__":
    cluster = distributed.LocalCluster(
            n_workers=1,
            threads_per_worker=16,
        )

    print("Running task_scaling_test, but with a dask cluster task_runner")
    sizes = [1000, 2000, 4000, 8000, 16000, 32000, 64000, 80000]
    overall_dask_task_scaling_results = []

    for size in sizes:
        timing_results = task_scaling_test(size, 0.001, 0.01)
        save_timing_results(pd.DataFrame.from_dict(timing_results))
        overall_dask_task_scaling_results.append(timing_results[0])

    # Artifact for overall task_scaling_test results:
    create_table_artifact(
        table=overall_dask_task_scaling_results, key="dask-task-scaling-results"
    )

[Jan-Willem]

The Dask tests were taking a long time due to log printouts. After adding:

import logging
#Set higher log level to suppress INFO messages
logging.getLogger("prefect").setLevel(logging.WARNING)
logging.getLogger("dask").setLevel(logging.WARNING)
logging.getLogger("distributed").setLevel(logging.ERROR)

It is completing a lot quicker now.

[amcnicho]

I think this is ready to merge into main and then run tests in all the dev environments.

[krlberry]

I'd agree about the merge to main. All the tests ran successfully in my environment earlier today.