diff --git a/server/automation/service/session_test.go b/server/automation/service/session_test.go index 1fe67da2f..c684ff044 100644 --- a/server/automation/service/session_test.go +++ b/server/automation/service/session_test.go @@ -40,5 +40,152 @@ func TestSession_Start(t *testing.T) { g.AddStep(wfexec.NewGenericStep(nil), s) _, _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 42, Invoker: auth.Anonymous()}) req.EqualError(err, "cannot start workflow on a step with parents") - +} + +func TestSessionStackTraces(t *testing.T) { + stepID := uint64(0) + getStepID := func() uint64 { + stepID++ + return uint64(stepID) + } + + iter1 := &wfexec.Frame{StepID: getStepID(), ParentID: 0, Action: "iterator initialized"} + iter1stepExpr := &wfexec.Frame{StepID: getStepID(), ParentID: iter1.StepID, Action: ""} + iter1stepCnt := &wfexec.Frame{StepID: getStepID(), ParentID: iter1stepExpr.StepID, Action: "loop continue"} + + iter2 := &wfexec.Frame{StepID: getStepID(), ParentID: iter1.StepID, Action: "iterator initialized"} + iter2stepExpr := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""} + iter2stepCnt := &wfexec.Frame{StepID: getStepID(), ParentID: iter2stepExpr.StepID, Action: "loop continue"} + + rando1 := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""} + rando2 := &wfexec.Frame{StepID: getStepID(), ParentID: iter2.StepID, Action: ""} + + trace := []*wfexec.Frame{ + // first loop + iter1, + iter1stepExpr, + iter1stepCnt, + iter1, + iter1stepExpr, + iter1stepCnt, + iter1, + + // second loop + iter2, + iter2stepExpr, + iter2stepCnt, + iter2, + iter2stepExpr, + iter2stepCnt, + iter2, + + rando1, + rando2, + } + + ses := &types.Session{} + + for _, f := range trace { + ses.AppendRuntimeStacktrace(f) + } + + require.Len(t, ses.RuntimeStacktrace, 10) + + require.Equal(t, iter1.StepID, ses.RuntimeStacktrace[0].StepID) + require.Equal(t, iter1stepExpr.StepID, ses.RuntimeStacktrace[1].StepID) + require.Equal(t, iter1stepCnt.StepID, ses.RuntimeStacktrace[2].StepID) + require.Equal(t, iter1.StepID, ses.RuntimeStacktrace[3].StepID) + require.Equal(t, iter2.StepID, ses.RuntimeStacktrace[4].StepID) + require.Equal(t, iter2stepExpr.StepID, ses.RuntimeStacktrace[5].StepID) + require.Equal(t, iter2stepCnt.StepID, ses.RuntimeStacktrace[6].StepID) + require.Equal(t, iter2.StepID, ses.RuntimeStacktrace[7].StepID) + require.Equal(t, rando1.StepID, ses.RuntimeStacktrace[8].StepID) + require.Equal(t, rando2.StepID, ses.RuntimeStacktrace[9].StepID) +} + +// Before: +// goos: darwin +// goarch: arm64 +// pkg: github.com/cortezaproject/corteza/server/automation/service +// BenchmarkSessionStackTraces_1000-12 42758 28594 ns/op 87288 B/op 15 allocs/op +// BenchmarkSessionStackTraces_10000-12 3430 348308 ns/op 1160442 B/op 23 allocs/op +// BenchmarkSessionStackTraces_100000-12 229 5299188 ns/op 13317368 B/op 33 allocs/op +// BenchmarkSessionStackTraces_1000000-12 19 52896750 ns/op 128431352 B/op 43 allocs/op +// BenchmarkSessionStackTraces_10000000-12 3 436311292 ns/op 1202337018 B/op 53 allocs/op + +// After: +// goos: darwin +// goarch: arm64 +// pkg: github.com/cortezaproject/corteza/server/automation/service +// BenchmarkSessionStackTraces_1000-12 40142 29860 ns/op 120 B/op 4 allocs/op +// BenchmarkSessionStackTraces_10000-12 4387 264930 ns/op 120 B/op 4 allocs/op +// BenchmarkSessionStackTraces_100000-12 447 2735284 ns/op 120 B/op 4 allocs/op +// BenchmarkSessionStackTraces_1000000-12 42 26596636 ns/op 120 B/op 4 allocs/op +// BenchmarkSessionStackTraces_10000000-12 4 265553667 ns/op 120 B/op 4 allocs/op +func benchmarkSessionStackTraces(b *testing.B, iters int) { + stepID := uint64(0) + getStepID := func() uint64 { + stepID++ + return uint64(stepID) + } + + iter := &wfexec.Frame{ + StepID: getStepID(), + Action: "iterator initialized", + } + + loop := []*wfexec.Frame{iter} + loop = append(loop, &wfexec.Frame{ + StepID: getStepID(), + ParentID: iter.ParentID, + Action: "", + }) + loop = append(loop, &wfexec.Frame{ + StepID: getStepID(), + ParentID: loop[len(loop)-1].StepID, + Action: "loop continue", + }) + + frames := []*wfexec.Frame{} + for x := 0; x < iters; x++ { + frames = append(frames, loop...) + } + + frames = append(frames, iter) + frames = append(frames, &wfexec.Frame{ + StepID: getStepID(), + ParentID: iter.ParentID, + Action: "termination", + }) + + b.StopTimer() + b.ResetTimer() + for n := 0; n < b.N; n++ { + ses := &types.Session{} + b.StartTimer() + for _, f := range frames { + ses.AppendRuntimeStacktrace(f) + } + b.StopTimer() + } +} + +func BenchmarkSessionStackTraces_1000(b *testing.B) { + benchmarkSessionStackTraces(b, 1000) +} + +func BenchmarkSessionStackTraces_10000(b *testing.B) { + benchmarkSessionStackTraces(b, 10000) +} + +func BenchmarkSessionStackTraces_100000(b *testing.B) { + benchmarkSessionStackTraces(b, 100000) +} + +func BenchmarkSessionStackTraces_1000000(b *testing.B) { + benchmarkSessionStackTraces(b, 1000000) +} + +func BenchmarkSessionStackTraces_10000000(b *testing.B) { + benchmarkSessionStackTraces(b, 10000000) } diff --git a/server/automation/types/session.go b/server/automation/types/session.go index b97b8599b..50f0d4f69 100644 --- a/server/automation/types/session.go +++ b/server/automation/types/session.go @@ -184,13 +184,55 @@ func (s *Session) Apply(ssp SessionStartParams) { } } +// AppendRuntimeStacktrace adds a new frame to the runtime stacktrace +// +// This does have some smartness +// If the workflow uses longer iterators, the memory pressure got too high. +// To counter this, only the frames of the last iteration are preserved to +// better match what programming languages do. func (s *Session) AppendRuntimeStacktrace(frame *wfexec.Frame) { - if !s.runtimeOpts.disableStacktrace { - s.l.RLock() - defer s.l.RUnlock() - - s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame) + if s.runtimeOpts.disableStacktrace { + return } + + s.l.RLock() + defer s.l.RUnlock() + + // The only way to get to the same stepID is when we're in a loop + // Find where the first frame of the iterator is and slice the stack. + if frame.Action != "iterator initialized" && s.hasDuplicate(frame.StepID) { + var ( + i = 0 + f *wfexec.Frame + ) + for i, f = range s.RuntimeStacktrace { + if f.StepID == frame.StepID { + break + } + } + + // @todo this might cause a memory leak; investigate further + s.RuntimeStacktrace = s.RuntimeStacktrace[0:i] + } + + // Push to the newly done trace + s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame) +} + +// @todo potentially optimize this; it'll probably be fine for now but +// might degrade performance for larger workflows. +// To investigate and potentially optimize it. +func (s *Session) hasDuplicate(stepID uint64) bool { + s.l.RLock() + defer s.l.RUnlock() + + for _, f := range s.RuntimeStacktrace { + if f.StepID == stepID { + return true + } + } + + return false } func (s *Session) CopyRuntimeStacktrace() {