From cb7f2dee547df319e89ebd54ef77abb088050d45 Mon Sep 17 00:00:00 2001 From: ston1th Date: Thu, 7 Apr 2022 16:13:08 +0200 Subject: [PATCH] fix preload alignment for custom buffer sizes --- TODO.txt | 3 +-- pkg/fs/file.go | 12 +++++------- pkg/provider/crypto/file.go | 26 +++----------------------- pkg/provider/crypto/reader.go | 3 --- pkg/provider/crypto/writer.go | 6 ------ pkg/provider/sftp/sftp.go | 8 +++----- 6 files changed, 12 insertions(+), 46 deletions(-) diff --git a/TODO.txt b/TODO.txt index 90a2f37..a53ee57 100644 --- a/TODO.txt +++ b/TODO.txt @@ -3,10 +3,9 @@ * crypto * flush last chunk in streaming mode - * curl download does not finish - missing EOF?? * sftp - * implement read ahead buffer + * implement read ahead buffer? * add breadcrumb * add copyright footer diff --git a/pkg/fs/file.go b/pkg/fs/file.go index 4cb7380..8cd7342 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -43,6 +43,9 @@ func (p *preload) Read(data []byte) (n int, err error) { return } n, err = p.f.readToCache(data) + if n < len(data) { + _, err = p.f.Seek(p.f.offset, io.SeekStart) + } p.written++ return } @@ -60,11 +63,6 @@ func (f *File) Preload(ctx context.Context, unlock func()) { return } - // TODO - //fi, _ := f.Stat() - //s := fi.Size() - //log.V(2).Info("preload started", "size", s, "crypto.Size", crypto.Size(s), "crypto.RealSize", crypto.RealSize(crypto.Size(s))) - log.V(2).Info("preload started") p := &preload{f: f, ctx: ctx} _, err := io.Copy(Discard, p) @@ -128,9 +126,9 @@ func (f *File) readToCache(p []byte) (n int, err error) { log := f.log n, err = f.readSource(p) if n > 0 { - if n, err := f.md.WriteAt(p[:n], f.offset); err != nil { + if n, err = f.md.WriteAt(p[:n], f.offset); err != nil { log.Error(err, "error writing cache file") - return n, err + return } } if err != nil && err != io.EOF { diff --git a/pkg/provider/crypto/file.go b/pkg/provider/crypto/file.go index 8e58b42..027132f 100644 --- a/pkg/provider/crypto/file.go +++ b/pkg/provider/crypto/file.go @@ -64,28 +64,12 @@ func (f *file) Stat() (fs.FileInfo, error) { if f.dir { return dir{FileInfo: fi, name: f.name}, nil } - - rs := RealSize(fi.Size()) f.stat = &stat{ FileInfo: fi, - size: rs, + size: RealSize(fi.Size()), name: f.name, } return f.stat, nil - - //fmt.Println("rs", rs) - //_, err = f.ReadAt(nil, rs) - //if err != nil && err != io.EOF { - // return nil, err - //} - //f.stat = &stat{ - // FileInfo: fi, - // size: rs - (ChunkSize - int64(len(f.r.unread))), - // name: f.name, - //} - //fmt.Println("size", f.stat.Size()) - //f.Seek(0, io.SeekStart) - //return f.stat, nil } func (f *file) Name() string { @@ -146,7 +130,8 @@ func (f *file) Seek(offset int64, whence int) (n int64, err error) { _, err = f.File.Seek(f.offset, io.SeekStart) n = offset } - f.reset() + f.r.err = nil + f.r.unread = f.r.unread[:0] if err != nil { return } @@ -158,11 +143,6 @@ func (f *file) Seek(offset int64, whence int) (n int64, err error) { return } -func (f *file) reset() { - f.r.err = nil - f.r.unread = f.r.unread[:0] -} - func (f *file) Read(p []byte) (n int, err error) { n, err = f.r.Read(p) return diff --git a/pkg/provider/crypto/reader.go b/pkg/provider/crypto/reader.go index ba89ea8..4262aa6 100644 --- a/pkg/provider/crypto/reader.go +++ b/pkg/provider/crypto/reader.go @@ -3,7 +3,6 @@ package crypto import ( "crypto/cipher" "errors" - "fmt" "io" "golang.org/x/crypto/chacha20poly1305" @@ -87,9 +86,7 @@ func (r *reader) readChunk() (last bool, err error) { // The last chunk can be short. in = in[:n] last = true - fmt.Println("last nonce", r.nonce) setLastChunkFlag(&r.nonce) - fmt.Println("last nonce flag", r.nonce) case err != nil: return false, err } diff --git a/pkg/provider/crypto/writer.go b/pkg/provider/crypto/writer.go index 1efa359..c52d57a 100644 --- a/pkg/provider/crypto/writer.go +++ b/pkg/provider/crypto/writer.go @@ -4,7 +4,6 @@ import ( "cachefs/pkg/chunk" "crypto/cipher" "errors" - "fmt" "io" "golang.org/x/crypto/chacha20poly1305" @@ -72,9 +71,7 @@ func (c *chunkWriter) full() bool { func (c *chunkWriter) flush(last bool) (n int, err error) { wb := c.buf[:0] if last { - fmt.Println("last nonce", c.nonce) setLastChunkFlag(&c.nonce) - fmt.Println("last nonce flag", c.nonce) s := c.chunks.Size() wb = c.buf[:s] n = int(s) @@ -115,16 +112,13 @@ func (w *writer) Write(p []byte) (n int, err error) { } func (w *writer) Close() error { - fmt.Println("writer closed", w.err) if w.err != nil { return w.err } if w.last != nil { - fmt.Println("flushing last chunk") _, w.err = w.last.flush(lastChunk) if w.err != nil { - fmt.Println("error flushing last chunk", w.err) return w.err } w.flush = false diff --git a/pkg/provider/sftp/sftp.go b/pkg/provider/sftp/sftp.go index 07397b5..0f0f357 100644 --- a/pkg/provider/sftp/sftp.go +++ b/pkg/provider/sftp/sftp.go @@ -92,8 +92,6 @@ func (fs *FS) dial() (err error) { func (fs *FS) connect(ctx context.Context) { log := fs.log - t := time.NewTicker(time.Second * 2) - defer t.Stop() for { log.V(2).Info("connecting") err := fs.dial() @@ -106,8 +104,7 @@ func (fs *FS) connect(ctx context.Context) { fs.client, err = sftp.NewClient(fs.c, sftp.UseFstat(true), sftp.UseConcurrentReads(true), - //sftp.UseConcurrentWrites(true), - //sftp.MaxPacket(23552), + sftp.UseConcurrentWrites(true), ) select { case <-ctx.Done(): @@ -124,7 +121,7 @@ func (fs *FS) connect(ctx context.Context) { select { case <-ctx.Done(): return - case <-t.C: + case <-time.After(time.Second * 2): } } } @@ -142,6 +139,7 @@ func (fs *FS) wait(ctx context.Context) <-chan error { func (fs *FS) keepAlive(ctx context.Context) { t := time.NewTicker(time.Second * 15) + defer t.Stop() for { if fs.c != nil { _, _, err := fs.c.SendRequest("keepalive@cachefs", true, nil)