From f9c03263903b3bad74f703f3bcfd4770e6b1171d Mon Sep 17 00:00:00 2001 From: Lawrence Schlosser Date: Tue, 10 Apr 2018 15:21:08 -0700 Subject: [PATCH] CT-187 Fix for thread termination issues - Now properly terminates the following threads after uploading a job: ErrorThread MetricStore PrintStatusThread ReporterThread - Disclaimer: this code really stinks. Needs an entire re-write. --- conductor/lib/uploader.py | 64 +++++++++++++++-------------------- conductor/lib/worker.py | 70 +++++++++++++++++++++++++++++++-------- 2 files changed, 82 insertions(+), 52 deletions(-) diff --git a/conductor/lib/uploader.py b/conductor/lib/uploader.py index cc9b1044..75cd56e2 100644 --- a/conductor/lib/uploader.py +++ b/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,6 +149,7 @@ 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() @@ -156,7 +157,7 @@ def __init__(self, *args, **kwargs): 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,7 +467,6 @@ def upload_status_text(self): return formatted_text - def print_status(self): logger.debug('starting print_status thread') update_interval = 3 @@ -475,15 +474,13 @@ def print_status(self): 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,16 +493,15 @@ 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): @@ -513,8 +509,8 @@ def mark_upload_failed(self, error_message, upload_id): # 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 - - - - - - - - diff --git a/conductor/lib/worker.py b/conductor/lib/worker.py index d9db7f14..8dac70ca 100644 --- a/conductor/lib/worker.py +++ b/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")