Added DCRDT Support

consys-core/src/main/scala/de/tuda/stg/consys/core/ConsistencyLabel.scala

@@ -24,6 +24,7 @@ object ConsistencyLabel {
 	/* CRDTs */
 	case object CvRDT extends ConsistencyLabel
 	case object CmRDT extends ConsistencyLabel
+	case object DCRDT extends ConsistencyLabel
 
 
 	case class Cassandra(replicas : Int) extends ConsistencyLabel

consys-core/src/main/scala/de/tuda/stg/consys/core/akka/AkkaReplicaSystemFactory.scala

@@ -30,6 +30,7 @@ object AkkaReplicaSystemFactory extends ReplicaSystemFactory {
 			with CausalAkkaReplicaSystem
 			with CmRDTAkkaReplicaSystem
 			with CvRDTAkkaReplicaSystem
+			with DeltaCRDTAkkaReplicaSystem
 	{
 		override protected def freshAddr() : String =
 			"$" + String.valueOf(Random.alphanumeric.take(16).toArray)

consys-core/src/main/scala/de/tuda/stg/consys/core/akka/DeltaCRDTAkkaReplicaSystem.scala

@@ -0,0 +1,138 @@
+package de.tuda.stg.consys.core.akka
+import java.io
+
+import akka.actor.{ActorRef, ActorSystem}
+import de.tuda.stg.consys.core.Address
+
+import scala.reflect.runtime.universe._
+import de.tuda.stg.consys.core.akka.DeltaCRDTAkkaReplicaSystem.{DeltaCRDTReplicatedObject, DeltaUpdateReq}
+import de.tuda.stg.consys.core.ConsistencyLabel.DCRDT
+import de.tuda.stg.consys.core.akka.Requests.{InvokeOp, NoAnswerRequest, Operation, Request, SynchronousRequest}
+
+import scala.concurrent.duration.FiniteDuration
+import scala.reflect.runtime.universe
+/**
+@author: Kris Frühwein und Julius Näumann
+ */
+trait DeltaCRDTAkkaReplicaSystem extends AkkaReplicaSystem {
+
+
+//creates master replica
+  override protected def createMasterReplica[T <: Obj : TypeTag](l: ConsistencyLevel, addr: Addr, obj: T): AkkaReplicatedObject[Addr, T] = {
+    val result = l match {
+
+      case DCRDT => new DeltaCRDTReplicatedObject[Addr, T](obj, addr, this)
+      case _ => super.createMasterReplica[T](l, addr, obj)
+    }
+    println("created master replica")
+    result
+  }
+
+  //creates follower replica
+  override protected def createFollowerReplica[T <: Obj : TypeTag](l: ConsistencyLevel, addr: Addr, obj: T, masterRef: ActorRef): AkkaReplicatedObject[Addr, T] = {
+    val result = l match {
+      case DCRDT => new DeltaCRDTReplicatedObject[Addr, T](obj, addr, this)
+      case _ => super.createFollowerReplica[T](l, addr, obj, masterRef)
+    }
+    println("created follower replicas")
+    result
+  }
+}
+
+
+trait DeltaHandler {
+  def transmitDelta(delta: Any)
+}
+
+    object DeltaCRDTAkkaReplicaSystem {
+
+      private case class RequestOperation(op: Operation[_]) extends SynchronousRequest[Unit]
+
+      private case class RequestSync(tx: Transaction) extends SynchronousRequest[Unit]
+
+      case class Message(obj: Any) extends NoAnswerRequest
+
+      case class DeltaUpdateReq(obj: Any) extends NoAnswerRequest
+
+
+
+      //Class for an Replicated Object. must be serializable
+      class DeltaCRDTReplicatedObject[Loc, T]
+      (
+        init: T, val addr: Loc, val replicaSystem: AkkaReplicaSystem {type Addr = Loc}
+      )(
+        protected implicit val ttt: TypeTag[T]
+      ) extends AkkaSECReplicatedObject[Loc, T]
+        with Lockable[T]
+        with Serializable
+        with DeltaHandler
+      {
+        setObject(init)
+        val t = init.asInstanceOf[DeltaCRDT]
+
+        override final def consistencyLevel: ConsistencyLevel = DCRDT
+
+        override def handleRequest[R](request: Request[R]): R = request match {
+          case DeltaUpdateReq(state: AkkaReplicaSystem#Obj) =>
+            getObject.asInstanceOf[DeltaCRDT].merge(state)
+
+            None.asInstanceOf[R]
+          case _ =>
+            super.handleRequest(request)
+
+        }
+
+        override def internalInvoke[R](tx: Transaction, methodName: String, args: Seq[Seq[Any]]): R = {
+          val result = super.internalInvoke[R](tx, methodName, args)
+          if (result.isInstanceOf[Delta]) {
+            val d = result.asInstanceOf[Delta]
+            replicaSystem.foreachOtherReplica(handler => handler.request(addr, DeltaUpdateReq(d.delta)))
+          }
+          result
+        }
+
+
+
+        override def sync(): Unit = {
+          // todo:
+          // traditional sync method does not make sense in the context of dcrdt
+          // should there be an option to force sync?
+          println(s"DCRDT '$addr' sync")
+        }
+
+        override protected def transactionStarted(tx: Transaction): Unit = {
+          super.transactionStarted(tx)
+        }
+
+        override protected def transactionFinished(tx: Transaction): Unit = {
+          super.transactionFinished(tx)
+        }
+
+        override def toString: String = s"@DCRDT($addr, $getObject)"
+
+        override def transmitDelta(delta: Any): Unit = {
+          replicaSystem.foreachOtherReplica(handler => handler.request(addr, DeltaUpdateReq(delta)))
+        }
+      }
+    }
+
+//abstract class for all deltaCRDT
+abstract class DeltaCRDT extends DeltaMergeable {
+
+}
+
+//general delta return type
+class Delta (
+  d: AkkaReplicaSystem#Obj
+  )  {
+    var delta :AkkaReplicaSystem#Obj = d
+  }
+
+//delta return type if method returns something too
+class ResultWrapper[T <: Object] (v: T, d: AkkaReplicaSystem#Obj)
+extends Delta(d) {
+  val value: T = v
+}
+
+
+

consys-core/src/main/scala/de/tuda/stg/consys/core/akka/DeltaMergeable.scala

@@ -0,0 +1,13 @@
+package de.tuda.stg.consys.core.akka
+
+/**
+ * Classes with this trait allow being updated with a delta state
+ * @tparam T delta state type
+ */
+trait DeltaMergeable {
+
+  private[akka] def merge(other:AkkaReplicaSystem#Obj)
+
+}
+
+

consys-japi/src/main/java/de/tuda/stg/consys/japi/JConsistencyLevels.java

@@ -16,5 +16,5 @@ public interface JConsistencyLevels {
 
 	@Local ConsistencyLabel CMRDT = ConsistencyLabel.CmRDT$.MODULE$;
 	@Local ConsistencyLabel CVRDT = ConsistencyLabel.CvRDT$.MODULE$;
-
+	@Local ConsistencyLabel DCRDT = ConsistencyLabel.DCRDT$.MODULE$;
 }

demos/counter/src/main/java/de/tuda/stg/consys/demo/counter/CounterBenchmark.java → demos/counter/src/main/java/de/tuda/stg/consys/demo/dcrdt/CounterBenchmark.java

@@ -1,19 +1,13 @@
-package de.tuda.stg.consys.demo.counter;
+package de.tuda.stg.consys.demo.dcrdt;
 
 import com.typesafe.config.Config;
 import de.tuda.stg.consys.bench.OutputFileResolver;
 import de.tuda.stg.consys.demo.DemoBenchmark;
-import de.tuda.stg.consys.demo.counter.schema.Counter;
+import de.tuda.stg.consys.demo.dcrdt.schema.Counter;
 import de.tuda.stg.consys.japi.JRef;
-import de.tuda.stg.consys.japi.impl.JReplicaSystems;
-import de.tuda.stg.consys.japi.impl.akka.JAkkaReplicaSystem;
 import org.checkerframework.com.google.common.collect.Sets;
 import scala.Option;
 
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
 /**
  * Created on 10.10.19.
  *
@@ -32,6 +26,7 @@ public CounterBenchmark(Config config, Option<OutputFileResolver> outputResolver
 
 	@Override
 	public void setup() {
+
 		if (processId() == 0) {
 			counter = system().replicate("counter", new Counter(0), getWeakLevel());
 		} else {

demos/counter/src/main/java/de/tuda/stg/consys/demo/counter/Demo.java → demos/counter/src/main/java/de/tuda/stg/consys/demo/dcrdt/Demo.java

@@ -1,4 +1,4 @@
-package de.tuda.stg.consys.demo.counter;
+package de.tuda.stg.consys.demo.dcrdt;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;

demos/counter/src/main/java/de/tuda/stg/consys/demo/counter/JMHBenchmark.java → demos/counter/src/main/java/de/tuda/stg/consys/demo/dcrdt/JMHBenchmark.java

@@ -1,7 +1,7 @@
-package de.tuda.stg.consys.demo.counter;
+package de.tuda.stg.consys.demo.dcrdt;
 
 import de.tuda.stg.consys.core.ConsistencyLabel;
-import de.tuda.stg.consys.demo.counter.schema.Counter;
+import de.tuda.stg.consys.demo.dcrdt.schema.Counter;
 import de.tuda.stg.consys.japi.JConsistencyLevels;
 import de.tuda.stg.consys.japi.JRef;
 import de.tuda.stg.consys.japi.JReplicaSystem;

demos/counter/src/main/java/de/tuda/stg/consys/demo/counter/schema/Counter.java → demos/counter/src/main/java/de/tuda/stg/consys/demo/dcrdt/schema/Counter.java

@@ -1,4 +1,4 @@
-package de.tuda.stg.consys.demo.counter.schema;
+package de.tuda.stg.consys.demo.dcrdt.schema;
 
 import java.io.Serializable;
 

demos/dcrdt-demo/README.md

@@ -0,0 +1,81 @@
+# Implementing Delta-CRDTs in Consys
+
+Consys supports implementing custom Delta-CRDTs, as presented in `Almeida et al. - 2018 - Delta State Replicated Data Types`. 
+
+## Delta-CRDT overview
+Delta-CRDTs are an extension of CRDTs, which are a replicated data structure synchronized using a `merge` method, which merges two states into a new one. 
+
+Delta-CRDTs provide the advantage of reduced data transfer, as they are synchronized through _delta-states_. If a replica has changed, it will only need to send a difference. 
+Other replicas can incorporate these changes by implementing the `merge` method, which takes a _delta-state_ parameter. 
+
+The structure must implement a _join-semilattice_; Please see the referenced publication for more details on the formal requirements it has to meet.
+## Implementing a Delta-CRDT
+
+Contrary to the typical workflow of Consys, using a Delta-CRDT structure requires implementing a custom class defining a `merge` method. Additionally, methods must follow a certain convention to convey whether they result in a delta state. 
+Automatically inferring delta states is not currently in scope of this project.
+
+The following exemplifies a DCRDT implementation using an AddOnlySet. 
+
+```
+public class AddOnlySetString extends DeltaCRDT implements Serializable {
+
+    private Set<String> set = new HashSet<String>();
+
+
+    @Override
+    public void merge(Object other) {
+        if (other instanceof Set) {
+            Set<String> s = (Set<String>) other;
+            set.addAll(s);
+        }
+    }
+
+    // adds a new String to this set
+    public Delta addElement(String str) {
+        set.add(str);
+        Set<String> s = new HashSet<String>();
+        s.add(str);
+        return new Delta(s);
+    }
+
+    // adds a new String to this set. 
+    // Returns whether the current local set did not yet include this string.
+    public ResultWrapper<Boolean> addElement2(String str) {
+        boolean result = set.add(str);
+        Set<String> s = new HashSet<String>();
+        s.add(str);
+        return new ResultWrapper<>(result, s);
+    }
+
+
+}
+```
+The `addElement` method returns a Delta instance containing _delta-state_. 
+`merge()` takes an `Object` parameter. Any value wrapped in a `Delta` or `ResultWrapper` instance will be passed here.
+
+
+Things to note:
+* The class must be Serializable. When a replica is initially registered using `replicate()`, it is transmitted to other clients as a whole, without the use of delta states.
+* As of yet, akka does not support generics, which is why the merge method only takes an `Object`. This is also why this example explicitly uses strings. 
+* If a method results in a changed state, it must return a `Delta` instance that includes the delta state. Akka will transmit this Delta to other replicas by invoking their `merge` method.
+* If a method intended to return a value results in a changed state, it must return a `ResultWrapper` object, which allows setting a delta state and an arbitrary value. `ResultWrapper` takes a type parameter, as akka's generics limitation does not apply here.
+* Please not that DCRDT methods are currently not yet atomic. This can pose a problem if a set being iterated over is modified by a `merge()` call. This issue will be fixed in a future version.
+
+Once implemented correctly, instances of DCRDT classes can be used just like any other data type in akka:
+
+```
+// ...
+if (master) {
+    set = system().replicate("counter", new AddOnlySetString(), JConsistencyLevels.DCRDT);
+} else {
+    set = system().lookup("counter", AddOnlySetString.class, JConsistencyLevels.DCRDT);
+}
+
+// ...
+set.ref().addElement("Hello");
+
+// ...
+if (! set.ref().addElement2("Hello").value) {
+    System.out.println("element already in set");
+}
+```