From e1eecae6d3a9d247c706a6a7e5da629c7294126f Mon Sep 17 00:00:00 2001 From: sawka Date: Thu, 18 Aug 2022 21:54:42 -0700 Subject: [PATCH] implement WriteAt, refactored and tested append, writeat still needs testing --- pkg/cirfile/cirfile.go | 175 +++++++++++++++++++++++++++--------- pkg/cirfile/cirfile_test.go | 8 +- 2 files changed, 133 insertions(+), 50 deletions(-) diff --git a/pkg/cirfile/cirfile.go b/pkg/cirfile/cirfile.go index d40fdceb..5992b876 100644 --- a/pkg/cirfile/cirfile.go +++ b/pkg/cirfile/cirfile.go @@ -20,6 +20,7 @@ const InitialLockDelay = 10 * time.Millisecond const InitialLockTries = 5 const LockDelay = 100 * time.Millisecond +// File objects are *not* multithread safe, operations must be externally synchronized type File struct { OSFile *os.File Version byte @@ -72,8 +73,10 @@ func (f *File) flock(ctx context.Context, lockType int) error { } func (f *File) unflock() { - syscall.Flock(int(f.OSFile.Fd()), syscall.LOCK_UN) // ignore error (nothing to do about it anyway) - f.FlockStatus = 0 + if f.FlockStatus != 0 { + syscall.Flock(int(f.OSFile.Fd()), syscall.LOCK_UN) // ignore error (nothing to do about it anyway) + f.FlockStatus = 0 + } return } @@ -127,6 +130,10 @@ func CreateCirFile(fileName string, maxSize int64) (*File, error) { return rtn, nil } +func (f *File) Close() error { + return f.OSFile.Close() +} + func (f *File) ReadMeta(ctx context.Context) error { err := f.flock(ctx, syscall.LOCK_SH) if err != nil { @@ -341,16 +348,13 @@ func (f *File) ReadNext(ctx context.Context, buf []byte, offset int64) (int64, i } func (f *File) ensureFreeSpace(requiredSpace int64) error { - if f.StartPos == FilePosEmpty { - return nil - } chunks := f.getFileChunks() curSpace := f.MaxSize - totalChunksSize(chunks) if curSpace >= requiredSpace { return nil } neededSpace := requiredSpace - curSpace - if requiredSpace >= f.MaxSize { + if requiredSpace >= f.MaxSize || f.StartPos == FilePosEmpty { f.StartPos = FilePosEmpty f.EndPos = 0 f.FileOffset += neededSpace @@ -361,6 +365,126 @@ func (f *File) ensureFreeSpace(requiredSpace int64) error { return f.writeMeta() } +// does not implement io.WriterAt (needs context) +func (f *File) WriteAt(ctx context.Context, buf []byte, writePos int64) error { + if writePos < 0 { + return fmt.Errorf("WriteAt got invalid writePos[%d]", writePos) + } + err := f.flock(ctx, syscall.LOCK_EX) + if err != nil { + return err + } + defer f.unflock() + err = f.readMeta() + if err != nil { + return err + } + chunks := f.getFileChunks() + currentSize := totalChunksSize(chunks) + if writePos < f.FileOffset { + negOffset := f.FileOffset - writePos + if negOffset >= int64(len(buf)) { + return nil + } + buf = buf[negOffset:] + writePos = f.FileOffset + } + if writePos > f.FileOffset+currentSize { + // fill gap with zero bytes + posOffset := writePos - (f.FileOffset + currentSize) + err = f.ensureFreeSpace(int64(posOffset)) + if err != nil { + return err + } + var zeroBuf []byte + if posOffset >= f.MaxSize { + zeroBuf = make([]byte, f.MaxSize) + } else { + zeroBuf = make([]byte, posOffset) + } + err = f.internalAppendData(zeroBuf) + if err != nil { + return err + } + // recalc chunks/currentSize + chunks = f.getFileChunks() + currentSize = totalChunksSize(chunks) + // after writing the zero bytes, writePos == f.FileOffset+currentSize (the rest is a straight append) + } + // now writePos >= f.FileOffset && writePos <= f.FileOffset+currentSize (check invariant) + if writePos < f.FileOffset || writePos > f.FileOffset+currentSize { + panic(fmt.Sprintf("invalid writePos, invariant violated writepos[%d] fileoffset[%d] currentsize[%d]", writePos, f.FileOffset, currentSize)) + } + // overwrite existing data (in chunks). advance by writePosOffset + writePosOffset := writePos - f.FileOffset + if writePosOffset < currentSize { + advChunks := advanceChunks(chunks, writePosOffset) + nw, err := f.writeToChunks(buf, advChunks, false) + if err != nil { + return err + } + buf = buf[nw:] + if len(buf) == 0 { + return nil + } + } + // buf contains what was unwritten. this unwritten data is now just a straight append + return f.internalAppendData(buf) +} + +// try writing to chunks, returns (nw, error) +func (f *File) writeToChunks(buf []byte, chunks []fileChunk, updatePos bool) (int64, error) { + var numWrite int64 + for _, chunk := range chunks { + if numWrite >= int64(len(buf)) { + break + } + if chunk.Len == 0 { + continue + } + toWrite := int64(len(buf)) - numWrite + if toWrite > chunk.Len { + toWrite = chunk.Len + } + nw, err := f.OSFile.WriteAt(buf[numWrite:numWrite+toWrite], chunk.StartPos+HeaderLen) + if err != nil { + return 0, err + } + if updatePos { + if chunk.StartPos+int64(nw) > f.FileDataSize { + f.FileDataSize = chunk.StartPos + int64(nw) + } + if f.StartPos == FilePosEmpty { + f.StartPos = chunk.StartPos + } + f.EndPos = chunk.StartPos + int64(nw) - 1 + } + numWrite += int64(nw) + } + return numWrite, nil +} + +func (f *File) internalAppendData(buf []byte) error { + err := f.ensureFreeSpace(int64(len(buf))) + if err != nil { + return err + } + if len(buf) >= int(f.MaxSize) { + buf = buf[len(buf)-int(f.MaxSize):] + } + chunks := f.getFreeChunks() + // don't track nw because we know we have enough free space to write entire buf + _, err = f.writeToChunks(buf, chunks, true) + if err != nil { + return err + } + err = f.writeMeta() + if err != nil { + return err + } + return nil +} + func (f *File) AppendData(ctx context.Context, buf []byte) error { err := f.flock(ctx, syscall.LOCK_EX) if err != nil { @@ -371,42 +495,5 @@ func (f *File) AppendData(ctx context.Context, buf []byte) error { if err != nil { return err } - err = f.ensureFreeSpace(int64(len(buf))) - if err != nil { - return err - } - if len(buf) >= int(f.MaxSize) { - buf = buf[len(buf)-int(f.MaxSize):] - } - chunks := f.getFreeChunks() - numWrite := 0 - for _, chunk := range chunks { - if numWrite >= len(buf) { - break - } - if chunk.Len == 0 { - continue - } - toRead := len(buf) - numWrite - if toRead > int(chunk.Len) { - toRead = int(chunk.Len) - } - nw, err := f.OSFile.WriteAt(buf[numWrite:numWrite+toRead], chunk.StartPos+HeaderLen) - if err != nil { - return err - } - if chunk.StartPos+int64(nw) > f.FileDataSize { - f.FileDataSize = chunk.StartPos + int64(nw) - } - if f.StartPos == FilePosEmpty { - f.StartPos = chunk.StartPos - } - f.EndPos = chunk.StartPos + int64(nw) - 1 - numWrite += nw - } - err = f.writeMeta() - if err != nil { - return err - } - return nil + return f.internalAppendData(buf) } diff --git a/pkg/cirfile/cirfile_test.go b/pkg/cirfile/cirfile_test.go index 4e4f555c..1e655470 100644 --- a/pkg/cirfile/cirfile_test.go +++ b/pkg/cirfile/cirfile_test.go @@ -96,17 +96,13 @@ func TestFile(t *testing.T) { t.Fatalf("cannot append data: %v", err) } validateFileSize(t, f1Name, HeaderLen) - if f.StartPos != FilePosEmpty || f.EndPos != 0 || f.FileDataSize != 0 { - t.Fatalf("metadata error (1): %#v", f) - } + validateMeta(t, "1", f, FilePosEmpty, 0, 0, 0) err = f.AppendData(context.Background(), []byte("hello")) if err != nil { t.Fatalf("cannot append data: %v", err) } validateFileSize(t, f1Name, HeaderLen+5) - if f.StartPos != 0 || f.EndPos != 4 || f.FileDataSize != 5 { - t.Fatalf("metadata error (2): %#v", f) - } + validateMeta(t, "2", f, 0, 4, 5, 0) err = f.AppendData(context.Background(), []byte(" foo")) if err != nil { t.Fatalf("cannot append data: %v", err)