anton-vinogradov commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3589127975
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -1241,6 +1264,10 @@ private void onMessage0(UUID nodeId, GridIoMessage msg,
IgniteRunnable msgC) {
}
}
+ // Deliberately below the waitMap gate: replayed delayed messages
pass through this method twice,
Review Comment:
Reworded — now: "Kept after the delayed-message gate: a message held back
and later replayed enters this method a second time, and the NIO-thread
unmarshal of its routing header must happen once, on the replay only."
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -1175,10 +1188,20 @@ private void onChannelOpened0(UUID rmtNodeId,
GridIoMessage initMsg, Channel cha
byte plc = initMsg.policy();
+ MessageMarshaller.unmarshalNio(ctx.messageFactory(), initMsg, ctx);
+
pools.poolForPolicy(plc).execute(new Runnable() {
@Override public void run() {
- processOpenedChannel(initMsg.topic(), rmtNodeId,
(SessionChannelMessage)initMsg.message(),
- (SocketChannel)channel);
+ try {
+ MessageMarshaller.unmarshal(ctx.messageFactory(),
initMsg, ctx);
Review Comment:
Added a comment — it's a two-stage unmarshal, not a duplicate: the
`@NioField` routing header is restored on the NIO thread so the message can be
dispatched, then the full payload is restored on a pool thread. The two touch
disjoint fields.
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -1450,11 +1468,23 @@ private void processRegularMessage0(GridIoMessage msg,
UUID nodeId) {
if (lsnr == null)
return;
- Object obj = msg.message();
+ unmarshalPayload(msg);
+
+ invokeListener(msg.policy(), lsnr, nodeId, msg.message());
+ }
- assert obj != null;
+ /** */
+ private void unmarshalPayload(GridIoMessage msg) {
+ // Unmarshalled by GridCacheIoManager with the deployment loader; the
loader here can't see its peer classes.
Review Comment:
Linked it — the comment now names `GridCacheIoManager.unmarshall` and
`GridCacheDeploymentManager#globalLoader` (the peer-deployment loader).
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -1450,11 +1468,23 @@ private void processRegularMessage0(GridIoMessage msg,
UUID nodeId) {
if (lsnr == null)
return;
- Object obj = msg.message();
+ unmarshalPayload(msg);
+
+ invokeListener(msg.policy(), lsnr, nodeId, msg.message());
+ }
- assert obj != null;
+ /** */
+ private void unmarshalPayload(GridIoMessage msg) {
+ // Unmarshalled by GridCacheIoManager with the deployment loader; the
loader here can't see its peer classes.
+ if (msg.message() instanceof GridCacheMessage)
Review Comment:
Expanded the comment: cache messages are unmarshalled later by
`GridCacheIoManager` with the peer-deployment loader; the generic pass here
only has the configuration class loader, which can't see peer-loaded classes —
so it must skip them.
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -2011,25 +2042,102 @@ else if (async)
ackC.apply(null);
}
else {
+ MessageMarshaller.marshal(ctx.messageFactory(), ioMsg, ctx,
null);
+
+ sendMarshalled(node, ioMsg, ackC);
+ }
+ }
+ }
+
+ /**
+ * Wraps {@code msg} into a marshalled {@link GridIoMessage} without
sending it. Message marshalling is not
+ * idempotent (see {@code MessageMarshalOnceTest}), so a caller that
retries transmission must prepare the
+ * message once and re-send it via {@link #sendPrepared} on each attempt.
+ */
+ public GridIoMessage prepare(Object topic, Message msg, byte plc, boolean
ordered, long timeout,
+ boolean skipOnTimeout) throws IgniteCheckedException {
+ assert !ordered || timeout > 0 || skipOnTimeout;
+
+ GridIoMessage ioMsg = createGridIoMessage(topic, msg, plc, ordered,
timeout, skipOnTimeout);
+
+ MessageMarshaller.marshal(ctx.messageFactory(), ioMsg, ctx, null);
+
+ return ioMsg;
+ }
+
+ /**
+ * Transmits a message created by {@link #prepare} to a remote node.
Transmission does not mutate the message,
+ * so it can be safely retried.
+ */
+ public void sendPrepared(ClusterNode node, GridIoMessage ioMsg) throws
IgniteCheckedException {
+ assert !locNodeId.equals(node.id()) : node;
Review Comment:
Added — `GridIoMessage` now carries a transient `prepared` flag set by
`prepare()`, and `sendPrepared()` asserts it (`assert ioMsg.prepared()`).
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java:
##########
@@ -1942,6 +1972,8 @@ private IgniteInternalFuture<Channel> openChannel(
false
);
+ MessageMarshaller.marshal(ctx.messageFactory(), ioMsg, ctx, null);
Review Comment:
It's needed here: `TcpCommunicationSpi.openChannel` serializes this `ioMsg`
(the `SessionChannelMessage`) straight into the channel handshake, so it must
be marshalled before that call — there's no later point on this path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
