diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java index c3ec21f64b..bdb099a60c 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java @@ -19,11 +19,7 @@ import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; -import org.apache.ratis.util.CollectionUtils; -import org.apache.ratis.util.JavaUtils; -import org.apache.ratis.util.LogUtils; -import org.apache.ratis.util.Preconditions; -import org.apache.ratis.util.TaskQueue; +import org.apache.ratis.util.*; import org.apache.ratis.util.function.CheckedFunction; import org.apache.ratis.util.function.CheckedSupplier; import org.slf4j.Logger; @@ -31,7 +27,9 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.nio.channels.SeekableByteChannel; +import java.nio.channels.WritableByteChannel; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.StandardOpenOption; @@ -88,6 +86,28 @@ ByteString read(CheckedFunction resolver, long offset, } } + void streamRead(CheckedFunction resolver, long offset, long length, + WritableByteChannel stream) throws IOException { + if (offset + length > getWriteSize()) { + throw new IOException("Failed to read: offset (=" + offset + + " + length (=" + length + ") > size = " + getWriteSize() + + ", path=" + getRelativePath()); + } + + try (FileChannel in = FileUtils.newFileChannel( + resolver.apply(getRelativePath()), StandardOpenOption.READ)) { + long transferred = 0; + while (transferred < length) { + final long n = in.transferTo(offset + transferred, length - transferred, stream); + Preconditions.assertTrue(n >= 0); + transferred += n; + } + Preconditions.assertSame(length, transferred, "transferred"); + } finally { + stream.close(); + } + } + UnderConstruction asUnderConstruction() { throw new UnsupportedOperationException( "File " + getRelativePath() + " is not under construction."); @@ -121,6 +141,12 @@ static class ReadOnly extends FileInfo { this.writeSize = f.getWriteSize(); } + ReadOnly(Path relativePath, long size) { + super(relativePath); + this.committedSize = size; + this.writeSize = size; + } + @Override long getCommittedSize() { return committedSize; diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java index a930170ecd..585660e1e7 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java @@ -43,6 +43,7 @@ import java.io.IOException; import java.io.RandomAccessFile; import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; @@ -102,6 +103,11 @@ void putNew(UnderConstruction uc) { } } + void putReadOnly(ReadOnly ro) { + LOG.trace("{}: putReadOnly {}", name, ro.getRelativePath()); + map.put(ro.getRelativePath(), ro); + } + ReadOnly close(UnderConstruction uc) { LOG.trace("{}: close {}", name, uc.getRelativePath()); final ReadOnly ro = new ReadOnly(uc); @@ -208,6 +214,12 @@ CompletableFuture read(String relative, long offset, long length return submit(task, reader); } + void streamRead(String relative, long offset, long length, WritableByteChannel stream) + throws IOException { + final FileInfo info = files.get(relative); + info.streamRead(this::resolve, offset, length, stream); + } + CompletableFuture delete(long index, String relative) { final Supplier name = () -> "delete(" + relative + ") @" + getId() + ":" + index; final CheckedSupplier task = LogUtils.newCheckedSupplier(LOG, () -> { @@ -284,13 +296,18 @@ public void close() { CompletableFuture streamCommit(String p, long bytesWritten) { return CompletableFuture.supplyAsync(() -> { + final Path relative = normalize(p); final long len; - try (RandomAccessFile file = new RandomAccessFile(resolve(normalize(p)).toFile(), "r")) { + try (RandomAccessFile file = new RandomAccessFile(resolve(relative).toFile(), "r")) { len = file.length(); - return StreamWriteReplyProto.newBuilder().setIsSuccess(len == bytesWritten).setByteWritten(len).build(); } catch (IOException e) { throw new CompletionException("Failed to commit stream " + p + " with " + bytesWritten + " B.", e); } + final boolean success = len == bytesWritten; + if (success) { + files.putReadOnly(new ReadOnly(relative, len)); + } + return StreamWriteReplyProto.newBuilder().setIsSuccess(success).setByteWritten(len).build(); }, committer); } diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java index fb0ee49dc2..8eb6e3c17f 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java @@ -18,6 +18,7 @@ package org.apache.ratis.examples.filestore; import org.apache.ratis.client.RaftClient; +import org.apache.ratis.client.api.DataStreamInput; import org.apache.ratis.client.api.DataStreamOutput; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.proto.ExamplesProtos.DeleteReplyProto; @@ -29,6 +30,8 @@ import org.apache.ratis.proto.ExamplesProtos.WriteReplyProto; import org.apache.ratis.proto.ExamplesProtos.WriteRequestHeaderProto; import org.apache.ratis.proto.ExamplesProtos.WriteRequestProto; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.protocol.DataStreamReply; import org.apache.ratis.protocol.Message; import org.apache.ratis.protocol.RaftClientReply; import org.apache.ratis.protocol.RaftGroup; @@ -39,6 +42,7 @@ import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.Preconditions; import org.apache.ratis.util.ProtoUtils; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.function.CheckedFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +50,7 @@ import java.io.Closeable; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.function.Function; @@ -197,6 +202,49 @@ public DataStreamOutput getStreamOutput(String path, long dataSize, RoutingTable return client.getDataStreamApi().stream(request.toByteString().asReadOnlyByteBuffer(), routingTable); } + public DataStreamInput getStreamInput(String path, long offset, long length) { + final ReadRequestProto read = ReadRequestProto.newBuilder() + .setPath(ProtoUtils.toByteString(path)) + .setOffset(offset) + .setLength(length) + .build(); + return client.getDataStreamApi().streamReadOnly(read.toByteString().asReadOnlyByteBuffer()); + } + + /** + * Read file data using streaming read and write it to the given channel. + * + * @return total number of bytes read. + */ + public long streamRead(String path, long offset, long length, WritableByteChannel channel) + throws IOException { + long total = 0; + try (DataStreamInput in = getStreamInput(path, offset, length)) { + while (true) { + final ReferenceCountedObject ref = in.readAsync().join(); + try { + final DataStreamReply reply = ref.get(); + if (reply.getType() == Type.STREAM_HEADER) { + Preconditions.assertTrue(reply.isSuccess(), + () -> "Failed to stream read " + path + ", reply=" + reply); + return total; + } else { + Preconditions.assertTrue(reply.isSuccess(), + () -> "Failed to stream read " + path + ", reply=" + reply); + Preconditions.assertEquals(Type.STREAM_DATA, reply.getType(), + "reply type for stream read " + path); + final ByteBuffer data = reply.nioBuffer(); + while (data.hasRemaining()) { + total += channel.write(data); + } + } + } finally { + ref.release(); + } + } + } + } + public CompletableFuture writeAsync(String path, long offset, boolean close, ByteBuffer buffer, boolean sync) { return writeImpl(this::sendAsync, path, offset, close, buffer, sync ).thenApply(reply -> JavaUtils.supplyAndWrapAsCompletionException( diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java index d9a1463b93..4ff83369ac 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java @@ -43,6 +43,7 @@ import org.apache.ratis.util.FileUtils; import java.io.IOException; +import java.nio.channels.WritableByteChannel; import java.nio.file.Path; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; @@ -106,6 +107,24 @@ public CompletableFuture query(Message request) { .thenApply(reply -> Message.valueOf(reply.toByteString())); } + @Override + public void query(Message request, WritableByteChannel stream) { + try { + final ReadRequestProto proto = ReadRequestProto.parseFrom(request.getContent()); + if (proto.getIsWatch()) { + throw new IOException("Watch is not supported for streaming read: " + proto); + } + files.streamRead(proto.getPath().toStringUtf8(), proto.getOffset(), proto.getLength(), stream); + } catch (Exception e) { + LOG.error(getId() + ": Failed streaming read for " + request, e); + try { + stream.close(); + } catch (IOException ignored) { + // ignore + } + } + } + @Override public TransactionContext startTransaction(RaftClientRequest request) throws IOException { final ByteString content = request.getMessage().getContent(); diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java index 9d50e3421b..c62d5a10c4 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java @@ -35,6 +35,7 @@ public static List getSubCommands() { commands.add(new Server()); commands.add(new LoadGen()); commands.add(new DataStream()); + commands.add(new Read()); return commands; } } diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java new file mode 100644 index 0000000000..128a9abbac --- /dev/null +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java @@ -0,0 +1,158 @@ +/* + * 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.ratis.examples.filestore.cli; + +import com.beust.jcommander.Parameter; +import com.beust.jcommander.Parameters; +import org.apache.ratis.RaftConfigKeys; +import org.apache.ratis.client.RaftClient; +import org.apache.ratis.client.RaftClientConfigKeys; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.datastream.SupportedDataStreamType; +import org.apache.ratis.examples.common.SubCommandBase; +import org.apache.ratis.examples.filestore.FileStoreClient; +import org.apache.ratis.grpc.GrpcConfigKeys; +import org.apache.ratis.grpc.GrpcFactory; +import org.apache.ratis.protocol.ClientId; +import org.apache.ratis.protocol.RaftGroup; +import org.apache.ratis.protocol.RaftGroupId; +import org.apache.ratis.rpc.SupportedRpcType; +import org.apache.ratis.server.RaftServerConfigKeys; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; +import org.apache.ratis.util.FileUtils; +import org.apache.ratis.util.SizeInBytes; +import org.apache.ratis.util.TimeDuration; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * Subcommand to read files from FileStore using streaming read. + */ +@Parameters(commandDescription = "Read files from FileStore using streaming read") +public class Read extends SubCommandBase { + + @Parameter(names = {"--files"}, description = "Comma-separated file names on the server", required = true) + private String files; + + @Parameter(names = {"--size"}, description = "Number of bytes to read per file", required = true) + private long fileSizeInBytes; + + @Parameter(names = {"--offset"}, description = "Offset to start reading from", required = false) + private long offset; + + @Override + public void run() throws Exception { + final List fileNames = Arrays.asList(files.split(",")); + if (fileNames.isEmpty()) { + throw new IllegalArgumentException("No files specified in --files"); + } + + final RaftProperties raftProperties = newRaftProperties(); + + System.out.println("Starting streaming read now"); + + final long startTime = System.currentTimeMillis(); + long totalReadBytes = 0; + final List results = new ArrayList<>(); + try (FileStoreClient client = newClient(raftProperties)) { + for (String fileName : fileNames) { + final String name = fileName.trim(); + final ReadResult result = readFile(client, name); + results.add(result); + if (result.bytesRead != fileSizeInBytes) { + System.err.println("Error: file:" + name + " read:" + result.bytesRead + + " mismatch expected size:" + fileSizeInBytes); + } + totalReadBytes += result.bytesRead; + } + } + + final long endTime = System.currentTimeMillis(); + System.out.println("Total files read: " + fileNames.size()); + for (ReadResult result : results) { + System.out.println(" " + result.fileName + " -> " + result.tempFile + + " (" + result.bytesRead + " bytes)"); + } + System.out.println("Each file size: " + fileSizeInBytes); + System.out.println("Total data read: " + totalReadBytes + " bytes"); + System.out.println("Total time taken: " + (endTime - startTime) + " millis"); + } + + private ReadResult readFile(FileStoreClient client, String fileName) throws IOException { + final File tempFile = Files.createTempFile("filestore-read-", "-" + fileName).toFile(); + try (FileChannel out = FileUtils.newFileChannel(tempFile, + StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) { + final long bytesRead = client.streamRead(fileName, offset, fileSizeInBytes, out); + return new ReadResult(fileName, tempFile.getAbsolutePath(), bytesRead); + } finally { + FileUtils.deleteFully(tempFile); + } + } + + private static final class ReadResult { + private final String fileName; + private final String tempFile; + private final long bytesRead; + + private ReadResult(String fileName, String tempFile, long bytesRead) { + this.fileName = fileName; + this.tempFile = tempFile; + this.bytesRead = bytesRead; + } + } + + private static RaftProperties newRaftProperties() { + final int raftSegmentPreallocatedSize = 1024 * 1024 * 1024; + final RaftProperties raftProperties = new RaftProperties(); + RaftConfigKeys.Rpc.setType(raftProperties, SupportedRpcType.GRPC); + GrpcConfigKeys.setMessageSizeMax(raftProperties, SizeInBytes.valueOf(raftSegmentPreallocatedSize)); + RaftServerConfigKeys.Log.Appender.setBufferByteLimit(raftProperties, + SizeInBytes.valueOf(raftSegmentPreallocatedSize)); + RaftServerConfigKeys.Log.setWriteBufferSize(raftProperties, + SizeInBytes.valueOf(raftSegmentPreallocatedSize)); + RaftServerConfigKeys.Log.setPreallocatedSize(raftProperties, + SizeInBytes.valueOf(raftSegmentPreallocatedSize)); + RaftServerConfigKeys.Log.setSegmentSizeMax(raftProperties, SizeInBytes.valueOf(raftSegmentPreallocatedSize)); + RaftConfigKeys.DataStream.setType(raftProperties, SupportedDataStreamType.NETTY); + RaftServerConfigKeys.Log.setSegmentCacheNumMax(raftProperties, 2); + RaftClientConfigKeys.Rpc.setRequestTimeout(raftProperties, + TimeDuration.valueOf(50000, TimeUnit.MILLISECONDS)); + return raftProperties; + } + + private FileStoreClient newClient(RaftProperties raftProperties) throws IOException { + final RaftGroup raftGroup = RaftGroup.valueOf( + RaftGroupId.valueOf(ByteString.copyFromUtf8(getRaftGroupId())), getPeers()); + final RaftClient client = RaftClient.newBuilder() + .setProperties(raftProperties) + .setRaftGroup(raftGroup) + .setClientRpc(new GrpcFactory(new org.apache.ratis.conf.Parameters()) + .newRaftClientRpc(ClientId.randomId(), raftProperties)) + .setPrimaryDataStreamServer(getPeers()[0]) + .build(); + return new FileStoreClient(client); + } +} diff --git a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java index cdcee0ef0e..d70efaaa9e 100644 --- a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java +++ b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java @@ -104,6 +104,42 @@ public void testFileStoreStreamMultipleFiles() throws Exception { cluster.shutdown(); } + @Test + public void testFileStoreStreamReadAfterStreamWrite() throws Exception { + final CLUSTER cluster = newCluster(NUM_PEERS); + cluster.start(); + RaftTestUtil.waitForLeader(cluster); + + final RaftGroup raftGroup = cluster.getGroup(); + final Collection peers = raftGroup.getPeers(); + final RaftPeer primary = cluster.getLeader().getPeer(); + + final CheckedSupplier newClient = + () -> new FileStoreClient(cluster.getGroup(), getProperties(), primary); + + final RoutingTable routingTable = DataStreamTestUtils.getRoutingTableChainTopology(peers, primary); + testSingleFileReadAfterStreamWrite("foo", SizeInBytes.valueOf("2M"), 10_000, newClient, routingTable); + + cluster.shutdown(); + } + + private static void testSingleFileReadAfterStreamWrite( + String path, SizeInBytes fileLength, int bufferSize, + CheckedSupplier newClient, RoutingTable routingTable) + throws Exception { + LOG.info("testSingleFileAfterStreamWrite with path={}, fileLength={}", path, fileLength); + FileStoreWriter.newBuilder() + .setFileName(path) + .setFileSize(fileLength) + .setBufferSize(bufferSize) + .setFileStoreClientSupplier(newClient) + .build() + .streamWrite(routingTable) + .streamRead() + .delete() + .close(); + } + private void testSingleFile( String path, SizeInBytes fileLength, int bufferSize, CheckedSupplier newClient, RoutingTable routingTable) diff --git a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java index 480fe40d1b..18f8f89abe 100644 --- a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java +++ b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java @@ -36,6 +36,7 @@ import java.io.Closeable; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; import java.util.ArrayList; import java.util.List; import java.util.Objects; @@ -133,7 +134,7 @@ FileStoreWriter write(boolean sync) throws IOException { return this; } - public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) { + public FileStoreWriter streamWrite(RoutingTable routingTable) { final int size = fileSize.getSizeInt(); final DataStreamOutput dataStreamOutput = client.getStreamOutput(fileName, size, routingTable); final List> futures = new ArrayList<>(); @@ -156,8 +157,6 @@ public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) { DataStreamReply reply = dataStreamOutput.closeAsync().join(); Assertions.assertTrue(reply.isSuccess()); - // TODO: handle when any of the writeAsync has failed. - // check writeAsync requests for (int i = 0; i < futures.size(); i++) { reply = futures.get(i).join(); Assertions.assertTrue(reply.isSuccess()); @@ -168,6 +167,37 @@ public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) { return this; } + public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) { + return streamWrite(routingTable); + } + + FileStoreWriter streamRead() throws IOException { + final long expected = fileSize.getSizeInt(); + final long bytesRead = client.streamRead(fileName, 0, expected, DISCARD); + Assertions.assertEquals(expected, bytesRead, + () -> "stream read " + fileName + ": expected " + expected + " bytes"); + LOG.info("Stream read successful: {} bytes from {}", bytesRead, fileName); + return this; + } + + private static final WritableByteChannel DISCARD = new WritableByteChannel() { + @Override + public int write(ByteBuffer src) { + final int n = src.remaining(); + src.position(src.limit()); + return n; + } + + @Override + public boolean isOpen() { + return true; + } + + @Override + public void close() { + } + }; + CompletableFuture writeAsync(boolean sync) { Objects.requireNonNull(asyncExecutor, "asyncExecutor == null"); final Random r = new Random(seed);