core: object chunks; chunk manifest

* lz4-compress chunk manifest with trailing xxhash checksum
* each chunk entry now includes its own path
* remove byte-packer; use SGL buffers instead
* set maximum manifest size to 64K (vs. LOM's 4K)
* part two, prev. commit: c413336

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

cmn/cos/unsafe.go

@@ -18,6 +18,8 @@ const (
 	SizeofI64 = int(unsafe.Sizeof(uint64(0)))
 	SizeofI32 = int(unsafe.Sizeof(uint32(0)))
 	SizeofI16 = int(unsafe.Sizeof(uint16(0)))
+
+	SizeXXHash64 = SizeofI64
 )
 
 // Unsafe cast (string => []byte) and ([]byte => string)

cmn/jsp/io.go

@@ -20,7 +20,6 @@ import (
 )
 
 const (
-	sizeXXHash64  = cos.SizeofI64
 	lz4BufferSize = lz4.Block64Kb
 )
 
@@ -57,7 +56,7 @@ func Encode(ws cos.WriterAt, v any, opts Options) error {
 		debug.Assert(off == prefLen)
 	}
 	if opts.Checksum {
-		var cksum [sizeXXHash64]byte
+		var cksum [cos.SizeXXHash64]byte
 		w.Write(cksum[:]) // reserve for checksum
 	}
 	if opts.Compress {
@@ -69,7 +68,7 @@ func Encode(ws cos.WriterAt, v any, opts Options) error {
 	}
 	if opts.Checksum {
 		h = onexxh.New64()
-		cos.Assert(h.Size() == sizeXXHash64)
+		cos.Assert(h.Size() == cos.SizeXXHash64)
 		w = io.MultiWriter(h, w)
 	}
 
@@ -136,7 +135,7 @@ func Decode(r io.Reader, v any, opts Options, tag string) (*cos.Cksum, error) {
 }
 
 func _decksum(r io.Reader, v any, opts Options, tag string) (*cos.Cksum, error) {
-	var cksum [sizeXXHash64]byte
+	var cksum [cos.SizeXXHash64]byte
 	if _, err := r.Read(cksum[:]); err != nil {
 		return nil, err
 	}

core/lchunk.go

@@ -5,14 +5,20 @@
 package core
 
 import (
+	"bytes"
+	"encoding/binary"
+	"errors"
 	"fmt"
+	"io"
 	"os"
 
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/fs"
+	"github.com/NVIDIA/aistore/memsys"
 
 	onexxh "github.com/OneOfOne/xxhash"
+	"github.com/pierrec/lz4/v4"
 )
 
 const (
@@ -22,148 +28,245 @@ const (
 type (
 	Uchunk struct {
 		Siz      int64
+		Path     string // (may become v2 _location_)
 		CksumVal string
 	}
 	Ufest struct {
-		Size     int64
 		Num      uint16
 		CksumTyp string
 		Chunks   []Uchunk
+		// runtime state
+		Size int64
 	}
 )
 
 // on-disk xattr
 const (
 	xattrChunk = "user.ais.chunk"
-)
 
-const (
-	utag = "chunk-manifest"
+	xattrChunkDflt = memsys.DefaultBufSize
+	xattrChunkMax  = memsys.DefaultBuf2Size // maximum 64K
 )
 
-// interface guard
-var (
-	_ cos.Unpacker = (*Ufest)(nil)
-	_ cos.Packer   = (*Ufest)(nil)
+const (
+	utag     = "chunk-manifest"
+	itag     = "invalid " + utag
+	tooShort = "failed to unpack: too short"
 )
 
-// TODO -- FIXME: 4K only; reuse memsys
 func (u *Ufest) Load(lom *LOM) error {
-	b, err := fs.GetXattr(lom.FQN, xattrChunk)
+	cbuf, cslab := g.pmm.AllocSize(xattrChunkMax)
+	defer cslab.Free(cbuf)
+
+	data, err := lom.getXchunk(cbuf)
 	if err != nil {
 		return _ucerr(lom.Cname(), err)
 	}
-	unpacker := cos.NewUnpacker(b)
-	if err := unpacker.ReadAny(u); err != nil {
+
+	// validate and strip/remove the trailing checksum
+	checksumOffset := len(data) - cos.SizeXXHash64
+	expectedChecksum := binary.BigEndian.Uint64(data[checksumOffset:])
+	compressedData := data[:checksumOffset]
+	actualChecksum := onexxh.Checksum64S(compressedData, cos.MLCG32)
+	if expectedChecksum != actualChecksum {
+		return cos.NewErrMetaCksum(expectedChecksum, actualChecksum, utag)
+	}
+
+	// decompress into a 2nd buffer
+	dbuf, dslab := g.pmm.AllocSize(xattrChunkMax)
+	err = u._load(lom, compressedData, dbuf)
+	dslab.Free(dbuf)
+	return err
+}
+
+func (u *Ufest) _load(lom *LOM, compressedData, buf []byte) error {
+	data := buf[:0]
+	zr := lz4.NewReader(bytes.NewReader(compressedData))
+
+	for {
+		if len(data) == cap(data) {
+			return fmt.Errorf("%s too large", utag)
+		}
+		n, err := zr.Read(data[len(data):cap(data)])
+		data = data[:len(data)+n]
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return fmt.Errorf("%s decompression failed: %v", utag, err)
+		}
+	}
+
+	if err := u.unpack(data); err != nil {
 		return _ucerr(lom.Cname(), err)
 	}
+	if u.Size != lom.Lsize(true) {
+		return fmt.Errorf("%s load size mismatch: %d vs %d", _utag(lom.Cname()), u.Size, lom.Lsize(true))
+	}
 	return nil
 }
 
 func _ucerr(cname string, err error) error {
-	tag := utag + "[" + cname + "]"
 	if cos.IsErrXattrNotFound(err) {
-		return cmn.NewErrLmetaNotFound(tag, err)
+		return cmn.NewErrLmetaNotFound(_utag(cname), err)
 	}
-	return os.NewSyscallError(getxattr, fmt.Errorf("%s, err: %w", tag, err))
+	return os.NewSyscallError(getxattr, fmt.Errorf("%s, err: %w", _utag(cname), err))
 }
 
+func _utag(cname string) string { return itag + "[" + cname + "]" }
+
 func (u *Ufest) Store(lom *LOM) error {
 	// validate
+	var total int64
 	if u.Num == 0 || int(u.Num) != len(u.Chunks) {
-		return fmt.Errorf("%s[%s]: invalid manifest num=%d, len=%d", utag, lom.Cname(), u.Num, len(u.Chunks))
+		return fmt.Errorf("%s: num %d vs %d", _utag(lom.Cname()), u.Num, len(u.Chunks))
+	}
+	lsize := lom.Lsize(true)
+	if u.Size != 0 && u.Size != lsize {
+		return fmt.Errorf("%s store size: %d vs %d", _utag(lom.Cname()), u.Size, lsize)
 	}
-	var total int64
 	for i := range u.Num {
 		total += u.Chunks[i].Siz
 	}
-	if u.Size == 0 {
-		u.Size = total
-	} else if total != u.Size {
-		return fmt.Errorf("%s total size mismatch: %d vs %d", utag, total, u.Size)
+	if total != lsize {
+		return fmt.Errorf("%s: total size mismatch (%d vs %d)", _utag(lom.Cname()), total, lsize)
 	}
 
 	// pack
-	psize := u.PackedSize()
-	if psize > xattrMaxSize { // TODO -- FIXME: see 4K above
-		return fmt.Errorf("%s[%s]: manifest too large (%d > %d)", utag, lom.Cname(), psize, xattrMaxSize)
-	}
+	sgl := g.pmm.NewSGL(xattrChunkDflt)
+	defer sgl.Free()
+
+	zw := lz4.NewWriter(sgl)
+	u.pack(zw)
+	zw.Close()
+
+	// compute and write trailing checksum
+	data := sgl.Bytes()
+	h := onexxh.Checksum64S(data, cos.MLCG32)
+	var checksumBuf [cos.SizeXXHash64]byte
+	binary.BigEndian.PutUint64(checksumBuf[:], h)
+	sgl.Write(checksumBuf[:])
 
-	packer := cos.NewPacker(nil, psize)
-	u.Pack(packer)
+	if sgl.Len() > xattrChunkMax {
+		return fmt.Errorf("%s: too large (%d > %d max)", _utag(lom.Cname()), sgl.Len(), xattrChunkMax)
+	}
 
 	// write
-	if err := fs.SetXattr(lom.FQN, xattrChunk, packer.Bytes()); err != nil {
+	if err := lom.setXchunk(sgl.Bytes()); err != nil {
 		return err
 	}
-
 	lom.md.lid.setlmfl(lmflChunk) // TODO -- FIXME: persist
 	return nil
 }
 
-func (u *Ufest) Pack(packer *cos.BytePack) {
-	packer.WriteByte(umetaver)
-	packer.WriteInt64(u.Size)
-	packer.WriteUint16(u.Num)
-	packer.WriteString(u.CksumTyp)
+func (u *Ufest) pack(w io.Writer) {
+	// meta-version
+	w.Write([]byte{umetaver})
+
+	// number of chunks
+	var buf [cos.SizeofI16]byte
+	binary.BigEndian.PutUint16(buf[:], u.Num)
+	w.Write(buf[:])
+
+	// checksum type
+	_packStr(w, u.CksumTyp)
+
+	// chunks
 	for _, c := range u.Chunks {
-		packer.WriteInt64(c.Siz)
-		packer.WriteString(c.CksumVal)
+		// chunk size
+		var sizeBuf [cos.SizeofI64]byte
+		binary.BigEndian.PutUint64(sizeBuf[:], uint64(c.Siz))
+		w.Write(sizeBuf[:])
+
+		// chunk path
+		_packStr(w, c.Path)
+
+		// chunk checksum
+		_packStr(w, c.CksumVal)
 	}
-	h := onexxh.Checksum64S(packer.Bytes(), cos.MLCG32)
-	packer.WriteUint64(h)
 }
 
-func (u *Ufest) PackedSize() (ll int) {
-	for _, c := range u.Chunks {
-		ll += cos.SizeofI64 + cos.PackedStrLen(c.CksumVal)
-	}
-	return ll + 1 + cos.SizeofI64 + cos.SizeofI16 + cos.PackedStrLen(u.CksumTyp) + cos.SizeofI64
+func _packStr(w io.Writer, s string) {
+	var lenBuf [cos.SizeofI16]byte
+	binary.BigEndian.PutUint16(lenBuf[:], uint16(len(s)))
+	w.Write(lenBuf[:])
+	w.Write(cos.UnsafeB(s))
 }
 
-func (u *Ufest) Unpack(unpacker *cos.ByteUnpack) (err error) {
-	var metaver byte
-	if metaver, err = unpacker.ReadByte(); err != nil {
-		return fmt.Errorf("invalid %s (too short)", utag)
+func (u *Ufest) unpack(data []byte) (err error) {
+	if len(data) < 1 {
+		return errors.New(tooShort)
 	}
+
+	var offset int
+
+	// meta-version
+	metaver := data[offset]
+	offset++
 	if metaver != umetaver {
 		return fmt.Errorf("unsupported %s meta-version %d (expecting %d)", utag, metaver, umetaver)
 	}
-	if u.Size, err = unpacker.ReadInt64(); err != nil {
-		return fmt.Errorf("invalid %s size: %v", utag, err)
-	}
-	if u.Num, err = unpacker.ReadUint16(); err != nil {
-		return fmt.Errorf("invalid %s num: %v", utag, err)
+
+	// number of chunks
+	if len(data) < offset+cos.SizeofI16 {
+		return errors.New(tooShort)
 	}
-	if u.CksumTyp, err = unpacker.ReadString(); err != nil {
-		return fmt.Errorf("invalid %s cksum type: %v", utag, err)
+	u.Num = binary.BigEndian.Uint16(data[offset:])
+	offset += cos.SizeofI16
+
+	// checksum type
+	if u.CksumTyp, offset, err = _unpackStr(data, offset); err != nil {
+		return err
 	}
 
+	// Read chunks
 	u.Chunks = make([]Uchunk, u.Num)
-	var total int64
+	u.Size = 0
 	for i := range u.Num {
 		c := &u.Chunks[i]
-		if c.Siz, err = unpacker.ReadInt64(); err != nil {
-			return fmt.Errorf("invalid %s chunk idx %d size: %v", utag, i, err)
+
+		// chunk size
+		if len(data) < offset+cos.SizeofI64 {
+			return errors.New(tooShort)
 		}
-		if c.CksumVal, err = unpacker.ReadString(); err != nil {
-			return fmt.Errorf("invalid %s chunk idx %d cksum val: %v", utag, i, err)
+		c.Siz = int64(binary.BigEndian.Uint64(data[offset:]))
+		offset += cos.SizeofI64
+		u.Size += c.Siz
+		// chunk path
+		if c.Path, offset, err = _unpackStr(data, offset); err != nil {
+			return err
+		}
+		// chunk checksum
+		if c.CksumVal, offset, err = _unpackStr(data, offset); err != nil {
+			return err
 		}
-		total += c.Siz
-	}
-	if total != u.Size {
-		return fmt.Errorf("%s total size mismatch: %d vs %d", utag, total, u.Size)
 	}
+	return nil
+}
 
-	var h uint64
-	if h, err = unpacker.ReadUint64(); err != nil {
-		return fmt.Errorf("%s corrupted: %v", utag, err)
+func _unpackStr(data []byte, offset int) (string, int, error) {
+	if len(data) < offset+cos.SizeofI16 {
+		return "", offset, errors.New(tooShort)
 	}
-	b := unpacker.Bytes()
-	hh := onexxh.Checksum64S(b[:len(b)-cos.SizeofI64], cos.MLCG32)
-	if h != hh {
-		return cos.NewErrMetaCksum(h, hh, utag)
+
+	l := int(binary.BigEndian.Uint16(data[offset:]))
+	offset += cos.SizeofI16
+	if len(data) < offset+l {
+		return "", offset, errors.New(tooShort)
 	}
+	str := string(data[offset : offset+l])
+	offset += l
+	return str, offset, nil
+}
 
-	return nil
+//
+// additional lom
+//
+
+func (lom *LOM) getXchunk(buf []byte) ([]byte, error) {
+	return fs.GetXattrBuf(lom.FQN, xattrChunk, buf)
+}
+
+func (lom *LOM) setXchunk(data []byte) error {
+	return fs.SetXattr(lom.FQN, xattrChunk, data)
 }