replace synchronized methods of TaskDataStore with ReadWriteLock
Change-Id: Iff82eae848243acf98f3e1a2e4aff8b6de254d3c
diff --git a/org.eclipse.mylyn.tasks.core.tests/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStoreTest.java b/org.eclipse.mylyn.tasks.core.tests/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStoreTest.java
new file mode 100644
index 0000000..73a9591
--- /dev/null
+++ b/org.eclipse.mylyn.tasks.core.tests/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStoreTest.java
@@ -0,0 +1,333 @@
+/*******************************************************************************
+ * Copyright (c) 2017 Tasktop Technologies and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * Tasktop Technologies - initial API and implementation
+ *******************************************************************************/
+
+package org.eclipse.mylyn.internal.tasks.core.data;
+
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import org.eclipse.core.runtime.CoreException;
+import org.eclipse.mylyn.internal.tasks.core.TaskRepositoryManager;
+import org.eclipse.mylyn.tasks.core.TaskRepository;
+import org.eclipse.mylyn.tasks.core.data.ITaskDataWorkingCopy;
+import org.eclipse.mylyn.tasks.core.data.TaskAttributeMapper;
+import org.eclipse.mylyn.tasks.core.data.TaskData;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.xml.sax.SAXException;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.ByteStreams;
+
+public class TaskDataStoreTest {
+
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder();
+
+ private static final int THREAD_COUNT = 10;
+
+ private static final long DELAY = 100;
+
+ private static final String DATA_XML_CONTENT = "test";
+
+ private static final TaskData TASK_DATA = newTaskData();
+
+ private static final TaskDataState TEST_STATE = new TaskDataState("connectorKind", "repositoryUrl", "taskId");
+
+ @Test
+ public void getTaskDataStateByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(getTaskDataState(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void discardEditsByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(discardEdits(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void putEditsByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(putEdits(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void putTaskDataSetLastReadUserByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(putTaskData(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void setTaskDataByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(setTaskData(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void putTaskDataByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(putTaskDataState(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void deleteTaskDataByMultipleThreads() throws Exception {
+ File file = newTaskDataZipFile();
+
+ List<Throwable> failures = runSameByMultipleThreads(deleteTaskData(file));
+
+ assertNoFailures(failures);
+ }
+
+ @Test
+ public void manipulateTaskDataConcurrently() throws Exception {
+ File file = newTaskDataZipFile();
+
+ // run multiple times to increase chance of a failure
+ for (int i = 0; i <= 10; i++) {
+ List<Throwable> failures = runConcurrently( //
+ getTaskDataState(file), //
+ discardEdits(file), //
+ putEdits(file), //
+ putTaskData(file), //
+ setTaskData(file), //
+ putTaskDataState(file), //
+ deleteTaskData(file));
+ assertNoFailures(failures);
+ }
+ }
+
+ @Test
+ public void manipulateTaskDataByFixedNumberOfThreads() throws Exception {
+ File file = newTaskDataZipFile();
+ TaskDataStore store = newTaskDataStore();
+ List<Throwable> failures = new ArrayList<>();
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+
+ List<Thread> threads = new ArrayList<>();
+ threads.addAll(Collections.nCopies(3, thread(getTaskDataState(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(discardEdits(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(putEdits(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(putTaskData(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(setTaskData(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(putTaskDataState(file), store, Optional.empty(), failures).get()));
+ threads.addAll(Collections.nCopies(3, thread(deleteTaskData(file), store, Optional.empty(), failures).get()));
+ Collections.shuffle(threads);
+
+ threads.stream().forEach(thread -> executor.submit(thread));
+
+ executor.shutdown();
+
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ assertNoFailures(failures);
+ }
+
+ private static TaskDataStore newTaskDataStore() {
+ TaskRepositoryManager manager = new TaskRepositoryManager();
+ TaskDataExternalizer externalizer = new TaskDataExternalizer(manager) {
+ @Override
+ public TaskDataState readState(InputStream in) throws IOException, SAXException {
+ // read the input stream fully but do not return the result, not relevant for the purpose of the test
+ assertEquals(DATA_XML_CONTENT, new String(ByteStreams.toByteArray(in), Charsets.UTF_8));
+ in.close();
+ // pretend that reading state takes more than the blink of an eye
+ sleep();
+ return TEST_STATE;
+ };
+
+ @Override
+ public void writeState(OutputStream out, ITaskDataWorkingCopy state) throws IOException {
+ out.write(DATA_XML_CONTENT.getBytes(Charsets.UTF_8));
+ // pretend that writing state takes more than the blink of an eye
+ sleep();
+ }
+ };
+ return new TaskDataStore(externalizer);
+ }
+
+ private static void sleep() {
+ try {
+ Thread.sleep(DELAY);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private File newTaskDataZipFile() throws IOException {
+ File file = folder.newFile("test.zip");
+ try (ZipOutputStream outputStream = new ZipOutputStream(new FileOutputStream(file))) {
+ ZipEntry entry = new ZipEntry("data.xml");
+ outputStream.putNextEntry(entry);
+ byte[] data = DATA_XML_CONTENT.getBytes();
+ outputStream.write(data, 0, data.length);
+ outputStream.closeEntry();
+ }
+ return file;
+ }
+
+ private static TaskData newTaskData() {
+ TaskRepository repository = new TaskRepository("connectorKind", "repositoryUrl");
+ TaskAttributeMapper mapper = new TaskAttributeMapper(repository);
+ return new TaskData(mapper, "connectorKind", "repositoryUrl", "taskId");
+ }
+
+ @FunctionalInterface
+ public interface ConsumerWithCoreException<T> {
+ public void accept(T t) throws CoreException;
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> getTaskDataState(File file) {
+ return store -> store.getTaskDataState(file);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> discardEdits(File file) {
+ return store -> store.discardEdits(file);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> putEdits(File file) {
+ return store -> store.putEdits(file, TASK_DATA);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> putTaskData(File file) {
+ return store -> store.putTaskData(file, TASK_DATA, true, true);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> setTaskData(File file) {
+ return store -> store.setTaskData(file, TASK_DATA);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> putTaskDataState(File file) {
+ return store -> store.putTaskData(file, TEST_STATE);
+ }
+
+ private ConsumerWithCoreException<TaskDataStore> deleteTaskData(File file) {
+ return store -> store.deleteTaskData(file);
+ }
+
+ private static List<Throwable> runSameByMultipleThreads(ConsumerWithCoreException<TaskDataStore> consumer)
+ throws Exception {
+ TaskDataStore store = newTaskDataStore();
+ CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT + 1);
+ List<Throwable> failures = new ArrayList<>(THREAD_COUNT);
+
+ List<Thread> threads = Stream.generate(thread(consumer, store, Optional.of(barrier), failures))
+ .limit(THREAD_COUNT)
+ .collect(toList());
+
+ startThreadsSimultaneously(barrier, threads);
+ return failures;
+ }
+
+ @SafeVarargs
+ private static List<Throwable> runConcurrently(ConsumerWithCoreException<TaskDataStore>... consumers)
+ throws Exception {
+ TaskDataStore store = newTaskDataStore();
+ CyclicBarrier barrier = new CyclicBarrier(consumers.length + 1);
+ List<Throwable> failures = new ArrayList<>(consumers.length);
+
+ List<Thread> threads = Arrays.asList(consumers)
+ .stream()
+ .map(c -> thread(c, store, Optional.of(barrier), failures).get())
+ .collect(toList());
+
+ startThreadsSimultaneously(barrier, threads);
+ return failures;
+ }
+
+ private static void startThreadsSimultaneously(CyclicBarrier barrier, List<Thread> threads) throws Exception {
+ threads.stream().forEach(Thread::start);
+ barrier.await();
+ for (Thread thread : threads) {
+ thread.join();
+ }
+ }
+
+ private static Supplier<Thread> thread(ConsumerWithCoreException<TaskDataStore> consumer, TaskDataStore store,
+ Optional<CyclicBarrier> barrier, List<Throwable> failures) {
+ return new Supplier<Thread>() {
+
+ @Override
+ public Thread get() {
+ return new Thread(new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ barrier.ifPresent(b -> {
+ try {
+ b.await();
+ } catch (InterruptedException | BrokenBarrierException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ consumer.accept(store);
+ } catch (Throwable e) {
+ failures.add(e);
+ }
+ }
+ });
+ }
+ };
+ }
+
+ private static void assertNoFailures(List<Throwable> failures) {
+ assertTrue(format("expected no failures but found %d:\n%s", failures.size(), collectMessages(failures)),
+ failures.isEmpty());
+ }
+
+ private static String collectMessages(List<Throwable> failures) {
+ return failures.stream().map(e -> e.getClass().getSimpleName()).collect(Collectors.joining("\n"));
+ }
+
+}
diff --git a/org.eclipse.mylyn.tasks.core/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStore.java b/org.eclipse.mylyn.tasks.core/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStore.java
index defad4d..85c6107 100644
--- a/org.eclipse.mylyn.tasks.core/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStore.java
+++ b/org.eclipse.mylyn.tasks.core/src/org/eclipse/mylyn/internal/tasks/core/data/TaskDataStore.java
@@ -17,6 +17,8 @@
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import java.util.zip.ZipOutputStream;
@@ -40,11 +42,17 @@
private final TaskDataExternalizer externalizer;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
public TaskDataStore(IRepositoryManager taskRepositoryManager) {
- this.externalizer = new TaskDataExternalizer(taskRepositoryManager);
+ this(new TaskDataExternalizer(taskRepositoryManager));
}
- public synchronized TaskDataState discardEdits(File file) throws CoreException {
+ TaskDataStore(TaskDataExternalizer externalizer) {
+ this.externalizer = externalizer;
+ }
+
+ public TaskDataState discardEdits(File file) throws CoreException {
TaskDataState state = readState(file);
if (state != null) {
state.setEditsData(null);
@@ -53,11 +61,11 @@
return state;
}
- public synchronized TaskDataState getTaskDataState(File file) throws CoreException {
+ public TaskDataState getTaskDataState(File file) throws CoreException {
return readState(file);
}
- public synchronized void putEdits(File file, TaskData data) throws CoreException {
+ public void putEdits(File file, TaskData data) throws CoreException {
Assert.isNotNull(file);
Assert.isNotNull(data);
TaskDataState state = readState(file);
@@ -68,8 +76,7 @@
writeState(file, state);
}
- public synchronized TaskDataState putTaskData(File file, TaskData data, boolean setLastRead, boolean user)
- throws CoreException {
+ public TaskDataState putTaskData(File file, TaskData data, boolean setLastRead, boolean user) throws CoreException {
Assert.isNotNull(file);
Assert.isNotNull(data);
TaskDataState state = null;
@@ -93,7 +100,7 @@
return state;
}
- public synchronized TaskDataState setTaskData(File file, TaskData data) throws CoreException {
+ public TaskDataState setTaskData(File file, TaskData data) throws CoreException {
Assert.isNotNull(file);
Assert.isNotNull(data);
@@ -123,6 +130,7 @@
}
private TaskDataState readState(File file) throws CoreException {
+ lock.readLock().lock();
try {
if (file.exists()) {
try {
@@ -146,10 +154,13 @@
} catch (IOException e) {
throw new CoreException(new Status(IStatus.ERROR, ITasksCoreConstants.ID_PLUGIN, "Error reading task data", //$NON-NLS-1$
e));
+ } finally {
+ lock.readLock().unlock();
}
}
private void writeState(File file, TaskDataState state) throws CoreException {
+ lock.writeLock().lock();
try {
try (ZipOutputStream out = new ZipOutputStream(new BufferedOutputStream(new FileOutputStream(file)))) {
out.setMethod(ZipOutputStream.DEFLATED);
@@ -162,15 +173,22 @@
} catch (IOException e) {
throw new CoreException(new Status(IStatus.ERROR, ITasksCoreConstants.ID_PLUGIN, "Error writing task data", //$NON-NLS-1$
e));
+ } finally {
+ lock.writeLock().unlock();
}
}
- public synchronized void putTaskData(File file, TaskDataState state) throws CoreException {
+ public void putTaskData(File file, TaskDataState state) throws CoreException {
writeState(file, state);
}
- public synchronized boolean deleteTaskData(File file) {
- return file.delete();
+ public boolean deleteTaskData(File file) {
+ lock.writeLock().lock();
+ try {
+ return file.delete();
+ } finally {
+ lock.writeLock().unlock();
+ }
}
}
