[BUG]: Python source stages not shutting down if another stage raises an exception

[dagardner-nv]

Version

24.10

Which installation method(s) does this occur on?

Source

Describe the bug.

This primarily impacts sources which run indefinitely (HttpClientSourceStage, AppShieldSourceStage, RSSSourceStage)

C++ sources avoid this issue as they have access to the subscriber object and a well behaved source stage will routinely check the subscriber.is_subscribed() method at the beginning of each poll iteration.

Python sources don't have access to the subscriber, and while the pymrc C++ wrapper (python/mrc/_pymrc/src/segment.cpp) can check this there are two problems:

1. It doesn't stop iterating when it returns a false (easily fixed).
2. It blocks on each iteration of for (auto next_val : iter_wrapper), so unless the source yields, the C++ code is blocked.

Ideal solution:
The MRC executor should accept an optional on_stage_change callback method
pymrc should wrap this allowing for a Python impl for the callback
Morpheus' pipeline.py should supply a callback which calls Pipeline.stop() when the status changes to a terminal state.

Minimum reproducible example

   pipeline.set_source(
                HttpClientSourceStage(config, url="http://localhost:8080/api/v1/data", sleep_time=sleep_time))

    @stage
    def print_msg(msg: typing.Any, *, raise_error: bool = False) -> typing.Any:
        if isinstance(msg, MessageMeta):
            print(f"Received:\n{msg.df}")
            if raise_error:
                print("Raising error")
                raise ValueError("Error")
        else:
            print(f"Received:\n{msg}")

        return msg

    pipeline.add_stage(print_msg(config, raise_error=True))

Relevant log output

Click here to see error details

[Paste the error here, it will be hidden by default]

Full env printout

Click here to see environment details

[Paste the results of print_env.sh here, it will be hidden by default]

Other/Misc.

No response

Code of Conduct

• I agree to follow Morpheus' Code of Conduct
• I have searched the open bugs and have found no duplicates for this bug report