diff options
Diffstat (limited to 'internal/pkg/pkg.go')
| -rw-r--r-- | internal/pkg/pkg.go | 459 |
1 files changed, 348 insertions, 111 deletions
diff --git a/internal/pkg/pkg.go b/internal/pkg/pkg.go index f5bd9d3a..199a6177 100644 --- a/internal/pkg/pkg.go +++ b/internal/pkg/pkg.go @@ -3,6 +3,7 @@ package pkg import ( "bytes" + "context" "crypto/sha512" "encoding/base64" "encoding/binary" @@ -12,11 +13,14 @@ import ( "os" "path" "path/filepath" + "runtime" "slices" "strings" "sync" + "syscall" "hakurei.app/container/check" + "hakurei.app/message" ) type ( @@ -53,45 +57,123 @@ func MustDecode(s string) Checksum { } } -// CureContext is passed to [Artifact.Cure] and contains information and methods -// useful for curing the [Artifact], like requesting the data of [File], or that -// other artifacts be cured. +// TContext is passed to [TrivialArtifact.Cure] and provides information and +// methods required for curing the [TrivialArtifact]. // -// Methods of CureContext are safe for concurrent use. CureContext is valid -// until [Artifact.Cure] returns. -type CureContext struct { - // Address of underlying [Cache], should be zeroed after [Artifact.Cure] - // returns and must not be exposed directly. +// Methods of TContext are safe for concurrent use. TContext is valid +// until [TrivialArtifact.Cure] returns. +type TContext struct { + // Address of underlying [Cache], should be zeroed or made unusable after + // [TrivialArtifact.Cure] returns and must not be exposed directly. cache *Cache // Populated during [Cache.Cure]. work, temp *check.Absolute } +// destroy destroys the temporary directory and joins its errors with the error +// referred to by errP. If the error referred to by errP is non-nil, the work +// directory is removed similarly. [Cache] is responsible for making sure work +// is never left behind for a successful [Cache.Cure]. +// +// destroy must be deferred by [Cache.Cure] if [TContext] is passed to any Cure +// implementation. It should not be called prior to that point. +func (t *TContext) destroy(errP *error) { + if chmodErr, removeErr := removeAll(t.temp); chmodErr != nil || removeErr != nil { + *errP = errors.Join(*errP, chmodErr, removeErr) + return + } + + if *errP != nil { + chmodErr, removeErr := removeAll(t.work) + if chmodErr != nil || removeErr != nil { + *errP = errors.Join(*errP, chmodErr, removeErr) + } else if errors.Is(*errP, os.ErrExist) { + // two artifacts may be backed by the same file + *errP = nil + } + } +} + +// Unwrap returns the underlying [context.Context]. +func (t *TContext) Unwrap() context.Context { return t.cache.ctx } + +// GetMessage returns [message.Msg] held by the underlying [Cache]. +func (t *TContext) GetMessage() message.Msg { return t.cache.msg } + // GetWorkDir returns a pathname to a directory which [Artifact] is expected to // write its output to. This is not the final resting place of the [Artifact] // and this pathname should not be directly referred to in the final contents. -func (c *CureContext) GetWorkDir() *check.Absolute { return c.work } +func (t *TContext) GetWorkDir() *check.Absolute { return t.work } // GetTempDir returns a pathname which implementations may use as scratch space. // A directory is not created automatically, implementations are expected to // create it if they wish to use it, using [os.MkdirAll]. -func (c *CureContext) GetTempDir() *check.Absolute { return c.temp } +func (t *TContext) GetTempDir() *check.Absolute { return t.temp } -// Cure cures the [Artifact] and returns its pathname and [Checksum]. -func (c *CureContext) Cure(a Artifact) ( - pathname *check.Absolute, - checksum Checksum, - err error, -) { - return c.cache.Cure(a) +// Open tries to open [Artifact] for reading. If a implements [File], its data +// might be used directly, eliminating the roundtrip to vfs. Otherwise, it must +// cure into a directory containing a single regular file. +// +// If err is nil, the caller is responsible for closing the resulting +// [io.ReadCloser]. +func (t *TContext) Open(a Artifact) (r io.ReadCloser, err error) { + if f, ok := a.(File); ok { + return t.cache.openFile(f) + } + + var pathname *check.Absolute + if pathname, _, err = t.cache.Cure(a); err != nil { + return + } + + var entries []os.DirEntry + if entries, err = os.ReadDir(pathname.String()); err != nil { + return + } + + if len(entries) != 1 || !entries[0].Type().IsRegular() { + err = errors.New( + "input directory does not contain a single regular file", + ) + return + } else { + return os.Open(pathname.Append(entries[0].Name()).String()) + } } -// OpenFile tries to load [File] from [Cache], and if that fails, obtains it via -// [File.Data] instead. Notably, it does not cure [File]. If err is nil, the -// caller is responsible for closing the resulting [io.ReadCloser]. -func (c *CureContext) OpenFile(f File) (r io.ReadCloser, err error) { - return c.cache.openFile(f) +// FContext is passed to [FloodArtifact.Cure] and provides information and +// methods required for curing the [FloodArtifact]. +// +// Methods of FContext are safe for concurrent use. FContext is valid +// until [FloodArtifact.Cure] returns. +type FContext struct { + TContext + + // Cured top-level dependencies looked up by Pathname. + deps map[ID]*check.Absolute +} + +// InvalidLookupError is the identifier of non-dependency [Artifact] looked up +// via [FContext.Pathname] by a misbehaving [Artifact] implementation. +type InvalidLookupError ID + +func (e InvalidLookupError) Error() string { + return "attempting to look up non-dependency artifact " + Encode(e) +} + +var _ error = InvalidLookupError{} + +// Pathname returns the identifier pathname of an [Artifact]. Calling Pathname +// with an [Artifact] not part of the slice returned by [Artifact.Dependencies] +// panics. +func (f *FContext) Pathname(a Artifact) *check.Absolute { + id := Ident(a) + if p, ok := f.deps[id]; ok { + return p + } else { + panic(InvalidLookupError(id)) + } } // An Artifact is a read-only reference to a piece of data that may be created @@ -119,16 +201,35 @@ type Artifact interface { // // Result must remain identical across multiple invocations. Dependencies() []Artifact +} +// FloodArtifact refers to an [Artifact] requiring its entire dependency graph +// to be cured prior to curing itself. +type FloodArtifact interface { // Cure cures the current [Artifact] to the working directory obtained via - // [CureContext.GetWorkDir]. + // [TContext.GetWorkDir] embedded in [FContext]. // - // If the implementation produces a single file, it must implement [File] - // as well. In that case, Cure must produce a single regular file with - // contents identical to that returned by [File.Data]. + // Implementations must not retain c. + Cure(f *FContext) (err error) + + Artifact +} + +// TrivialArtifact refers to an [Artifact] that cures without requiring that +// any other [Artifact] is cured before it. Its dependency tree is ignored after +// computing its identifier. +// +// TrivialArtifact is unable to cure any other [Artifact] and it cannot access +// pathnames. This type of [Artifact] is primarily intended for dependency-less +// artifacts or direct dependencies that only consists of [File]. +type TrivialArtifact interface { + // Cure cures the current [Artifact] to the working directory obtained via + // [TContext.GetWorkDir]. // // Implementations must not retain c. - Cure(c *CureContext) (err error) + Cure(t *TContext) (err error) + + Artifact } // KnownIdent is optionally implemented by [Artifact] and is used instead of @@ -157,12 +258,14 @@ type KnownChecksum interface { // A File refers to an [Artifact] backed by a single file. type File interface { - // Data returns the full contents of [Artifact]. If [Artifact.Checksum] - // returns a non-nil address, Data is responsible for validating any data - // it produces and must return [ChecksumMismatchError] if validation fails. + // Cure returns the full contents of [File]. If [File] implements + // [KnownChecksum], Cure is responsible for validating any data it produces + // and must return [ChecksumMismatchError] if validation fails. // // Callers must not modify the returned byte slice. - Data() ([]byte, error) + // + // Result must remain identical across multiple invocations. + Cure() ([]byte, error) Artifact } @@ -242,9 +345,47 @@ const ( checksumLinknamePrefix = "../" + dirChecksum + "/" ) +// A pendingArtifactDep is a dependency [Artifact] pending concurrent curing, +// subject to the cures limit. Values pointed to by result addresses are safe +// to access after the [sync.WaitGroup] associated with this pendingArtifactDep +// is done. pendingArtifactDep must not be reused or modified after it is sent +// to Cache.cureDep. +type pendingArtifactDep struct { + // Dependency artifact populated during [Cache.Cure]. + a Artifact + + // Address of result pathname populated during [Cache.Cure] and dereferenced + // if curing succeeds. + resP **check.Absolute + + // Address of result error slice populated during [Cache.Cure], dereferenced + // after acquiring errsMu if curing fails. No additional action is taken, + // [Cache] and its caller are responsible for further error handling. + errs *[]error + // Address of mutex synchronising access to errs. + errsMu *sync.Mutex + + // For synchronising access to result buffer. + *sync.WaitGroup +} + // Cache is a support layer that implementations of [Artifact] can use to store // cured [Artifact] data in a content addressed fashion. type Cache struct { + // Work for curing dependency [Artifact] is sent here and cured concurrently + // while subject to the cures limit. Invalid after the context is canceled. + cureDep chan<- *pendingArtifactDep + + // [context.WithCancel] over caller-supplied context, used by [Artifact] and + // all dependency curing goroutines. + ctx context.Context + // Cancels ctx. + cancel context.CancelFunc + // For waiting on dependency curing goroutines. + wg sync.WaitGroup + // Reports new cures and passed to [Artifact]. + msg message.Msg + // Directory where all [Cache] related files are placed. base *check.Absolute @@ -560,7 +701,9 @@ func (c *Cache) finaliseIdent( close(done) } -// openFile provides [CureContext.OpenFile] for [Artifact.Cure]. +// openFile tries to load [File] from [Cache], and if that fails, obtains it via +// [File.Cure] instead. Notably, it does not cure [File]. If err is nil, the +// caller is responsible for closing the resulting [io.ReadCloser]. func (c *Cache) openFile(f File) (r io.ReadCloser, err error) { if kc, ok := f.(KnownChecksum); ok { c.checksumMu.RLock() @@ -583,7 +726,7 @@ func (c *Cache) openFile(f File) (r io.ReadCloser, err error) { return } var data []byte - if data, err = f.Data(); err != nil { + if data, err = f.Cure(); err != nil { return } r = io.NopCloser(bytes.NewReader(data)) @@ -691,7 +834,16 @@ func (fsys dotOverrideFS) Stat(name string) (fi fs.FileInfo, err error) { return } -// Cure cures the [Artifact] and returns its pathname and [Checksum]. +// InvalidArtifactError describes an artifact that does not implement a +// supported Cure method. +type InvalidArtifactError ID + +func (e InvalidArtifactError) Error() string { + return "artifact " + Encode(e) + " cannot be cured" +} + +// Cure cures the [Artifact] and returns its pathname and [Checksum]. Direct +// calls to Cure are not subject to the cures limit. func (c *Cache) Cure(a Artifact) ( pathname *check.Absolute, checksum Checksum, @@ -764,6 +916,11 @@ func (c *Cache) Cure(a Artifact) ( } } + if c.msg.IsVerbose() { + c.msg.Verbosef("curing %s...", Encode(id)) + } + + // cure File outside type switch to skip TContext initialisation if f, ok := a.(File); ok { if checksumFi != nil { if !checksumFi.Mode().IsRegular() { @@ -774,7 +931,7 @@ func (c *Cache) Cure(a Artifact) ( } var data []byte - data, err = f.Data() + data, err = f.Cure() if err != nil { return } @@ -823,107 +980,163 @@ func (c *Cache) Cure(a Artifact) ( c.checksumMu.Unlock() return - } else { - if checksumFi != nil { - if !checksumFi.Mode().IsDir() { - // unreachable - err = InvalidFileModeError(checksumFi.Mode()) - } - return + } + + if checksumFi != nil { + if !checksumFi.Mode().IsDir() { + // unreachable + err = InvalidFileModeError(checksumFi.Mode()) } + return + } - cc := CureContext{ - cache: c, - work: c.base.Append(dirWork, ids), - temp: c.base.Append(dirTemp, ids), + t := TContext{c, c.base.Append(dirWork, ids), c.base.Append(dirTemp, ids)} + switch ca := a.(type) { + case TrivialArtifact: + defer t.destroy(&err) + if err = ca.Cure(&t); err != nil { + return } - defer func() { - if chmodErr, removeErr := removeAll(cc.temp); chmodErr != nil || removeErr != nil { - err = errors.Join(err, chmodErr, removeErr) + break + + case FloodArtifact: + deps := a.Dependencies() + f := FContext{t, make(map[ID]*check.Absolute, len(deps))} + + var wg sync.WaitGroup + wg.Add(len(deps)) + res := make([]*check.Absolute, len(deps)) + errs := make([]error, 0, len(deps)) + var errsMu sync.Mutex + for i, d := range deps { + pending := pendingArtifactDep{d, &res[i], &errs, &errsMu, &wg} + select { + case c.cureDep <- &pending: + break + + case <-c.ctx.Done(): + err = c.ctx.Err() return } + } + wg.Wait() - if err != nil { - chmodErr, removeErr := removeAll(cc.work) - if chmodErr != nil || removeErr != nil { - err = errors.Join(err, chmodErr, removeErr) - } else if errors.Is(err, os.ErrExist) { - // two artifacts may be backed by the same file - err = nil - } + if len(errs) > 0 { + err = errors.Join(errs...) + if err == nil { + // unreachable + err = syscall.ENOTRECOVERABLE } - }() - - if err = a.Cure(&cc); err != nil { return } - cc.cache = nil - var fi os.FileInfo - if fi, err = os.Lstat(cc.work.String()); err != nil { - if errors.Is(err, os.ErrNotExist) { - err = NoOutputError{} - } - return + for i, p := range res { + f.deps[Ident(deps[i])] = p } - if !fi.IsDir() { - if !fi.Mode().IsRegular() { - err = InvalidFileModeError(fi.Mode()) - } else { - err = errors.New("non-file artifact produced regular file") - } + defer f.destroy(&err) + if err = ca.Cure(&f); err != nil { return } + break - var gotChecksum Checksum - if gotChecksum, err = HashFS( - dotOverrideFS{os.DirFS(cc.work.String()).(dirFS)}, - ".", - ); err != nil { - return + default: + err = InvalidArtifactError(id) + return + } + t.cache = nil + + var fi os.FileInfo + if fi, err = os.Lstat(t.work.String()); err != nil { + if errors.Is(err, os.ErrNotExist) { + err = NoOutputError{} } + return + } - if checksumPathname == nil { - checksum = gotChecksum - checksums = Encode(checksum) - checksumPathname = c.base.Append( - dirChecksum, - checksums, - ) + if !fi.IsDir() { + if !fi.Mode().IsRegular() { + err = InvalidFileModeError(fi.Mode()) } else { - if gotChecksum != checksum { - err = &ChecksumMismatchError{ - Got: gotChecksum, - Want: checksum, - } - return - } + err = errors.New("non-file artifact produced regular file") } + return + } - if err = os.Chmod(cc.work.String(), 0700); err != nil { + var gotChecksum Checksum + if gotChecksum, err = HashFS( + dotOverrideFS{os.DirFS(t.work.String()).(dirFS)}, + ".", + ); err != nil { + return + } + + if checksumPathname == nil { + checksum = gotChecksum + checksums = Encode(checksum) + checksumPathname = c.base.Append( + dirChecksum, + checksums, + ) + } else { + if gotChecksum != checksum { + err = &ChecksumMismatchError{ + Got: gotChecksum, + Want: checksum, + } return } - c.checksumMu.Lock() - if err = os.Rename( - cc.work.String(), - checksumPathname.String(), - ); err != nil { - if !errors.Is(err, os.ErrExist) { - c.checksumMu.Unlock() - return - } - // err is zeroed during deferred cleanup - } else { - err = os.Chmod(checksumPathname.String(), 0500) + } + + if err = os.Chmod(t.work.String(), 0700); err != nil { + return + } + c.checksumMu.Lock() + if err = os.Rename( + t.work.String(), + checksumPathname.String(), + ); err != nil { + if !errors.Is(err, os.ErrExist) { + c.checksumMu.Unlock() + return } - c.checksumMu.Unlock() + // err is zeroed during deferred cleanup + } else { + err = os.Chmod(checksumPathname.String(), 0500) + } + c.checksumMu.Unlock() + return +} +// cure cures the pending [Artifact], stores its result and notifies the caller. +func (pending *pendingArtifactDep) cure(c *Cache) { + defer pending.Done() + + pathname, _, err := c.Cure(pending.a) + if err == nil { + *pending.resP = pathname return } + + pending.errsMu.Lock() + *pending.errs = append(*pending.errs, err) + pending.errsMu.Unlock() } -// New returns the address to a new instance of [Cache]. -func New(base *check.Absolute) (*Cache, error) { +// Close cancels all pending cures and waits for them to clean up. +func (c *Cache) Close() { c.cancel(); c.wg.Wait() } + +// New returns the address to a new instance of [Cache]. Concurrent cures of +// dependency [Artifact] is limited to the caller-supplied value, however direct +// calls to [Cache.Cure] is not subject to this limitation. +// +// A cures value of 0 or lower is equivalent to the value returned by +// [runtime.NumCPU]. +func New( + ctx context.Context, + msg message.Msg, + cures int, + base *check.Absolute, +) (*Cache, error) { for _, name := range []string{ dirIdentifier, dirChecksum, @@ -935,11 +1148,35 @@ func New(base *check.Absolute) (*Cache, error) { } } - return &Cache{ + c := Cache{ + msg: msg, base: base, ident: make(map[ID]Checksum), identErr: make(map[ID]error), identPending: make(map[ID]<-chan struct{}), - }, nil + } + c.ctx, c.cancel = context.WithCancel(ctx) + cureDep := make(chan *pendingArtifactDep, cures) + c.cureDep = cureDep + + if cures < 1 { + cures = runtime.NumCPU() + } + for i := 0; i < cures; i++ { + c.wg.Go(func() { + for { + select { + case <-c.ctx.Done(): + return + + case pending := <-cureDep: + pending.cure(&c) + break + } + } + }) + } + + return &c, nil } |