def worker(worker_id, pool_id, task_queue, result_queue, worker_queue):
    """

    Put request token into queue
    Get task from task_queue
    Pop request from queue
    Put result into result_queue
    """
    start_file_logger('{}/{}/worker_{}.log'.format(args.logdir, pool_id, worker_id),
                      worker_id,
                      name="worker_log",
                      level=logging.DEBUG if args.debug else logging.INFO)

    # Sync worker with master
    logger.info('Worker {} started'.format(worker_id))
    if args.debug:
        logger.debug("Debug logging enabled")

    while True:
        worker_queue.put(worker_id)

        # The worker will receive {'task_id':<tid>, 'buffer':<buf>}
        req = task_queue.get()
        tid = req['task_id']
        logger.info("Received task {}".format(tid))

        try:
            worker_queue.get()
        except queue.Empty:
            logger.warning("Worker ID: {} failed to remove itself from ready_worker_queue".format(worker_id))
            pass

        try:
            result = execute_task(req['buffer'])
            serialized_result = serialize_object(result)
        except Exception:
            result_package = {'task_id': tid, 'exception': serialize_object(RemoteExceptionWrapper(*sys.exc_info()))}
        else:
            result_package = {'task_id': tid, 'result': serialized_result}
            # logger.debug("Result: {}".format(result))

        logger.info("Completed task {}".format(tid))
        pkl_package = pickle.dumps(result_package)

        result_queue.put(pkl_package)
def worker ( worker_id , pool_id , task_queue , result_queue , worker_queue ) : start_file_logger ( '{}/{}/worker_{}.log' . format ( args . logdir , pool_id , worker_id ) , worker_id , name = "worker_log" , level = logging . DEBUG if args . debug else logging . INFO ) logger . info ( 'Worker {} started' . format ( worker_id ) ) if args . debug : logger . debug ( "Debug logging enabled" ) while True : worker_queue . put ( worker_id ) req = task_queue . get ( ) tid = req [ 'task_id' ] logger . info ( "Received task {}" . format ( tid ) ) try : worker_queue . get ( ) except queue . Empty : logger . warning ( "Worker ID: {} failed to remove itself from ready_worker_queue" . format ( worker_id ) ) pass try : result = execute_task ( req [ 'buffer' ] ) serialized_result = serialize_object ( result ) except Exception : result_package = { 'task_id' : tid , 'exception' : serialize_object ( RemoteExceptionWrapper ( * sys . exc_info ( ) ) ) } else : result_package = { 'task_id' : tid , 'result' : serialized_result } logger . info ( "Completed task {}" . format ( tid ) ) pkl_package = pickle . dumps ( result_package ) result_queue . put ( pkl_package )
Put request token into queue
    Get task from task_queue
    Pop request from queue
    Put result into result_queue