[1.2.0] Event ordering [API-1102]

internal/cluster/connection_listener_binder.go

@@ -46,7 +46,6 @@ type ConnectionListenerBinder struct {
 	regs                  map[types.UUID]listenerRegistration
 	correlationIDs        map[types.UUID][]int64
 	subscriptionToMembers map[types.UUID]map[types.UUID]struct{}
-	memberSubscriptions   map[types.UUID][]types.UUID
 	regsMu                *sync.RWMutex
 	connectionCount       int32
 	smart                 bool
@@ -67,7 +66,6 @@ func NewConnectionListenerBinder(
 		regs:                  map[types.UUID]listenerRegistration{},
 		correlationIDs:        map[types.UUID][]int64{},
 		subscriptionToMembers: map[types.UUID]map[types.UUID]struct{}{},
-		memberSubscriptions:   map[types.UUID][]types.UUID{},
 		regsMu:                &sync.RWMutex{},
 		logger:                logger,
 		smart:                 smart,
@@ -123,9 +121,6 @@ func (b *ConnectionListenerBinder) Remove(ctx context.Context, id types.UUID) er
 	b.logger.Trace(func() string {
 		return fmt.Sprintf("removing listener %s:\nconns: %v,\nregs: %v", id, conns, b.regs)
 	})
-	for _, conn := range conns {
-		b.removeMemberSubscriptions(conn.memberUUID)
-	}
 	return b.sendRemoveListenerRequests(ctx, reg.removeRequest, conns...)
 }
 
@@ -269,9 +264,6 @@ func (b *ConnectionListenerBinder) handleConnectionOpened(e *ConnectionStateChan
 
 func (b *ConnectionListenerBinder) handleConnectionClosed(e *ConnectionStateChangedEvent) {
 	atomic.AddInt32(&b.connectionCount, -1)
-	b.regsMu.Lock()
-	b.removeMemberSubscriptions(e.Conn.memberUUID)
-	b.regsMu.Unlock()
 }
 
 func (b *ConnectionListenerBinder) connExists(conn *Connection, subID types.UUID) bool {
@@ -291,17 +283,4 @@ func (b *ConnectionListenerBinder) addSubscriptionToMember(subID types.UUID, mem
 		b.subscriptionToMembers[subID] = mems
 	}
 	mems[memberUUID] = struct{}{}
-	b.memberSubscriptions[memberUUID] = append(b.memberSubscriptions[memberUUID], subID)
-}
-
-func (b *ConnectionListenerBinder) removeMemberSubscriptions(memberUUID types.UUID) {
-	// this method should be called under lock
-	subs, found := b.memberSubscriptions[memberUUID]
-	if !found {
-		return
-	}
-	for _, sub := range subs {
-		delete(b.subscriptionToMembers, sub)
-	}
-	delete(b.memberSubscriptions, memberUUID)
 }

internal/invocation/invocation_service.go

@@ -41,23 +41,24 @@ type Handler interface {
 }
 
 type Service struct {
+	handler         Handler
+	logger          ilogger.Logger
 	requestCh       chan Invocation
-	urgentRequestCh chan Invocation
 	responseCh      chan *proto.ClientMessage
-	// removeCh carries correlationIDs to be removed
-	removeCh        chan int64
 	doneCh          chan struct{}
 	groupLostCh     chan *GroupLostEvent
 	invocations     map[int64]Invocation
-	handler         Handler
+	urgentRequestCh chan Invocation
 	eventDispatcher *event.DispatchService
-	logger          ilogger.Logger
-	state           int32
+	// removeCh carries correlationIDs to be removed
+	removeCh chan int64
+	executor stripeExecutor
+	state    int32
 }
 
 func NewService(
 	handler Handler,
-	eventDispacher *event.DispatchService,
+	eventDispatcher *event.DispatchService,
 	logger ilogger.Logger) *Service {
 	s := &Service{
 		requestCh:       make(chan Invocation),
@@ -68,15 +69,17 @@ func NewService(
 		groupLostCh:     make(chan *GroupLostEvent),
 		invocations:     map[int64]Invocation{},
 		handler:         handler,
-		eventDispatcher: eventDispacher,
+		eventDispatcher: eventDispatcher,
 		logger:          logger,
 		state:           ready,
+		executor:        newStripeExecutor(5, 100),
 	}
 	s.eventDispatcher.Subscribe(EventGroupLost, serviceSubID, func(event event.Event) {
 		go func() {
 			s.groupLostCh <- event.(*GroupLostEvent)
 		}()
 	})
+	s.executor.start()
 	go s.processIncoming()
 	return s
 }
@@ -85,6 +88,7 @@ func (s *Service) Stop() {
 	if !atomic.CompareAndSwapInt32(&s.state, ready, stopped) {
 		return
 	}
+	s.executor.stop()
 	close(s.doneCh)
 }
 
@@ -183,7 +187,16 @@ func (s *Service) handleClientMessage(msg *proto.ClientMessage) {
 				return fmt.Sprintf("invocation with unknown correlation ID: %d", correlationID)
 			})
 		} else if inv.EventHandler() != nil {
-			go inv.EventHandler()(msg)
+			handler := func() {
+				inv.EventHandler()(msg)
+			}
+			if inv.PartitionID() == -1 {
+				// Execute on a random worker
+				s.executor.dispatchRandom(handler)
+				return
+			}
+			partitionID := uint32(inv.PartitionID())
+			s.executor.dispatch(partitionID, handler)
 		}
 		return
 	}

internal/invocation/stripe_executor.go

@@ -0,0 +1,73 @@
+package invocation
+
+import (
+	"math/rand"
+	"sync"
+)
+
+// stripeExecutor executes given "tasks" preserving the order among the ones
+// that are given with the same key
+type stripeExecutor struct {
+	quit            chan struct{}
+	wg              *sync.WaitGroup
+	executeFunction func(queue chan func(), quit chan struct{}, wg *sync.WaitGroup)
+	tasks           []chan func()
+	queueCount      uint32
+}
+
+// newStripeExecutor returns a new stripeExecutor with configured queueCount and queueSize
+func newStripeExecutor(queueCount, queueSize uint32) stripeExecutor {
+	se := stripeExecutor{
+		tasks:      make([]chan func(), queueCount),
+		queueCount: queueCount,
+	}
+	for ind := range se.tasks {
+		se.tasks[ind] = make(chan func(), queueSize)
+	}
+	se.quit = make(chan struct{})
+	se.wg = &sync.WaitGroup{}
+	se.executeFunction = defaultExecuteFnc
+	return se
+}
+
+// start fires up the workers for each queue
+func (se stripeExecutor) start() {
+	se.wg.Add(int(se.queueCount))
+	for ind := range se.tasks {
+		ind := ind
+		go se.executeFunction(se.tasks[ind], se.quit, se.wg)
+	}
+}
+
+// dispatch sends the handler "task" to the appropriate queue, "tasks"
+// with the same key end up on the same queue
+func (se stripeExecutor) dispatch(key uint32, handler func()) {
+	se.tasks[key%se.queueCount] <- handler
+}
+
+func (se stripeExecutor) dispatchRandom(handler func()) {
+	key := rand.Int31n(int32(se.queueCount))
+	se.dispatch(uint32(key), handler)
+}
+
+// stop blocks until all workers are stopped.
+func (se stripeExecutor) stop() {
+	close(se.quit)
+	se.wg.Wait()
+}
+
+func (se stripeExecutor) setExecutorFnc(custom func(queue chan func(), quit chan struct{}, wg *sync.WaitGroup)) {
+	se.executeFunction = custom
+}
+
+func defaultExecuteFnc(queue chan func(), quit chan struct{}, wg *sync.WaitGroup) {
+	defer wg.Done()
+	for {
+		select {
+		case task := <-queue:
+			task()
+		case <-quit:
+			return
+		}
+	}
+}

internal/invocation/stripe_executor_test.go

@@ -0,0 +1,139 @@
+package invocation
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/goleak"
+)
+
+type orderChecker struct {
+	*testing.T
+	previousCallArg int
+}
+
+func (oc *orderChecker) call(arg int) {
+	assert.Equal(oc.T, oc.previousCallArg+1, arg, "order of the tasks is not preserved")
+	oc.previousCallArg = arg
+}
+
+func Test_defaultExecuteFnc(t *testing.T) {
+	oc := &orderChecker{T: t}
+	tasks := make(chan func(), 3)
+	for i := 1; i < 4; i++ {
+		tmp := i
+		tasks <- func() {
+			oc.call(tmp)
+		}
+	}
+	quit := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go defaultExecuteFnc(tasks, quit, &wg)
+	assert.Eventually(t, func() bool {
+		return oc.previousCallArg == 3
+	}, time.Second, time.Millisecond*200, "execute function could not finish the tasks")
+	close(quit)
+	go func() {
+		wg.Wait()
+		// Just to see if goroutine finished
+		oc.previousCallArg = 10
+	}()
+	assert.Eventually(t, func() bool {
+		return oc.previousCallArg == 10
+	}, time.Second, time.Millisecond*200, "execute function did not notify about its finish")
+}
+
+func Test_serialExecutor_dispatch(t *testing.T) {
+	tests := []struct {
+		queueCount    uint32
+		key           uint32
+		expectedIndex int32
+	}{
+		{
+			queueCount:    4,
+			key:           2,
+			expectedIndex: 2,
+		},
+		{
+			queueCount:    2,
+			key:           2,
+			expectedIndex: 0,
+		},
+		{
+			queueCount:    2,
+			key:           3,
+			expectedIndex: 1,
+		},
+		{
+			queueCount:    2,
+			key:           4,
+			expectedIndex: 0,
+		},
+		{
+			queueCount:    1,
+			key:           5,
+			expectedIndex: 0,
+		},
+	}
+	for ind, tt := range tests {
+		t.Run(fmt.Sprintf("QueueCount: %d, Key: %d", tt.queueCount, tt.key), func(t *testing.T) {
+			se := newStripeExecutor(tt.queueCount, 0)
+			tmpHandler := func() {
+				panic(ind)
+			}
+			go se.dispatch(tt.key, tmpHandler)
+			select {
+			case <-se.tasks[tt.expectedIndex]:
+			case <-time.After(time.Second):
+				assert.FailNow(t, "dispatcher did not dispatch to correct queue")
+			}
+		})
+	}
+}
+
+func Test_serialExecutor_start(t *testing.T) {
+	t.Logf("enabled leak check")
+	defer goleak.VerifyNone(t)
+	t.Run("Functionality test", func(t *testing.T) {
+		var orderCheckers []*orderChecker
+		type pair struct {
+			handler func()
+			key     uint32
+		}
+		// create orderCheckers, index corresponding to key
+		for i := 1; i <= 100; i++ {
+			orderCheckers = append(orderCheckers, &orderChecker{T: t})
+		}
+		// populate task queues
+		// assume we have orderCheckers a,b,c, we will have
+		// a1,b1,c1,b2,c2,a2,a3,c3,b3
+		var tasks []pair
+		for i := 1; i <= 3; i++ {
+			tmp := i
+			for _, perm := range rand.Perm(100) {
+				key := perm
+				tasks = append(tasks, pair{key: uint32(key), handler: func() {
+					orderCheckers[key].call(tmp)
+				}})
+			}
+		}
+
+		se := newStripeExecutor(3, 3)
+		se.start()
+		go func() {
+			for _, task := range tasks {
+				se.dispatch(task.key, task.handler)
+			}
+		}()
+		time.Sleep(time.Second * 1)
+		for _, oc := range orderCheckers {
+			assert.Equal(t, 3, oc.previousCallArg, "task did not complete")
+		}
+		se.stop()
+	})
+}