def _synchronizer(self):
        """
        **Purpose**: Thread in the master process to keep the workflow data
        structure in appmanager up to date. We receive pipelines, stages and
        tasks objects directly. The respective object is updated in this master
        process.

        Details: Important to note that acknowledgements of the type
        channel.basic_ack() is an acknowledgement to the server that the msg
        was received. This is not to be confused with the Ack sent to the
        enqueuer/dequeuer/task_manager through the sync-ack queue.
        """

        try:

            self._prof.prof('synchronizer started', uid=self._uid)

            self._logger.info('synchronizer thread started')

            def task_update(msg, reply_to, corr_id, mq_channel):

                completed_task = Task()
                completed_task.from_dict(msg['object'])
                self._logger.info('Received %s with state %s' % (completed_task.uid, completed_task.state))

                found_task = False

                # Traverse the entire workflow to find the correct task
                for pipe in self._workflow:

                    if not pipe.completed:
                        if completed_task.parent_pipeline['uid'] == pipe.uid:

                            for stage in pipe.stages:

                                if completed_task.parent_stage['uid'] == stage.uid:

                                    for task in stage.tasks:

                                        if (completed_task.uid == task.uid)and(completed_task.state != task.state):

                                            task.state = str(completed_task.state)
                                            self._logger.debug('Found task %s with state %s' %
                                                               (task.uid, task.state))

                                            if completed_task.path:
                                                task.path = str(completed_task.path)

                                            mq_channel.basic_publish(exchange='',
                                                                     routing_key=reply_to,
                                                                     properties=pika.BasicProperties(
                                                                         correlation_id=corr_id),
                                                                     body='%s-ack' % task.uid)

                                            self._prof.prof('publishing sync ack for obj with state %s' %
                                                            msg['object']['state'],
                                                            uid=msg['object']['uid']
                                                            )

                                            mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)
                                            self._report.ok('Update: ')
                                            self._report.info('%s state: %s\n' % (task.luid, task.state))

                                            found_task = True

                                    if not found_task:

                                        # If there was a Stage update, but the Stage was not found in any of the Pipelines. This
                                        # means that this was a Stage that was added during runtime and the AppManager does not
                                        # know about it. The current solution is going to be: add it to the workflow object in the
                                        # AppManager via the synchronizer.

                                        self._prof.prof('Adap: adding new task')

                                        self._logger.info('Adding new task %s to parent stage: %s' % (completed_task.uid,
                                                                                                      stage.uid))

                                        stage.add_tasks(completed_task)
                                        mq_channel.basic_publish(exchange='',
                                                                 routing_key=reply_to,
                                                                 properties=pika.BasicProperties(
                                                                     correlation_id=corr_id),
                                                                 body='%s-ack' % completed_task.uid)

                                        self._prof.prof('Adap: added new task')

                                        self._prof.prof('publishing sync ack for obj with state %s' %
                                                        msg['object']['state'],
                                                        uid=msg['object']['uid']
                                                        )

                                        mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)
                                        self._report.ok('Update: ')
                                        self._report.info('%s state: %s\n' %
                                                          (completed_task.luid, completed_task.state))

            def stage_update(msg, reply_to, corr_id, mq_channel):

                completed_stage = Stage()
                completed_stage.from_dict(msg['object'])
                self._logger.info('Received %s with state %s' % (completed_stage.uid, completed_stage.state))

                found_stage = False

                # Traverse the entire workflow to find the correct stage
                for pipe in self._workflow:

                    if not pipe.completed:

                        if completed_stage.parent_pipeline['uid'] == pipe.uid:
                            self._logger.info('Found parent pipeline: %s' % pipe.uid)

                            for stage in pipe.stages:

                                if (completed_stage.uid == stage.uid)and(completed_stage.state != stage.state):

                                    self._logger.debug('Found stage %s' % stage.uid)

                                    stage.state = str(completed_stage.state)

                                    mq_channel.basic_publish(exchange='',
                                                             routing_key=reply_to,
                                                             properties=pika.BasicProperties(
                                                                 correlation_id=corr_id),
                                                             body='%s-ack' % stage.uid)

                                    self._prof.prof('publishing sync ack for obj with state %s' %
                                                    msg['object']['state'],
                                                    uid=msg['object']['uid']
                                                    )

                                    mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)
                                    self._report.ok('Update: ')
                                    self._report.info('%s state: %s\n' % (stage.luid, stage.state))

                                    found_stage = True

                            if not found_stage:

                                # If there was a Stage update, but the Stage was not found in any of the Pipelines. This
                                # means that this was a Stage that was added during runtime and the AppManager does not
                                # know about it. The current solution is going to be: add it to the workflow object in the
                                # AppManager via the synchronizer.

                                self._prof.prof('Adap: adding new stage', uid=self._uid)

                                self._logger.info('Adding new stage %s to parent pipeline: %s' % (completed_stage.uid,
                                                                                                  pipe.uid))

                                pipe.add_stages(completed_stage)
                                mq_channel.basic_publish(exchange='',
                                                         routing_key=reply_to,
                                                         properties=pika.BasicProperties(
                                                             correlation_id=corr_id),
                                                         body='%s-ack' % completed_stage.uid)

                                self._prof.prof('Adap: adding new stage', uid=self._uid)

                                self._prof.prof('publishing sync ack for obj with state %s' %
                                                msg['object']['state'],
                                                uid=msg['object']['uid']
                                                )

                                mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)

            def pipeline_update(msg, reply_to, corr_id, mq_channel):

                completed_pipeline = Pipeline()
                completed_pipeline.from_dict(msg['object'])

                self._logger.info('Received %s with state %s' % (completed_pipeline.uid, completed_pipeline.state))

                # Traverse the entire workflow to find the correct pipeline
                for pipe in self._workflow:

                    if not pipe.completed:

                        if (completed_pipeline.uid == pipe.uid)and(completed_pipeline.state != pipe.state):

                            pipe.state = str(completed_pipeline.state)

                            self._logger.info('Found pipeline %s, state %s, completed %s' % (pipe.uid,
                                                                                             pipe.state,
                                                                                             pipe.completed)
                                              )

                            # Reply with ack msg to the sender
                            mq_channel.basic_publish(exchange='',
                                                     routing_key=reply_to,
                                                     properties=pika.BasicProperties(
                                                         correlation_id=corr_id),
                                                     body='%s-ack' % pipe.uid)

                            self._prof.prof('publishing sync ack for obj with state %s' %
                                            msg['object']['state'],
                                            uid=msg['object']['uid']
                                            )

                            mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                            # Keep the assignment of the completed flag after sending the acknowledgment
                            # back. Otherwise the MainThread takes lock over the pipeline because of logging
                            # and profiling
                            if completed_pipeline.completed:
                                pipe._completed_flag.set()
                            self._report.ok('Update: ')
                            self._report.info('%s state: %s\n' % (pipe.luid, pipe.state))

            mq_connection = pika.BlockingConnection(pika.ConnectionParameters(host=self._mq_hostname, port=self._port))
            mq_channel = mq_connection.channel()

            last = time.time()

            while not self._terminate_sync.is_set():

                #-------------------------------------------------------------------------------------------------------
                # Messages between tmgr Main thread and synchronizer -- only Task objects

                method_frame, props, body = mq_channel.basic_get(queue='%s-tmgr-to-sync' % self._sid)

                """
                The message received is a JSON object with the following structure:

                msg = {
                        'type': 'Pipeline'/'Stage'/'Task',
                        'object': json/dict
                        }
                """

                if body:

                    msg = json.loads(body)

                    self._prof.prof('received obj with state %s for sync' %
                                    msg['object']['state'], uid=msg['object']['uid'])

                    self._logger.debug('received %s with state %s for sync' %
                                       (msg['object']['uid'], msg['object']['state']))

                    if msg['type'] == 'Task':
                        task_update(msg, '%s-sync-to-tmgr' % self._sid, props.correlation_id, mq_channel)

                #-------------------------------------------------------------------------------------------------------

                #-------------------------------------------------------------------------------------------------------
                # Messages between callback thread and synchronizer -- only Task objects

                method_frame, props, body = mq_channel.basic_get(queue='%s-cb-to-sync' % self._sid)

                """
                The message received is a JSON object with the following structure:

                msg = {
                        'type': 'Pipeline'/'Stage'/'Task',
                        'object': json/dict
                        }
                """

                if body:

                    msg = json.loads(body)

                    self._prof.prof('received obj with state %s for sync' %
                                    msg['object']['state'], uid=msg['object']['uid'])

                    self._logger.debug('received %s with state %s for sync' %
                                       (msg['object']['uid'], msg['object']['state']))

                    if msg['type'] == 'Task':
                        task_update(msg, '%s-sync-to-cb' % self._sid, props.correlation_id, mq_channel)

                #-------------------------------------------------------------------------------------------------------

                #-------------------------------------------------------------------------------------------------------
                # Messages between enqueue thread and synchronizer -- Task, Stage or Pipeline
                method_frame, props, body = mq_channel.basic_get(queue='%s-enq-to-sync' % self._sid)

                if body:

                    msg = json.loads(body)

                    self._prof.prof('received obj with state %s for sync' %
                                    msg['object']['state'], uid=msg['object']['uid'])

                    self._logger.debug('received %s with state %s for sync' %
                                       (msg['object']['uid'], msg['object']['state']))

                    if msg['type'] == 'Task':
                        task_update(msg, '%s-sync-to-enq' % self._sid, props.correlation_id, mq_channel)

                    elif msg['type'] == 'Stage':
                        stage_update(msg, '%s-sync-to-enq' % self._sid, props.correlation_id, mq_channel)

                    elif msg['type'] == 'Pipeline':
                        pipeline_update(msg, '%s-sync-to-enq' % self._sid, props.correlation_id, mq_channel)
                #-------------------------------------------------------------------------------------------------------

                #-------------------------------------------------------------------------------------------------------
                # Messages between dequeue thread and synchronizer -- Task, Stage or Pipeline
                method_frame, props, body = mq_channel.basic_get(queue='%s-deq-to-sync' % self._sid)

                if body:

                    msg = json.loads(body)

                    self._prof.prof('received obj with state %s for sync' %
                                    msg['object']['state'], uid=msg['object']['uid'])

                    self._logger.debug('received %s with state %s for sync' %
                                       (msg['object']['uid'], msg['object']['state']))

                    if msg['type'] == 'Task':
                        task_update(msg, '%s-sync-to-deq' % self._sid, props.correlation_id, mq_channel)

                    elif msg['type'] == 'Stage':
                        stage_update(msg, '%s-sync-to-deq' % self._sid, props.correlation_id, mq_channel)

                    elif msg['type'] == 'Pipeline':
                        pipeline_update(msg, '%s-sync-to-deq' % self._sid, props.correlation_id, mq_channel)
                #-------------------------------------------------------------------------------------------------------

                # Appease pika cos it thinks the connection is dead
                now = time.time()
                if now - last >= self._rmq_ping_interval:
                    mq_connection.process_data_events()
                    last = now

            self._prof.prof('terminating synchronizer', uid=self._uid)

        except KeyboardInterrupt:

            self._logger.exception('Execution interrupted by user (you probably hit Ctrl+C), ' +
                               'trying to terminate synchronizer thread gracefully...')

            raise KeyboardInterrupt

        except Exception, ex:

            self._logger.exception('Unknown error in synchronizer: %s. \n Terminating thread' % ex)
            raise
def _synchronizer ( self ) : try : self . _prof . prof ( 'synchronizer started' , uid = self . _uid ) self . _logger . info ( 'synchronizer thread started' ) def task_update ( msg , reply_to , corr_id , mq_channel ) : completed_task = Task ( ) completed_task . from_dict ( msg [ 'object' ] ) self . _logger . info ( 'Received %s with state %s' % ( completed_task . uid , completed_task . state ) ) found_task = False for pipe in self . _workflow : if not pipe . completed : if completed_task . parent_pipeline [ 'uid' ] == pipe . uid : for stage in pipe . stages : if completed_task . parent_stage [ 'uid' ] == stage . uid : for task in stage . tasks : if ( completed_task . uid == task . uid ) and ( completed_task . state != task . state ) : task . state = str ( completed_task . state ) self . _logger . debug ( 'Found task %s with state %s' % ( task . uid , task . state ) ) if completed_task . path : task . path = str ( completed_task . path ) mq_channel . basic_publish ( exchange = '' , routing_key = reply_to , properties = pika . BasicProperties ( correlation_id = corr_id ) , body = '%s-ack' % task . uid ) self . _prof . prof ( 'publishing sync ack for obj with state %s' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) mq_channel . basic_ack ( delivery_tag = method_frame . delivery_tag ) self . _report . ok ( 'Update: ' ) self . _report . info ( '%s state: %s\n' % ( task . luid , task . state ) ) found_task = True if not found_task : self . _prof . prof ( 'Adap: adding new task' ) self . _logger . info ( 'Adding new task %s to parent stage: %s' % ( completed_task . uid , stage . uid ) ) stage . add_tasks ( completed_task ) mq_channel . basic_publish ( exchange = '' , routing_key = reply_to , properties = pika . BasicProperties ( correlation_id = corr_id ) , body = '%s-ack' % completed_task . uid ) self . _prof . prof ( 'Adap: added new task' ) self . _prof . prof ( 'publishing sync ack for obj with state %s' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) mq_channel . basic_ack ( delivery_tag = method_frame . delivery_tag ) self . _report . ok ( 'Update: ' ) self . _report . info ( '%s state: %s\n' % ( completed_task . luid , completed_task . state ) ) def stage_update ( msg , reply_to , corr_id , mq_channel ) : completed_stage = Stage ( ) completed_stage . from_dict ( msg [ 'object' ] ) self . _logger . info ( 'Received %s with state %s' % ( completed_stage . uid , completed_stage . state ) ) found_stage = False for pipe in self . _workflow : if not pipe . completed : if completed_stage . parent_pipeline [ 'uid' ] == pipe . uid : self . _logger . info ( 'Found parent pipeline: %s' % pipe . uid ) for stage in pipe . stages : if ( completed_stage . uid == stage . uid ) and ( completed_stage . state != stage . state ) : self . _logger . debug ( 'Found stage %s' % stage . uid ) stage . state = str ( completed_stage . state ) mq_channel . basic_publish ( exchange = '' , routing_key = reply_to , properties = pika . BasicProperties ( correlation_id = corr_id ) , body = '%s-ack' % stage . uid ) self . _prof . prof ( 'publishing sync ack for obj with state %s' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) mq_channel . basic_ack ( delivery_tag = method_frame . delivery_tag ) self . _report . ok ( 'Update: ' ) self . _report . info ( '%s state: %s\n' % ( stage . luid , stage . state ) ) found_stage = True if not found_stage : self . _prof . prof ( 'Adap: adding new stage' , uid = self . _uid ) self . _logger . info ( 'Adding new stage %s to parent pipeline: %s' % ( completed_stage . uid , pipe . uid ) ) pipe . add_stages ( completed_stage ) mq_channel . basic_publish ( exchange = '' , routing_key = reply_to , properties = pika . BasicProperties ( correlation_id = corr_id ) , body = '%s-ack' % completed_stage . uid ) self . _prof . prof ( 'Adap: adding new stage' , uid = self . _uid ) self . _prof . prof ( 'publishing sync ack for obj with state %s' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) mq_channel . basic_ack ( delivery_tag = method_frame . delivery_tag ) def pipeline_update ( msg , reply_to , corr_id , mq_channel ) : completed_pipeline = Pipeline ( ) completed_pipeline . from_dict ( msg [ 'object' ] ) self . _logger . info ( 'Received %s with state %s' % ( completed_pipeline . uid , completed_pipeline . state ) ) for pipe in self . _workflow : if not pipe . completed : if ( completed_pipeline . uid == pipe . uid ) and ( completed_pipeline . state != pipe . state ) : pipe . state = str ( completed_pipeline . state ) self . _logger . info ( 'Found pipeline %s, state %s, completed %s' % ( pipe . uid , pipe . state , pipe . completed ) ) mq_channel . basic_publish ( exchange = '' , routing_key = reply_to , properties = pika . BasicProperties ( correlation_id = corr_id ) , body = '%s-ack' % pipe . uid ) self . _prof . prof ( 'publishing sync ack for obj with state %s' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) mq_channel . basic_ack ( delivery_tag = method_frame . delivery_tag ) if completed_pipeline . completed : pipe . _completed_flag . set ( ) self . _report . ok ( 'Update: ' ) self . _report . info ( '%s state: %s\n' % ( pipe . luid , pipe . state ) ) mq_connection = pika . BlockingConnection ( pika . ConnectionParameters ( host = self . _mq_hostname , port = self . _port ) ) mq_channel = mq_connection . channel ( ) last = time . time ( ) while not self . _terminate_sync . is_set ( ) : method_frame , props , body = mq_channel . basic_get ( queue = '%s-tmgr-to-sync' % self . _sid ) """
                The message received is a JSON object with the following structure:
                msg = {
                        'type': 'Pipeline'/'Stage'/'Task',
                        'object': json/dict
                        }
                """ if body : msg = json . loads ( body ) self . _prof . prof ( 'received obj with state %s for sync' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) self . _logger . debug ( 'received %s with state %s for sync' % ( msg [ 'object' ] [ 'uid' ] , msg [ 'object' ] [ 'state' ] ) ) if msg [ 'type' ] == 'Task' : task_update ( msg , '%s-sync-to-tmgr' % self . _sid , props . correlation_id , mq_channel ) method_frame , props , body = mq_channel . basic_get ( queue = '%s-cb-to-sync' % self . _sid ) """
                The message received is a JSON object with the following structure:
                msg = {
                        'type': 'Pipeline'/'Stage'/'Task',
                        'object': json/dict
                        }
                """ if body : msg = json . loads ( body ) self . _prof . prof ( 'received obj with state %s for sync' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) self . _logger . debug ( 'received %s with state %s for sync' % ( msg [ 'object' ] [ 'uid' ] , msg [ 'object' ] [ 'state' ] ) ) if msg [ 'type' ] == 'Task' : task_update ( msg , '%s-sync-to-cb' % self . _sid , props . correlation_id , mq_channel ) method_frame , props , body = mq_channel . basic_get ( queue = '%s-enq-to-sync' % self . _sid ) if body : msg = json . loads ( body ) self . _prof . prof ( 'received obj with state %s for sync' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) self . _logger . debug ( 'received %s with state %s for sync' % ( msg [ 'object' ] [ 'uid' ] , msg [ 'object' ] [ 'state' ] ) ) if msg [ 'type' ] == 'Task' : task_update ( msg , '%s-sync-to-enq' % self . _sid , props . correlation_id , mq_channel ) elif msg [ 'type' ] == 'Stage' : stage_update ( msg , '%s-sync-to-enq' % self . _sid , props . correlation_id , mq_channel ) elif msg [ 'type' ] == 'Pipeline' : pipeline_update ( msg , '%s-sync-to-enq' % self . _sid , props . correlation_id , mq_channel ) method_frame , props , body = mq_channel . basic_get ( queue = '%s-deq-to-sync' % self . _sid ) if body : msg = json . loads ( body ) self . _prof . prof ( 'received obj with state %s for sync' % msg [ 'object' ] [ 'state' ] , uid = msg [ 'object' ] [ 'uid' ] ) self . _logger . debug ( 'received %s with state %s for sync' % ( msg [ 'object' ] [ 'uid' ] , msg [ 'object' ] [ 'state' ] ) ) if msg [ 'type' ] == 'Task' : task_update ( msg , '%s-sync-to-deq' % self . _sid , props . correlation_id , mq_channel ) elif msg [ 'type' ] == 'Stage' : stage_update ( msg , '%s-sync-to-deq' % self . _sid , props . correlation_id , mq_channel ) elif msg [ 'type' ] == 'Pipeline' : pipeline_update ( msg , '%s-sync-to-deq' % self . _sid , props . correlation_id , mq_channel ) now = time . time ( ) if now - last >= self . _rmq_ping_interval : mq_connection . process_data_events ( ) last = now self . _prof . prof ( 'terminating synchronizer' , uid = self . _uid ) except KeyboardInterrupt : self . _logger . exception ( 'Execution interrupted by user (you probably hit Ctrl+C), ' + 'trying to terminate synchronizer thread gracefully...' ) raise KeyboardInterrupt except Exception , ex : self . _logger . exception ( 'Unknown error in synchronizer: %s. \n Terminating thread' % ex ) raise
**Purpose**: Thread in the master process to keep the workflow data
        structure in appmanager up to date. We receive pipelines, stages and
        tasks objects directly. The respective object is updated in this master
        process.

        Details: Important to note that acknowledgements of the type
        channel.basic_ack() is an acknowledgement to the server that the msg
        was received. This is not to be confused with the Ack sent to the
        enqueuer/dequeuer/task_manager through the sync-ack queue.