Vladsz83 commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3587050181
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageMessage.java:
##########
Review Comment:
Do we need any TODO now over this? Like `revise the filters marshalling`?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java:
##########
@@ -741,7 +743,7 @@ else if (cacheMsg instanceof
GridNearAtomicCheckUpdateRequest)
* @param plc Message policy.
* @throws IgniteCheckedException If failed.
*/
- private void processFailedMessage(UUID nodeId,
+ public void processFailedMessage(UUID nodeId,
Review Comment:
Minor codestyle suggestion: `UUID nodeId,` to the next line.
##########
modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/CacheIdAware.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.plugin.extensions.communication;
+
+/**
+ * Implemented by messages that carry cache data ({@code CacheObject}s) to
transfer. The cache ID lets the generated
Review Comment:
Suggestion: `{@code CacheObject}s` -> ``{@code CacheObject}
##########
modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java:
##########
@@ -1318,8 +1318,8 @@ else if (e instanceof IgniteCheckedException)
startTimer.finishGlobalStage("Await exchange");
}
- /** */
- private void initMessageFactory() throws IgniteCheckedException {
+ /** Public for a cross-package test; the production call is {@link #start}
below. */
+ public void initMessageFactory() throws IgniteCheckedException {
Review Comment:
We might add special `public` method with smth. like `@TestOnly`.
##########
modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageUnmarshalOnceTest.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 org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.managers.communication.MessageUnmarshalDedup;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Verifies the no-double-unmarshal check itself, not its coverage: {@link
MessageUnmarshalDedup#firstUnmarshal} must
+ * detect a second finish-unmarshal of the same instance within one pass,
while allowing the two legitimate passes
+ * (cache-free and cache-aware), and the check must be enabled for every test
— otherwise the suite-wide guard would
+ * silently turn off and pass every test vacuously. The actual coverage (no
real receive path unmarshals an instance
+ * twice in the same pass) comes from running the check across the whole suite
via {@code GridAbstractTest}.
+ */
+public class MessageUnmarshalOnceTest extends GridCommonAbstractTest {
+ /** The suite-wide guard must be on, so a silently-disabled check cannot
pass every test without verifying anything. */
+ @Test
+ public void testCheckEnabled() {
Review Comment:
MIght be `beforeTestStarted` with an `assert`. WDYT?
##########
modules/codegen/src/main/java/org/apache/ignite/internal/MarshalledCollection.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/** Marks a {@code Collection} field whose wire form is a companion {@code
@Order} array field named by {@link #value()}. */
+@Retention(RetentionPolicy.CLASS)
+@Target(ElementType.FIELD)
+public @interface MarshalledCollection {
Review Comment:
Let's do as for `@Order`. It is gracefully used for all the message's field
types: arrays, collections, lists, sets, maps and so on. Can we? I've prepared
a
[patch](https://github.com/user-attachments/files/30059225/MarshallableColl.patch).
It is a scratch with the general idea. At least
`MessageProcessorTest#testMarshalledCollectionMessage()` works if was previousy
assembled from the master branch.
##########
modules/codegen/src/main/java/org/apache/ignite/internal/MessageMarshallerGenerator.java:
##########
@@ -0,0 +1,1073 @@
+/*
+ * 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.io.IOException;
+import java.io.StringWriter;
+import java.io.Writer;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import javax.annotation.processing.ProcessingEnvironment;
+import javax.lang.model.element.Element;
+import javax.lang.model.element.Modifier;
+import javax.lang.model.element.QualifiedNameable;
+import javax.lang.model.element.TypeElement;
+import javax.lang.model.element.VariableElement;
+import javax.lang.model.type.ArrayType;
+import javax.lang.model.type.DeclaredType;
+import javax.lang.model.type.TypeKind;
+import javax.lang.model.type.TypeMirror;
+import javax.lang.model.type.TypeVariable;
+import javax.lang.model.util.ElementFilter;
+import javax.tools.Diagnostic;
+import
org.apache.ignite.internal.systemview.SystemViewRowAttributeWalkerProcessor;
+
+import static
org.apache.ignite.internal.MessageProcessor.MARSHALLABLE_MESSAGE_INTERFACE;
+import static org.apache.ignite.internal.MessageProcessor.MESSAGE_INTERFACE;
+import static
org.apache.ignite.internal.MessageProcessor.NON_MARSHALLABLE_MESSAGE_INTERFACE;
+
+/**
+ * Generates {@code *Marshaller} classes for {@code Message} types that are
not {@code NonMarshallableMessage}.
+ */
+public class MessageMarshallerGenerator extends MessageGenerator {
+ /** Accumulated source lines for all generated marshal/unmarshal methods.
*/
+ private final List<String> marshall = new ArrayList<>();
+
+ /** */
+ private final TypeMirror marshallableMsgType;
+
+ /** */
+ private final TypeMirror messageMirror;
+
+ /** */
+ private final TypeMirror cacheObjectMirror;
+
+ /** */
+ private final TypeMirror nonMarshallableMirror;
+
+ /** */
+ private final TypeMirror cacheGroupIdMsgMirror;
+
+ /** */
+ private final TypeMirror mapMirror;
+
+ /** */
+ private final TypeMirror collectionMirror;
+
+ /** */
+ private boolean marshallable;
+
+ /** */
+ private boolean hasMarshalled;
+
+ /** Whether any generated method got a non-empty body; a marshaller
without one is skipped entirely. */
+ private boolean hasStatements;
+
+ /** Enclosed fields of the currently processed type. Computed once per
{@link #generateBody} call. */
+ private Map<String, VariableElement> enclosed;
+
+ /** Nesting depth of the current marshal for-loop; names loop variables
{@code e}, {@code e1}, {@code e2}… */
+ private int loopDepth;
+
+ /** */
+ MessageMarshallerGenerator(ProcessingEnvironment env) {
+ super(env);
+
+ marshallableMsgType = type(MARSHALLABLE_MESSAGE_INTERFACE);
+ messageMirror = type(MESSAGE_INTERFACE);
+ cacheObjectMirror =
type("org.apache.ignite.internal.processors.cache.CacheObject");
+ nonMarshallableMirror = type(NON_MARSHALLABLE_MESSAGE_INTERFACE);
+ cacheGroupIdMsgMirror =
type("org.apache.ignite.internal.processors.cache.GridCacheGroupIdMessage");
+ mapMirror = type("java.util.Map");
+ collectionMirror = type("java.util.Collection");
+ }
+
+ /** {@inheritDoc} */
+ @Override String typeSuffix() {
+ return "Marshaller";
+ }
+
+ /** {@inheritDoc} */
+ @Override boolean shouldSkip(TypeElement type, List<VariableElement>
fields) {
+ return isNonMarshallableMessage(type);
+ }
+
+ /** {@inheritDoc} */
+ @Override void generateBody(List<VariableElement> fields) throws Exception
{
+ enclosed = enclosedFields();
+ marshallable = marshallableMsgType != null &&
assignableFrom(type.asType(), marshallableMsgType);
+ hasMarshalled = enclosed.values().stream().anyMatch(f ->
+ f.getAnnotation(Marshalled.class) != null ||
f.getAnnotation(MarshalledObjects.class) != null);
+
+ generateMarshalMethod(fields);
+ generateUnmarshalMethods(fields);
+ }
+
+ /** {@inheritDoc} */
+ @Override String buildClassCode(String marshallerClsName) throws
IOException {
+ if (!hasStatements)
+ return null;
+
+ try (Writer writer = new StringWriter()) {
+ imports.add(type.toString());
+
imports.add("org.apache.ignite.plugin.extensions.communication.MessageMarshaller");
+
+ if (marshallable || hasMarshalled)
+ imports.add("org.apache.ignite.marshaller.Marshaller");
+
+ writeClassHeader(writer, "MessageMarshaller", marshallerClsName);
+
+ writer.write(" {" + NL);
+
+ writeConstructor(writer, marshallerClsName);
+
+ for (String line : marshall)
+ writer.write(line + NL);
+
+ writer.write("}");
+
+ return writer.toString();
+ }
+ }
+
+ /** Writes the {@code marshaller} field and the constructor initializing
it, when the marshaller is needed. */
+ private void writeConstructor(Writer writer, String marshallerClsName)
throws IOException {
+ if (!marshallable && !hasMarshalled)
+ return;
+
+ writer.write(indentedLine(METHOD_JAVADOC));
+ writer.write(NL);
+ writer.write(indentedLine("private final Marshaller marshaller;"));
+ writer.write(NL + NL);
+
+ writer.write(indentedLine(METHOD_JAVADOC));
+ writer.write(NL);
+ writer.write(indentedLine("public " + marshallerClsName + "(Marshaller
marshaller) {"));
+ writer.write(NL);
+
+ indent++;
+
+ writer.write(indentedLine("this.marshaller = marshaller;"));
+ writer.write(NL);
+
+ indent--;
+
+ writer.write(indentedLine("}"));
+ writer.write(NL + NL);
+ }
+
+ /** Generates the {@code marshal} method body and appends it to {@link
#marshall}. */
+ private void generateMarshalMethod(List<VariableElement> orderedFields) {
+ imports.add("org.apache.ignite.IgniteCheckedException");
+ imports.add("org.apache.ignite.internal.GridKernalContext");
+
imports.add("org.apache.ignite.internal.processors.cache.CacheObjectContext");
+
+ String signature = "marshal(" + simpleNameWithGeneric(type) + " msg,
GridKernalContext kctx, CacheObjectContext cacheObjCtx)";
+
+ hasStatements |= emitMethod(marshall, signature, body -> {
+ if (needsCtx(orderedFields))
+ appendBlock(body, List.of(ctxResolutionLine()));
+
+ appendMarshalledFieldsPrepare(body);
+ appendMarshalledPrepare(body);
+
+ if (marshallable)
+ appendBlock(body,
List.of(indentedLine("msg.marshal(marshaller);")));
+
+ appendFields(body, orderedFields, MarshalMode.MARSHAL);
+ });
+ }
+
+ /** Generates all {@code unmarshal} overloads and appends them to {@link
#marshall}. */
+ private void generateUnmarshalMethods(List<VariableElement> orderedFields)
{
+ List<VariableElement> nioFields = new ArrayList<>();
+ List<VariableElement> workerFields = new ArrayList<>();
+
+ for (VariableElement f : orderedFields) {
+ boolean nioField = isNioField(f);
+
+ if (nioField && isMessage(f.asType()) &&
!nestedNeedsCtx(f.asType()))
+ nioFields.add(f);
+ else {
+ if (nioField && !isMessage(f.asType()))
+ env.getMessager().printMessage(Diagnostic.Kind.ERROR,
+ "@NioField has no effect on non-Message field '" +
f.getSimpleName() + "' of type " + f.asType(), f);
+ else if (nioField)
+ env.getMessager().printMessage(Diagnostic.Kind.ERROR,
+ "@NioField field '" + f.getSimpleName() + "' of type "
+ f.asType() + " needs a cache object " +
+ "context to unmarshal, but the NIO thread has
none; only context-free messages may be @NioField", f);
+
+ workerFields.add(f);
+ }
+ }
+
+ // U is referenced only by
@Marshalled/@MarshalledObjects/@MarshalledCollection/@MarshalledMap handling.
+ boolean usesU = hasMarshalled || enclosed.values().stream().anyMatch(f
->
+ f.getAnnotation(MarshalledCollection.class) != null ||
f.getAnnotation(MarshalledMap.class) != null);
+
+ if (usesU)
+ imports.add("org.apache.ignite.internal.util.typedef.internal.U");
+
+ String msgParam = simpleNameWithGeneric(type) + " msg,
GridKernalContext kctx";
+
+ generateUnmarshalMethod(msgParam + ", CacheObjectContext cacheObjCtx,
ClassLoader clsLdr", workerFields);
+
+ if (!nioFields.isEmpty())
+ generateUnmarshalNioMethod(msgParam, nioFields);
+ }
+
+ /** Generates the cache-aware {@code unmarshal} overload: the full field
set, with cache context and deployment class loader. */
+ private void generateUnmarshalMethod(String params, List<VariableElement>
fields) {
+ hasStatements |= emitMethod(marshall, "unmarshal(" + params + ")",
body -> {
+ Set<String> wireFieldSkip = marshalledWireFieldsToSkip();
+
+ if (needsCtx(fields) || !wireFieldSkip.isEmpty())
+ appendBlock(body, List.of(ctxResolutionLine()));
+
+ appendFields(body, fields, MarshalMode.UNMARSHAL, wireFieldSkip);
+
+ if (marshallable)
+ appendBlock(body,
List.of(indentedLine("msg.unmarshal(marshaller, clsLdr);")));
+
+ appendMarshalledFinish(body);
+
+ appendMarshalledCollectionFinish(body);
+ appendMarshalledMapFinish(body);
+ appendMarshalledObjectsFinish(body);
+ });
+ }
+
+ /** Generates the {@code unmarshalNio} method for NIO-eligible {@code
@Message} fields. */
+ private void generateUnmarshalNioMethod(String params,
List<VariableElement> nioFields) {
+ hasStatements |= emitMethod(marshall, "unmarshalNio(" + params + ")",
body -> {
+ for (VariableElement f : nioFields)
+ appendBlock(body, unmarshalNioField(fieldAccessor(f)));
+ });
+ }
+
+ /** Cache-free unmarshal of a {@code @NioField} message field on the NIO
thread (no cache context available). */
+ private List<String> unmarshalNioField(String accessor) {
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null)", accessor));
+
+ indent++;
+
+
code.add(indentedLine("MessageMarshaller.unmarshal(kctx.messageFactory(), %s,
kctx);", accessor));
+
+ indent--;
+
+ return code;
+ }
+
+ /** Generates logical→wire conversions for all {@code
@MarshalledCollection} and {@code @MarshalledMap} fields. */
+ private void appendMarshalledFieldsPrepare(List<String> body) {
+ for (VariableElement field : enclosed.values()) {
+ appendCollectionPrepare(body, field);
+ appendMapPrepare(body, field);
+ appendObjectsPrepare(body, field);
+ }
+ }
+
+ /** Generates the {@code Collection<byte[]>} build-up for a {@code
@MarshalledObjects} field in marshal. */
+ private void appendObjectsPrepare(List<String> body, VariableElement
field) {
+ MarshalledObjects ann = field.getAnnotation(MarshalledObjects.class);
+
+ if (ann == null)
+ return;
+
+ String objField = "msg." + field.getSimpleName();
+ String bytesField = "msg." + ann.value();
+
+ imports.add("java.util.ArrayList");
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null && %s == null) {", objField,
bytesField));
+
+ indent++;
+
+ code.add(indentedLine("%s = new ArrayList<>(%s.size());", bytesField,
objField));
+ code.add(EMPTY);
+ code.add(indentedLine("for (Object e : %s)", objField));
+
+ indent++;
+
+ code.add(indentedLine("%s.add(U.marshal(marshaller, e));",
bytesField));
+
+ indent--;
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ appendBlock(body, code);
+ }
+
+ /** Appends a {@code toArray} assignment for a {@code
@MarshalledCollection} field, if present. */
+ private void appendCollectionPrepare(List<String> body, VariableElement
field) {
+ MarshalledCollection ann =
field.getAnnotation(MarshalledCollection.class);
+
+ if (ann == null)
+ return;
+
+ String colField = "msg." + field.getSimpleName();
+ String arrField = "msg." + ann.value();
+ String compName = arrayComponentName(requireEnclosed(enclosed,
ann.value(), "@MarshalledCollection"));
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null && %s == null)", colField,
arrField));
+
+ indent++;
+
+ code.add(indentedLine("%s = %s.toArray(new %s[0]);", arrField,
colField, compName));
+
+ indent--;
+
+ appendBlock(body, code);
+ }
+
+ /** Appends key/value array assignments for a {@code @MarshalledMap}
field, if present. */
+ private void appendMapPrepare(List<String> body, VariableElement field) {
+ MarshalledMap ann = field.getAnnotation(MarshalledMap.class);
+
+ if (ann == null)
+ return;
+
+ String mapField = "msg." + field.getSimpleName();
+ String keysField = "msg." + ann.keys();
+ String valuesField = "msg." + ann.values();
+ VariableElement keysEl = requireEnclosed(enclosed, ann.keys(),
"@MarshalledMap");
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null && %s == null) {", mapField,
keysField));
+
+ indent++;
+
+ code.addAll(keysEl.asType().getKind() == TypeKind.ARRAY
+ ? arrayMapBody(ann, mapField, keysField, keysEl, valuesField)
+ : viewBasedMapBody(keysField, mapField, valuesField));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ appendBlock(body, code);
+ }
+
+ /** Generates {@code U.marshal} calls for all {@code @Marshalled} fields
in marshal. */
+ private void appendMarshalledPrepare(List<String> body) {
+ forEachMarshalled((bytesAcc, objAcc) -> {
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null && %s == null)", objAcc,
bytesAcc));
+
+ indent++;
+
+ code.add(indentedLine("%s = U.marshal(marshaller, %s);", bytesAcc,
objAcc));
+
+ indent--;
+
+ return code;
+ }, body);
+ }
+
+ /** Generates {@code U.unmarshal} calls for all {@code @Marshalled} fields
in the cache-aware unmarshal. */
+ private void appendMarshalledFinish(List<String> body) {
+ forEachMarshalled((bytesAcc, objAcc) -> {
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null) {", bytesAcc));
+
+ indent++;
+
+ code.add(indentedLine("%s = U.unmarshal(marshaller, %s, clsLdr);",
objAcc, bytesAcc));
+ code.add(EMPTY);
+
+ // Drop the serialized cache once the object is restored: keeping
both the deserialized value and its bytes
+ // on every received message doubles retained memory (e.g.
topology history nodes) and can exhaust the heap.
+ code.add(indentedLine("%s = null;", bytesAcc));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ return code;
+ }, body);
+ }
+
+ /** Generates Set reconstruction for all {@code @MarshalledCollection}
fields. */
+ private void appendMarshalledCollectionFinish(List<String> body) {
+ for (VariableElement field : enclosed.values()) {
+ MarshalledCollection colAnn =
field.getAnnotation(MarshalledCollection.class);
+
+ if (colAnn == null)
+ continue;
+
+ String colField = "msg." + field.getSimpleName();
+ String arrField = "msg." + colAnn.value();
+ VariableElement wireField = requireEnclosed(enclosed,
colAnn.value(), "@MarshalledCollection");
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null) {", arrField));
+
+ indent++;
+
+ code.add(indentedLine("%s = U.newHashSet(%s.length);", colField,
arrField));
+ code.add(EMPTY);
+ code.addAll(collectionFinishForBlock(wireField, colField,
arrField, field.getSimpleName().toString()));
+ code.add(EMPTY);
+ code.add(indentedLine("%s = null;", arrField));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ appendBlock(body, code);
+ }
+ }
+
+ /** Generates Collection reconstruction for all {@code @MarshalledObjects}
fields (cache-aware pass only). */
+ private void appendMarshalledObjectsFinish(List<String> body) {
+ for (VariableElement field : enclosed.values()) {
+ MarshalledObjects ann =
field.getAnnotation(MarshalledObjects.class);
+
+ if (ann == null)
+ continue;
+
+ String objField = "msg." + field.getSimpleName();
+ String bytesField = "msg." + ann.value();
+
+ imports.add("java.util.ArrayList");
+ imports.add("java.util.Map");
+
imports.add("org.apache.ignite.internal.processors.cache.KeyCacheObject");
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("if (%s != null) {", bytesField));
+
+ indent++;
+
+ code.add(indentedLine("%s = new ArrayList<>(%s.size());",
objField, bytesField));
+ code.add(EMPTY);
+ code.add(indentedLine("for (byte[] e : %s) {", bytesField));
+
+ indent++;
+
+ code.add(indentedLine("Object o = U.unmarshal(marshaller, e,
clsLdr);"));
+ code.add(EMPTY);
+ code.add(indentedLine("if (o instanceof Map.Entry) {"));
+
+ indent++;
+
+ code.add(indentedLine("Object key = ((Map.Entry<?,
?>)o).getKey();"));
+ code.add(EMPTY);
+ code.add(indentedLine("if (key instanceof KeyCacheObject)"));
+
+ indent++;
+
+ code.add(indentedLine("((KeyCacheObject)key).unmarshal(ctx,
clsLdr);"));
+
+ indent--;
+ indent--;
+
+ code.add(indentedLine("}"));
+ code.add(EMPTY);
+ code.add(indentedLine("%s.add(o);", objField));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+ code.add(EMPTY);
+ code.add(indentedLine("%s = null;", bytesField));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ appendBlock(body, code);
+ }
+ }
+
+ /** Generates the {@code for} loop body: per-element unmarshal + try/catch
add into the collection. */
+ private List<String> collectionFinishForBlock(VariableElement wireField,
String colField, String arrField, String fieldName) {
+ String compName = arrayComponentName(wireField);
+ TypeMirror compType =
((ArrayType)wireField.asType()).getComponentType();
+
+ List<String> code = new ArrayList<>();
+
+ code.add(indentedLine("for (%s e : %s) {", compName, arrField));
+
+ indent++;
+
+ code.addAll(codeFor(compType, "e", MarshalMode.UNMARSHAL));
+ code.add(EMPTY);
+ code.add(indentedLine("%s.add(e);", colField));
+
+ indent--;
+
+ code.add(indentedLine("}"));
+
+ return code;
+ }
+
+ /** Returns names of wire fields skipped by {@link #appendFields} in
UNMARSHAL mode. */
+ private Set<String> marshalledWireFieldsToSkip() {
+ Set<String> names = new HashSet<>();
+
+ for (VariableElement f : enclosed.values()) {
+ MarshalledCollection colAnn =
f.getAnnotation(MarshalledCollection.class);
Review Comment:
Let's use next lines. In the other places too.
##########
modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TestDelayMessageSerializer.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.spi.communication.tcp;
+
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/** Serializer for {@link TestDelayMessage} that injects an optional write
delay for testing. */
+public class TestDelayMessageSerializer implements
MessageSerializer<TestDelayMessage> {
Review Comment:
How it is used now?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java:
##########
@@ -1136,11 +1138,13 @@ public void send(ClusterNode node, GridCacheMessage
msg, byte plc) throws Ignite
if (log.isDebugEnabled())
log.debug("Sending cache message [msg=" + msg + ", node=" +
U.toShortString(node) + ']');
+ GridIoMessage ioMsg = cctx.gridIO().prepare(TOPIC_CACHE, msg, plc,
false, 0, false);
Review Comment:
Can we enclose the message preparing, preprocessing, pre-marshalling logic
into even lower transport level? Somewhere into `send()`.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java:
##########
@@ -1197,11 +1201,19 @@ public void sendOrderedMessage(ClusterNode node, Object
topic, GridCacheMessage
msg.lastAffinityChangedTopologyVersion(cctx.exchange().lastAffinityChangedTopologyVersion(msg.topologyVersion()));
+ if (node.isLocal()) {
+ cctx.gridIO().sendOrderedMessage(node, topic, msg, plc, timeout,
false);
+
+ return;
+ }
+
+ GridIoMessage ioMsg = cctx.gridIO().prepare(topic, msg, plc, true,
timeout, false);
Review Comment:
Maybe we should keep some transient `prepared` flag in a generated message
base class or in a message wrap (like `GridIoMessage`) and use proper abstract
`prepare/prepare0` methods. This might allow us to call the `prepare` logic
deeper in i/o, less visible and requiring less knowledge of how to send a
message. Especially messages or certain kinds like `GridCacheMessage`.
##########
modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/Message.java:
##########
@@ -23,7 +23,23 @@
import
org.apache.ignite.internal.managers.communication.UnknownMessageException;
/**
- * Base class for all communication messages.
+ * Base type for all messages sent between nodes, both over the communication
SPI and via discovery.
+ * <p>
+ * Serialized fields are declared by annotating instance fields; {@ignitelink
org.apache.ignite.internal.MessageProcessor} then
Review Comment:
Why we use `@ignitelink `? Normally, it is `@link`
##########
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:
`Map<Class<? extends Message>, Short>`?
##########
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<>();
+ /** Per-class cache over {@link #REGISTRATIONS}; keeps {@link
#directType()} off the map lookup done for every sent message. */
+ ClassValue<Short> DIRECT_TYPES = new ClassValue<>() {
+ @Override protected Short computeValue(Class<?> type) {
+ Short directType = REGISTRATIONS.get(type);
+
+ if (directType == null)
+ throw new
UnknownMessageException(type.asSubclass(Message.class));
+
+ return directType;
+ }
+ };
+
/**
* Gets message type.
*
* @return Message type.
*/
default short directType() {
- var clazz = getClass();
- Short type = REGISTRATIONS.get(clazz);
-
- if (type == null)
- throw new UnknownMessageException(clazz);
-
- return type;
+ return DIRECT_TYPES.get(getClass());
Review Comment:
```suggestion
return REGISTRATIONS.get(getClass());
```
##########
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<>();
+ /** Per-class cache over {@link #REGISTRATIONS}; keeps {@link
#directType()} off the map lookup done for every sent message. */
+ ClassValue<Short> DIRECT_TYPES = new ClassValue<>() {
Review Comment:
Why changed? let's use just `REGISTRATIONS`.
##########
modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/Message.java:
##########
Review Comment:
This is about serialization btw. It should be somewhere near to a
`Serializer` or a `MessageGnerator`.
##########
modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteMessageFactory.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.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 extends MessageFactory {
+ /** {@inheritDoc} */
+ @Override default void register(short directType, Supplier<Message>
supplier, MessageSerializer serializer)
+ throws IgniteException {
+ register(directType, supplier, serializer, null, null);
+ }
+
+ /**
+ * Register message factory with given direct type, serializer, and
marshaller. All messages must be registered
+ * during construction of class which implements this interface.
+ *
+ * @param directType Direct type.
+ * @param supplier Message factory.
+ * @param serializer Message serializer.
+ * @param marshaller Message marshaller, or {@code null} for
non-marshallable messages.
+ * @throws IgniteException In case of attempt to register message with
direct type which is already registered.
+ */
+ default void register(short directType, Supplier<Message> supplier,
MessageSerializer serializer,
+ @Nullable MessageMarshaller marshaller) throws IgniteException {
+ register(directType, supplier, serializer, marshaller, null);
+ }
+
+ /**
+ * Register message factory with given direct type, serializer,
marshaller, and deployer. All messages must be
+ * registered during construction of class which implements this interface.
+ *
+ * @param directType Direct type.
+ * @param supplier Message factory.
+ * @param serializer Message serializer.
+ * @param marshaller Message marshaller, or {@code null} for
non-marshallable messages.
+ * @param deployer Message deployer, or {@code null} for messages without
deployable fields.
+ * @throws IgniteException In case of attempt to register message with
direct type which is already registered.
+ */
+ public void register(short directType, Supplier<Message> supplier,
MessageSerializer serializer,
Review Comment:
The same here. Pls. see the patch in the other comment
--
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]
