Skip to content

Commit 5e7eb80

Browse files
plastikfanplastikfan
committed
feat(nav): add With to session runner (#333)
1 parent 360b292 commit 5e7eb80

7 files changed

+247
-19
lines changed

‎xfs/nav/helpers_test.go

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,25 @@ func universalCallback(name string, extended bool) nav.LabelledTraverseCallback
174174
}
175175
}
176176

177+
func universalCallbackNoAssert(name string, extended bool) nav.LabelledTraverseCallback {
178+
ex := lo.Ternary(extended, "-EX", "")
179+
180+
return nav.LabelledTraverseCallback{
181+
Label: "test universal callback",
182+
Fn: func(item *nav.TraverseItem) error {
183+
depth := lo.TernaryF(extended,
184+
func() int { return item.Extension.Depth },
185+
func() int { return 9999 },
186+
)
187+
GinkgoWriter.Printf(
188+
"---> 🌊 UNIVERSAL//%v-CALLBACK%v: (depth:%v) '%v'\n", name, ex, depth, item.Path,
189+
)
190+
191+
return nil
192+
},
193+
}
194+
}
195+
177196
func foldersCallback(name string, extended bool) nav.LabelledTraverseCallback {
178197
ex := lo.Ternary(extended, "-EX", "")
179198

‎xfs/nav/navigation-async_test.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919

2020
var (
2121
navigatorRoutineName = boost.GoRoutineName("✨ observable-navigator")
22+
outputChTimeout = time.Second
2223
)
2324

2425
type (
@@ -279,7 +280,7 @@ var _ = Describe("navigation", Ordered, func() {
279280
given: "Primary Session Consume",
280281
should: "enable output to be consumed externally",
281282
operator: func(op nav.AccelerationOperators) nav.AccelerationOperators {
282-
return op.NoW(4).Consume(outputCh)
283+
return op.NoW(4).Consume(outputCh, outputChTimeout)
283284
},
284285
},
285286
}, SpecTimeout(time.Second*2)),
@@ -290,7 +291,7 @@ var _ = Describe("navigation", Ordered, func() {
290291
should: "enable output to be consumed externally",
291292
operator: func(op nav.AccelerationOperators) nav.AccelerationOperators {
292293
outputCh = nav.CreateTraverseOutputCh(3)
293-
return op.NoW(4).Consume(outputCh)
294+
return op.NoW(4).Consume(outputCh, outputChTimeout)
294295
},
295296
// 🔥 panic: send on closed channel;
296297
//

‎xfs/nav/navigation-runner.go

Lines changed: 96 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,35 @@ package nav
33
import (
44
"fmt"
55
"runtime"
6+
"time"
67

78
"github.com/samber/lo"
89
"github.com/snivilised/lorax/boost"
910
)
1011

12+
type CreateNewRunnerWith int
13+
14+
const (
15+
RunnerDefault CreateNewRunnerWith = 0
16+
RunnerWithResume CreateNewRunnerWith = 1
17+
RunnerWithPool CreateNewRunnerWith = 2
18+
)
19+
20+
type Acceleration struct {
21+
WgAn boost.WaitGroupAn
22+
RoutineName boost.GoRoutineName
23+
NoW int
24+
JobsChOut TraverseItemJobStream
25+
JobResultsCh boost.JobOutputStream[TraverseOutput]
26+
OutputChTimeout time.Duration
27+
}
28+
29+
type RunnerInfo struct {
30+
ResumeInfo *Resumption
31+
PrimeInfo *Prime
32+
AccelerationInfo *Acceleration
33+
}
34+
1135
const (
1236
MinNoWorkers = 1
1337
MaxNoWorkers = 100
@@ -31,10 +55,11 @@ type Runnable interface {
3155
type AccelerationOperators interface {
3256
Runnable
3357
NoW(now int) AccelerationOperators
34-
Consume(outputCh boost.JobOutputStream[TraverseOutput]) AccelerationOperators
58+
Consume(outputCh boost.JobOutputStream[TraverseOutput], timeout time.Duration) AccelerationOperators
3559
}
3660

3761
type SessionRunner interface {
62+
With(with CreateNewRunnerWith, info *RunnerInfo) NavigationRunner
3863
Primary(info *Prime) NavigationRunner
3964
Resume(info *Resumption) NavigationRunner
4065
}
@@ -54,6 +79,71 @@ type runner struct {
5479
sync *acceleratedSync
5580
}
5681

82+
func (r *runner) With(with CreateNewRunnerWith, info *RunnerInfo) NavigationRunner {
83+
lo.TernaryF(with&RunnerWithResume == 0,
84+
func() NavigationRunner {
85+
return r.Primary(&Prime{
86+
Path: info.PrimeInfo.Path,
87+
OptionsFn: info.PrimeInfo.OptionsFn,
88+
})
89+
},
90+
func() NavigationRunner {
91+
return r.Resume(&Resumption{
92+
RestorePath: info.ResumeInfo.RestorePath,
93+
Restorer: info.ResumeInfo.Restorer,
94+
Strategy: info.ResumeInfo.Strategy,
95+
})
96+
},
97+
)
98+
99+
lo.TernaryF(with&RunnerWithPool == 0,
100+
func() AccelerationOperators {
101+
return r
102+
},
103+
func() AccelerationOperators {
104+
if info.AccelerationInfo == nil {
105+
// As this is not a user facing issue (ie programming error),
106+
// it does not have to be i18n error
107+
//
108+
panic("internal: acceleration info missing from runner info")
109+
}
110+
111+
if info.AccelerationInfo.JobsChOut == nil {
112+
panic("internal: job channel not set on acceleration info")
113+
}
114+
115+
return r.WithPool(&AsyncInfo{
116+
NavigatorRoutineName: info.AccelerationInfo.RoutineName,
117+
WaitAQ: info.AccelerationInfo.WgAn,
118+
JobsChanOut: info.AccelerationInfo.JobsChOut,
119+
})
120+
},
121+
)
122+
123+
if info.AccelerationInfo != nil && with&RunnerWithPool > 0 {
124+
if info.AccelerationInfo.JobResultsCh != nil {
125+
r.Consume(info.AccelerationInfo.JobResultsCh, info.AccelerationInfo.OutputChTimeout)
126+
}
127+
128+
if info.AccelerationInfo.NoW > 0 {
129+
r.NoW(info.AccelerationInfo.NoW)
130+
}
131+
}
132+
133+
return r
134+
}
135+
136+
func IfWithPoolUseContext(with CreateNewRunnerWith, args ...any) []any {
137+
return lo.TernaryF(with&RunnerWithPool > 0,
138+
func() []any {
139+
return args
140+
},
141+
func() []any {
142+
return []any{}
143+
},
144+
)
145+
}
146+
57147
func (r *runner) Primary(info *Prime) NavigationRunner {
58148
r.session = &Primary{
59149
Path: info.Path,
@@ -99,8 +189,12 @@ func (r *runner) NoW(now int) AccelerationOperators {
99189
return r
100190
}
101191

102-
func (r *runner) Consume(outputCh boost.JobOutputStream[TraverseOutput]) AccelerationOperators {
192+
func (r *runner) Consume(
193+
outputCh boost.JobOutputStream[TraverseOutput],
194+
timeout time.Duration,
195+
) AccelerationOperators {
103196
r.sync.outputChOut = outputCh
197+
r.sync.outputChTimeout = timeout
104198

105199
return r
106200
}

‎xfs/nav/navigation-sync.go

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"fmt"
66
"reflect"
7+
"time"
78

89
"github.com/snivilised/lorax/boost"
910
)
@@ -59,10 +60,11 @@ func (s *inlineSync) Run(callback sessionCallback, _ syncable, _ ...any) (*Trave
5960

6061
type acceleratedSync struct {
6162
baseSync
62-
ai *AsyncInfo
63-
noWorkers int
64-
outputChOut boost.JobOutputStream[TraverseOutput]
65-
pool *boost.WorkerPool[TraverseItemInput, TraverseOutput]
63+
ai *AsyncInfo
64+
noWorkers int
65+
outputChOut boost.JobOutputStream[TraverseOutput]
66+
outputChTimeout time.Duration
67+
pool *boost.WorkerPool[TraverseItemInput, TraverseOutput]
6668
}
6769

6870
func (s *acceleratedSync) Run(callback sessionCallback, nc syncable, args ...any) (*TraverseResult, error) {
@@ -85,10 +87,11 @@ func (s *acceleratedSync) Run(callback sessionCallback, nc syncable, args ...any
8587
func (s *acceleratedSync) start(ctx context.Context, cancel context.CancelFunc) {
8688
s.pool = boost.NewWorkerPool[TraverseItemInput, TraverseOutput](
8789
&boost.NewWorkerPoolParams[TraverseItemInput, TraverseOutput]{
88-
NoWorkers: s.noWorkers,
89-
Exec: workerExecutive,
90-
JobsCh: s.ai.JobsChanOut,
91-
WaitAQ: s.ai.WaitAQ,
90+
NoWorkers: s.noWorkers,
91+
OutputChTimeout: s.outputChTimeout,
92+
Exec: workerExecutive,
93+
JobsCh: s.ai.JobsChanOut,
94+
WaitAQ: s.ai.WaitAQ,
9295
})
9396

9497
// We are handing over ownership of this channel (ai.OutputsChIn) to the pool as

‎xfs/nav/navigation-with-runner_test.go

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
package nav_test
2+
3+
import (
4+
"context"
5+
"time"
6+
7+
. "github.com/onsi/ginkgo/v2"
8+
. "github.com/onsi/gomega"
9+
10+
"github.com/snivilised/extendio/internal/helpers"
11+
"github.com/snivilised/extendio/xfs/nav"
12+
"github.com/snivilised/lorax/boost"
13+
14+
. "github.com/snivilised/extendio/i18n"
15+
)
16+
17+
var _ = Describe("NavigationWithRunner", Ordered, func() {
18+
var (
19+
root string
20+
jroot string
21+
fromJSONPath string
22+
)
23+
24+
BeforeAll(func() {
25+
root = musico()
26+
jroot = helpers.JoinCwd("Test", "json")
27+
fromJSONPath = helpers.Path(jroot, "resume-state.json")
28+
})
29+
30+
BeforeEach(func() {
31+
if err := Use(func(o *UseOptions) {
32+
o.Tag = DefaultLanguage.Get()
33+
}); err != nil {
34+
Fail(err.Error())
35+
}
36+
})
37+
38+
Context("resume and worker pool acceleration", func() {
39+
When("universal: listen pending(logged)", func() {
40+
It("should: ...", SpecTimeout(time.Second*5), func(ctxSpec SpecContext) {
41+
ctx, cancel := context.WithCancel(ctxSpec)
42+
path := helpers.Path(root, "RETRO-WAVE")
43+
restorer := func(o *nav.TraverseOptions, active *nav.ActiveState) {
44+
// synthetic assignments
45+
//
46+
active.Root = helpers.Path(root, "RETRO-WAVE")
47+
active.NodePath = helpers.Path(root, ResumeAtTeenageColor)
48+
active.Listen = nav.ListenPending
49+
o.Store.Subscription = nav.SubscribeAny
50+
//
51+
// end of synthetic assignments
52+
o.Callback = universalCallbackNoAssert(
53+
"universal: listen pending(Resume-WithRunner)",
54+
NotExtended,
55+
)
56+
}
57+
58+
wgan := boost.NewAnnotatedWaitGroup("🍂 traversal")
59+
wgan.Add(1, navigatorRoutineName)
60+
createWith := nav.RunnerWithResume | nav.RunnerWithPool
61+
now := 3
62+
JobsChOut := make(boost.JobStream[nav.TraverseItemInput], DefaultJobsChSize)
63+
outputChTimeout := time.Second
64+
jobsOutputChOut := make(boost.JobOutputStream[nav.TraverseOutput], DefaultJobsChSize)
65+
66+
result, err := nav.New().With(createWith, &nav.RunnerInfo{
67+
PrimeInfo: &nav.Prime{
68+
Path: path,
69+
OptionsFn: func(o *nav.TraverseOptions) {
70+
o.Notify.OnBegin = begin("🛡️")
71+
o.Store.Subscription = nav.SubscribeAny
72+
o.Callback = universalCallbackNoAssert(
73+
"universal: Path contains folders(Prime-WithRunner)",
74+
NotExtended,
75+
)
76+
},
77+
},
78+
ResumeInfo: &nav.Resumption{
79+
RestorePath: fromJSONPath,
80+
Restorer: restorer,
81+
Strategy: nav.ResumeStrategySpawnEn,
82+
},
83+
AccelerationInfo: &nav.Acceleration{
84+
WgAn: wgan,
85+
RoutineName: navigatorRoutineName,
86+
NoW: now,
87+
JobsChOut: JobsChOut,
88+
JobResultsCh: jobsOutputChOut,
89+
OutputChTimeout: outputChTimeout,
90+
},
91+
}).Run(
92+
nav.IfWithPoolUseContext(createWith, ctx, cancel)...,
93+
)
94+
95+
if createWith&nav.RunnerWithPool > 0 {
96+
wgan.Wait("👾 test-main")
97+
}
98+
99+
Expect(err).Error().To(BeNil())
100+
_ = result.Session.StartedAt()
101+
_ = result.Session.Elapsed()
102+
})
103+
})
104+
})
105+
})

‎xfs/nav/resume-strategy_test.go

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -281,10 +281,12 @@ var _ = Describe("Resume", Ordered, func() {
281281
}
282282
}
283283

284-
result, _ := nav.New().Resume(&nav.Resumption{
285-
RestorePath: fromJSONPath,
286-
Restorer: restorer,
287-
Strategy: strategyEn,
284+
result, _ := nav.New().With(nav.RunnerWithResume, &nav.RunnerInfo{
285+
ResumeInfo: &nav.Resumption{
286+
RestorePath: fromJSONPath,
287+
Restorer: restorer,
288+
Strategy: strategyEn,
289+
},
288290
}).Run()
289291

290292
if profile.mandatory != nil {

‎xfs/nav/traverse-navigator_test.go

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -314,9 +314,13 @@ var _ = Describe("TraverseNavigator(logged)", Ordered, func() {
314314
o.Store.DoExtend = entry.extended
315315
o.Callback = once
316316
}
317-
result, _ := nav.New().Primary(&nav.Prime{
318-
Path: path,
319-
OptionsFn: optionFn,
317+
318+
createWith := nav.RunnerDefault
319+
result, _ := nav.New().With(createWith, &nav.RunnerInfo{
320+
PrimeInfo: &nav.Prime{
321+
Path: path,
322+
OptionsFn: optionFn,
323+
},
320324
}).Run()
321325

322326
if entry.visit {

0 commit comments

Comments
 (0)