1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
/*
* SPDX-FileCopyrightText: 2023 The Refinery Authors <https://refinery.tools/>
*
* SPDX-License-Identifier: EPL-2.0
*/
package tools.refinery.language.web.xtext.server;
import com.google.inject.Singleton;
import org.eclipse.xtext.ide.ExecutorServiceProvider;
import org.eclipse.xtext.web.server.model.XtextWebDocumentAccess;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tools.refinery.language.web.generator.ModelGenerationService;
import tools.refinery.language.web.semantics.SemanticsService;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
@Singleton
public class ThreadPoolExecutorServiceProvider extends ExecutorServiceProvider {
private static final Logger LOG = LoggerFactory.getLogger(ThreadPoolExecutorServiceProvider.class);
private static final String DOCUMENT_LOCK_EXECUTOR;
private static final AtomicInteger POOL_ID = new AtomicInteger(1);
private final Map<String, ScheduledExecutorService> scheduledInstanceCache =
Collections.synchronizedMap(new HashMap<>());
private final int executorThreadCount;
private final int lockExecutorThreadCount;
private final int semanticsExecutorThreadCount;
private final int generatorExecutorThreadCount;
static {
var lookup = MethodHandles.lookup();
MethodHandle getter;
try {
var privateLookup = MethodHandles.privateLookupIn(XtextWebDocumentAccess.class, lookup);
getter = privateLookup.findStaticGetter(XtextWebDocumentAccess.class, "DOCUMENT_LOCK_EXECUTOR",
String.class);
} catch (IllegalAccessException | NoSuchFieldException e) {
throw new IllegalStateException("Failed to find getter", e);
}
try {
DOCUMENT_LOCK_EXECUTOR = (String) getter.invokeExact();
} catch (Error e) {
// Rethrow JVM errors.
throw e;
} catch (Throwable e) {
throw new IllegalStateException("Failed to get DOCUMENT_LOCK_EXECUTOR", e);
}
}
public ThreadPoolExecutorServiceProvider() {
executorThreadCount = getCount("REFINERY_XTEXT_THREAD_COUNT").orElse(0);
lockExecutorThreadCount = getCount("REFINERY_XTEXT_LOCKING_THREAD_COUNT").orElse(executorThreadCount);
int semanticsCount = getCount("REFINERY_XTEXT_SEMANTICS_THREAD_COUNT").orElse(0);
if (semanticsCount == 0 || executorThreadCount == 0) {
semanticsExecutorThreadCount = 0;
} else {
semanticsExecutorThreadCount = Math.max(semanticsCount, executorThreadCount);
}
if (semanticsExecutorThreadCount != semanticsCount) {
LOG.warn("Setting REFINERY_XTEXT_SEMANTICS_THREAD_COUNT to {} to avoid deadlock. This value must be " +
"either 0 or at least as large as REFINERY_XTEXT_THREAD_COUNT to avoid lock contention.",
semanticsExecutorThreadCount);
}
generatorExecutorThreadCount = getCount("REFINERY_MODEL_GENERATION_THREAD_COUNT").orElse(executorThreadCount);
}
private static Optional<Integer> getCount(String name) {
return Optional.ofNullable(System.getenv(name)).map(Integer::parseUnsignedInt);
}
public ScheduledExecutorService getScheduled(String key) {
return scheduledInstanceCache.computeIfAbsent(key, this::createScheduledInstance);
}
@Override
protected ExecutorService createInstance(String key) {
String name = "xtext-" + POOL_ID.getAndIncrement();
if (key != null) {
name = name + "-" + key;
}
var threadFactory = new Factory(name, 5);
int size = getSize(key);
if (size == 0) {
return Executors.newCachedThreadPool(threadFactory);
}
return Executors.newFixedThreadPool(size, threadFactory);
}
protected ScheduledExecutorService createScheduledInstance(String key) {
String name = "xtext-scheduled-" + POOL_ID.getAndIncrement();
if (key != null) {
name = name + "-" + key;
}
var threadFactory = new Factory(name, 5);
return Executors.newScheduledThreadPool(1, threadFactory);
}
private int getSize(String key) {
if (SemanticsService.SEMANTICS_EXECUTOR.equals(key)) {
return semanticsExecutorThreadCount;
} else if (ModelGenerationService.MODEL_GENERATION_EXECUTOR.equals(key)) {
return generatorExecutorThreadCount;
} else if (DOCUMENT_LOCK_EXECUTOR.equals(key)) {
return lockExecutorThreadCount;
} else {
return executorThreadCount;
}
}
@Override
public void dispose() {
super.dispose();
synchronized (scheduledInstanceCache) {
for (var instance : scheduledInstanceCache.values()) {
instance.shutdown();
}
scheduledInstanceCache.clear();
}
}
private static class Factory implements ThreadFactory {
// We have to explicitly store the {@link ThreadGroup} to create a {@link ThreadFactory}.
@SuppressWarnings("squid:S3014")
private final ThreadGroup threadGroup = Thread.currentThread().getThreadGroup();
private final AtomicInteger threadId = new AtomicInteger(1);
private final String namePrefix;
private final int priority;
public Factory(String name, int priority) {
namePrefix = name + "-thread-";
this.priority = priority;
}
@Override
public Thread newThread(@NotNull Runnable runnable) {
var thread = new Thread(threadGroup, runnable, namePrefix + threadId.getAndIncrement());
if (thread.isDaemon()) {
thread.setDaemon(false);
}
if (thread.getPriority() != priority) {
thread.setPriority(priority);
}
return thread;
}
}
}
|
