umamaheswararao commented on code in PR #3504:
URL: https://github.com/apache/ozone/pull/3504#discussion_r898492177
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java:
##########
@@ -347,6 +349,37 @@ public void setMaxSize(int maxSize) {
this.maxSize = maxSize;
}
+ public void setStaleThreshold(long threshold) {
+ this.staleThreshold = threshold;
+ }
+
+ }
+
+ /**
+ * Builder of XceiverClientManagerConfig.
Review Comment:
done. Initially I created another class of this. Did not want to exposed
ScmClientConfig as it's name not quite right when we use this in DN. I changed
that in java doc.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ec/reconstruction/ECReconstructionCoordinator.java:
##########
@@ -0,0 +1,417 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.container.ec.reconstruction;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.ByteStringConversion;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
+import org.apache.hadoop.hdds.scm.storage.BufferPool;
+import org.apache.hadoop.hdds.scm.storage.ECBlockOutputStream;
+import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
+import org.apache.hadoop.io.ByteBufferPool;
+import org.apache.hadoop.io.ElasticByteBufferPool;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.client.io.BlockInputStreamFactory;
+import org.apache.hadoop.ozone.client.io.BlockInputStreamFactoryImpl;
+import org.apache.hadoop.ozone.client.io.ECBlockInputStreamProxy;
+import org.apache.hadoop.ozone.client.io.ECBlockReconstructedStripeInputStream;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * The Coordinator implements the main flow of reconstructing
+ * missing container replicas.
+ * <p>
+ * For a container reconstruction task, the main flow is:
+ * - ListBlock from all healthy replicas
+ * - calculate effective block group len for all blocks
+ * - for each block
+ * - build a ReconstructInputStream to read healthy chunks
+ * - build a ECBlockOutputStream to write out decoded chunks
+ * - for each stripe
+ * - use ReconstructInputStream.readStripe to decode missing chunks
+ * - use ECBlockOutputStream.write to write decoded chunks to TargetDNs
+ * - PutBlock
+ * - CloseContainer
Review Comment:
done. Thanks
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ec/reconstruction/ECReconstructionCoordinator.java:
##########
@@ -0,0 +1,417 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.container.ec.reconstruction;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.ByteStringConversion;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
+import org.apache.hadoop.hdds.scm.storage.BufferPool;
+import org.apache.hadoop.hdds.scm.storage.ECBlockOutputStream;
+import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
+import org.apache.hadoop.io.ByteBufferPool;
+import org.apache.hadoop.io.ElasticByteBufferPool;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.client.io.BlockInputStreamFactory;
+import org.apache.hadoop.ozone.client.io.BlockInputStreamFactoryImpl;
+import org.apache.hadoop.ozone.client.io.ECBlockInputStreamProxy;
+import org.apache.hadoop.ozone.client.io.ECBlockReconstructedStripeInputStream;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * The Coordinator implements the main flow of reconstructing
+ * missing container replicas.
+ * <p>
+ * For a container reconstruction task, the main flow is:
+ * - ListBlock from all healthy replicas
+ * - calculate effective block group len for all blocks
+ * - for each block
+ * - build a ReconstructInputStream to read healthy chunks
+ * - build a ECBlockOutputStream to write out decoded chunks
+ * - for each stripe
+ * - use ReconstructInputStream.readStripe to decode missing chunks
+ * - use ECBlockOutputStream.write to write decoded chunks to TargetDNs
+ * - PutBlock
+ * - CloseContainer
+ */
+public class ECReconstructionCoordinator implements Closeable {
+
+ static final Logger LOG =
+ LoggerFactory.getLogger(ECReconstructionCoordinator.class);
+
+ private static final int EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE = 3;
+
+ private final ECContainerOperationClient containerOperationClient;
+
+ private final ConfigurationSource config;
+
+ private final ByteBufferPool byteBufferPool;
+
+ private ExecutorService ecReconstructExecutor;
+
+ private BlockInputStreamFactory blockInputStreamFactory;
+
+ public ECReconstructionCoordinator(ECContainerOperationClient
containerClient,
+ ConfigurationSource conf, ByteBufferPool byteBufferPool,
+ ExecutorService reconstructExecutor,
+ BlockInputStreamFactory streamFactory) {
+ this.containerOperationClient = containerClient;
+ this.config = conf;
+ this.byteBufferPool = byteBufferPool;
+ this.blockInputStreamFactory = streamFactory;
+ this.ecReconstructExecutor = reconstructExecutor;
+ }
+
+ public ECReconstructionCoordinator(ConfigurationSource conf,
+ CertificateClient certificateClient) throws IOException {
+ this(new ECContainerOperationClient(conf, certificateClient), conf,
+ new ElasticByteBufferPool(), null, null);
+ this.ecReconstructExecutor =
+ new ThreadPoolExecutor(EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE,
+ config.getObject(OzoneClientConfig.class)
+ .getEcReconstructStripeReadPoolLimit(), 60, TimeUnit.SECONDS,
+ new SynchronousQueue<>(), new ThreadFactoryBuilder()
+ .setNameFormat("ec-reconstruct-reader-TID-%d").build(),
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ this.blockInputStreamFactory = BlockInputStreamFactoryImpl
+ .getInstance(byteBufferPool, () -> ecReconstructExecutor);
+ }
+
+ public void reconstructECContainerGroup(long containerID,
+ ECReplicationConfig repConfig,
+ SortedMap<Integer, DatanodeDetails> sourceNodeMap,
+ SortedMap<Integer, DatanodeDetails> targetNodeMap) throws IOException {
+
+ Pipeline pipeline = rebuildInputPipeline(repConfig, sourceNodeMap);
+
+ SortedMap<Long, BlockData[]> blockDataMap =
+ getBlockDataMap(containerID, repConfig, sourceNodeMap);
+
+ SortedMap<Long, BlockLocationInfo> blockLocationInfoMap =
+ calcBlockLocationInfoMap(containerID, blockDataMap, pipeline);
+
+ // 1. create target recovering containers.
+ Set<Map.Entry<Integer, DatanodeDetails>> targetIndexDns =
+ targetNodeMap.entrySet();
+ Iterator<Map.Entry<Integer, DatanodeDetails>> iterator =
+ targetIndexDns.iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<Integer, DatanodeDetails> indexDnPair = iterator.next();
Review Comment:
sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
