beehive

README

Beehive

Beehive is a distributed programming framework that comes with built-in transactions, replication, fault-tolerance, runtime instrumentation, and optimized placement.

• Installation
• Hello World
  • Message
  • Message Handler
  • Emit Hello
  • Start
  • Run a Cluster
• Deep Dive
  • Hives
  • Applications, Dictionaries, and Message Handlers
  • Map and Consistent Concurrency
  • Bees
  • Life of a Message
  • Reply Messages
  • HTTP Handlers
  • Detached Handlers
  • Transactions
  • Runtime Instrumentation and Optimization
• Projects using Beehive
• Mailing List
• Publications

Installation

Option 1. Install Beehive using goget:

curl -sL https://git.io/goget | bash -s -- github.com/kandoo/beehive

Option 2. Install go 1.4+, set up your GOPATH, and install Beehive using go get:

go get github.com/kandoo/beehive

Test Your Setup. Enter Beehive's root directory ($GOPATH/src/github.com/kandoo/beehive) and run:

go test -v

Hello World

Let's write a simple example that counts the number of times we have said hello to each person. You can find the complete example in the GoDoc.

Message

Beehive is based on asynchronous message passing. Naturally, the first step is to define a Hello message:

// Hello represents a message in our hello world example.
type Hello struct {
	Name string // Name is the name of the person saying hello.
}

Message Handler

To handle Hello messages, we need to write an application that has a message handler for Hello. Pretty analogous to HTTP handlers, except we are processing messages not HTTP requests.

A message handler in Beehive consists of two functions: (i) Rcv and (ii) Map. Rcv is the function that actually processes a message. Since Beehive provides a generic runtime Map function that works for all applications, let's skip the Map function for now, and we will explain it in next section.

This is a simple Rcv function that handles Hello messages (don't panic it's all comments ;) ):

// Rcvf receives the message and the context.
func Rcvf(msg beehive.Msg, ctx beehive.RcvContext) error {
    // msg is an envelope around the Hello message.
    // You can retrieve the Hello, using msg.Data() and then
    // you need to assert that its a Hello.
    hello := msg.Data().(Hello)
    // Using ctx.Dict you can get (or create) a dictionary.
    dict := ctx.Dict("hello_dict")
    // Using Get(), you can get the value associated with
    // a key in the dictionary. Keys are always string
    // and values are generic interface{}'s.
    v, err := dict.Get(hello.Name)
    // If there is an error, the entry is not in the
    // dictionary. Otherwise, we set cnt based on
    // the value we already have in the dictionary
    // for that name.
    cnt := 0
    if err == nil {
        cnt = v.(int)
    }
    // Now we increment the count.
    cnt++
    // And then we print the hello message.
    ctx.Printf("hello %s (%d)!\n", hello.Name, cnt)
    // Finally we update the count stored in the dictionary.
    return dict.Put(hello.Name, cnt)
}

To register a message handler, we first create an application and then we register the Hello handler for our application:

// Create the hello world application and make sure .
app := beehive.NewApp("hello-world", beehive.Persistent(1))
// Register the handler for Hello messages.
app.HandleFunc(Hello{}, beehive.RuntimeMap(Rcvf), Rcvf)

Note that our application is persistent and will save its state on 1 node (i.e., persistent but not replicated).

Emit Hello

Now, to send a Hello message, you can emit it:

// Emit simply emits a message, here a
// string of your name.
go beehive.Emit(Hello{Name: "your name"})
// Emit another message with the same name
// to test the counting feature.
go beehive.Emit(Hello{Name: "your name"})

Whenever you emit a Hello message, it will be processed by all applications that have a handler for Hello. Here, we have only one application, but you could create different applications with different handlers for Hello. All of them would receive the Hello message.

Start

Finally, we need to start Beehive:

beehive.Start()

When you run the application (say go run helloworld.go), you will have the following output:

bee 1/HelloWorld/0000000000000402> hello your name (1)!
bee 1/HelloWorld/0000000000000402> hello your name (2)!

When you run the application one more time, you will see the following output:

bee 1/HelloWorld/0000000000000402> hello your name (3)!
bee 1/HelloWorld/0000000000000402> hello your name (4)!

Note that the counter is saved on disk, so you can safely restart your application.

Run a Cluster

This simple hello world application is actually a distributed application. The message handler is automatically sharded by Hello.Name. Later, we will explain how that happens. For now, let's just try to run our hello world application in a cluster.

Run the first node as you have done previously (go run helloworld.go). Wait until you see the hello messages:

bee 1/HelloWorld/0000000000000402> hello your name (5)!
bee 1/HelloWorld/0000000000000402> hello your name (6)!

Then, run a new node using the following command:

go run helloworld.go -addr localhost:7678 -paddrs localhost:7677 -statepath /tmp/beehive2

After you connect the second node, the first node should generate the following output:

bee 1/HelloWorld/0000000000000402> hello your name (7)!
bee 1/HelloWorld/0000000000000402> hello your name (8)!

Note that in the last command, -addr sets the listening address of the beehive server, -paddrs sets the address of the peers (the first node is listening on the default port, 7677), and -statepath sets where beehive should store its state and the dictionaries.

Note: You can reinitializing the cluster by removing both /tmp/beehive and /tmp/beehive2, and re-running the commands.

Deep Dive

Hives

A Hive is basically a Beehive server, representing one logical unit of computing (say, a physical or a virtual machine). Hives can form, join to, and leave a cluster. Beehive clusters are homogeneous, meaning that all hives in the same cluster are running the same set of applications.

Applications, Dictionaries, and Message Handlers

A Beehive application is a set of asynchronous message handlers. Message handlers simply process async messages and store their state in dictionaries. A dictionary is basically a hash map. Behind the scenes, these dictionaries are saved to disk and are replicated. A message handler is composed of a Rcv function that actually processes the message and a Map function declaring how messages should be sharded/partitioned. Beehive provides a generic Map functions (as you saw in the Hello World example) and also has a compiler that can generate Map functions based on your Rcv functions. Having said that, Map functions are almost always one-liners and are pretty easy to implement.

Map and Consistent Concurrency

To make the distributed and concurrent version of of message handlers, we want to balance the load of message processing among multiple go-routines across multiple hives. We need to do this in a way that the application's behavior remains identical to when we use only a single, centralized go-routine. To do so, we need to preserve the consistency of application dictionaries.

In other words, we want to make sure that each entry (or as we call them, cell) in an application dictionary is always accessed on the same logical go-routine. Otherwise, we can't guarantee that the application behaves consistently when distributed over multiple hives. For example, what would happen to our hello world application if two different go-routines could read and modify the same entry concurrently?

To that end, for each message, we need to know what are the keys used to process the message in the Rcv function of a message handler. We call this the mapped cells of that message. Each message handler, in addition to its Rcv function, needs to provide a Map function that maps an incoming message to cells or simply keys in application dictionaries. Map functions are usually very simple to implement, but you can also use Beehive's generic RuntimeMap function or the Beehive compiler to generate the Map function.

Bees

Applications and their message handlers are passive in Beehive. Internally, each hive has a set of go-routines called bees that run the message handlers for each application. Each bee exclusively owns a set of cells. These cells are the cells that must be accessed by the same go-routine to preserve consistency. Cells are locked by bees using an internal distributed consensus mechanism implemented using Raft. Bees persist their cells if needed and, when a hive restarts, we reload all the bees.

Moreover, for replicated applications, bees will form a colony of bees (itself and some other bees on other hives) and will consistently replicate its cells using raft. When a bee fails, we hand its cells and workload to other bees in the colony. The size of a colony is equal to the application's replication factor.

Life of a Message

When a message is emitted on a hive, we first pass that message to the Map function of the registered message handlers for that type of message. The Map function returns the mapped cells of that message. Then, we relay the message to the bee that has any of the keys in the mapped cell. That bee in response calls the Rcv function of that message handler. This bee may be on the same hive or can be on another hive.

If there is no such bee (this is the first message mapped to those cells), we elect a hive, create one bee on it, and relay the message. By default, we create the bee on the local hive, but applications can register custom placement methods to change this behavior. For example, using this option, one can implement a random placement.

Reply Messages

So far, we have seen emitted messages but we cannot use that for communication between applications, say to implement a request response system. For example, a better way to implement our hello world application would be emitting a Hello message and waiting for a response that contains the count.

In Beehive, you can reply to a message using ReplyTo method. For example, we can rewrite our hello world application using reply messages:

func Rcvf(msg beehive.Msg, ctx beehive.RcvContext) error {
	hello := msg.Data().(Hello)
	dict := ctx.Dict("hello_dict")
	v, err := dict.Get(hello.Name)
	cnt := 0
	if err == nil {
			cnt = v.(int)
	}
	cnt++
	// Reply to the message with the count of hellos.
	ctx.ReplyTo(msg, cnt)
	return dict.Put(name, cnt)
}

To use this version, we also need to emit Hello messages and wait for the application's response. We can implement this using Sync:

go beehive.Start()
defer beehive.Stop()

name = "your name"
// Sync sends the Hello message and waits until it receives the reply.
res, err := beehive.Sync(context.TODO(), Hello{Name: name})
if err != nil {
	...
	return
}
cnt := res.(int)
fmt.Printf("%s (%d)!\n", name, cnt)

HTTP Handlers

Beehive applications can register custom HTTP handlers, handling requests to URLs with the /apps/APP_NAME/ prefix. To register a HTTP handler, applications can use HandleHTTP:

app.HandleHTTP("/", httpHandler)

Internally we use Gorilla mux, and we expose the sub-router of each application. As a result, applications can match against regular expression, use parameters in the URL, and specify HTTP methods:

app.HandleHTTP("/{name}", httpHandler).Methods("POST")

HTTP handlers usually communicate with the application using synchronous messaging. For example, we can implement a HTTP handler for our hello world application as follows:

type HelloHTTPHandler struct {}

func (h *HelloHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	name, ok := vars["name"]
	if !ok {
		http.Error(w, "no name", http.StatusBadRequest)
		return
	}

	res, err := beehive.Sync(context.TODO(), Hello{Name: name})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	fmt.Fprintf(w, "hello %s (%d)\n", name, res.(int))
}

Now you can send HTTP requests to this handler using curl:

curl -X POST localhost:7677/apps/hello-world/yourname

And you should see the following output:

hello yourname (1)

Detached Handlers

Sometimes you need to create go-routines that read data from network connections or file system and generate messages. For example, to implement a network listener for a custom protocol, we need to run a network listener and then run go-routines for each established connection. To implement such functionalities in Beehive, we can use detached handlers. A detached handler is different than a message handler in a sense that it is started in its Start method and only receives replies.

The Detached Example demonstrates how we used detached handlers to implement a text-based protocol in Beehive. When you run this example, you can telnet to port 6789 for sending names to the hello world application:

telnet localhost 6789
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
your name
hello your name (1)!
your name
hello your name (2)!
your name
hello your name (3)!

Transactions

By default, all applications are transactional in Beehive, meaning that the bee opens up a transaction for each call to the Rcv function. If the Rcv panics or returns an error the transaction is automatically aborted, and otherwise is committed.

In Beehive, transactions include all the modification to the dictionaries and all the messages emitted in a Rcv function. That is, when a transaction is aborted, all the messages and all dictionary modifications are dropped.

Transactions are by default implicit, but message handlers can explicitly control transactions. To open a transaction in a Rcv function you can use BeginTx(), CommitTx() and AbortTx():

func Rcvf(msg bh.Msg, ctx bh.RcvContext) error {
	ctx.BeginTx()
	d1 := ctx.Dict("d1")
	d1.Put("k", []byte{1})
	ctx.Emit(MyMsg("test1"))
	// Update d1/k and emit test1.
	ctx.CommitTx()

	ctx.BeginTx()
	d2 := ctx.Dict("d2")
	d2.Put("k", []byte{2})
	ctx.Emit(MyMsg("test2"))
	// DO NOT update d2/k and DO NOT emit test2.
	ctx.AbortTx()
}

To disable automatic transactions for an application (say, for performance reasons), use the NonTransactional option.

Runtime Instrumentation and Optimization

Beehive is capable of automatic runtime instrumentation. It measures the messages exchanged between different bees and can use it to live migrate bees to minimize latency. You can enable instrumentation and optimization by passing -instrument command line arguments to your Beehive program. You can also access the instrumentation data on Beehive's web interface (by default, http://localhost:7677/).

Projects using Beehive

• Beehive Distributed SDN Controller
• TaskQ

Mailing List

Google group: https://groups.google.com/forum/#!forum/beehive-dev

Please report bugs in github, not in the group.

Publications

• Soheil Hassas Yeganeh, Yashar Ganjali, Beehive: Beehive: Simple Distributed Programming in Software-Defined Networks, SOSR 2016.
• Soheil Hassas Yeganeh, Yashar Ganjali, Beehive: Towards a Simple Abstraction for Scalable Software-Defined Networking, HotNets XIII, 2014.

Documentation

Overview

Example

package main

import "github.com/kandoo/beehive"

// Hello represents a message in our hello world example.
type Hello struct {
	Name string // Name is the name of the person saying hello.
}

// Rcvf receives the message and the context.
func Rcvf(msg beehive.Msg, ctx beehive.RcvContext) error {
	// msg is an envelope around the Hello message.
	// You can retrieve the Hello, using msg.Data() and then
	// you need to assert that its a Hello.
	hello := msg.Data().(Hello)
	// Using ctx.Dict you can get (or create) a dictionary.
	dict := ctx.Dict("hello_dict")
	// Using Get(), you can get the value associated with
	// a key in the dictionary. Keys are always string
	// and values are generic interface{}'s.
	v, err := dict.Get(hello.Name)
	// If there is an error, the entry is not in the
	// dictionary. Otherwise, we set cnt based on
	// the value we already have in the dictionary
	// for that name.
	cnt := 0
	if err == nil {
		cnt = v.(int)
	}
	// Now we increment the count.
	cnt++
	// And then we print the hello message.
	ctx.Printf("hello %s (%d)!\n", hello.Name, cnt)
	// Finally we update the count stored in the dictionary.
	return dict.Put(hello.Name, cnt)
}

func main() {
	// Create the hello world application and make sure .
	app := beehive.NewApp("hello-world", beehive.Persistent(1))
	// Register the handler for Hello messages.
	app.HandleFunc(Hello{}, beehive.RuntimeMap(Rcvf), Rcvf)
	// Emit simply emits a message, here a
	// string of your name.
	go beehive.Emit(Hello{Name: "your name"})
	// Emit another message with the same name
	// to test the counting feature.
	go beehive.Emit(Hello{Name: "your name"})
	// Start the DefualtHive.
	beehive.Start()
}

Example (Detached)

package main

import (
	"bufio"
	"encoding/gob"
	"errors"
	"fmt"
	"net"

	"github.com/kandoo/beehive"
	"github.com/kandoo/beehive/Godeps/_workspace/src/github.com/golang/glog"
)

// HelloDetached represents a message in our hello world example.
type HelloDetached struct {
	Name string // Name is the name of the person saying hello.
}

// HelloCount represents a message sent as a reply to a HelloDetached.
type HelloCount struct {
	Name  string // Name of the person.
	Count int    // Number of times we have said hello.
}

// RcvfDetached receives the message and the context.
func RcvfDetached(msg beehive.Msg, ctx beehive.RcvContext) error {
	// msg is an envelope around the Hello message.
	// You can retrieve the Hello, using msg.Data() and then
	// you need to assert that its a Hello.
	hello := msg.Data().(HelloDetached)
	// Using ctx.Dict you can get (or create) a dictionary.
	dict := ctx.Dict("hello_dict")
	// Using Get(), you can get the value associated with
	// a key in the dictionary. Keys are always string
	// and values are generic interface{}'s.
	v, err := dict.Get(hello.Name)
	// If there is an error, the entry is not in the
	// dictionary. Otherwise, we set cnt based on
	// the value we already have in the dictionary
	// for that name.
	cnt := 0
	if err == nil {
		cnt = v.(int)
	}
	// Now we increment the count.
	cnt++
	// Reply to the message with the count of hellos.
	ctx.Reply(msg, HelloCount{Name: hello.Name, Count: cnt})
	// Finally we update the count stored in the dictionary.
	return dict.Put(hello.Name, cnt)
}

// HelloListener is a detached handler that acts as a newtork listener for
// our example.
type HelloListener struct {
	lis net.Listener
}

// NewHelloListener creates a new HelloListener.
func NewHelloListener() *HelloListener {
	lis, err := net.Listen("tcp", ":6789")
	if err != nil {
		glog.Fatalf("cannot start listener: %v", err)
	}

	return &HelloListener{lis: lis}
}

// Start is called once the detached handler starts.
func (h *HelloListener) Start(ctx beehive.RcvContext) {
	defer h.lis.Close()

	for {
		c, err := h.lis.Accept()
		if err != nil {
			return
		}
		// Start a new detached handler for the connection.
		go ctx.StartDetached(&HelloConn{conn: c})
	}
}

// Stop is called when the hive is stopping.
func (h *HelloListener) Stop(ctx beehive.RcvContext) {
	h.lis.Close()
}

// Rcv receives replies to HelloListener which we do not expect to receive.
// Note that HelloConn emits hellos and should receives replies.
func (h *HelloListener) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	return errors.New("unexpected message")
}

// HelloConn is a detached handler that handles a connection.
type HelloConn struct {
	conn net.Conn
}

// Start is called once the detached handler starts.
func (h *HelloConn) Start(ctx beehive.RcvContext) {
	defer h.conn.Close()

	r := bufio.NewReader(h.conn)
	for {
		name, _, err := r.ReadLine()
		if err != nil {
			return
		}
		ctx.Emit(HelloDetached{Name: string(name)})
	}
}

// Stop is called when the hive is stopping.
func (h *HelloConn) Stop(ctx beehive.RcvContext) {
	h.conn.Close()
}

// Rcv receives HelloCount messages.
func (h *HelloConn) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	reply := msg.Data().(HelloCount)
	_, err := fmt.Fprintf(h.conn, "hello %s (%d)!\n", reply.Name, reply.Count)
	return err
}

func main() {
	// Create the hello world application and make sure .
	app := beehive.NewApp("hello-world", beehive.Persistent(1))
	// Register the handler for Hello messages.
	app.HandleFunc(HelloDetached{},
		beehive.RuntimeMap(RcvfDetached), RcvfDetached)
	// Register the detached handler for the hello world listener.
	app.Detached(NewHelloListener())
	// Start the DefaultHive.
	beehive.Start()
}

func init() {
	// We need to register HelloCount on gob.
	gob.Register(HelloCount{})
}

Example (HTTP)

package main

import (
	"fmt"
	"net/http"

	"github.com/kandoo/beehive"
	"github.com/kandoo/beehive/Godeps/_workspace/src/github.com/gorilla/mux"
	"github.com/kandoo/beehive/Godeps/_workspace/src/golang.org/x/net/context"
)

// HelloHTTP represents a message in our hello world example.
type HelloHTTP struct {
	Name string // Name is the name of the person saying hello.
}

// RcvfHTTP receives the message and the context.
func RcvfHTTP(msg beehive.Msg, ctx beehive.RcvContext) error {
	// msg is an envelope around the Hello message.
	// You can retrieve the Hello, using msg.Data() and then
	// you need to assert that its a Hello.
	hello := msg.Data().(HelloHTTP)
	// Using ctx.Dict you can get (or create) a dictionary.
	dict := ctx.Dict("hello_dict")
	// Using Get(), you can get the value associated with
	// a key in the dictionary. Keys are always string
	// and values are generic interface{}'s.
	v, err := dict.Get(hello.Name)
	// If there is an error, the entry is not in the
	// dictionary. Otherwise, we set cnt based on
	// the value we already have in the dictionary
	// for that name.
	cnt := 0
	if err == nil {
		cnt = v.(int)
	}
	// Now we increment the count.
	cnt++
	// Reply to the message with the count of hellos.
	ctx.Reply(msg, cnt)
	// Finally we update the count stored in the dictionary.
	return dict.Put(hello.Name, cnt)
}

type HelloHTTPHandler struct{}

func (h *HelloHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	name, ok := vars["name"]
	if !ok {
		http.Error(w, "no name", http.StatusBadRequest)
		return
	}

	res, err := beehive.Sync(context.TODO(), HelloHTTP{Name: name})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	fmt.Fprintf(w, "hello %s (%d)\n", name, res.(int))
}

func main() {
	// Create the hello world application and make sure .
	app := beehive.NewApp("hello-world", beehive.Persistent(1))
	// Register the handler for Hello messages.
	app.HandleFunc(HelloHTTP{}, beehive.RuntimeMap(RcvfHTTP), RcvfHTTP)
	// Register the HTTP handler for the hello world application.
	app.HandleHTTP("/{name}", &HelloHTTPHandler{}).Methods("POST")
	// Start the DefaultHive.
	beehive.Start()
}

Example (Reply)

package main

import (
	"fmt"

	"github.com/kandoo/beehive"
	"github.com/kandoo/beehive/Godeps/_workspace/src/github.com/golang/glog"
	"github.com/kandoo/beehive/Godeps/_workspace/src/golang.org/x/net/context"
)

// HelloReply represents a message in our hello world example.
type HelloReply struct {
	Name string // Name is the name of the person saying hello.
}

// RcvfReply receives the message and the context.
func RcvfReply(msg beehive.Msg, ctx beehive.RcvContext) error {
	// msg is an envelope around the Hello message.
	// You can retrieve the Hello, using msg.Data() and then
	// you need to assert that its a Hello.
	hello := msg.Data().(HelloReply)
	// Using ctx.Dict you can get (or create) a dictionary.
	dict := ctx.Dict("hello_dict")
	// Using Get(), you can get the value associated with
	// a key in the dictionary. Keys are always string
	// and values are generic interface{}'s.
	v, err := dict.Get(hello.Name)
	// If there is an error, the entry is not in the
	// dictionary. Otherwise, we set cnt based on
	// the value we already have in the dictionary
	// for that name.
	cnt := 0
	if err == nil {
		cnt = v.(int)
	}
	// Now we increment the count.
	cnt++
	// Reply to the message with the count of hellos.
	ctx.Reply(msg, cnt)
	// Finally we update the count stored in the dictionary.
	return dict.Put(hello.Name, cnt)
}

func main() {
	// Create the hello world application and make sure .
	app := beehive.NewApp("hello-world", beehive.Persistent(1))
	// Register the handler for Hello messages.
	app.HandleFunc(HelloReply{}, beehive.RuntimeMap(RcvfReply), RcvfReply)

	// Start the default hive.
	go beehive.Start()
	defer beehive.Stop()

	name := "your name"
	for i := 0; i < 2; i++ {
		// Sync sends the Hello message and waits until it receives the reply.
		res, err := beehive.Sync(context.TODO(), HelloReply{Name: name})
		if err != nil {
			glog.Fatalf("error in sending Hello: %v", err)
		}
		cnt := res.(int)
		fmt.Printf("hello %s (%d)!\n", name, cnt)
	}
}

Index

• Constants
• Variables
• func Emit(msgData interface{})
• func MsgType(d interface{}) string
• func Start() error
• func Stop() error
• func Sync(ctx context.Context, req interface{}) (res interface{}, err error)
• type App
  • func NewApp(name string, options ...AppOption) App
• type AppCellKey
  • func (ack AppCellKey) Cell() CellKey
  • func (ack AppCellKey) IsNil() bool
• type AppOption
  • func InRate(rate bucket.Rate, max uint64) AppOption
  • func NonTransactional() AppOption
  • func OutRate(rate bucket.Rate, max uint64) AppOption
  • func Persistent(replicationFactor int) AppOption
  • func Placement(m PlacementMethod) AppOption
  • func Sticky() AppOption
  • func Transactional() AppOption
• type BeeInfo
• type CellKey
• type Colony
  • func ColonyFromBytes(b []byte) (Colony, error)
  • func (c *Colony) AddFollower(id uint64) bool
  • func (c *Colony) Bytes() ([]byte, error)
  • func (c Colony) Contains(id uint64) bool
  • func (c Colony) DeepCopy() Colony
  • func (c *Colony) DelFollower(id uint64) bool
  • func (c Colony) Equals(thatC Colony) bool
  • func (c Colony) IsFollower(id uint64) bool
  • func (c Colony) IsLeader(id uint64) bool
  • func (c Colony) IsNil() bool
  • func (c Colony) String() string
• type Context
• type DetachedHandler
  • func NewTimer(tick time.Duration, fn func()) DetachedHandler
• type Emitter
• type Handler
• type Hive
  • func NewHive(opts ...HiveOption) Hive
• type HiveConfig
  • func (c HiveConfig) RaftElectTimeout() time.Duration
  • func (c HiveConfig) RaftHBTimeout() time.Duration
• type HiveInfo
• type HiveOption
  • func Addr(a string) HiveOption
  • func BatchSize(s uint) HiveOption
  • func CmdChBufSize(s uint) HiveOption
  • func ConnTimeout(t time.Duration) HiveOption
  • func DataChBufSize(s uint) HiveOption
  • func InstrumentOptimize(i bool) HiveOption
  • func OptimizeThresh(t uint) HiveOption
  • func PeerAddrs(pa ...string) HiveOption
  • func Pprof(p bool) HiveOption
  • func RaftElectTicks(e int) HiveOption
  • func RaftFsyncTick(t time.Duration) HiveOption
  • func RaftHbeatTicks(h int) HiveOption
  • func RaftInFlights(f int) HiveOption
  • func RaftMaxMsgSize(s uint64) HiveOption
  • func RaftTick(t time.Duration) HiveOption
  • func RaftTickDelta(d time.Duration) HiveOption
  • func StatePath(p string) HiveOption
  • func SyncPoolSize(s uint) HiveOption
• type HiveState
• type MapContext
• type MapFunc
  • func RuntimeMap(rcv RcvFunc) MapFunc
• type MappedCells
  • func (mc MappedCells) Len() int
  • func (mc MappedCells) Less(i, j int) bool
  • func (mc MappedCells) LocalBroadcast() bool
  • func (mc MappedCells) String() string
  • func (mc MappedCells) Swap(i, j int)
• type MockMsg
  • func (m MockMsg) Data() interface{}
  • func (m MockMsg) From() uint64
  • func (m MockMsg) IsBroadCast() bool
  • func (m MockMsg) IsUnicast() bool
  • func (m MockMsg) NoReply() bool
  • func (m MockMsg) To() uint64
  • func (m MockMsg) Type() string
• type MockRcvContext
  • func (m MockRcvContext) AbortTx() error
  • func (m MockRcvContext) App() string
  • func (m MockRcvContext) BeeLocal() interface{}
  • func (m MockRcvContext) BeginTx() error
  • func (m MockRcvContext) CommitTx() error
  • func (m MockRcvContext) DeferReply(msg Msg) Repliable
  • func (m *MockRcvContext) Dict(name string) state.Dict
  • func (m *MockRcvContext) Emit(msgData interface{})
  • func (m MockRcvContext) Hive() Hive
  • func (m MockRcvContext) ID() uint64
  • func (m MockRcvContext) LockCells(keys []CellKey) error
  • func (m MockRcvContext) Printf(format string, a ...interface{})
  • func (m *MockRcvContext) Reply(msg Msg, replyData interface{}) error
  • func (m *MockRcvContext) SendToBee(msgData interface{}, to uint64)
  • func (m MockRcvContext) SendToCell(msgData interface{}, app string, cell CellKey)
  • func (m MockRcvContext) SetBeeLocal(d interface{})
  • func (m MockRcvContext) Snooze(d time.Duration)
  • func (m MockRcvContext) StartDetached(h DetachedHandler) uint64
  • func (m MockRcvContext) StartDetachedFunc(start StartFunc, stop StopFunc, rcv RcvFunc) uint64
  • func (m MockRcvContext) Sync(ctx context.Context, req interface{}) (res interface{}, err error)
• type Msg
• type PlacementMethod
• type RandomPlacement
  • func (r RandomPlacement) Place(cells MappedCells, thisHive Hive, liveHives []HiveInfo) HiveInfo
• type RcvContext
• type RcvFunc
• type Receiver
• type Repliable
  • func (r *Repliable) Reply(ctx RcvContext, replyData interface{})
• type StartFunc
• type StopFunc
• type Typed

Examples

• Package
• Package (Detached)
• Package (HTTP)
• Package (Reply)

Constants

const (
	Nil uint64 = 0
)

Variables

var (
	ErrOldTx       = errors.New("transaction has an old term")
	ErrIsNotMaster = errors.New("bee is not master")
)

var (
	ErrUnsupportedRequest = errors.New("registry: unsupported request")
	ErrInvalidParam       = errors.New("registry: invalid parameter")
	ErrNoSuchHive         = errors.New("registry: no such hive")
	ErrDuplicateHive      = errors.New("registry: duplicate hive")
	ErrNoSuchBee          = errors.New("registry: no such bee")
	ErrDuplicateBee       = errors.New("registry: duplicate bee")
)

var (
	// ErrSyncStopped returned when the sync handler is stopped before receiving
	// the response.
	ErrSyncStopped = bhgob.Error("sync: stopped")
	// ErrSyncNoSuchRequest returned when we cannot find the request for that
	// response.
	ErrSyncNoSuchRequest = bhgob.Error("sync: request not found")
	// ErrSyncDuplicateResponse returned when there is a duplicate repsonse to the
	// sync request.
	ErrSyncDuplicateResponse = bhgob.Error("sync: duplicate response")
)

var ErrInvalidCmd = errors.New("invalid command")

ErrInvalidCmd is returned when the requested command is invalid or is not supported by the receiver of the command.

Functions

func Emit

func Emit(msgData interface{})

Emit emits a message on DefaultHive.

func MsgType

func MsgType(d interface{}) string

MsgType returns the message type for d.

func Start

func Start() error

Start starts the DefaultHive. This method blocks.

func Stop

func Stop() error

Stop stops the DefaultHive. This method blocks.

func Sync

func Sync(ctx context.Context, req interface{}) (res interface{}, err error)

Sync processes a synchrounous message (req) and blocks until the response is recieved on DefaultHive.

Types

type App

type App interface {
	// Returns the app name.
	Name() string

	// Handles a specific message type using the handler. If msgType is an
	// name of msgType's reflection type.
	// instnace of MsgType, we use it as the type. Otherwise, we use the qualified
	Handle(msgType interface{}, h Handler) error
	// Hanldes a specific message type using the map and receive functions. If
	// msgType is an instnace of MsgType, we use it as the type. Otherwise, we use
	// the qualified name of msgType's reflection type.
	HandleFunc(msgType interface{}, m MapFunc, r RcvFunc) error

	// Regsiters the app's detached handler.
	Detached(h DetachedHandler)
	// Registers the detached handler using functions.
	DetachedFunc(start StartFunc, stop StopFunc, r RcvFunc)

	// Returns the state of this app that is used in the map function. This state
	// is NOT thread-safe and apps must synchronize for themselves.
	Dict(name string) state.Dict

	// HandleHTTP registers an HTTP handler for this application on
	// "/apps/name/path".
	//
	// Note: Gorilla mux is used internally. As such, it is legal to use path
	// parameters.
	HandleHTTP(path string, handler http.Handler) *mux.Route
	// HandleHTTPFunc registers an HTTP handler func for this application on
	// "/app/name/path".
	//
	// Note: Gorilla mux is used internally. As such, it is legal to use path
	// parameters.
	HandleHTTPFunc(path string,
		handler func(http.ResponseWriter, *http.Request)) *mux.Route
}

App represents an application in beehive. An app is a collection of stateful message handlers.

Methods in this interface are not thread-safe and must be called before the Hive starts.

func NewApp

func NewApp(name string, options ...AppOption) App

NewApp creates a new application on DefaultHive.

type AppCellKey

type AppCellKey struct {
	App  string
	Dict string
	Key  string
}

AppCellKey represents a key in a dictionary of a specific app.

func (AppCellKey) Cell

func (ack AppCellKey) Cell() CellKey

Cell returns the CellKey of this AppCellKey.

func (AppCellKey) IsNil

func (ack AppCellKey) IsNil() bool

IsNil returns whether the AppCellKey represents no cells.

type AppOption

type AppOption func(a *app)

AppOption represents an option for applications.

func InRate

func InRate(rate bucket.Rate, max uint64) AppOption

InRate is an application option that limits the rate of incoming messages of each bee of an application using a token bucket with the given rate and the given maximum.

func NonTransactional

func NonTransactional() AppOption

NonTransactional is an application option that makes the application non-transactional.

func OutRate

func OutRate(rate bucket.Rate, max uint64) AppOption

OutRate is an application option that limits the rate of outgoing messages of each bee of an application using a token bucket with the given rate and the given maximum.

func Persistent

func Persistent(replicationFactor int) AppOption

Persistent is an application option that makes the application's state persistent. The app state will be replciated on "replicationFactor" hives. This option also makes the application transactional.

func Placement

func Placement(m PlacementMethod) AppOption

Placement is an application option that customizes the default placement strategy for the application.

func Sticky

func Sticky() AppOption

Sticky is an application option that makes the application sticky. Bees of sticky apps are not migrated by the optimizer.

func Transactional

func Transactional() AppOption

Transactional is an application option that makes the application transactional. Transactions embody both application messages and its state.

type BeeInfo

type BeeInfo struct {
	ID       uint64 `json:"id"`
	Hive     uint64 `json:"hive"`
	App      string `json:"app"`
	Colony   Colony `json:"colony"`
	Detached bool   `json:"detached"`
}

BeeInfo stores the metadata about a bee.

type CellKey

type CellKey struct {
	Dict string
	Key  string
}

CellKey represents a key in a dictionary.

type Colony

type Colony struct {
	ID        uint64   `json:"id"`
	Leader    uint64   `json:"leader"`
	Followers []uint64 `json:"followers"`
}

Colony is the colony of bees that maintain a consistent state.

func ColonyFromBytes

func ColonyFromBytes(b []byte) (Colony, error)

ColonyFromBytes creates a colony from its []byte representation.

func (*Colony) AddFollower

func (c *Colony) AddFollower(id uint64) bool

AddFollower adds a follower to the colony. Returns false if id is already a follower.

func (*Colony) Bytes

func (c *Colony) Bytes() ([]byte, error)

Bytes returns the []byte representation of this colony.

func (Colony) Contains

func (c Colony) Contains(id uint64) bool

Contains returns whether id is the leader or a follower in this colony.

func (Colony) DeepCopy

func (c Colony) DeepCopy() Colony

DeepCopy creates a cloned copy of the colony.

func (*Colony) DelFollower

func (c *Colony) DelFollower(id uint64) bool

DelFollower deletes id from the followers of this colony. Returns false if id is not already a follower.

func (Colony) Equals

func (c Colony) Equals(thatC Colony) bool

Equals return whether c is equal to thatC.

func (Colony) IsFollower

func (c Colony) IsFollower(id uint64) bool

IsFollower retursn whether id is the follower in this colony.

func (Colony) IsLeader

func (c Colony) IsLeader(id uint64) bool

IsLeader returns whether id is the leader of this colony.

func (Colony) IsNil

func (c Colony) IsNil() bool

IsNil returns whether the colony does not represent a valid colony.

func (Colony) String

func (c Colony) String() string

type Context

type Context interface {
	// Hive returns the Hive of this context.
	Hive() Hive
	// App returns the name of the application of this context.
	App() string
	// Dict is a helper function that returns the specific dict within the state.
	Dict(name string) state.Dict
	// Sync processes a synchrounous message (req) and blocks until the response
	// is recieved.
	Sync(ctx context.Context, req interface{}) (res interface{}, err error)
	// Printf formats according to format string and writes the string on
	// standard output.
	//
	// Note: This method is solely for debugging your message handlers.
	// For proper logging, use glog.
	Printf(format string, a ...interface{})
}

Context is the interface shared between MapContext and RcvContext. It wraps Hive(), App() and Dict().

type DetachedHandler

type DetachedHandler interface {
	Receiver
	// Starts the handler. Note that this will run in a separate goroutine, and
	// you can block.
	Start(ctx RcvContext)
	// Stops the handler. This should notify the start method perhaps using a
	// channel.
	Stop(ctx RcvContext)
}

DetachedHandler in contrast to normal Handlers with Map and Rcv, starts in their own go-routine and emit messages. They do not listen on a particular message and only recv replys in their receive functions. Note that each app can have only one detached handler.

func NewTimer

func NewTimer(tick time.Duration, fn func()) DetachedHandler

NewTimer returns a detached handler that calls fn per tick.

type Emitter

type Emitter interface {
	Emit(msgData interface{})
}

type Handler

type Handler interface {
	Receiver
	Map(m Msg, c MapContext) MappedCells
}

Handler represents a message handler.

type Hive

type Hive interface {
	// ID of the hive. Valid only if the hive is started.
	ID() uint64
	// Config returns the hive configuration.
	Config() HiveConfig

	// Start starts the hive. This function blocks.
	Start() error
	// Stop stops the hive and all its apps. It blocks until the hive is actually
	// stopped.
	Stop() error

	// Creates an app with the given name and the provided options.
	// Note that apps are not active until the hive is started.
	NewApp(name string, opts ...AppOption) App

	// Emits a message containing msgData from this hive.
	Emit(msgData interface{})
	// Sends a message to a specific bee that owns a specific dictionary key.
	SendToCellKey(msgData interface{}, to string, dk CellKey)
	// Sends a message to a sepcific bee.
	SendToBee(msgData interface{}, to uint64)
	// Reply replies to the message.
	Reply(msg Msg, replyData interface{}) error
	// Sync processes a synchrounous message (req) and blocks until the response
	// is recieved.
	Sync(ctx context.Context, req interface{}) (res interface{}, err error)

	// Registers a message for encoding/decoding. This method should be called
	// only on messages that have no active handler. Such messages are almost
	// always replies to some detached handler.
	RegisterMsg(msg interface{})
}

Hive represents is the main active entity of beehive. It mananges all messages, apps and bees.

var DefaultHive Hive

DefaultHive is the hive used by Start() and NewApp().

func NewHive

func NewHive(opts ...HiveOption) Hive

NewHive creates a new hive based on the given configuration options.

type HiveConfig

type HiveConfig struct {
	Addr      string   // public address of the hive.
	PeerAddrs []string // peer addresses.
	StatePath string   // where to store state data.

	DataChBufSize uint // buffer size of the data channels.
	CmdChBufSize  uint // buffer size of the control channels.
	BatchSize     uint // number of messages to batch.
	SyncPoolSize  uint // number of sync go-routines.

	Pprof          bool // whether to enable pprof web handlers.
	Instrument     bool // whether to instrument apps on the hive.
	OptimizeThresh uint // when to notify the optimizer (in msg/s).

	RaftTick       time.Duration // the raft tick interval.
	RaftTickDelta  time.Duration // the maximum random delta added to the tick.
	RaftFsyncTick  time.Duration // the frequency of Fsync.
	RaftHBTicks    int           // number of raft ticks that fires a heartbeat.
	RaftElectTicks int           // number of raft ticks that fires election.
	RaftInFlights  int           // maximum number of inflights to a node.
	RaftMaxMsgSize uint64        // maximum size of an append message.

	ConnTimeout time.Duration // timeout for connections between hives.
}

HiveConfig represents the configuration of a hive.

func (HiveConfig) RaftElectTimeout

func (c HiveConfig) RaftElectTimeout() time.Duration

RaftElectTimeout returns the raft election timeout as RaftTick*RaftElectTicks.

func (HiveConfig) RaftHBTimeout

func (c HiveConfig) RaftHBTimeout() time.Duration

RaftHBTimeout returns the raft heartbeat timeout as RaftTick*RaftHBTicks.

type HiveInfo

type HiveInfo struct {
	ID   uint64 `json:"id"`
	Addr string `json:"addr"`
}

type HiveOption

type HiveOption args.V

HiveOption represents a configuration option of a hive.

func Addr

func Addr(a string) HiveOption

Addr represents the listening address of the hive used for both inter-hive RPC and its HTTP/web interface.

func BatchSize

func BatchSize(s uint) HiveOption

BatchSize represents the maximum batch size used for batching messages in a hive.

func CmdChBufSize

func CmdChBufSize(s uint) HiveOption

CmdChBufSize represents the size of the command channel used by hives, queen bees and bees.

func ConnTimeout

func ConnTimeout(t time.Duration) HiveOption

ConnTimeout represents the connection timeout for RPC connections.

func DataChBufSize

func DataChBufSize(s uint) HiveOption

DataChBufSize represents the size of the message channels used by hives, queen bees and bees.

func InstrumentOptimize

func InstrumentOptimize(i bool) HiveOption

InstrumentOptimize represents whether the hive should perform runtime intstrumentation and optimization.

func OptimizeThresh

func OptimizeThresh(t uint) HiveOption

OptimizeThresh represents the minimum message rate (i.e., the number of messages per second) after which we notify the optimizer.

func PeerAddrs

func PeerAddrs(pa ...string) HiveOption

PeerAddrs represents the peer addresses of hive.

func Pprof

func Pprof(p bool) HiveOption

Pprof represents whether the hive should enable pprof handlers on its HTTP interface.

func RaftElectTicks

func RaftElectTicks(e int) HiveOption

RaftElectTicks represents the number of ticks to start a new raft election.

func RaftFsyncTick

func RaftFsyncTick(t time.Duration) HiveOption

RaftFsyncTick represents when the hive should call fsync on written entires. 0 means immidiately after each write.

func RaftHbeatTicks

func RaftHbeatTicks(h int) HiveOption

RaftHbeatTicks represents the number of ticks to send a new raft heartbeat.

func RaftInFlights

func RaftInFlights(f int) HiveOption

RaftInFlights represents the maximum number of raft messages in flight.

func RaftMaxMsgSize

func RaftMaxMsgSize(s uint64) HiveOption

RaftMaxMsgSize represents the maximum number of entries in a raft message.

func RaftTick

func RaftTick(t time.Duration) HiveOption

RaftTick represents the raft tick.

func RaftTickDelta

func RaftTickDelta(d time.Duration) HiveOption

RaftTickDelta represents the random tick to add to the main raft tick.

func StatePath

func StatePath(p string) HiveOption

StatePath represents where the hive should save its state.

func SyncPoolSize

func SyncPoolSize(s uint) HiveOption

SyncPoolSize represents the number of sync go-routines running in a hive. These go-routine handle sync requests.

type HiveState

type HiveState struct {
	ID    uint64     `json:"id"`    // ID is the ID of the hive.
	Addr  string     `json:"addr"`  // Addr is the hive's address.
	Peers []HiveInfo `json:"peers"` // Peers of the hive.
}

HiveState represents the state of a hive.

type MapContext

type MapContext interface {
	Context

	// LocalMappedCells returns a mapped cell unique to the hive of this map
	// context.
	LocalMappedCells() MappedCells
}

MapContext is passed to the map functions of message handlers. It provides all the platform-level functions required to implement the map function.

type MapFunc

type MapFunc func(m Msg, c MapContext) MappedCells

MapFunc is a map function that maps a specific message to the set of keys in state dictionaries. This method is assumed not to be thread-safe and is called sequentially. If the return value is an empty set the message is broadcasted to all local bees. Also, if the return value is nil, the message is drop.

func RuntimeMap

func RuntimeMap(rcv RcvFunc) MapFunc

RuntimeMap generates an automatic runtime map function based on the given rcv function.

If there was an error in the rcv function, it will return "nil" and the message will be dropped.

type MappedCells

type MappedCells []CellKey

MappedCells is the list of dictionary keys returned by the map functions.

func (MappedCells) Len

func (mc MappedCells) Len() int

func (MappedCells) Less

func (mc MappedCells) Less(i, j int) bool

func (MappedCells) LocalBroadcast

func (mc MappedCells) LocalBroadcast() bool

LocalBroadcast returns whether the mapped cells indicate a local broadcast. An empty set means a local broadcast of message. Note that nil means drop.

func (MappedCells) String

func (mc MappedCells) String() string

func (MappedCells) Swap

func (mc MappedCells) Swap(i, j int)

type MockMsg

type MockMsg msg

MockMsg is a mock for Msg.

func (MockMsg) Data

func (m MockMsg) Data() interface{}

func (MockMsg) From

func (m MockMsg) From() uint64

func (MockMsg) IsBroadCast

func (m MockMsg) IsBroadCast() bool

func (MockMsg) IsUnicast

func (m MockMsg) IsUnicast() bool

func (MockMsg) NoReply

func (m MockMsg) NoReply() bool

func (MockMsg) To

func (m MockMsg) To() uint64

func (MockMsg) Type

func (m MockMsg) Type() string

type MockRcvContext

type MockRcvContext struct {
	CtxHive  Hive
	CtxApp   string
	CtxDicts *state.InMem
	CtxID    uint64
	CtxMsgs  []Msg
}

MockRcvContext is a mock for RcvContext.

func (MockRcvContext) AbortTx

func (m MockRcvContext) AbortTx() error

func (MockRcvContext) App

func (m MockRcvContext) App() string

func (MockRcvContext) BeeLocal

func (m MockRcvContext) BeeLocal() interface{}

func (MockRcvContext) BeginTx

func (m MockRcvContext) BeginTx() error

func (MockRcvContext) CommitTx

func (m MockRcvContext) CommitTx() error

func (MockRcvContext) DeferReply

func (m MockRcvContext) DeferReply(msg Msg) Repliable

func (*MockRcvContext) Dict

func (m *MockRcvContext) Dict(name string) state.Dict

func (*MockRcvContext) Emit

func (m *MockRcvContext) Emit(msgData interface{})

func (MockRcvContext) Hive

func (m MockRcvContext) Hive() Hive

func (MockRcvContext) ID

func (m MockRcvContext) ID() uint64

func (MockRcvContext) LockCells

func (m MockRcvContext) LockCells(keys []CellKey) error

func (MockRcvContext) Printf

func (m MockRcvContext) Printf(format string, a ...interface{})

func (*MockRcvContext) Reply

func (m *MockRcvContext) Reply(msg Msg, replyData interface{}) error

func (*MockRcvContext) SendToBee

func (m *MockRcvContext) SendToBee(msgData interface{}, to uint64)

func (MockRcvContext) SendToCell

func (m MockRcvContext) SendToCell(msgData interface{}, app string,
	cell CellKey)

func (MockRcvContext) SetBeeLocal

func (m MockRcvContext) SetBeeLocal(d interface{})

func (MockRcvContext) Snooze

func (m MockRcvContext) Snooze(d time.Duration)

func (MockRcvContext) StartDetached

func (m MockRcvContext) StartDetached(h DetachedHandler) uint64

func (MockRcvContext) StartDetachedFunc

func (m MockRcvContext) StartDetachedFunc(start StartFunc, stop StopFunc,
	rcv RcvFunc) uint64

func (MockRcvContext) Sync

func (m MockRcvContext) Sync(ctx context.Context, req interface{}) (
	res interface{}, err error)

type Msg

type Msg interface {
	// Type of the data in this message.
	Type() string
	// Data stored in the message.
	Data() interface{}
	// From returns the ID of the sender of this message.
	From() uint64
	// To returns the ID of the receiver of this message.
	To() uint64

	// NoReply returns whether we can reply to the message.
	NoReply() bool
	// IsBroadCast returns whether the message is a broadcast.
	IsBroadCast() bool
	// IsUnicast returns whether the message is a unicast.
	IsUnicast() bool
}

Msg is a generic interface for messages emitted in the system. Messages are defined for each type.

type PlacementMethod

type PlacementMethod interface {
	// Place returns the metadata of the hive chosen for cells. cells is the
	// mapped cells of a message according to the map function of the
	// application's message handler. thisHive is the local hive and liveHives
	// contains the meta data about live hives. Note that liveHives contains
	// thisHive.
	Place(cells MappedCells, thisHive Hive, liveHives []HiveInfo) HiveInfo
}

PlacementMethod represents a placement algorithm that chooses a hive among live hives for the given mapped cells. This interface is used only for the first message that is mapped to those cells.

The elected hive might go down, while the system is assigning the mapped cells to it. In such a case, the message will be placed locally after receiving an error.

type RandomPlacement

type RandomPlacement struct {
	*rand.Rand
}

RandomPlacement is a placement method that place mapped cells on a random hive.

func (RandomPlacement) Place

func (r RandomPlacement) Place(cells MappedCells, thisHive Hive,
	liveHives []HiveInfo) HiveInfo

type RcvContext

type RcvContext interface {
	Context

	// ID returns the bee id of this context.
	ID() uint64

	// Emit emits a message.
	Emit(msgData interface{})
	// SendToCell sends a message to the bee of the give app that owns the
	// given cell.
	SendToCell(msgData interface{}, app string, cell CellKey)
	// SendToBee sends a message to the given bee.
	SendToBee(msgData interface{}, to uint64)
	// Reply replies to a message: Sends a message from the current bee to the
	// bee that emitted msg.
	Reply(msg Msg, replyData interface{}) error
	// DeferReply returns a Repliable that can be used to reply to a
	// message (either a sync or a async message) later.
	DeferReply(msg Msg) Repliable

	// StartDetached spawns a detached handler.
	StartDetached(h DetachedHandler) uint64
	// StartDetachedFunc spawns a detached handler using the provide function.
	StartDetachedFunc(start StartFunc, stop StopFunc, rcv RcvFunc) uint64

	// LockCells proactively locks the cells in the given cell keys.
	LockCells(keys []CellKey) error

	// Snooze exits the Rcv function, and schedules the current message to be
	// enqued again after at least duration d.
	Snooze(d time.Duration)

	// BeeLocal returns the bee-local storage. It is an ephemeral memory that is
	// just visible to the current bee. Very similar to thread-locals in the scope
	// of a bee.
	BeeLocal() interface{}
	// SetBeeLocal sets a data in the bee-local storage.
	SetBeeLocal(d interface{})

	// Starts a transaction in this context. Transactions span multiple
	// dictionaries and buffer all messages. When a transaction commits all the
	// side effects will be applied. Note that since handlers are called in a
	// single bee, transactions are mostly for programming convinience and easy
	// atomocity.
	BeginTx() error
	// Commits the current transaction.
	// If the application has a 2+ replication factor, calling commit also means
	// that we will wait until the transaction is sufficiently replicated and then
	// commits the transaction.
	CommitTx() error
	// Aborts the transaction.
	AbortTx() error
}

RcvContext is passed to the rcv functions of message handlers. It provides all the platform-level functions required to implement the rcv function.

type RcvFunc

type RcvFunc func(m Msg, c RcvContext) error

RcvFunc is the function that handles a message. This method is called in parallel for different map-sets and sequentially within a map-set.

type Receiver

type Receiver interface {
	// Receives replies to messages emitted in this handler.
	Rcv(m Msg, c RcvContext) error
}

Receiver wraps Rcv.

type Repliable

type Repliable struct {
	From   uint64 // The ID of the bee that originally sent the message.
	SyncID uint64 // The sync message ID if the message was sync, otherwise 0.
}

Repliable is a serializable structure that can be used to reply to a message at any time. Repliable is always created using RcvContext.DeferReply().

Note: The fields in the Repliable are public for serialization. It is not advisable to modify these fields.

func (*Repliable) Reply

func (r *Repliable) Reply(ctx RcvContext, replyData interface{})

Reply replies to the Repliable using replyData.

type StartFunc

type StartFunc func(ctx RcvContext)

StartFunc is the start function of a detached handler.

type StopFunc

type StopFunc func(ctx RcvContext)

StopFunc is the stop function of a detached handler.

type Typed

type Typed interface {
	Type() string
}

Typed is a message data with an explicit type.