support raft cluster

Author: smallnest

bitmaps.go

@@ -2,17 +2,30 @@ package basalt
 
 import (
 	"encoding/binary"
+	"fmt"
 	"io"
 	"sync"
 
 	"github.com/RoaringBitmap/roaring"
 	"github.com/smallnest/log"
 )
 
+// OP bitmaps operations
+type OP byte
+
+const (
+	BmOpAdd     OP = 1
+	BmOpAddMany    = 2
+	BmOpRemove     = 3
+	BmOpDrop       = 4
+	BmOpClear      = 5
+)
+
 // Bitmaps contains all bitmaps of namespace.
 type Bitmaps struct {
-	mu      sync.RWMutex
-	bitmaps map[string]*Bitmap
+	mu            sync.RWMutex
+	bitmaps       map[string]*Bitmap
+	writeCallback func(op OP, value string)
 }
 
 // NewBitmaps creates a Bitmaps.
@@ -29,7 +42,12 @@ type Bitmap struct {
 }
 
 // Add adds a value.
-func (bs *Bitmaps) Add(name string, v uint32) {
+func (bs *Bitmaps) Add(name string, v uint32, callback bool) {
+	if bs.writeCallback != nil && callback {
+		bs.writeCallback(BmOpAdd, fmt.Sprintf("%s,%d", name, v))
+		return
+	}
+
 	bs.mu.Lock()
 	bm := bs.bitmaps[name]
 	if bm == nil {
@@ -46,7 +64,12 @@ func (bs *Bitmaps) Add(name string, v uint32) {
 }
 
 // AddMany adds multiple values.
-func (bs *Bitmaps) AddMany(name string, v []uint32) {
+func (bs *Bitmaps) AddMany(name string, v []uint32, callback bool) {
+	if bs.writeCallback != nil && callback {
+		bs.writeCallback(BmOpAddMany, fmt.Sprintf("%s,%d", name, ints2str(v)))
+		return
+	}
+
 	bs.mu.Lock()
 	bm := bs.bitmaps[name]
 	if bm == nil {
@@ -63,7 +86,12 @@ func (bs *Bitmaps) AddMany(name string, v []uint32) {
 }
 
 // Remove removes a value.
-func (bs *Bitmaps) Remove(name string, v uint32) {
+func (bs *Bitmaps) Remove(name string, v uint32, callback bool) {
+	if bs.writeCallback != nil && callback {
+		bs.writeCallback(BmOpRemove, fmt.Sprintf("%s,%d", name, v))
+		return
+	}
+
 	bs.mu.Lock()
 	bm := bs.bitmaps[name]
 	if bm == nil {
@@ -80,14 +108,24 @@ func (bs *Bitmaps) Remove(name string, v uint32) {
 }
 
 // RemoveBitmap removes a bitmap.
-func (bs *Bitmaps) RemoveBitmap(name string) {
+func (bs *Bitmaps) RemoveBitmap(name string, callback bool) {
+	if bs.writeCallback != nil && callback {
+		bs.writeCallback(BmOpDrop, name)
+		return
+	}
+
 	bs.mu.Lock()
 	delete(bs.bitmaps, name)
 	bs.mu.Unlock()
 }
 
 // ClearBitmap clear a bitmap.
-func (bs *Bitmaps) ClearBitmap(name string) {
+func (bs *Bitmaps) ClearBitmap(name string, callback bool) {
+	if bs.writeCallback != nil && callback {
+		bs.writeCallback(BmOpClear, name)
+		return
+	}
+
 	bs.mu.RLock()
 	bm := bs.bitmaps[name]
 	if bm == nil {

cmd/raft_server/README.md

@@ -0,0 +1,30 @@
+## 分布式位图服务
+
+基于raft的位图服务，可以搭建一个n个节点的raft集群，保证数据一致性。
+
+位图服务是一个写少读多的服务，所以基于raft的实现可以满足性能的要求。
+
+同时，考虑到位图服务的应用场景并不是严格强一致性的场景， 读操作并不基于raft的线性读或者lease read，而是保证最终一致性。
+
+
+### 测试集群
+
+以三个节点的集群为例:
+
+```
+basalt --id 1 --peers http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --addr :18972 --data bitmaps1.bdb
+basalt --id 2 --peers http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --addr :28972 --data bitmaps2.bdb
+basalt --id 3 --peers http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --addr :38972 --data bitmaps3.bdb
+```
+
+
+测试在第一个节点增加一个数据:
+```sh
+ basalt git:(master) ✗ curl "http://127.0.0.1:18972/add/test/1000"
+```
+
+在第二个节点检查这个数据是否存在，正常应该返回`200 OK`
+```
+➜  basalt git:(master) ✗ curl -v "http://127.0.0.1:28972/exists/test/1000"
+< HTTP/1.1 200 OK
+```

cmd/raft_server/basalt.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+	"strings"
+
+	"github.com/rpcxio/basalt"
+	"github.com/rpcxio/etcd/raft/raftpb"
+)
+
+var (
+	addr     = flag.String("addr", ":18972", "the listened address")
+	dataFile = flag.String("data", "bitmaps.bdb", "the persisted file")
+
+	peers = flag.String("peers", "http://127.0.0.1:12379", "comma separated peers in a cluster")
+	id    = flag.Int("id", 1, "node ID")
+	join  = flag.Bool("join", false, "join an existing cluster")
+)
+
+func main() {
+	flag.Parse()
+
+	if _, err := os.Stat(*dataFile); os.IsNotExist(err) {
+		f, err := os.Create(*dataFile)
+		if err != nil {
+			log.Fatalf("failed to create file %s: %v", *dataFile, err)
+		}
+		f.Close()
+	}
+
+	// bitmap
+	bitmaps := basalt.NewBitmaps()
+	srv := basalt.NewServer(*addr, bitmaps, nil, *dataFile)
+
+	// raft
+	proposeC := make(chan string)
+	defer close(proposeC)
+	confChangeC := make(chan raftpb.ConfChange)
+	defer close(confChangeC)
+
+	var raftServer *basalt.RaftServer
+	getSnapshot := func() ([]byte, error) { return raftServer.GetSnapshot() }
+	commitC, errorC, snapshotterReady := basalt.NewRaftNode(*id, strings.Split(*peers, ","), *join, getSnapshot, proposeC, confChangeC)
+
+	raftServer = basalt.NewRaftServer(srv, <-snapshotterReady, proposeC, commitC, errorC)
+
+	if err := srv.Serve(); err != nil {
+		log.Fatalf("failed to start basalt services:%v", err)
+	}
+	srv.Close()
+}

go.mod

@@ -1,14 +1,15 @@
 module github.com/rpcxio/basalt
 
-go 1.13
+go 1.14
 
 require (
 	github.com/RoaringBitmap/roaring v0.4.21
 	github.com/go-redis/redis v6.15.7+incompatible
 	github.com/julienschmidt/httprouter v1.3.0
+	github.com/rpcxio/etcd v0.0.0-20200729120139-f9cde972fd94
 	github.com/smallnest/log v0.0.0-20190128090703-5dc5752d8772
 	github.com/smallnest/rpcx v0.0.0-20200213044823-78d7a4d32e2a
 	github.com/soheilhy/cmux v0.1.4
 	github.com/tidwall/redcon v1.2.1
-	golang.org/x/text v0.3.2
+	go.uber.org/zap v1.14.1
 )