def execute()

in hasher-matcher-actioner/hmalib/scripts/cli/storm.py [0:0]

• 51 lines of code
• 16 McCabe index (conditional complexity)

    def execute(self, api: utils.HasherMatcherActionerAPI) -> None:
        sent_message_count = 0
        jobs = []

        execution_times = []

        if self.retry:
            retry_strategy = Retry(
                total=5,
                backoff_factor=1,
                status_forcelist=[429, 500, 502, 503, 504],
                method_whitelist=[
                    "GET",
                    "PUT",
                    "POST",
                ],  # Including POST as the API should not perform an insert if an error is returned
            )
            api.add_transport_adapter(HTTPAdapter(max_retries=retry_strategy))

        if self.submit_method in ["sns-s3", "sns-url"] and not self.sns_topic:
            print(
                f"submit_method: {self.submit_method} require's the sns topic arn be provided to `--sns_topic`"
            )
            sys.exit(1)
        # Need to compute script level RequestsPerSecond so that we can estimate
        # benchmark performance. For that, storing the start time every 200
        # requests and reporting the QPS between that and current.

        with concurrent.futures.ThreadPoolExecutor(max_workers=48) as executor:
            if self.verbose:
                print("Started adding tasks to executor")
            chunk_start_time_200 = perf_counter()

            while sent_message_count < self.count:
                jobs.append(
                    executor.submit(
                        self._submit,
                        api,
                    )
                )

                sent_message_count += 1
                if self.verbose:
                    print(f"{sent_message_count} requests prepared", end="\r")

            if self.verbose:
                print("\nDone adding tasks to executor")

            for i, completed_future in enumerate(concurrent.futures.as_completed(jobs)):
                execution_times.append(completed_future.result())
                if self.verbose:
                    progress_report_string = f"{i} of {self.count} sent!"

                    rps_report_string = f"Current 200 chunk has QPS of {(i % 200) / (perf_counter() - chunk_start_time_200)}"

                    # Report progress and RPS
                    print(f"{progress_report_string}! {rps_report_string}", end="\r")

                if i % 200 == 0:
                    # Reset chunk start time
                    chunk_start_time_200 = perf_counter()

        if self.verbose:
            print(f"\nSent all {self.count} submissions.")

            # Compute some beginner stats.
            execution_times = sorted(execution_times)
            print(
                f"""Percentiles in ms:
                    p75: {execution_times[int(len(execution_times)*0.75)]}
                    p95: {execution_times[int(len(execution_times)*0.95)]}
                    p99: {execution_times[int(len(execution_times)*0.99)]}
                """
            )