

def run_task(self, msg):
    proc = Popen(self.task_cmd, stdout=PIPE, shell=True)
    stdout_reader = AsynchronousFileReader(proc.stdout, self.io_queue)
    stdout_reader.start()
    while (not stdout_reader.eof()):
        while True:
            try:
                line = self.io_queue.get_nowait()
            except Empty:
                break
            else:
                self.job_log.info(line)
        lease_age = (datetime.now() - self.lease_start)
        if (lease_age.seconds > (self.ackdeadline - 20)):
            log.debug('extending lease')
            try:
                resp = self.extend_lease(msg)
                self.extend_error_ct = 0
                self.lease_start = datetime.now()
            except HttpError as e:
                if (e.resp.status == 503):
                    log.warning('PubSub returned 503')
                    self.extend_error_ct += 1
                    if (self.extend_error_ct > 5):
                        log.critical('Too many error responses to extend request')
                        raise
        time.sleep(1)
    retcode = proc.poll()
    if (retcode is not None):
        log.debug('process ended')
    return retcode

dataset/ETHPy150Open GoogleCloudPlatform/reliable-task-scheduling-compute-engine-sample/gce/cron_executor.py Executor.run_task/original