Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Deliver messages even when events are skipped #1015

Merged
merged 2 commits into from
Jan 30, 2023
Merged

Deliver messages even when events are skipped #1015

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bdd5878
Deliver messages even when events are skipped
Jan 28, 2023
23b0854
Avoid sorting messages to index
Jan 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions internal/internal_message.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 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 internal

import (
"sort"

protocolpb "go.temporal.io/api/protocol/v1"
)

type eventMsgIndex []*protocolpb.Message

// indexMessagesByEventID creates an index over a set of input messages that allows for
// fast access to messages with an event ID less than or equal to a specific
// upper bound. The order of messages with the same event ID will be preserved.
func indexMessagesByEventID(msgs []*protocolpb.Message) *eventMsgIndex {
// implementor note: the order preservation requirement is why we can't use
// the heap package from the Go SDK here.

sorted := make(eventMsgIndex, len(msgs))
copy(sorted, msgs)
sort.SliceStable(sorted, func(i, j int) bool {
return sorted[i].GetEventId() < sorted[j].GetEventId()
})
return &sorted
}

// takeLTE removes and returns the messages in this index that have an event ID
// less than or equal to the input argument.
func (emi *eventMsgIndex) takeLTE(eventID int64) []*protocolpb.Message {
indexOfFirstGreater := len(*emi)
for i, msg := range *emi {
if msg.GetEventId() > eventID {
indexOfFirstGreater = i
break
}
}
var out []*protocolpb.Message
out, *emi = (*emi)[0:indexOfFirstGreater], (*emi)[indexOfFirstGreater:]
return out
}
81 changes: 81 additions & 0 deletions internal/internal_message_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 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 internal

import (
"testing"

"github.com/stretchr/testify/require"
protocolpb "go.temporal.io/api/protocol/v1"
)

func TestEventMessageIndex(t *testing.T) {
newMsg := func(id string, eventID int64) *protocolpb.Message {
return &protocolpb.Message{
Id: id,
SequencingId: &protocolpb.Message_EventId{EventId: eventID},
}
}
messages := []*protocolpb.Message{
newMsg("00", 0),
newMsg("01", 2),
newMsg("02", 2),
newMsg("03", 101),
newMsg("04", 100),
newMsg("05", 50),
newMsg("06", 3),
newMsg("07", 0),
newMsg("08", 100),
}

emi := indexMessagesByEventID(messages)

batch := emi.takeLTE(2)
require.Len(t, batch, 4)
for i := 0; i < len(batch)-1; i++ {
if batch[i].GetEventId() == batch[i+1].GetEventId() {
require.Less(t, batch[i].Id, batch[i+1].Id)
}
}

batch = emi.takeLTE(2)
require.Empty(t, batch)

batch = emi.takeLTE(3)
require.Len(t, batch, 1)

batch = emi.takeLTE(100)
require.Len(t, batch, 3)
for i := 0; i < len(batch)-1; i++ {
if batch[i].GetEventId() == batch[i+1].GetEventId() {
require.Less(t, batch[i].Id, batch[i+1].Id)
}
}

emi = indexMessagesByEventID(messages)
batch = emi.takeLTE(9000)
require.Len(t, batch, len(messages))
require.Empty(t, emi)
}
30 changes: 15 additions & 15 deletions internal/internal_task_handlers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -815,19 +815,6 @@ processWorkflowLoop:
return
}

// indexInvocations builds a map of the interaction invocations contained in a
// workflow task where the index is the interaction's event level (i.e. the
// event after which the interaction can execute) and the value is the
// invocation itself. This function may return an empty map but it will not
// return nil.
func indexMessages(workflowTask *workflowTask) map[int64][]*protocolpb.Message {
out := map[int64][]*protocolpb.Message{}
for _, msg := range workflowTask.task.Messages {
out[msg.GetEventId()] = append(out[msg.GetEventId()], msg)
}
return out
}

func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflowTask) (interface{}, error) {
task := workflowTask.task
historyIterator := workflowTask.historyIterator
Expand All @@ -841,7 +828,7 @@ func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflo
var replayCommands []*commandpb.Command
var respondEvents []*historypb.HistoryEvent

msgs := indexMessages(workflowTask)
msgs := indexMessagesByEventID(workflowTask.task.GetMessages())
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it have value to ask the server to always deliver these in event ID order?

Also, is the number of these so great that you really need to sort because of fear of the cost of takeLTE? I think you can just make takeLTE iterate over all messages every time and https://github.com/golang/go/wiki/SliceTricks#filter-in-place to keep the ones greater than while extracting the ones less than or equal.

So something like this (untested, just typed here in comments):

func (emi *eventMsgIndex) takeLTE(eventID int64) (out []*protocolpb.Message) {
	n := 0
	for _, msg := range *emi {
		if msg.GetEventID() > eventID {
			(*emi)[n] = msg
			n++
		} else {
			out = append(out, msg)
		}
	}
	*emi = *emi[:n]
	return;
}

Copy link
Contributor Author

@mmcshane mmcshane Jan 30, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm fine with doing a little more work in takeLTE. I would expect the number of messages per WFT to be low (which to be fair, also makes sorting cheap) but there may be a substantial number when we rebundle into a bigger WFT for replay. I'd like to preserve the transport's ability to reorder messages until we learn more about what is desirable at that layer.


skipReplayCheck := w.skipReplayCheck()
shouldForceReplayCheck := func() bool {
Expand Down Expand Up @@ -912,12 +899,25 @@ ProcessEvents:
return nil, err
}

// because we don't run all events through this code path, we have
// to run ProcessMessages both before and after ProcessEvent to
// catch any messages that should have been delivered _before_ this
// event but perhaps were not because there were attached to an
// event (e.g. WFTScheduledEvent) that does not come through this
// loop.
for _, msg := range msgs.takeLTE(event.GetEventId() - 1) {
err := eventHandler.ProcessMessage(msg, isInReplay, isLast)
if err != nil {
return nil, err
}
}

err = eventHandler.ProcessEvent(event, isInReplay, isLast)
if err != nil {
return nil, err
}

for _, msg := range msgs[event.GetEventId()] {
for _, msg := range msgs.takeLTE(event.GetEventId()) {
err := eventHandler.ProcessMessage(msg, isInReplay, isLast)
if err != nil {
return nil, err
Expand Down
35 changes: 0 additions & 35 deletions internal/internal_task_handlers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@ import (
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
historypb "go.temporal.io/api/history/v1"
protocolpb "go.temporal.io/api/protocol/v1"
querypb "go.temporal.io/api/query/v1"
"go.temporal.io/api/serviceerror"
taskqueuepb "go.temporal.io/api/taskqueue/v1"
Expand Down Expand Up @@ -1892,40 +1891,6 @@ func Test_IsMemoMatched(t *testing.T) {
}
}

func TestMessageIndexing(t *testing.T) {
wft := &workflowTask{
task: &workflowservice.PollWorkflowTaskQueueResponse{
Messages: []*protocolpb.Message{
{
Id: "ID.1",
SequencingId: &protocolpb.Message_EventId{EventId: 3},
},
{
Id: "ID.2",
SequencingId: &protocolpb.Message_EventId{EventId: 5},
},
{
Id: "ID.3",
SequencingId: &protocolpb.Message_EventId{EventId: 3},
},
},
},
}
index := indexMessages(wft)

event3Interactions := index[3]
event4Interactions := index[4]
event5Interactions := index[5]

require.Len(t, event3Interactions, 2)
require.Len(t, event4Interactions, 0)
require.Len(t, event5Interactions, 1)

require.Equal(t, event3Interactions[0].Id, "ID.1")
require.Equal(t, event3Interactions[1].Id, "ID.3")
require.Equal(t, event5Interactions[0].Id, "ID.2")
}

func TestHeartbeatThrottleInterval(t *testing.T) {
assertInterval := func(timeoutSec, defaultIntervalSec, maxIntervalSec, expectedSec int) {
a := &activityTaskHandlerImpl{
Expand Down