Ability to check whether we are on the last task of a SequentialTaskSet is gone

[cyberw]

Prerequisites

• [X] I am using the latest version of Locust
• [X] I am reporting a bug, not asking a question

Description

Some of our users (@tdadela) are dependant on being able to determine if they are on the last task of a SequentialTaskSet (by checking the index), but that ability disappeared now that it is implemented as a cycle (since #2742).

@bakhtos can I bother you to take a look at this?

Command line

Locustfile contents

class MeasuredSequentialTaskSet(TaggedSequentialTaskSet):
    one_iteration_only = False

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.user.last_task = False

    def on_stop(self):
        self.user.last_task = True
        super().on_stop()

    @property
    def _task_number(self):
        return self._task_index % len(self.tasks)

    @property
    def _is_first_task(self):
        return self._task_number == 1 or len(self.tasks) == 1

    @property
    def _is_last_task(self):
        return self._task_number == 0

    def execute_task(self, task):
        if self._is_first_task:
            self.user.last_task = False
            self.user.start_time = time()
        try:
            super().execute_task(task)
        except InterruptTaskSet:
            if self.one_iteration_only:
                raise StopUser
            raise

        if self._is_last_task:
            self.user.last_task = True
            if self.one_iteration_only:
                raise StopUser()

Python version

Locust version

latest

Operating system

[bakhtos]

Hello!

Since the SequentialTaskSet still stores the list of all tasks, it should be possible to modify the custom class definition:

class MeasuredSequentialTaskSet(TaggedSequentialTaskSet):
    one_iteration_only = False

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.user.last_task = False
        self._last_task = self.tasks[-1]
        self._first_task = self.tasks[0]

    def on_stop(self):
        self.user.last_task = True
        super().on_stop()

    def _is_first_task(self, task):
        return task is self._first_task

    def _is_last_task(self, task):
        return task is self._last_task

    def execute_task(self, task):
        if self._is_first_task(task):
            self.user.last_task = False
            self.user.start_time = time()
        try:
            super().execute_task(task)
        except InterruptTaskSet:
            if self.one_iteration_only:
                raise StopUser
            raise

        if self._is_last_task(task):
            self.user.last_task = True
            if self.one_iteration_only:
                raise StopUser()

So the class should store the actual function objects that are the first/last task in the list, and check the identity of the task that execute_task receives.

[cyberw]

Is that good enough for you @domik82 ?

[github-actions[bot]]

This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 10 days.

[github-actions[bot]]

This issue was closed because it has been stalled for 10 days with no activity. This does not necessarily mean that the issue is bad, but it most likely means that nobody is willing to take the time to fix it. If you have found Locust useful, then consider contributing a fix yourself!

[tdadela]

Hi, @cyberw I don't know details of your conversation with Dominik (which resulted in opening this issue) but I don't want to leave this without a response from our team. The PR that extended functionality of SequentialTaskSet wasn't compatible with our extensions of it. However, breaking changes are somehow expected because our project is relatively big and complex and uses not only public API but also a lot of internal implementation details. It was easy to adjust to this change. Thank you, @bakhtos, for support and providing example implementation. Sorry for late response.