Initial Version of RDMA based client Library

code/hpdos_rdma_sal/.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="bin/main" path="src/main/java">
+		<attributes>
+			<attribute name="gradle_scope" value="main"/>
+			<attribute name="gradle_used_by_scope" value="main,test"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="bin/main" path="src/main/resources">
+		<attributes>
+			<attribute name="gradle_scope" value="main"/>
+			<attribute name="gradle_used_by_scope" value="main,test"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="bin/test" path="src/test/resources">
+		<attributes>
+			<attribute name="gradle_scope" value="test"/>
+			<attribute name="gradle_used_by_scope" value="test"/>
+			<attribute name="test" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-11/"/>
+	<classpathentry kind="con" path="org.eclipse.buildship.core.gradleclasspathcontainer"/>
+	<classpathentry kind="output" path="bin/default"/>
+</classpath>

code/hpdos_rdma_sal/.gitattributes

+#
+# https://help.github.com/articles/dealing-with-line-endings/
+#
+# These are explicitly windows files and should use crlf
+*.bat           text eol=crlf
+

code/hpdos_rdma_sal/.gitignore

+# Ignore Gradle project-specific cache directory
+.gradle
+
+# Ignore Gradle build output directory
+build

code/hpdos_rdma_sal/.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>hpdos_rdma_sal</name>
+	<comment>Project hpdos_rdma_sal created by Buildship.</comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.buildship.core.gradleprojectbuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.buildship.core.gradleprojectnature</nature>
+	</natures>
+	<filteredResources>
+		<filter>
+			<id>1632245738513</id>
+			<name></name>
+			<type>30</type>
+			<matcher>
+				<id>org.eclipse.core.resources.regexFilterMatcher</id>
+				<arguments>node_modules|.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__</arguments>
+			</matcher>
+		</filter>
+	</filteredResources>
+</projectDescription>

code/hpdos_rdma_sal/.settings/org.eclipse.buildship.core.prefs

+arguments=
+auto.sync=false
+build.scans.enabled=false
+connection.gradle.distribution=GRADLE_DISTRIBUTION(WRAPPER)
+connection.project.dir=
+eclipse.preferences.version=1
+gradle.user.home=
+java.home=/usr/lib/jvm/java-11-openjdk-amd64
+jvm.arguments=
+offline.mode=false
+override.workspace.settings=true
+show.console.view=true
+show.executions.view=true

code/hpdos_rdma_sal/README.md

+1. Without load Balancing
+2. See Threading
\ No newline at end of file

code/hpdos_rdma_sal/build.gradle

+/*
+ * This file was generated by the Gradle 'init' task.
+ *
+ * This generated file contains a sample Java Library project to get you started.
+ * For more details take a look at the Java Libraries chapter in the Gradle
+ * User Manual available at https://docs.gradle.org/6.0/userguide/java_library_plugin.html
+ */
+
+// plugins {
+//     // Apply the java-library plugin to add support for Java Library
+//     id 'java-library'
+//     id 'application'
+// }
+
+// repositories {
+//     // Use jcenter for resolving dependencies.
+//     // You can declare any Maven/Ivy/file repository here.
+//     jcenter()
+//     mavenCentral()
+// }
+
+// application {
+//     // Define the main class for the application.
+//     mainClass = 'MetadataClient'
+// }
+
+// dependencies {
+//     // This dependency is exported to consumers, that is to say found on their compile classpath.
+//     api 'org.apache.commons:commons-math3:3.6.1'
+
+//     // This dependency is used internally, and not exposed to consumers on their own compile classpath.
+//     implementation 'com.google.guava:guava:28.0-jre'
+//     implementation 'com.ibm.darpc:darpc:1.9'
+
+//     // Use JUnit test framework
+//     testImplementation 'junit:junit:4.12'
+// }
+
+apply plugin: "application"
+apply plugin: "java-library"
+
+mainClassName = "MetadataClient"
+mainClassName = "StandaloneRDMAClient"
+version '1.0-SNAPSHOT'
+
+repositories {
+    mavenCentral()
+    jcenter()
+}
+
+dependencies {
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+    api 'org.apache.commons:commons-math3:3.6.1'
+
+    // This dependency is used internally, and not exposed to consumers on their own compile classpath.
+    implementation 'com.google.guava:guava:28.0-jre'
+    implementation 'com.ibm.darpc:darpc:1.9'
+    implementation 'com.github.ben-manes.caffeine:caffeine:3.0.4'
+    // Use JUnit test framework
+    // testImplementation 'junit:junit:4.12'
+}
\ No newline at end of file

code/hpdos_rdma_sal/gradle/wrapper/gradle-wrapper.properties

+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.0-bin.zip
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists

code/hpdos_rdma_sal/gradlew

+#!/usr/bin/env sh
+
+#
+# Copyright 2015 the original author or authors.
+#
+# 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
+#
+#      https://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.
+#
+
+##############################################################################
+##
+##  Gradle start up script for UN*X
+##
+##############################################################################
+
+# Attempt to set APP_HOME
+# Resolve links: $0 may be a link
+PRG="$0"
+# Need this for relative symlinks.
+while [ -h "$PRG" ] ; do
+    ls=`ls -ld "$PRG"`
+    link=`expr "$ls" : '.*-> \(.*\)$'`
+    if expr "$link" : '/.*' > /dev/null; then
+        PRG="$link"
+    else
+        PRG=`dirname "$PRG"`"/$link"
+    fi
+done
+SAVED="`pwd`"
+cd "`dirname \"$PRG\"`/" >/dev/null
+APP_HOME="`pwd -P`"
+cd "$SAVED" >/dev/null
+
+APP_NAME="Gradle"
+APP_BASE_NAME=`basename "$0"`
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD="maximum"
+
+warn () {
+    echo "$*"
+}
+
+die () {
+    echo
+    echo "$*"
+    echo
+    exit 1
+}
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "`uname`" in
+  CYGWIN* )
+    cygwin=true
+    ;;
+  Darwin* )
+    darwin=true
+    ;;
+  MINGW* )
+    msys=true
+    ;;
+  NONSTOP* )
+    nonstop=true
+    ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+        # IBM's JDK on AIX uses strange locations for the executables
+        JAVACMD="$JAVA_HOME/jre/sh/java"
+    else
+        JAVACMD="$JAVA_HOME/bin/java"
+    fi
+    if [ ! -x "$JAVACMD" ] ; then
+        die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+else
+    JAVACMD="java"
+    which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if [ "$cygwin" = "false" -a "$darwin" = "false" -a "$nonstop" = "false" ] ; then
+    MAX_FD_LIMIT=`ulimit -H -n`
+    if [ $? -eq 0 ] ; then
+        if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
+            MAX_FD="$MAX_FD_LIMIT"
+        fi
+        ulimit -n $MAX_FD
+        if [ $? -ne 0 ] ; then
+            warn "Could not set maximum file descriptor limit: $MAX_FD"
+        fi
+    else
+        warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT"
+    fi
+fi
+
+# For Darwin, add options to specify how the application appears in the dock
+if $darwin; then
+    GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
+fi
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
+    APP_HOME=`cygpath --path --mixed "$APP_HOME"`
+    CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+    JAVACMD=`cygpath --unix "$JAVACMD"`
+
+    # We build the pattern for arguments to be converted via cygpath
+    ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`
+    SEP=""
+    for dir in $ROOTDIRSRAW ; do
+        ROOTDIRS="$ROOTDIRS$SEP$dir"
+        SEP="|"
+    done
+    OURCYGPATTERN="(^($ROOTDIRS))"
+    # Add a user-defined pattern to the cygpath arguments
+    if [ "$GRADLE_CYGPATTERN" != "" ] ; then
+        OURCYGPATTERN="$OURCYGPATTERN|($GRADLE_CYGPATTERN)"
+    fi
+    # Now convert the arguments - kludge to limit ourselves to /bin/sh
+    i=0
+    for arg in "$@" ; do
+        CHECK=`echo "$arg"|egrep -c "$OURCYGPATTERN" -`
+        CHECK2=`echo "$arg"|egrep -c "^-"`                                 ### Determine if an option
+
+        if [ $CHECK -ne 0 ] && [ $CHECK2 -eq 0 ] ; then                    ### Added a condition
+            eval `echo args$i`=`cygpath --path --ignore --mixed "$arg"`
+        else
+            eval `echo args$i`="\"$arg\""
+        fi
+        i=`expr $i + 1`
+    done
+    case $i in
+        0) set -- ;;
+        1) set -- "$args0" ;;
+        2) set -- "$args0" "$args1" ;;
+        3) set -- "$args0" "$args1" "$args2" ;;
+        4) set -- "$args0" "$args1" "$args2" "$args3" ;;
+        5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
+        6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
+        7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
+        8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
+        9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
+    esac
+fi
+
+# Escape application args
+save () {
+    for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
+    echo " "
+}
+APP_ARGS=`save "$@"`
+
+# Collect all arguments for the java command, following the shell quoting and substitution rules
+eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS"
+
+exec "$JAVACMD" "$@"

code/hpdos_rdma_sal/gradlew.bat

+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
+@if "%DEBUG%" == "" @echo off
+@rem ##########################################################################
+@rem
+@rem  Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%" == "" set DIRNAME=.
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if "%ERRORLEVEL%" == "0" goto init
+
+echo.
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto init
+
+echo.
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:init
+@rem Get command-line arguments, handling Windows variants
+
+if not "%OS%" == "Windows_NT" goto win9xME_args
+
+:win9xME_args
+@rem Slurp the command line arguments.
+set CMD_LINE_ARGS=
+set _SKIP=2
+
+:win9xME_args_slurp
+if "x%~1" == "x" goto execute
+
+set CMD_LINE_ARGS=%*
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
+
+:end
+@rem End local scope for the variables with windows NT shell
+if "%ERRORLEVEL%"=="0" goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+if  not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
+exit /b 1
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega

code/hpdos_rdma_sal/settings.gradle

+/*
+ * This file was generated by the Gradle 'init' task.
+ *
+ * The settings file is used to specify which projects to include in your build.
+ *
+ * Detailed information about configuring a multi-project build in Gradle can be found
+ * in the user manual at https://docs.gradle.org/6.0/userguide/multi_project_builds.html
+ */
+
+rootProject.name = 'hpdos_rdma_sal'

code/hpdos_rdma_sal/src/main/java/MetadataClient.java

+import java.lang.reflect.InaccessibleObjectException;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.ibm.darpc.DaRPCClientEndpoint;
+import com.ibm.darpc.DaRPCClientGroup;
+import com.ibm.darpc.DaRPCStream;
+
+import hpdos.cache.SalCache;
+import hpdos.handlers.NetworkHandler;
+import hpdos.invalidationServer.InvalidationServer;
+import hpdos.protocol.AckType;
+import hpdos.protocol.Request;
+import hpdos.protocol.RequestType;
+import hpdos.protocol.Response;
+import hpdos.protocol.RpcProtocol;
+
+public class MetadataClient{
+    private DaRPCStream<Request, Response> stream;
+    private DaRPCClientEndpoint<Request, Response> clientEp;
+    private  InvalidationServer invalidationServer;
+    public MetadataClient() throws Exception
+    {
+        Cache<byte[],byte[]> cache = Caffeine.newBuilder()
+                                            .expireAfterWrite(30, TimeUnit.SECONDS)
+                                            .maximumSize(10_000)
+                                            .build();
+        invalidationServer = new InvalidationServer(cache);
+        invalidationServer.acceptSingleConnection();
+        System.out.println("started Invalidation server");
+        RpcProtocol rpcProtocol = new RpcProtocol();
+        DaRPCClientGroup<Request, Response> group = DaRPCClientGroup.createClientGroup(rpcProtocol, 100, 0, 16, 16); 
+        InetSocketAddress address = new InetSocketAddress("192.168.200.20", 1920);
+        clientEp = group.createEndpoint();
+        clientEp.connect(address, 1000);
+
+        stream = clientEp.createStream();
+    }
+
+    public byte[] get(byte[] key) throws Exception
+    {
+        Request request= new Request();
+        request.setRequestType(RequestType.GET);
+        request.setKey(key);
+        
+        Response response = new Response();
+        stream.request(request, response, false).get();
+        if(response.getAck() == AckType.SUCCESS_WITH_VALUE)
+            return response.getValue();
+        return null;
+    }
+
+    public int put(byte[] key,byte[] value) throws Exception
+    {
+        Request request= new Request();
+        request.setRequestType(RequestType.PUT);
+        request.setKey(key);
+        request.setValue(value);
+        
+        Response response = new Response();
+        stream.request(request, response, false).get();
+        return response.getAck();
+    }
+
+    public int delete(byte[] key) throws Exception
+    {
+        Request request= new Request();
+        request.setRequestType(RequestType.DELETE);
+        request.setKey(key);
+        
+        Response response = new Response();
+        stream.request(request, response, false).get();
+        return response.getAck();
+    }
+}
\ No newline at end of file

code/hpdos_rdma_sal/src/main/java/StandaloneRDMAClient.java

+import java.util.Scanner;
+import hpdos.protocol.AckType;
+
+public class StandaloneRDMAClient {
+
+        public static void main(String[] args) throws Exception {
+            MetadataClient  client = new MetadataClient();
+
+            Scanner scanner = new Scanner(System.in);
+            int option;
+            while (true) {
+                System.out.println("Enter 1. for read and 2 for write 3 for delete: ");
+                option = scanner.nextInt();
+                if(option == 1)
+                {
+                    System.out.println("Enter key: ");
+                    String inputkey = scanner.next();
+                    byte[] key = new byte[128];
+                    for(int i=0;i<inputkey.length();i++)
+                    {
+                        key[i]  = (byte)inputkey.charAt(i);   
+                    }
+
+                   byte[] value =  client.get(key);
+                   if(value != null)
+                       System.out.println(new String(value));
+                }
+                else if(option == 2)
+                {
+                    System.out.println("Enter key: ");
+                    String inputkey = scanner.next();
+                    System.out.println("Enter value: ");
+                    String inputvalue = scanner.next();
+                    byte[] key = new byte[128];
+                    byte[] value = new byte[128];
+                    for(int i=0;i<inputkey.length();i++)
+                    {
+                        key[i]  = (byte)inputkey.charAt(i);   
+                    }
+
+                    for(int i=0;i<inputvalue.length();i++)
+                    {
+                        value[i]  = (byte)inputvalue.charAt(i);   
+                    }
+                    int ack = client.put(key, value);
+                    System.out.println("Response Ack got " +ack);
+                    if (ack == AckType.SUCCESS) {
+                        System.out.println("Success");
+                    }
+                       
+                }else if(option == 3)
+                {
+                    System.out.println("Enter key: ");
+                    String inputkey = scanner.next();
+                    byte[] key = new byte[128];
+                    for(int i=0;i<inputkey.length();i++)
+                    {
+                        key[i]  = (byte)inputkey.charAt(i);   
+                    }
+                    int ack = client.delete(key);
+                    System.out.println("Response Ack got " +ack);
+                    if (ack == AckType.SUCCESS) {
+                        System.out.println("Success");
+                    }
+                       
+                }
+                else
+                {
+                    System.out.println("Invalid option try again");
+                    continue;
+                }
+                /*Stopwatch stopwatch = Stopwatch.createUnstarted();
+                stopwatch.start(); 
+                stopwatch.stop();
+                long micros = stopwatch.elapsed(TimeUnit.MICROSECONDS);
+                 System.out.println("Request took: "+micros+" microseconds.");
+                */
+            }  
+            
+            
+        }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/cache/SalCache.java

+package hpdos.cache;
+
+import java.util.concurrent.TimeUnit;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+
+public class SalCache {
+    public  static Cache<byte[], byte[]> cache;
+    public SalCache()
+    {
+        SalCache.cache = Caffeine.newBuilder()
+        .expireAfterWrite(30, TimeUnit.SECONDS)
+        .maximumSize(10_000)
+        .build();
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/handlers/NetworkHandler.java

+package hpdos.handlers;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+import com.ibm.darpc.DaRPCClientEndpoint;
+import com.ibm.darpc.DaRPCClientGroup;
+import com.ibm.darpc.DaRPCFuture;
+import com.ibm.darpc.DaRPCStream;
+import com.ibm.darpc.DaRPCClientGroup.RpcClientFactory;
+
+import org.apache.commons.math3.util.Pair;
+
+import hpdos.cache.SalCache;
+import hpdos.packet.Packet;
+import hpdos.protocol.Request;
+import hpdos.protocol.RequestType;
+import hpdos.protocol.Response;
+import hpdos.protocol.RpcProtocol;
+
+
+public class NetworkHandler {
+
+    public HashMap<String, DaRPCStream<Request, Response>> streams;
+    public int masterPartitions = 1;
+    public List<String> ip_addresses = Arrays.asList("192.168.200.50");
+
+    public NetworkHandler(){
+        this.streams = null;
+
+    }
+    
+    public boolean connectToServer() throws Exception{
+        RpcProtocol rpcProtocol = new RpcProtocol();
+        DaRPCClientGroup<Request, Response> group = DaRPCClientGroup.createClientGroup(rpcProtocol, 100, 0, 16, 16); 
+        InetSocketAddress address = new InetSocketAddress("192.168.200.50", 1919);
+        DaRPCClientEndpoint<Request, Response> clientEp = group.createEndpoint();
+        clientEp.connect(address, 1000);
+        DaRPCStream<Request, Response> stream = clientEp.createStream();
+        
+        if(!stream.isEmpty()){
+            this.streams.put("192.168.200.50", stream);
+            return true;
+        }
+        else{
+            return false;
+        }
+    }
+
+    public Response create(Packet packet) throws IOException, InterruptedException, ExecutionException{
+        SalCache.cache.put(packet.getKey(), packet.getValue());
+        Request request = new Request();
+        Response response = new Response();
+        request.setRequestType(RequestType.GET);
+        request.setKey(packet.getKey());
+        request.setValue(packet.getValue());
+        int partition = partition(packet.getKey());
+        response = streams.get(ip_addresses.get(partition)).request(request, response, false).get();
+        return response;
+    }
+
+    public Pair<byte[], byte[]> read(byte[] key) throws InterruptedException, ExecutionException, IOException{
+        Pair<byte[], byte[]> kPair;
+        byte[] value = SalCache.cache.getIfPresent(key);
+        if (value == null) {
+            // Add code to read from NIC/SSD
+            Request request = new Request();
+            Response response = new Response();
+            request.setRequestType(RequestType.GET);
+            request.setKey(key);
+            request.setValue(null);
+            int partition = partition(key);
+            response = streams.get(ip_addresses.get(partition)).request(request, response, false).get();
+            value = response.getValue();
+            kPair = new Pair<byte[],byte[]>(key, value);
+            return kPair;
+        }
+        else {
+            // Add code to return key,value from sal cache
+            kPair = new Pair<byte[],byte[]>(key, value);
+            return kPair;
+        }
+        
+    }
+
+    public Response update(Packet packet) throws InterruptedException, ExecutionException, IOException{
+        SalCache.cache.put(packet.getKey(), packet.getValue());
+        Request request = new Request();
+        Response response = new Response();
+        request.setRequestType(RequestType.PUT);
+        request.setKey(packet.getKey());
+        request.setValue(packet.getValue());
+        int partition = partition(packet.getKey());
+        response = streams.get(ip_addresses.get(partition)).request(request, response, false).get();
+        return response;
+    }
+
+    public Response delete(Packet packet) throws InterruptedException, ExecutionException, IOException{
+        SalCache.cache.put(packet.getKey(), packet.getValue());
+        Request request = new Request();
+        Response response = new Response();
+        request.setRequestType(RequestType.DELETE);
+        request.setKey(packet.getKey());
+        request.setValue(null);
+        int partition = partition(packet.getKey());
+        response = streams.get(ip_addresses.get(partition)).request(request, response, false).get();
+        return response;
+    }
+
+    public int partition(byte[] key)
+    {
+        int hash = 7;
+        hash = 31 * hash + (key == null ? 0 : key.hashCode());
+        return hash % masterPartitions;
+    }
+
+}

code/hpdos_rdma_sal/src/main/java/hpdos/invalidationServer/InvalidationServer.java

+package hpdos.invalidationServer;
+
+import java.net.InetSocketAddress;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.ibm.darpc.DaRPCServerEndpoint;
+import com.ibm.darpc.DaRPCServerGroup;
+import com.ibm.disni.RdmaServerEndpoint;
+
+import hpdos.protocol.InvalidationRequest;
+import hpdos.protocol.InvalidationResponse;
+import hpdos.services.InvalidationService;
+
+public class InvalidationServer
+{
+    Cache<byte[], byte[]> cache;
+    RdmaServerEndpoint<DaRPCServerEndpoint<InvalidationRequest, InvalidationResponse>> serverEp;
+    public InvalidationServer(Cache<byte[], byte[]> cache)throws Exception
+    {
+            this.cache = cache; 
+            long[] clusterAffinities =  new long[1];
+            clusterAffinities[0] = 1<<1;
+            DaRPCServerGroup<InvalidationRequest, InvalidationResponse> group = null;
+            InvalidationService  service = new InvalidationService(this.cache);
+
+            group = DaRPCServerGroup.createServerGroup(service, clusterAffinities, -1, 1, true, 2, 2, 2, 1);    
+            serverEp = group.createServerEndpoint();
+            InetSocketAddress address = new InetSocketAddress("192.168.200.20", 1921);
+            serverEp.bind(address, 100);
+    }
+    public void acceptSingleConnection() throws Exception
+    {
+        System.out.println("Waiting for server to send connection request for invalidation");
+        Runnable runnable = ()->
+            {
+                try{
+                serverEp.accept();
+                }catch(Exception e){
+                    e.printStackTrace();
+                }
+            };
+        new Thread(runnable).start();
+        
+        System.out.println("Got Connected for Invalidation");
+    }
+    
+}

code/hpdos_rdma_sal/src/main/java/hpdos/packet/Packet.java

+package hpdos.packet;
+
+public class Packet {
+    byte[] key;
+    byte[] value;
+    int operationType;
+    
+    public Packet(int operationType, byte[] key, byte[] value){
+        this.key = key;
+        this.value = value;
+        this.operationType = operationType;
+    }
+
+    public byte[] getKey(){
+        return this.key;
+    }
+
+    public byte[] getValue(){
+        return this.value;
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/AckType.java

+package hpdos.protocol;
+
+public interface AckType
+{
+    public static int SUCCESS = 0;
+    public static int NOTFOUND = 1;
+    public static int NOTALLOWED = 2;
+    public static int DBFAILED = 3;
+    public static int SUCCESS_WITH_VALUE = 4;
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/InvalidationRequest.java

+package hpdos.protocol;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import com.ibm.darpc.DaRPCMessage;
+
+public class InvalidationRequest implements DaRPCMessage {
+    public byte[] key;
+    private static int SERIALIZED_SIZE = 128 ;
+    @Override
+    public int write(ByteBuffer buffer) throws IOException {
+        buffer.put(key);
+        return SERIALIZED_SIZE;
+    }
+
+    @Override
+    public void update(ByteBuffer buffer) throws IOException {
+        if(key == null)
+            key = new byte[128];
+        buffer.get(key);
+    }
+
+    @Override
+    public int size() {
+        return SERIALIZED_SIZE;
+    }
+
+    public byte[] getKey() {
+        return key;
+    }
+
+    public void setKey(byte[] key) {
+        this.key = key;
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/InvalidationResponse.java

+package hpdos.protocol;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import com.ibm.darpc.DaRPCMessage;
+
+public class InvalidationResponse implements DaRPCMessage{
+    public static int SERIALIZED_SIZE = 4;
+    private int ack;
+
+    @Override
+    public int write(ByteBuffer buffer) throws IOException
+    {
+        buffer.putInt(ack);   
+        return 0;
+    }
+
+    @Override
+    public void update(ByteBuffer buffer) throws IOException
+    {
+        buffer.getInt(ack);
+    }
+
+    @Override
+    public int size() {
+        
+        return SERIALIZED_SIZE;
+    }
+    public void setAck(int ack){
+        this.ack = ack;
+    }
+    public int getAck()
+    {
+        return this.ack;
+    }
+}
+

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/InvalidationRpcProtocol.java

+package hpdos.protocol;
+
+import com.ibm.darpc.DaRPCProtocol;
+
+public class InvalidationRpcProtocol implements DaRPCProtocol<InvalidationRequest, InvalidationResponse> {
+
+    @Override
+    public InvalidationRequest createRequest() {
+        return new InvalidationRequest();
+    }
+
+    @Override
+    public InvalidationResponse createResponse() {
+        return new InvalidationResponse();
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/Request.java

+package hpdos.protocol;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import com.ibm.darpc.DaRPCMessage;
+
+public class Request implements DaRPCMessage
+{
+    private static int SERIALIZED_SIZE = 260;
+    public int requestType;
+    public byte[] key;
+    public byte[] value;
+
+
+    @Override
+    public int write(ByteBuffer buffer) throws IOException
+    {
+        System.out.println("Request Write Method");
+        buffer.putInt(requestType);
+        buffer.put(key);
+        System.out.println("length : "+key.length);
+        //if operation type is get and delete then value is not required
+        if(requestType == RequestType.PUT)
+        {
+            buffer.put(value);
+            System.out.println(" "+value.length);
+            //size of key+value+operationType
+            return 4+key.length+value.length;
+        }
+        //else return  size of Key+operationType
+        return 4+key.length;
+    }
+
+    @Override
+    public void update(ByteBuffer buffer) throws IOException
+    {
+        System.out.println("Request update method"+buffer.capacity());
+        requestType = buffer.getInt();
+        if(key == null || key.length != 128)
+           this.key  = new byte[128];
+        buffer.get(key);
+        if(requestType == RequestType.PUT)
+        {
+            if(value == null || value.length != 128)
+                this.value = new byte[128];
+            buffer.get(value);    
+        }
+    }
+
+    @Override
+    public int size() {
+        //This method gives max size of request
+        return SERIALIZED_SIZE;
+    }
+    
+    public byte[] getKey() {
+        return key;
+    }
+
+    public void setKey(byte[] key) {
+        this.key = key;
+    }
+
+    public byte[] getValue() {
+        return value;
+    }
+
+    public void setValue(byte[] value) {
+        this.value = value;
+    }
+
+    public int getRequestType() {
+        return requestType;
+    }
+
+    public void setRequestType(int requestType) {
+        this.requestType = requestType;
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/RequestType.java

+package hpdos.protocol;
+
+public interface RequestType
+{
+    public static int GET =101;
+    public static int PUT =100;
+    public static int DELETE =102;
+    public static int INVALIDATE = 103;
+    
+};
\ No newline at end of file

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/Response.java

+package hpdos.protocol;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import com.ibm.darpc.DaRPCMessage;
+
+public class Response implements DaRPCMessage{
+    public static int SERIALIZED_SIZE = 260;
+    private int ack;
+    public byte[] key;
+    public byte[] value;
+    
+    @Override
+    public int write(ByteBuffer buffer) throws IOException 
+    {
+        System.out.println("Response write Method");
+        buffer.putInt(ack);
+        if(ack == AckType.SUCCESS_WITH_VALUE)
+        {
+            System.out.println("length "+value.length);
+            buffer.put(value);
+            return 4+value.length;
+        }
+        return 4;
+    }
+
+    @Override
+    public void update(ByteBuffer buffer) throws IOException
+    {
+        System.out.println("Response update method "+buffer.limit());
+        ack = buffer.getInt();
+        if(ack == AckType.SUCCESS_WITH_VALUE)
+        {
+            if(value == null || value.length <128)
+                value = new byte[128];
+            buffer.get(value);
+        }
+    }
+
+    @Override
+    public int size() {
+        
+        return SERIALIZED_SIZE;
+    }
+    
+    public int getAck(){
+        return ack;
+    }
+
+    public void setAck(int ack){
+        this.ack = ack;
+    }
+
+    public byte[] getKey() {
+        return key;
+    }
+
+    public void setKey(byte[] key) {
+        this.key = key;
+    }
+
+    public byte[] getValue() {
+        return value;
+    }
+
+    public void setValue(byte[] value) {
+        this.value = value;
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/protocol/RpcProtocol.java

+package hpdos.protocol;
+
+import com.ibm.darpc.DaRPCProtocol;
+
+public class RpcProtocol implements DaRPCProtocol<Request, Response> {
+
+    @Override
+    public Request createRequest() {
+        return new Request();
+    }
+
+    @Override
+    public Response createResponse() {
+        return new Response();
+    }
+}

code/hpdos_rdma_sal/src/main/java/hpdos/services/InvalidationService.java

+package hpdos.services;
+
+import java.io.IOException;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.ibm.darpc.DaRPCServerEndpoint;
+import com.ibm.darpc.DaRPCServerEvent;
+import com.ibm.darpc.DaRPCService;
+
+import hpdos.protocol.InvalidationRequest;
+import hpdos.protocol.InvalidationResponse;
+import hpdos.protocol.InvalidationRpcProtocol;
+
+public class InvalidationService extends InvalidationRpcProtocol implements DaRPCService<InvalidationRequest, InvalidationResponse>{
+    Cache<byte[], byte[]> cache;
+    public InvalidationService(Cache<byte[], byte[]> cache)
+    {
+        this.cache  = cache;
+    }
+    @Override
+    public void processServerEvent(DaRPCServerEvent<InvalidationRequest, InvalidationResponse> event) throws IOException {
+        System.out.println("Got Invalidation Request");
+        InvalidationRequest request = event.getReceiveMessage();
+        InvalidationResponse response = event.getSendMessage();
+        try
+        {
+            byte[] key = request.getKey();
+            cache.invalidate(key);
+            response.setAck(1);
+        }
+        catch(Exception e)
+        {
+            e.printStackTrace();
+            response.setAck(0);
+        }
+        event.triggerResponse();
+    }
+      
+    public void open(DaRPCServerEndpoint<InvalidationRequest,InvalidationResponse> serverEp)
+    {
+
+    }
+    public void close(DaRPCServerEndpoint<InvalidationRequest,InvalidationResponse> serverEp)
+    {
+
+    }
+}