Add lightweight subqueues for priority #7372

dnr · 2025-02-21T02:06:40Z

Compare to #7371

What changed?

Add subqueues for priority using a new backlog manager with less overhead than a full queue:
- metadata (ack levels and approximate count) is shared in one database row, minimizing transactions
- task writer can write to multiple subqueues at once
- there's one task reader, ack manager, and task gc per subqueue
This forks backlog_manager.go and task_writer.go. You can check it out and look at the diff between the old and new files for reviewing.

Why?

Support simple priority and later fairness.

How did you test it?

new functional test, unit tests are broken, will be fixed in follow-up PR

dnr · 2025-02-28T20:07:57Z

service/matching/task.go

@@ -40,7 +40,8 @@ type (
 	// genericTaskInfo contains the info for an activity or workflow task
 	genericTaskInfo struct {
 		*persistencespb.AllocatedTaskInfo
-		completionFunc func(*internalTask, taskResponse)
+		completionFunc  func(*internalTask, taskResponse)
+		backlogSubqueue int


This is going to go away in a later PR, actually

dnr · 2025-02-28T20:08:21Z

service/matching/task_gc.go

+	tqCtx    context.Context
+	db       *taskQueueDB
+	config   *taskQueueConfig
+	subqueue int


This is going to go away in a later PR (that merges taskGC into task reader)

dnr · 2025-02-28T20:13:38Z

common/persistence/sql/task.go

 	idBytes = append(idBytes, namespaceID...)
 	idBytes = append(idBytes, []byte(name)...)
 	idBytes = append(idBytes, uint8(taskType))
+	if subqueue > 0 {
+		// FIXME: this should be done so that it's invertible


I'll pull this fix into this PR since it's small

stephanos · 2025-02-27T18:12:58Z

common/persistence/cassandra/matching_task_store.go

+// We steal some upper bits of the "row type" field to hold a subqueue index.
+// Subqueue 0 must be the same as rowTypeTask (before subqueues were introduced).
+// 00000000: task in subqueue 0
+// 00000001: task queue metadata
+// xxxxxx1x: reserved
+// 00000100: task in subqueue 1
+// nnnnnn00: task in subqueue n, etc.


Suggested change

// We steal some upper bits of the "row type" field to hold a subqueue index.

// Subqueue 0 must be the same as rowTypeTask (before subqueues were introduced).

// 00000000: task in subqueue 0

// 00000001: task queue metadata

// xxxxxx1x: reserved

// 00000100: task in subqueue 1

// nnnnnn00: task in subqueue n, etc.

// We steal some upper bits of the "row type" field to hold a subqueue index.

// Subqueue 0 must be the same as rowTypeTask (before subqueues were introduced).

// 00000000: task in subqueue 0 (`rowTypeTask`)

// 00000001: task queue metadata (`rowTypeTaskQueue`)

// xxxxxx1x: reserved

// 00000100: task in subqueue 1

// nnnnnn00: task in subqueue n, etc.

stephanos · 2025-02-27T22:47:29Z

common/persistence/sql/task.go

 ) ([]byte, uint32) {
-	id := m.taskQueueId(namespaceID, name, taskType)
+	id := m.taskQueueId(namespaceID, name, taskType, subqueue)


Could we rename name to taskqueueName and truncate the comment as the parameter types would be sufficient?

stephanos · 2025-02-28T00:36:31Z

proto/internal/temporal/server/api/persistence/v1/tasks.proto

@@ -65,6 +65,35 @@ message TaskQueueInfo {
    google.protobuf.Timestamp expiry_time = 6;
    google.protobuf.Timestamp last_update_time = 7;
    int64 approximate_backlog_count = 8;


Could we add a note here and for ack_level, too, that mentions it's a copy of the subqueue 0's data? It's mentioned on subqueues, but worth mentioning twice, I think.

stephanos · 2025-02-28T00:51:24Z

service/matching/task.go

+func (task *internalTask) getPriority() *commonpb.Priority {
+	if task.event != nil {
+		return task.event.AllocatedTaskInfo.GetData().GetPriority()
+	} else if task.query != nil {
+		return task.query.request.GetPriority()
+	}
+	return nil
+}
+


nit: how about dropping the else?

Suggested change

func (task *internalTask) getPriority() *commonpb.Priority {

if task.event != nil {

return task.event.AllocatedTaskInfo.GetData().GetPriority()

} else if task.query != nil {

return task.query.request.GetPriority()

}

return nil

}

func (task *internalTask) getPriority() *commonpb.Priority {

if task.event != nil {

return task.event.AllocatedTaskInfo.GetData().GetPriority()

}

if task.query != nil {

return task.query.request.GetPriority()

}

return nil

}

EDIT: After reading more code, I see the same pattern more often and am unlikely to convince you otherwise ;) But I'll leave my original one here.

stephanos · 2025-02-28T00:52:57Z

service/matching/task.go

@@ -263,6 +266,15 @@ func (task *internalTask) getResponse() (taskResponse, bool) {
 	return <-task.responseC, true
 }

+func (task *internalTask) getPriority() *commonpb.Priority {
+	if task.event != nil {
+		return task.event.AllocatedTaskInfo.GetData().GetPriority()


side note about existing code:
event is quite a surprising/confusing naming choice.

stephanos · 2025-02-28T20:33:27Z

service/matching/pri_backlog_manager.go

+
+	c.loadSubqueuesLocked(subqueues)
+
+	// this should be here now


Is this a TODO comment?

stephanos · 2025-02-28T21:16:48Z

service/matching/db.go

+	hasDefault := false
+	for _, s := range subqueues {
+		if hasDefault = proto.Equal(s.Key, defKey); hasDefault {
+			break
+		}
+	}


Could consider this now:

Suggested change

hasDefault := false

for _, s := range subqueues {

if hasDefault = proto.Equal(s.Key, defKey); hasDefault {

break

}

}

hasDefault := slices.ContainsFunc(subqueues, func(s *dbSubqueue) bool {

return proto.Equal(s.Key, defKey)

})

stephanos · 2025-02-28T21:27:44Z

service/matching/db.go

+	// start ack level + max read level just before the current block
+	initAckLevel := rangeIDToTaskIDBlock(db.rangeID, db.config.RangeSize).start - 1


Could we make this a function on taskQueueDB? It exists twice and the two lines are the same AFAICT but have/don't have comments and use different variable names.

stephanos · 2025-02-28T22:21:11Z

service/matching/task.go

+		return task.event.AllocatedTaskInfo.GetData().GetPriority()
+	} else if task.query != nil {
+		return task.query.request.GetPriority()
+	}


Could we add a comment mentioning Nexus?

stephanos · 2025-02-28T22:38:38Z

service/matching/db.go

-	return taskQueueState{rangeID: db.rangeID, ackLevel: db.ackLevel}, nil
+	return taskQueueState{
+		rangeID:   db.rangeID,
+		ackLevel:  db.subqueues[0].AckLevel, // TODO(pri): cleanup, only used by old backlog manager


As discussed, it would be nice to have the db.subqueues[0] meaning spelled out more explicitly for the "legacy" context.

Add lightweight subqueues for priority

36c85f5

dnr mentioned this pull request Feb 21, 2025

Add heavy subqueues for priority #7371

Closed

dnr marked this pull request as ready for review February 28, 2025 19:06

dnr requested a review from a team as a code owner February 28, 2025 19:06

improve sql tqid generation

0ae78d9

dnr commented Feb 28, 2025

View reviewed changes

stephanos reviewed Feb 28, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add lightweight subqueues for priority #7372

Add lightweight subqueues for priority #7372

dnr commented Feb 21, 2025 •

edited

Loading

dnr Feb 28, 2025

dnr Feb 28, 2025

dnr Feb 28, 2025

stephanos Feb 27, 2025

stephanos Feb 27, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

stephanos Feb 28, 2025

		// start ack level + max read level just before the current block
		initAckLevel := rangeIDToTaskIDBlock(db.rangeID, db.config.RangeSize).start - 1

Add lightweight subqueues for priority #7372

Are you sure you want to change the base?

Add lightweight subqueues for priority #7372

Conversation

dnr commented Feb 21, 2025 • edited Loading

What changed?

Why?

How did you test it?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dnr commented Feb 21, 2025 •

edited

Loading