[1.2.0] Event ordering [API-1102]

client_it_test.go

@@ -23,6 +23,7 @@ import (
 	"log"
 	"reflect"
 	"runtime"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -35,7 +36,9 @@ import (
 	"github.com/hazelcast/hazelcast-go-client/hzerrors"
 	"github.com/hazelcast/hazelcast-go-client/internal"
 	"github.com/hazelcast/hazelcast-go-client/internal/it"
+	"github.com/hazelcast/hazelcast-go-client/internal/murmur"
 	"github.com/hazelcast/hazelcast-go-client/internal/proxy"
+	"github.com/hazelcast/hazelcast-go-client/internal/serialization"
 	"github.com/hazelcast/hazelcast-go-client/logger"
 	"github.com/hazelcast/hazelcast-go-client/types"
 )
@@ -140,6 +143,115 @@ func TestClientMemberEvents(t *testing.T) {
 	})
 }
 
+func TestClientEventOrder(t *testing.T) {
+	it.MapTester(t, func(t *testing.T, m *hz.Map) {
+		ctx := context.Background()
+		// events should be processed in this order
+		const (
+			noPrevEvent = 0
+			addEvent    = 1
+			removeEvent = 2
+		)
+		// populate event order checkers
+		var checkers []*int32
+		for i := 0; i < 20; i++ {
+			var state int32
+			checkers = append(checkers, &state)
+		}
+		// init listener conf
+		var c hz.MapEntryListenerConfig
+		c.NotifyEntryAdded(true)
+		c.NotifyEntryRemoved(true)
+		var tasks sync.WaitGroup
+		// add and remove are separate tasks
+		tasks.Add(len(checkers) * 2)
+		it.MustValue(m.AddEntryListener(ctx, c, func(e *hz.EntryNotified) {
+			state := checkers[e.Key.(int64)]
+			switch e.EventType {
+			case hz.EntryAdded:
+				if !atomic.CompareAndSwapInt32(state, noPrevEvent, noPrevEvent) {
+					panic("order is not preserved")
+				}
+				// keep the executor busy, make sure remove event is not processed before this
+				time.Sleep(500 * time.Millisecond)
+				if !atomic.CompareAndSwapInt32(state, noPrevEvent, addEvent) {
+					panic("order is not preserved")
+				}
+				tasks.Done()
+			case hz.EntryRemoved:
+				if !atomic.CompareAndSwapInt32(state, addEvent, removeEvent) {
+					panic("order is not preserved")
+				}
+				tasks.Done()
+			}
+		}))
+		for i := range checkers {
+			tmp := i
+			go func(index int) {
+				it.MustValue(m.Put(ctx, index, "test"))
+				it.MustValue(m.Remove(ctx, index))
+			}(tmp)
+		}
+		tasks.Wait()
+	})
+}
+
+func calculatePartitionID(ss *serialization.Service, key interface{}) (int32, error) {
+	kd, err := ss.ToData(key)
+	if err != nil {
+		return 0, err
+	}
+	return murmur.HashToIndex(kd.PartitionHash(), 271), nil
+}
+
+func TestClientEventHandlingOrder(t *testing.T) {
+	// Create custom cluster, and client from it
+	cls := it.StartNewClusterWithOptions("event-order-test-cluster", 15701, it.MemberCount())
+	defer cls.Shutdown()
+	conf := cls.DefaultConfig()
+	ctx := context.Background()
+	c := it.MustValue(hz.StartNewClientWithConfig(ctx, conf)).(*hz.Client)
+	defer c.Shutdown(ctx)
+	ss := it.MustValue(serialization.NewService(&conf.Serialization)).(*serialization.Service)
+	// Create test map
+	m := it.MustValue(c.GetMap(ctx, "TestClientEventHandlingOrder")).(*hz.Map)
+	var lc hz.MapEntryListenerConfig
+	lc.NotifyEntryAdded(true)
+	var (
+		// have 271 partitions by default
+		partitionToEvent = make([][]int, 271)
+		// wait for all events to be processed
+		wg sync.WaitGroup
+		// access it with atomic package
+		count int32
+	)
+	wg.Add(1)
+	handler := func(event *hz.EntryNotified) {
+		atomic.AddInt32(&count, 1)
+		// it is okay to use conversion, since greatest key is 1000
+		key := int(event.Key.(int64))
+		pid, err := calculatePartitionID(ss, key)
+		if err != nil {
+			panic(err)
+		}
+		partitionToEvent[pid] = append(partitionToEvent[pid], key)
+		if count == 1000 {
+			// last event processed
+			wg.Done()
+		}
+	}
+	it.MustValue(m.AddEntryListener(ctx, lc, handler))
+	for i := 1; i <= 1000; i++ {
+		it.MustValue(m.Put(ctx, i, "test"))
+	}
+	wg.Wait()
+	for _, keys := range partitionToEvent {
+		if !sort.IntsAreSorted(keys) {
+			t.Fatalf("events are not processed in order, event keys:\n%v\n", keys)
+		}
+	}
+}
+
 func TestClientHeartbeat(t *testing.T) {
 	// Slow test.
 	t.SkipNow()
@@ -618,7 +730,7 @@ func TestClientStartShutdownMemoryLeak(t *testing.T) {
 		ctx := context.Background()
 		var max uint64
 		var m runtime.MemStats
-		const limit = 8 * 1024 * 1024 // 8 MB
+		const limit = 8 * 1024 * 1024 // 16 MB
 		runtime.GC()
 		runtime.ReadMemStats(&m)
 		base := m.Alloc
@@ -634,8 +746,6 @@ func TestClientStartShutdownMemoryLeak(t *testing.T) {
 			t.Logf("memory allocation: %d at iteration: %d", m.Alloc, i)
 			if m.Alloc > base && m.Alloc-base > limit {
 				max = m.Alloc - base
-			}
-			if max > limit {
 				t.Fatalf("memory allocation: %d > %d (base: %d) at iteration: %d", max, limit, base, i)
 			}
 		}

internal/invocation/invocation_service.go

@@ -41,23 +41,24 @@ type Handler interface {
 }
 
 type Service struct {
+	handler         Handler
 	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          logger.LogAdaptor
-	state           int32
+	// removeCh carries correlationIDs to be removed
+	removeCh chan int64
+	executor *stripeExecutor
+	logger   logger.LogAdaptor
+	state    int32
 }
 
 func NewService(
 	handler Handler,
-	eventDispacher *event.DispatchService,
+	eventDispatcher *event.DispatchService,
 	logger logger.LogAdaptor) *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(),
 	}
 	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,15 @@ 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)
+			}
+			partitionID := msg.PartitionID()
+			// no specific partition (-1) are dispatched randomly in dispatch func.
+			ok := s.executor.dispatch(int(partitionID), handler)
+			if !ok {
+				s.logger.Warnf("event could not be processed, corresponding queue is full. PartitionID: %d, CorrelationID: %d", partitionID, correlationID)
+			}
 		}
 		return
 	}

internal/invocation/stripe_executor.go

@@ -0,0 +1,93 @@
+package invocation
+
+import (
+	"math/rand"
+	"runtime"
+	"sync"
+)
+
+var (
+	// Default values differ from java impl. Also queue size is calculated differently.
+	// Java Client: queueSize per worker = defaultEventQueueCapacity / defaultEventWorkerCount
+	// Go Client: queueSize per worker = defaultEventQueueCapacity
+	defaultEventQueueCapacity = 10000
+	defaultEventWorkerCount   = runtime.NumCPU()
+)
+
+// executor represents the function that will run on workers of stripeExecutor.
+type executor func(queue chan func(), quit chan struct{}, wg *sync.WaitGroup)
+
+// stripeExecutor executes given "tasks" preserving the order among the ones that are given with the same key.
+type stripeExecutor struct {
+	quit       chan struct{}
+	execFn     executor
+	taskQueues []chan func()
+	queueCount int
+	wg         sync.WaitGroup
+}
+
+// newStripeExecutor returns a new stripeExecutor with default configuration.
+func newStripeExecutor() *stripeExecutor {
+	return newStripeExecutorWithConfig(defaultEventWorkerCount, defaultEventQueueCapacity)
+}
+
+// newStripeExecutor returns a new stripeExecutor with configured queueCount and queueSize. If parameters are not greater than zero, it panics.
+func newStripeExecutorWithConfig(queueCount, queueSize int) *stripeExecutor {
+	if queueCount <= 0 {
+		panic("queueCount must be greater than 0")
+	}
+	if queueSize <= 0 {
+		panic("queueSize must be greater than 0")
+	}
+	se := stripeExecutor{
+		taskQueues: make([]chan func(), queueCount),
+		queueCount: queueCount,
+	}
+	for i := range se.taskQueues {
+		se.taskQueues[i] = make(chan func(), queueSize)
+	}
+	se.quit = make(chan struct{})
+	se.execFn = defaultExecFn
+	return &se
+}
+
+// start fires up the workers for each queue.
+func (se *stripeExecutor) start() {
+	se.wg.Add(se.queueCount)
+	for i := range se.taskQueues {
+		go se.execFn(se.taskQueues[i], se.quit, &se.wg)
+	}
+}
+
+// dispatch sends the handler "task" to one of the appropriate taskQueues, "tasks" with the same key end up on the same queue. Returns false if queue is full and could not dispatch.
+func (se *stripeExecutor) dispatch(key int, task func()) bool {
+	if key < 0 {
+		// dispatch random.
+		key = rand.Intn(se.queueCount)
+	}
+	select {
+	case se.taskQueues[key%se.queueCount] <- task:
+	default:
+		// do not block if queue is full.
+		return false
+	}
+	return true
+}
+
+// stop blocks until all workers are stopped.
+func (se *stripeExecutor) stop() {
+	close(se.quit)
+	se.wg.Wait()
+}
+
+func defaultExecFn(queue chan func(), quit chan struct{}, wg *sync.WaitGroup) {
+	defer wg.Done()
+	for {
+		select {
+		case task := <-queue:
+			task()
+		case <-quit:
+			return
+		}
+	}
+}