CT-187 Fix for thread termination issues

conductor/lib/uploader.py

@@ -33,6 +33,7 @@ def do_work(self, job, thread_int):
         logger.debug('job is %s', job)
         filename, submission_time_md5 = job
         assert isinstance(filename, (str, unicode)), "Filepath not of expected type. Got %s" % type(filename)
+
         filename = str(filename)
         current_md5 = self.get_md5(filename)
         # if a submission time md5 was provided then check against it
@@ -74,7 +75,6 @@ def get_md5(self, filepath):
         logger.debug("Using md5 cache for file: %s", filepath)
         return file_cache["md5"]
 
-
     def cache_file_info(self, file_info):
         '''
         Store the given file_info into the database
@@ -84,12 +84,12 @@ def cache_file_info(self, file_info):
                                    thread_safe=True)
 
 
-
 class MD5OutputWorker(worker.ThreadWorker):
     '''
     This worker will batch the computed md5's into self.batch_size chunks.
     It will send a partial batch after waiting self.wait_time seconds
     '''
+
     def __init__(self, *args, **kwargs):
         worker.ThreadWorker.__init__(self, *args, **kwargs)
         self.batch_size = 20  # the controlls the batch size for http get_signed_urls
@@ -149,14 +149,15 @@ class HttpBatchWorker(worker.ThreadWorker):
 
     Each item in the return list is added to the out_queue.
     '''
+
     def __init__(self, *args, **kwargs):
         worker.ThreadWorker.__init__(self, *args, **kwargs)
         self.api_client = api_client.ApiClient()
         self.project = kwargs.get('project')
 
     def make_request(self, job):
         uri_path = '/api/files/get_upload_urls'
-        headers = {'Content-Type':'application/json'}
+        headers = {'Content-Type': 'application/json'}
         data = {"upload_files": job,
                 "project": self.project}
 
@@ -191,7 +192,10 @@ def do_work(self, job, thread_int):
 to be uploaded. Note: This is stored as an [int] in order to pass it by
 reference, as it needs to be accessed and reset by the caller.
 '''
+
+
 class FileStatWorker(worker.ThreadWorker):
+
     def __init__(self, *args, **kwargs):
         worker.ThreadWorker.__init__(self, *args, **kwargs)
 
@@ -224,6 +228,7 @@ class UploadWorker(worker.ThreadWorker):
     This worker receives a (filepath: signed_upload_url) pair and performs an upload
     of the specified file to the provided url.
     '''
+
     def __init__(self, *args, **kwargs):
         worker.ThreadWorker.__init__(self, *args, **kwargs)
         self.chunk_size = 1048576  # 1M
@@ -257,9 +262,6 @@ def do_work(self, job, thread_int):
             logger.error(error_message)
             raise
 
-
-
-
     @common.DecRetry(retry_exceptions=api_client.CONNECTION_EXCEPTIONS, tries=5)
     def do_upload(self, upload_url, filename, md5):
         '''
@@ -327,7 +329,7 @@ def create_manager(self, project, md5_only=False):
     def report_status(self):
         logger.debug('started report_status thread')
         update_interval = 5
-        while True:
+        while self.working:
 
             # don't report status if we are doing a local_upload
             if not self.upload_id:
@@ -401,13 +403,12 @@ def convert_byte_count_to_string(byte_count, transfer_rate=False):
     @staticmethod
     def convert_time_to_string(time_remaining):
         if time_remaining > 3600:
-            return str(round(time_remaining / float(3600) , 1)) + ' hours'
+            return str(round(time_remaining / float(3600), 1)) + ' hours'
         elif time_remaining > 60:
-            return str(round(time_remaining / float(60) , 1)) + ' minutes'
+            return str(round(time_remaining / float(60), 1)) + ' minutes'
         else:
             return str(round(time_remaining, 1)) + ' seconds'
 
-
     def upload_status_text(self):
         num_files_to_upload = self.manager.metric_store.get('num_files_to_upload')
         files_to_upload = str(num_files_to_upload)
@@ -430,7 +431,6 @@ def upload_status_text(self):
         else:
             transfer_rate = 0
 
-
         unformatted_text = '''
 ################################################################################
      files to process: {files_to_analyze}
@@ -467,23 +467,20 @@ def upload_status_text(self):
 
         return formatted_text
 
-
     def print_status(self):
         logger.debug('starting print_status thread')
         update_interval = 3
 
         def sleep():
             time.sleep(update_interval)
 
-        while True:
-            if self.working:
-                try:
-                    logger.info(self.manager.worker_queue_status_text())
-                    logger.info(self.upload_status_text())
-                except Exception, e:
-                    print e
-                    print traceback.format_exc()
-                    # pass
+        while self.working:
+            try:
+                logger.info(self.manager.worker_queue_status_text())
+                logger.info(self.upload_status_text())
+            except Exception, e:
+                print e
+                print traceback.format_exc()
             sleep()
 
     def create_print_status_thread(self):
@@ -496,25 +493,24 @@ def create_print_status_thread(self):
         # start thread
         thd.start()
 
-
     def mark_upload_finished(self, upload_id, upload_files):
 
-        data = {'upload_id':upload_id,
+        data = {'upload_id': upload_id,
                 'status': 'server_pending',
                 'upload_files': upload_files}
 
         self.api_client.make_request('/uploads/%s/finish' % upload_id,
-                                                           data=json.dumps(data),
-                                                           verb='POST', use_api_key=True)
+                                     data=json.dumps(data),
+                                     verb='POST', use_api_key=True)
         return True
 
     def mark_upload_failed(self, error_message, upload_id):
         logger.error('failing upload due to: \n%s' % error_message)
 
         # report error_message to the app
         self.api_client.make_request('/uploads/%s/fail' % upload_id,
-                                                        data=error_message,
-                                                        verb='POST', use_api_key=True)
+                                     data=error_message,
+                                     verb='POST', use_api_key=True)
 
         return True
 
@@ -536,6 +532,7 @@ def handle_upload_response(self, project, upload_files, upload_id=None, md5_only
             logger.info('upload_files %s:(truncated)\n\t%s',
                         len(upload_files), "\n\t".join(upload_files.keys()[:5]))
 
+
             # reset counters
             self.num_files_to_process = len(upload_files)
             self.job_start_time = int(time.time())
@@ -583,7 +580,6 @@ def handle_upload_response(self, project, upload_files, upload_id=None, md5_only
         except:
             return traceback.format_exc()
 
-
     def main(self, run_one_loop=False):
         logger.info('Uploader Started. Checking for uploads...')
 
@@ -637,7 +633,6 @@ def main(self, run_one_loop=False):
 
         logger.info('exiting uploader')
 
-
     def return_md5s(self):
         '''
         Return a dictionary of the filepaths and their md5s that were generated
@@ -655,6 +650,7 @@ def set_logging(level=None, log_dirpath=None):
                                      file_formatter=LOG_FORMATTER,
                                      log_filepath=log_filepath)
 
+
 def run_uploader(args):
     '''
     Start the uploader process. This process will run indefinitely, polling
@@ -691,6 +687,7 @@ def get_file_info(filepath):
             "modtime": modtime,
             "size": stat.st_size}
 
+
 def resolve_args(args):
     '''
     Resolve all arguments, reconsiling differences between command line args
@@ -703,7 +700,6 @@ def resolve_args(args):
     return args
 
 
-
 def resolve_arg(arg_name, args, config):
     '''
     Helper function to resolve the value of an argument.
@@ -797,11 +793,3 @@ def resolve_arg(arg_name, args, config):
 #
 #     logger.debug("Complete")
 #     return md5s
-
-
-
-
-
-
-
-

conductor/lib/worker.py

@@ -14,7 +14,9 @@
 '''
 WORKING = True
 
+
 class Reporter():
+
     def __init__(self, metric_store=None):
         self.metric_store = metric_store
         self.api_helper = api_client.ApiClient()
@@ -48,8 +50,6 @@ def start(self):
         return self.thread
 
 
-
-
 class ThreadWorker(object):
     '''
     Abstract worker class.
@@ -79,9 +79,6 @@ def __init__(self, **kwargs):
         # create a list to hold the threads that we create
         self.threads = []
 
-
-
-
     def do_work(self, job):
         '''
         This needs to be implemented for each worker type. The work task from
@@ -214,6 +211,7 @@ def put_job(self, job):
         self.out_queue.put(job)
         return True
 
+
 class MetricStore():
     '''
     This provides a thread-safe integer store that can be used by workers to
@@ -226,8 +224,10 @@ def __init__(self):
         self.metric_store = {}
         self.update_queue = Queue.Queue()
         self.started = False
+        self.terminate = False
 
     def join(self):
+        empty_queue(self.update_queue)
         self.update_queue.join()
         return True
 
@@ -239,11 +239,11 @@ def start(self):
             logger.debug('metric_store already started')
             return None
         logger.debug('starting metric_store')
-        thd = threading.Thread(target=self.target, name=self.__class__.__name__)
-        thd.daemon = True
-        thd.start()
+        self.thread = threading.Thread(target=self.target, name=self.__class__.__name__)
+        self.thread.daemon = True
+        self.thread.start()
         self.started = True
-        return thd
+        return self.thread
 
     def set(self, key, value):
         self.metric_store[key] = value
@@ -312,9 +312,11 @@ def get_list(self, list_name):
 
     def target(self):
         logger.debug('created metric_store target thread')
-        while True:
-            # block until update given
-            update_tuple = self.update_queue.get(True)
+        while not self.terminate:
+
+            update_tuple = safe_get(self.update_queue)
+            if not update_tuple:
+                continue
 
             method = update_tuple[0]
             method_args = update_tuple[1:]
@@ -331,6 +333,9 @@ def target(self):
             # mark task done
             self.update_queue.task_done()
 
+    def kill(self):
+        self.terminate = True
+        self.thread.join()
 
 
 class JobManager():
@@ -379,18 +384,27 @@ def kill_reporters(self):
             logger.debug('killing reporter %s', reporter)
             reporter.kill()
 
+    def kill_metricstore(self):
+        logger.debug('killing metric store %s', self.metric_store)
+        self.metric_store.kill()
+
+
     def stop_work(self):
         global WORKING
         WORKING = False  # stop any new jobs from being created
         self.drain_queues()  # clear out any jobs in queue
         self.kill_workers()  # kill all threads
+        self.kill_metricstore()
         self.kill_reporters()
         self.mark_all_tasks_complete()  # reset task counts
 
     def error_handler_target(self):
+        global WORKING
 
-        while True:
-            error = self.error_queue.get(True)
+        while  WORKING:
+            error = safe_get(self.error_queue)
+            if not error:
+                continue
             logger.error('got something from the error queue')
             self.error.append(error)
             self.stop_work()
@@ -482,6 +496,7 @@ def join(self):
         if self.error:
             return self.error
         self.kill_workers()
+        self.kill_metricstore()
         self.kill_reporters()
         return None
 
@@ -496,3 +511,30 @@ def worker_queue_status_text(self):
             msg += '\t\t%s threads' % num_active_threads
             msg += '\n'
         return msg
+
+
+def empty_queue(queue):
+    '''
+    Remove and return all items from the given Queue object
+    '''
+    items = []
+
+    while True:
+        item = safe_get(queue)
+        if not item:
+            break
+        items.append(item)
+    return items
+
+
+def safe_get(queue):
+    '''
+    Get and return an item from the given queue.
+    If the queue is empty, reurn None (supressing the exception).
+    '''
+    try:
+        return (queue.get_nowait())
+    except Queue.Empty:
+        return
+    except Exception:
+        logger.exception("recovered from exception:\n%s")