Refactor dispatch pipeline in C++ #1923

bernardnormier · 2024-03-10T21:23:08Z

This PR refactor the dispatch pipeline in C++.

It replaces Incoming and ResponseHandler with an IceRPC-like dispatch pipeline. Ice's dispatcher "interface" is simply Object, with a public dispatch function.

It also introduces two internal middleware: ObserverMiddleware and LoggerMiddleware. ServantManager also becomes a dispatcher.

You'll notice that this PR doesn't change any test: the external behavior remains identical.

…uest, sendResponse) on Object.

bernardnormier · 2024-03-10T21:24:37Z

cpp/src/Ice/Incoming.cpp

-}
-
-OutputStream*
-Incoming::startWriteParams()


All the "write response" code was moved or rewritten in OutgoingResponse.cpp.

bernardnormier · 2024-03-10T21:25:26Z

cpp/src/Ice/Incoming.cpp

-        assert(_responseHandler);
-        if (_isTwoWay)
-        {
-            _observer.reply(static_cast<int32_t>(_os.b.size() - headerSize - 4));


All the observer code is now in the ObserverMiddleware.

bernardnormier · 2024-03-10T21:26:19Z

cpp/src/Ice/Incoming.cpp

-        if (_isTwoWay)
-        {
-            _observer.reply(static_cast<int32_t>(_os.b.size() - headerSize - 4));
-            _responseHandler->sendResponse(_current.requestId, &_os, _compress);


The sendResponse/sendNoResponse logic is now in the new sendResponse(OutgoingResponse) function implemented by ConnectionI and CollocatedRequestHandler.

bernardnormier · 2024-03-10T21:26:42Z

cpp/src/Ice/Incoming.cpp

-}
-
-void
-Incoming::warning(const Exception& ex) const


This code moved to the LoggerMiddleware.

bernardnormier · 2024-03-10T21:27:09Z

cpp/src/Ice/Incoming.cpp

-}
-
-bool
-Incoming::servantLocatorFinished()


The servant locator handling moved to ServantManager::dispatch.

bernardnormier · 2024-03-10T21:27:31Z

cpp/src/Ice/Incoming.cpp

-}
-
-void
-Incoming::handleException(std::exception_ptr exc)


This largely moved to OutgoingResponse.cpp - except for the logging code (=> LoggerMiddleware) and the observer code (=> ObserverMiddleware).

bernardnormier · 2024-03-10T21:28:32Z

cpp/src/Ice/Incoming.cpp

-}
-
-void
-Incoming::invoke(const ServantManagerPtr& servantManager, InputStream* stream)


This dispatch logic moved to the IncomingRequest constructor and ServantManager::dispatch.

bernardnormier · 2024-03-10T21:32:54Z

cpp/src/Ice/ConnectionI.h

@@ -268,6 +258,11 @@ namespace Ice
        void initiateShutdown();
        void sendHeartbeatNow();

+        void sendResponse(OutgoingResponse, std::uint8_t compress);
+        void sendResponse(std::int32_t, Ice::OutputStream*, std::uint8_t);


These 3 functions used to be the ResponseHandler. They are called by the first sendResponse; we could refactor further and merge sendResponse + sendNoResponse into the main sendResponse.

bernardnormier · 2024-03-10T21:35:01Z

cpp/src/Ice/ConnectionI.h

@@ -320,7 +308,6 @@ namespace Ice
        mutable Ice::ConnectionInfoPtr _info;

        ObjectAdapterPtr _adapter;
-        IceInternal::ServantManagerPtr _servantManager;


It is not clear to me why we stored and passed around the servant manager. _servantManager is an immutable field of ObjectAdapter, so we could just ask ObjectAdapter whenever we needed it. That's what this PR does for dispatcher.

bernardnormier · 2024-03-10T21:37:30Z

cpp/include/Ice/LocalException.h

-     * Indicates the application attempted to call the callbacks for an async dispatch more than once, or called an
-     * async dispatch callback after throwing an exception from the dispatch thread. \headerfile Ice/Ice.h
-     */
-    class ICE_CLASS(ICE_API) ResponseSentException : public LocalExceptionHelper<ResponseSentException, LocalException>


Removed this exception. If you attempt to call twice an AMD callback, the second call will be no-op. I want the AMD callbacks to be logically noexcept, just like sendResponse.

pepone

Looks good, just a few comments.

pepone · 2024-03-11T09:22:20Z

cpp/src/Ice/CollocatedRequestHandler.cpp

    {
-        invokeException(requestId, current_exception(), invokeNum); // Fatal invocation exception
+        invokeException(requestId, current_exception()); // Fatal invocation exception
    }

    _adapter->decDirectCount();


We call decDirectCount in invokeException implementation and again here, which doesn't seem correct. We can end up with a negative count.

I didn't change this code.

I believe it's correct, even if the style is not the best.

This _adapter->decDirectCount() "releases" the count that we "acquired" before the call to invokeAll. It's unrelated to the count acquired in the while loop.

cpp/src/Ice/Object.cpp

pepone · 2024-03-11T09:44:41Z

cpp/src/Ice/ObjectAdapterI.h

@@ -137,7 +137,8 @@ namespace Ice
        IceInternal::ObjectAdapterFactoryPtr _objectAdapterFactory;
        IceInternal::ThreadPoolPtr _threadPool;
        IceInternal::ACMConfig _acm;
-        IceInternal::ServantManagerPtr _servantManager;
+        const IceInternal::ServantManagerPtr _servantManager;
+        ObjectPtr _dispatcher;


Shouldn't it be const?

I update it in the constructor, so making it const is not convenient. I also plan to move the "composition root" somewhere else in a follow-up PR.

We could use const_cast, that's what we do in other constructors.

pepone · 2024-03-11T09:46:56Z

cpp/src/Ice/ObserverMiddleware.cpp

+void
+ObserverMiddleware::dispatch(Ice::IncomingRequest& request, function<void(OutgoingResponse)> sendResponse)
+{
+    auto observerPtr = _communicatorObserver->getDispatchObserver(request.current(), request.size());


I would use ObserverPtr observer = ... here

Fixed. It's actually a DispatchObserverPtr.

pepone · 2024-03-11T09:53:54Z

cpp/src/Ice/ObserverMiddleware.cpp

+            observerPtr->detach();
+
+            // If we get a synchronous exception, sendResponse was not called.
+            sendResponse(std::move(response));


Isn't this preventing the logger middleware from seeing the exceptions? Where are these exceptions handled when we don't have an observer?

No, absolutely not. Each middleware must handle both path:

the exception path (when an exception is thrown synchronously)

the sendResponse path, when the response carries an exception

And OutgoingResponse carries all the information any middleware may need to log/observe (...) any response, including responses with an exception payload.

I don't follow the logic, here we call sendResponse for exception that were throw synchronously. In LoggerMiddlware we don't and rethrow them.

The preferred style when you catch an exception synchronously is to observe it and rethrow, because that's the least amount of work.

In the observer middleware, we use another style because we need to build the OutgoingResponse when catching an exception. We build this OutgoingResponse to report the size of this response. So since we've built the response, we "return" it with sendResponse and don't rethrow.

pepone · 2024-03-11T09:55:41Z

cpp/src/Ice/OutgoingResponse.cpp

+            replyStatus = ReplyStatus::UnknownLocalException;
+            ostringstream str;
+            str << ex;
+            if (IceUtilInternal::printStackTraces)


Wonder if we should remove this setting, and always provide stack traces.

We should definitely not do that. Stack traces can provide sensitive information from the server to the client.

cpp/src/Ice/OutgoingResponse.cpp

pepone · 2024-03-11T10:03:48Z

cpp/src/Ice/OutgoingResponse.cpp

+    {
+        OutputStream ostr(current.adapter->getCommunicator(), Ice::currentProtocolEncoding);
+
+        if (current.requestId != 0)


Why do we need to call this method when current.requestId == 0? I suppose we still need to construct the OutgoingResponse even if we aren't sending it back to the client, to ensure it's available for use in the dispatch pipeline middleware. Is that correct?

Yes, that's correct. The response carries the reply status and error message (and more) and middleware (and dispatchers such as ServantManager) need it even for oneway request.

cpp/src/Ice/OutgoingResponse.cpp

externl

LGTM!

externl · 2024-03-11T13:02:33Z

cpp/include/Ice/AsyncResponseHandler.h

+            {
+                _sendResponse(makeEmptyOutgoingResponse(_current));
+            }
+            // else we ignore this call.


Maybe we should we debug assert these to make sure this never happens since it's otherwise silent.

externl · 2024-03-11T13:04:51Z

cpp/include/Ice/IncomingRequest.h

+    class InputStream;
+
+    /**
+     * Represent a request received by a connection. It's the argument to the dispatch function on Object.


Suggested change

* Represent a request received by a connection. It's the argument to the dispatch function on Object.

* Represents a request received by a connection. It's the argument to the dispatch function on Object.

We actually use this style (no s) for every other doc-comment, and I'd rather not update them all.

cpp/include/Ice/Object.h

externl · 2024-03-11T15:03:13Z

cpp/src/Ice/ObjectAdapterI.cpp

+    // _dispatcher is immutable, so no need to lock _mutex. There is no need to clear _dispatcher during destroy
+    // because _dispatcher does not hold onto this object adapter directly. It can hold onto a communicator that holds
+    // onto this object adapter, but the communicator will release this refcount when it is destroyed or when the
+    // object adapter is destroyed.


I think much of this comment belongs next to the declaration in the header file.

externl · 2024-03-11T15:18:02Z

cpp/src/Ice/OutgoingResponse.cpp

+    _replyStatus = other._replyStatus;
+    _exceptionId = std::move(other._exceptionId);
+    _errorMessage = std::move(other._errorMessage);
+    _outputStream.swap(other._outputStream);


If we implemented the move assignment operator on OuputStream could we then use the = default implementations for some these classes?

Yes, and it would be less error-prone. It's easy to forget to update all these functions when you add a new field to OutgoingResponse.

This reverts commit 464269a.

bentoi

I'm approving because the implementation looks corect. But while design is closer to IceRPC and cleaner, to me it doesn't fall into the category of a simple refactoring.

It introduces public APIs that are not totally safe to use (the application can now mess around with the input/output stream of the incoming request and outgoing response that contains the reply header). Some APIs are public but it's unclear whether or not there's really a use-cases for them.

It also makes the server-side and client-side inconsistent and now that we refactored the dispatch pipeline, I don't see why we wouldn't also refactor the client-side invocation pipeline later (not in 3.8).

Should we add a demo to show how to write a middleware? Do we need to add back a test like the the dispatch interceptor test?

bentoi · 2024-03-13T09:50:58Z

cpp/include/Ice/Object.h

+         * Dispatch an incoming request and return the corresponding outgoing response.
+         * @param request The incoming request.
+         * @param sendResponse A callback that the implementation calls to return the response. sendResponse does not
+         * throw any exception and any sendResponse wrapper must not throw any exception. sendResponse can be called by


It could be good to document what happens if the application doesn't respect the contract (raising exceptions and calling twice sendResponse).

With connections, it messes up with the dispatch count of the connection. Presumably, the result is the connection shutdown wouldn't wait for all outstanding dispatches (since the dispatch count is too low). I could not think of a simple way to describe this situation. It's also different for collocated invocations. When I accidentally called it twice, I was actually getting hangs at shutdown in some tests.

cpp/include/Ice/Object.h

cpp/include/Ice/IncomingRequest.h

cpp/include/Ice/ObjectAdapter.h

cpp/include/Ice/OutgoingResponse.h

bentoi · 2024-03-13T10:26:17Z

cpp/include/Ice/OutgoingResponse.h

+         * response.
+         * @param current A reference to the current object of the request.
+         */
+        OutgoingResponse(


I would make this constructor private since it's really not clear how the application could use it. Or we have to document that the given output stream should also contain the reply header.

Another option would be to get rid of the makeOutgoingReponse factory methods and like the IncomingRequest constructor is reading the reply header, the OutgoingResponse constructor would write the reply header.

I think this constructor should be public. The makeOutgoingResponse are helpers that the user could implement. The core functionality is this constructor.

bentoi · 2024-03-13T10:51:48Z

cpp/src/Ice/ObjectAdapterI.h

@@ -137,7 +137,8 @@ namespace Ice
        IceInternal::ObjectAdapterFactoryPtr _objectAdapterFactory;
        IceInternal::ThreadPoolPtr _threadPool;
        IceInternal::ACMConfig _acm;
-        IceInternal::ServantManagerPtr _servantManager;
+        const IceInternal::ServantManagerPtr _servantManager;
+        ObjectPtr _dispatcher;


We could use const_cast, that's what we do in other constructors.

cpp/src/Ice/Object.cpp

bentoi · 2024-03-13T11:27:57Z

cpp/include/Ice/AsyncResponseHandler.h

+    {
+    public:
+        // This class typically holds a _copy_ of the incoming request current object.
+        AsyncResponseHandler(std::function<void(Ice::OutgoingResponse)> sendResponse, Ice::Current current)


I probably miss something but I don't clearly understand why it's useful. Is it only about ensuring that we don't send twice the response? Could we have a internal flag on OutgoingResponse for this purpose instead?

Could we have a internal flag on OutgoingResponse for this purpose instead?

How would this work?

OutgoingResponse is a value that each thread can construct independently. Your response callback and exception callback could each construct an OutgoingResponse and pass it to sendResponse.

Refactor dispatch pipeline with the addition of a public dispatch(req…

1cd1e0e

…uest, sendResponse) on Object.

bernardnormier requested review from pepone, externl and bentoi March 10, 2024 21:23

bernardnormier commented Mar 10, 2024

View reviewed changes

bernardnormier added 4 commits March 10, 2024 17:43

Windows build fixes

f66a15e

Fix gcc build failure

6d7c7af

Remove ResponseSentException

281d5d4

Small fix to OutgoingResponse::size

2af6e74

pepone approved these changes Mar 11, 2024

View reviewed changes

bernardnormier added 2 commits March 11, 2024 09:22

Fix copilot hallucination

a296d14

Fix review comments

64423eb

externl approved these changes Mar 11, 2024

View reviewed changes

Work-around for Ruby test crash

6d9abc5

bernardnormier mentioned this pull request Mar 11, 2024

Ruby test crash on macos adding to value factory manager #1924

Open

bernardnormier added 2 commits March 11, 2024 21:56

Add support for inserting dispatcher in other OA

464269a

Revert "Add support for inserting dispatcher in other OA"

fdfa8e1

This reverts commit 464269a.

bernardnormier mentioned this pull request Mar 12, 2024

More dispatch pipeline refactoring #1925

Closed

bentoi approved these changes Mar 13, 2024

View reviewed changes

Fix review comments

c4618eb

bernardnormier merged commit ab839c8 into zeroc-ice:main Mar 13, 2024
24 checks passed

bernardnormier mentioned this pull request Mar 13, 2024

More dispatch pipeline refactoring #1929

Merged

bernardnormier deleted the object-dispatch branch May 10, 2024 23:43

	* Represent a request received by a connection. It's the argument to the dispatch function on Object.
	* Represents a request received by a connection. It's the argument to the dispatch function on Object.

Refactor dispatch pipeline in C++ #1923

Refactor dispatch pipeline in C++ #1923

Conversation

bernardnormier commented Mar 10, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pepone left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

externl left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bentoi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bernardnormier Mar 13, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bernardnormier Mar 13, 2024 •

edited

Loading