Vladsz83 commented on code in PR #13095: URL: https://github.com/apache/ignite/pull/13095#discussion_r3630826986
########## modules/core/src/test/resources/codegen/NioFieldOnNonMessageMessage.java: ########## @@ -0,0 +1,30 @@ +/* + * 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.plugin.extensions.communication.Message; + +public class NioFieldOnNonMessageMessage implements Message { Review Comment: Suggestion: `NioFieldOnFieldMessage` ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryCloseMessage.java: ########## @@ -18,13 +18,13 @@ package org.apache.ignite.internal.processors.query.calcite.message; import java.util.UUID; +import org.apache.ignite.internal.DeferredUnmarshalMessage; import org.apache.ignite.internal.Order; -import org.apache.ignite.plugin.extensions.communication.Message; /** * */ -public class QueryCloseMessage implements Message { +public class QueryCloseMessage implements DeferredUnmarshalMessage { Review Comment: Why it is a `DeferredUnmarshalMessage`? And many other who `implements DeferredUnmarshalMessage` ########## modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteErrorMessage.java: ########## @@ -18,11 +18,12 @@ package org.apache.ignite.internal.processors.query.calcite.message; import java.util.UUID; +import org.apache.ignite.internal.DeferredUnmarshalMessage; import org.apache.ignite.internal.Order; import org.apache.ignite.internal.managers.communication.ErrorMessage; /** */ -public class CalciteErrorMessage extends ErrorMessage { +public class CalciteErrorMessage extends ErrorMessage implements DeferredUnmarshalMessage { Review Comment: Why it is a `DeferredUnmarshalMessage`? ########## 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: Let's use just a `ConcurrentMap`? WDYT? ########## 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: Maybe shared `if( ctx != null )` ########## modules/core/src/test/resources/codegen/TestMarshallableMessageMarshallableSerializer.java: ########## @@ -108,13 +104,20 @@ public TestMarshallableMessageMarshallableSerializer(Marshaller marshaller, Clas reader.incrementState(); } - try { - msg.finishUnmarshal(marshaller, clsLdr); - } - catch (IgniteCheckedException e) { - throw new IgniteException("Failed to unmarshal object " + msg.getClass().getSimpleName(), e); - } - return true; } -} + + /** */ + @Override public void marshal(TestMarshallableMessage msg, GridKernalContext kctx, GridCacheContext<?, ?> nested) throws IgniteCheckedException { + GridCacheContext<?, ?> ctx = nested; Review Comment: Isn't actually used. Some other places too. ########## modules/core/src/test/resources/codegen/TestMarshalledCollectionMessage.java: ########## @@ -0,0 +1,34 @@ +/* + * 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 java.util.Set; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; +import org.apache.ignite.plugin.extensions.communication.Message; + +public class TestMarshalledCollectionMessage implements Message { Review Comment: Do we need tests/test messages for `Collection`/`Set`/`List`? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java: ########## @@ -71,6 +73,9 @@ public class GridIoMessage implements Message, SpanTransport, MessageWrapper { @GridToStringInclude public @Nullable OperationContextMessage opCtxMsg; + /** Set once the payload is marshalled; guards double marshal and unmarshalled transmit. Not on the wire. */ + private transient boolean marshalled; Review Comment: `transient` isnt required ########## 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: We should not manually marshall. Cat we revise the design? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java: ########## @@ -315,83 +305,18 @@ public boolean skipCompletedVersion() { return txLbl; } - /** - * {@inheritDoc} - * - * @param ctx - */ - @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException { - super.prepareMarshal(ctx); - + /** {@inheritDoc} */ + @Override public void deploy(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException { if (owned != null && ownedKeys == null) { - ownedKeys = owned.keySet(); - - ownedVals = owned.values(); - - for (IgniteTxKey key: ownedKeys) { + for (IgniteTxKey key : owned.keySet()) { GridCacheContext<?, ?> cctx = ctx.cacheContext(key.cacheId()); - key.prepareMarshal(cctx); - if (addDepInfo) - prepareObject(key, cctx); + deployObject(key, cctx); Review Comment: Why manually-deployed? No by a generated deployer ########## modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceSingleNodeDeploymentResult.java: ########## @@ -76,8 +77,14 @@ public Collection<Throwable> errors() { * @param errors Exceptions. */ public void errors(@Nullable Collection<Throwable> errors) { - if (!F.isEmpty(errors)) - this.errors = F.viewReadOnly(errors, ErrorMessage::new); Review Comment: @anton-vinogradov could you resolve this pls? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageMessage.java: ########## Review Comment: @anton-vinogradov ? ########## modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/Message.java: ########## @@ -32,19 +48,25 @@ public interface Message { /** Registry of message class to direct type mappings, populated during factory initialization. */ Map<Class<?>, Short> REGISTRATIONS = new ConcurrentHashMap<>(); Review Comment: @anton-vinogradov resolve pls. ########## 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: To be more clear, `MessageMarshaller` might use additional type parameter `MM implements MarshallableMessage` ########## 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: why the package is `nio`? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java: ########## @@ -1033,9 +1033,6 @@ private void map(Iterable<GridDhtCacheEntry> entries) { } assert added.dhtLocal(); - - if (added.ownerVersion() != null) Review Comment: Why changed? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java: ########## @@ -124,24 +125,14 @@ public Collection<GridCacheEntryInfo> preloadEntries() { } /** {@inheritDoc} */ - @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException { - super.prepareMarshal(ctx); + @Override public void deploy(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException { + if (preloadEntries != null) { + GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId); - GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId); - - if (preloadEntries != null) - marshalInfos(preloadEntries, cctx.shared(), cctx.cacheObjectContext()); - } - - /** {@inheritDoc} */ - @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException { - super.finishUnmarshal(ctx, ldr); - - if (preloadEntries != null) - unmarshalInfos(preloadEntries, ctx.cacheContext(cacheId), ldr); + deployInfos(preloadEntries, cctx); Review Comment: Why manually-deployed? Not by a generated deployer. ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java: ########## @@ -201,26 +191,8 @@ public int marshalledSize(CacheObjectContext ctx) throws IgniteCheckedException return SIZE_OVERHEAD + size; } - /** - * @param ctx Cache context. - * @throws IgniteCheckedException In case of error. - */ - public void marshal(GridCacheContext ctx) throws IgniteCheckedException { - marshal(ctx.cacheObjectContext()); - } - - /** - * @param ctx Cache context. - * @throws IgniteCheckedException In case of error. - */ - public void marshal(CacheObjectContext ctx) throws IgniteCheckedException { - assert key != null; - - key.prepareMarshal(ctx); - - if (val != null) - val.prepareMarshal(ctx); - + /** {@inheritDoc} */ + @Override public void marshal(Marshaller marsh) throws IgniteCheckedException { Review Comment: Good Place to fire a 'revise' ticket. Revise certain message logic or shared message preparation/finalization. -- 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]
