Skip to content

Commit 5d2c0b7

Browse files
tglmantglman
committed
fix: changed static thread local allocation after recent failures
1 parent f4f094b commit 5d2c0b7

File tree

16 files changed

+119
-92
lines changed

16 files changed

+119
-92
lines changed

‎client/src/main/java/com/orientechnologies/orient/client/remote/db/document/ODatabaseDocumentRemote.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -574,7 +574,7 @@ public OIdentifiable beforeCreateOperations(OIdentifiable id, String iClusterNam
574574
}
575575
return id;
576576
} else if (res == ORecordHook.RESULT.RECORD_REPLACED) {
577-
ORecord replaced = OHookReplacedRecordThreadLocal.INSTANCE.get();
577+
ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
578578
if (replaced instanceof ODocument) {
579579
((ODocument) replaced).validate();
580580
}
@@ -593,7 +593,7 @@ public OIdentifiable beforeUpdateOperations(OIdentifiable id, String iClusterNam
593593
}
594594
return id;
595595
} else if (res == ORecordHook.RESULT.RECORD_REPLACED) {
596-
ORecord replaced = OHookReplacedRecordThreadLocal.INSTANCE.get();
596+
ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
597597
if (replaced instanceof ODocument) {
598598
((ODocument) replaced).validate();
599599
}

‎core/src/main/java/com/orientechnologies/orient/core/db/OExecutionThreadLocal.java‎

Lines changed: 1 addition & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -20,23 +20,13 @@
2020
package com.orientechnologies.orient.core.db;
2121

2222
import com.orientechnologies.common.thread.OSoftThread;
23-
import com.orientechnologies.orient.core.OOrientListenerAbstract;
24-
import com.orientechnologies.orient.core.Orient;
2523

2624
/**
2725
* Thread Local to store execution setting.
2826
*
2927
* @author Luca Garulli (l.garulli--(at)--orientdb.com)
3028
*/
31-
public class OExecutionThreadLocal extends ThreadLocal<OExecutionThreadLocal.OExecutionThreadData> {
32-
public class OExecutionThreadData {}
33-
34-
@Override
35-
protected OExecutionThreadData initialValue() {
36-
return new OExecutionThreadData();
37-
}
38-
39-
public static volatile OExecutionThreadLocal INSTANCE = new OExecutionThreadLocal();
29+
public class OExecutionThreadLocal {
4030

4131
public static boolean isInterruptCurrentOperation() {
4232
final Thread t = Thread.currentThread();
@@ -52,20 +42,4 @@ public static void setInterruptCurrentOperation() {
5242
final Thread t = Thread.currentThread();
5343
if (t instanceof OSoftThread) ((OSoftThread) t).softShutdown();
5444
}
55-
56-
static {
57-
final Orient inst = Orient.instance();
58-
inst.registerListener(
59-
new OOrientListenerAbstract() {
60-
@Override
61-
public void onStartup() {
62-
if (INSTANCE == null) INSTANCE = new OExecutionThreadLocal();
63-
}
64-
65-
@Override
66-
public void onShutdown() {
67-
INSTANCE = null;
68-
}
69-
});
70-
}
7145
}

‎core/src/main/java/com/orientechnologies/orient/core/db/OHookReplacedRecordThreadLocal.java‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -33,14 +33,11 @@ public class OHookReplacedRecordThreadLocal extends ThreadLocal<ORecord> {
3333
Orient.instance()
3434
.registerListener(
3535
new OOrientListenerAbstract() {
36-
@Override
37-
public void onStartup() {
38-
if (INSTANCE == null) INSTANCE = new OHookReplacedRecordThreadLocal();
39-
}
40-
4136
@Override
4237
public void onShutdown() {
43-
INSTANCE = null;
38+
synchronized (OHookReplacedRecordThreadLocal.class) {
39+
INSTANCE = null;
40+
}
4441
}
4542
});
4643
}
@@ -52,4 +49,15 @@ public ORecord getIfDefined() {
5249
public boolean isDefined() {
5350
return super.get() != null;
5451
}
52+
53+
public static OHookReplacedRecordThreadLocal instance() {
54+
if (INSTANCE == null) {
55+
synchronized (OHookReplacedRecordThreadLocal.class) {
56+
if (INSTANCE == null) {
57+
INSTANCE = new OHookReplacedRecordThreadLocal();
58+
}
59+
}
60+
}
61+
return INSTANCE;
62+
}
5563
}

‎core/src/main/java/com/orientechnologies/orient/core/db/OScenarioThreadLocal.java‎

Lines changed: 21 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -44,14 +44,11 @@ public static class RunContext {
4444
Orient.instance()
4545
.registerListener(
4646
new OOrientListenerAbstract() {
47-
@Override
48-
public void onStartup() {
49-
if (INSTANCE == null) INSTANCE = new OScenarioThreadLocal();
50-
}
51-
5247
@Override
5348
public void onShutdown() {
54-
INSTANCE = null;
49+
synchronized (OScenarioThreadLocal.class) {
50+
INSTANCE = null;
51+
}
5552
}
5653
});
5754
}
@@ -60,12 +57,23 @@ public OScenarioThreadLocal() {
6057
setRunMode(RUN_MODE.DEFAULT);
6158
}
6259

60+
public static OScenarioThreadLocal instance() {
61+
if (INSTANCE == null) {
62+
synchronized (OScenarioThreadLocal.class) {
63+
if (INSTANCE == null) {
64+
INSTANCE = new OScenarioThreadLocal();
65+
}
66+
}
67+
}
68+
return INSTANCE;
69+
}
70+
6371
public static <T> Object executeAsDefault(final Callable<T> iCallback) {
6472
final OScenarioThreadLocal.RUN_MODE currentDistributedMode =
65-
OScenarioThreadLocal.INSTANCE.getRunMode();
73+
OScenarioThreadLocal.instance().getRunMode();
6674
if (currentDistributedMode == OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED)
6775
// ASSURE SCHEMA CHANGES ARE NEVER PROPAGATED ON CLUSTER
68-
OScenarioThreadLocal.INSTANCE.setRunMode(RUN_MODE.DEFAULT);
76+
OScenarioThreadLocal.instance().setRunMode(RUN_MODE.DEFAULT);
6977

7078
try {
7179
return (T) iCallback.call();
@@ -76,16 +84,17 @@ public static <T> Object executeAsDefault(final Callable<T> iCallback) {
7684
} finally {
7785
if (currentDistributedMode == OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED)
7886
// RESTORE PREVIOUS MODE
79-
OScenarioThreadLocal.INSTANCE.setRunMode(OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED);
87+
OScenarioThreadLocal.instance()
88+
.setRunMode(OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED);
8089
}
8190
}
8291

8392
public static Object executeAsDistributed(final Callable<? extends Object> iCallback) {
8493
final OScenarioThreadLocal.RUN_MODE currentDistributedMode =
85-
OScenarioThreadLocal.INSTANCE.getRunMode();
94+
OScenarioThreadLocal.instance().getRunMode();
8695
if (currentDistributedMode != OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED)
8796
// ASSURE SCHEMA CHANGES ARE NEVER PROPAGATED ON CLUSTER
88-
OScenarioThreadLocal.INSTANCE.setRunMode(OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED);
97+
OScenarioThreadLocal.instance().setRunMode(OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED);
8998

9099
try {
91100
return iCallback.call();
@@ -96,7 +105,7 @@ public static Object executeAsDistributed(final Callable<? extends Object> iCall
96105
} finally {
97106
if (currentDistributedMode != OScenarioThreadLocal.RUN_MODE.RUNNING_DISTRIBUTED)
98107
// RESTORE PREVIOUS MODE
99-
OScenarioThreadLocal.INSTANCE.setRunMode(OScenarioThreadLocal.RUN_MODE.DEFAULT);
108+
OScenarioThreadLocal.instance().setRunMode(OScenarioThreadLocal.RUN_MODE.DEFAULT);
100109
}
101110
}
102111

‎core/src/main/java/com/orientechnologies/orient/core/db/document/ODatabaseDocumentAbstract.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -488,7 +488,7 @@ public ORecordHook.RESULT callbackHooks(final ORecordHook.TYPE type, final OIden
488488
final ORecord rec = id.getRecord();
489489
if (rec == null) return ORecordHook.RESULT.RECORD_NOT_CHANGED;
490490

491-
final OScenarioThreadLocal.RUN_MODE runMode = OScenarioThreadLocal.INSTANCE.getRunMode();
491+
final OScenarioThreadLocal.RUN_MODE runMode = OScenarioThreadLocal.instance().getRunMode();
492492

493493
boolean recordChanged = false;
494494
for (ORecordHook hook : hooksByScope[scopeOrdinal]) {

‎core/src/main/java/com/orientechnologies/orient/core/db/document/ODatabaseDocumentEmbedded.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -950,7 +950,7 @@ public OIdentifiable beforeCreateOperations(OIdentifiable id, String iClusterNam
950950
return id;
951951
} else if (res == ORecordHook.RESULT.RECORD_REPLACED
952952
|| triggerChanged == ORecordHook.RESULT.RECORD_REPLACED) {
953-
ORecord replaced = OHookReplacedRecordThreadLocal.INSTANCE.get();
953+
ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
954954
if (replaced instanceof ODocument) {
955955
((ODocument) replaced).validate();
956956
}
@@ -1007,7 +1007,7 @@ public OIdentifiable beforeUpdateOperations(OIdentifiable id, String iClusterNam
10071007
return id;
10081008
} else if (res == ORecordHook.RESULT.RECORD_REPLACED
10091009
|| triggerChanged == ORecordHook.RESULT.RECORD_REPLACED) {
1010-
ORecord replaced = OHookReplacedRecordThreadLocal.INSTANCE.get();
1010+
ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
10111011
if (replaced instanceof ODocument) {
10121012
((ODocument) replaced).validate();
10131013
}

‎core/src/main/java/com/orientechnologies/orient/core/serialization/serializer/record/ORecordSaveThreadLocal.java‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,27 +37,37 @@ public class ORecordSaveThreadLocal extends ThreadLocal<ORecord> {
3737
Orient.instance()
3838
.registerListener(
3939
new OOrientListenerAbstract() {
40-
@Override
41-
public void onStartup() {
42-
if (INSTANCE == null) INSTANCE = new ORecordSaveThreadLocal();
43-
}
44-
4540
@Override
4641
public void onShutdown() {
47-
INSTANCE = null;
42+
synchronized (ORecordSaveThreadLocal.class) {
43+
INSTANCE = null;
44+
}
4845
}
4946
});
5047
}
5148

49+
private static void init() {
50+
if (INSTANCE == null) {
51+
synchronized (ORecordSaveThreadLocal.class) {
52+
if (INSTANCE == null) {
53+
INSTANCE = new ORecordSaveThreadLocal();
54+
}
55+
}
56+
}
57+
}
58+
5259
public static ORecord getLast() {
60+
init();
5361
return INSTANCE.get();
5462
}
5563

5664
public static void setLast(final ORecord document) {
65+
init();
5766
INSTANCE.set(document);
5867
}
5968

6069
public static void removeLast() {
70+
init();
6171
INSTANCE.set(null);
6272
}
6373
}

‎core/src/main/java/com/orientechnologies/orient/core/serialization/serializer/record/OSerializationThreadLocal.java‎

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -25,24 +25,32 @@
2525
import java.util.Set;
2626

2727
public class OSerializationThreadLocal extends ThreadLocal<Set<Integer>> {
28-
public static volatile OSerializationThreadLocal INSTANCE = new OSerializationThreadLocal();
28+
private static volatile OSerializationThreadLocal INSTANCE = new OSerializationThreadLocal();
2929

3030
static {
3131
Orient.instance()
3232
.registerListener(
3333
new OOrientListenerAbstract() {
34-
@Override
35-
public void onStartup() {
36-
if (INSTANCE == null) INSTANCE = new OSerializationThreadLocal();
37-
}
38-
3934
@Override
4035
public void onShutdown() {
41-
INSTANCE = null;
36+
synchronized (OSerializationThreadLocal.class) {
37+
INSTANCE = null;
38+
}
4239
}
4340
});
4441
}
4542

43+
public static OSerializationThreadLocal instance() {
44+
if (INSTANCE == null) {
45+
synchronized (OSerializationThreadLocal.class) {
46+
if (INSTANCE == null) {
47+
INSTANCE = new OSerializationThreadLocal();
48+
}
49+
}
50+
}
51+
return INSTANCE;
52+
}
53+
4654
@Override
4755
protected Set<Integer> initialValue() {
4856
return new HashSet<Integer>();

‎core/src/main/java/com/orientechnologies/orient/core/sql/functions/OSQLFunctionAbstract.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ public Object mergeDistributedResult(List<Object> resultsToMerge) {
9696
}
9797

9898
protected boolean returnDistributedResult() {
99-
return OScenarioThreadLocal.INSTANCE.isRunModeDistributed();
99+
return OScenarioThreadLocal.instance().isRunModeDistributed();
100100
}
101101

102102
protected String getDistributedStorageId() {

‎core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/paginated/ORecordSerializationContext.java‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -40,48 +40,55 @@ public class ORecordSerializationContext {
4040
Orient.instance()
4141
.registerListener(
4242
new OOrientListenerAbstract() {
43-
@Override
44-
public void onStartup() {
45-
if (SERIALIZATION_CONTEXT_STACK == null)
46-
SERIALIZATION_CONTEXT_STACK = new SerializationContextThreadLocal();
47-
}
48-
4943
@Override
5044
public void onShutdown() {
51-
SERIALIZATION_CONTEXT_STACK = null;
45+
synchronized (ORecordSerializationContext.class) {
46+
SERIALIZATION_CONTEXT_STACK = null;
47+
}
5248
}
5349
});
5450
}
5551

5652
private final Deque<ORecordSerializationOperation> operations = new ArrayDeque<>();
5753

5854
public static int getDepth() {
59-
return ORecordSerializationContext.SERIALIZATION_CONTEXT_STACK.get().size();
55+
return getStack().size();
6056
}
6157

6258
public static ORecordSerializationContext pushContext() {
63-
final Deque<ORecordSerializationContext> stack = SERIALIZATION_CONTEXT_STACK.get();
59+
final Deque<ORecordSerializationContext> stack = getStack();
6460

6561
final ORecordSerializationContext context = new ORecordSerializationContext();
6662
stack.push(context);
6763
return context;
6864
}
6965

7066
public static ORecordSerializationContext getContext() {
71-
final Deque<ORecordSerializationContext> stack = SERIALIZATION_CONTEXT_STACK.get();
67+
final Deque<ORecordSerializationContext> stack = getStack();
7268
if (stack.isEmpty()) return null;
7369

7470
return stack.peek();
7571
}
7672

7773
public static ORecordSerializationContext pullContext() {
78-
final Deque<ORecordSerializationContext> stack = SERIALIZATION_CONTEXT_STACK.get();
74+
final Deque<ORecordSerializationContext> stack = getStack();
7975
if (stack.isEmpty())
8076
throw new IllegalStateException("Cannot find current serialization context");
8177

8278
return stack.poll();
8379
}
8480

81+
private static Deque<ORecordSerializationContext> getStack() {
82+
if (SERIALIZATION_CONTEXT_STACK == null) {
83+
synchronized (ORecordSerializationContext.class) {
84+
if (SERIALIZATION_CONTEXT_STACK == null) {
85+
SERIALIZATION_CONTEXT_STACK = new SerializationContextThreadLocal();
86+
}
87+
}
88+
}
89+
return SERIALIZATION_CONTEXT_STACK.get();
90+
}
91+
8592
public void push(ORecordSerializationOperation operation) {
8693
operations.push(operation);
8794
}

0 commit comments

Comments
 (0)