[Predator] Run predator.

appengine/findit/util_scripts/script_util.py

Index: appengine/findit/util_scripts/script_util.py
diff --git a/appengine/findit/util_scripts/script_util.py b/appengine/findit/util_scripts/script_util.py
index 1d9846d437f3a7cb4731b5c728d05b11631c2e02..7d335123473298e972e84ebc895ecc341fa5212c 100644
--- a/appengine/findit/util_scripts/script_util.py
+++ b/appengine/findit/util_scripts/script_util.py
@@ -4,8 +4,19 @@
 
 """This module contains util functions that local scripts can use."""
 
+import atexit
+import cgi
+import json
 import os
+import Queue
+import re
 import sys
+import threading
+import traceback
+import time
+
+MAX_THREAD_NUMBER = 10
+TASK_QUEUE = None
 
 
 def SetUpSystemPaths():
@@ -25,3 +36,80 @@ def SetUpSystemPaths():
 
   # Add Findit root dir to sys.path so that modules in Findit is available.
   sys.path.insert(1, findit_root_dir)
+
+
+def SignalWorkerThreads():

[stgao, 2016/11/02 01:45:49]

Is this a copy from the version in ClusterFuzz? Or are there some significant
modification?

[Sharu Jiang, 2016/11/11 23:10:26]

It is a copy from the Clusterfuzz.

+  """Puts signal worker threads into task queue."""
+  global TASK_QUEUE  # pylint: disable=W0602
+  if not TASK_QUEUE:
+    return
+
+  for _ in range(MAX_THREAD_NUMBER):
+    TASK_QUEUE.put(None)
+
+  # Give worker threads a chance to exit.
+  # Workaround the harmless bug in python 2.7 below.
+  time.sleep(1)
+
+
+atexit.register(SignalWorkerThreads)
+
+
+def Worker():
+  global TASK_QUEUE  # pylint: disable=W0602
+  while True:
+    try:
+      task = TASK_QUEUE.get()
+      if not task:
+        return
+    except TypeError:
+      # According to http://bugs.python.org/issue14623, this is a harmless bug
+      # in python 2.7 which won't be fixed.
+      # The exception is raised on daemon threads when python interpreter is
+      # shutting down.
+      return
+
+    function, args, kwargs, result_semaphore = task
+    try:
+      function(*args, **kwargs)
+    except Exception:
+      print traceback.format_exc()
+      # Continue to process tasks in queue, in case every thread fails, the
+      # main thread will be waiting forever.
+      continue
+    finally:
+      # Signal one task is done in case of exception.
+      result_semaphore.release()
+
+
+def RunTasks(tasks):
+  """Run given tasks. Not thread-safe: no concurrent calls of this function.
+
+  Return after all tasks were completed. A task is a dict as below:
+    {
+      'function': the function to call,
+      'args': the positional argument to pass to the function,
+      'kwargs': the key-value arguments to pass to the function,
+    }
+  """
+  if not tasks:
+    return
+
+  global TASK_QUEUE
+  if not TASK_QUEUE:
+    TASK_QUEUE = Queue.Queue()
+    for index in range(MAX_THREAD_NUMBER):
+      thread = threading.Thread(target=Worker, name='worker_%s' % index)
+      # Set as daemon, so no join is needed.
+      thread.daemon = True
+      thread.start()
+
+  result_semaphore = threading.Semaphore(0)
+  # Push task to task queue for execution.
+  for task in tasks:
+    TASK_QUEUE.put((task['function'], task.get('args', []),
+                    task.get('kwargs', {}), result_semaphore))
+
+  # Wait until all tasks to be executed.
+  for _ in tasks:
+    result_semaphore.acquire()