You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "ChenSammi (via GitHub)" <gi...@apache.org> on 2024/09/24 08:43:32 UTC
[PR] HDDS-10685. Short circuit read support in Ozone [ozone]
ChenSammi opened a new pull request, #7236:
URL: https://github.com/apache/ozone/pull/7236
## What changes were proposed in this pull request?
support short-circuit read for datanode.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-10685
## How was this patch tested?
new unit tests
- TestDomainSocketFactory
- TestXceiverServerDomainSocket
- TestShortCircuitChunkInputStream
manual test with freon
```
sammi@SAMMICHEN-MB0 ozone-1.5.0-SNAPSHOT % bin/ozone --jvmargs -Djava.library.path=/Users/sammi/workspace/hadoop-ozone/hadoop-hdds/client/src/test/resources freon rk --num-of-volumes=1 --num-of-buckets=10 --num-of-keys=1000 --replication=ONE --replication-type=RATIS --validate-writes --validate-channel short-circuit --num-of-validate-threads=10 --key-size 1MB
SLF4J(W): Class path contains multiple SLF4J providers.
SLF4J(W): Found provider [org.slf4j.reload4j.Reload4jServiceProvider@18ef96]
SLF4J(W): Found provider [org.slf4j.simple.SimpleServiceProvider@6956de9]
SLF4J(W): See https://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J(I): Actual provider is of type [org.slf4j.reload4j.Reload4jServiceProvider@18ef96]
2024-09-24 15:37:01,844 [main] INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties
2024-09-24 15:37:01,877 [main] INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).
2024-09-24 15:37:01,877 [main] INFO impl.MetricsSystemImpl: ozone-freon metrics system started
2024-09-24 15:37:02,469 [main] INFO storage.DomainSocketFactory: Trying to load the custom-built native-hadoop library...
2024-09-24 15:37:02,469 [main] INFO storage.DomainSocketFactory: Loaded the native-hadoop library
2024-09-24 15:37:02,469 [main] INFO storage.DomainSocketFactory: short-circuit local reads is enabled within 294209 ns.
2024-09-24 15:37:02,516 [main] INFO freon.RandomKeyGenerator: Number of Threads: 10
2024-09-24 15:37:02,524 [main] INFO freon.RandomKeyGenerator: Number of Volumes: 1.
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: Number of Buckets per Volume: 10.
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: Number of Keys per Bucket: 1000.
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: Key size: 1048576 bytes
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: Buffer size: 4096 bytes
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: validateWrites : true
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: Number of Validate Threads: 10
2024-09-24 15:37:02,525 [main] INFO freon.RandomKeyGenerator: cleanObjects : false
2024-09-24 15:37:02,527 [main] INFO freon.RandomKeyGenerator: Data validation is enabled.
2024-09-24 15:37:02,527 [main] INFO freon.RandomKeyGenerator: Starting progress bar Thread.
0.00% |█ | 0/10000 Time: 0:00:00| 2024-09-24 15:37:02,550 [pool-3-thread-1] INFO rpc.RpcClient: Creating Volume: vol-0-30931, with sammi as owner and space quota set to -1 bytes, counts quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-8] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-6-04589, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-2] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-0-10826, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-5] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-3-67064, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-7] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-5-65340, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-1] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-9-04405, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,616 [pool-3-thread-4] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-2-22496, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,619 [pool-3-thread-3] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-1-81033, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,619 [pool-3-thread-6] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-4-66996, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,621 [pool-3-thread-10] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-8-83075, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,626 [pool-3-thread-9] INFO rpc.RpcClient: Creating Bucket: vol-0-30931/bucket-7-63535, with server-side default bucket layout, sammi as owner, Versioning false, Storage Type set to DISK and Encryption set to false, Replication Type set to server-side default replication type, Namespace Quota set to -1, Space Quota set to -1
2024-09-24 15:37:02,817 [pool-3-thread-6] WARN impl.MetricsSystemImpl: ozone-freon metrics system already initialized!
2024-09-24 15:37:03,022 [pool-3-thread-4] INFO metrics.MetricRegistries: Loaded MetricRegistries class org.apache.ratis.metrics.dropwizard3.Dm3MetricRegistriesImpl
0.10% |█ | 10/10000 Time: 0:00:01| 2024-09-24 15:37:03,773 [pool-5-thread-7] INFO scm.XceiverClientShortCircuit: Pipeline-df315eb9-30ee-4765-9e68-7fc16335ce48-XceiverClientShortCircuit is created
2024-09-24 15:37:03,775 [pool-5-thread-7] INFO storage.DomainSocketFactory: DomainSocket(fd=311,path=/Users/sammi/ozone_dn_socket) is created within 1572500 ns
100.00% |█████████████████████████████████████████████████████████████████████████████████████████████████████| 10000/10000 Time: 0:00:24|
2024-09-24 15:37:27,539 [main] INFO freon.RandomKeyGenerator: Data generation is completed
2024-09-24 15:37:27,539 [main] INFO freon.RandomKeyGenerator: Data validation is completed
2024-09-24 15:37:31,231 [Pipeline-df315eb9-30ee-4765-9e68-7fc16335ce48-XceiverClientShortCircuit-ReceiveResponse] INFO scm.XceiverClientShortCircuit: receiveResponseTask is closed with java.nio.channels.ClosedChannelException
2024-09-24 15:37:31,243 [main] INFO scm.XceiverClientShortCircuit: DomainSocket(fd=311,path=/Users/sammi/ozone_dn_socket) is closed for 249539d9-f678-4ecf-aa6d-fcbcea75f627(localhost/127.0.0.1)
2024-09-24 15:37:31,243 [Pipeline-df315eb9-30ee-4765-9e68-7fc16335ce48-XceiverClientShortCircuit-SendRequest] INFO scm.XceiverClientShortCircuit: sendRequestTask is interrupted
***************************************************
Status: Success
Git Base Revision: bab77db646e17201dd953cc9f0a805502d24ea45
Number of Volumes created: 1
Number of Buckets created: 10
Number of Keys added: 10000
Replication: RATIS/ONE
Average Time spent in volume creation: 00:00:00,008
Average Time spent in bucket creation: 00:00:00,014
Average Time spent in key creation: 00:00:03,689
Average Time spent in key write: 00:00:02,927
Total bytes written: 10485760000
Total number of writes validated: 10000
Writes validated: 100.0 %
Successful validation: 10000
Unsuccessful validation: 0
Total Execution time: 00:00:29,326
***************************************************
```
Still working on whether it's possible to test it with robot test.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "jojochuang (via GitHub)" <gi...@apache.org>.
jojochuang commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1821253567
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java:
##########
@@ -0,0 +1,645 @@
+/*
+ * 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.hadoop.hdds.scm;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.client.ReplicationType;
+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.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.storage.DomainSocketFactory;
+import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
+import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.net.unix.DomainSocket;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.util.Daemon;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.ClosedChannelException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.UUID;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.hadoop.hdds.HddsUtils.processForDebug;
+import static org.apache.hadoop.hdds.scm.OzoneClientConfig.DATA_TRANSFER_VERSION;
+
+/**
+ * {@link XceiverClientSpi} implementation, the client to read local replica through short circuit.
+ */
+public class XceiverClientShortCircuit extends XceiverClientSpi {
+ public static final Logger LOG =
+ LoggerFactory.getLogger(XceiverClientShortCircuit.class);
+ private final Pipeline pipeline;
+ private final ConfigurationSource config;
+ private final XceiverClientMetrics metrics;
+ private int readTimeoutMs;
+ private int writeTimeoutMs;
+ // Cache the stream of blocks
+ private final Map<String, FileInputStream> blockStreamCache;
+ private final Map<String, RequestEntry> sentRequests;
+ private final BlockingDeque<RequestEntry> pendingRequests;
+ private final Daemon writeDaemon;
+ private final Daemon readDaemon;
+ private final Timer timer;
+
+ private boolean closed = false;
+ private final DatanodeDetails dn;
+ private final InetSocketAddress dnAddr;
+ private final DomainSocketFactory domainSocketFactory;
+ private DomainSocket domainSocket;
+ private DataOutputStream dataOut;
+ private DataInputStream dataIn;
+ private final int bufferSize;
+ private final ByteString clientId = ByteString.copyFrom(UUID.randomUUID().toString().getBytes());
+ private final AtomicLong callId = new AtomicLong(0);
+ private final String prefix;
+
+ /**
+ * Constructs a client that can communicate with the Container framework on local datanode through DomainSocket
+ */
+ public XceiverClientShortCircuit(Pipeline pipeline, ConfigurationSource config, DatanodeDetails dn) {
+ super();
+ Preconditions.checkNotNull(config);
+ this.readTimeoutMs = (int) config.getTimeDuration(OzoneConfigKeys.OZONE_CLIENT_READ_TIMEOUT,
+ OzoneConfigKeys.OZONE_CLIENT_READ_TIMEOUT_DEFAULT, TimeUnit.MILLISECONDS);
+ this.writeTimeoutMs = (int) config.getTimeDuration(OzoneConfigKeys.OZONE_CLIENT_WRITE_TIMEOUT,
+ OzoneConfigKeys.OZONE_CLIENT_WRITE_TIMEOUT_DEFAULT, TimeUnit.MILLISECONDS);
+
+ this.pipeline = pipeline;
+ this.dn = dn;
+ this.domainSocketFactory = DomainSocketFactory.getInstance(config);
+ this.config = config;
+ this.metrics = XceiverClientManager.getXceiverClientMetrics();
+ this.blockStreamCache = new ConcurrentHashMap<>();
+ this.sentRequests = new ConcurrentHashMap<>();
+ this.pendingRequests = new LinkedBlockingDeque<>();
+ int port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
+ this.dnAddr = NetUtils.createSocketAddr(dn.getIpAddress(), port);
+ this.bufferSize = config.getObject(OzoneClientConfig.class).getShortCircuitBufferSize();
+ this.prefix = "Pipeline-" + pipeline.getId().getId() + "-" + XceiverClientShortCircuit.class.getSimpleName();
+ this.timer = new Timer(prefix + "-Timer");
+ this.writeDaemon = new Daemon(new SendRequestTask());
+ this.readDaemon = new Daemon(new ReceiveResponseTask());
+ LOG.info("{} is created", prefix);
+ }
+
+ /**
+ * Create the DomainSocket to connect to the local DataNode.
+ */
+ @Override
+ public void connect() throws IOException {
+ if (domainSocket != null && domainSocket.isOpen()) {
+ return;
+ }
+ domainSocket = domainSocketFactory.createSocket(readTimeoutMs, writeTimeoutMs, dnAddr);
+ dataOut = new DataOutputStream(new BufferedOutputStream(domainSocket.getOutputStream(), bufferSize));
+ dataIn = new DataInputStream(new BufferedInputStream(domainSocket.getInputStream(),bufferSize));
+ writeDaemon.start();
+ readDaemon.start();
+ }
+
+ /**
+ * Close the DomainSocket.
+ */
+ @Override
+ public synchronized void close() {
+ closed = true;
+ timer.cancel();
+ if (domainSocket != null) {
+ try {
+ dataOut.close();
+ dataIn.close();
+ domainSocket.close();
+ dataOut = null;
+ dataIn = null;
+ LOG.info("{} is closed for {}", domainSocket.toString(), dn);
+ } catch (IOException e) {
+ LOG.warn("Failed to close domain socket for datanode {}", dn, e);
+ }
+ }
+ writeDaemon.interrupt();
+ readDaemon.interrupt();
+ try {
+ writeDaemon.join();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ try {
+ readDaemon.join();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return pipeline;
+ }
+
+ public DatanodeDetails getDn() {
+ return this.dn;
+ }
+
+ public ByteString getClientId() {
+ return clientId;
+ }
+
+ public long getCallId() {
+ return callId.incrementAndGet();
+ }
+
+ @Override
+ public ContainerCommandResponseProto sendCommand(ContainerCommandRequestProto request) throws IOException {
+ try {
+ return sendCommandWithTraceID(request, null).
+ getResponse().get();
+ } catch (ExecutionException e) {
+ throw getIOExceptionForSendCommand(request, e);
+ } catch (InterruptedException e) {
+ LOG.error("Command execution was interrupted.");
+ Thread.currentThread().interrupt();
+ throw (IOException) new InterruptedIOException(
+ "Command " + processForDebug(request) + " was interrupted.")
+ .initCause(e);
+ }
+ }
+
+ @Override
+ public Map<DatanodeDetails, ContainerCommandResponseProto>
+ sendCommandOnAllNodes(
+ ContainerCommandRequestProto request) throws IOException {
+ throw new UnsupportedOperationException("Operation Not supported for " +
+ DomainSocketFactory.FEATURE + " client");
+ }
+
+ @Override
+ public ContainerCommandResponseProto sendCommand(
+ ContainerCommandRequestProto request, List<Validator> validators)
+ throws IOException {
+ try {
+ XceiverClientReply reply;
+ reply = sendCommandWithTraceID(request, validators);
+ return reply.getResponse().get();
+ } catch (ExecutionException e) {
+ throw getIOExceptionForSendCommand(request, e);
+ } catch (InterruptedException e) {
+ LOG.error("Command execution was interrupted.");
+ Thread.currentThread().interrupt();
+ throw (IOException) new InterruptedIOException(
+ "Command " + processForDebug(request) + " was interrupted.")
+ .initCause(e);
+ }
+ }
+
+ private XceiverClientReply sendCommandWithTraceID(
+ ContainerCommandRequestProto request, List<Validator> validators)
+ throws IOException {
+ String spanName = "XceiverClientGrpc." + request.getCmdType().name();
+ return TracingUtil.executeInNewSpan(spanName,
+ () -> {
+ ContainerCommandRequestProto finalPayload =
+ ContainerCommandRequestProto.newBuilder(request)
+ .setTraceID(TracingUtil.exportCurrentSpan()).build();
+ ContainerCommandResponseProto responseProto = null;
+ IOException ioException = null;
+
+ // In case of an exception or an error, we will try to read from the
+ // datanodes in the pipeline in a round-robin fashion.
+ XceiverClientReply reply = new XceiverClientReply(null);
+
+ if (request.getCmdType() != ContainerProtos.Type.GetBlock &&
+ request.getCmdType() != ContainerProtos.Type.Echo) {
+ throw new UnsupportedOperationException("Command " + request.getCmdType() +
+ " is not supported for " + DomainSocketFactory.FEATURE + " client");
+ }
+
+ try {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing command {} on datanode {}", request, dn);
+ }
+ reply.addDatanode(dn);
+ responseProto = sendCommandInternal(finalPayload).getResponse().get();
+ if (validators != null && !validators.isEmpty()) {
+ for (Validator validator : validators) {
+ validator.accept(request, responseProto);
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("request {} {} {} finished", request.getCmdType(),
+ request.getClientId().toStringUtf8(), request.getCallId());
+ }
+ } catch (IOException e) {
+ ioException = e;
+ responseProto = null;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Failed to execute command {} on datanode {}", request, dn, e);
+ }
+ } catch (ExecutionException e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Failed to execute command {} on datanode {}", request, dn, e);
+ }
+ if (Status.fromThrowable(e.getCause()).getCode()
+ == Status.UNAUTHENTICATED.getCode()) {
+ throw new SCMSecurityException("Failed to authenticate with "
+ + "datanode DomainSocket XceiverServer with Ozone block token.");
+ }
+ ioException = new IOException(e);
+ } catch (InterruptedException e) {
+ LOG.error("Command execution was interrupted ", e);
+ Thread.currentThread().interrupt();
+ }
+
+ if (responseProto != null) {
+ reply.setResponse(CompletableFuture.completedFuture(responseProto));
+ return reply;
+ } else {
+ Objects.requireNonNull(ioException);
+ String message = "Failed to execute command {}";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(message + " on the pipeline {}.", request, pipeline);
+ } else {
+ LOG.error(message + " on the pipeline {}.", request.getCmdType(), pipeline);
+ }
+ throw ioException;
+ }
+ });
+ }
+
+ @VisibleForTesting
+ public XceiverClientReply sendCommandInternal(ContainerCommandRequestProto request)
+ throws IOException, InterruptedException {
+ checkOpen();
+ final CompletableFuture<ContainerCommandResponseProto> replyFuture =
+ new CompletableFuture<>();
+ RequestEntry entry = new RequestEntry(request, replyFuture);
+ // pendingRequests.add(entry);
+ // CompletableFuture.runAsync(() -> sendRequest(entry), poolExecutor);
+ sendRequest(entry);
+ return new XceiverClientReply(replyFuture);
+ }
+
+ @Override
+ public XceiverClientReply sendCommandAsync(
+ ContainerCommandRequestProto request)
+ throws IOException, ExecutionException, InterruptedException {
+ throw new UnsupportedOperationException("Operation Not supported for " + DomainSocketFactory.FEATURE + " client");
+ }
+
+ public synchronized void checkOpen() throws IOException {
+ if (closed) {
+ throw new IOException("This DomainSocket is not connected.");
+ }
+
+ // try to connect again
+ if (!domainSocket.isOpen()) {
+ connect();
+ }
+ }
+
+ @Override
+ public CompletableFuture<XceiverClientReply> watchForCommit(long index) {
Review Comment:
the latest master changed the interface defintion of this method. It is now returning XceiverClientReply instead.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi closed pull request #7236: HDDS-10685. Short circuit read support in Ozone
URL: https://github.com/apache/ozone/pull/7236
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "jojochuang (via GitHub)" <gi...@apache.org>.
jojochuang commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1821213900
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java:
##########
@@ -658,16 +675,29 @@ ContainerCommandResponseProto handleGetBlock(
}
ContainerProtos.BlockData responseData;
+ boolean shortCircuitGranted = false;
try {
- BlockID blockID = BlockID.getFromProtobuf(
- request.getGetBlock().getBlockID());
- if (replicaIndexCheckRequired(request)) {
- BlockUtils.verifyReplicaIdx(kvContainer, blockID);
Review Comment:
shouldn't we keep this line to verify replica index?
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822021767
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java:
##########
@@ -39,6 +39,51 @@ public class OzoneClientConfig {
private static final Logger LOG =
LoggerFactory.getLogger(OzoneClientConfig.class);
+ public static final String OZONE_READ_SHORT_CIRCUIT = "ozone.client.read.short-circuit";
+ public static final boolean OZONE_READ_SHORT_CIRCUIT_DEFAULT = false;
+ public static final String OZONE_DOMAIN_SOCKET_PATH = "ozone.domain.socket.path";
+ public static final String OZONE_DOMAIN_SOCKET_PATH_DEFAULT = "/var/lib/ozone/dn_socket";
+ public static final String SHORT_CIRCUIT_PREFIX = "read.short-circuit.";
+ public static final int DATA_TRANSFER_VERSION = 28;
+
+ @Config(key = "read.short-circuit",
+ defaultValue = "false",
+ type = ConfigType.BOOLEAN,
+ description = "Whether read short-circuit is enabled or not",
+ tags = { ConfigTag.CLIENT, ConfigTag.DATANODE })
+ private boolean shortCircuitEnabled = OZONE_READ_SHORT_CIRCUIT_DEFAULT;
+
+ @Config(key = SHORT_CIRCUIT_PREFIX + "buffer.size",
+ defaultValue = "128KB",
+ type = ConfigType.SIZE,
+ description = "Buffer size of reader/writer.",
+ tags = { ConfigTag.CLIENT, ConfigTag.DATANODE })
+ private int shortCircuitBufferSize = 128 * 1024;
Review Comment:
The short-circuit channel only exchange getBlock request and response. The value is determined by how big will a request and a response of a 256MB block size, 4MB chunk size, 16KB checksum size block. The request is round 500 bytes, and the response is around
30 + 53 * 64 + 6 * 16384 ~ 100k
block size (exclude chunks) - 30 bytes
chunk size (one checksums) - 53 bytes
one checksum size - 6 bytes
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822031029
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverClientProtocol.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.common.transport.server;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
+import org.apache.hadoop.security.KerberosInfo;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_KERBEROS_PRINCIPAL_KEY;
+
+/**
+ * Security protocol for a secure OzoneManager.
+ */
+@KerberosInfo(
+ serverPrincipal = DFS_DATANODE_KERBEROS_PRINCIPAL_KEY
+)
Review Comment:
removed
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822052684
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java:
##########
@@ -39,6 +39,51 @@ public class OzoneClientConfig {
private static final Logger LOG =
LoggerFactory.getLogger(OzoneClientConfig.class);
+ public static final String OZONE_READ_SHORT_CIRCUIT = "ozone.client.read.short-circuit";
+ public static final boolean OZONE_READ_SHORT_CIRCUIT_DEFAULT = false;
+ public static final String OZONE_DOMAIN_SOCKET_PATH = "ozone.domain.socket.path";
+ public static final String OZONE_DOMAIN_SOCKET_PATH_DEFAULT = "/var/lib/ozone/dn_socket";
Review Comment:
I plan to add a document about this feature. Will explain all these new properties there.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822025150
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java:
##########
@@ -150,26 +198,49 @@ public void releaseClient(XceiverClientSpi client, boolean invalidateClient,
}
}
- protected XceiverClientSpi getClient(Pipeline pipeline, boolean topologyAware)
+ protected XceiverClientSpi getClient(Pipeline pipeline, boolean topologyAware, boolean allowShortCircuit)
throws IOException {
try {
// create different client different pipeline node based on
// network topology
- String key = getPipelineCacheKey(pipeline, topologyAware);
- return clientCache.get(key, () -> newClient(pipeline));
+ String key = getPipelineCacheKey(pipeline, topologyAware, allowShortCircuit);
+ if (key.endsWith(DomainSocketFactory.FEATURE_FLAG)) {
+ final Pipeline newPipeline = Pipeline.newBuilder(pipeline).setReplicationConfig(
+ ReplicationConfig.fromTypeAndFactor(ReplicationType.SHORT_CIRCUIT,
+ ReplicationFactor.valueOf(pipeline.getReplicationConfig().getReplication()))).build();
+ return clientCache.get(key, () -> newClient(newPipeline, localDNCache.get(key)));
+ } else {
+ return clientCache.get(key, () -> newClient(pipeline));
+ }
} catch (Exception e) {
throw new IOException(
"Exception getting XceiverClient: " + e, e);
}
}
- private String getPipelineCacheKey(Pipeline pipeline,
- boolean topologyAware) {
- String key = pipeline.getId().getId().toString() + pipeline.getType();
+ private String getPipelineCacheKey(Pipeline pipeline, boolean topologyAware, boolean allowShortCircuit) {
+ String key = pipeline.getId().getId().toString() + "-" + pipeline.getType();
boolean isEC = pipeline.getType() == HddsProtos.ReplicationType.EC;
- if (topologyAware || isEC) {
+ if ((!isEC) && allowShortCircuit && isShortCircuitEnabled()) {
+ int port = 0;
+ for (DatanodeDetails dn : pipeline.getNodes()) {
+ // read port from the data node, on failure use default configured port.
+ port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
+ InetSocketAddress addr = NetUtils.createSocketAddr(dn.getIpAddress(), port);
+ if (OzoneNetUtils.isAddressLocal(addr) &&
+ dn.getCurrentVersion() >= SHORT_CIRCUIT_READS.toProtoValue()) {
Review Comment:
Sure. It will be addressed with a new patch.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822030488
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java:
##########
@@ -29,6 +30,10 @@
import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature;
import org.apache.hadoop.hdds.utils.db.BatchOperation;
import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DatanodeUtil;
Review Comment:
removed
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on code in PR #7236:
URL: https://github.com/apache/ozone/pull/7236#discussion_r1822028643
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/FilePerBlockStrategy.java:
##########
@@ -27,6 +27,10 @@
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DatanodeUtil;
Review Comment:
removed.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on PR #7236:
URL: https://github.com/apache/ozone/pull/7236#issuecomment-2446090546
> See also https://issues.apache.org/jira/browse/HDFS-16198 we will need to handle block token expirations in secure environment too.
Same KeyValueHandler#handleGetBlock is used to process the getBlock request from the short-circuit channel. Block token will be verified. If its expires, block will not be opened at server side. We don't use shared memory in current implementation. So I think we should be fine.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-10685. Short circuit read support in Ozone [ozone]
Posted by "ChenSammi (via GitHub)" <gi...@apache.org>.
ChenSammi commented on PR #7236:
URL: https://github.com/apache/ozone/pull/7236#issuecomment-2446247340
I will seperate the patch into sub patches. Close this one.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org