• Diego Albo Martínez
• Joris Quist
• Tomasz Motyka
• Wouter Zorgdrager

With Orleans our biggest goal is to provide developers with an easy-to-use framework that allows for implementing highly concurrent systems and deploy those systems with a minimum effort while achieving good performance.

The system is centered around the concept of Grains, which are the minimum computing unit, which developers can extend and implement their own functionality in an easy manner. These grains reside in Silos or servers, who are responsible from making Grains accessible and executable by clients by delivering them the messages sent.

Grains can also communicate among them in a straightforward way by means of fire-and-forget messages or asynchronous messages in which they wait for a response from other party.

We provide Grains as an abstract class that developers can extend and implement their functionality with. Grains provide one method to developers, the receive method, to which all messages destined for that grain are passed and have to be matched by the developer and implement its logic.

// Define the type of function that the Grains have
object Grain{
  type Receive = PartialFunction[Any, Unit]
}

// Grain abstract class extended by the users
abstract class Grain(val _id: String) extends Serializable  {
  def receive : Grain.Receive
  def store() = {
    println(s"Executing store function in grain with id ${_id}")
  }
}

As an example, a developer could program a grain to answer to the user in case they receive a certain message, below we provide a possible use a developer could do of that Grain abstract class to implement their own logic.

class GreeterGrain(_id: String) 
	extends Grain(_id)
  	with LazyLogging {

  /**
   *Receive method of the grain
   */
  def receive = {
    case ("hello", sender: Sender) =>
      logger.info("Replying to the sender!")
      sender ! "Hello World!"
    case _ => logger.error("Unexpected message")
  }
}

As can be seen, Grains and clients communicate through two different methods. Drawing a parallel with the akka framework, grains can make use of the ! (fire-and-forget) and ? (asynchronous response) methods to send a message to another grain. This makes the communication straightforward and simple and concise to program.

The sender of the message is also facilitated alongside the message itself so the grain always has a way of answering to the originator of the message.

Silos are the servers that store the grains and that make sure that clients can make reference to a grain, and are also in charge of load balancing and grain replication and storage. Silos can be differentiated into two different kinds.

• Master Silo. In charge of distributing the load between the slaves as well as storing the index of which grain is stored in which server. It is also the server who the clients address when wanting to perform control operations such as creating, deleting or searching for a grain.

• Slave Silos. In charge of holding the grains and delivering to them their corresponding messages. They also hold multiple Dispatchers (one per grain type), which can be created at runtime to deliver messages sent to a particular type of grain. This means that the system doesn't have to be restarted when trying to add a different kind of grain; just by calling the CreateGrain method in the Master, a new grain will be created given that class' definition and a new Dispatcher created to deliver messages to those grains.

Both Master and Silos have a Control Grain in charge of performing these cluster management operations.

Once the Silos are initiated, a client can connect to the master silo and create and start its own grains simply in no more than a couple lines of code.

// Create reference to the runtime and register the grain
val runtime = OrleansRuntime()
      .registerGrain[GreeterGrain]
      .setHost("localhost")
      .setPort(1400)
      .build()

// Create grain and wait for a response with that grain's ID 
val createGrainFuture: Future[GrainRef] =
	runtime.createGrain[GreeterGrain]()
val grainRef = Await.result(createGrainFuture, 5 seconds)

// Send a fire and forget message to the newly created grain
grainref ! "hello"

// or
// Send an async message and register a callback
grainref ? "how are you doing?" onComplete {
      case Success(response) =>
        println(response)
      case _ => println("got no response")
    }

In order to prevent concurrency issues, each grain is executed in a single thread, which guarantees that its inner variables will not be accessed concurrently. Furthermore, many grains can receive messages concurrently in a Silo, to increase throughput of the system.

Moreover, if a grain is under heavy load, the silos will be able to detect that through the gathered metrics of the dispatcher and replicate that grain so load can be distributed. This grain might be duplicated to the same silo or to a different silo altogether, and requests split between the two.

Also, if a grain is passive for a long time, that grain with its state will be persisted to persistent storage in a MongoDB database. Once that grain wants to be referenced by a client, the master and the server silo will activate the grain and load it in memory so it can receive messages again from the user.

For each silo we have to notion of a 'master' and a 'slave'. Currently only one master is supported. Both master and slave run using two threads: 1) control thread to verify slaves/master is still alive and send heartbeats, 2) packet-manager thread which receives packets.

Each silo is configured using a host and a port for UDP commmunication. Keep in mind that if run on the same computer, different ports per silo need to be used! A packet has the following form:

 case class Packet(packetType: String,
                    uuid: String,
                    timestamp: Long,
                    data: List[String] = List())

In the table below you can see the packets and its usages:

Note: All communication goes through the master. Although they are aware, slaves won't directly communicate to each other.