You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ae...@apache.org on 2018/10/22 06:20:06 UTC
hadoop git commit: HDDS-544. Unconditional wait findbug warning from
ReplicationSupervisor. Contributed by Arpit Agarwal.
Repository: hadoop
Updated Branches:
refs/heads/trunk c9077a9f5 -> 7db3bb3ac
HDDS-544. Unconditional wait findbug warning from ReplicationSupervisor.
Contributed by Arpit Agarwal.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/7db3bb3a
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/7db3bb3a
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/7db3bb3a
Branch: refs/heads/trunk
Commit: 7db3bb3ac185929fbe8f5b9b59cdd157eec2ac64
Parents: c9077a9
Author: Anu Engineer <ae...@apache.org>
Authored: Sun Oct 21 23:18:38 2018 -0700
Committer: Anu Engineer <ae...@apache.org>
Committed: Sun Oct 21 23:18:38 2018 -0700
----------------------------------------------------------------------
.../statemachine/DatanodeStateMachine.java | 1 -
.../replication/ReplicationSupervisor.java | 137 +++++++++----------
.../replication/TestReplicationSupervisor.java | 31 ++---
3 files changed, 73 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7db3bb3a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
index 1bade8e..85fa304 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
@@ -308,7 +308,6 @@ public class DatanodeStateMachine implements Closeable {
public void startDaemon() {
Runnable startStateMachineTask = () -> {
try {
- supervisor.start();
start();
LOG.info("Ozone container server started.");
} catch (Exception ex) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7db3bb3a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
index 1d8d5f6..c59d643 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java
@@ -17,11 +17,13 @@
*/
package org.apache.hadoop.ozone.container.replication;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentHashMap.KeySetView;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
import org.apache.hadoop.ozone.container.replication.ReplicationTask.Status;
@@ -37,106 +39,91 @@ public class ReplicationSupervisor {
private static final Logger LOG =
LoggerFactory.getLogger(ReplicationSupervisor.class);
- private final Set<Worker> threadPool = new HashSet<>();
-
- private final Map<Long, ReplicationTask> queue = new TreeMap();
-
private final ContainerSet containerSet;
-
private final ContainerReplicator replicator;
+ private final ThreadPoolExecutor executor;
- private final int poolSize;
+ /**
+ * A set of container IDs that are currently being downloaded
+ * or queued for download. Tracked so we don't schedule > 1
+ * concurrent download for the same container.
+ */
+ private final KeySetView<Object, Boolean> containersInFlight;
public ReplicationSupervisor(
ContainerSet containerSet,
ContainerReplicator replicator, int poolSize) {
this.containerSet = containerSet;
this.replicator = replicator;
- this.poolSize = poolSize;
- }
-
- public synchronized void addTask(ReplicationTask task) {
- queue.putIfAbsent(task.getContainerId(), task);
- synchronized (threadPool) {
- threadPool.notify();
- }
+ this.containersInFlight = ConcurrentHashMap.newKeySet();
+ this.executor = new ThreadPoolExecutor(
+ 0, poolSize, 60, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new ThreadFactoryBuilder().setDaemon(true)
+ .setNameFormat("ContainerReplicationThread-%d")
+ .build());
}
- public void start() {
- for (int i = 0; i < poolSize; i++) {
- Worker worker = new Worker();
- Thread thread = new Thread(worker, "ContainerReplication-" + i);
- thread.setDaemon(true);
- thread.start();
- threadPool.add(worker);
- }
- }
-
- public synchronized ReplicationTask selectTask() {
- for (ReplicationTask task : queue.values()) {
- if (task.getStatus() == Status.QUEUED) {
- if (containerSet.getContainer(task.getContainerId()) == null) {
- task.setStatus(Status.DOWNLOADING);
- return task;
- } else {
- LOG.debug("Container {} has already been downloaded.",
- task.getContainerId());
- queue.remove(task.getContainerId());
- }
- } else if (task.getStatus() == Status.FAILED) {
- LOG.error(
- "Container {} can't be downloaded from any of the datanodes.",
- task.getContainerId());
- queue.remove(task.getContainerId());
- } else if (task.getStatus() == Status.DONE) {
- queue.remove(task.getContainerId());
- LOG.info("Container {} is replicated.", task.getContainerId());
- }
+ /**
+ * Queue an asynchronous download of the given container.
+ */
+ public void addTask(ReplicationTask task) {
+ if (containersInFlight.add(task.getContainerId())) {
+ executor.submit(new TaskRunner(task));
}
- //no available task.
- return null;
}
public void stop() {
- for (Worker worker : threadPool) {
- worker.stop();
+ try {
+ executor.shutdown();
+ if (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException ie) {
+ // Ignore, we don't really care about the failure.
+ Thread.currentThread().interrupt();
}
}
+ /**
+ * Get the number of containers currently being downloaded
+ * or scheduled for download.
+ * @return Count of in-flight replications.
+ */
@VisibleForTesting
- public int getQueueSize() {
- return queue.size();
+ public int getInFlightReplications() {
+ return containersInFlight.size();
}
- private class Worker implements Runnable {
+ private final class TaskRunner implements Runnable {
+ private final ReplicationTask task;
- private boolean running = true;
+ private TaskRunner(ReplicationTask task) {
+ this.task = task;
+ }
@Override
public void run() {
try {
- while (running) {
- ReplicationTask task = selectTask();
- if (task == null) {
- synchronized (threadPool) {
- threadPool.wait();
- }
- } else {
- replicator.replicate(task);
- }
+ if (containerSet.getContainer(task.getContainerId()) != null) {
+ LOG.debug("Container {} has already been downloaded.",
+ task.getContainerId());
+ return;
}
- } catch (Exception ex) {
- LOG.error("Error on doing replication", ex);
- try {
- Thread.sleep(200);
- } catch (InterruptedException e) {
- LOG.error("Error on waiting after failed replication task", e);
+
+ task.setStatus(Status.DOWNLOADING);
+ replicator.replicate(task);
+
+ if (task.getStatus() == Status.FAILED) {
+ LOG.error(
+ "Container {} can't be downloaded from any of the datanodes.",
+ task.getContainerId());
+ } else if (task.getStatus() == Status.DONE) {
+ LOG.info("Container {} is replicated.", task.getContainerId());
}
+ } finally {
+ containersInFlight.remove(task.getContainerId());
}
}
-
- public void stop() {
- running = false;
- }
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7db3bb3a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
index d433319..030412c 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java
@@ -29,6 +29,7 @@ import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.apache.hadoop.test.LambdaTestUtils;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -41,7 +42,7 @@ public class TestReplicationSupervisor {
private OzoneConfiguration conf = new OzoneConfiguration();
@Test
- public void normal() {
+ public void normal() throws Exception {
//GIVEN
ContainerSet set = new ContainerSet();
@@ -54,7 +55,6 @@ public class TestReplicationSupervisor {
.collect(Collectors.toList());
try {
- supervisor.start();
//WHEN
supervisor.addTask(new ReplicationTask(1L, datanodes));
supervisor.addTask(new ReplicationTask(1L, datanodes));
@@ -62,16 +62,11 @@ public class TestReplicationSupervisor {
supervisor.addTask(new ReplicationTask(2L, datanodes));
supervisor.addTask(new ReplicationTask(2L, datanodes));
supervisor.addTask(new ReplicationTask(3L, datanodes));
- try {
- Thread.sleep(300);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
//THEN
- System.out.println(replicator.replicated.get(0));
+ LambdaTestUtils.await(200_000, 1000,
+ () -> supervisor.getInFlightReplications() == 0);
- Assert
- .assertEquals(3, replicator.replicated.size());
+ Assert.assertEquals(3, replicator.replicated.size());
} finally {
supervisor.stop();
@@ -79,7 +74,7 @@ public class TestReplicationSupervisor {
}
@Test
- public void duplicateMessageAfterAWhile() throws InterruptedException {
+ public void duplicateMessageAfterAWhile() throws Exception {
//GIVEN
ContainerSet set = new ContainerSet();
@@ -92,22 +87,18 @@ public class TestReplicationSupervisor {
.collect(Collectors.toList());
try {
- supervisor.start();
//WHEN
supervisor.addTask(new ReplicationTask(1L, datanodes));
- Thread.sleep(400);
+ LambdaTestUtils.await(200_000, 1000,
+ () -> supervisor.getInFlightReplications() == 0);
supervisor.addTask(new ReplicationTask(1L, datanodes));
- Thread.sleep(300);
+ LambdaTestUtils.await(200_000, 1000,
+ () -> supervisor.getInFlightReplications() == 0);
//THEN
System.out.println(replicator.replicated.get(0));
- Assert
- .assertEquals(1, replicator.replicated.size());
-
- //the last item is still in the queue as we cleanup the queue during the
- // selection
- Assert.assertEquals(1, supervisor.getQueueSize());
+ Assert.assertEquals(1, replicator.replicated.size());
} finally {
supervisor.stop();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org