feat(queuev2): wire CachedScheduledQueue into the timer queue factory

Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>

Author: arzonus

service/history/queuev2/queue_scheduled_cached.go

@@ -0,0 +1,63 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package queuev2
+
+import (
+	"context"
+
+	hcommon "github.com/uber/cadence/service/history/common"
+)
+
+type cachedScheduledQueue struct {
+	*scheduledQueue
+	reader CachedQueueReader
+}
+
+func newCachedScheduledQueue(inner *scheduledQueue, reader CachedQueueReader) Queue {
+	// Wrap the queue state update so the reader evicts tasks below the current
+	// ack level whenever the queue persists its progress. GetMinReadLevel returns
+	// MaximumHistoryTaskKey when no virtual queues exist yet; UpdateReadLevel
+	// normalises that sentinel so it is treated as a no-op advance.
+	originalUpdateFn := inner.base.updateQueueStateFn
+	inner.base.updateQueueStateFn = func(ctx context.Context) {
+		originalUpdateFn(ctx)
+		reader.UpdateReadLevel(inner.base.virtualQueueManager.GetMinReadLevel())
+	}
+
+	return &cachedScheduledQueue{scheduledQueue: inner, reader: reader}
+}
+
+func (q *cachedScheduledQueue) NotifyNewTask(clusterName string, info *hcommon.NotifyTaskInfo) {
+	q.reader.Inject(info.Tasks)
+	q.scheduledQueue.NotifyNewTask(clusterName, info)
+}
+
+func (q *cachedScheduledQueue) Start() {
+	q.reader.Start()
+	q.scheduledQueue.Start()
+}
+
+func (q *cachedScheduledQueue) Stop() {
+	q.scheduledQueue.Stop()
+	q.reader.Stop()
+}

service/history/queuev2/queue_scheduled_cached_test.go

@@ -0,0 +1,231 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package queuev2
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"go.uber.org/mock/gomock"
+
+	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/persistence"
+	hcommon "github.com/uber/cadence/service/history/common"
+	"github.com/uber/cadence/service/history/config"
+	"github.com/uber/cadence/service/history/shard"
+	"github.com/uber/cadence/service/history/task"
+)
+
+func TestCachedScheduledQueue_Construction(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	ctrl := gomock.NewController(t)
+
+	mockShard := shard.NewTestContext(
+		t, ctrl,
+		&persistence.ShardInfo{ShardID: 10, RangeID: 1, TransferAckLevel: 0},
+		config.NewForTest(),
+	)
+
+	options := testScheduledQueueOptions()
+	mockReader := NewMockCachedQueueReader(ctrl)
+
+	inner := newScheduledQueue(mockShard, persistence.HistoryTaskCategoryTimer,
+		task.NewMockProcessor(ctrl), task.NewMockExecutor(ctrl),
+		mockShard.GetLogger(), metrics.NoopClient, metrics.NoopScope,
+		options, mockReader)
+
+	q := newCachedScheduledQueue(inner, mockReader)
+
+	require.NotNil(t, q)
+	csq, ok := q.(*cachedScheduledQueue)
+	require.True(t, ok, "expected *cachedScheduledQueue")
+	assert.NotNil(t, csq.reader)
+	assert.NotNil(t, csq.scheduledQueue)
+}
+
+func TestCachedScheduledQueue_NotifyNewTask_Empty(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockReader := NewMockCachedQueueReader(ctrl)
+
+	// Inject is always called even with nil tasks; the real implementation handles it.
+	mockReader.EXPECT().Inject([]persistence.Task(nil)).Times(1)
+
+	csq := &cachedScheduledQueue{
+		scheduledQueue: &scheduledQueue{
+			base: &queueBase{
+				metricsScope: metrics.NoopScope,
+			},
+			newTimerCh: make(chan struct{}, 1),
+		},
+		reader: mockReader,
+	}
+
+	csq.NotifyNewTask("test-cluster", &hcommon.NotifyTaskInfo{
+		Tasks: nil,
+	})
+}
+
+func TestCachedScheduledQueue_NotifyNewTask_WithTasks(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockReader := NewMockCachedQueueReader(ctrl)
+
+	tasks := []persistence.Task{
+		&persistence.DecisionTimeoutTask{
+			TaskData: persistence.TaskData{
+				VisibilityTimestamp: time.Now(),
+			},
+		},
+	}
+
+	mockReader.EXPECT().Inject(tasks).Times(1)
+
+	csq := &cachedScheduledQueue{
+		scheduledQueue: &scheduledQueue{
+			base: &queueBase{
+				metricsScope: metrics.NoopScope,
+			},
+			newTimerCh: make(chan struct{}, 1),
+		},
+		reader: mockReader,
+	}
+
+	csq.NotifyNewTask("test-cluster", &hcommon.NotifyTaskInfo{
+		Tasks: tasks,
+	})
+}
+
+// TestCachedScheduledQueue_StartStop verifies that Start and Stop delegate to both
+// the reader and the inner scheduledQueue.
+func TestCachedScheduledQueue_StartStop(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	ctrl := gomock.NewController(t)
+
+	mockShard := shard.NewTestContext(
+		t, ctrl,
+		&persistence.ShardInfo{ShardID: 10, RangeID: 1, TransferAckLevel: 0},
+		config.NewForTest(),
+	)
+
+	options := testScheduledQueueOptions()
+	mockReader := NewMockCachedQueueReader(ctrl)
+
+	// processEventLoop calls LookAHead after the timer gate fires, and GetTask
+	// when processing new tasks. Both can fire multiple times.
+	mockReader.EXPECT().LookAHead(gomock.Any(), gomock.Any()).Return(&LookAHeadResponse{}, nil).AnyTimes()
+	mockReader.EXPECT().GetTask(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, req *GetTaskRequest) (*GetTaskResponse, error) {
+			return &GetTaskResponse{
+				Progress: &GetTaskProgress{
+					Range:       req.Progress.Range,
+					NextTaskKey: req.Progress.ExclusiveMaxTaskKey,
+				},
+			}, nil
+		},
+	).AnyTimes()
+	mockReader.EXPECT().Start().Times(1)
+	mockReader.EXPECT().Stop().Times(1)
+
+	inner := newScheduledQueue(mockShard, persistence.HistoryTaskCategoryTimer,
+		task.NewMockProcessor(ctrl), task.NewMockExecutor(ctrl),
+		mockShard.GetLogger(), metrics.NoopClient, metrics.NoopScope,
+		options, mockReader)
+
+	q := newCachedScheduledQueue(inner, mockReader)
+
+	q.Start()
+	q.Stop()
+}
+
+// TestCachedScheduledQueue_StartStopDelegation verifies that cachedScheduledQueue.Start
+// and Stop delegate to the embedded reader.
+func TestCachedScheduledQueue_StartStopDelegation(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockReader := NewMockCachedQueueReader(ctrl)
+	mockReader.EXPECT().Start().Times(1)
+	mockReader.EXPECT().Stop().Times(1)
+
+	csq := &cachedScheduledQueue{
+		reader: mockReader,
+	}
+
+	csq.reader.Start()
+	csq.reader.Stop()
+}
+
+func TestCachedScheduledQueue_EvictionHookWired(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	ctrl := gomock.NewController(t)
+
+	mockShard := shard.NewTestContext(
+		t, ctrl,
+		&persistence.ShardInfo{ShardID: 10, RangeID: 1, TransferAckLevel: 0},
+		config.NewForTest(),
+	)
+
+	options := testScheduledQueueOptions()
+	mockReader := NewMockCachedQueueReader(ctrl)
+
+	inner := newScheduledQueue(mockShard, persistence.HistoryTaskCategoryTimer,
+		task.NewMockProcessor(ctrl), task.NewMockExecutor(ctrl),
+		mockShard.GetLogger(), metrics.NoopClient, metrics.NoopScope,
+		options, mockReader)
+
+	// Replace the original updateQueueStateFn with a no-op so the hook closure
+	// can be called without triggering real shard persistence.
+	inner.base.updateQueueStateFn = func(ctx context.Context) {}
+
+	q := newCachedScheduledQueue(inner, mockReader)
+	csq := q.(*cachedScheduledQueue)
+
+	// The eviction hook should be wired into updateQueueStateFn.
+	assert.NotNil(t, csq.scheduledQueue.base.updateQueueStateFn)
+
+	// Calling the hooked function exercises the closure (covers the inner body).
+	mockReader.EXPECT().UpdateReadLevel(gomock.Any()).Times(1)
+	csq.scheduledQueue.base.updateQueueStateFn(context.Background())
+}
+
+func testScheduledQueueOptions() *Options {
+	return &Options{
+		DeleteBatchSize:                      dynamicproperties.GetIntPropertyFn(100),
+		RedispatchInterval:                   dynamicproperties.GetDurationPropertyFn(10 * time.Second),
+		PageSize:                             dynamicproperties.GetIntPropertyFn(100),
+		PollBackoffInterval:                  dynamicproperties.GetDurationPropertyFn(10 * time.Second),
+		MaxPollInterval:                      dynamicproperties.GetDurationPropertyFn(10 * time.Second),
+		MaxPollIntervalJitterCoefficient:     dynamicproperties.GetFloatPropertyFn(0.1),
+		UpdateAckInterval:                    dynamicproperties.GetDurationPropertyFn(10 * time.Second),
+		UpdateAckIntervalJitterCoefficient:   dynamicproperties.GetFloatPropertyFn(0.1),
+		MaxPollRPS:                           dynamicproperties.GetIntPropertyFn(100),
+		MaxPendingTasksCount:                 dynamicproperties.GetIntPropertyFn(100),
+		PollBackoffIntervalJitterCoefficient: dynamicproperties.GetFloatPropertyFn(0.0),
+		VirtualSliceForceAppendInterval:      dynamicproperties.GetDurationPropertyFn(10 * time.Second),
+		CriticalPendingTaskCount:             dynamicproperties.GetIntPropertyFn(90),
+		EnablePendingTaskCountAlert:          func() bool { return true },
+		MaxVirtualQueueCount:                 dynamicproperties.GetIntPropertyFn(2),
+	}
+}

service/history/queuev2/timer_queue_factory.go

@@ -151,14 +151,19 @@ func (f *timerQueueFactory) createQueuev2(
 		MaxVirtualQueueCount:                 config.QueueMaxVirtualQueueCount,
 	}
 
+	var cachedReader CachedQueueReader
 	reader := NewQueueReader(
 		shard,
 		persistence.HistoryTaskCategoryTimer,
 		options.MaxPollInterval,
 		options.MaxPollIntervalJitterCoefficient,
 	)
+	if config.TimerProcessorEnableCachedScheduledQueue() {
+		cachedReader = newCachedQueueReader(reader, newInMemQueue(), shard, metricsScope)
+		reader = cachedReader
+	}
 
-	return newScheduledQueue(
+	base := newScheduledQueue(
 		shard,
 		persistence.HistoryTaskCategoryTimer,
 		f.taskProcessor,
@@ -169,4 +174,10 @@ func (f *timerQueueFactory) createQueuev2(
 		options,
 		reader,
 	)
+
+	if cachedReader != nil {
+		return newCachedScheduledQueue(base, cachedReader)
+	}
+
+	return base
 }

service/history/queuev2/timer_queue_factory_test.go

@@ -7,6 +7,7 @@ import (
 	"go.uber.org/goleak"
 	"go.uber.org/mock/gomock"
 
+	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/reconciliation/invariant"
 	"github.com/uber/cadence/service/history/config"
@@ -50,6 +51,34 @@ func TestTimerQueueFactory_Category(t *testing.T) {
 	assert.Equal(t, persistence.HistoryTaskCategoryTimer, category)
 }
 
+func TestTimerQueueFactory_CreateQueueV2_Cached(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	ctrl := gomock.NewController(t)
+
+	cfg := config.NewForTest()
+	cfg.TimerProcessorEnableCachedScheduledQueue = dynamicproperties.GetBoolPropertyFn(true)
+
+	mockShard := shard.NewTestContext(
+		t, ctrl, &persistence.ShardInfo{
+			ShardID:          10,
+			RangeID:          1,
+			TransferAckLevel: 0,
+		},
+		cfg)
+
+	factory := &timerQueueFactory{
+		taskProcessor:  task.NewMockProcessor(ctrl),
+		archivalClient: archiver.NewMockClient(ctrl),
+	}
+
+	processor := factory.createQueuev2(mockShard, execution.NewMockCache(ctrl), invariant.NewMockInvariant(ctrl))
+
+	assert.NotNil(t, processor)
+	assert.Implements(t, (*queue.Processor)(nil), processor)
+	_, ok := processor.(*cachedScheduledQueue)
+	assert.True(t, ok, "expected *cachedScheduledQueue when config is enabled")
+}
+
 func TestTimerQueueFactory_IsQueueV2Enabled(t *testing.T) {
 	defer goleak.VerifyNone(t)
 	ctrl := gomock.NewController(t)