Support block compression API.

Author: scunningham

README.md

@@ -26,7 +26,7 @@ This project attempts to support all of the features enumerated in the [LZ4 Fram
 - [Sparse](./pkg/sparse) write support
 - Random read access (see [caveats](#random-read-access))
 
-
+While the primary purpose of plz4 is to support parallel processing, the raw block API's have also been supported for cases where the payloads are very small and do not benefit from LZ4 Framing.
 
 ## Design
 
@@ -57,6 +57,11 @@ There is another LZ4 Frame feature that is problematic at scale.  By default, pl
 
 Another advantage of independent blocks is the potential to support random read access.  This is possible because each block can be independently decompressed.  To support this, plz4 provides an optional progress callback that emits both the source offset and corresponding block offset during compression.  An implementation can use this information to build lookup tables that can later be used to skip ahead during decompression to a known block offset.  plz4 provides the 'WithReadOffset' option on the NewReader API to skip ahead and start decompression at a known block offset.
 
+### CGO
+
+
+This package uses CGO to call the canonical LZ4 library which is written in C.  There may be be cases where CGO is not desired, and in those cases the package also supports building with the environment variable "CGO_ENABLED=0".  In general, the library runs a bit slower in that mode and not all features are available.
+
 
 ## Install
 

cmd/plz4/internal/ops/bakeoff.go

@@ -31,7 +31,7 @@ func RunBakeoff() error {
 	)
 
 	// Consume into RAM; must be able to seek
-	if rdr == os.Stdin {
+	if rdr == os.Stdin || CLI.Bakeoff.RAM {
 		var buf bytes.Buffer
 		n, err := io.Copy(&buf, rdr)
 		if err != nil {
@@ -125,16 +125,22 @@ func outputOptions() error {
 		dict = CLI.Dict
 	}
 
-	t.AppendRows([]table.Row{
-		{"File name", fn},
-		{"Dictionary", dict},
-		{"Concurrency", CLI.Cpus},
-		{"Block Size", CLI.Bakeoff.BS},
-		{"Block Checksum", CLI.Bakeoff.BX},
-		{"Blocks Linked", CLI.Bakeoff.BD},
-		{"Content Checksum", CLI.Bakeoff.CS},
-		{"Content Size", CLI.Bakeoff.CX},
-	})
+	if CLI.Bakeoff.BlockMode {
+		t.AppendRows([]table.Row{
+			{"File name", fn},
+		})
+	} else {
+		t.AppendRows([]table.Row{
+			{"File name", fn},
+			{"Dictionary", dict},
+			{"Concurrency", CLI.Cpus},
+			{"Block Size", CLI.Bakeoff.BS},
+			{"Block Checksum", CLI.Bakeoff.BX},
+			{"Blocks Linked", CLI.Bakeoff.BD},
+			{"Content Checksum", CLI.Bakeoff.CS},
+			{"Content Size", CLI.Bakeoff.CX},
+		})
+	}
 
 	t.Render()
 	return nil
@@ -143,8 +149,13 @@ func outputOptions() error {
 func outputResults(srcSz int64, plz4Results, lz4Results []resultT) error {
 	fmt.Println()
 
+	mode := "frame mode"
+	if CLI.Bakeoff.BlockMode {
+		mode = "block mode"
+	}
+
 	t := table.NewWriter()
-	t.SetTitle("Bakeoff Results")
+	t.SetTitle(fmt.Sprintf("Bakeoff Results [%s]", mode))
 	t.SetStyle(table.StyleColoredBright)
 	t.SetOutputMirror(os.Stdout)
 	t.AppendHeader(table.Row{"Algo", "Level", "SrcSize", "Compressed", "Ratio", "Compress", "Decompress"})
@@ -208,6 +219,18 @@ func _prepLz4(rd io.ReadSeeker, srcSz int64, pw progress.Writer) (bakeFuncT, err
 
 	opts = append(opts, lz4.OnBlockDoneOption(cbHandler))
 
+	var srcBlock []byte
+	if CLI.Bakeoff.BlockMode {
+		srcBlock, err = io.ReadAll(rd)
+		if err != nil {
+			return nil, err
+		}
+		if _, err := rd.Seek(0, io.SeekStart); err != nil {
+			return nil, err
+		}
+
+	}
+
 	bakeFunc := func() ([]resultT, error) {
 		defer tr.MarkAsDone()
 
@@ -216,19 +239,33 @@ func _prepLz4(rd io.ReadSeeker, srcSz int64, pw progress.Writer) (bakeFuncT, err
 		for ; i < 10; i++ {
 			start := time.Now()
 
-			if _, err := rd.Seek(0, io.SeekStart); err != nil {
-				return nil, err
-			}
+			var (
+				split time.Time
+				cnt   int64
+				err   error
+			)
 
-			// Last one wins; so append is ok.
 			lvl, err := lz4Level(i)
 			if err != nil {
 				return nil, err
 			}
 
-			opts = append(opts, lz4.CompressionLevelOption(lvl))
+			if srcBlock != nil {
+				// Block mode
+				split, cnt, err = lz4BakeOneBlock(srcBlock, lvl)
+
+			} else {
+
+				if _, err := rd.Seek(0, io.SeekStart); err != nil {
+					return nil, err
+				}
+
+				// Last one wins; so append is ok.
+				opts = append(opts, lz4.CompressionLevelOption(lvl))
+
+				split, cnt, err = lz4BakeOne(rd, opts...)
+			}
 
-			split, cnt, err := lz4BakeOne(rd, opts...)
 			if err != nil {
 				return nil, err
 			}
@@ -335,21 +372,44 @@ func _prepPlz4(rd io.ReadSeeker, srcSz int64, pw progress.Writer) (bakeFuncT, er
 			plz4.WithWorkerPool(wp),
 		)
 
+		var srcBlock []byte
+		if CLI.Bakeoff.BlockMode {
+			srcBlock, err = io.ReadAll(rd)
+			if err != nil {
+				return nil, err
+			}
+			if _, err := rd.Seek(0, io.SeekStart); err != nil {
+				return nil, err
+			}
+		}
+
 		var results []resultT
 
 		for ; i < 12; i++ {
 			start := time.Now()
 
-			if _, err := rd.Seek(0, io.SeekStart); err != nil {
-				return nil, err
-			}
-
-			// Last one wins; so append is ok.
-			opts = append(opts,
-				plz4.WithLevel(plz4.LevelT(i+1)),
+			var (
+				split time.Time
+				cnt   int64
+				err   error
 			)
 
-			split, cnt, err := plz4BakeOne(rd, opts...)
+			if srcBlock != nil {
+				// Block mode
+				split, cnt, err = plz4BakeOneBlock(srcBlock, plz4.LevelT(i+1))
+
+			} else {
+				// Last one wins; so append is ok.
+
+				if _, err := rd.Seek(0, io.SeekStart); err != nil {
+					return nil, err
+				}
+
+				opts = append(opts,
+					plz4.WithLevel(plz4.LevelT(i+1)),
+				)
+				split, cnt, err = plz4BakeOne(rd, opts...)
+			}
 			if err != nil {
 				return nil, err
 			}
@@ -373,6 +433,7 @@ func _prepPlz4(rd io.ReadSeeker, srcSz int64, pw progress.Writer) (bakeFuncT, er
 }
 
 func plz4BakeOne(src io.Reader, opts ...plz4.OptT) (split time.Time, cnt int64, err error) {
+
 	var (
 		fh *os.File
 		wr io.Writer
@@ -445,6 +506,46 @@ func _plz4Decompress(rd io.Reader) error {
 	return err
 }
 
+func lz4BakeOneBlock(src []byte, level lz4.CompressionLevel) (split time.Time, cnt int64, err error) {
+
+	var (
+		sz  = lz4.CompressBlockBound(len(src))
+		dst = make([]byte, sz)
+		n   int
+	)
+
+	if level == lz4.Fast {
+		n, err = lz4.CompressBlock(src, dst, nil)
+	} else {
+		n, err = lz4.CompressBlockHC(src, dst, level, nil, nil)
+	}
+	if err != nil {
+		return
+	}
+
+	dst = dst[:n]
+	split = time.Now()
+	cnt = int64(n)
+
+	tmp := make([]byte, len(src))
+
+	_, err = lz4.UncompressBlock(dst, tmp)
+	return
+}
+
+func plz4BakeOneBlock(src []byte, level plz4.LevelT) (split time.Time, cnt int64, err error) {
+
+	dst, err := plz4.CompressBlock(src, plz4.WithBlockCompressionLevel(level))
+	if err != nil {
+		return
+	}
+
+	split = time.Now()
+	_, err = plz4.DecompressBlock(dst)
+	cnt = int64(len(dst))
+	return
+}
+
 func _lz4Decompress(rd io.Reader) error {
 
 	frd := lz4.NewReader(rd)

cmd/plz4/internal/ops/cli.go

@@ -25,13 +25,14 @@ var CLI struct {
 		Skip bool   `help:"Skip decompress" short:"s"`
 	} `cmd:"" aliases:"v,ver" help:"Verify lz4 data"`
 	Bakeoff struct {
-		File string `optional:"" arg:"" type:"existingfile"`
-		BS   string `help:"Block size [4MB, 1MB, 256KB, 64KB]" default:"4MB"`
-		BD   bool   `help:"Enable linked blocks"`
-		BX   bool   `help:"Enable block checksum"`
-		CX   bool   `help:"Enable content checksum"`
-		CS   bool   `help:"Enable content size; fails on stdin"`
-		RAM  bool   `help:"Process data in RAM"`
+		File      string `optional:"" arg:"" type:"existingfile"`
+		BS        string `help:"Block size [4MB, 1MB, 256KB, 64KB]" default:"4MB"`
+		BD        bool   `help:"Enable linked blocks"`
+		BX        bool   `help:"Enable block checksum"`
+		CX        bool   `help:"Enable content checksum"`
+		CS        bool   `help:"Enable content size; fails on stdin"`
+		RAM       bool   `help:"Process data in RAM"`
+		BlockMode bool   `help:"Use block API instead of frame API" short:"B"`
 	} `cmd:"" aliases:"b,bake" help:"Compare performance to github.com/pierrec/lz4"`
 
 	Cpus int    `help:"Concurrency [0 synchronous] [-1 auto]" default:"-1" short:"c"`

internal/pkg/clz4/clz4.go

@@ -24,6 +24,10 @@ func byteSliceToCharPointer(b []byte) *C.char {
 	return (*C.char)(unsafe.Pointer(&b[0]))
 }
 
+func CompressBound(sz int) int {
+	return int(C.LZ4_compressBound(C.int(sz)))
+}
+
 func CompressFast(source, dest []byte, acceleration int) (int, error) {
 	ret := int(C.LZ4_compress_fast(
 		byteSliceToCharPointer(source),

internal/pkg/compress/compress.go

@@ -68,10 +68,14 @@ func (f CompressorFactory) newIndie() Compressor {
 }
 
 func (f CompressorFactory) newLinked() Compressor {
-	switch {
-	case f.level == 1:
+	switch f.level {
+	case 1:
 		return newLinkedCompressor(f.dictCtx)
 	default:
 		return newLinkedCompressorHC(f.level, f.dictCtxHC)
 	}
 }
+
+func CompressBound(sz int) int {
+	return clz4.CompressBound(sz)
+}

internal/pkg/compress/nocgo_compress.go

@@ -47,30 +47,27 @@ func (f CompressorFactory) NewCompressor() Compressor {
 }
 
 type fastCompressor struct {
-	cmp lz4.Compressor
 }
 
 func (c *fastCompressor) Compress(src, dst, dict []byte) (int, error) {
-	return c.cmp.CompressBlock(src, dst)
+	return lz4.CompressBlock(src, dst, nil)
 }
 
 func NewCompressorHC(level int) Compressor {
 	if level > 9 {
 		level = 9
 	}
 	return &hcCompressor{
-		cmp: lz4.CompressorHC{
-			Level: lz4Level(level),
-		},
+		level: lz4Level(level),
 	}
 }
 
 type hcCompressor struct {
-	cmp lz4.CompressorHC
+	level lz4.CompressionLevel
 }
 
 func (c *hcCompressor) Compress(src, dst, dict []byte) (int, error) {
-	return c.cmp.CompressBlock(src, dst)
+	return lz4.CompressBlockHC(src, dst, c.level, nil, nil)
 }
 
 type failedCompressor struct {
@@ -110,3 +107,7 @@ func lz4Level(l int) lz4.CompressionLevel {
 	}
 	return lz4Level
 }
+
+func CompressBound(sz int) int {
+	return lz4.CompressBlockBound(sz)
+}