anton-vinogradov commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3649848845
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java:
##########
@@ -58,25 +58,25 @@ public class CacheContinuousQueryEntry implements
GridCacheDeployable, Marshalla
@GridToStringInclude
KeyCacheObject key;
- /** */
+ /** {@code null} for a filtered entry. */
@Order(2)
- byte[] keyBytes;
+ KeyCacheObject keyWire;
/** New value. */
@GridToStringInclude
CacheObject newVal;
- /** */
+ /** {@code null} for a filtered entry. */
@Order(3)
- byte[] newValBytes;
+ CacheObject newValWire;
/** Old value. */
@GridToStringInclude
CacheObject oldVal;
- /** */
+ /** {@code null} for a filtered entry. */
@Order(4)
- byte[] oldValBytes;
+ CacheObject oldValWire;
Review Comment:
Same as the thread at line 63: the master's local/wire pair (key +
keyBytes), only with a typed wire form.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java:
##########
@@ -359,35 +324,38 @@ CacheObject oldValue() {
return depInfo;
}
-
/** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(CacheContinuousQueryEntry.class, this);
+ @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
Review Comment:
These are the MarshallableMessage hooks; the contract is "populate the wire
form before send / restore the runtime form after receive", not necessarily
byte serialization. Other implementors do the same: QueryBatchMessage.marshal()
only wraps rows into mRows message objects, and GridDhtPartitionsFullMessage's
hook only does locParts = copyPartitionsMap(parts) — on master that same body
now lives in prepareMarshal(GridCacheSharedContext) after IGNITE-28914, still
with no serialization in it. On master this entry's hook produced byte[] only
because the wire form was a marshaller blob; with typed wire fields the step
degenerates to a reference transfer, but it is the same contract. Unlike
GridDhtPartitionsFullMessage this entry is not a GridCacheMessage, so there is
no cache-level hook to move this to.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java:
##########
@@ -359,35 +324,38 @@ CacheObject oldValue() {
return depInfo;
}
-
/** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(CacheContinuousQueryEntry.class, this);
+ @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
+ if (!isFiltered()) {
+ keyWire = key;
+ newValWire = newVal;
+ oldValWire = oldVal;
+ }
}
/** {@inheritDoc} */
- @Override public void prepareMarshal(Marshaller marsh) throws
IgniteCheckedException {
- if (!isFiltered()) {
- if (key != null)
- keyBytes = marsh.marshal(key);
+ @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
Review Comment:
Same as above — the unmarshal side of the same wire-prepare contract.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java:
##########
@@ -207,6 +208,19 @@ private void processRequest(final UUID nodeId, final
DataStreamerRequest req) {
}
}
+ // The generic receive pass skips this request (entries must stay
serialized for the update job and its
+ // peer-deployment loader), so the response topic is restored
here; it carries only internal classes.
+ try {
+ if (req.resTopicMsg != null)
+ MessageMarshalling.unmarshal(req.resTopicMsg, ctx);
Review Comment:
DataStreamerRequest is a DeferredUnmarshalMessage: the generic inbound pass
deliberately skips it (GridIoManager.unmarshalPayload), because the entries
must stay serialized until DataStreamerUpdateJob unmarshals them with the
peer-deployment loader. The flip side of that contract is that the consumer
itself unmarshals what it needs, and the response topic is needed before the
job is ever submitted — it is how deployment and updater-unmarshal failures are
reported back to the sender, cases where the job never runs. This is not
hand-rolled marshalling: MessageMarshalling.unmarshal(req.resTopicMsg, ctx) is
exactly the registered-marshaller dispatch the generic pass would have applied,
just scoped to the one nested message needed early. The alternatives look
worse: unmarshalling the whole request here defeats the deferred contract, and
turning resTopicMsg into a @NioField would move its deserialization onto NIO
threads.
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteMessageFactory.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.managers.communication;
+
+import java.util.function.Supplier;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheMessageDeployer;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.apache.ignite.plugin.extensions.communication.MessageMarshaller;
+import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Internal extension of {@link MessageFactory} that also registers and
resolves the marshal and deployment
+ * companions of a message. Kept out of the public factory interface:
marshallers and deployers are bound to
+ * kernal and cache internals ({@code GridKernalContext}, {@code
GridCacheSharedContext}) the public SPI module
+ * must not depend on.
+ *
+ * @see MessageMarshaller
+ * @see GridCacheMessageDeployer
+ */
+public interface IgniteMessageFactory<M extends Message, CM extends
GridCacheMessage> extends MessageFactory<M> {
+ /** {@inheritDoc} */
+ @Override default void register(short directType, Supplier<M> supplier,
MessageSerializer<M> serializer)
+ throws IgniteException {
+ register(directType, supplier, serializer, null, null);
+ }
+
+ /**
+ * Registers a message with the given direct type, serializer, and
marshaller. All messages must be registered
+ * during construction of the class that implements this interface.
+ *
+ * @param directType Direct type ({@link Message#directType()}) to
register the message under.
+ * @param supplier Message supplier.
+ * @param serializer Message serializer.
+ * @param marshaller Message marshaller, or {@code null} for
non-marshallable messages.
+ * @throws IgniteException If a message is already registered under the
given direct type.
+ */
+ default void register(short directType, Supplier<M> supplier,
MessageSerializer<M> serializer,
+ @Nullable MessageMarshaller<M> marshaller) throws IgniteException {
Review Comment:
A marshaller is not tied to MarshallableMessage here. Generated marshallers
exist for any message with marshalling work — CacheObject fields, nested
messages, @Marshalled fields: on this branch it is 195 generated *Marshaller
classes vs ~20 classes implementing MarshallableMessage (e.g. TestMessage in
the goldens is a plain Message with a generated marshaller).
MarshallableMessage only opts a message into the custom marshal(Marshaller)
hook, which the generated marshaller calls among its other statements. So an MM
extends MarshallableMessage bound would not typecheck for most registrations,
and since the registry is keyed by directType at runtime, the lookup cast in
MessageMarshalling.resolve stays regardless of the bound. Keeping <M extends
Message> as the accurate bound.
##########
modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageRegistry.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.plugin.extensions.communication;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.IgniteException;
+import
org.apache.ignite.internal.managers.communication.UnknownMessageException;
+
+/**
+ * Registry of message class to direct type mappings behind {@link
Message#directType()}, populated during factory
+ * initialization via {@link Message#registerAsDirectType(short)}. Kept
package-private so the map is mutable only
+ * through the validated registration path.
+ */
+final class MessageRegistry {
+ /** Message class to direct type mappings. */
+ private static final Map<Class<? extends Message>, Short> REGISTRATIONS =
new ConcurrentHashMap<>();
+
+ /** Per-class cache over {@link #REGISTRATIONS}; keeps {@link #directType}
off the map lookup done for every sent message. */
+ private static final ClassValue<Short> DIRECT_TYPES = new ClassValue<>() {
Review Comment:
REGISTRATIONS already is a ConcurrentHashMap; the ClassValue is not a map
substitute but a read cache in front of it. Message.directType() resolves
through it for every message written and read — writeHeader(msg.directType())
in every generated serializer plus the serializer lookup in
MessageSerialization.resolve, nested messages included. ClassValue stores the
entry on the Class itself with a constant-time fast path (no hashing) — exactly
the JDK primitive for per-class lookups; a ConcurrentMap.get on that path is
strictly more work. And the map alone cannot replace it in the other direction
either: register() needs putIfAbsent + conflict detection, which ClassValue
cannot express.
##########
modules/core/src/test/resources/codegen/TestMessageMarshaller.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.TestMessage;
+import org.apache.ignite.internal.managers.communication.MessageMarshalling;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.plugin.extensions.communication.MessageMarshaller;
+
+/**
+ * This class is generated automatically.
+ *
+ * @see org.apache.ignite.internal.MessageProcessor
+ */
+public class TestMessageMarshaller implements MessageMarshaller<TestMessage> {
+ /** */
+ @Override public void marshal(TestMessage msg, GridKernalContext kctx,
CacheObjectContext cacheObjCtx) throws IgniteCheckedException {
+ CacheObjectContext ctx = cacheObjCtx;
+
+ if (msg.keyCacheObject != null && ctx != null)
Review Comment:
Doable, but the win is narrow: the ctx guard is shareable only for adjacent
top-level CacheObject fields — 17 of 195 generated marshallers; in the common
collection/map shapes the guard lives inside per-element loops and cannot be
hoisted (and here in marshal() the third statement is not ctx-guarded, so a
method-level early-out doesn't work either). The generator currently emits
every field as one independent self-contained block; grouping adjacent fields
adds emitter branching — two output shapes for the same field kind depending on
neighbors — for a purely cosmetic change in machine-generated code. I'd keep
the uniform per-field emission.
##########
modules/nio/src/main/java/org/apache/ignite/internal/util/nio/MessageSerialization.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.util.nio;
Review Comment:
It sits next to its consumers: writeTo/readFrom are called from
GridNioServer and GridDirectParser — same org.apache.ignite.internal.util.nio
package of ignite-nio — plus DirectByteBufferStream and the discovery IO
sessions downstream. ignite-nio cannot see core, so the dispatcher must live in
the nio module, and within it util.nio is where its two primary callers are.
This mirrors MessageMarshalling, which lives in managers.communication next to
its consumer GridIoManager.
--
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]
