def METHOD_NAME(
    ctx,
    builders,
    verbosity,
    reporting_store,
    num_workers,
    url,
    port,
    num_chunks,
    no_bars,
    num_processes,
    rabbitmq,
    queue_prefix,
    memray,
    memray_dir,
    memray_file=None,
    follow_fork=False,
):
    # Import profiler and setup directories to dump profiler output
    if memray:
        from memray import FileDestination, Tracker
        if memray_dir:
            import os
            os.makedirs(memray_dir, exist_ok=True)
            memray_file = f"{memray_dir}/{builders[0]}_{datetime.now().isoformat()}.bin"
        else:
            memray_file = f"{settings.TEMP_DIR}/{builders[0]}_{datetime.now().isoformat()}.bin"
        if num_processes > 1:
            follow_fork = True
        # Click context manager handles creation and clean up of profiler dump files for memray tracker
        ctx.obj = ctx.with_resource(
            Tracker(
                destination=FileDestination(memray_file),
                native_traces=False,
                trace_python_allocators=False,
                follow_fork=follow_fork,
            )
        )
    # Import proper manager and worker
    if rabbitmq:
        from maggma.cli.rabbitmq import manager, worker
    else:
        from maggma.cli.distributed import manager, worker
    # Set Logging
    levels = [logging.WARNING, logging.INFO, logging.DEBUG]
    level = levels[min(len(levels) - 1, verbosity)]  # capped to number of levels
    root = logging.getLogger()
    root.setLevel(level)
    ch = TqdmLoggingHandler()
    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    ch.setFormatter(formatter)
    root.addHandler(ch)
    builder_objects = []
    for b in builders:
        if str(b).endswith(".py") or str(b).endswith(".ipynb"):
            builder_objects.append(load_builder_from_source(b))
        else:
            builder_objects.append(loadfn(b))
    builder_objects = [b if isinstance(b, list) else [b] for b in builder_objects]
    builder_objects = list(chain.from_iterable(builder_objects))
    if reporting_store:
        reporting_store = loadfn(reporting_store)
        root.addHandler(ReportingHandler(reporting_store))
    if url:
        if num_chunks > 0:
            # Manager
            if port is None:
                port = find_port()
                root.critical(f"Using random port for mrun manager: {port}")
            if rabbitmq:
                manager(
                    url=url,
                    port=port,
                    builders=builder_objects,
                    num_chunks=num_chunks,
                    num_workers=num_workers,
                    queue_prefix=queue_prefix,
                )
            else:
                manager(
                    url=url,
                    port=port,
                    builders=builder_objects,
                    num_chunks=num_chunks,
                    num_workers=num_workers,
                )
        else:
            # Worker
            if rabbitmq:
                worker(
                    url=url,
                    port=port,
                    num_processes=num_processes,
                    no_bars=no_bars,
                    queue_prefix=queue_prefix,
                )
            else:
                worker(url=url, port=port, num_processes=num_processes, no_bars=no_bars)
    else:
        if num_processes == 1:
            for builder in builder_objects:
                serial(builder, no_bars)
        else:
            loop = asyncio.get_event_loop()
            for builder in builder_objects:
                loop.run_until_complete(multi(builder=builder, num_processes=num_processes, no_bars=no_bars))
    if memray_file:
        import subprocess
        subprocess.METHOD_NAME(["memray", "flamegraph", memray_file], shell=False)