You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by am...@apache.org on 2020/01/15 05:01:03 UTC
svn commit: r1872803 - in /jackrabbit/oak/trunk/oak-blob-plugins/src:
main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java
test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java
Author: amitj
Date: Wed Jan 15 05:01:02 2020
New Revision: 1872803
URL: http://svn.apache.org/viewvc?rev=1872803&view=rev
Log:
OAK-8780 - Remove the synchronized block on the FSBackend
Slightly modified patch from Ankita Agarwal
Moved the copy of file to a temp file outside the synchronized block
Added:
jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java (with props)
Modified:
jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java
Modified: jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java?rev=1872803&r1=1872802&r2=1872803&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java (original)
+++ jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackend.java Wed Jan 15 05:01:02 2020
@@ -43,6 +43,7 @@ import org.apache.jackrabbit.core.data.D
import org.apache.jackrabbit.core.data.LazyFileInputStream;
import org.apache.jackrabbit.oak.spi.blob.AbstractDataRecord;
import org.apache.jackrabbit.oak.spi.blob.AbstractSharedBackend;
+import org.apache.jackrabbit.util.TransientFileFactory;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -104,20 +105,38 @@ public class FSBackend extends AbstractS
@Override
public void write(DataIdentifier identifier, File file) throws DataStoreException {
+ TransientFileFactory fileFactory = TransientFileFactory.getInstance();
+ File tmpFile = null;
+
File dest = getFile(identifier, fsPathDir);
- synchronized (this) {
- if (dest.exists()) {
- long now = System.currentTimeMillis();
- if (getLastModified(dest) < now + ACCESS_TIME_RESOLUTION) {
- setLastModified(dest, now + ACCESS_TIME_RESOLUTION);
+ if (dest.exists()) {
+ long now = System.currentTimeMillis();
+ if (getLastModified(dest) < now + ACCESS_TIME_RESOLUTION) {
+ setLastModified(dest, now + ACCESS_TIME_RESOLUTION);
+ }
+ } else {
+ try {
+ tmpFile = fileFactory.createTransientFile("fsbackend", null, fsPathDir);
+ FileUtils.copyFile(file, tmpFile);
+
+ File parent = dest.getParentFile();
+ parent.mkdirs();
+ synchronized (this) {
+ if (tmpFile.renameTo(dest)) {
+ // no longer need to delete the temporary file
+ tmpFile = null;
+ } else {
+ throw new IOException(
+ "Can not rename " + tmpFile.getAbsolutePath()
+ + " to " + dest.getAbsolutePath()
+ + " (media read only?)");
+ }
}
- } else {
- try {
- FileUtils.copyFile(file, dest);
- } catch (IOException ie) {
- LOG.error("failed to copy [{}] to [{}]", file.getAbsolutePath(),
- dest.getAbsolutePath());
- throw new DataStoreException("Not able to write file [" + identifier + "]", ie);
+ } catch (IOException e) {
+ throw new DataStoreException("Could not add record", e);
+ } finally {
+ if (tmpFile != null) {
+ tmpFile.delete();
}
}
}
Added: jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java?rev=1872803&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java (added)
+++ jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java Wed Jan 15 05:01:02 2020
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.jackrabbit.oak.plugins.blob.datastore;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.core.data.DataIdentifier;
+import org.apache.jackrabbit.core.data.DataRecord;
+import org.apache.jackrabbit.core.data.DataStore;
+import org.apache.jackrabbit.core.data.DataStoreException;
+import org.apache.jackrabbit.core.data.util.NamedThreadFactory;
+import org.apache.jackrabbit.oak.commons.FileIOUtils;
+import org.apache.jackrabbit.oak.commons.PropertiesUtil;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreUtils.randomStream;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class FSBackendIT {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FSBackendIT.class);
+
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder(new File("target"));
+
+ private Properties props;
+ private FSBackend backend;
+ private String dataStoreDir;
+ private DataStore ds;
+ private ListeningExecutorService executor;
+ private Random rand = new Random(0);
+
+ @Before
+ public void setUp() throws Exception {
+ dataStoreDir = folder.newFolder().getAbsolutePath();
+ props = new Properties();
+ props.setProperty("cacheSize", "0");
+ props.setProperty("fsBackendPath", dataStoreDir);
+ ds = createDataStore();
+ backend = (FSBackend) ((CachingFileDataStore) ds).getBackend();
+ this.executor = MoreExecutors.listeningDecorator(Executors
+ .newFixedThreadPool(25, new NamedThreadFactory("oak-backend-test-write-thread")));
+ }
+
+ protected DataStore createDataStore() {
+ CachingFileDataStore ds = null;
+ try {
+ ds = new CachingFileDataStore();
+ Map<String, ?> config = DataStoreUtils.getConfig();
+ props.putAll(config);
+ PropertiesUtil.populate(ds, Maps.fromProperties(props), false);
+ ds.setProperties(props);
+ ds.init(dataStoreDir);
+ } catch (Exception e) {
+ LOG.error("Exception creating DataStore", e);
+ }
+ return ds;
+ }
+
+ /**
+ * Test for write single threaded to FSBackend.
+ */
+ @Test
+ public void testSingleThreadFSBackend() {
+ try {
+ long start = System.currentTimeMillis();
+ LOG.info("Testcase: " + this.getClass().getName() + "#testSingleThread, testDir=" + dataStoreDir);
+ doTest(ds, 1, true);
+ LOG.info("Testcase: " + this.getClass().getName() + "#testSingleThread finished, time taken = [" + (
+ System.currentTimeMillis() - start) + "]ms");
+ } catch (Exception e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Tests for multi-threaded write of same file to FSBackend
+ */
+ @Test
+ public void testMultiThreadedSame() {
+ try {
+ long start = System.currentTimeMillis();
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedSame, testDir=" + dataStoreDir);
+ doTest(ds, 10, true);
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedSame finished, time taken = [" + (
+ System.currentTimeMillis() - start) + "]ms");
+ } catch (Exception e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Tests for multi-threaded write of same file to FSBackend
+ */
+ @Test
+ public void testMultiThreadedSameLarge() {
+ try {
+ long start = System.currentTimeMillis();
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedSameLarge, testDir=" + dataStoreDir);
+ doTest(ds, 100, true);
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedSameLarge finished, time taken = [" + (
+ System.currentTimeMillis() - start) + "]ms");
+ } catch (Exception e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Tests for multi-threaded write of different file to FSBackend
+ */
+ @Test
+ public void testMultiThreadedDifferent() {
+ try {
+ long start = System.currentTimeMillis();
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedDifferent, testDir=" + dataStoreDir);
+ doTest(ds, 10, false);
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedDifferent finished, time taken = [" + (
+ System.currentTimeMillis() - start) + "]ms");
+ } catch (Exception e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Tests for multi-threaded write of different file to FSBackend
+ */
+ @Test
+ public void testMultiThreadedDifferentLarge() {
+ try {
+ long start = System.currentTimeMillis();
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedDifferentLarge, testDir=" + dataStoreDir);
+ doTest(ds, 100, false);
+ LOG.info("Testcase: " + this.getClass().getName() + "#testMultiThreadedDifferentLarge finished, time taken = [" + (
+ System.currentTimeMillis() - start) + "]ms");
+ } catch (Exception e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ @After
+ public void tearDown() {
+ try {
+ new ExecutorCloser(executor).close();
+ ds.close();
+ } catch (DataStoreException e) {
+ LOG.error("error", e);
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Method to assert record while writing and deleting record from FSBackend
+ */
+ void doTest(DataStore ds, int concurrency, boolean same) throws Exception {
+ List<ListenableFuture<Integer>> futures = Lists.newArrayList();
+ CountDownLatch latch = new CountDownLatch(concurrency);
+
+ int seed = 0;
+ for (int i = 0; i < concurrency; i++) {
+ if (!same) {
+ seed = rand.nextInt(1000);
+ }
+ put(folder, futures, seed, latch);
+ }
+
+ for (int i = 0; i < concurrency; i++) {
+ latch.countDown();
+ }
+
+ assertFuture(futures);
+ }
+
+ private List<ListenableFuture<Integer>> put(TemporaryFolder folder, List<ListenableFuture<Integer>> futures,
+ int seed, CountDownLatch writeLatch)
+ throws IOException {
+
+ File f = copyToFile(randomStream(seed, 4 * 1024 * 1024), folder.newFile());
+
+ ListenableFuture<Integer> future = executor.submit(() -> {
+ try {
+ writeLatch.await();
+ backend.write(new DataIdentifier("0000ID" + seed), f);
+ LOG.info("Added file to backend");
+ return seed;
+ } catch (Exception e) {
+ LOG.error("Error adding file to backend", e);
+ throw e;
+ }
+ });
+ futures.add(future);
+
+ return futures;
+ }
+
+ private void waitFinish(List<ListenableFuture<Integer>> futures) {
+ ListenableFuture<List<Integer>> listenableFutures = Futures.successfulAsList(futures);
+ try {
+ listenableFutures.get();
+ } catch (Exception e) {
+ LOG.error("Error in finishing threads", e);
+ }
+ }
+
+ private void assertFuture(List<ListenableFuture<Integer>> futures) throws Exception {
+ waitFinish(futures);
+
+ for (ListenableFuture future : futures) {
+ assertFile((Integer) future.get(), folder);
+ }
+ }
+
+ private void assertFile(int seed, TemporaryFolder folder) throws IOException, DataStoreException {
+ DataRecord backendRecord = backend.getRecord(new DataIdentifier("0000ID" + seed));
+
+ assertEquals(backendRecord.getLength(), 4 * 1024 * 1024);
+ File original = copyToFile(randomStream(seed, 4 * 1024 * 1024), folder.newFile());
+ assertTrue("Backend file content differs",
+ FileUtils.contentEquals(original, copyToFile(backendRecord.getStream(), folder.newFile())));
+ }
+
+ static File copyToFile(InputStream stream, File file) throws IOException {
+ FileIOUtils.copyInputStreamToFile(stream, file);
+ return file;
+ }
+}
Propchange: jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/FSBackendIT.java
------------------------------------------------------------------------------
svn:eol-style = native