Skip to content

Commit 7a61b2d

Browse files
Vladsz83Vladsz83
authored
IGNITE-28808 Restricted distributed Operation Context attribute registration after node started (#13275)
1 parent ff6d9eb commit 7a61b2d

14 files changed

Lines changed: 214 additions & 154 deletions

File tree

‎modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -687,7 +687,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
687687

688688
// [13400 - 13500]: Operation context messages.
689689
msgIdx = 13400;
690-
withNoSchema(DistributedOperationContextMessage.class);
690+
withNoSchema(OperationContextMessage.class);
691691

692692
// [13500 - 13600]: Rolling Upgrade messages.
693693
msgIdx = 13500;

‎modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@
7878
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
7979
import org.apache.ignite.internal.processors.tracing.Tracing;
8080
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
81+
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
8182
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
8283
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
8384
import org.apache.ignite.internal.worker.WorkersRegistry;
@@ -211,12 +212,19 @@ public interface GridKernalContext extends Iterable<GridComponent> {
211212
public MaintenanceRegistry maintenanceRegistry();
212213

213214
/**
214-
* Gets core message factoy.
215+
* Gets core message factory.
215216
*
216217
* @return Core message factory.
217218
*/
218219
public MessageFactory messageFactory();
219220

221+
/**
222+
* Gets the distributed operation context dispatcher.
223+
*
224+
* @return The distributed operation context dispatcher.
225+
*/
226+
public OperationContextDispatcher operationContextDispatcher();
227+
220228
/**
221229
* Gets transformation processor.
222230
*

‎modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@
101101
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
102102
import org.apache.ignite.internal.processors.tracing.Tracing;
103103
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
104+
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
104105
import org.apache.ignite.internal.thread.pool.IgniteForkJoinPool;
105106
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
106107
import org.apache.ignite.internal.util.spring.IgniteSpringHelper;
@@ -697,6 +698,11 @@ else if (!(comp instanceof DiscoveryNodeValidationProcessor
697698
return grid.messageFactory();
698699
}
699700

701+
/** {@inheritDoc} */
702+
@Override public OperationContextDispatcher operationContextDispatcher() {
703+
return grid.operationContextDispatcher();
704+
}
705+
700706
/** {@inheritDoc} */
701707
@Override public CacheObjectTransformerProcessor transformer() {
702708
return transProc;

‎modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,7 @@
174174
import org.apache.ignite.internal.suggestions.JvmConfigurationSuggestions;
175175
import org.apache.ignite.internal.suggestions.OsConfigurationSuggestions;
176176
import org.apache.ignite.internal.systemview.ConfigurationViewWalker;
177+
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
177178
import org.apache.ignite.internal.util.TimeBag;
178179
import org.apache.ignite.internal.util.future.GridCompoundFuture;
179180
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -444,6 +445,9 @@ public class IgniteKernal implements IgniteEx, Externalizable {
444445
/** Core message factory. */
445446
private MessageFactory msgFactory;
446447

448+
/** Distributed operation context dispatcher. */
449+
private OperationContextDispatcher operationCtxDispatcher;
450+
447451
/**
448452
* No-arg constructor is required by externalization.
449453
*/
@@ -930,6 +934,8 @@ public void start(
930934
longJVMPauseDetector
931935
);
932936

937+
operationCtxDispatcher = new OperationContextDispatcher();
938+
933939
startProcessor(new DiagnosticProcessor(ctx));
934940

935941
mBeansMgr = new IgniteMBeansManager(this);
@@ -1152,6 +1158,8 @@ public void start(
11521158
// All components exept Discovery are started, time to check if maintenance is still needed.
11531159
mntcProc.prepareAndExecuteMaintenance();
11541160

1161+
operationCtxDispatcher.finishRegistration();
1162+
11551163
gw.writeLock();
11561164

11571165
try {
@@ -3059,6 +3067,11 @@ MessageFactory messageFactory() {
30593067
return msgFactory;
30603068
}
30613069

3070+
/** @return Distributed operation context dispatcher. */
3071+
OperationContextDispatcher operationContextDispatcher() {
3072+
return operationCtxDispatcher;
3073+
}
3074+
30623075
/**
30633076
* Method is responsible for handling the {@link EventType#EVT_CLIENT_NODE_DISCONNECTED} event. Notify all the
30643077
* GridComponents that the such even has been occurred (e.g. if the local client node disconnected from the cluster

modules/core/src/main/java/org/apache/ignite/internal/DistributedOperationContextMessage.java renamed to modules/core/src/main/java/org/apache/ignite/internal/OperationContextMessage.java

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,16 @@
1717

1818
package org.apache.ignite.internal;
1919

20-
import org.apache.ignite.internal.thread.context.DistributedOperationContextManager;
2120
import org.apache.ignite.internal.thread.context.OperationContext;
21+
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
2222
import org.apache.ignite.plugin.extensions.communication.Message;
2323

2424
/**
25-
* Transport for {@link OperationContext} distributed attributes.
25+
* Message for {@link OperationContext} distributed attributes.
2626
*
27-
* @see DistributedOperationContextManager
27+
* @see OperationContextDispatcher
2828
*/
29-
public class DistributedOperationContextMessage implements Message {
29+
public class OperationContextMessage implements Message {
3030
/** Values of operation context attributes. */
3131
@Order(0)
3232
public Message[] vals;
@@ -36,7 +36,7 @@ public class DistributedOperationContextMessage implements Message {
3636
public byte idBitmap;
3737

3838
/** Empty constructor for serialization purposes. */
39-
public DistributedOperationContextMessage() {
39+
public OperationContextMessage() {
4040
// No-op.
4141
}
4242
}

‎modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,6 @@
102102
import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
103103
import org.apache.ignite.internal.processors.tracing.Span;
104104
import org.apache.ignite.internal.processors.tracing.SpanTags;
105-
import org.apache.ignite.internal.thread.context.DistributedOperationContextManager;
106105
import org.apache.ignite.internal.thread.context.Scope;
107106
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashSet;
108107
import org.apache.ignite.internal.util.IgniteUtils;
@@ -461,7 +460,7 @@ public void resetMetrics() {
461460
try {
462461
GridIoMessage msg0 = (GridIoMessage)msg;
463462

464-
try (Scope ignored = DistributedOperationContextManager.instance().restoreDistributedAttributes(msg0.opCtxMsg)) {
463+
try (Scope ignored = ctx.operationContextDispatcher().restoreDistributedAttributes(msg0.opCtxMsg)) {
465464
onMessage0(nodeId, msg0, msgC);
466465
}
467466
}
@@ -2055,7 +2054,7 @@ private long getInverseConnectionWaitTimeout() {
20552054
else
20562055
res = new GridIoMessage(plc, topic, msg, ordered, timeout, skipOnTimeout);
20572056

2058-
res.opCtxMsg = DistributedOperationContextManager.instance().collectDistributedAttributes();
2057+
res.opCtxMsg = ctx.operationContextDispatcher().collectDistributedAttributes();
20592058

20602059
return res;
20612060
}

‎modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,9 @@
1717

1818
package org.apache.ignite.internal.managers.communication;
1919

20-
import org.apache.ignite.internal.DistributedOperationContextMessage;
2120
import org.apache.ignite.internal.ExecutorAwareMessage;
2221
import org.apache.ignite.internal.GridTopicMessage;
22+
import org.apache.ignite.internal.OperationContextMessage;
2323
import org.apache.ignite.internal.Order;
2424
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
2525
import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
@@ -65,10 +65,10 @@ public class GridIoMessage implements Message, SpanTransport {
6565
@Order(6)
6666
byte[] span;
6767

68-
/** Effective operation context attributes. */
68+
/** Effective operation context attributes to propagate. */
6969
@Order(7)
7070
@GridToStringInclude
71-
public @Nullable DistributedOperationContextMessage opCtxMsg;
71+
public @Nullable OperationContextMessage opCtxMsg;
7272

7373
/**
7474
* Default constructor.

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@
103103
import org.apache.ignite.internal.processors.tracing.NoopTracing;
104104
import org.apache.ignite.internal.processors.tracing.Tracing;
105105
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
106+
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
106107
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
107108
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
108109
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -176,6 +177,9 @@ public class StandaloneGridKernalContext implements GridKernalContext {
176177
/** Marshaller. */
177178
private final BinaryMarshaller marsh;
178179

180+
/** Operation context dispacther. */
181+
private final OperationContextDispatcher opCtxDispatcher = new OperationContextDispatcher();
182+
179183
/**
180184
* @param log Logger.
181185
* @param ft Node file tree.
@@ -454,6 +458,11 @@ private void setField(IgniteEx kernal, String name, Object val) throws NoSuchFie
454458
return null;
455459
}
456460

461+
/** {@inheritDoc} */
462+
@Override public OperationContextDispatcher operationContextDispatcher() {
463+
return opCtxDispatcher;
464+
}
465+
457466
/** {@inheritDoc} */
458467
@Override public CacheObjectTransformerProcessor transformer() {
459468
return transProc;

modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedOperationContextManager.java renamed to modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextDispatcher.java

Lines changed: 32 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import java.util.Map;
2222
import java.util.concurrent.ConcurrentSkipListMap;
2323
import org.apache.ignite.IgniteException;
24-
import org.apache.ignite.internal.DistributedOperationContextMessage;
24+
import org.apache.ignite.internal.OperationContextMessage;
2525
import org.apache.ignite.internal.util.typedef.F;
2626
import org.apache.ignite.plugin.extensions.communication.Message;
2727
import org.jetbrains.annotations.Nullable;
@@ -37,71 +37,62 @@
3737
* {@link OperationContextAttribute} instance that is consistent across all cluster nodes.</p>
3838
*
3939
* <p>To enable propagation of an {@link OperationContextAttribute} value across cluster nodes, the
40-
* attribute must be created using the {@link #createDistributedAttribute(byte, Message)} method.
40+
* attribute must be registered with the {@link #registerDistributedAttribute(int, OperationContextAttribute)} method.
4141
*
42-
* <p> Note, that the maximum number of distributed attribute instances that can be created is currently limited to
43-
* {@link #MAX_DISTRIBUTED_ATTR_CNT} for implementation reasons.</p>
42+
* <p> Note, that the maximum number of distributed attributes to register is currently limited to
43+
* {@link #MAX_ATTRS_CNT} for implementation reasons.</p>
4444
*
4545
* @see OperationContext
46-
* @see DistributedOperationContextMessage
46+
* @see OperationContextMessage
4747
*/
48-
public class DistributedOperationContextManager {
49-
/** */
50-
private static final DistributedOperationContextManager INSTANCE = new DistributedOperationContextManager();
51-
48+
public class OperationContextDispatcher {
5249
/** Maximal number of supported distributed attributes. */
53-
static final byte MAX_DISTRIBUTED_ATTR_CNT = Byte.SIZE;
50+
static final byte MAX_ATTRS_CNT = Byte.SIZE;
5451

5552
/** Registered distributed attributes by their cluster-wide id. */
56-
private final Map<Byte, OperationContextAttribute<Message>> attrs = new ConcurrentSkipListMap<>();
53+
private final Map<Byte, OperationContextAttribute<? extends Message>> attrs = new ConcurrentSkipListMap<>();
5754

58-
/** */
59-
public static DistributedOperationContextManager instance() {
60-
return INSTANCE;
61-
}
55+
/** Whether the registration of new distributed attributes is allowed. */
56+
private volatile boolean regFinished;
6257

6358
/**
64-
* Creates a new {@link OperationContext} attribute with the specified distributed ID and initial value.
59+
* Registers an attribute of {@link OperationContext} with the specified distributed ID.
6560
*
6661
* <p>The distributed ID is used to consistently identify the attribute across all nodes in the cluster.
67-
* It must be unique, and its value must be in the range from {@code 0} (inclusive) to {@code Byte.SIZE} (exclusive).</p>
68-
*
69-
* <p>The value of the created attribute is automatically captured and propagated between cluster nodes
70-
* during message transmission.</p>
62+
* It must be unique, and its value must be in the range [{@code 0} : {@code Byte.SIZE}).</p>
7163
*
72-
* @see OperationContextAttribute#newInstance(Object)
64+
* <p>Registered attribute value is automatically captured and propagated between cluster nodes
65+
* during the messages transmission.</p>
7366
*/
74-
public <T extends Message> OperationContextAttribute<T> createDistributedAttribute(byte id, @Nullable T initVal) {
75-
assert id >= 0 && id < MAX_DISTRIBUTED_ATTR_CNT : "Invalid distributed attributed id [id=" + id + ']';
67+
public <T extends Message> void registerDistributedAttribute(int id, OperationContextAttribute<T> attr) {
68+
if (regFinished)
69+
throw new IgniteException("Initialization of distributed operation context attributes has already finished.");
7670

77-
return (OperationContextAttribute<T>)attrs.compute(id, (id0, attr0) -> {
78-
if (attr0 != null)
79-
throw new IgniteException("Duplicated distributed attribute id [id=" + id + ']');
71+
assert id >= 0 && id < MAX_ATTRS_CNT : "Invalid distributed attributed id [id=" + id + ']';
8072

81-
return OperationContextAttribute.newInstance(initVal);
82-
});
73+
if (attrs.putIfAbsent((byte)id, attr) != null)
74+
throw new IgniteException("Duplicated distributed attribute id [id=" + id + ']');
8375
}
8476

8577
/**
86-
* Collects the values of all distributed {@link OperationContextAttribute}s registered by this manager in a format
87-
* suitable for transmission between cluster nodes.
78+
* Collects the values of all distributed {@link OperationContextAttribute}s registered by this dispatcher.
8879
*
8980
* @see OperationContext#get(OperationContextAttribute)
9081
*/
91-
public @Nullable DistributedOperationContextMessage collectDistributedAttributes() {
92-
DistributedOperationContextMessage res = null;
82+
public @Nullable OperationContextMessage collectDistributedAttributes() {
83+
OperationContextMessage res = null;
9384
List<Message> vals = null;
9485

95-
for (Map.Entry<Byte, OperationContextAttribute<Message>> e : attrs.entrySet()) {
86+
for (Map.Entry<Byte, OperationContextAttribute<? extends Message>> e : attrs.entrySet()) {
9687
OperationContextAttribute<? extends Message> attr = e.getValue();
9788

9889
Message curVal = OperationContext.get(attr);
9990

10091
if (curVal != attr.initialValue()) {
10192
if (res == null) {
102-
res = new DistributedOperationContextMessage();
93+
res = new OperationContextMessage();
10394

104-
vals = new ArrayList<>(MAX_DISTRIBUTED_ATTR_CNT / 2);
95+
vals = new ArrayList<>(MAX_ATTRS_CNT / 2);
10596
}
10697

10798
byte mask = (byte)(1 << e.getKey());
@@ -120,13 +111,13 @@ public <T extends Message> OperationContextAttribute<T> createDistributedAttribu
120111
}
121112

122113
/** Restores distributed {@link OperationContextAttribute} values received from a remote node. */
123-
public Scope restoreDistributedAttributes(@Nullable DistributedOperationContextMessage msg) {
114+
public Scope restoreDistributedAttributes(@Nullable OperationContextMessage msg) {
124115
if (msg == null)
125116
return Scope.NOOP_SCOPE;
126117

127118
assert msg.idBitmap != 0;
128119
assert !F.isEmpty(msg.vals);
129-
assert msg.vals.length <= MAX_DISTRIBUTED_ATTR_CNT;
120+
assert msg.vals.length <= MAX_ATTRS_CNT;
130121

131122
OperationContext.ContextUpdater updater = OperationContext.ContextUpdater.create();
132123

@@ -136,7 +127,7 @@ public Scope restoreDistributedAttributes(@Nullable DistributedOperationContextM
136127
while ((msg.idBitmap & (1 << maskIdx)) == 0)
137128
++maskIdx;
138129

139-
OperationContextAttribute<Message> attr = attrs.get(maskIdx++);
130+
OperationContextAttribute<Message> attr = (OperationContextAttribute<Message>)attrs.get(maskIdx++);
140131

141132
assert attr != null;
142133

@@ -146,8 +137,8 @@ public Scope restoreDistributedAttributes(@Nullable DistributedOperationContextM
146137
return updater.apply();
147138
}
148139

149-
/** For testing purposes mostly. */
150-
void clear() {
151-
attrs.clear();
140+
/** Restricts further registration of distributed attributes. */
141+
public void finishRegistration() {
142+
regFinished = true;
152143
}
153144
}

‎modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,6 @@
7070
import org.apache.ignite.internal.processors.tracing.messages.SpanContainer;
7171
import org.apache.ignite.internal.processors.tracing.messages.TraceableMessage;
7272
import org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable;
73-
import org.apache.ignite.internal.thread.context.DistributedOperationContextManager;
7473
import org.apache.ignite.internal.thread.context.Scope;
7574
import org.apache.ignite.internal.util.future.GridFutureAdapter;
7675
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
@@ -1312,7 +1311,7 @@ private class SocketWriter extends IgniteSpiThread {
13121311
* @param msg Message.
13131312
*/
13141313
private void sendMessage(TcpDiscoveryAbstractMessage msg) {
1315-
msg.opCtxMsg = DistributedOperationContextManager.instance().collectDistributedAttributes();
1314+
msg.opCtxMsg = operationCtxDispatcher.collectDistributedAttributes();
13161315

13171316
synchronized (mux) {
13181317
queue.add(msg);
@@ -1765,8 +1764,7 @@ private MessageWorker(IgniteLogger log) {
17651764
? (TcpDiscoveryAbstractMessage)msg
17661765
: null;
17671766

1768-
try (Scope ignored = DistributedOperationContextManager.instance()
1769-
.restoreDistributedAttributes(dm == null ? null : dm.opCtxMsg)) {
1767+
try (Scope ignored = operationCtxDispatcher.restoreDistributedAttributes(dm == null ? null : dm.opCtxMsg)) {
17701768
if (msg instanceof JoinTimeout) {
17711769
int joinCnt0 = ((JoinTimeout)msg).joinCnt;
17721770

0 commit comments

Comments
 (0)