ston1th/cachefs
0
0
Fork 0
Code Issues Releases Activity

fix preload alignment for custom buffer sizes

Browse source
This commit is contained in:
ston1th 2022-04-07 16:13:08 +02:00
commit cb7f2dee54
6 changed files with 12 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

3
TODO.txt
View file

@ -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

12
pkg/fs/file.go
View file

@ -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 {

26
pkg/provider/crypto/file.go
View file

@ -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

3
pkg/provider/crypto/reader.go
View file

@ -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
}

6
pkg/provider/crypto/writer.go
View file

@ -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

8
pkg/provider/sftp/sftp.go
View file

@ -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)