Skip to content

Commit a77eaa8

Browse files
wyhasanywyhasany
authored
Align to use virtual threads (#540)
Align to use virtual threads - In scope of this ticket I removed synchronized blocks as they pin virtual thread to carrier thread. - Expose user way to inject ThreadFactory for creating virtual threads
1 parent 3afa432 commit a77eaa8

File tree

10 files changed

+234
-78
lines changed

10 files changed

+234
-78
lines changed

‎core-api/src/main/java/com/optimizely/ab/Optimizely.java

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import com.optimizely.ab.optimizelyconfig.OptimizelyConfigService;
3636
import com.optimizely.ab.optimizelydecision.*;
3737
import com.optimizely.ab.optimizelyjson.OptimizelyJSON;
38+
import java.util.concurrent.locks.ReentrantLock;
3839
import org.slf4j.Logger;
3940
import org.slf4j.LoggerFactory;
4041

@@ -101,6 +102,8 @@ public class Optimizely implements AutoCloseable {
101102
@Nullable
102103
private final ODPManager odpManager;
103104

105+
private final ReentrantLock lock = new ReentrantLock();
106+
104107
private Optimizely(@Nonnull EventHandler eventHandler,
105108
@Nonnull EventProcessor eventProcessor,
106109
@Nonnull ErrorHandler errorHandler,
@@ -1451,8 +1454,11 @@ public List<String> fetchQualifiedSegments(String userId, @Nonnull List<ODPSegme
14511454
return null;
14521455
}
14531456
if (odpManager != null) {
1454-
synchronized (odpManager) {
1457+
lock.lock();
1458+
try {
14551459
return odpManager.getSegmentManager().getQualifiedSegments(userId, segmentOptions);
1460+
} finally {
1461+
lock.unlock();
14561462
}
14571463
}
14581464
logger.error("Audience segments fetch failed (ODP is not enabled).");

‎core-api/src/main/java/com/optimizely/ab/config/PollingProjectConfigManager.java

Lines changed: 56 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import com.optimizely.ab.optimizelyconfig.OptimizelyConfig;
2323
import com.optimizely.ab.optimizelyconfig.OptimizelyConfigManager;
2424
import com.optimizely.ab.optimizelyconfig.OptimizelyConfigService;
25+
import java.util.concurrent.locks.ReentrantLock;
26+
import javax.annotation.Nullable;
2527
import org.slf4j.Logger;
2628
import org.slf4j.LoggerFactory;
2729

@@ -60,6 +62,7 @@ public abstract class PollingProjectConfigManager implements ProjectConfigManage
6062
private volatile String sdkKey;
6163
private volatile boolean started;
6264
private ScheduledFuture<?> scheduledFuture;
65+
private ReentrantLock lock = new ReentrantLock();
6366

6467
public PollingProjectConfigManager(long period, TimeUnit timeUnit) {
6568
this(period, timeUnit, Long.MAX_VALUE, TimeUnit.MILLISECONDS, new NotificationCenter());
@@ -70,6 +73,15 @@ public PollingProjectConfigManager(long period, TimeUnit timeUnit, NotificationC
7073
}
7174

7275
public PollingProjectConfigManager(long period, TimeUnit timeUnit, long blockingTimeoutPeriod, TimeUnit blockingTimeoutUnit, NotificationCenter notificationCenter) {
76+
this(period, timeUnit, blockingTimeoutPeriod, blockingTimeoutUnit, notificationCenter, null);
77+
}
78+
79+
public PollingProjectConfigManager(long period,
80+
TimeUnit timeUnit,
81+
long blockingTimeoutPeriod,
82+
TimeUnit blockingTimeoutUnit,
83+
NotificationCenter notificationCenter,
84+
@Nullable ThreadFactory customThreadFactory) {
7385
this.period = period;
7486
this.timeUnit = timeUnit;
7587
this.blockingTimeoutPeriod = blockingTimeoutPeriod;
@@ -78,7 +90,7 @@ public PollingProjectConfigManager(long period, TimeUnit timeUnit, long blocking
7890
if (TimeUnit.SECONDS.convert(period, this.timeUnit) < 30) {
7991
logger.warn("Polling intervals below 30 seconds are not recommended.");
8092
}
81-
final ThreadFactory threadFactory = Executors.defaultThreadFactory();
93+
final ThreadFactory threadFactory = customThreadFactory != null ? customThreadFactory : Executors.defaultThreadFactory();
8294
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(runnable -> {
8395
Thread thread = threadFactory.newThread(runnable);
8496
thread.setDaemon(true);
@@ -176,43 +188,58 @@ public String getSDKKey() {
176188
return this.sdkKey;
177189
}
178190

179-
public synchronized void start() {
180-
if (started) {
181-
logger.warn("Manager already started.");
182-
return;
183-
}
191+
public void start() {
192+
lock.lock();
193+
try {
194+
if (started) {
195+
logger.warn("Manager already started.");
196+
return;
197+
}
184198

185-
if (scheduledExecutorService.isShutdown()) {
186-
logger.warn("Not starting. Already in shutdown.");
187-
return;
188-
}
199+
if (scheduledExecutorService.isShutdown()) {
200+
logger.warn("Not starting. Already in shutdown.");
201+
return;
202+
}
189203

190-
Runnable runnable = new ProjectConfigFetcher();
191-
scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(runnable, 0, period, timeUnit);
192-
started = true;
204+
Runnable runnable = new ProjectConfigFetcher();
205+
scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(runnable, 0, period, timeUnit);
206+
started = true;
207+
} finally {
208+
lock.unlock();
209+
}
193210
}
194211

195-
public synchronized void stop() {
196-
if (!started) {
197-
logger.warn("Not pausing. Manager has not been started.");
198-
return;
199-
}
212+
public void stop() {
213+
lock.lock();
214+
try {
215+
if (!started) {
216+
logger.warn("Not pausing. Manager has not been started.");
217+
return;
218+
}
200219

201-
if (scheduledExecutorService.isShutdown()) {
202-
logger.warn("Not pausing. Already in shutdown.");
203-
return;
204-
}
220+
if (scheduledExecutorService.isShutdown()) {
221+
logger.warn("Not pausing. Already in shutdown.");
222+
return;
223+
}
205224

206-
logger.info("pausing project watcher");
207-
scheduledFuture.cancel(true);
208-
started = false;
225+
logger.info("pausing project watcher");
226+
scheduledFuture.cancel(true);
227+
started = false;
228+
} finally {
229+
lock.unlock();
230+
}
209231
}
210232

211233
@Override
212-
public synchronized void close() {
213-
stop();
214-
scheduledExecutorService.shutdownNow();
215-
started = false;
234+
public void close() {
235+
lock.lock();
236+
try {
237+
stop();
238+
scheduledExecutorService.shutdownNow();
239+
started = false;
240+
} finally {
241+
lock.unlock();
242+
}
216243
}
217244

218245
protected void setSdkKey(String sdkKey) {

‎core-api/src/main/java/com/optimizely/ab/event/BatchEventProcessor.java

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.optimizely.ab.event.internal.UserEvent;
2222
import com.optimizely.ab.internal.PropertyUtils;
2323
import com.optimizely.ab.notification.NotificationCenter;
24+
import java.util.concurrent.locks.ReentrantLock;
2425
import org.slf4j.Logger;
2526
import org.slf4j.LoggerFactory;
2627

@@ -67,6 +68,7 @@ public class BatchEventProcessor implements EventProcessor, AutoCloseable {
6768

6869
private Future<?> future;
6970
private boolean isStarted = false;
71+
private final ReentrantLock lock = new ReentrantLock();
7072

7173
private BatchEventProcessor(BlockingQueue<Object> eventQueue, EventHandler eventHandler, Integer batchSize, Long flushInterval, Long timeoutMillis, ExecutorService executor, NotificationCenter notificationCenter) {
7274
this.eventHandler = eventHandler;
@@ -78,15 +80,20 @@ private BatchEventProcessor(BlockingQueue<Object> eventQueue, EventHandler event
7880
this.executor = executor;
7981
}
8082

81-
public synchronized void start() {
82-
if (isStarted) {
83-
logger.info("Executor already started.");
84-
return;
85-
}
83+
public void start() {
84+
lock.lock();
85+
try {
86+
if (isStarted) {
87+
logger.info("Executor already started.");
88+
return;
89+
}
8690

87-
isStarted = true;
88-
EventConsumer runnable = new EventConsumer();
89-
future = executor.submit(runnable);
91+
isStarted = true;
92+
EventConsumer runnable = new EventConsumer();
93+
future = executor.submit(runnable);
94+
} finally {
95+
lock.unlock();
96+
}
9097
}
9198

9299
@Override

‎core-api/src/main/java/com/optimizely/ab/internal/DefaultLRUCache.java

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,11 @@
1919
import com.optimizely.ab.annotations.VisibleForTesting;
2020

2121
import java.util.*;
22+
import java.util.concurrent.locks.ReentrantLock;
2223

2324
public class DefaultLRUCache<T> implements Cache<T> {
2425

25-
private final Object lock = new Object();
26+
private final ReentrantLock lock = new ReentrantLock();
2627

2728
private final Integer maxSize;
2829

@@ -51,8 +52,11 @@ public void save(String key, T value) {
5152
return;
5253
}
5354

54-
synchronized (lock) {
55+
lock.lock();
56+
try {
5557
linkedHashMap.put(key, new CacheEntity(value));
58+
} finally {
59+
lock.unlock();
5660
}
5761
}
5862

@@ -62,7 +66,8 @@ public T lookup(String key) {
6266
return null;
6367
}
6468

65-
synchronized (lock) {
69+
lock.lock();
70+
try {
6671
if (linkedHashMap.containsKey(key)) {
6772
CacheEntity entity = linkedHashMap.get(key);
6873
Long nowMs = new Date().getTime();
@@ -75,12 +80,17 @@ public T lookup(String key) {
7580
linkedHashMap.remove(key);
7681
}
7782
return null;
83+
} finally {
84+
lock.unlock();
7885
}
7986
}
8087

8188
public void reset() {
82-
synchronized (lock) {
89+
lock.lock();
90+
try {
8391
linkedHashMap.clear();
92+
} finally {
93+
lock.unlock();
8494
}
8595
}
8696

‎core-api/src/main/java/com/optimizely/ab/notification/NotificationManager.java

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
*/
1717
package com.optimizely.ab.notification;
1818

19+
import java.util.concurrent.locks.ReentrantLock;
1920
import org.slf4j.Logger;
2021
import org.slf4j.LoggerFactory;
2122

@@ -36,6 +37,7 @@ public class NotificationManager<T> {
3637

3738
private final Map<Integer, NotificationHandler<T>> handlers = Collections.synchronizedMap(new LinkedHashMap<>());
3839
private final AtomicInteger counter;
40+
private final ReentrantLock lock = new ReentrantLock();
3941

4042
public NotificationManager() {
4143
this(new AtomicInteger());
@@ -48,13 +50,16 @@ public NotificationManager(AtomicInteger counter) {
4850
public int addHandler(NotificationHandler<T> newHandler) {
4951

5052
// Prevent registering a duplicate listener.
51-
synchronized (handlers) {
53+
lock.lock();
54+
try {
5255
for (NotificationHandler<T> handler : handlers.values()) {
5356
if (handler.equals(newHandler)) {
5457
logger.warn("Notification listener was already added");
5558
return -1;
5659
}
5760
}
61+
} finally {
62+
lock.unlock();
5863
}
5964

6065
int notificationId = counter.incrementAndGet();
@@ -64,14 +69,17 @@ public int addHandler(NotificationHandler<T> newHandler) {
6469
}
6570

6671
public void send(final T message) {
67-
synchronized (handlers) {
72+
lock.lock();
73+
try {
6874
for (Map.Entry<Integer, NotificationHandler<T>> handler: handlers.entrySet()) {
6975
try {
7076
handler.getValue().handle(message);
7177
} catch (Exception e) {
7278
logger.warn("Catching exception sending notification for class: {}, handler: {}", message.getClass(), handler.getKey());
7379
}
7480
}
81+
} finally {
82+
lock.unlock();
7583
}
7684
}
7785

0 commit comments

Comments
 (0)