Bring back parallel signing for tworkers. (#4447)

It's important for speed, probably safe on tworkers, and it will be very
easy to notice if things go wrong since the preprocess queue will pile
up.
This PR also fixes a bad bug where the max_pool_size was not obeyed, but
in practice since this code never runs on machines with more than 2
cores, it's unlikely to matter.
Partial undoes #4430

Author: jonathanmetzman

src/clusterfuzz/_internal/base/concurrency.py

@@ -36,7 +36,7 @@ def map(self, f, l):
 def make_pool(pool_size=POOL_SIZE, cpu_bound=False, max_pool_size=None):
   """Returns a pool that can (usually) execute tasks concurrently."""
   if max_pool_size is not None:
-    pool_size = max(pool_size, max_pool_size)
+    pool_size = min(pool_size, max_pool_size)
 
   # Don't use processes on Windows and unittests to avoid hangs.
   if (environment.get_value('PY_UNITTESTS') or

src/clusterfuzz/_internal/google_cloud_utils/storage.py

@@ -1369,7 +1369,7 @@ def sign_urls_for_existing_files(urls,
                                  include_delete_urls) -> List[Tuple[str, str]]:
   logs.info('Signing URLs for existing files.')
   args = ((url, include_delete_urls) for url in urls)
-  result = list(map(_sign_urls_for_existing_file, args))
+  result = maybe_parallel_map(_sign_urls_for_existing_file, args)
   logs.info('Done signing URLs for existing files.')
   return result
 
@@ -1379,6 +1379,19 @@ def get_arbitrary_signed_upload_url(remote_directory):
       get_arbitrary_signed_upload_urls(remote_directory, num_uploads=1))[0]
 
 
+def maybe_parallel_map(func, arguments):
+  """Wrapper around pool.map so we don't do it on OSS-Fuzz hosts which
+  will OOM."""
+  if not environment.is_tworker():
+    # TODO(b/metzman): When the rearch is done, internal google CF won't have
+    # tworkers, but maybe should be using parallel.
+    return list(map(func, arguments))
+
+  max_size = 2
+  with concurrency.make_pool(cpu_bound=True, max_pool_size=max_size) as pool:
+    return list(pool.map(func, arguments))
+
+
 def get_arbitrary_signed_upload_urls(remote_directory: str,
                                      num_uploads: int) -> List[str]:
   """Returns |num_uploads| number of signed upload URLs to upload files with
@@ -1404,6 +1417,6 @@ def get_arbitrary_signed_upload_urls(remote_directory: str,
 
   urls = (f'{base_path}-{idx}' for idx in range(num_uploads))
   logs.info('Signing URLs for arbitrary uploads.')
-  result = list(map(get_signed_upload_url, urls))
+  result = maybe_parallel_map(get_signed_upload_url, urls)
   logs.info('Done signing URLs for arbitrary uploads.')
   return result