5951900512

Committed 23 Aug 2023 01:19PM UTC coverage: 90.747% (-0.4%) from 91.118%

Build # 5951900512

Build Type

push

github

Committed by

plastikfan

Commit Message

feat(nav): use context for concurrent navigation (#292)

Run Details

109 of 109 new or added lines in 8 files covered. (100.0%)

2295 of 2529 relevant lines covered (90.75%)

4.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.93

/xfs/nav/navigation-session.go

package nav

import (
        "fmt"
        "runtime"
        "time"

        xi18n "github.com/snivilised/extendio/i18n"
)

const (
        MinNoWorkers = 1
        MaxNoWorkers = 100
)

type TraverseSession interface {
        Init() NavigationRunner
        Run(ai ...*AsyncInfo) (*TraverseResult, error)
        StartedAt() time.Time
        Elapsed() time.Duration
}

type NavigationRunner interface {
        Run(ai ...*AsyncInfo) (*TraverseResult, error)
        WithCPUPool() NavigationRunner
        WithPool(now int) NavigationRunner
}

type sessionRunner struct {
        session     TraverseSession
        accelerator navigationAccelerator
}

// Run executes the traversal session
func (r *sessionRunner) Run(ai ...*AsyncInfo) (*TraverseResult, error) {
        if r.accelerator.noWorkers > 0 && len(ai) > 0 {
                r.accelerator.start(ai[0])
                return r.session.Run(ai[0])
        }

        return r.session.Run()
}

func (r *sessionRunner) WithPool(now int) NavigationRunner {
        if now >= MinNoWorkers && now <= MaxNoWorkers {
                r.accelerator.noWorkers = now
        }

        return r
}

func (r *sessionRunner) WithCPUPool() NavigationRunner {
        r.accelerator.noWorkers = runtime.NumCPU()
        return r
}

type primaryRunner struct {
        sessionRunner
}

type resumeRunner struct {
        sessionRunner
}

type session struct {
        startAt  time.Time
        duration time.Duration
}

func (s *session) start() {
        s.startAt = time.Now()
}

func (s *session) finish(_ *TraverseResult, _ error, ai ...*AsyncInfo) {
        defer func() {
                if len(ai) > 0 {
                        fmt.Printf("---> 😈😈😈 defer session.finish\n")
                        close(ai[0].JobsChanOut)
                }
        }()

        s.duration = time.Since(s.startAt)
}

// PrimarySession
type PrimarySession struct {
        session
        Path      string
        OptionFn  TraverseOptionFn
        navigator TraverseNavigator
}

func (s *PrimarySession) Init() NavigationRunner {
        s.navigator = navigatorFactory{}.new(s.OptionFn)

        return &primaryRunner{
                sessionRunner: sessionRunner{
                        session: s,
                },
        }
}

// Save persists the current state for a primary session, that allows
// a subsequent run to complete the resume.
func (s *PrimarySession) Save(path string) error {
        return s.navigator.save(path)
}

func (s *PrimarySession) Run(ai ...*AsyncInfo) (result *TraverseResult, err error) {
        defer s.finish(result, err, ai...)

        s.session.start()

        return s.navigator.walk(s.Path, ai...)
}

func (s *PrimarySession) StartedAt() time.Time {
        return s.startAt
}

func (s *PrimarySession) Elapsed() time.Duration {
        return s.duration
}

func (s *PrimarySession) finish(result *TraverseResult, err error, ai ...*AsyncInfo) {
        defer s.session.finish(result, err, ai...)

        _ = s.navigator.finish()
}

// ResumeSession represents a traversal that is invoked as a result
// of the user needing to resume a previously interrupted navigation
// session.
type ResumeSession struct {
        session
        Path     string
        Restorer func(o *TraverseOptions, active *ActiveState)
        Strategy ResumeStrategyEnum
        rc       *resumeController
}

func (s *ResumeSession) Init() NavigationRunner {
        var err error

        s.rc, err = resumerFactory{}.new(&ResumerInfo{
                RestorePath: s.Path,
                Restorer:    s.Restorer,
                Strategy:    s.Strategy,
        })

        if err != nil {
                panic(xi18n.NewFailedToResumeFromFileError(s.Path, err))
        }

        return &resumeRunner{
                sessionRunner: sessionRunner{
                        session: s,
                },
        }
}

// Restore is the pre run stage for a resume session
func (s *ResumeSession) Restore(restore func(o *TraverseOptions, active *ActiveState)) NavigationRunner {
        var err error

        s.rc, err = resumerFactory{}.new(&ResumerInfo{
                RestorePath: s.Path,
                Restorer:    restore,
                Strategy:    s.Strategy,
        })

        if err != nil {
                panic(xi18n.NewFailedToResumeFromFileError(s.Path, err))
        }

        return &resumeRunner{
                sessionRunner: sessionRunner{
                        session: s,
                },
        }
}

// Save persists the current state for a resume session, that allows
// a subsequent run to complete the resume.
func (s *ResumeSession) Save(path string) error {
        return s.rc.navigator.save(path)
}

func (s *ResumeSession) Run(ai ...*AsyncInfo) (result *TraverseResult, err error) {
        defer s.finish(result, err, ai...)

        s.session.start()

        return s.rc.Continue(ai...)
}

func (s *ResumeSession) StartedAt() time.Time {
        return s.startAt
}

func (s *ResumeSession) Elapsed() time.Duration {
        return s.duration
}

func (s *ResumeSession) finish(result *TraverseResult, err error, ai ...*AsyncInfo) {
        defer s.session.finish(result, err, ai...)

        _ = s.rc.finish()
}

1	package nav
2
3	import (
4	"fmt"
5	"runtime"
6	"time"
7
8	xi18n "github.com/snivilised/extendio/i18n"
9	)
10
11	const (
12	MinNoWorkers = 1
13	MaxNoWorkers = 100
14	)
15
16	type TraverseSession interface {
17	Init() NavigationRunner
18	Run(ai ...AsyncInfo) (TraverseResult, error)
19	StartedAt() time.Time
20	Elapsed() time.Duration
21	}
22
23	type NavigationRunner interface {
24	Run(ai ...AsyncInfo) (TraverseResult, error)
25	WithCPUPool() NavigationRunner
26	WithPool(now int) NavigationRunner
27	}
28
29	type sessionRunner struct {
30	session TraverseSession
31	accelerator navigationAccelerator
32	}
33
34	// Run executes the traversal session
35	func (r sessionRunner) Run(ai ...AsyncInfo) (*TraverseResult, error) {	4✔
36	if r.accelerator.noWorkers > 0 && len(ai) > 0 {	8✔
37	r.accelerator.start(ai[0])	4✔
38	return r.session.Run(ai[0])	4✔
39	}	4✔
40
41	return r.session.Run()	4✔
42	}
43
44	func (r *sessionRunner) WithPool(now int) NavigationRunner {	4✔
45	if now >= MinNoWorkers && now <= MaxNoWorkers {	8✔
46	r.accelerator.noWorkers = now	4✔
47	}	4✔
48
49	return r	4✔
50	}
51
52	func (r *sessionRunner) WithCPUPool() NavigationRunner {	4✔
53	r.accelerator.noWorkers = runtime.NumCPU()	4✔
54	return r	4✔
55	}	4✔
56
57	type primaryRunner struct {
58	sessionRunner
59	}
60
61	type resumeRunner struct {
62	sessionRunner
63	}
64
65	type session struct {
66	startAt time.Time
67	duration time.Duration
68	}
69
70	func (s *session) start() {	4✔
71	s.startAt = time.Now()	4✔
72	}	4✔
73
74	func (s session) finish(_ TraverseResult, _ error, ai ...*AsyncInfo) {	4✔
75	defer func() {	8✔
76	if len(ai) > 0 {	8✔
77	fmt.Printf("---> 😈😈😈 defer session.finish\n")	4✔
78	close(ai[0].JobsChanOut)	4✔
79	}	4✔
80	}()
81
82	s.duration = time.Since(s.startAt)	4✔
83	}
84
85	// PrimarySession
86	type PrimarySession struct {
87	session
88	Path string
89	OptionFn TraverseOptionFn
90	navigator TraverseNavigator
91	}
92
93	func (s *PrimarySession) Init() NavigationRunner {	4✔
94	s.navigator = navigatorFactory{}.new(s.OptionFn)	4✔
95		4✔
96	return &primaryRunner{	4✔
97	sessionRunner: sessionRunner{	4✔
98	session: s,	4✔
99	},	4✔
100	}	4✔
101	}	4✔
102
103	// Save persists the current state for a primary session, that allows
104	// a subsequent run to complete the resume.
105	func (s *PrimarySession) Save(path string) error {	4✔
106	return s.navigator.save(path)	4✔
107	}	4✔
108
109	func (s PrimarySession) Run(ai ...AsyncInfo) (result *TraverseResult, err error) {	4✔
110	defer s.finish(result, err, ai...)	4✔
111		4✔
112	s.session.start()	4✔
113		4✔
114	return s.navigator.walk(s.Path, ai...)	4✔
115	}	4✔
116
117	func (s *PrimarySession) StartedAt() time.Time {	4✔
118	return s.startAt	4✔
119	}	4✔
120
121	func (s *PrimarySession) Elapsed() time.Duration {	4✔
122	return s.duration	4✔
123	}	4✔
124
125	func (s PrimarySession) finish(result TraverseResult, err error, ai ...*AsyncInfo) {	4✔
126	defer s.session.finish(result, err, ai...)	4✔
127		4✔
128	_ = s.navigator.finish()	4✔
129	}	4✔
130
131	// ResumeSession represents a traversal that is invoked as a result
132	// of the user needing to resume a previously interrupted navigation
133	// session.
134	type ResumeSession struct {
135	session
136	Path string
137	Restorer func(o TraverseOptions, active ActiveState)
138	Strategy ResumeStrategyEnum
139	rc *resumeController
140	}
141
142	func (s *ResumeSession) Init() NavigationRunner {	4✔
143	var err error	4✔
144		4✔
145	s.rc, err = resumerFactory{}.new(&ResumerInfo{	4✔
146	RestorePath: s.Path,	4✔
147	Restorer: s.Restorer,	4✔
148	Strategy: s.Strategy,	4✔
149	})	4✔
150		4✔
151	if err != nil {	4✔
152	panic(xi18n.NewFailedToResumeFromFileError(s.Path, err))	×
153	}
154
155	return &resumeRunner{	4✔
156	sessionRunner: sessionRunner{	4✔
157	session: s,	4✔
158	},	4✔
159	}	4✔
160	}
161
162	// Restore is the pre run stage for a resume session
163	func (s ResumeSession) Restore(restore func(o TraverseOptions, active *ActiveState)) NavigationRunner {	×
164	var err error	×
165		×
166	s.rc, err = resumerFactory{}.new(&ResumerInfo{	×
167	RestorePath: s.Path,	×
168	Restorer: restore,	×
169	Strategy: s.Strategy,	×
170	})	×
171		×
172	if err != nil {	×
173	panic(xi18n.NewFailedToResumeFromFileError(s.Path, err))	×
174	}
175
176	return &resumeRunner{	×
177	sessionRunner: sessionRunner{	×
178	session: s,	×
179	},	×
180	}	×
181	}
182
183	// Save persists the current state for a resume session, that allows
184	// a subsequent run to complete the resume.
185	func (s *ResumeSession) Save(path string) error {	×
186	return s.rc.navigator.save(path)	×
187	}	×
188
189	func (s ResumeSession) Run(ai ...AsyncInfo) (result *TraverseResult, err error) {	4✔
190	defer s.finish(result, err, ai...)	4✔
191		4✔
192	s.session.start()	4✔
193		4✔
194	return s.rc.Continue(ai...)	4✔
195	}	4✔
196
197	func (s *ResumeSession) StartedAt() time.Time {	×
198	return s.startAt	×
199	}	×
200
201	func (s *ResumeSession) Elapsed() time.Duration {	×
202	return s.duration	×
203	}	×
204
205	func (s ResumeSession) finish(result TraverseResult, err error, ai ...*AsyncInfo) {	4✔
206	defer s.session.finish(result, err, ai...)	4✔
207		4✔
208	_ = s.rc.finish()	4✔
209	}	4✔

snivilised / extendio / 5951900512

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous