Fixed a race condition with mounts that caused them to fail

README.md

@@ -17,6 +17,4 @@ Thanks also to [distri's squashfs library](https://github.com/distr1/distri/tree
 
 ## Performance
 
-This library, decompressing the Firefox AppImage and using go tests, takes about twice as long as `unsquashfs` on my quad core laptop. (~1 second with the library and about half a second with `unsquashfs`).
+Testing on a zstd compressed file, my library is anywhere from 5x ~ 7x slower then `unsquashfs`
-
-**My recents tests have shown the Firefox AppImage might be an outlier and this library might be considerably slower (4x ~ 6x time slower then `unsquashfs`)**

fuse.go

@@ -3,6 +3,7 @@ package squashfs
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io"
 
 	"github.com/CalebQ42/fuse"
@@ -10,15 +11,38 @@ import (
 	"github.com/CalebQ42/squashfs/internal/inode"
 )
 
-func (r *Reader) Mount(mountpoint string) (con *fuse.Conn, err error) {
+// Mounts the archive to the given mountpoint using fuse3.
-	con, err = fuse.Mount(mountpoint, fuse.ReadOnly())
+// Blocks until the arhive is unmounted.
+// Hightly suggested to run in a goroutine.
+// Will take a moment before MountWait and Unmount will work correctly.
+func (r *Reader) Mount(mountpoint string) (err error) {
+	if r.con != nil {
+		return errors.New("squashfs archive already mounted")
+	}
+	r.con, err = fuse.Mount(mountpoint, fuse.ReadOnly())
 	if err != nil {
 		return
 	}
-	err = fs.Serve(con, &squashFuse{r: r})
+	err = fs.Serve(r.con, &squashFuse{r: r})
 	return
 }
 
+// Blocks until the mount ends.
+func (r *Reader) MountWait() {
+	if r.con != nil {
+		<-r.con.Ready
+	}
+}
+
+// Unmounts the archive.
+func (r *Reader) Unmount() error {
+	if r.con != nil {
+		defer func() { r.con = nil }()
+		return r.con.Close()
+	}
+	return errors.New("squashfs archive is not mounted")
+}
+
 type squashFuse struct {
 	r *Reader
 }

internal/data/fullreader.go

@@ -2,6 +2,7 @@ package data
 
 import (
 	"io"
+	"sync"
 
 	"github.com/CalebQ42/squashfs/internal/decompress"
 	"github.com/CalebQ42/squashfs/internal/toreader"
@@ -26,9 +27,9 @@ func NewFullReader(r io.ReaderAt, start uint64, d decompress.Decompressor, block
 	}
 }
 
-func (r *FullReader) AddFragment(rdr func() (io.Reader, error)) {
+func (r *FullReader) AddFragment(rdr func() (io.Reader, error), size uint32) {
 	r.fragRdr = rdr
-	r.sizes = append(r.sizes, 0)
+	r.sizes = append(r.sizes, size)
 }
 
 type outDat struct {
@@ -37,50 +38,50 @@ type outDat struct {
 	i    int
 }
 
-func (r FullReader) process(index int, offset int64, out chan outDat) {
+func (r FullReader) process(index int, offset int64, od *outDat, out chan *outDat) {
-	var err error
+	defer func() {
-	var dat []byte
+		out <- od
-	var rdr io.ReadCloser
+	}()
+	od.i = index
 	size := realSize(r.sizes[index])
 	if size == 0 {
-		out <- outDat{
+		od.err = nil
-			i:    index,
+		od.data = make([]byte, r.blockSize)
-			err:  nil,
-			data: make([]byte, r.blockSize),
-		}
 		return
 	}
-	// rdr := io.LimitReader(toreader.NewReader(r.r, offset), int64(size))
 	if size == r.sizes[index] {
-		//Special workaround for zstd for increased performancce.
+		if dec, ok := r.d.(decompress.Decoder); ok {
-		if zstd, ok := r.d.(*decompress.Zstd); ok {
+			dat := make([]byte, size)
-			dat = make([]byte, size)
+			_, od.err = r.r.ReadAt(dat, offset)
-			_, err = r.r.ReadAt(dat, offset)
+			if od.err != nil {
-			if err == nil {
+				return
-				dat, err = zstd.Decode(dat)
 			}
+			od.data, od.err = dec.Decode(dat, int(r.blockSize))
+			return
+		}
+		var rdr io.ReadCloser
+		rdr, od.err = r.d.Reader(io.LimitReader(toreader.NewReader(r.r, offset), int64(size)))
+		if od.err != nil {
+			return
+		}
+		od.data = make([]byte, r.blockSize)
+		var read int
+		read, od.err = rdr.Read(od.data)
+		od.data = od.data[:read]
+		rdr.Close()
 	} else {
-			rdr, err = r.d.Reader(io.LimitReader(toreader.NewReader(r.r, offset), int64(size)))
+		od.data = make([]byte, size)
-			if err == nil {
+		_, od.err = r.r.ReadAt(od.data, offset)
-				dat, err = io.ReadAll(rdr)
-			}
-		}
-	} else {
-		dat = make([]byte, size)
-		_, err = r.r.ReadAt(dat, offset)
-	}
-	out <- outDat{
-		i:    index,
-		err:  err,
-		data: dat,
-	}
-	if clr, ok := rdr.(io.Closer); ok {
-		clr.Close()
 	}
 }
 
 func (r FullReader) ReadAt(p []byte, off int64) (n int, err error) {
-	out := make(chan outDat, len(r.sizes))
+	pol := &sync.Pool{
+		New: func() any {
+			return new(outDat)
+		},
+	}
+	out := make(chan *outDat, len(r.sizes))
 	offset := r.start
 	num := len(r.sizes)
 	start := off / int64(r.blockSize)
@@ -100,40 +101,42 @@ func (r FullReader) ReadAt(p []byte, off int64) (n int, err error) {
 			offset += uint64(realSize(r.sizes[i]))
 			continue
 		}
+		od := pol.Get().(*outDat)
 		if i == num-1 && r.fragRdr != nil {
 			go func() {
+				defer func() {
+					out <- od
+				}()
 				rdr, e := r.fragRdr()
 				if err != nil {
-					out <- outDat{
+					od.i = num - 1
-						i:   num - 1,
+					od.err = e
-						err: e,
-					}
 					return
 				}
-				dat, e := io.ReadAll(rdr)
+				od.data = make([]byte, r.sizes[num-1])
-				out <- outDat{
+				_, e = rdr.Read(od.data)
-					i:    num - 1,
+				od.i = num - 1
-					err:  e,
+				od.err = e
-					data: dat,
-				}
 				if clr, ok := rdr.(io.Closer); ok {
 					clr.Close()
 				}
 			}()
 			continue
 		}
-		go r.process(i, int64(offset), out)
+		go r.process(i, int64(offset), od, out)
 		offset += uint64(realSize(r.sizes[i]))
 	}
+	cur := start
 	cache := make(map[int]outDat)
-	for cur := start; cur < int64(end); {
+	for dat := range out {
-		dat := <-out
 		if dat.err != nil {
 			err = dat.err
+			pol.Put(dat)
 			return
 		}
 		if dat.i != int(cur) {
-			cache[dat.i] = dat
+			cache[dat.i] = *dat
+			pol.Put(dat)
 			continue
 		}
 		if cur == start {
@@ -144,16 +147,18 @@ func (r FullReader) ReadAt(p []byte, off int64) (n int, err error) {
 		}
 		n += len(dat.data)
 		cur++
+		pol.Put(dat)
 		var ok bool
+		var curDat outDat
 		for {
-			dat, ok = cache[int(cur)]
+			curDat, ok = cache[int(cur)]
 			if !ok {
 				break
 			}
-			for i := range dat.data {
+			for i := range curDat.data {
-				p[n+i] = dat.data[i]
+				p[n+i] = curDat.data[i]
 			}
-			n += len(dat.data)
+			n += len(curDat.data)
 			cur++
 			delete(cache, int(cur))
 		}
@@ -165,45 +170,56 @@ func (r FullReader) ReadAt(p []byte, off int64) (n int, err error) {
 }
 
 func (r FullReader) WriteTo(w io.Writer) (n int64, err error) {
-	out := make(chan outDat, len(r.sizes))
+	pol := &sync.Pool{
+		New: func() any {
+			return new(outDat)
+		},
+	}
+	out := make(chan *outDat, len(r.sizes))
 	offset := r.start
 	num := len(r.sizes)
 	for i := 0; i < num; i++ {
+		od := pol.Get().(*outDat)
 		if i == num-1 && r.fragRdr != nil {
 			go func() {
+				defer func() {
+					out <- od
+				}()
 				rdr, e := r.fragRdr()
 				if err != nil {
-					out <- outDat{
+					od.i = num - 1
-						i:   num - 1,
+					od.err = e
-						err: e,
-					}
 					return
 				}
-				dat, e := io.ReadAll(rdr)
+				buf := make([]byte, r.sizes[num-1])
-				out <- outDat{
+				_, e = rdr.Read(buf)
-					i:    num - 1,
+				od.i = num - 1
-					err:  e,
+				od.err = e
-					data: dat,
+				od.data = buf
-				}
 				if clr, ok := rdr.(io.Closer); ok {
 					clr.Close()
 				}
 			}()
 			continue
 		}
-		go r.process(i, int64(offset), out)
+		go r.process(i, int64(offset), od, out)
 		offset += uint64(realSize(r.sizes[i]))
 	}
+	wt, ok := w.(io.WriterAt)
+	if !ok {
+		var cur int
 		cache := make(map[int]outDat)
 		var tmpN int
-	for cur := 0; cur < num; {
+		var dat *outDat
-		dat := <-out
+		for cur < len(r.sizes) {
+			dat = <-out
+			defer pol.Put(dat)
 			if dat.err != nil {
 				err = dat.err
 				return
 			}
 			if dat.i != cur {
-			cache[dat.i] = dat
+				cache[dat.i] = *dat
 				continue
 			}
 			tmpN, err = w.Write(dat.data)
@@ -213,12 +229,13 @@ func (r FullReader) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			cur++
 			var ok bool
+			var curDat outDat
 			for {
-			dat, ok = cache[cur]
+				curDat, ok = cache[cur]
 				if !ok {
 					break
 				}
-			tmpN, err = w.Write(dat.data)
+				tmpN, err = w.Write(curDat.data)
 				n += int64(tmpN)
 				if err != nil {
 					return
@@ -226,5 +243,22 @@ func (r FullReader) WriteTo(w io.Writer) (n int64, err error) {
 				cur++
 			}
 		}
+	} else {
+		var done int
+		var dat *outDat
+		for done < len(r.sizes) {
+			dat = <-out
+			defer pol.Put(dat)
+			if dat.err != nil {
+				err = dat.err
+				return
+			}
+			_, err = wt.WriteAt(dat.data, int64(dat.i*int(r.blockSize)))
+			if err != nil {
+				return
+			}
+			done++
+		}
+	}
 	return
 }

internal/data/reader.go

@@ -53,14 +53,14 @@ func (r *Reader) advance() (err error) {
 		} else {
 			r.cur = io.LimitReader(r.master, int64(size))
 			if size == r.blockSizes[0] {
-				if r.d.Resetable() {
+				if rs, ok := r.d.(decompress.Resetable); ok {
 					if r.comRdr == nil {
 						r.cur, err = r.d.Reader(r.cur)
 						if err != nil {
 							return
 						}
 					} else {
-						err = r.d.Reset(r.comRdr, r.cur)
+						err = rs.Reset(r.comRdr, r.cur)
 						r.cur = r.comRdr
 					}
 				} else {

internal/decompress/gzip.go

@@ -12,8 +12,6 @@ func (g GZip) Reader(src io.Reader) (io.ReadCloser, error) {
 	return zlib.NewReader(src)
 }
 
-func (g GZip) Resetable() bool { return true }
-
 func (g GZip) Reset(old, src io.Reader) error {
 	return old.(zlib.Resetter).Reset(src, nil)
 }

internal/decompress/interface.go

@@ -1,19 +1,22 @@
 package decompress
 
 import (
-	"errors"
 	"io"
 )
 
-var ErrNotResetable = errors.New("decompressor not resetable")
-
 type Decompressor interface {
 	//Creates a new decompressor reading from src.
 	Reader(src io.Reader) (io.ReadCloser, error)
-	//Reports whether Reset will work or not.
+}
-	Resetable() bool
+
+type Resetable interface {
 	//Reset attempts to re-use an old decompressor with new data.
 	//Will return ErrNotResetable if not Resetable().
 	//Must ALWAYS be provided with a reader created with Reader.
 	Reset(old, src io.Reader) error
 }
+
+type Decoder interface {
+	//Decodes a chunk of data all at once.
+	Decode(in []byte, outSize int) ([]byte, error)
+}

internal/decompress/lz4.go

@@ -12,9 +12,16 @@ func (l Lz4) Reader(r io.Reader) (io.ReadCloser, error) {
 	return io.NopCloser(lz4.NewReader(r)), nil
 }
 
-func (l Lz4) Resetable() bool { return true }
-
 func (l Lz4) Reset(old, src io.Reader) error {
 	old.(*lz4.Reader).Reset(src)
 	return nil
 }
+
+func (l Lz4) Decode(in []byte, outSize int) (out []byte, err error) {
+	out = make([]byte, outSize)
+	outLen, err := lz4.UncompressBlock(in, out)
+	if outLen < outSize {
+		out = out[:outLen]
+	}
+	return
+}

internal/decompress/lzma.go

@@ -12,7 +12,3 @@ func (l Lzma) Reader(r io.Reader) (io.ReadCloser, error) {
 	rdr, err := lzma.NewReader(r)
 	return io.NopCloser(rdr), err
 }
-
-func (l Lzma) Resetable() bool { return false }
-
-func (l Lzma) Reset(old, src io.Reader) error { return ErrNotResetable }

internal/decompress/lzo.go

@@ -16,7 +16,3 @@ func (l Lzo) Reader(r io.Reader) (io.ReadCloser, error) {
 	}
 	return io.NopCloser(bytes.NewReader(cache)), nil
 }
-
-func (l Lzo) Resetable() bool { return false }
-
-func (l Lzo) Reset(old, src io.Reader) error { return ErrNotResetable }

internal/decompress/xz.go

@@ -13,8 +13,6 @@ func (x Xz) Reader(r io.Reader) (io.ReadCloser, error) {
 	return io.NopCloser(rdr), err
 }
 
-func (x Xz) Resetable() bool { return true }
-
 func (x Xz) Reset(old, src io.Reader) error {
 	return old.(*xz.Reader).Reset(src)
 }

internal/decompress/zstd.go

@@ -15,15 +15,13 @@ func (z Zstd) Reader(src io.Reader) (io.ReadCloser, error) {
 	return r.IOReadCloser(), err
 }
 
-func (z Zstd) Resetable() bool { return true }
-
 func (z Zstd) Reset(old, src io.Reader) error {
 	return old.(*zstd.Decoder).Reset(src)
 }
 
-func (z *Zstd) Decode(in []byte) (out []byte, err error) {
+func (z Zstd) Decode(in []byte, outSize int) ([]byte, error) {
 	if z.writeToReader == nil {
 		z.writeToReader, _ = zstd.NewReader(nil)
 	}
-	return z.writeToReader.DecodeAll(in, nil)
+	return z.writeToReader.DecodeAll(in, make([]byte, outSize))
 }

internal/metadata/reader.go

@@ -26,7 +26,7 @@ func realSize(siz uint16) uint16 {
 }
 
 func (r *Reader) advance() (err error) {
-	if !r.d.Resetable() {
+	if _, ok := r.d.(decompress.Resetable); !ok {
 		if clr, ok := r.cur.(io.Closer); ok {
 			clr.Close()
 		}
@@ -39,14 +39,14 @@ func (r *Reader) advance() (err error) {
 	size := realSize(raw)
 	r.cur = io.LimitReader(r.master, int64(size))
 	if size == raw {
-		if r.d.Resetable() {
+		if rs, ok := r.d.(decompress.Resetable); ok {
 			if r.comRdr == nil {
 				r.cur, err = r.d.Reader(r.cur)
 				if err != nil {
 					return
 				}
 			} else {
-				err = r.d.Reset(r.comRdr, r.cur)
+				err = rs.Reset(r.comRdr, r.cur)
 				r.cur = r.comRdr
 			}
 		} else {

internal/toreader/offsetreader.go

@@ -0,0 +1,19 @@
+package toreader
+
+import "io"
+
+type OffsetReader struct {
+	r   io.ReaderAt
+	off int64
+}
+
+func NewOffsetReader(r io.ReaderAt, off int64) *OffsetReader {
+	return &OffsetReader{
+		r:   r,
+		off: off,
+	}
+}
+
+func (r OffsetReader) ReadAt(p []byte, off int64) (n int, e error) {
+	return r.r.ReadAt(p, off+r.off)
+}

internal/toreader/readerat.go

@@ -6,7 +6,8 @@ type ReaderAt struct {
 	d []byte
 }
 
-func NewReaderAt(r io.Reader) (ra ReaderAt, err error) {
+func NewReaderAt(r io.Reader) (ra *ReaderAt, err error) {
+	ra = new(ReaderAt)
 	ra.d, err = io.ReadAll(r)
 	return
 }

reader.go

@@ -7,6 +7,7 @@ import (
 	"math"
 	"time"
 
+	"github.com/CalebQ42/fuse"
 	"github.com/CalebQ42/squashfs/internal/decompress"
 	"github.com/CalebQ42/squashfs/internal/directory"
 	"github.com/CalebQ42/squashfs/internal/inode"
@@ -16,6 +17,7 @@ import (
 
 type Reader struct {
 	*FS
+	con         *fuse.Conn
 	d           decompress.Decompressor
 	r           io.ReaderAt
 	fragEntries []fragEntry
@@ -40,6 +42,10 @@ const (
 	ZSTDCompression
 )
 
+func NewReaderAtOffset(r io.ReaderAt, off int64) (*Reader, error) {
+	return NewReader(toreader.NewOffsetReader(r, off))
+}
+
 // Creates a new squashfs.Reader from the given io.Reader. NOTE: All data from the io.Reader will be read and stored in memory.
 func NewReaderFromReader(r io.Reader) (*Reader, error) {
 	rdr, err := toreader.NewReaderAt(r)

reader_inode.go

@@ -70,7 +70,7 @@ func (r Reader) getReaders(i inode.Inode) (full *data.FullReader, rdr *data.Read
 			}
 			fragRdr = io.LimitReader(fragRdr, int64(fragSize))
 			return fragRdr, nil
-		})
+		}, fragSize)
 		var fragRdr io.Reader
 		fragRdr, err = r.fragReader(fragInd)
 		if err != nil {

squashfs_test.go

@@ -73,7 +73,6 @@ func TestMisc(t *testing.T) {
 }
 
 func BenchmarkRace(b *testing.B) {
-	// tmpDir := b.TempDir()
 	tmpDir := "testing"
 	fil, err := preTest(tmpDir)
 	if err != nil {
@@ -197,11 +196,11 @@ func TestFuse(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	con, err := rdr.Mount("testing/fuseTest")
+	err = rdr.Mount("testing/fuseTest")
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer con.Close()
+	defer rdr.Unmount()
-	<-con.Ready
+	rdr.MountWait()
 	t.Fatal("testing")
 }