You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sz...@apache.org on 2020/07/17 18:51:17 UTC
[incubator-ratis] branch master updated: RATIS-997: Benchmarking
protobuffers and flatbuffers for streaming in GRPC (#139)
This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ratis.git
The following commit(s) were added to refs/heads/master by this push:
new 94e043a RATIS-997: Benchmarking protobuffers and flatbuffers for streaming in GRPC (#139)
94e043a is described below
commit 94e043a78b346f4c50eda2543f50c9ab5c34a7de
Author: anshkhannasbu <54...@users.noreply.github.com>
AuthorDate: Fri Jul 17 14:51:08 2020 -0400
RATIS-997: Benchmarking protobuffers and flatbuffers for streaming in GRPC (#139)
RATIS-997: Benchmarking protobuffers and flatbuffers for streaming in GRPC. Contributed by Ansh Khanna
---
pom.xml | 16 ++
ratis-experiments/README.md | 49 +++++++
ratis-experiments/pom.xml | 162 +++++++++++++++++++++
ratis-experiments/scripts/flatbuf-compile.sh | 30 ++++
.../src/main/flatbufs/FileTransfer.fbs | 34 +++++
.../experiments/flatbuffers/client/ClientFlat.java | 133 +++++++++++++++++
.../flatbuffers/client/ClientProto.java | 132 +++++++++++++++++
.../flatbuffers/server/FileTransferFlatbufs.java | 66 +++++++++
.../flatbuffers/server/FileTransferProtobufs.java | 60 ++++++++
.../experiments/flatbuffers/server/ServerFlat.java | 46 ++++++
.../flatbuffers/server/ServerProto.java | 47 ++++++
ratis-proto/src/main/proto/Experiments.proto | 36 +++++
12 files changed, 811 insertions(+)
diff --git a/pom.xml b/pom.xml
index 87c43f6..3dd19b2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -298,6 +298,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <artifactId>ratis-experiments</artifactId>
+ <groupId>org.apache.ratis</groupId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
<artifactId>ratis-metrics</artifactId>
<groupId>org.apache.ratis</groupId>
<version>${project.version}</version>
@@ -845,6 +850,17 @@
<profiles>
<profile>
+ <id>experiments-build</id>
+ <activation>
+ <property>
+ <name>buildExp</name>
+ </property>
+ </activation>
+ <modules>
+ <module>ratis-experiments</module>
+ </modules>
+ </profile>
+ <profile>
<id>os.linux</id>
<activation>
<os>
diff --git a/ratis-experiments/README.md b/ratis-experiments/README.md
new file mode 100644
index 0000000..ee45a3a
--- /dev/null
+++ b/ratis-experiments/README.md
@@ -0,0 +1,49 @@
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+
+# Apache Ratis Experiments
+
+This module has been created to house experimental projects.
+Details for running the experiment are mentioned below.
+
+### Building and Running The Experiments
+
+#### Requirements
+The flatbuffers compiler(flatc) is required for building this module.
+You can refer to this [link](https://rwinslow.com/posts/how-to-install-flatbuffers/) for instructions on how to build from flatbuffers source.
+
+#### Build Commands
+The repository can be complied using `mvn clean package -DskipTests -DbuildExp` under the project root directory;
+see also [BUILDING.md](../BUILDING.md).
+
+
+## Flatbuffers
+
+This program was developed to test the efficiency of flatbuffers vs protobuffers in terms of intermediate buffer copying.
+The Client creates a 1MB buffer, fills it with data and transfers it over GRPC.
+
+### Run Datatransfer Server and Client
+There two pairs of Server and Client: (ClientProto, ServerProto) and (ClientFlat,ServerFlat).
+
+run the client and servers by these commands from ratis-experiments directory:
+
+for server: `java -cp target/ratis-experiments-0.6.0-SNAPSHOT.jar org.apache.ratis.experiments.flatbuffers.server.ServerFlat`
+
+for client: `java -cp target/ratis-experiments-0.6.0-SNAPSHOT.jar org.apache.ratis.experiments.flatbuffers.client.ClientFlat {numberOfReps}`
+
+replace noOfReps with the number of times you want to transfer data(defaults to 100,000).
+
+### Findings:
+Current releases of flatbuffers with GRPC, do not provide zero-copy semantics in Java and have similar performance characteristics as compared to protobufs.
+Clarified in [flatbuffers issue #6023](https://github.com/google/flatbuffers/issues/6023).
\ No newline at end of file
diff --git a/ratis-experiments/pom.xml b/ratis-experiments/pom.xml
new file mode 100644
index 0000000..3c5aaf9
--- /dev/null
+++ b/ratis-experiments/pom.xml
@@ -0,0 +1,162 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>ratis</artifactId>
+ <groupId>org.apache.ratis</groupId>
+ <version>0.6.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>ratis-experiments</artifactId>
+ <name>Apache Ratis Experiments</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.ratis</groupId>
+ <artifactId>ratis-thirdparty-misc</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ratis</groupId>
+ <artifactId>ratis-proto</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.6.0</version>
+ <executions>
+ <execution>
+ <id>compile-flatbufs</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>exec</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <executable>bash</executable>
+ <arguments>scripts/flatbuf-compile.sh</arguments>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>com.google.code.maven-replacer-plugin</groupId>
+ <artifactId>replacer</artifactId>
+ <version>1.5.3</version>
+ <executions>
+ <execution>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>replace</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <basedir>${project.build.directory}/generated-sources</basedir>
+ <includes>
+ <include>**/*.java</include>
+ </includes>
+ <replacements>
+ <replacement>
+ <token>([^\.])com.google</token>
+ <value>$1org.apache.ratis.thirdparty.com.google</value>
+ </replacement>
+ <replacement>
+ <token>([^\.])io.grpc</token>
+ <value>$1org.apache.ratis.thirdparty.io.grpc</value>
+ </replacement>
+ </replacements>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <dependencyReducedPomLocation>${project.build.directory}/dependency-reduced-pom.xml</dependencyReducedPomLocation>
+ <shadedArtifactAttached>true</shadedArtifactAttached>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exlcude>META-INF/*.SF</exlcude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>com.coderplus.maven.plugins</groupId>
+ <artifactId>copy-rename-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-file</id>
+ <phase>package</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ <configuration>
+ <sourceFile>target/${project.artifactId}-${project.version}-shaded.jar</sourceFile>
+ <destinationFile>target/${project.artifactId}-${project.version}.jar</destinationFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/ratis-experiments/scripts/flatbuf-compile.sh b/ratis-experiments/scripts/flatbuf-compile.sh
new file mode 100644
index 0000000..bb84939
--- /dev/null
+++ b/ratis-experiments/scripts/flatbuf-compile.sh
@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+# 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.
+
+set -eu
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+DOCDIR="$DIR/.."
+
+if [ "$(which flatc)" ]; then
+ DESTDIR="$DOCDIR/target/generated-sources/"
+ mkdir -p "$DESTDIR"
+ cd "$DOCDIR"
+ flatc --grpc --java --gen-mutable -o ./target/generated-sources/ ./src/main/flatbufs/*.fbs
+ cd -
+fi
+
+
diff --git a/ratis-experiments/src/main/flatbufs/FileTransfer.fbs b/ratis-experiments/src/main/flatbufs/FileTransfer.fbs
new file mode 100644
index 0000000..a6211d2
--- /dev/null
+++ b/ratis-experiments/src/main/flatbufs/FileTransfer.fbs
@@ -0,0 +1,34 @@
+/**
+ * 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.
+ */
+namespace org.apache.ratis.flatbufs;
+
+table TransferMsg {
+partId: uint32;
+data: [byte];
+}
+
+table TransferReply {
+partId: uint32;
+message: string;
+}
+
+rpc_service FileTransfer{
+SendData(TransferMsg):TransferReply (streaming: "bidi");
+}
+
+root_type TransferMsg;
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientFlat.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientFlat.java
new file mode 100644
index 0000000..09eafb5
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientFlat.java
@@ -0,0 +1,133 @@
+/**
+ * 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.experiments.flatbuffers.client;
+
+import org.apache.ratis.thirdparty.com.google.flatbuffers.FlatBufferBuilder;
+import org.apache.ratis.thirdparty.io.grpc.Channel;
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannelBuilder;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
+import org.apache.ratis.flatbufs.FileTransferGrpc;
+import org.apache.ratis.flatbufs.TransferMsg;
+import org.apache.ratis.flatbufs.TransferReply;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Sets up a client to send flatbuffers messages over GRPC to Server.
+ * Uses a semaphore to manage current outbound data
+ * Uses a flatbuffer builder interface to create message.
+ */
+
+public class ClientFlat {
+
+ private FileTransferGrpc.FileTransferStub asyncStubFlat;
+ // semaphore to manage current outbound data
+ private final Semaphore available = new Semaphore(3000, true);
+
+ private final long[] recv = new long[1];
+ private int partId = 0;
+
+ public long[] getRecv() {
+ return recv;
+ }
+
+ public int getPartId(){
+ return partId;
+ }
+
+ public ClientFlat(Channel channel){
+ asyncStubFlat = FileTransferGrpc.newStub(channel);
+ recv[0] = 0;
+ }
+
+ public void execFlatClient(int reps) throws Exception{
+ System.out.println("Starting streaming with Flatbuffers");
+ StreamObserver<TransferMsg> requestObserver = asyncStubFlat.sendData(new StreamObserver<TransferReply>(){
+ @Override
+ public void onNext(TransferReply msg) {
+ recv[0]++;
+ available.release();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ Status status = Status.fromThrowable(t);
+ System.out.println(status);
+ System.out.println("Finished streaming with errors");
+ }
+
+ @Override
+ public void onCompleted() {
+ System.out.println("Finished streaming");
+ }
+ });
+ try{
+ int i = 0;
+ // allocate a byte buffer containing message data.
+ ByteBuffer bf = ByteBuffer.allocateDirect(1024*1024);
+ if(bf.hasArray()){
+ Arrays.fill(bf.array(), (byte) 'a');
+ }
+ while(i < reps) {
+ partId++;
+ available.acquire();
+ // start a builder and reset position of databuffer to 0.
+ FlatBufferBuilder builder = new FlatBufferBuilder();
+ bf.position(0).limit(bf.capacity());
+ // create string of the message data
+ // the datacopy happens here, significant CPU time spent.
+ int dataOff = builder.createString(bf);
+ // Use flatbuffers generated message builder.
+ int off = TransferMsg.createTransferMsg(builder, partId, dataOff);
+ builder.finish(off);
+ TransferMsg msg = TransferMsg.getRootAsTransferMsg(builder.dataBuffer());
+ requestObserver.onNext(msg);
+ i++;
+ }
+
+ } catch (Exception e){
+ System.out.println(e);
+ }
+ requestObserver.onCompleted();
+ Thread.sleep(1000*100);
+ if(recv[0] == partId){
+ System.out.println("Transfer Successfull....");
+ } else{
+ System.out.println("Some error occurred...");
+ }
+ }
+
+ public static void main(String[] args) throws Exception{
+ int times = 100000;
+ if(args.length != 0){
+ times = Integer.parseInt(args[0]);
+ }
+ String target = "localhost:50051";
+ ManagedChannel channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
+ ClientFlat c = new ClientFlat(channel);
+ c.execFlatClient(times);
+ channel.shutdown();
+ channel.awaitTermination(1, TimeUnit.SECONDS);
+ }
+}
\ No newline at end of file
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientProto.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientProto.java
new file mode 100644
index 0000000..c165329
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/client/ClientProto.java
@@ -0,0 +1,132 @@
+/**
+ * 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.experiments.flatbuffers.client;
+
+import org.apache.ratis.proto.ExperimentsProtos.TransferReplyProto;
+import org.apache.ratis.proto.ExperimentsProtos.TransferMsgProto;
+
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannelBuilder;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
+import org.apache.ratis.thirdparty.io.grpc.Channel;
+import org.apache.ratis.proto.FileTransferExampleServiceGrpc;
+import org.apache.ratis.thirdparty.com.google.protobuf.UnsafeByteOperations;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Sets up a client to send protobuf messages over GRPC to server.
+ * Uses a semaphore to manage current outbound data
+ * Uses UnsafeByteOperations to avoid data copy during message creation.
+ *
+ */
+
+public class ClientProto {
+ private FileTransferExampleServiceGrpc.FileTransferExampleServiceStub asyncStubProto;
+ private final long[] recv = new long[1];
+ private int partId = 0;
+ // semaphore to manage current outbound data
+ private final Semaphore available = new Semaphore(3000, true);
+
+ public long[] getRecv() {
+ return recv;
+ }
+
+ public int getPartId(){
+ return partId;
+ }
+
+ public ClientProto(Channel channel){
+ asyncStubProto = FileTransferExampleServiceGrpc.newStub(channel);
+ recv[0] = 0;
+ }
+
+ public void execProto(int reps) throws Exception{
+ System.out.println("Starting streaming with Protobuffers");
+
+ StreamObserver<TransferMsgProto> requestObserver = asyncStubProto.sendData(new StreamObserver<TransferReplyProto>(){
+
+ @Override
+ public void onNext(TransferReplyProto msg) {
+ available.release();
+ recv[0]++;
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ Status status = Status.fromThrowable(t);
+ System.out.println(status);
+ System.out.println("Finished streaming with errors");
+ }
+
+ @Override
+ public void onCompleted() {
+ System.out.println("Finished streaming");
+ }
+ });
+ try{
+ int i = 0;
+ // allocate a byte buffer containing message data.
+ ByteBuffer bf = ByteBuffer.allocate(1024*1024);
+ if(bf.hasArray()){
+ Arrays.fill(bf.array(), (byte) 'a');
+ }
+ while(i < reps) {
+ partId++;
+ available.acquire();
+ // using unsafewrap operations
+ // creates a ByteString refrencing buffer data. Avoids Copying.
+ // Something similar is missing in flatbuffers.
+ TransferMsgProto msg = TransferMsgProto.newBuilder().
+ setPartId(partId).
+ setData(UnsafeByteOperations.unsafeWrap(bf)).build();
+ requestObserver.onNext(msg);
+ i++;
+ }
+
+ } catch (Exception e){
+ System.out.println(e);
+ }
+ requestObserver.onCompleted();
+ Thread.sleep(1000*100);
+ if(recv[0] == partId){
+ System.out.println("Transfer Successfull....");
+ } else{
+ System.out.println("Some error occurred...");
+ }
+ }
+
+ public static void main(String[] args) throws Exception{
+ int times = 100000;
+ if(args.length != 0){
+ times = Integer.parseInt(args[0]);
+ }
+ String target = "localhost:50051";
+ ManagedChannel channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
+ ClientProto c = new ClientProto(channel);
+ c.execProto(times);
+ channel.shutdown();
+ channel.awaitTermination(1, TimeUnit.SECONDS);
+ }
+
+}
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferFlatbufs.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferFlatbufs.java
new file mode 100644
index 0000000..480475d
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferFlatbufs.java
@@ -0,0 +1,66 @@
+/**
+ * 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.experiments.flatbuffers.server;
+
+import org.apache.ratis.thirdparty.com.google.flatbuffers.FlatBufferBuilder;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.flatbufs.FileTransferGrpc;
+import org.apache.ratis.flatbufs.TransferMsg;
+import org.apache.ratis.flatbufs.TransferReply;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
+
+/**
+ * Server code responding to messages of flatbuffer based Client.
+ */
+
+public class FileTransferFlatbufs extends FileTransferGrpc.FileTransferImplBase {
+ @Override
+ public StreamObserver<TransferMsg> sendData(final StreamObserver<TransferReply> responseObserver){
+ return new StreamObserver<TransferMsg>(){
+ private long rcvdDataSize = 0;
+
+ public long getRcvdDataSize(){
+ return rcvdDataSize;
+ }
+
+ @Override
+ public void onNext(TransferMsg msg){
+ rcvdDataSize += msg.dataLength();
+ FlatBufferBuilder builder = new FlatBufferBuilder();
+ int off = TransferReply.createTransferReply(builder, msg.partId(), builder.createString("OK"));
+ builder.finish(off);
+ TransferReply rep = TransferReply.getRootAsTransferReply(builder.dataBuffer());
+ responseObserver.onNext(rep);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ Status status = Status.fromThrowable(t);
+ System.out.println(status);
+ System.out.println("Finished streaming with errors");
+ }
+
+ @Override
+ public void onCompleted(){
+ System.out.println(rcvdDataSize);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+}
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferProtobufs.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferProtobufs.java
new file mode 100644
index 0000000..631e94b
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/FileTransferProtobufs.java
@@ -0,0 +1,60 @@
+/**
+ * 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.experiments.flatbuffers.server;
+
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
+import org.apache.ratis.proto.ExperimentsProtos.TransferReplyProto;
+import org.apache.ratis.proto.ExperimentsProtos.TransferMsgProto;
+import org.apache.ratis.proto.FileTransferExampleServiceGrpc;
+
+/**
+ * Server code responding to messages of protobuffers based Client.
+ */
+
+public class FileTransferProtobufs extends FileTransferExampleServiceGrpc.FileTransferExampleServiceImplBase {
+ @Override
+ public StreamObserver<TransferMsgProto> sendData(final StreamObserver<TransferReplyProto> responseObserver) {
+ return new StreamObserver<TransferMsgProto>() {
+ private long rcvdDataSize = 0;
+
+ public long getRcvdDataSize(){
+ return rcvdDataSize;
+ }
+
+ @Override
+ public void onNext(TransferMsgProto msg){
+ rcvdDataSize += msg.getData().size();
+ TransferReplyProto rep = TransferReplyProto.newBuilder().setPartId(msg.getPartId()).setMessage("OK").build();
+ responseObserver.onNext(rep);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ Status status = Status.fromThrowable(t);
+ System.out.println(status);
+ System.out.println("Finished streaming with errors");
+ }
+
+ @Override
+ public void onCompleted(){
+ responseObserver.onCompleted();
+ }
+ };
+ }
+}
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerFlat.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerFlat.java
new file mode 100644
index 0000000..0df1f6e
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerFlat.java
@@ -0,0 +1,46 @@
+/**
+ * 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.experiments.flatbuffers.server;
+
+import org.apache.ratis.thirdparty.io.grpc.BindableService;
+import org.apache.ratis.thirdparty.io.grpc.ServerBuilder;
+import org.apache.ratis.thirdparty.io.grpc.Server;
+
+public final class ServerFlat {
+ private ServerFlat(){
+ }
+
+ public static void main( String[] args ) throws Exception {
+ BindableService service;
+ service = (BindableService) new FileTransferFlatbufs();
+ Server svr = ServerBuilder.forPort(50051)
+ .addService(service)
+ .build();
+ svr.start();
+ System.out.println("Server started");
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ System.out.println("Server shutting down");
+ svr.shutdown();
+ }
+ });
+ svr.awaitTermination();
+ }
+}
diff --git a/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerProto.java b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerProto.java
new file mode 100644
index 0000000..7db3a07
--- /dev/null
+++ b/ratis-experiments/src/main/java/org/apache/ratis/experiments/flatbuffers/server/ServerProto.java
@@ -0,0 +1,47 @@
+/**
+ * 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.experiments.flatbuffers.server;
+
+import org.apache.ratis.thirdparty.io.grpc.BindableService;
+import org.apache.ratis.thirdparty.io.grpc.Server;
+import org.apache.ratis.thirdparty.io.grpc.ServerBuilder;
+
+public final class ServerProto {
+ private ServerProto(){
+
+ }
+
+ public static void main( String[] args ) throws Exception {
+ BindableService service;
+ service = (BindableService) new FileTransferProtobufs();
+ Server svr = ServerBuilder.forPort(50051)
+ .addService(service)
+ .build();
+ svr.start();
+ System.out.println("Server started");
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ System.out.println("Server shutting down");
+ svr.shutdown();
+ }
+ });
+ svr.awaitTermination();
+ }
+}
diff --git a/ratis-proto/src/main/proto/Experiments.proto b/ratis-proto/src/main/proto/Experiments.proto
new file mode 100644
index 0000000..a3eec28
--- /dev/null
+++ b/ratis-proto/src/main/proto/Experiments.proto
@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+syntax = "proto3";
+option java_package = "org.apache.ratis.proto";
+option java_outer_classname = "ExperimentsProtos";
+option java_generate_equals_and_hash = true;
+package ratis.experiments;
+
+message TransferMsgProto {
+ uint32 partId = 1;
+ bytes data = 2;
+}
+
+message TransferReplyProto {
+ uint32 partId = 1;
+ string message = 2;
+}
+
+service FileTransferExampleService {
+ rpc sendData(stream TransferMsgProto) returns (stream TransferReplyProto);
+}
\ No newline at end of file