Vladsz83 commented on code in PR #13095: URL: https://github.com/apache/ignite/pull/13095#discussion_r3580270038
########## modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageMarshaller.java: ########## @@ -0,0 +1,154 @@ +/* + * 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 org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactory; +import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Handles {@code marshal}/{@code unmarshal} for a {@link Message} that requires custom serialization. + * + * @param <M> A message this marshaller handles. + */ +public interface MessageMarshaller<M extends Message> { + /** + * Marshals the message on the user thread before sending. + * + * @param msg Message to marshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + */ + public void marshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx) + throws IgniteCheckedException; + + /** + * Unmarshals the message with full cache context and class loader. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + * @param clsLdr Class loader for unmarshalling. + */ + public void unmarshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx, ClassLoader clsLdr) + throws IgniteCheckedException; + + /** + * Unmarshals the message without a cache context, using the configuration class loader — the cache-free receive + * path (e.g. the generic {@code GridIoManager} pass). Delegates to the cache-aware overload with a {@code null} + * context, so per-message marshallers need only implement the cache-aware method. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshal(M msg, GridKernalContext kctx) throws IgniteCheckedException { + unmarshal(msg, kctx, null, U.resolveClassLoader(kctx.config())); + } + + /** + * Unmarshals only {@code @NioField}-annotated fields in the NIO/IO thread. No-op by default. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshalNio(M msg, GridKernalContext kctx) throws IgniteCheckedException { + } + + /** + * Null-safe {@code unmarshalNio} — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. + * @param factory Message factory. + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + static <M extends Message> void unmarshalNio(MessageFactory factory, M msg, GridKernalContext kctx) + throws IgniteCheckedException { + MessageMarshaller<M> m = resolve(factory, msg); + + if (m != null) + m.unmarshalNio(msg, kctx); + } + + /** + * Null-safe {@code marshal} — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. + * @param factory Message factory. + * @param msg Message to marshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + */ + static <M extends Message> void marshal(MessageFactory factory, M msg, GridKernalContext kctx, + @Nullable CacheObjectContext cacheObjCtx) throws IgniteCheckedException { + MessageMarshaller<M> m = resolve(factory, msg); + + if (m != null) + m.marshal(msg, kctx, cacheObjCtx); + } + + /** + * Null-safe {@code unmarshal} — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. + * @param factory Message factory. + * @param msg Message to unmarshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + * @param clsLdr Class loader for unmarshalling. + */ + static <M extends Message> void unmarshal(MessageFactory factory, M msg, GridKernalContext kctx, + @Nullable CacheObjectContext cacheObjCtx, ClassLoader clsLdr) throws IgniteCheckedException { + assert !MessageUnmarshalDedup.ENABLED || MessageUnmarshalDedup.firstUnmarshal(msg, true) + : "Finish-unmarshalled more than once: " + msg.getClass().getName(); + + MessageMarshaller<M> m = resolve(factory, msg); + + if (m != null) + m.unmarshal(msg, kctx, cacheObjCtx, clsLdr); + } + + /** + * Null-safe {@code unmarshal} (cache-free) — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. + * @param factory Message factory. + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + static <M extends Message> void unmarshal(MessageFactory factory, M msg, GridKernalContext kctx) + throws IgniteCheckedException { + assert !MessageUnmarshalDedup.ENABLED || MessageUnmarshalDedup.firstUnmarshal(msg, false) + : "Finish-unmarshalled more than once: " + msg.getClass().getName(); + + MessageMarshaller<M> m = resolve(factory, msg); + + if (m != null) + m.unmarshal(msg, kctx); + } + + /** @return the marshaller registered for {@code msg}'s direct type, or {@code null} if none. */ + @SuppressWarnings("unchecked") Review Comment: If `{@code null} if none` let's declare `@Nullable MessageMarshaller<M>`. ########## modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageMarshaller.java: ########## @@ -0,0 +1,154 @@ +/* + * 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 org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactory; +import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Handles {@code marshal}/{@code unmarshal} for a {@link Message} that requires custom serialization. + * + * @param <M> A message this marshaller handles. + */ +public interface MessageMarshaller<M extends Message> { + /** + * Marshals the message on the user thread before sending. + * + * @param msg Message to marshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + */ + public void marshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx) + throws IgniteCheckedException; + + /** + * Unmarshals the message with full cache context and class loader. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + * @param clsLdr Class loader for unmarshalling. + */ + public void unmarshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx, ClassLoader clsLdr) + throws IgniteCheckedException; + + /** + * Unmarshals the message without a cache context, using the configuration class loader — the cache-free receive + * path (e.g. the generic {@code GridIoManager} pass). Delegates to the cache-aware overload with a {@code null} + * context, so per-message marshallers need only implement the cache-aware method. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshal(M msg, GridKernalContext kctx) throws IgniteCheckedException { + unmarshal(msg, kctx, null, U.resolveClassLoader(kctx.config())); + } + + /** + * Unmarshals only {@code @NioField}-annotated fields in the NIO/IO thread. No-op by default. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshalNio(M msg, GridKernalContext kctx) throws IgniteCheckedException { + } + + /** + * Null-safe {@code unmarshalNio} — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. + * @param factory Message factory. + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + static <M extends Message> void unmarshalNio(MessageFactory factory, M msg, GridKernalContext kctx) + throws IgniteCheckedException { + MessageMarshaller<M> m = resolve(factory, msg); + + if (m != null) + m.unmarshalNio(msg, kctx); + } + + /** + * Null-safe {@code marshal} — skips when no marshaller is registered. + * + * @param <M> A message this marshaller handles. Review Comment: Interferes with `Message to marshal` below. Let's say smth like `Message type`. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { Review Comment: Do we need a special unit-test for it? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass Review Comment: Suggestion: lets simplify and shorten. We can keep smth. like `Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once`. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ Review Comment: `(identity, pass)` -> `(hash, cacheMode)`? Using {@code}. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); Review Comment: Seems not be Ignite-style naming just `Q`. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ Review Comment: What? Let's rephrase more clean. ########## modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageMarshaller.java: ########## @@ -0,0 +1,154 @@ +/* + * 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 org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactory; +import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Handles {@code marshal}/{@code unmarshal} for a {@link Message} type that requires custom serialization. + * + * @param <M> Message type. + */ +public interface MessageMarshaller<M extends Message> { + /** + * Marshals the message on the user thread before sending. + * + * @param msg Message to marshal. Review Comment: True. But we already have tons of 'marshall'. `org.apache.ignite.marshaller`, `interface Marshaller`, `MarshallerContext`. And so on. Search for it pls. Let's not break the namings. This class is already names `MessageMarshaller`. ########## modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageMarshaller.java: ########## @@ -0,0 +1,154 @@ +/* + * 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 org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactory; +import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Handles {@code marshal}/{@code unmarshal} for a {@link Message} that requires custom serialization. + * + * @param <M> A message this marshaller handles. + */ +public interface MessageMarshaller<M extends Message> { + /** + * Marshals the message on the user thread before sending. + * + * @param msg Message to marshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + */ + public void marshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx) + throws IgniteCheckedException; + + /** + * Unmarshals the message with full cache context and class loader. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + * @param cacheObjCtx Cache object context of the enclosing message, or {@code null} at the top level. + * @param clsLdr Class loader for unmarshalling. + */ + public void unmarshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext cacheObjCtx, ClassLoader clsLdr) + throws IgniteCheckedException; + + /** + * Unmarshals the message without a cache context, using the configuration class loader — the cache-free receive + * path (e.g. the generic {@code GridIoManager} pass). Delegates to the cache-aware overload with a {@code null} + * context, so per-message marshallers need only implement the cache-aware method. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshal(M msg, GridKernalContext kctx) throws IgniteCheckedException { + unmarshal(msg, kctx, null, U.resolveClassLoader(kctx.config())); + } + + /** + * Unmarshals only {@code @NioField}-annotated fields in the NIO/IO thread. No-op by default. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshalNio(M msg, GridKernalContext kctx) throws IgniteCheckedException { + } + + /** + * Null-safe {@code unmarshalNio} — skips when no marshaller is registered. Review Comment: The description of what it actually does and how differ from another `unmarshal()s` . ########## 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: Double unmarshal look wierd. Why? Or maybe a comment here? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { Review Comment: `Dedup` -> `Deduplicator`? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ + private final boolean cacheMode; + + /** + * @param msg Tracked message. + * @param cacheMode Unmarshal pass. + */ + IdRef(Message msg, boolean cacheMode) { + super(msg, Q); + + this.cacheMode = cacheMode; + hash = 31 * System.identityHashCode(msg) + (cacheMode ? 1 : 0); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return hash; + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (!(o instanceof IdRef)) + return false; + + IdRef ref = (IdRef)o; + + Message m = get(); Review Comment: Do we need `Message m = get();`? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ + private final boolean cacheMode; + + /** + * @param msg Tracked message. + * @param cacheMode Unmarshal pass. + */ + IdRef(Message msg, boolean cacheMode) { Review Comment: private? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ + private final boolean cacheMode; + + /** + * @param msg Tracked message. + * @param cacheMode Unmarshal pass. + */ + IdRef(Message msg, boolean cacheMode) { + super(msg, Q); + + this.cacheMode = cacheMode; + hash = 31 * System.identityHashCode(msg) + (cacheMode ? 1 : 0); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return hash; + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (!(o instanceof IdRef)) Review Comment: `o instanceof IdRef` -> `o instanceof IdRef ref`. And use `ref` below. ########## 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: Let's human-write this comment. ########## modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageMarshaller.java: ########## @@ -0,0 +1,154 @@ +/* + * 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 org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactory; +import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Handles {@code marshal}/{@code unmarshal} for a {@link Message} type that requires custom serialization. + * + * @param <M> Message type. + */ +public interface MessageMarshaller<M extends Message> { + /** + * Marshals the message on the user thread before sending. + * + * @param msg Message to marshal. + * @param kctx Kernal context. + * @param nested Cache object context, or {@code null} if not applicable. + */ + public void marshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext nested) + throws IgniteCheckedException; + + /** + * Unmarshals the message with full cache context and class loader. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + * @param nested Cache object context, or {@code null} if not applicable. + * @param clsLdr Class loader for unmarshalling. + */ + public void unmarshal(M msg, GridKernalContext kctx, @Nullable CacheObjectContext nested, ClassLoader clsLdr) + throws IgniteCheckedException; + + /** + * Unmarshals the message without a cache context, using the configuration class loader — the cache-free receive + * path (e.g. the generic {@code GridIoManager} pass). Delegates to the cache-aware overload with a {@code null} + * context, so per-message marshallers need only implement the cache-aware method. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshal(M msg, GridKernalContext kctx) throws IgniteCheckedException { + unmarshal(msg, kctx, null, U.resolveClassLoader(kctx.config())); + } + + /** + * Unmarshals only {@code @NioField}-annotated fields in the NIO/IO thread. No-op by default. + * + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + default void unmarshalNio(M msg, GridKernalContext kctx) throws IgniteCheckedException { + } + + /** + * Null-safe {@code unmarshalNio} — skips when no marshaller is registered. + * + * @param <M> Message type. + * @param factory Message factory. + * @param msg Message to unmarshal. + * @param kctx Kernal context. + */ + static <M extends Message> void unmarshalNio(MessageFactory factory, M msg, GridKernalContext kctx) Review Comment: I know. Let's not stop with `wont compile`. We can rename or refactor as we wish. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { Review Comment: Do we need also a `MessageMarshalDedup`? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ + private final boolean cacheMode; + + /** + * @param msg Tracked message. + * @param cacheMode Unmarshal pass. + */ + IdRef(Message msg, boolean cacheMode) { + super(msg, Q); + + this.cacheMode = cacheMode; + hash = 31 * System.identityHashCode(msg) + (cacheMode ? 1 : 0); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return hash; + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (!(o instanceof IdRef)) + return false; + + IdRef ref = (IdRef)o; + + Message m = get(); + + return m != null && m == ref.get() && cacheMode == ref.cacheMode; Review Comment: What if `ref.get()` is also null? ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { + /** + * When {@code true}, the no-double-unmarshal check runs. {@code static final} so the JIT folds the guard away + * in production (even with assertions on); enabled only by tests via {@code IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK}. + */ + public static final boolean ENABLED = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_MESSAGE_UNMARSHAL_ONCE_CHECK); + + /** Queue of collected referents, drained on each call to evict stale {@link IdRef}s from {@link #SEEN}. */ + private static final ReferenceQueue<Message> Q = new ReferenceQueue<>(); + + /** Finish-unmarshalled instances, held weakly and keyed by identity so they vanish with the message. */ + private static final Set<IdRef> SEEN = ConcurrentHashMap.newKeySet(); + + /** */ + private MessageUnmarshalDedup() { + // No-op. + } + + /** + * @param msg Message about to be finish-unmarshalled. + * @param cacheMode {@code true} for the cache-aware pass, {@code false} for the cache-free pass; the two passes + * over one message are legitimate and tracked separately, so only a repeat of the same pass is reported. + * @return {@code true} if {@code msg} is not a {@link MarshallableMessage} or is finish-unmarshalled the first + * time in this pass. + */ + public static boolean firstUnmarshal(Message msg, boolean cacheMode) { + if (!(msg instanceof MarshallableMessage)) + return true; + + for (Reference<? extends Message> r; (r = Q.poll()) != null; ) + SEEN.remove(r); + + return SEEN.add(new IdRef(msg, cacheMode)); + } + + /** Weak reference to a message keyed by (identity, pass), so distinct messages and the two passes stay distinct. */ + private static final class IdRef extends WeakReference<Message> { + /** Referent identity hash folded with the pass, captured up front since the referent may be cleared later. */ + private final int hash; + + /** Unmarshal pass: cache-aware vs cache-free. Keeps the two legitimate passes over one message distinct. */ + private final boolean cacheMode; + + /** + * @param msg Tracked message. + * @param cacheMode Unmarshal pass. + */ + IdRef(Message msg, boolean cacheMode) { + super(msg, Q); Review Comment: Why we need this queue? I've removed it: `super(msg);`. Test `MessageUnmarshalOnceTest` passes. ########## modules/core/src/main/java/org/apache/ignite/internal/managers/communication/MessageUnmarshalDedup.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; + +/** + * Detects a {@link MarshallableMessage} instance being finish-unmarshalled more than once within the same pass + * (cache-aware or cache-free) — a class-loader or receive-path bug. The two passes over one message (e.g. the + * generic {@code GridIoManager} pass plus a subsystem's cache-aware pass) are legitimate and tracked separately. + * Gated by {@link #ENABLED}, so it runs only under tests and is folded away in production. + * + * @see MessageMarshaller + */ +public class MessageUnmarshalDedup { Review Comment: Can we add this in a separate ticket? After or before. As a subticket with a prepared patch? -- 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]
