run_benchmark Namespace Reference

Functions
def	verify_destinations
def	get_connection
def	get_run_vars
def	get_gpu_info
def	get_machine_info
def	read_query_files
def	read_setup_teardown_query_files
def	validate_setup_teardown_query_file
def	validate_query_file
def	execute_query
def	calculate_query_times
def	clear_memory
def	clear_system_caches
def	get_mem_usage
def	run_query
def	run_setup_teardown_query
def	json_format_handler
def	create_results_dataset
def	send_results_db
def	send_results_file_json
def	send_results_jenkins_bench
def	send_results_output
def	process_arguments
def	benchmark

Function Documentation

def run_benchmark.benchmark

(

input_arguments

)

Definition at line 1629 of file run_benchmark.py.

References File_Namespace.append(), create_results_dataset(), get_connection(), get_gpu_info(), get_machine_info(), get_run_vars(), process_arguments(), read_query_files(), read_setup_teardown_query_files(), run_query(), run_setup_teardown_query(), send_results_db(), send_results_file_json(), send_results_jenkins_bench(), send_results_output(), split(), to_string(), and verify_destinations().

def benchmark(input_arguments):
    # Set input args to vars
    args = process_arguments(input_arguments)
    verbose = args.verbose
    quiet = args.quiet
    source_db_user = args.user
    source_db_passwd = args.passwd
    source_db_server = args.server
    source_db_port = args.port
    source_db_name = args.name
    source_table = args.table
    label = args.label
    queries_dir = args.queries_dir
    iterations = args.iterations
    gpu_count = args.gpu_count
    gpu_name = args.gpu_name
    no_gather_conn_gpu_info = args.no_gather_conn_gpu_info
    no_gather_nvml_gpu_info = args.no_gather_nvml_gpu_info
    gather_nvml_gpu_info = args.gather_nvml_gpu_info
    machine_name = args.machine_name
    machine_uname = args.machine_uname
    destinations = args.destination
    dest_db_user = args.dest_user
    dest_db_passwd = args.dest_passwd
    dest_db_server = args.dest_server
    dest_db_port = args.dest_port
    dest_db_name = args.dest_name
    dest_table = args.dest_table
    dest_table_schema_file = args.dest_table_schema_file
    output_file_json = args.output_file_json
    output_file_jenkins = args.output_file_jenkins
    output_tag_jenkins = args.output_tag_jenkins
    setup_teardown_queries_dir = args.setup_teardown_queries_dir
    run_setup_teardown_per_query = args.run_setup_teardown_per_query
    foreign_table_filename = args.foreign_table_filename
    jenkins_thresholds_name = args.jenkins_thresholds_name
    jenkins_thresholds_field = args.jenkins_thresholds_field
    clear_all_memory_pre_query = args.clear_all_memory_pre_query
    cuda_block_grid_perf_test = args.cuda_block_grid_perf_test
    show_simplified_result = args.show_simplified_result
    if args.cuda_grid_sizes is None:
        grid_size_range = [2]
    else:
        grid_size_range = args.cuda_grid_sizes
    if args.cuda_block_sizes is None:
        block_size_range = [1024]
    else:
        block_size_range = args.cuda_block_sizes

    # Hard-coded vars
    trim = 0.15

    # Set logging output level
    if verbose:
        logging.basicConfig(level=logging.DEBUG)
    elif quiet:
        logging.basicConfig(level=logging.WARNING)
    else:
        logging.basicConfig(level=logging.INFO)

    # Input validation
    if (iterations > 1) is not True:
        # Need > 1 iteration as first iteration is dropped from calculations
        logging.error("Iterations must be greater than 1")
        exit(1)
    if verify_destinations(
            destinations=destinations,
            dest_db_server=dest_db_server,
            output_file_json=output_file_json,
            output_file_jenkins=output_file_jenkins,
    ):
        logging.debug("Destination(s) have been verified.")
    else:
        logging.error("No valid destination(s) have been set. Exiting.")
        exit(1)

    # Establish connection to mapd db
    con = get_connection(
        db_user=source_db_user,
        db_passwd=source_db_passwd,
        db_server=source_db_server,
        db_port=source_db_port,
        db_name=source_db_name,
    )
    if not con:
        exit(1)  # Exit if cannot connect to db
    # Set run-specific variables (time, uid, etc.)
    run_vars = get_run_vars(con=con)
    # Set GPU info depending on availability
    gpu_info = get_gpu_info(
        gpu_name=gpu_name,
        no_gather_conn_gpu_info=no_gather_conn_gpu_info,
        con=con,
        conn_machine_name=run_vars["conn_machine_name"],
        no_gather_nvml_gpu_info=no_gather_nvml_gpu_info,
        gather_nvml_gpu_info=gather_nvml_gpu_info,
        gpu_count=gpu_count,
    )
    # Set run machine info
    machine_info = get_machine_info(
        conn_machine_name=run_vars["conn_machine_name"],
        machine_name=machine_name,
        machine_uname=machine_uname,
    )
    # Read queries from files, set to queries dir in PWD if not passed in
    if not queries_dir:
        queries_dir = os.path.join(os.path.dirname(__file__), "queries")
    query_list = read_query_files(
        queries_dir=queries_dir, source_table=source_table
    )
    if not query_list:
        exit(1)
    # Read setup/teardown queries if they exist
    setup_query_list, teardown_query_list = \
        read_setup_teardown_query_files(queries_dir=setup_teardown_queries_dir,
                                        source_table=source_table,
                                        foreign_table_filename=foreign_table_filename)
    # Check at what granularity we want to run setup or teardown queries at
    run_global_setup_queries = setup_query_list is not None and not run_setup_teardown_per_query
    run_per_query_setup_queries = setup_query_list is not None and run_setup_teardown_per_query
    run_global_teardown_queries = teardown_query_list is not None and not run_setup_teardown_per_query
    run_per_query_teardown_queries = teardown_query_list is not None and run_setup_teardown_per_query
    # Run global setup queries if they exist
    queries_results = []
    st_qr = run_setup_teardown_query(queries=setup_query_list,
                                     do_run=run_global_setup_queries, trim=trim, con=con)
    queries_results.extend(st_qr)
    new_query_list = {"query_group": "", "queries": []}
    new_query_list.update(query_group=query_list["query_group"])
    if cuda_block_grid_perf_test:
        for query_info in query_list["queries"]:
            query = query_info["mapdql"]
            for block_size in block_size_range:
                for grid_size in grid_size_range:
                    query_hint = "SELECT /*+ g_cuda_block_size(" + str(block_size) + "), "
                    query_hint += "g_cuda_grid_size_multiplier(" + str(grid_size) + ") */ "
                    query_name = query_info["name"] + "_block_size_" + str(block_size) + "_grid_size_" + str(grid_size)
                    new_query = re.sub("select", "SELECT", query, re.IGNORECASE)
                    new_query = new_query.replace("SELECT", query_hint, 1)
                    new_query_list["queries"].append(
                        {"name": query_name, "mapdql": new_query})
            cuda_opt_query_hint = "SELECT /*+ cuda_opt_param */ "
            cuda_opt_query_name = query_info["name"] + "_block_size_-1_grid_size_-1"
            cuda_opt_new_query = re.sub("select", "SELECT", query, re.IGNORECASE)
            cuda_opt_new_query = cuda_opt_new_query.replace("SELECT", cuda_opt_query_hint, 1)
            new_query_list["queries"].append(
                {"name": cuda_opt_query_name, "mapdql": cuda_opt_new_query})
        query_list = new_query_list

    # Run queries
    for query in query_list["queries"]:
        # Run setup queries
        st_qr = run_setup_teardown_query(
            queries=setup_query_list, do_run=run_per_query_setup_queries, trim=trim, con=con)
        queries_results.extend(st_qr)
        # Run benchmark query
        query_result = run_query(
            query=query, iterations=iterations, trim=trim, con=con,
            clear_all_memory_pre_query=clear_all_memory_pre_query
        )
        queries_results.append(query_result)
        # Run tear-down queries
        st_qr = run_setup_teardown_query(
            queries=teardown_query_list, do_run=run_per_query_teardown_queries, trim=trim, con=con)
        queries_results.extend(st_qr)
    logging.info("Completed all queries.")
    # Run global tear-down queries if they exist
    st_qr = run_setup_teardown_query(queries=teardown_query_list,
                                     do_run=run_global_teardown_queries, trim=trim, con=con)
    queries_results.extend(st_qr)
    logging.debug("Closing source db connection.")
    con.close()
    # Generate results dataset
    results_dataset = create_results_dataset(
        run_guid=run_vars["run_guid"],
        run_timestamp=run_vars["run_timestamp"],
        run_connection=run_vars["run_connection"],
        run_machine_name=machine_info["run_machine_name"],
        run_machine_uname=machine_info["run_machine_uname"],
        run_driver=run_vars["run_driver"],
        run_version=run_vars["run_version"],
        run_version_short=run_vars["run_version_short"],
        label=label,
        source_db_gpu_count=gpu_info["source_db_gpu_count"],
        source_db_gpu_driver_ver=gpu_info["source_db_gpu_driver_ver"],
        source_db_gpu_name=gpu_info["source_db_gpu_name"],
        source_db_gpu_mem=gpu_info["source_db_gpu_mem"],
        source_table=source_table,
        trim=trim,
        iterations=iterations,
        query_group=query_list["query_group"],
        queries_results=queries_results,
    )
    results_dataset_json = json.dumps(
        results_dataset, default=json_format_handler, indent=2
    )
    successful_results_dataset = [
        x for x in results_dataset if x["succeeded"] is not False
    ]
    successful_results_dataset_results = []
    for results_dataset_entry in successful_results_dataset:
        successful_results_dataset_results.append(
            results_dataset_entry["results"]
        )

    # Send results to destination(s)
    sent_destination = True
    if "mapd_db" in destinations:
        if not send_results_db(
                results_dataset=successful_results_dataset_results,
                table=dest_table,
                db_user=dest_db_user,
                db_passwd=dest_db_passwd,
                db_server=dest_db_server,
                db_port=dest_db_port,
                db_name=dest_db_name,
                table_schema_file=dest_table_schema_file,
        ):
            sent_destination = False
    if "file_json" in destinations:
        if not send_results_file_json(
                results_dataset_json=results_dataset_json,
                output_file_json=output_file_json,
        ):
            sent_destination = False
    if "jenkins_bench" in destinations:
        if not send_results_jenkins_bench(
                results_dataset=successful_results_dataset_results,
                thresholds_name=jenkins_thresholds_name,
                thresholds_field=jenkins_thresholds_field,
                output_tag_jenkins=output_tag_jenkins,
                output_file_jenkins=output_file_jenkins,
        ):
            sent_destination = False
    if "output" in destinations:
        if not send_results_output(results_dataset_json=results_dataset_json):
            sent_destination = False
    if not sent_destination:
        logging.error("Sending results to one or more destinations failed")
        exit(1)
    else:
        logging.info(
            "Succesfully loaded query results info into destination(s)"
        )
    if show_simplified_result:
        res_header = ['Query', 'Block', 'Grid', 'First', 'Min', 'Max', 'Avg']
        for i in range(1, iterations):
            res_header.append("Run-" + str(i))
        res_data = []
        for perf_info in results_dataset:
            tok = perf_info["name"].split("_")
            cur_query_perf = [tok[0], tok[3], tok[6], str(perf_info["results"]["query_exec_first"]),
                              str(perf_info["results"]["query_exec_min"]), str(perf_info["results"]["query_exec_max"]),
                              str(perf_info["results"]["query_exec_avg"])]
            for query_time in perf_info["debug"]["query_exec_times"]:
                cur_query_perf.append(str(query_time))
            res_data.append(cur_query_perf)
        res_df = pd.DataFrame(res_data, columns=res_header)
        res_df['Query'] = res_df['Query'].astype(str)
        res_df['Block'] = res_df['Block'].astype(int)
        res_df['Grid'] = res_df['Grid'].astype(float)
        res_df['Min'] = res_df['Min'].astype(int)
        res_str = res_df.to_string(header=True, index=False)
        col_desc = "(Block: cuda block size, Grid: cuda grid size multiplier (cuda grid size = # SMs * multiplier))"
        print("[Benchmark result in ms.]\n" + col_desc + "\n" + res_str)

        df1 = res_df.groupby(["Query", "Block"]).apply(
            lambda v: v.sort_values(by=['Min'], ascending=[True])).reset_index(drop=True)
        df2 = df1.groupby(["Query", "Block"]).head(1)
        per_block_str = df2[['Query', 'Block', 'Grid', 'Min']].to_string(index=False)
        per_block_str = per_block_str.replace("-1.00", "opt")
        per_block_str = per_block_str.replace("-1", "opt")
        print("[Best performance per block size in ms.]\n" + per_block_str)

        df3 = res_df.groupby(["Query", "Grid"]).apply(
            lambda v: v.sort_values(by=['Min'], ascending=[True])).reset_index(drop=True)
        df4 = df3.groupby(["Query", "Grid"]).head(1)
        per_grid_str = df4[['Query', 'Grid', 'Block', 'Min']].to_string(index=False)
        per_grid_str = per_grid_str.replace("-1.00", "opt")
        per_grid_str = per_grid_str.replace("-1", "opt")
        print("[Best performance per grid size]\n" + per_grid_str)

        overall_best = res_df.sort_values('Min', ascending=True)[['Query', 'Block', 'Grid', 'Min']].head(1)
        print("[Best performance in all conditions in ms.]\n" + overall_best.to_string(index=False))

Here is the call graph for this function: