Switch to using a CloseForLongOperation around the sleep
This commit is contained in:
parent
cf35da30bc
commit
7ee00b83cb
2 changed files with 18 additions and 27 deletions
|
@ -1,6 +1,5 @@
|
|||
from data.database import BUILD_PHASE, UseThenDisconnect
|
||||
from data.database import BUILD_PHASE
|
||||
from data import model
|
||||
from app import app
|
||||
|
||||
class StatusHandler(object):
|
||||
""" Context wrapper for writing status to build logs. """
|
||||
|
@ -44,10 +43,9 @@ class StatusHandler(object):
|
|||
self._append_log_message(phase, self._build_logs.PHASE, extra_data)
|
||||
|
||||
# Update the repository build with the new phase
|
||||
with UseThenDisconnect(app.config):
|
||||
repo_build = model.get_repository_build(self._uuid)
|
||||
repo_build.phase = phase
|
||||
repo_build.save()
|
||||
repo_build = model.get_repository_build(self._uuid)
|
||||
repo_build.phase = phase
|
||||
repo_build.save()
|
||||
|
||||
return True
|
||||
|
||||
|
|
|
@ -120,18 +120,16 @@ class BuilderServer(object):
|
|||
self._session_factory.remove(component)
|
||||
|
||||
def _job_heartbeat(self, build_job):
|
||||
with database.UseThenDisconnect(app.config):
|
||||
WorkQueue.extend_processing(build_job.job_item, seconds_from_now=JOB_TIMEOUT_SECONDS,
|
||||
retry_count=1, minimum_extension=MINIMUM_JOB_EXTENSION)
|
||||
WorkQueue.extend_processing(build_job.job_item, seconds_from_now=JOB_TIMEOUT_SECONDS,
|
||||
retry_count=1, minimum_extension=MINIMUM_JOB_EXTENSION)
|
||||
|
||||
def _job_complete(self, build_job, job_status):
|
||||
with database.UseThenDisconnect(app.config):
|
||||
if job_status == BuildJobResult.INCOMPLETE:
|
||||
self._queue.incomplete(build_job.job_item, restore_retry=True, retry_after=30)
|
||||
elif job_status == BuildJobResult.ERROR:
|
||||
self._queue.incomplete(build_job.job_item, restore_retry=False)
|
||||
else:
|
||||
self._queue.complete(build_job.job_item)
|
||||
if job_status == BuildJobResult.INCOMPLETE:
|
||||
self._queue.incomplete(build_job.job_item, restore_retry=True, retry_after=30)
|
||||
elif job_status == BuildJobResult.ERROR:
|
||||
self._queue.incomplete(build_job.job_item, restore_retry=False)
|
||||
else:
|
||||
self._queue.complete(build_job.job_item)
|
||||
|
||||
self._job_count = self._job_count - 1
|
||||
|
||||
|
@ -144,21 +142,18 @@ class BuilderServer(object):
|
|||
logger.debug('Checking for more work for %d active workers',
|
||||
self._lifecycle_manager.num_workers())
|
||||
|
||||
with database.UseThenDisconnect(app.config):
|
||||
job_item = self._queue.get(processing_time=self._lifecycle_manager.setup_time())
|
||||
|
||||
job_item = self._queue.get(processing_time=self._lifecycle_manager.setup_time())
|
||||
if job_item is None:
|
||||
logger.debug('No additional work found. Going to sleep for %s seconds', WORK_CHECK_TIMEOUT)
|
||||
yield From(trollius.sleep(WORK_CHECK_TIMEOUT))
|
||||
continue
|
||||
with database.CloseForLongOperation(app.config):
|
||||
yield From(trollius.sleep(WORK_CHECK_TIMEOUT))
|
||||
continue
|
||||
|
||||
try:
|
||||
build_job = BuildJob(job_item)
|
||||
except BuildJobLoadException as irbe:
|
||||
logger.exception(irbe)
|
||||
|
||||
with database.UseThenDisconnect(app.config):
|
||||
self._queue.incomplete(job_item, restore_retry=False)
|
||||
self._queue.incomplete(job_item, restore_retry=False)
|
||||
|
||||
logger.debug('Build job found. Checking for an avaliable worker.')
|
||||
scheduled = yield From(self._lifecycle_manager.schedule(build_job))
|
||||
|
@ -167,9 +162,7 @@ class BuilderServer(object):
|
|||
logger.debug('Build job scheduled. Running: %s', self._job_count)
|
||||
else:
|
||||
logger.debug('All workers are busy. Requeuing.')
|
||||
|
||||
with database.UseThenDisconnect(app.config):
|
||||
self._queue.incomplete(job_item, restore_retry=True, retry_after=0)
|
||||
self._queue.incomplete(job_item, restore_retry=True, retry_after=0)
|
||||
|
||||
@trollius.coroutine
|
||||
def _initialize(self, loop, host, websocket_port, controller_port, ssl=None):
|
||||
|
|
Reference in a new issue