tar_uncompress.go

package genesis

// go-gzip, no hashing   elapsed=49.274027963s
// pgzip.NewReaderN(reader, 1024, 100*1024) , no hashing,  elapsed=1m13.959573387s

import (
	"archive/tar"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"github.com/Fantom-foundation/go-opera/cmd/sonictool/db"
	"github.com/Fantom-foundation/go-opera/opera/genesisstore/filelog"
	"github.com/ethereum/go-ethereum/log"
	"github.com/klauspost/pgzip"
	"io"
	"os"
	"path/filepath"
	"runtime"
	"runtime/pprof"
	"time"
)

func SonicImport(dataDir string, genesisFile *os.File) error {
	f, err := os.Create("cpuprofile")
	if err != nil {
		panic(err)
	}
	defer f.Close()
	if err := pprof.StartCPUProfile(f); err != nil {
		panic(err)
	}
	defer pprof.StopCPUProfile()
	runtime.SetBlockProfileRate(1)

	info, err := genesisFile.Stat()
	if err != nil {
		return fmt.Errorf("failed to get genesis file stats: %w", err)
	}

	if err := db.RemoveDatabase(dataDir); err != nil {
		return fmt.Errorf("failed to remove existing data from the datadir: %w", err)
	}

	chaindataDir := filepath.Join(dataDir, "chaindata")
	err = os.MkdirAll(chaindataDir, 0700)
	if err != nil {
		return fmt.Errorf("failed to create chaindataDir directory: %w", err)
	}
	setGenesisProcessing(chaindataDir)

	hasher := parallelHasher{
		inputs: make(chan []byte, 10),
		freeSlices: make(chan []byte, 15),
		result: make(chan string),
	}
	go hasher.run()

	start := time.Now()
	log.Info("Unpacking Sonic genesis")
	reader := filelog.Wrap(genesisFile, "sonic-genesis", uint64(info.Size()), time.Minute)
	teeReader := io.TeeReader(reader, &hasher)
	uncompressedStream, err := pgzip.NewReaderN(teeReader, 1024, 100*1024)
	if err != nil {
		return err
	}
	tarReader := tar.NewReader(uncompressedStream)
	for {
		header, err := tarReader.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			return fmt.Errorf("untar failed: %w", err)
		}
		switch header.Typeflag {
		case tar.TypeDir:
			if err := os.MkdirAll(filepath.Join(dataDir, header.Name), 0700); err != nil {
				return fmt.Errorf("mkdir failed: %w", err)
			}
		case tar.TypeReg:
			outFile, err := os.Create(filepath.Join(dataDir, header.Name))
			if err != nil {
				return fmt.Errorf("create file failed: %w", err)
			}
			if _, err := io.Copy(outFile, tarReader); err != nil {
				outFile.Close()
				return fmt.Errorf("file write failed: %w", err)
			}
			outFile.Close()
		default:
			return fmt.Errorf("uknown tar content type: %x in %s", header.Typeflag, header.Name)
		}
	}
	hasher.Close()
	hash := <- hasher.result

	f2, err := os.Create("blocking")
	if err != nil {
		panic(err)
	}
	defer f2.Close()
	pprof.Lookup("block").WriteTo(f2, 0)

	name, ok := allowedSonicGenesisHashes[hash]
	if !ok {
		_ = db.RemoveDatabase(dataDir)
		return fmt.Errorf("hash of the genesis file does not match any allowed value: %s", hash)
	}

	setGenesisComplete(chaindataDir)
	log.Info("Successfully imported Sonic genesis", "name", name, "elapsed", time.Since(start))
	return nil
}

const inputMaxSize = 100 * 1024

type parallelHasher struct {
	inputs chan []byte
	freeSlices chan []byte
	result chan string
}

func (h *parallelHasher) Write(b []byte) (n int, err error) {
	if len(b) == 0 {
		return 0, nil
	}
	var cp []byte
	select {
	case cp = <-h.freeSlices:
		// reused
	default:
		cp = make([]byte, inputMaxSize)
	}
	n = copy(cp, b)
	h.inputs <- cp
	return n, nil
}

func (h *parallelHasher) Close() {
	close(h.inputs)
}

func (h *parallelHasher) run() {
	var hasher = sha256.New()
	for input := range h.inputs {
		_, err := hasher.Write(input)
		if err != nil {
			panic(err)
		}
		h.freeSlices <- input
	}
	h.result <- hex.EncodeToString(hasher.Sum(nil))
}