nil

TODO.md

@@ -2,7 +2,6 @@
 
 ### TODO
 
-- [ ] inc replication recv buffer size
 - [ ] tweak of batcher yield
 - [ ] pack.Payload reuse memory, json.NewEncoder(os.Stdout)
 - [ ] metrics isolation by cluster
@@ -21,6 +20,7 @@
 - [ ] add Operator for Filter
   - count, filter, regex, sort, split, rename
 - [ ] RowsEvent avro
+- [X] inc binlog replication recv buffer size
 - [X] alert mysql binlog lags
 - [X] dbc participants -i // show internal buffers
 - [X] model.RowsEvent add dbus timestamp
@@ -129,4 +129,6 @@
   - It might exceed max event size: 1MB
     mysql seems to auto-chunk the big event into chunks of small events
   - It might malloc a very big memory in RowsEvent struct
+  - mysql packet max payload len = (1<<24 -1)
 - OSC tools will make 'ALTER' very complex, whence dbusd not able to clear table columns cache
+  - use SQL comment to solve it 

cmd/dbusd/chaos.sh

@@ -11,7 +11,7 @@
 # term2$ sh chaos.sh
 
 for i in `seq 1000`; do
-    ./dbusd -cluster -pprof :10120 -rpc 9877 -api 9897 &
+    ./dbusd -cluster -c test -pprof :10120 -rpc 9877 -api 9897 &
     pid=$!
     sleep 10
     sleep $[ ( $RANDOM % 10 ) + 10 ]s

engine/packet.go

@@ -3,8 +3,6 @@ package engine
 import (
 	"fmt"
 	"sync/atomic"
-
-	"github.com/funkygao/dbus/pkg/sys"
 )
 
 // Payloader defines the contract of Packet payload.
@@ -31,29 +29,21 @@ type KeyValuer interface {
 //
 // TODO hide it to private.
 type Packet struct {
-	_padding0   [sys.CacheLineSize / 8]uint64 // avoid false sharing
 	recycleChan chan *Packet
 
-	_padding1 [sys.CacheLineSize / 8]uint64
-	refCount  int32
+	refCount int32
+	acker    Acker // the Input it originates from
+	// buf []byte TODO reuse memory
 
-	_padding2 [sys.CacheLineSize / 8]uint64
 	// Ident is used for routing.
 	Ident string
 
-	_padding3 [sys.CacheLineSize / 8]uint64 // TODO [7]uint64 should be enough
 	// Metadata is used to hold arbitrary data you wish to include.
 	// Engine completely ignores this field and is only to be used for
 	// pass-through data.
 	Metadata interface{}
 
-	_padding4 [sys.CacheLineSize / 8]uint64
-	acker     Acker
-
-	_padding5 [sys.CacheLineSize / 8]uint64
-	Payload   Payloader
-
-	//	buf     []byte TODO reuse memory
+	Payload Payloader
 }
 
 func newPacket(recycleChan chan *Packet) *Packet {

pkg/batcher/disruptor.go

@@ -11,7 +11,7 @@ import (
 // disruptor is a batched lock free queue that borrows design from disruptor.
 // It maintains a queue with the sematics of all succeed and advance or any fails and retry.
 type disruptor struct {
-	_padding0 [sys.CacheLineSize / 8]uint64
+	_padding0 [sys.CacheLineSize / 8]uint64 // avoid false sharing
 	capacity  uint32
 
 	_padding1 [sys.CacheLineSize / 8]uint64