[G-API] Handle exceptions in streaming executor #21660

TolyaTalamanov · 2022-02-24T15:41:31Z

Pull Request Readiness Checklist

See details at https://github.com/opencv/opencv/wiki/How_to_contribute#making-a-good-pull-request

I agree to contribute to the project under Apache 2 License.
To the best of my knowledge, the proposed patch is not based on a code under GPL or another license that is incompatible with OpenCV
The PR is proposed to the proper branch
There is a reference to the original bug report and related work
There is accuracy test, performance test and test data in opencv_extra repository, if applicable
Patch to opencv_extra has the same branch name.
The feature is well documented and sample code can be built with the project CMake

Build configuration

force_builders=Custom,Custom Win,Custom Mac
build_gapi_standalone:Linux x64=ade-0.1.1f
build_gapi_standalone:Win64=ade-0.1.1f
build_gapi_standalone:Mac=ade-0.1.1f
build_gapi_standalone:Linux x64 Debug=ade-0.1.1f
Xbuild_image:Custom=centos:7
Xbuildworker:Custom=linux-1
build_gapi_standalone:Custom=ade-0.1.1f
build_image:Custom=ubuntu-openvino-2021.4.1:20.04
build_image:Custom Win=openvino-2021.4.1
build_image:Custom Mac=openvino-2021.4.1
test_modules:Custom=gapi,python2,python3,java
test_modules:Custom Win=gapi,python2,python3,java
test_modules:Custom Mac=gapi,python2,python3,java
buildworker:Custom=linux-1
# disabled due high memory usage: test_opencl:Custom=ON
test_opencl:Custom=OFF
test_bigdata:Custom=1
test_filter:Custom=*

TolyaTalamanov · 2022-02-25T08:54:55Z

modules/gapi/include/opencv2/gapi/cpu/gcpukernel.hpp



 #ifndef OPENCV_GAPI_GCPUKERNEL_HPP
 #define OPENCV_GAPI_GCPUKERNEL_HPP

+#ifdef _MSC_VER


Seems to be false positive warning, but need to double check

TolyaTalamanov · 2022-02-25T08:57:58Z

modules/gapi/src/executor/gstreamingexecutor.cpp

@@ -889,6 +929,33 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
            }
        }
    }
+
+    virtual void post(cv::gimpl::Exception&& error) override


Copy paste from post(EndOfStream&&) except incrementing m_stops_sent

IMHO:
From my perspective there are a lot of sort-out and shitfing from queue to queue code here as a legacy part.
But it this function just copied & pasted then I would recommend introduce self-documented private function here like
dispatch<Message>()
because it would be easier to understand full-body of post function when see that post just makes dispatch to subsequent

That's the problem, this function isn't fully copy-pasted. (m_stops_sent++ removed)

i assumed some customization here, for example

post(Exception ) { (void)dispatch<Exception> () } post(Stop) { stop_sent += dispatch <Stop>(); }

Please consider this comment as optional

TolyaTalamanov · 2022-02-25T08:58:40Z

modules/gapi/src/executor/gstreamingexecutor.cpp

-            GAPI_ITT_AUTO_TRACE_GUARD(collector_push_hndl);
-            out_queue.push(Cmd{Result{std::move(this_result), flags}});
+            case QueueReader::V::index_of<Stop>():
+                if (handle_stop)


@AsyaPronina
Does it make sense to put traces here?

* Not only (std::exception& e)

TolyaTalamanov · 2022-03-09T13:41:54Z

@sivanov-work Could you have a look?

sivanov-work

IMHO: there a lot of switch-like code block which handles different types of variant and it's hard to understand & read it. Furthermore, for future customization ( adding new message type) it requires to sort-out a lot of similar code and doesn't look upgraded easily

I'd propose to introduce full-descriptive-class of Message of ControlMessage as part of current std::variant at least then introduce message_handle(ControlMessage &) to encapsulate operations with it - to make code clear

Also, there might be a sort of problem in working with queue capacity: when exception (and other command messages) cannot be pushed due to queue fullness. I'd propose to consider to modify queue to some kind of smart queue: which allow to push controlMessages when it's full as high priority messages

sivanov-work · 2022-03-09T13:44:43Z

modules/gapi/src/backends/ie/giebackend.cpp

+    try {
+        t.run(request);
+    } catch (...) {
+        request.SetCompletionCallback([](){});


i'd put LOG_WARNING/ERROR here something like:

request by id: " << id << " failed, error" << ex.what()

Don't think that log is needed here, user will see exception message when call pull.
I'd propose to make exception msg a bit clearer.

it's optional comment: please feel free to ignore it
in defense of logger I could say that it may provide more detailed information about what actually happened and what current states we have which will help in further troubleshooting

sivanov-work · 2022-03-09T13:51:32Z

modules/gapi/src/backends/ie/giebackend.cpp

    GAPI_ITT_STATIC_LOCAL_HANDLE(ie_cb_post_outputs_hndl, "IE_async_callback_PostOutputs");
    GAPI_ITT_AUTO_TRACE_GUARD(ie_cb_post_outputs_hndl);

+    if (code != IE::StatusCode::OK) {
+        ctx->eptr = std::make_exception_ptr(
+               std::logic_error("IE::InferRequest finished with not OK status"));


from my perspective it would be great also to print status value

sivanov-work · 2022-03-09T14:13:18Z

modules/gapi/src/compiler/gislandmodel.hpp

+
+struct Exception {
+    std::exception_ptr eptr;
+};


IMHO:
encapsulation in some kind of ServiceMessage will simplify future extensibility:

struct ServiceMessage { enum { EndOfStream, Exception } .... }; using StreamMsg = cv::util::variant<ServiceMessage, cv::GRunArgs>;

then

if (cv::util::holds_alternative<cv::gimpl::EndOfStream>(in_msg)) { } if (cv::util::holds_alternative<cv::gimpl::Exception>(in_msg)) { }

might be modified as

if (cv::util::holds_alternative<cv::gimpl::ServiceMessage>(in_msg)) { out.post(in_msg); }

It's really make code a bit clearer, but

Is there any other possible message types? (I can't imagine)

EndOfStream & Exception should be handled in different ways (almost the same but not)

Is there any other possible message types?

there had been only one Stop message before you introduced Exception - so hypothetically if any new messages are required we will need to repeat this effort again for adding new variant types

EndOfStream & Exception should be handled in different ways (almost the same but not)

that's because we would separate "traffic" messages & "service" messages in processing. Something like that:

if (cv::util::holds_alternative<cv::gimpl::ServiceMessage>(in_msg)) { //encapsulate switch-case process_service_message(in_msg); //dispatch for other islands, stop, exception and so on... } else { .. frame processing }

P.S. optional comment

sivanov-work · 2022-03-10T06:39:30Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+               break;
+           }
+           default:
+                cv::util::throw_error(std::logic_error("Unsupported cmd type in getInputVector()"));


i think it had better to put GAPI_DbgAssert here in consideration that it acts as c-like assert

What do you think about GAPI_Assert instead? In order to keep consistency with other code (see above)
GAPI_Assert(stop.kind == Stop::Kind::HARD); and etc...

sivanov-work · 2022-03-10T06:47:01Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+               oq->push(Cmd{cv::gimpl::Exception{eptr}});
+           }
+           // NB: Go to the next iteration.
+           continue;


Will we hang here forever?

when emitter produces exception constantly then we catch it and go to 'continue' for while (true)

If source produces exceptions constantly then these exceptions will be propagated through the graph and rethrown in GStreamingExecutor::pull to the user.

Emitter shouldn't be stopped if source caused exception, see the test where source throw exception every second frame

sivanov-work · 2022-03-10T06:48:27Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+                    break;
+                case Posting::V::index_of<cv::gimpl::EndOfStream>():
+                    cmd = Cmd{Stop{}};
+                    m_stops_sent++;


Does it make sense to introduce m_exceptions_caught++ for exception case, doesn't it?

I didn't get the idea, why it's needed?

not sure, for statistics/analytics maybe - in case of multiple exception are allowed in pipeline without stopping it

If this case is not supported - then please ignore this comment

Multiple exceptions aren't supported.

In this PR, I try not to allow exceptions to be thrown from not main thread.

sivanov-work · 2022-03-10T06:53:22Z

modules/gapi/src/executor/gstreamingexecutor.cpp

@@ -889,6 +929,33 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
            }
        }
    }
+
+    virtual void post(cv::gimpl::Exception&& error) override


IMHO:
From my perspective there are a lot of sort-out and shitfing from queue to queue code here as a legacy part.
But it this function just copied & pasted then I would recommend introduce self-documented private function here like
dispatch<Message>()
because it would be easier to understand full-body of post function when see that post just makes dispatch to subsequent

sivanov-work

LGTM if old & new tests are passed

sivanov-work · 2022-03-10T09:45:58Z

modules/gapi/src/backends/ie/giebackend.cpp

+    try {
+        t.run(request);
+    } catch (...) {
+        request.SetCompletionCallback([](){});


it's optional comment: please feel free to ignore it
in defense of logger I could say that it may provide more detailed information about what actually happened and what current states we have which will help in further troubleshooting

sivanov-work · 2022-03-10T09:55:04Z

modules/gapi/src/compiler/gislandmodel.hpp

+
+struct Exception {
+    std::exception_ptr eptr;
+};


Is there any other possible message types?

there had been only one Stop message before you introduced Exception - so hypothetically if any new messages are required we will need to repeat this effort again for adding new variant types

EndOfStream & Exception should be handled in different ways (almost the same but not)

that's because we would separate "traffic" messages & "service" messages in processing. Something like that:

if (cv::util::holds_alternative<cv::gimpl::ServiceMessage>(in_msg)) { //encapsulate switch-case process_service_message(in_msg); //dispatch for other islands, stop, exception and so on... } else { .. frame processing }

P.S. optional comment

sivanov-work · 2022-03-10T10:00:40Z

modules/gapi/src/executor/gexecutor.cpp

+    {
+        if (eptr)
+        {
+            std::rethrow_exception(eptr);


is it some kind of default/stub implementation?

because it now has a "state" and it affects multithreading behavior between simultaneous post and verify at least

GExecutor::IOutput methods might be called from different threads, but not concurrently.
verify() is supposed to be called after post()

https://github.com/TolyaTalamanov/opencv/blob/at/handle-exception-in-streamingexecutor/modules/gapi/src/executor/gexecutor.cpp#L410-L412

That's a good point, but I don't know how guarantee that's methods won't be called concurrently in future, except putting comment

sivanov-work · 2022-03-10T10:02:44Z

modules/gapi/src/executor/gstreamingexecutor.cpp

-bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
-                                 cv::GRunArgs    &in_constants,
-                                 cv::GRunArgs    &isl_inputs)
+cv::gimpl::StreamMsg QueueReader::getInputVector(std::vector<Q*> &in_queues,


can you clarify - why it changed on returning value?

Previously bool means != Stop{}, now except Stop there can be Exception message, so it was changed from bool to StreamMsg

sivanov-work · 2022-03-10T10:05:49Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+               oq->push(Cmd{cv::gimpl::Exception{eptr}});
+           }
+           // NB: Go to the next iteration.
+           continue;


when emitter produces exception constantly then we catch it and go to 'continue' for while (true)

sivanov-work · 2022-03-10T10:08:52Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+                    break;
+                case Posting::V::index_of<cv::gimpl::EndOfStream>():
+                    cmd = Cmd{Stop{}};
+                    m_stops_sent++;


not sure, for statistics/analytics maybe - in case of multiple exception are allowed in pipeline without stopping it

If this case is not supported - then please ignore this comment

sivanov-work · 2022-03-10T10:11:04Z

modules/gapi/src/executor/gstreamingexecutor.cpp

@@ -889,6 +929,33 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
            }
        }
    }
+
+    virtual void post(cv::gimpl::Exception&& error) override


i assumed some customization here, for example

post(Exception ) { (void)dispatch<Exception> () } post(Stop) { stop_sent += dispatch <Stop>(); }

Please consider this comment as optional

sivanov-work · 2022-03-10T10:14:04Z

modules/gapi/src/executor/gstreamingexecutor.cpp

+        case Cmd::index_of<Exception>(): {
+            std::rethrow_exception(cv::util::get<Exception>(cmd).eptr);
+            return true;
+        }


Is there any warning about missing 'default' ?

TolyaTalamanov · 2022-03-14T12:07:39Z

@dmatveev Could you have a look, please?

dmatveev

Updating every possible backend with this new exception handling is quite invasive.

Can you please offer some other way to do it?
Can those changes be only introduced at the executors level?

dmatveev · 2022-03-15T12:11:56Z

modules/gapi/src/backends/ie/giebackend.cpp

+    if (cv::util::holds_alternative<cv::gimpl::Exception>(in_msg))
+    {
+        // (4) If the Exception message is revieved, propagate it further.
+        out.post(std::move(cv::util::get<cv::gimpl::Exception>(in_msg)));
+        return;
+    }


Should this be done at this level? As well as every other island's level?

I mean, shouldn't this be handled uniformly for all possible islands? A level above.

Couldn't agree more. Handle the input exception in a general way.

dmatveev · 2022-03-15T12:14:10Z

modules/gapi/src/backends/oak/goakbackend.cpp

+    if (cv::util::holds_alternative<cv::gimpl::Exception>(in_msg)) {
+        out.post(std::move(cv::util::get<cv::gimpl::Exception>(in_msg)));
+        return;
+    }


well, if we need to update every our backend now, this change is quite invasive.

What's about our huge internal part?

Answered above

dmatveev

It is probably done right but I can't say it for sure until it is properly documented

modules/gapi/src/backends/ie/giebackend.cpp

dmatveev · 2022-03-22T18:52:13Z

modules/gapi/src/backends/ie/giebackend.cpp

+    try {
+        t.run(request);
+    } catch (...) {
+        request.SetCompletionCallback([](){});


A comment is required here. Previous single-liner was clear, now I have no idea why this should happen in catch. And I doubt if anybody knows.

dmatveev · 2022-03-22T18:53:22Z

modules/gapi/src/backends/ie/giebackend.cpp

 void cv::gimpl::ie::RequestPool::callback(cv::gimpl::ie::RequestPool::Task task,
                                          size_t id,
                                          IE::InferRequest request,
-                                          IE::StatusCode code) {
-    // FIXME: Any exception which is arrised here must not leave this callback,
-    // because it won't be handled.
-    try {
-        if (code != IE::StatusCode::OK) {
-            throw std::logic_error("IE::InferRequest finished with not OK status");
-        }
-        task.callback(request);
-        // NB: IE::InferRequest keeps the callback until the new one is set.
-        // Since user's callback might keep resources that should be released,
-        // need to destroy its after execution.
-        // Let's set the empty one to cause the destruction of a callback.
-        request.SetCompletionCallback([](){});
-        m_idle_ids.push(id);
-    } catch (const std::exception& e) {
-        GAPI_LOG_FATAL(NULL, "Callback failed with error: " << e.what());
-        //FIXME: Exception CAN't be rethrown here, since this callback works
-        // in separate IE thread and such scenarios aren't handled properly in
-        // G-API so far.
-    }
+                                          IE::StatusCode code) noexcept {
+    task.callback(request, code);
+    request.SetCompletionCallback([](){});
+    m_idle_ids.push(id);


What is this all about? Why? What's for? A simple comment would help to understand the need in this method.

dmatveev · 2022-03-22T18:56:33Z

modules/gapi/src/backends/streaming/gstreamingbackend.cpp

@@ -172,6 +172,7 @@ void Copy::Actor::run(cv::gimpl::GIslandExecutable::IInput  &in,
        return;
    }

+    GAPI_Assert(cv::util::holds_alternative<cv::GRunArgs>(in_msg));


Should this be a DgbAssert? get<> will fail anyway if the message is of other type.

Make sense, done

modules/gapi/test/infer/gapi_infer_ie_test.cpp

modules/gapi/src/executor/gstreamingexecutor.cpp

* Put more comments * Fix alignment * Move test outside of HAVE_NGRAPH

dmatveev

👍

modules/gapi/src/backends/ie/giebackend.cpp

TolyaTalamanov · 2022-03-25T08:07:23Z

@alalek Could you merge it, please?

…on-in-streamingexecutor [G-API] Handle exceptions in streaming executor * Handle exceptions in streaming executor * Rethrow exception in non-streaming executor * Clean up * Put more tests * Handle exceptions in IE backend * Handle exception in IE callbacks * Handle exception in GExecutor * Handle all exceptions in IE backend * Not only (std::exception& e) * Fix comments to review * Handle input exception in generic way * Fix comment * Clean up * Apply review comments * Put more comments * Fix alignment * Move test outside of HAVE_NGRAPH * Fix compilation

Handle exceptions in streaming executor

ffb05c4

TolyaTalamanov added this to the 4.6.0 milestone Feb 24, 2022

TolyaTalamanov added 2 commits February 24, 2022 20:02

Rethrow exception in non-streaming executor

ad80cbb

Clean up

086872a

TolyaTalamanov force-pushed the at/handle-exception-in-streamingexecutor branch from a6f530d to 086872a Compare February 25, 2022 09:09

TolyaTalamanov commented Feb 25, 2022

View reviewed changes

TolyaTalamanov added 5 commits February 28, 2022 15:59

Put more tests

21bad11

Handle exceptions in IE backend

d44e786

Handle exception in IE callbacks

1235c70

Handle exception in GExecutor

f98daa5

Handle all exceptions in IE backend

50fac32

* Not only (std::exception& e)

TolyaTalamanov requested a review from sivanov-work March 9, 2022 13:41

sivanov-work reviewed Mar 10, 2022

View reviewed changes

sivanov-work approved these changes Mar 10, 2022

View reviewed changes

Fix comments to review

8f8296c

TolyaTalamanov requested a review from dmatveev March 14, 2022 12:07

dmatveev self-assigned this Mar 15, 2022

dmatveev added the category: g-api / gapi label Mar 15, 2022

dmatveev reviewed Mar 15, 2022

View reviewed changes

TolyaTalamanov added 3 commits March 16, 2022 16:26

Handle input exception in generic way

c812bed

Fix comment

abdca09

Clean up

1e429d5

dmatveev reviewed Mar 22, 2022

View reviewed changes

TolyaTalamanov added 2 commits March 24, 2022 11:00

Apply review comments

7fc9980

* Put more comments * Fix alignment * Move test outside of HAVE_NGRAPH

Fix compilation

0a27b6b

dmatveev approved these changes Mar 25, 2022

View reviewed changes

modules/gapi/src/backends/ie/giebackend.cpp Show resolved Hide resolved

alalek merged commit d98e07c into opencv:4.x Mar 25, 2022

opencv-pushbot mentioned this pull request Apr 23, 2022

(5.x) Merge 4.x #21903

Merged

Uh oh!

[G-API] Handle exceptions in streaming executor #21660

[G-API] Handle exceptions in streaming executor #21660

Uh oh!

Conversation

TolyaTalamanov commented Feb 24, 2022

Pull Request Readiness Checklist

Build configuration

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TolyaTalamanov commented Mar 9, 2022

Uh oh!

sivanov-work left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sivanov-work Mar 10, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sivanov-work left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

sivanov-work Mar 10, 2022 •

edited

Loading

sivanov-work Mar 10, 2022 •

edited

Loading