def _process_tasks(self, task_queue, rmgr, logger, mq_hostname, port, local_prof, sid):


        '''
        **Purpose**: The new thread that gets spawned by the main tmgr process invokes this function. This
        function receives tasks from 'task_queue' and submits them to the RADICAL Pilot RTS.
        '''

        placeholder_dict = dict()

        def load_placeholder(task, rts_uid):

            parent_pipeline = str(task.parent_pipeline['name'])
            parent_stage = str(task.parent_stage['name'])

            if parent_pipeline not in placeholder_dict:
                placeholder_dict[parent_pipeline] = dict()

            if parent_stage not in placeholder_dict[parent_pipeline]:
                placeholder_dict[parent_pipeline][parent_stage] = dict()

            if None not in [parent_pipeline, parent_stage, task.name]:
                placeholder_dict[parent_pipeline][parent_stage][str(task.name)] = {'path': str(task.path),
                                                                                   'rts_uid': rts_uid}

        def unit_state_cb(unit, state):

            try:

                logger.debug('Unit %s in state %s' % (unit.uid, unit.state))

                if unit.state in rp.FINAL:

                    # Acquire a connection+channel to the rmq server
                    mq_connection = pika.BlockingConnection(
                        pika.ConnectionParameters(host=mq_hostname, port=port))
                    mq_channel = mq_connection.channel()

                    task = None
                    task = create_task_from_cu(unit, local_prof)

                    transition(obj=task,
                               obj_type='Task',
                               new_state=states.COMPLETED,
                               channel=mq_channel,
                               queue='%s-cb-to-sync' % sid,
                               profiler=local_prof,
                               logger=logger)

                    load_placeholder(task, unit.uid)

                    task_as_dict = json.dumps(task.to_dict())

                    mq_channel.basic_publish(exchange='',
                                             routing_key='%s-completedq-1' % sid,
                                             body=task_as_dict
                                             # properties=pika.BasicProperties(
                                             # make message persistent
                                             #    delivery_mode = 2,
                                             # )
                                             )

                    logger.info('Pushed task %s with state %s to completed queue %s-completedq-1' % (task.uid, task.state,
                                                                                                     sid))

                    mq_connection.close()

            except KeyboardInterrupt:
                logger.exception('Execution interrupted by user (you probably hit Ctrl+C), ' +
                                 'trying to exit callback thread gracefully...')

                raise KeyboardInterrupt

            except Exception, ex:
                logger.exception('Error in RP callback thread: %s' % ex)

        umgr = rp.UnitManager(session=rmgr._session)
        umgr.add_pilots(rmgr.pilot)
        umgr.register_callback(unit_state_cb)

        try:

            while not self._tmgr_terminate.is_set():

                body = None

                try:
                    body = task_queue.get(block=True, timeout=10)
                except Queue.Empty:
                    # Ignore empty exception, we don't always have new tasks to run
                    pass

                if body:

                    task_queue.task_done()

                    bulk_tasks = list()
                    bulk_cuds = list()

                    for task in body:
                        t = Task()
                        t.from_dict(task)
                        bulk_tasks.append(t)
                        bulk_cuds.append(create_cud_from_task(
                            t, placeholder_dict, local_prof))

                        mq_connection = pika.BlockingConnection(pika.ConnectionParameters(host=mq_hostname, port=port))
                        mq_channel = mq_connection.channel()

                        transition(obj=t,
                                   obj_type='Task',
                                   new_state=states.SUBMITTING,
                                   channel=mq_channel,
                                   queue='%s-tmgr-to-sync' % sid,
                                   profiler=local_prof,
                                   logger=logger)

                        mq_connection.close()


                    umgr.submit_units(bulk_cuds)

                    for task in bulk_tasks:

                        mq_connection = pika.BlockingConnection(pika.ConnectionParameters(host=mq_hostname, port=port))
                        mq_channel = mq_connection.channel()

                        transition(obj=task,
                                   obj_type='Task',
                                   new_state=states.SUBMITTED,
                                   channel=mq_channel,
                                   queue='%s-tmgr-to-sync' % sid,
                                   profiler=local_prof,
                                   logger=logger)

                        mq_connection.close()

        except KeyboardInterrupt as ex:
            logger.exception('Execution interrupted by user (you probably hit Ctrl+C), ' +
                            'trying to cancel task processor gracefully...')

        except Exception as ex:
            logger.exception('%s failed with %s'%(self._uid, ex))
            raise EnTKError(ex)
**Purpose**: The new thread that gets spawned by the main tmgr process invokes this function. This
        function receives tasks from 'task_queue' and submits them to the RADICAL Pilot RTS.