Skip to content

Update poll_workflow_queue #2499

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
henrikek merged 1 commit into from
Jun 2, 2026
Merged

Update poll_workflow_queue #2499

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion ESSArch_Core/WorkflowEngine/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import exceptions, viewsets
from rest_framework.decorators import action
from rest_framework.filters import OrderingFilter
from rest_framework.generics import GenericAPIView, get_object_or_404
from rest_framework.response import Response
from rest_framework_extensions.mixins import NestedViewSetMixin
Expand Down Expand Up @@ -85,8 +86,10 @@ class ProcessStepViewSet(viewsets.ModelViewSet):
queryset = ProcessStep.objects.none()
serializer_class = ProcessStepSerializer
permission_classes = (ActionPermissions,)
filter_backends = (DjangoFilterBackend,)
filter_backends = (DjangoFilterBackend, OrderingFilter,)
filterset_class = ProcessStepFilter
ordering_fields = ['time_created']
ordering = ['time_created']

def get_queryset(self):
user = self.request.user
Expand Down
191 changes: 76 additions & 115 deletions ESSArch_Core/workflow/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -521,102 +521,77 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):

cache_running_key = f'running_process_steps_{queue_name}'

#
# Handle workflows already running
#

for root_step in ProcessStep.objects.filter(
# Count all active steps
running_count = ProcessStep.objects.filter(
parent=None,
queue=queue_name,
run_state='STARTED',
):

#
# Root workflow with part_root children
#

if (
root_step.get_children().filter(part_root=True).exists() and
root_step.status not in [
celery_states.FAILURE,
celery_states.REVOKED,
]
):

logger.info(
'root_step %s (STARTED) has part_root steps [queue=%s]',
root_step,
queue_name,
)

#
# Mark completed part_root steps
#

for part_root_step in root_step.get_children().filter(
part_root=True,
run_state='STARTED',
):

if part_root_step.status == celery_states.SUCCESS:

logger.info(
'part_root_step %s SUCCESS [queue=%s]',
part_root_step,
queue_name,
)

part_root_step.run_state = 'SUCCESS'
part_root_step.save(update_fields=['run_state'])
).count()

#
# Start new part_root steps if capacity exists
#
logger.debug('Initial running_count=%s [queue=%s]', running_count, queue_name)
cache.set(cache_running_key, running_count)

running_count = ProcessStep.objects.filter(
queue=queue_name,
run_state='STARTED',
).count()
#
# 1. Handle started root steps with part_root children
#
for root_step in ProcessStep.objects.filter(parent=None, queue=queue_name, run_state='STARTED'):
part_root_children = root_step.get_children().filter(part_root=True)

if part_root_children.count():
logger.debug('root_step has part_root childs, set running_count=0')
running_count = 0
cache.set(cache_running_key, running_count)

for part_root_step in root_step.get_children().filter(
part_root=True,
run_state='PENDING',
):

if (
part_root_step.status == celery_states.PENDING and
cache.get(cache_running_key) < max_running_steps
):

logger.info(
'Starting part_root_step %s [queue=%s]',
part_root_step,
queue_name,
)

part_root_step.run()

part_root_step.run_state = 'STARTED'
part_root_step.save(update_fields=['run_state'])

cache.incr(cache_running_key)

else:

logger.info(
'Queue %s full (%s/%s)',
queue_name,
cache.get(cache_running_key),
max_running_steps,
)
# Mark completed children
for child in part_root_children.filter(run_state='STARTED'):
if child.status == 'SUCCESS':
child.run_state = 'SUCCESS'
child.save(update_fields=['run_state'])
# running_count -= 1
# cache.set(cache_running_key, running_count)
logger.info('Marked child %s %s SUCCESS [queue=%s]', child, child.information_package, queue_name)

# Count FAILED but still STARTED children
failed_started_count = 0
for child in part_root_children.filter(run_state='STARTED'):
if child.status == 'FAILURE':
failed_started_count += 1
else:
running_count += 1

break
if failed_started_count > 0:
logger.info(
'Queue %s has %s FAILED children, cannot start more children"',
queue_name,
failed_started_count
)
break
# running_count += failed_started_count
cache.set(cache_running_key, running_count)

# Start pending part_root children of STARTED roots
pending_children = part_root_children.filter(run_state='PENDING')

for child in pending_children:
if cache.get(cache_running_key) >= max_running_steps:
logger.info("Queue %s full (%s/%s), cannot start more children",
queue_name, cache.get(cache_running_key), max_running_steps)
break

logger.info("Starting child step %s %s [queue=%s]", child, child.information_package, queue_name)
child.run()
child.run_state = 'STARTED'
child.save(update_fields=['run_state'])
cache.incr(cache_running_key)

#
# Mark root workflow complete
#
#
# 2. Mark root workflow complete
#
for root_step in ProcessStep.objects.filter(
parent=None,
queue=queue_name,
run_state='STARTED',
):

if (
not root_step.get_children()
Expand All @@ -625,7 +600,6 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):
.exists() and
root_step.status == celery_states.SUCCESS
):

logger.info(
'root_step %s SUCCESS [queue=%s]',
root_step,
Expand All @@ -636,20 +610,19 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):
root_step.save(update_fields=['run_state'])

#
# Recalculate running count
# 3. Recalculate running count
#

running_count = ProcessStep.objects.filter(
parent=None,
queue=queue_name,
run_state='STARTED',
).count()

cache.set(cache_running_key, running_count)

#
# Start pending root workflows
# 4. Start pending root workflows
#

pending_root_steps = ProcessStep.objects.filter(
parent=None,
queue=queue_name,
Expand All @@ -658,28 +631,19 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):

for root_step in pending_root_steps:

if cache.get(cache_running_key) >= max_running_steps:
current_running = cache.get(cache_running_key) or 0

if current_running >= max_running_steps:
logger.info(
'Queue %s full (%s/%s)',
queue_name,
cache.get(cache_running_key),
current_running,
max_running_steps,
)

break

#
# Root workflow contains part_root lanes
#

if root_step.get_children().filter(part_root=True).exists():

#
# Preserve existing behaviour:
# only one root workflow with active part_root execution
#

if (
getattr(
settings,
Expand All @@ -693,12 +657,10 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):
child_steps__run_state__in=['STARTED', 'PENDING'],
).exists()
):

logger.info(
'Queue %s already has active root workflow with part_root steps',
queue_name,
)

continue

logger.info(
Expand All @@ -710,16 +672,19 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):
for part_root_step in root_step.get_children().filter(
part_root=True,
run_state='PENDING',
):
).order_by('time_created'):

if (
part_root_step.status == celery_states.PENDING and
cache.get(cache_running_key) < max_running_steps
):
current_running = cache.get(cache_running_key) or 0

if current_running >= max_running_steps:
break

if part_root_step.status == celery_states.PENDING:

logger.info(
'Starting part_root_step %s [queue=%s]',
'Starting part_root_step %s %s [queue=%s]',
part_root_step,
part_root_step.information_package,
queue_name,
)

Expand All @@ -730,16 +695,12 @@ def _poll_workflow_queue(queue_name, max_running_steps, logger):

cache.incr(cache_running_key)

else:
break

if (
root_step.get_children()
.filter(part_root=True, run_state='STARTED')
.exists() and
root_step.run_state == 'PENDING'
):

root_step.run_state = 'STARTED'
root_step.save(update_fields=['run_state'])

Expand Down
Loading