Commit 6d6224fc authored by Matija Obreza's avatar Matija Obreza

Execute update of Dataset and Subset counts in parallel

parent 5ae52dc6
......@@ -63,6 +63,7 @@ import org.genesys2.server.service.TaxonomyService;
import org.genesys2.server.service.filter.AccessionFilter;
import org.genesys2.server.service.filter.AppliedFiltersConverter;
import org.genesys2.server.service.impl.GenesysFilterServiceImpl.LabelValue;
import org.genesys2.spring.TransactionHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -70,6 +71,7 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
......@@ -136,6 +138,9 @@ public class AccessionServiceImpl implements AccessionService {
@Autowired
private DatasetAndSubsetCountAspect datasetAndSubsetCountAspect;
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
private static final int LOAD_CHUNK_SIZE = 200;
private <T extends AccessionData> T lazyLoad(T accession) {
......@@ -612,19 +617,27 @@ public class AccessionServiceImpl implements AccessionService {
}
@Override
@Transactional
@PreAuthorize("hasRole('ADMINISTRATOR')")
public void scanForPublishedDatasets() {
BooleanExpression expression = QDataset.dataset.state.eq(PublishState.PUBLISHED);
datasetRepository.findAll(expression).forEach(dataset -> datasetAndSubsetCountAspect.afterDatasetPersist(dataset));
datasetRepository.findAll(expression).forEach(dataset -> {
taskExecutor.submit(() -> TransactionHelper.executeInTransaction(false, () -> {
datasetAndSubsetCountAspect.afterDatasetPersist(dataset);
return true;
}));
});
}
@Override
@Transactional
@PreAuthorize("hasRole('ADMINISTRATOR')")
public void scanForPublishedSubsets() {
BooleanExpression expression = QSubset.subset.state.eq(PublishState.PUBLISHED);
subsetRepository.findAll(expression).forEach(subset -> datasetAndSubsetCountAspect.afterSubsetPersist(subset));
subsetRepository.findAll(expression).forEach(subset -> {
taskExecutor.submit(() -> TransactionHelper.executeInTransaction(false, () -> {
datasetAndSubsetCountAspect.afterSubsetPersist(subset);
return true;
}));
});
}
@Override
......
/*
* Copyright 2020 Global Crop Diversity Trust
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.genesys2.spring;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import org.genesys.blocks.util.CurrentApplicationContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
* The TransactionHelper.
*
* @author Matija Obreza
*/
public class TransactionHelper {
public static final Logger LOG = LoggerFactory.getLogger(TransactionHelper.class);
public static <T> T asUser(Authentication authentication, Callable<T> callable) throws Exception {
Authentication prevAuth = SecurityContextHolder.getContext().getAuthentication();
SecurityContextHolder.getContext().setAuthentication(authentication);
try {
return callable.call();
} finally {
SecurityContextHolder.getContext().setAuthentication(prevAuth);
}
}
public static <T> T asCurrentUser(Callable<T> callable) throws Exception {
Authentication prevAuth = SecurityContextHolder.getContext().getAuthentication();
try {
LOG.warn("Executing as {}", prevAuth);
return asUser(prevAuth, callable);
} finally {
SecurityContextHolder.getContext().setAuthentication(prevAuth);
}
}
public static <T> T executeInTransaction(boolean readonly, Callable<T> callable) {
final ThrowableHolder throwableHolder = new ThrowableHolder();
AtomicInteger retries = new AtomicInteger(0);
// do {
if (retries.get() > 0) {
LOG.warn("Transaction retry #{}", retries.get());
}
T result = transactionTemplate(readonly).execute(new TransactionCallback<T>() {
@Override
public T doInTransaction(TransactionStatus status) {
throwableHolder.throwable = null;
try {
return callable.call();
} catch (Throwable e) {
LOG.error("Transaction failed with {}", e.getMessage());
throwableHolder.throwable = e;
status.setRollbackOnly();
return null;
}
}
});
if (throwableHolder.throwable == null) {
return result;
}
// } while (throwableHolder.throwable != null && retries.incrementAndGet() < 3);
throw new RuntimeException("Could not execute transaction", throwableHolder.throwable);
}
private static TransactionTemplate transactionTemplate(boolean readonly) {
if (CurrentApplicationContext.getContext() == null) {
throw new RuntimeException("CurrentApplicationContext is not available.");
}
PlatformTransactionManager transactionManager = CurrentApplicationContext.getContext().getBean(PlatformTransactionManager.class);
if (transactionManager == null) {
throw new RuntimeException("PlatformTransactionManager is not available");
}
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.setReadOnly(readonly);
transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
transactionTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_DEFAULT);
return transactionTemplate;
}
/**
* Internal holder class for a Throwable in a callback transaction model.
*/
private static class ThrowableHolder {
public Throwable throwable;
}
}
......@@ -90,7 +90,7 @@ public class DatabaseConfig {
private int maxActive;
// max allowed age for any connection
private final long maxAgeSeconds = 60 * 5;
private final long maxAgeSeconds = 60 * 15;
@Value("${db.pool.maxIdleSeconds}")
private int maxIdleSeconds;
......@@ -123,7 +123,7 @@ public class DatabaseConfig {
dataSource.setAbandonWhenPercentageFull(80);
dataSource.setRemoveAbandoned(true);
dataSource.setRemoveAbandonedTimeout(60);
dataSource.setSuspectTimeout(40);
dataSource.setSuspectTimeout(30);
dataSource.setJdbcInterceptors("ResetAbandonedTimer");
return dataSource;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment