Merge branch 'master' into upgrade-base-image

experiment/build/builder.py

@@ -16,7 +16,6 @@
 
 import argparse
 import itertools
-import multiprocessing
 from multiprocessing import pool as mp_pool
 import os
 import random
@@ -35,15 +34,13 @@
 from common import logs
 
 from experiment.build import build_utils
+from experiment.run_experiment import DEFAULT_CONCURRENT_BUILDS
 
 if not experiment_utils.is_local_experiment():
     import experiment.build.gcb_build as buildlib
 else:
     import experiment.build.local_build as buildlib
 
-DEFAULT_MAX_CONCURRENT_BUILDS = max(min(2 * multiprocessing.cpu_count(), 150),
-                                    30)
-
 # Build attempts and wait interval.
 NUM_BUILD_ATTEMPTS = 3
 BUILD_FAIL_WAIT = 5 * 60
@@ -109,17 +106,13 @@ def build_measurer(benchmark: str) -> bool:
         return False
 
 
-def build_all_measurers(
-        benchmarks: List[str],
-        num_concurrent_builds: int = DEFAULT_MAX_CONCURRENT_BUILDS
-) -> List[str]:
+def build_all_measurers(benchmarks: List[str]) -> List[str]:
     """Build measurers for each benchmark in |benchmarks| in parallel
     Returns a list of benchmarks built successfully."""
     logger.info('Building measurers.')
     filesystem.recreate_directory(build_utils.get_coverage_binaries_dir())
     build_measurer_args = [(benchmark,) for benchmark in benchmarks]
-    successful_calls = retry_build_loop(build_measurer, build_measurer_args,
-                                        num_concurrent_builds)
+    successful_calls = retry_build_loop(build_measurer, build_measurer_args)
     logger.info('Done building measurers.')
     # Return list of benchmarks (like the list we were passed as an argument)
     # instead of returning a list of tuples each containing a benchmark.
@@ -142,12 +135,12 @@ def split_successes_and_failures(inputs: List,
     return successes, failures
 
 
-def retry_build_loop(build_func: Callable, inputs: List[Tuple],
-                     num_concurrent_builds: int) -> List:
+def retry_build_loop(build_func: Callable, inputs: List[Tuple]) -> List:
     """Calls |build_func| in parallel on |inputs|. Repeat on failures up to
     |NUM_BUILD_ATTEMPTS| times. Returns the list of inputs that |build_func| was
     called successfully on."""
     successes = []
+    num_concurrent_builds = os.getenv('CONCURRENT_BUILDS')
     logs.info('Concurrent builds: %d.', num_concurrent_builds)
     with mp_pool.ThreadPool(num_concurrent_builds) as pool:
         for _ in range(NUM_BUILD_ATTEMPTS):
@@ -185,11 +178,8 @@ def build_fuzzer_benchmark(fuzzer: str, benchmark: str) -> bool:
     return True
 
 
-def build_all_fuzzer_benchmarks(
-        fuzzers: List[str],
-        benchmarks: List[str],
-        num_concurrent_builds: int = DEFAULT_MAX_CONCURRENT_BUILDS
-) -> List[str]:
+def build_all_fuzzer_benchmarks(fuzzers: List[str],
+                                benchmarks: List[str]) -> List[str]:
     """Build fuzzer,benchmark images for all pairs of |fuzzers| and |benchmarks|
     in parallel. Returns a list of fuzzer,benchmark pairs that built
     successfully."""
@@ -200,8 +190,7 @@ def build_all_fuzzer_benchmarks(
     # TODO(metzman): Use an asynchronous unordered map variant to schedule
     # eagerly.
     successful_calls = retry_build_loop(build_fuzzer_benchmark,
-                                        build_fuzzer_benchmark_args,
-                                        num_concurrent_builds)
+                                        build_fuzzer_benchmark_args)
     logger.info('Done building fuzzer benchmarks.')
     return successful_calls
 
@@ -222,19 +211,19 @@ def main():
                         help='Fuzzer names.',
                         nargs='+',
                         required=True)
-
     parser.add_argument('-n',
                         '--num-concurrent-builds',
                         help='Max concurrent builds allowed.',
                         type=int,
-                        default=DEFAULT_MAX_CONCURRENT_BUILDS,
+                        default=DEFAULT_CONCURRENT_BUILDS,
                         required=False)
 
     logs.initialize()
     args = parser.parse_args()
+    os.environ['CONCURRENT_BUILDS'] = os.getenv('CONCURRENT_BUILDS',
+                                                args.num_concurrent_builds)
 
-    build_all_fuzzer_benchmarks(args.fuzzers, args.benchmarks,
-                                args.num_concurrent_builds)
+    build_all_fuzzer_benchmarks(args.fuzzers, args.benchmarks)
 
     return 0
 

experiment/build/gcb_build.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 """Module for building things on Google Cloud Build for use in trials."""
 
+import os
 import subprocess
 import tempfile
 from typing import Dict
@@ -30,10 +31,6 @@
 # Maximum time to wait for a GCB config to finish build.
 GCB_BUILD_TIMEOUT = 13 * 60 * 60  # 4 hours.
 
-# TODO(metzman): Make configurable.
-WORKER_POOL_NAME = (
-    'projects/fuzzbench/locations/us-central1/workerPools/buildpool')
-
 logger = logs.Logger('builder')  # pylint: disable=invalid-name
 
 
@@ -81,10 +78,8 @@ def _build(
         logger.debug('Using build configuration: %s' % config)
 
         config_arg = '--config=%s' % config_file.name
-
         # Use "s" suffix to denote seconds.
         timeout_arg = '--timeout=%ds' % timeout_seconds
-        worker_pool_arg = f'--worker-pool={WORKER_POOL_NAME}'
 
         command = [
             'gcloud',
@@ -93,9 +88,13 @@ def _build(
             str(utils.ROOT_DIR),
             config_arg,
             timeout_arg,
-            worker_pool_arg,
         ]
 
+        worker_pool_name = os.getenv('WORKER_POOL_NAME')
+        if worker_pool_name:
+            worker_pool_arg = (f'--worker-pool={worker_pool_name}')
+            command.append(worker_pool_arg)
+
         # Don't write to stdout to make concurrent building faster. Otherwise
         # writing becomes the bottleneck.
         result = new_process.execute(command,

experiment/dispatcher.py

@@ -89,7 +89,7 @@ def _initialize_trials_in_db(trials: List[models.Trial]):
     db_utils.bulk_save(trials)
 
 
-class Experiment:  # pylint: disable=too-many-instance-attributes
+class Experiment:
     """Class representing an experiment."""
 
     def __init__(self, experiment_config_filepath: str):
@@ -103,11 +103,9 @@ def __init__(self, experiment_config_filepath: str):
         self.preemptible = self.config.get('preemptible_runners')
 
 
-def build_images_for_trials(fuzzers: List[str],
-                            benchmarks: List[str],
+def build_images_for_trials(fuzzers: List[str], benchmarks: List[str],
                             num_trials: int,
-                            preemptible: bool,
-                            concurrent_builds=None) -> List[models.Trial]:
+                            preemptible: bool) -> List[models.Trial]:
     """Builds the images needed to run |experiment| and returns a list of trials
     that can be run for experiment. This is the number of trials specified in
     experiment times each pair of fuzzer+benchmark that builds successfully."""
@@ -116,14 +114,8 @@ def build_images_for_trials(fuzzers: List[str],
     builder.build_base_images()
 
     # Only build fuzzers for benchmarks whose measurers built successfully.
-    if concurrent_builds is None:
-        benchmarks = builder.build_all_measurers(benchmarks)
-        build_successes = builder.build_all_fuzzer_benchmarks(
-            fuzzers, benchmarks)
-    else:
-        benchmarks = builder.build_all_measurers(benchmarks, concurrent_builds)
-        build_successes = builder.build_all_fuzzer_benchmarks(
-            fuzzers, benchmarks, concurrent_builds)
+    benchmarks = builder.build_all_measurers(benchmarks)
+    build_successes = builder.build_all_fuzzer_benchmarks(fuzzers, benchmarks)
     experiment_name = experiment_utils.get_experiment_name()
     trials = []
     for fuzzer, benchmark in build_successes:
@@ -155,8 +147,7 @@ def dispatcher_main():
 
     trials = build_images_for_trials(experiment.fuzzers, experiment.benchmarks,
                                      experiment.num_trials,
-                                     experiment.preemptible,
-                                     experiment.config['concurrent_builds'])
+                                     experiment.preemptible)
     _initialize_trials_in_db(trials)
 
     create_work_subdirs(['experiment-folders', 'measurement-folders'])

experiment/resources/dispatcher-startup-script-template.sh

@@ -26,6 +26,8 @@ docker run --rm \
   -e POSTGRES_PASSWORD={{postgres_password}} \
   -e CLOUD_SQL_INSTANCE_CONNECTION_NAME={{cloud_sql_instance_connection_name}} \
   -e DOCKER_REGISTRY={{docker_registry}} \
+  -e CONCURRENT_BUILDS={{concurrent_builds}} \
+  -e WORKER_POOL_NAME={{worker_pool_name}} \
   --cap-add=SYS_PTRACE --cap-add=SYS_NICE \
   -v /var/run/docker.sock:/var/run/docker.sock --name=dispatcher-container \
   {{docker_registry}}/dispatcher-image /work/startup-dispatcher.sh