forked from golang/dep
-
Notifications
You must be signed in to change notification settings - Fork 0
/
source_manager.go
783 lines (670 loc) · 23.9 KB
/
source_manager.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
// Copyright 2017 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package gps
import (
"context"
"fmt"
"io/ioutil"
"log"
"net/url"
"os"
"os/signal"
"path/filepath"
"runtime"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/golang/dep/gps/pkgtree"
"github.com/golang/dep/internal/fs"
"github.com/nightlyone/lockfile"
"github.com/pkg/errors"
"github.com/sdboyer/constext"
)
// Used to compute a friendly filepath from a URL-shaped input.
var sanitizer = strings.NewReplacer("-", "--", ":", "-", "/", "-", "+", "-")
// A locker is responsible for preventing multiple instances of dep from
// interfering with one-another.
//
// Currently, anything that can either TryLock(), Unlock(), or GetOwner()
// satisfies that need.
type locker interface {
TryLock() error
Unlock() error
GetOwner() (*os.Process, error)
}
// A falselocker adheres to the locker interface and its purpose is to quietly
// fail to lock when the DEPNOLOCK environment variable is set.
//
// This allows dep to run on systems where file locking doesn't work --
// particularly those that use union mount type filesystems that don't
// implement hard links or fnctl() style locking.
type falseLocker struct{}
// Always returns an error to indicate there's no current ower PID for our
// lock.
func (fl falseLocker) GetOwner() (*os.Process, error) {
return nil, fmt.Errorf("falseLocker always fails")
}
// Does nothing and returns a nil error so caller believes locking succeeded.
func (fl falseLocker) TryLock() error {
return nil
}
// Does nothing and returns a nil error so caller believes unlocking succeeded.
func (fl falseLocker) Unlock() error {
return nil
}
// A SourceManager is responsible for retrieving, managing, and interrogating
// source repositories. Its primary purpose is to serve the needs of a Solver,
// but it is handy for other purposes, as well.
//
// gps's built-in SourceManager, SourceMgr, is intended to be generic and
// sufficient for any purpose. It provides some additional semantics around the
// methods defined here.
type SourceManager interface {
// SourceExists checks if a repository exists, either upstream or in the
// SourceManager's central repository cache.
SourceExists(ProjectIdentifier) (bool, error)
// SyncSourceFor will attempt to bring all local information about a source
// fully up to date.
SyncSourceFor(ProjectIdentifier) error
// ListVersions retrieves a list of the available versions for a given
// repository name.
ListVersions(ProjectIdentifier) ([]PairedVersion, error)
// RevisionPresentIn indicates whether the provided Version is present in
// the given repository.
RevisionPresentIn(ProjectIdentifier, Revision) (bool, error)
// ListPackages parses the tree of the Go packages at or below root of the
// provided ProjectIdentifier, at the provided version.
ListPackages(ProjectIdentifier, Version) (pkgtree.PackageTree, error)
// GetManifestAndLock returns manifest and lock information for the provided
// root import path.
//
// gps currently requires that projects be rooted at their repository root,
// necessitating that the ProjectIdentifier's ProjectRoot must also be a
// repository root.
GetManifestAndLock(ProjectIdentifier, Version, ProjectAnalyzer) (Manifest, Lock, error)
// ExportProject writes out the tree of the provided import path, at the
// provided version, to the provided directory.
ExportProject(context.Context, ProjectIdentifier, Version, string) error
// DeduceProjectRoot takes an import path and deduces the corresponding
// project/source root.
DeduceProjectRoot(ip string) (ProjectRoot, error)
// SourceURLsForPath takes an import path and deduces the set of source URLs
// that may refer to a canonical upstream source.
// In general, these URLs differ only by protocol (e.g. https vs. ssh), not path
SourceURLsForPath(ip string) ([]*url.URL, error)
// Release lets go of any locks held by the SourceManager. Once called, it is
// no longer safe to call methods against it; all method calls will
// immediately result in errors.
Release()
// InferConstraint tries to puzzle out what kind of version is given in a string -
// semver, a revision, or as a fallback, a plain tag
InferConstraint(s string, pi ProjectIdentifier) (Constraint, error)
}
// A ProjectAnalyzer is responsible for analyzing a given path for Manifest and
// Lock information. Tools relying on gps must implement one.
type ProjectAnalyzer interface {
// Perform analysis of the filesystem tree rooted at path, with the
// root import path importRoot, to determine the project's constraints, as
// indicated by a Manifest and Lock.
//
// Note that an error will typically cause the solver to treat the analyzed
// version as unusable. As such, an error should generally only be returned
// if the code tree is somehow malformed, but not if the implementor's
// expected files containing Manifest and Lock data are merely absent.
DeriveManifestAndLock(path string, importRoot ProjectRoot) (Manifest, Lock, error)
// Info reports this project analyzer's info.
Info() ProjectAnalyzerInfo
}
// ProjectAnalyzerInfo indicates a ProjectAnalyzer's name and version.
type ProjectAnalyzerInfo struct {
Name string
Version int
}
// String returns a string like: "<name>.<decimal version>"
func (p ProjectAnalyzerInfo) String() string {
return fmt.Sprintf("%s.%d", p.Name, p.Version)
}
// SourceMgr is the default SourceManager for gps.
//
// There's no (planned) reason why it would need to be reimplemented by other
// tools; control via dependency injection is intended to be sufficient.
type SourceMgr struct {
cachedir string // path to root of cache dir
lf locker // handle for the sm lock file on disk
suprvsr *supervisor // subsystem that supervises running calls/io
cancelAll context.CancelFunc // cancel func to kill all running work
deduceCoord *deductionCoordinator // subsystem that manages import path deduction
srcCoord *sourceCoordinator // subsystem that manages sources
sigmut sync.Mutex // mutex protecting signal handling setup/teardown
qch chan struct{} // quit chan for signal handler
relonce sync.Once // once-er to ensure we only release once
releasing int32 // flag indicating release of sm has begun
}
var _ SourceManager = &SourceMgr{}
// ErrSourceManagerIsReleased is the error returned by any SourceManager method
// called after the SourceManager has been released, rendering its methods no
// longer safe to call.
var ErrSourceManagerIsReleased = fmt.Errorf("this SourceManager has been released, its methods can no longer be called")
// SourceManagerConfig holds configuration information for creating SourceMgrs.
type SourceManagerConfig struct {
Cachedir string // Where to store local instances of upstream sources.
Logger *log.Logger // Optional info/warn logger. Discards if nil.
DisableLocking bool // True if the SourceManager should NOT use a lock file to protect the Cachedir from multiple processes.
}
// NewSourceManager produces an instance of gps's built-in SourceManager.
//
// The returned SourceManager aggressively caches information wherever possible.
// If tools need to do preliminary work involving upstream repository analysis
// prior to invoking a solve run, it is recommended that they create this
// SourceManager as early as possible and use it to their ends. That way, the
// solver can benefit from any caches that may have already been warmed.
//
// gps's SourceManager is intended to be threadsafe (if it's not, please file a
// bug!). It should be safe to reuse across concurrent solving runs, even on
// unrelated projects.
func NewSourceManager(c SourceManagerConfig) (*SourceMgr, error) {
if c.Logger == nil {
c.Logger = log.New(ioutil.Discard, "", 0)
}
err := fs.EnsureDir(filepath.Join(c.Cachedir, "sources"), 0777)
if err != nil {
return nil, err
}
// Fix for #820
//
// Consult https://godoc.org/github.com/nightlyone/lockfile for the lockfile
// behaviour. It's magic. It deals with stale processes, and if there is
// a process keeping the lock busy, it will pass back a temporary error that
// we can spin on.
glpath := filepath.Join(c.Cachedir, "sm.lock")
lockfile, err := func() (locker, error) {
if c.DisableLocking {
return falseLocker{}, nil
}
return lockfile.New(glpath)
}()
if err != nil {
return nil, CouldNotCreateLockError{
Path: glpath,
Err: errors.Wrapf(err, "unable to create lock %s", glpath),
}
}
process, err := lockfile.GetOwner()
if err == nil {
// If we didn't get an error, then the lockfile exists already. We should
// check to see if it's us already:
if process.Pid == os.Getpid() {
return nil, CouldNotCreateLockError{
Path: glpath,
Err: fmt.Errorf("lockfile %s already locked by this process", glpath),
}
}
// There is a lockfile, but it's owned by someone else. We'll try to lock
// it anyway.
}
// If it's a TemporaryError, we retry every second. Otherwise, we fail
// permanently.
//
// TODO: #534 needs to be implemented to provide a better way to log warnings,
// but until then we will just use stderr.
// Implicit Time of 0.
var lasttime time.Time
err = lockfile.TryLock()
for err != nil {
nowtime := time.Now()
duration := nowtime.Sub(lasttime)
// The first time this is evaluated, duration will be very large as lasttime is 0.
// Unless time travel is invented and someone travels back to the year 1, we should
// be ok.
if duration > 15*time.Second {
fmt.Fprintf(os.Stderr, "waiting for lockfile %s: %s\n", glpath, err.Error())
lasttime = nowtime
}
if t, ok := err.(interface {
Temporary() bool
}); ok && t.Temporary() {
time.Sleep(time.Second * 1)
} else {
return nil, CouldNotCreateLockError{
Path: glpath,
Err: errors.Wrapf(err, "unable to lock %s", glpath),
}
}
err = lockfile.TryLock()
}
ctx, cf := context.WithCancel(context.TODO())
superv := newSupervisor(ctx)
deducer := newDeductionCoordinator(superv)
sm := &SourceMgr{
cachedir: c.Cachedir,
lf: lockfile,
suprvsr: superv,
cancelAll: cf,
deduceCoord: deducer,
srcCoord: newSourceCoordinator(superv, deducer, c.Cachedir, c.Logger),
qch: make(chan struct{}),
}
return sm, nil
}
// Cachedir returns the location of the cache directory.
func (sm *SourceMgr) Cachedir() string {
return sm.cachedir
}
// UseDefaultSignalHandling sets up typical os.Interrupt signal handling for a
// SourceMgr.
func (sm *SourceMgr) UseDefaultSignalHandling() {
sigch := make(chan os.Signal, 1)
signal.Notify(sigch, os.Interrupt)
sm.HandleSignals(sigch)
}
// HandleSignals sets up logic to handle incoming signals with the goal of
// shutting down the SourceMgr safely.
//
// Calling code must provide the signal channel, and is responsible for calling
// signal.Notify() on that channel.
//
// Successive calls to HandleSignals() will deregister the previous handler and
// set up a new one. It is not recommended that the same channel be passed
// multiple times to this method.
//
// SetUpSigHandling() will set up a handler that is appropriate for most
// use cases.
func (sm *SourceMgr) HandleSignals(sigch chan os.Signal) {
sm.sigmut.Lock()
// always start by closing the qch, which will lead to any existing signal
// handler terminating, and deregistering its sigch.
if sm.qch != nil {
close(sm.qch)
}
sm.qch = make(chan struct{})
// Run a new goroutine with the input sigch and the fresh qch
go func(sch chan os.Signal, qch <-chan struct{}) {
defer signal.Stop(sch)
select {
case <-sch:
// Set up a timer to uninstall the signal handler after three
// seconds, so that the user can easily force termination with a
// second ctrl-c
time.AfterFunc(3*time.Second, func() {
signal.Stop(sch)
})
if opc := sm.suprvsr.count(); opc > 0 {
fmt.Printf("Signal received: waiting for %v ops to complete...\n", opc)
}
sm.Release()
case <-qch:
// quit channel triggered - deregister our sigch and return
}
}(sigch, sm.qch)
// Try to ensure handler is blocked in for-select before releasing the mutex
runtime.Gosched()
sm.sigmut.Unlock()
}
// StopSignalHandling deregisters any signal handler running on this SourceMgr.
//
// It's normally not necessary to call this directly; it will be called as
// needed by Release().
func (sm *SourceMgr) StopSignalHandling() {
sm.sigmut.Lock()
if sm.qch != nil {
close(sm.qch)
sm.qch = nil
runtime.Gosched()
}
sm.sigmut.Unlock()
}
// CouldNotCreateLockError describe failure modes in which creating a SourceMgr
// did not succeed because there was an error while attempting to create the
// on-disk lock file.
type CouldNotCreateLockError struct {
Path string
Err error
}
func (e CouldNotCreateLockError) Error() string {
return e.Err.Error()
}
// Release lets go of any locks held by the SourceManager. Once called, it is no
// longer safe to call methods against it; all method calls will immediately
// result in errors.
func (sm *SourceMgr) Release() {
atomic.StoreInt32(&sm.releasing, 1)
sm.relonce.Do(func() {
// Send the signal to the supervisor to cancel all running calls.
sm.cancelAll()
sm.suprvsr.wait()
// Close the source coordinator.
sm.srcCoord.close()
// Close the file handle for the lock file and remove it from disk
sm.lf.Unlock()
os.Remove(filepath.Join(sm.cachedir, "sm.lock"))
// Close the qch, if non-nil, so the signal handlers run out. This will
// also deregister the sig channel, if any has been set up.
if sm.qch != nil {
close(sm.qch)
}
})
}
// GetManifestAndLock returns manifest and lock information for the provided
// ProjectIdentifier, at the provided Version. The work of producing the
// manifest and lock is delegated to the provided ProjectAnalyzer's
// DeriveManifestAndLock() method.
func (sm *SourceMgr) GetManifestAndLock(id ProjectIdentifier, v Version, an ProjectAnalyzer) (Manifest, Lock, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return nil, nil, ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
return nil, nil, err
}
return srcg.getManifestAndLock(context.TODO(), id.ProjectRoot, v, an)
}
// ListPackages parses the tree of the Go packages at and below the ProjectRoot
// of the given ProjectIdentifier, at the given version.
func (sm *SourceMgr) ListPackages(id ProjectIdentifier, v Version) (pkgtree.PackageTree, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return pkgtree.PackageTree{}, ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
return pkgtree.PackageTree{}, err
}
return srcg.listPackages(context.TODO(), id.ProjectRoot, v)
}
// ListVersions retrieves a list of the available versions for a given
// repository name.
//
// The list is not sorted; while it may be returned in the order that the
// underlying VCS reports version information, no guarantee is made. It is
// expected that the caller either not care about order, or sort the result
// themselves.
//
// This list is always retrieved from upstream on the first call. Subsequent
// calls will return a cached version of the first call's results. if upstream
// is not accessible (network outage, access issues, or the resource actually
// went away), an error will be returned.
func (sm *SourceMgr) ListVersions(id ProjectIdentifier) ([]PairedVersion, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return nil, ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
// TODO(sdboyer) More-er proper-er errors
return nil, err
}
return srcg.listVersions(context.TODO())
}
// RevisionPresentIn indicates whether the provided Revision is present in the given
// repository.
func (sm *SourceMgr) RevisionPresentIn(id ProjectIdentifier, r Revision) (bool, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return false, ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
// TODO(sdboyer) More-er proper-er errors
return false, err
}
return srcg.revisionPresentIn(context.TODO(), r)
}
// SourceExists checks if a repository exists, either upstream or in the cache,
// for the provided ProjectIdentifier.
func (sm *SourceMgr) SourceExists(id ProjectIdentifier) (bool, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return false, ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
return false, err
}
ctx := context.TODO()
return srcg.existsInCache(ctx) || srcg.existsUpstream(ctx), nil
}
// SyncSourceFor will ensure that all local caches and information about a
// source are up to date with any network-acccesible information.
//
// The primary use case for this is prefetching.
func (sm *SourceMgr) SyncSourceFor(id ProjectIdentifier) error {
if atomic.LoadInt32(&sm.releasing) == 1 {
return ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
if err != nil {
return err
}
return srcg.syncLocal(context.TODO())
}
// ExportProject writes out the tree of the provided ProjectIdentifier's
// ProjectRoot, at the provided version, to the provided directory.
func (sm *SourceMgr) ExportProject(ctx context.Context, id ProjectIdentifier, v Version, to string) error {
if atomic.LoadInt32(&sm.releasing) == 1 {
return ErrSourceManagerIsReleased
}
srcg, err := sm.srcCoord.getSourceGatewayFor(ctx, id)
if err != nil {
return err
}
return srcg.exportVersionTo(ctx, v, to)
}
// DeduceProjectRoot takes an import path and deduces the corresponding
// project/source root.
//
// Note that some import paths may require network activity to correctly
// determine the root of the path, such as, but not limited to, vanity import
// paths. (A special exception is written for gopkg.in to minimize network
// activity, as its behavior is well-structured)
func (sm *SourceMgr) DeduceProjectRoot(ip string) (ProjectRoot, error) {
if atomic.LoadInt32(&sm.releasing) == 1 {
return "", ErrSourceManagerIsReleased
}
// TODO(sdboyer) refactor deduceRootPath() so that this validation can move
// back down below a cache point, rather than executing on every call.
if !pathvld.MatchString(ip) {
return "", errors.Errorf("%q is not a valid import path", ip)
}
pd, err := sm.deduceCoord.deduceRootPath(context.TODO(), ip)
return ProjectRoot(pd.root), err
}
// InferConstraint tries to puzzle out what kind of version is given in a
// string. Preference is given first for branches, then semver constraints, then
// plain tags, and then revisions.
func (sm *SourceMgr) InferConstraint(s string, pi ProjectIdentifier) (Constraint, error) {
if s == "" {
return Any(), nil
}
// Lookup the string in the repository
var version PairedVersion
versions, err := sm.ListVersions(pi)
if err != nil {
return nil, errors.Wrapf(err, "list versions for %s", pi) // means repo does not exist
}
SortPairedForUpgrade(versions)
for _, v := range versions {
if s == v.String() {
version = v
break
}
}
// Branch
if version != nil && version.Type() == IsBranch {
return version.Unpair(), nil
}
// Semver Constraint
c, err := NewSemverConstraintIC(s)
if c != nil && err == nil {
return c, nil
}
// Tag
if version != nil {
return version.Unpair(), nil
}
// Revision, possibly abbreviated
r, err := sm.disambiguateRevision(context.TODO(), pi, Revision(s))
if err == nil {
return r, nil
}
return nil, errors.Errorf("%s is not a valid version for the package %s(%s)", s, pi.ProjectRoot, pi.Source)
}
// SourceURLsForPath takes an import path and deduces the set of source URLs
// that may refer to a canonical upstream source.
// In general, these URLs differ only by protocol (e.g. https vs. ssh), not path
func (sm *SourceMgr) SourceURLsForPath(ip string) ([]*url.URL, error) {
deduced, err := sm.deduceCoord.deduceRootPath(context.TODO(), ip)
if err != nil {
return nil, err
}
return deduced.mb.possibleURLs(), nil
}
// disambiguateRevision looks up a revision in the underlying source, spitting
// it back out in an unabbreviated, disambiguated form.
//
// For example, if pi refers to a git-based project, then rev could be an
// abbreviated git commit hash. disambiguateRevision would return the complete
// hash.
func (sm *SourceMgr) disambiguateRevision(ctx context.Context, pi ProjectIdentifier, rev Revision) (Revision, error) {
srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), pi)
if err != nil {
return "", err
}
return srcg.disambiguateRevision(ctx, rev)
}
type timeCount struct {
count int
start time.Time
}
type durCount struct {
count int
dur time.Duration
}
type supervisor struct {
ctx context.Context
mu sync.Mutex // Guards all maps
cond sync.Cond // Wraps mu so callers can wait until all calls end
running map[callInfo]timeCount
ran map[callType]durCount
}
func newSupervisor(ctx context.Context) *supervisor {
supv := &supervisor{
ctx: ctx,
running: make(map[callInfo]timeCount),
ran: make(map[callType]durCount),
}
supv.cond = sync.Cond{L: &supv.mu}
return supv
}
// do executes the incoming closure using a conjoined context, and keeps
// counters to ensure the sourceMgr can't finish Release()ing until after all
// calls have returned.
func (sup *supervisor) do(inctx context.Context, name string, typ callType, f func(context.Context) error) error {
ci := callInfo{
name: name,
typ: typ,
}
octx, err := sup.start(ci)
if err != nil {
return err
}
cctx, cancelFunc := constext.Cons(inctx, octx)
err = f(cctx)
sup.done(ci)
cancelFunc()
return err
}
func (sup *supervisor) start(ci callInfo) (context.Context, error) {
sup.mu.Lock()
defer sup.mu.Unlock()
if err := sup.ctx.Err(); err != nil {
// We've already been canceled; error out.
return nil, err
}
if existingInfo, has := sup.running[ci]; has {
existingInfo.count++
sup.running[ci] = existingInfo
} else {
sup.running[ci] = timeCount{
count: 1,
start: time.Now(),
}
}
return sup.ctx, nil
}
func (sup *supervisor) count() int {
sup.mu.Lock()
defer sup.mu.Unlock()
return len(sup.running)
}
func (sup *supervisor) done(ci callInfo) {
sup.mu.Lock()
existingInfo, has := sup.running[ci]
if !has {
panic(fmt.Sprintf("sourceMgr: tried to complete a call that had not registered via run()"))
}
if existingInfo.count > 1 {
// If more than one is pending, don't stop the clock yet.
existingInfo.count--
sup.running[ci] = existingInfo
} else {
// Last one for this particular key; update metrics with info.
durCnt := sup.ran[ci.typ]
durCnt.count++
durCnt.dur += time.Since(existingInfo.start)
sup.ran[ci.typ] = durCnt
delete(sup.running, ci)
if len(sup.running) == 0 {
// This is the only place where we signal the cond, as it's the only
// time that the number of running calls could become zero.
sup.cond.Signal()
}
}
sup.mu.Unlock()
}
// wait until all active calls have terminated.
//
// Assumes something else has already canceled the supervisor via its context.
func (sup *supervisor) wait() {
sup.cond.L.Lock()
for len(sup.running) > 0 {
sup.cond.Wait()
}
sup.cond.L.Unlock()
}
type callType uint
const (
ctHTTPMetadata callType = iota
ctListVersions
ctGetManifestAndLock
ctListPackages
ctSourcePing
ctSourceInit
ctSourceFetch
ctExportTree
ctValidateLocal
)
func (ct callType) String() string {
switch ct {
case ctHTTPMetadata:
return "Retrieving go get metadata"
case ctListVersions:
return "Retrieving latest version list"
case ctGetManifestAndLock:
return "Reading manifest and lock data"
case ctListPackages:
return "Parsing PackageTree"
case ctSourcePing:
return "Checking for upstream existence"
case ctSourceInit:
return "Initializing local source cache"
case ctSourceFetch:
return "Fetching latest data into local source cache"
case ctExportTree:
return "Writing code tree out to disk"
default:
panic("unknown calltype")
}
}
// callInfo provides metadata about an ongoing call.
type callInfo struct {
name string
typ callType
}