Author: tolusha Date: 2011-03-18 10:13:57 -0400 (Fri, 18 Mar 2011) New Revision: 4133 Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractBackupJob.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractIncrementalBackupJob.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupChainImpl.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupManagerImpl.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/fs/IncrementalBackupJob.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/AbstractWorkspaceDataReceiver.java jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/WorkspaceDataTransmitter.java jcr/trunk/exo.jcr.component.ext/src/test/java/org/exoplatform/services/jcr/ext/backup/AbstractBackupUseCasesTest.java Log: EXOJCR-1081: * get FileCleaner from FileCleanerHolder * made IncrementalBackupJob work properly in multi-threads * comment out excluded tests Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractBackupJob.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractBackupJob.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractBackupJob.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -41,7 +41,7 @@ public abstract class AbstractBackupJob implements BackupJob { - protected int state; + protected volatile int state; protected int id; Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractIncrementalBackupJob.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractIncrementalBackupJob.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/AbstractIncrementalBackupJob.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -20,14 +20,17 @@ import org.exoplatform.services.jcr.dataflow.ItemStateChangesLog; import org.exoplatform.services.jcr.dataflow.persistent.ItemsPersistenceListener; +import org.exoplatform.services.jcr.impl.backup.SuspendException; import org.exoplatform.services.log.ExoLogger; import org.exoplatform.services.log.Log; -import java.io.FileNotFoundException; import java.io.IOException; import java.net.URL; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; /** * Created by The eXo Platform SARL .<br/> @@ -41,11 +44,24 @@ private static Log log = ExoLogger.getLogger("exo.jcr.component.ext.IncrementalBackupJob"); + /** + * Should be synchronized collection. + */ protected final List<ItemStateChangesLog> suspendBuffer; + /** + * The amount of current working threads. + */ + protected AtomicInteger workingThreads = new AtomicInteger(); + + /** + * Allows to make all threads waiting. + */ + protected CountDownLatch latcher = null; + public AbstractIncrementalBackupJob() { - this.suspendBuffer = new ArrayList<ItemStateChangesLog>(); + this.suspendBuffer = Collections.synchronizedList(new ArrayList<ItemStateChangesLog>()); this.id = 1; notifyListeners(); @@ -57,24 +73,45 @@ } /** - * @see org.exoplatform.services.jcr.dataflow.persistent.ItemsPersistenceListener#onSaveItems(org.exoplatform.services.jcr.dataflow.ItemStateChangesLog) + * {@inheritDoc} */ public void onSaveItems(ItemStateChangesLog chlog) { if (state == FINISHED) + { return; + } else if (state == WAITING) + { suspendBuffer.add(chlog); + } else if (state == WORKING) + { try { + // wait while suspended buffer is not clear + if (latcher != null && latcher.getCount() != 0) + { + latcher.await(); + } + + workingThreads.incrementAndGet(); + save(chlog); + + workingThreads.decrementAndGet(); } catch (IOException e) { log.error("Incremental backup: Can't save log ", e); notifyError("Incremental backup: Can't save log ", e); } + catch (InterruptedException e) + { + log.error("Incremental backup: Can't save log ", e); + notifyError("Incremental backup: Can't save log ", e); + } + } } /** @@ -101,30 +138,50 @@ { try { + // waiting until all current working threads finished + while (workingThreads.get() != 0) + { + try + { + Thread.sleep(50); + } + catch (InterruptedException e) + { + throw new SuspendException(e); + } + } + + // nobody write into storage (all threads add changes logs to suspendBuffer), so create new one url = createStorage(); + + // make all threads wait on save, that allow flush suspendBuffer + latcher = new CountDownLatch(1); + state = WORKING; + for (ItemStateChangesLog log : suspendBuffer) { save(log); } - suspendBuffer.clear(); - state = WORKING; if (config.getIncrementalJobNumber() != 0 && id == config.getIncrementalJobNumber() + 1) + { state = FINISHED; - - notifyListeners(); + } } - catch (FileNotFoundException e) + catch (Throwable e) { log.error("Incremental backup: resume failed ", e); notifyError("Incremental backup: resume failed ", e); } - catch (IOException e) + finally { - log.error("Incremental backup: resume failed +", e); - notifyError("Incremental backup: resume failed ", e); + // all threads now continue their work + suspendBuffer.clear(); + latcher.countDown(); } + notifyListeners(); + return url; } Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupChainImpl.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupChainImpl.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupChainImpl.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -313,6 +313,7 @@ { Thread starter = new Thread() { + @Override public void run() { restartIncrementalBackup(); Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupManagerImpl.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupManagerImpl.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/BackupManagerImpl.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -30,6 +30,7 @@ import org.exoplatform.services.jcr.config.SimpleParameterEntry; import org.exoplatform.services.jcr.config.WorkspaceEntry; import org.exoplatform.services.jcr.config.WorkspaceInitializerEntry; +import org.exoplatform.services.jcr.core.ManageableRepository; import org.exoplatform.services.jcr.core.WorkspaceContainerFacade; import org.exoplatform.services.jcr.ext.backup.BackupChain; import org.exoplatform.services.jcr.ext.backup.BackupChainLog; @@ -56,6 +57,7 @@ import org.exoplatform.services.jcr.impl.core.SysViewWorkspaceInitializer; import org.exoplatform.services.jcr.impl.dataflow.persistent.WorkspacePersistentDataManager; import org.exoplatform.services.jcr.impl.util.io.FileCleaner; +import org.exoplatform.services.jcr.impl.util.io.FileCleanerHolder; import org.exoplatform.services.jcr.util.IdGenerator; import org.exoplatform.services.log.ExoLogger; import org.exoplatform.services.log.Log; @@ -431,8 +433,21 @@ currentRepositoryBackups = Collections.synchronizedSet(new HashSet<RepositoryBackupChain>()); - fileCleaner = new FileCleaner(10000); + // get FileCleaner from container's component FileCleanerHolder + try + { + ManageableRepository repository = repoService.getCurrentRepository(); + String workspaceName = repository.getConfiguration().getSystemWorkspaceName(); + this.fileCleaner = + ((FileCleanerHolder)repository.getWorkspaceContainer(workspaceName).getComponent(FileCleanerHolder.class)) + .getFileCleaner(); + } + catch (RepositoryException e) + { + // do nothing. should not happens + } + messages = new BackupMessagesLog(MESSAGES_MAXSIZE); scheduler = new BackupScheduler(this, messages); Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/fs/IncrementalBackupJob.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/fs/IncrementalBackupJob.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/backup/impl/fs/IncrementalBackupJob.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -29,6 +29,7 @@ import org.exoplatform.services.jcr.ext.backup.impl.PendingChangesLog; import org.exoplatform.services.jcr.ext.replication.FixupStream; import org.exoplatform.services.jcr.impl.util.io.FileCleaner; +import org.exoplatform.services.jcr.impl.util.io.FileCleanerHolder; import org.exoplatform.services.log.ExoLogger; import org.exoplatform.services.log.Log; @@ -58,7 +59,6 @@ public IncrementalBackupJob() { - fileCleaner = new FileCleaner(10000); } public void init(ManageableRepository repository, String workspaceName, BackupConfig config, Calendar timeStamp) @@ -67,6 +67,9 @@ this.workspaceName = workspaceName; this.config = config; this.timeStamp = timeStamp; + this.fileCleaner = + ((FileCleanerHolder)repository.getWorkspaceContainer(workspaceName).getComponent(FileCleanerHolder.class)) + .getFileCleaner(); try { @@ -102,6 +105,7 @@ File backupFileData = fnp.getNextFile(); oosFileData = new ObjectOutputStream(PrivilegedFileHelper.fileOutputStream(backupFileData)); + return new URL("file:" + backupFileData.getAbsoluteFile()); } Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/AbstractWorkspaceDataReceiver.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/AbstractWorkspaceDataReceiver.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/AbstractWorkspaceDataReceiver.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -120,6 +120,7 @@ */ public AbstractWorkspaceDataReceiver() throws RepositoryConfigurationException { + // TODO: need to use FileCleaner from FileCleanerHolder this.fileCleaner = new FileCleaner(ReplicationService.FILE_CLEANRE_TIMEOUT); mapPendingBinaryFile = new HashMap<String, PendingBinaryFile>(); Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/WorkspaceDataTransmitter.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/WorkspaceDataTransmitter.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/replication/WorkspaceDataTransmitter.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -86,6 +86,7 @@ public WorkspaceDataTransmitter(CacheableWorkspaceDataManager dataManager) throws RepositoryConfigurationException { dataManager.addItemPersistenceListener(this); + // TODO: need to use FileCleaner from FileCleanerHolder this.fileCleaner = new FileCleaner(ReplicationService.FILE_CLEANRE_TIMEOUT); } Modified: jcr/trunk/exo.jcr.component.ext/src/test/java/org/exoplatform/services/jcr/ext/backup/AbstractBackupUseCasesTest.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/test/java/org/exoplatform/services/jcr/ext/backup/AbstractBackupUseCasesTest.java 2011-03-18 08:25:20 UTC (rev 4132) +++ jcr/trunk/exo.jcr.component.ext/src/test/java/org/exoplatform/services/jcr/ext/backup/AbstractBackupUseCasesTest.java 2011-03-18 14:13:57 UTC (rev 4133) @@ -427,7 +427,7 @@ fail("There are no backup files in " + backDir.getAbsolutePath()); } - public void _testAutoStopBackupFull() throws Exception + public void testAutoStopBackupFull() throws Exception { // backup File backDir = new File("target/backup/" + workspaceNameToBackup + getUUIndex()); @@ -455,7 +455,7 @@ fail("The backup with id '" + chain.getBackupId() + "' should not be active"); } - public void _testAutoStopBackupIncr() throws Exception + public void testAutoStopBackupIncr() throws Exception { // backup File backDir = new File("target/backup/" + workspaceNameToBackup + getUUIndex()); @@ -491,7 +491,7 @@ backup.stopBackup(bch); } - public void _testAutoStopBackupIncrRepetion() throws Exception + public void testAutoStopBackupIncrRepetion() throws Exception { // backup File backDir = new File("target/backup/" + workspaceNameToBackup + getUUIndex()); @@ -1102,7 +1102,7 @@ fail("There are no backup files in " + backDir.getAbsolutePath()); } - public void _testAutoStopRepositoryBackupIncrRepetion() throws Exception + public void testAutoStopRepositoryBackupIncrRepetion() throws Exception { // backup File backDir = new File("target/backup/" + getUUIndex());