<commit_before>package backup

import (
	"archive/tar"
	"context"
	"emperror.dev/errors"
	"github.com/apex/log"
	gzip "github.com/klauspost/pgzip"
	"github.com/remeh/sizedwaitgroup"
	"golang.org/x/sync/errgroup"
	"io"
	"os"
	"runtime"
	"strings"
	"sync"
)

type Archive struct {
	sync.Mutex

	TrimPrefix string
	Files      *IncludedFiles
}

// Creates an archive at dst with all of the files defined in the included files struct.
func (a *Archive) Create(dst string, ctx context.Context) error {
	f, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
	if err != nil {
		return errors.WithStackIf(err)
	}
	defer f.Close()

	maxCpu := runtime.NumCPU() / 2
	if maxCpu > 4 {
		maxCpu = 4
	}

	gzw, _ := gzip.NewWriterLevel(f, gzip.BestSpeed)
	_ = gzw.SetConcurrency(1<<20, maxCpu)

	defer gzw.Flush()
	defer gzw.Close()

	tw := tar.NewWriter(gzw)
	defer tw.Flush()
	defer tw.Close()

	wg := sizedwaitgroup.New(10)
	g, ctx := errgroup.WithContext(ctx)
	// Iterate over all of the files to be included and put them into the archive. This is
	// done as a concurrent goroutine to speed things along. If an error is encountered at
	// any step, the entire process is aborted.
	for _, p := range a.Files.All() {
		p := p
		g.Go(func() error {
			wg.Add()
			defer wg.Done()

			select {
			case <-ctx.Done():
				return errors.WithStackIf(ctx.Err())
			default:
				return a.addToArchive(p, tw)
			}
		})
	}

	// Block until the entire routine is completed.
	if err := g.Wait(); err != nil {
		f.Close()

		// Attempt to remove the archive if there is an error, report that error to
		// the logger if it fails.
		if rerr := os.Remove(dst); rerr != nil && !os.IsNotExist(rerr) {
			log.WithField("location", dst).Warn("failed to delete corrupted backup archive")
		}

		return errors.WithStackIf(err)
	}

	return nil
}

// Adds a single file to the existing tar archive writer.
func (a *Archive) addToArchive(p string, w *tar.Writer) error {
	f, err := os.Open(p)
	if err != nil {
		// If you try to backup something that no longer exists (got deleted somewhere during the process
		// but not by this process), just skip over it and don't kill the entire backup.
		if os.IsNotExist(err) {
			return nil
		}

		return errors.WithStackIf(err)
	}
	defer f.Close()

	s, err := f.Stat()
	if err != nil {
		// Same as above, don't kill the process just because the file no longer exists.
		if os.IsNotExist(err) {
			return nil
		}

		return errors.WithStackIf(err)
	}

	header, err := tar.FileInfoHeader(s, strings.TrimPrefix(p, a.TrimPrefix))
	if err != nil {
		return errors.WithStackIf(err)
	}

	// These actions must occur sequentially, even if this function is called multiple
	// in parallel. You'll get some nasty panic's otherwise.
	a.Lock()
	defer a.Unlock()

	if err := w.WriteHeader(header); err != nil {
		return errors.WithStackIf(err)
	}

	buf := make([]byte, 4*1024)
	if _, err := io.CopyBuffer(w, f, buf); err != nil {
		return errors.WithStackIf(err)
	}

	return nil
}<commit_msg>Oh, that's why it was done like that<commit_after>package backup

import (
	"archive/tar"
	"context"
	"emperror.dev/errors"
	"github.com/apex/log"
	gzip "github.com/klauspost/pgzip"
	"github.com/remeh/sizedwaitgroup"
	"golang.org/x/sync/errgroup"
	"io"
	"os"
	"runtime"
	"strings"
	"sync"
)

type Archive struct {
	sync.Mutex

	TrimPrefix string
	Files      *IncludedFiles
}

// Creates an archive at dst with all of the files defined in the included files struct.
func (a *Archive) Create(dst string, ctx context.Context) error {
	f, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
	if err != nil {
		return errors.WithStackIf(err)
	}
	defer f.Close()

	maxCpu := runtime.NumCPU() / 2
	if maxCpu > 4 {
		maxCpu = 4
	}

	gzw, _ := gzip.NewWriterLevel(f, gzip.BestSpeed)
	_ = gzw.SetConcurrency(1<<20, maxCpu)

	defer gzw.Flush()
	defer gzw.Close()

	tw := tar.NewWriter(gzw)
	defer tw.Flush()
	defer tw.Close()

	wg := sizedwaitgroup.New(10)
	g, ctx := errgroup.WithContext(ctx)
	// Iterate over all of the files to be included and put them into the archive. This is
	// done as a concurrent goroutine to speed things along. If an error is encountered at
	// any step, the entire process is aborted.
	for _, p := range a.Files.All() {
		p := p
		g.Go(func() error {
			wg.Add()
			defer wg.Done()

			select {
			case <-ctx.Done():
				return errors.WithStackIf(ctx.Err())
			default:
				return a.addToArchive(p, tw)
			}
		})
	}

	// Block until the entire routine is completed.
	if err := g.Wait(); err != nil {
		f.Close()

		// Attempt to remove the archive if there is an error, report that error to
		// the logger if it fails.
		if rerr := os.Remove(dst); rerr != nil && !os.IsNotExist(rerr) {
			log.WithField("location", dst).Warn("failed to delete corrupted backup archive")
		}

		return errors.WithStackIf(err)
	}

	return nil
}

// Adds a single file to the existing tar archive writer.
func (a *Archive) addToArchive(p string, w *tar.Writer) error {
	f, err := os.Open(p)
	if err != nil {
		// If you try to backup something that no longer exists (got deleted somewhere during the process
		// but not by this process), just skip over it and don't kill the entire backup.
		if os.IsNotExist(err) {
			return nil
		}

		return errors.WithStackIf(err)
	}
	defer f.Close()

	s, err := f.Stat()
	if err != nil {
		// Same as above, don't kill the process just because the file no longer exists.
		if os.IsNotExist(err) {
			return nil
		}

		return errors.WithStackIf(err)
	}

	header, err := tar.FileInfoHeader(s, strings.TrimPrefix(p, a.TrimPrefix))
	if err != nil {
		return errors.WithStackIf(err)
	}
	header.Name = strings.TrimPrefix(p, a.TrimPrefix)

	// These actions must occur sequentially, even if this function is called multiple
	// in parallel. You'll get some nasty panic's otherwise.
	a.Lock()
	defer a.Unlock()

	if err := w.WriteHeader(header); err != nil {
		return errors.WithStackIf(err)
	}

	buf := make([]byte, 4*1024)
	if _, err := io.CopyBuffer(w, io.LimitReader(f, header.Size), buf); err != nil {
		return errors.WithStackIf(err)
	}

	return nil
}
{'fraction_non_alphanumeric': 0.12653663333879692, 'fraction_numerical': 0.005572856908703491, 'mean_word_length': 3.3184713375796178, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4707260', 'n_tokens_mistral': 2348, 'n_tokens_neox': 2147, 'n_words': 689}