У меня есть приложение Java, работающее в Payara 5.
Мне нужно объединить некоторые объекты движка (из библиотеки), которые будут использовать мои компоненты.
Создание двигателей должно быть сделано в отдельном потоке.
Для этого я разработал свой EnginePool и мой EngineProducer.
Идея состоит в том, что EnginePool управляет двумя BlockingQueues. Один для доступных двигателей и один для двигателей, которые использовались компонентом и должны снова стать доступными.
EnginePool должен быть доступен только один раз, поэтому он является синглтоном.
@Singleton
@Startup
@LocalBean
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public class AbbyyEnginePool implements OcrEnginePool<IEngine> {
private static final Logger logger = LoggerFactory.getLogger(AbbyyEnginePool.class);
@Resource(lookup = "java:comp/DefaultManagedThreadFactory")
private ManagedThreadFactory threadFactory;
private static final int DEFAULT_ENGINE_COUNT = 3;
private BlockingQueue<EngineMetaInfo> availableEngines = new ArrayBlockingQueue<>(DEFAULT_ENGINE_COUNT);
private BlockingQueue<IEngine> enginesToRelease = new ArrayBlockingQueue<>(DEFAULT_ENGINE_COUNT);
private Map<IEngine, IEngine> proxiesMapping = new ConcurrentHashMap<>(DEFAULT_ENGINE_COUNT);
private int poolSize;
public AbbyyEnginePool() {
this(DEFAULT_ENGINE_COUNT);
}
public AbbyyEnginePool(int poolSize) {
this.poolSize = poolSize;
availableEngines = new ArrayBlockingQueue<>(poolSize);
enginesToRelease = new ArrayBlockingQueue<>(poolSize);
proxiesMapping = new ConcurrentHashMap<>(poolSize);
}
void setThreadFactory(ManagedThreadFactory threadFactory) {
this.threadFactory = threadFactory;
}
@PostConstruct
void init() {
EngineProducer engineProducer = new EngineProducer(availableEngines, enginesToRelease, poolSize);
Thread engineProducerThread = threadFactory.newThread(engineProducer);
engineProducerThread.setName("engineProducer");
engineProducerThread.start();
}
@Override
public IEngine get() throws EngineException {
try {
EngineMetaInfo engineMetaInfo = availableEngines.take();
IEngine engineProxy = IEngine.UnmarshalInterface(engineMetaInfo.engineHandle);
proxiesMapping.put(engineProxy, engineMetaInfo.engine);
return engineProxy;
} catch (InterruptedException e) {
throw new EngineException("Could not retrieve engine", e);
}
}
@Override
public void release(IEngine engineProxy) throws EngineException {
if (engineProxy != null) {
synchronized (proxiesMapping) {
if (proxiesMapping.containsKey(engineProxy)) {
try {
IEngine engine = proxiesMapping.remove(engineProxy);
enginesToRelease.put(engine);
} catch (Exception e) {
throw new EngineException("Could not release engine proxy.");
}
} else {
logger.warn("Engine proxy was not registered. Could not release proxy.");
}
}
}
}
static class EngineMetaInfo {
long engineHandle;
IEngine engine;
EngineMetaInfo(long engineHandle, IEngine engine) {
this.engineHandle = engineHandle;
this.engine = engine;
}
}
}
EngineProducer выглядит так:
public class EngineProducer implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(EngineProducer.class);
private static final String PROJECT_ID = "someId";
private final Integer MAX_ENGINE_COUNT;
private final BlockingQueue<AbbyyEnginePool.EngineMetaInfo> availableEngines;
private final BlockingQueue<IEngine> enginesToRelease;
private Boolean isRunning = Boolean.FALSE;
private List<EngineHolder> enginesHolder;
public EngineProducer(BlockingQueue<AbbyyEnginePool.EngineMetaInfo> availableEngines,
BlockingQueue<IEngine> enginesToRelease,
Integer maxEnginesCount) {
this.availableEngines = availableEngines;
this.enginesToRelease = enginesToRelease;
this.MAX_ENGINE_COUNT = maxEnginesCount;
this.enginesHolder = new ArrayList<>(MAX_ENGINE_COUNT);
}
private void initEngines() {
synchronized (availableEngines) {
if (availableEngines.size() == 0) {
try {
for (int i = 0; i < MAX_ENGINE_COUNT; i++) {
EngineHolder engineHolder = new EngineHolder(PROJECT_ID);
enginesHolder.add(engineHolder);
IEngine engine = engineHolder.getAndLockEngine();
long engineHandle = engine.MarshalInterface();
AbbyyEnginePool.EngineMetaInfo engineMetaInfo = new AbbyyEnginePool.EngineMetaInfo(engineHandle, engine);
availableEngines.put(engineMetaInfo);
}
logger.info("{} abbyy engines prepared for processing", MAX_ENGINE_COUNT);
EnginePool.setInitialized(Boolean.TRUE);
isRunning = Boolean.TRUE;
} catch (Exception e) {
logger.error("Could not instantiate engines.", e);
}
}
}
}
@Override
public void run() {
try {
initEngines();
while(isRunning) {
IEngine engineProxyToRelease = enginesToRelease.take();
releaseEngine(engineProxyToRelease);
}
availableEngines.clear();
for(int i = 0; i < enginesHolder.size(); i++) {
enginesHolder.get(i).unloadEngine();
}
} catch (Exception e) {
logger.error("EngineProducer encounter a problem.", e);
}
}
public void unloadEngines() {
isRunning = Boolean.FALSE;
}
private void releaseEngine( IEngine engineToRelease ) {
for (EngineHolder engineHolder : enginesHolder) {
if (engineHolder.containsEngine(engineToRelease)) {
engineHolder.unlockEngine();
IEngine engine = engineHolder.getAndLockEngine();
long engineHandle = engine.MarshalInterface();
AbbyyEnginePool.EngineMetaInfo engineMetaInfo = new AbbyyEnginePool.EngineMetaInfo(engineHandle, engine);
try {
availableEngines.put(engineMetaInfo);
} catch (InterruptedException e) {
logger.warn("could not add free engine");
}
break;
}
}
}
}
Когда я запускаю это в тесте, а не в Glassfish, он работает без проблем.
Но когда я запустил его в стеклянной рыбе, бобы зашли в тупик.
Бины используют этот код для получения и выпуска двигателей:
engine = enginePool.get();
ProcessingResult processingResult = null;
try {
this.parameters = parameters;
this.tmpDir = tmpDir;
Path customProfileFile = loadProfiles(parameters);
Instant processingStart = Instant.now();
processingResult = processFile();
Instant processingEnd = Instant.now();
enginePool.release(engine);
engine = null;
processingResult.setProcessingStartTime(processingStart);
processingResult.setProcessingEndTime(processingEnd);
logger.info("Processing took about {} milliseconds.", processingResult.getProcessDurationInMilliseconds());
customProfileFile.toFile().delete();
this.tmpDir.toFile().delete();
} catch (Exception e) {
logger.error("Ocr of document failed ",e );
enginePool.release(engine);
throw new EngineException("Ocr of document failed.", e);
}
В моем сценарии есть 4 бина, которые пытаются получить движки. 3 из них получат по одному, а последний бин будет ждать на engine = enginePool.get(); 3 бина, которые получили двигатели, будут работать и ждать на enginePool.release(engine);. Я взял дамп потока и увидел, что 3 бина ожидают блокировки, которую держит бин, у которого нет движка. Поэтому они не могут выпускать двигатели.
Моя проблема в том, что я этого не понимаю. Отпускание и получение двигателей работают с различными блоками блокировки, поэтому мне интересно, почему последний компонент, ожидающий получить двигатель, блокирует другие компоненты, которые пытаются освободить двигатель.