<commit_before>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package mysqlctl

import (
	"bufio"
	"code.google.com/p/vitess/go/relog"
	"compress/gzip"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"path"
)

// Use this to simulate failures in tests
var (
	simulateFailures = false
	failureCounter   = 0
)

func init() {
	_, statErr := os.Stat("/tmp/vtSimulateFetchFailures")
	simulateFailures = statErr == nil
}

// compressFile compresses a single file with gzip, leaving the src
// file intact.  Also computes the md5 hash on the fly.
func compressFile(srcPath, dstPath string) (*SnapshotFile, error) {
	// FIXME(msolomon) not sure how well Go will schedule cpu intensive tasks
	// might be better if this forked off workers.
	relog.Info("compressFile: starting to compress %v into %v", srcPath, dstPath)
	srcFile, err := os.OpenFile(srcPath, os.O_RDONLY, 0)
	if err != nil {
		return nil, err
	}
	defer srcFile.Close()

	src := bufio.NewReaderSize(srcFile, 2*1024*1024)

	dir, filePrefix := path.Split(dstPath)

	dstFile, err := ioutil.TempFile(dir, filePrefix)
	if err != nil {
		return nil, err
	}
	defer func() {
		// try to close and delete the file.
		// in the success case, the file will already be closed
		// and renamed, so all of this would fail anyway, no biggie
		dstFile.Close()
		os.Remove(dstFile.Name())
	}()

	dst := bufio.NewWriterSize(dstFile, 2*1024*1024)

	hasher := md5.New()
	tee := io.MultiWriter(dst, hasher)

	compressor := gzip.NewWriter(tee)
	defer compressor.Close()

	_, err = io.Copy(compressor, src)
	if err != nil {
		return nil, err
	}

	// close dst manually to flush all buffers to disk
	compressor.Close()
	dst.Flush()
	dstFile.Close()
	hash := hex.EncodeToString(hasher.Sum(nil))

	// atomically move completed compressed file
	err = os.Rename(dstFile.Name(), dstPath)
	if err != nil {
		return nil, err
	}

	relog.Info("clone data ready %v:%v", dstPath, hash)
	return &SnapshotFile{dstPath, hash}, nil
}

// compressFile compresses multiple files in parallel.
func compressFiles(sources, destinations []string, compressConcurrency int) ([]SnapshotFile, error) {
	if len(sources) != len(destinations) || len(sources) == 0 {
		panic(fmt.Errorf("bad array lengths: %v %v", len(sources), len(destinations)))
	}

	workQueue := make(chan int, len(sources))
	for i := 0; i < len(sources); i++ {
		workQueue <- i
	}
	close(workQueue)

	snapshotFiles := make([]SnapshotFile, len(sources))
	resultQueue := make(chan error, len(sources))
	for i := 0; i < compressConcurrency; i++ {
		go func() {
			for i := range workQueue {
				sf, err := compressFile(sources[i], destinations[i])
				if err == nil {
					snapshotFiles[i] = *sf
				}
				resultQueue <- err
			}
		}()
	}

	var err error
	for i := 0; i < len(sources); i++ {
		if compressErr := <-resultQueue; compressErr != nil {
			err = compressErr
		}
	}

	// clean up files if we had an error
	// FIXME(alainjobart) it seems extreme to delete all files if
	// the last one failed. Since we only move the file into
	// its destination when it worked, we could assume if the file
	// already exists it's good, and re-compute its md5.
	if err != nil {
		relog.Info("Error happened, deleting all the files we already compressed")
		for _, dest := range destinations {
			os.Remove(dest)
		}
		return nil, err
	}

	return snapshotFiles, nil
}

// fetchFile fetches data from the web server.  It then sends it to a
// tee, which on one side has an md5 checksum reader, and on the other
// a gunzip reader writing to a file.  It will compare the md5
// checksum after the copy is done.
func fetchFile(srcUrl, srcHash, dstFilename string) error {
	relog.Info("fetchFile: starting to fetch %v", dstFilename)

	// create destination directory
	dir, _ := path.Split(dstFilename)
	if dirErr := os.MkdirAll(dir, 0775); dirErr != nil {
		return dirErr
	}

	// open the URL
	resp, err := http.Get(srcUrl)
	if resp.StatusCode != 200 {
		return fmt.Errorf("failed fetching %v: %v", srcUrl, resp.Status)
	}
	defer resp.Body.Close()

	// create a temporary file to uncompress to
	dir, filePrefix := path.Split(dstFilename)
	dstFile, err := ioutil.TempFile(dir, filePrefix)
	if err != nil {
		return err
	}
	defer func() {
		// try to close and delete the file.
		// in the success case, the file will already be closed
		// and renamed, so all of this would fail anyway, no biggie
		dstFile.Close()
		os.Remove(dstFile.Name())
	}()

	// create a buffering output
	dst := bufio.NewWriterSize(dstFile, 2*1024*1024)

	// create md5 hash to write the compressed data to
	hasher := md5.New()

	// create a Tee: we split the HTTP input into the md5 hasher
	// and into the gunziper
	tee := io.TeeReader(resp.Body, hasher)

	// create the uncompresser
	decompressor, err := gzip.NewReader(tee)
	if err != nil {
		return err
	}
	defer decompressor.Close()

	// see if we need to introduce failures
	if simulateFailures {
		failureCounter++
		if failureCounter%3 == 0 {
			return fmt.Errorf("Simulated error")
		}
	}

	// copy the data. Will also write to the hasher
	if _, err = io.Copy(dst, decompressor); err != nil {
		return err
	}

	// check the md5
	hash := hex.EncodeToString(hasher.Sum(nil))
	if srcHash != hash {
		return fmt.Errorf("hash mismatch for %v, %v != %v", dstFilename, srcHash, hash)
	}

	// we're good
	relog.Info("fetched snapshot file: %v", dstFilename)
	dst.Flush()
	dstFile.Close()

	// atomically move uncompressed file
	if err := os.Chmod(dstFile.Name(), 0664); err != nil {
		return err
	}
	return os.Rename(dstFile.Name(), dstFilename)
}<commit_msg>Less simulated errors, to hopefully eliminate random test failures.<commit_after>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package mysqlctl

import (
	"bufio"
	"code.google.com/p/vitess/go/relog"
	"compress/gzip"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"path"
)

// Use this to simulate failures in tests
var (
	simulateFailures = false
	failureCounter   = 0
)

func init() {
	_, statErr := os.Stat("/tmp/vtSimulateFetchFailures")
	simulateFailures = statErr == nil
}

// compressFile compresses a single file with gzip, leaving the src
// file intact.  Also computes the md5 hash on the fly.
func compressFile(srcPath, dstPath string) (*SnapshotFile, error) {
	// FIXME(msolomon) not sure how well Go will schedule cpu intensive tasks
	// might be better if this forked off workers.
	relog.Info("compressFile: starting to compress %v into %v", srcPath, dstPath)
	srcFile, err := os.OpenFile(srcPath, os.O_RDONLY, 0)
	if err != nil {
		return nil, err
	}
	defer srcFile.Close()

	src := bufio.NewReaderSize(srcFile, 2*1024*1024)

	dir, filePrefix := path.Split(dstPath)

	dstFile, err := ioutil.TempFile(dir, filePrefix)
	if err != nil {
		return nil, err
	}
	defer func() {
		// try to close and delete the file.
		// in the success case, the file will already be closed
		// and renamed, so all of this would fail anyway, no biggie
		dstFile.Close()
		os.Remove(dstFile.Name())
	}()

	dst := bufio.NewWriterSize(dstFile, 2*1024*1024)

	hasher := md5.New()
	tee := io.MultiWriter(dst, hasher)

	compressor := gzip.NewWriter(tee)
	defer compressor.Close()

	_, err = io.Copy(compressor, src)
	if err != nil {
		return nil, err
	}

	// close dst manually to flush all buffers to disk
	compressor.Close()
	dst.Flush()
	dstFile.Close()
	hash := hex.EncodeToString(hasher.Sum(nil))

	// atomically move completed compressed file
	err = os.Rename(dstFile.Name(), dstPath)
	if err != nil {
		return nil, err
	}

	relog.Info("clone data ready %v:%v", dstPath, hash)
	return &SnapshotFile{dstPath, hash}, nil
}

// compressFile compresses multiple files in parallel.
func compressFiles(sources, destinations []string, compressConcurrency int) ([]SnapshotFile, error) {
	if len(sources) != len(destinations) || len(sources) == 0 {
		panic(fmt.Errorf("bad array lengths: %v %v", len(sources), len(destinations)))
	}

	workQueue := make(chan int, len(sources))
	for i := 0; i < len(sources); i++ {
		workQueue <- i
	}
	close(workQueue)

	snapshotFiles := make([]SnapshotFile, len(sources))
	resultQueue := make(chan error, len(sources))
	for i := 0; i < compressConcurrency; i++ {
		go func() {
			for i := range workQueue {
				sf, err := compressFile(sources[i], destinations[i])
				if err == nil {
					snapshotFiles[i] = *sf
				}
				resultQueue <- err
			}
		}()
	}

	var err error
	for i := 0; i < len(sources); i++ {
		if compressErr := <-resultQueue; compressErr != nil {
			err = compressErr
		}
	}

	// clean up files if we had an error
	// FIXME(alainjobart) it seems extreme to delete all files if
	// the last one failed. Since we only move the file into
	// its destination when it worked, we could assume if the file
	// already exists it's good, and re-compute its md5.
	if err != nil {
		relog.Info("Error happened, deleting all the files we already compressed")
		for _, dest := range destinations {
			os.Remove(dest)
		}
		return nil, err
	}

	return snapshotFiles, nil
}

// fetchFile fetches data from the web server.  It then sends it to a
// tee, which on one side has an md5 checksum reader, and on the other
// a gunzip reader writing to a file.  It will compare the md5
// checksum after the copy is done.
func fetchFile(srcUrl, srcHash, dstFilename string) error {
	relog.Info("fetchFile: starting to fetch %v", dstFilename)

	// create destination directory
	dir, _ := path.Split(dstFilename)
	if dirErr := os.MkdirAll(dir, 0775); dirErr != nil {
		return dirErr
	}

	// open the URL
	resp, err := http.Get(srcUrl)
	if resp.StatusCode != 200 {
		return fmt.Errorf("failed fetching %v: %v", srcUrl, resp.Status)
	}
	defer resp.Body.Close()

	// create a temporary file to uncompress to
	dir, filePrefix := path.Split(dstFilename)
	dstFile, err := ioutil.TempFile(dir, filePrefix)
	if err != nil {
		return err
	}
	defer func() {
		// try to close and delete the file.
		// in the success case, the file will already be closed
		// and renamed, so all of this would fail anyway, no biggie
		dstFile.Close()
		os.Remove(dstFile.Name())
	}()

	// create a buffering output
	dst := bufio.NewWriterSize(dstFile, 2*1024*1024)

	// create md5 hash to write the compressed data to
	hasher := md5.New()

	// create a Tee: we split the HTTP input into the md5 hasher
	// and into the gunziper
	tee := io.TeeReader(resp.Body, hasher)

	// create the uncompresser
	decompressor, err := gzip.NewReader(tee)
	if err != nil {
		return err
	}
	defer decompressor.Close()

	// see if we need to introduce failures
	if simulateFailures {
		failureCounter++
		if failureCounter%5 == 0 {
			return fmt.Errorf("Simulated error")
		}
	}

	// copy the data. Will also write to the hasher
	if _, err = io.Copy(dst, decompressor); err != nil {
		return err
	}

	// check the md5
	hash := hex.EncodeToString(hasher.Sum(nil))
	if srcHash != hash {
		return fmt.Errorf("hash mismatch for %v, %v != %v", dstFilename, srcHash, hash)
	}

	// we're good
	relog.Info("fetched snapshot file: %v", dstFilename)
	dst.Flush()
	dstFile.Close()

	// atomically move uncompressed file
	if err := os.Chmod(dstFile.Name(), 0664); err != nil {
		return err
	}
	return os.Rename(dstFile.Name(), dstFilename)
}
{'fraction_non_alphanumeric': 0.1231820571228316, 'fraction_numerical': 0.010513404590853337, 'mean_word_length': 3.544187898089172, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, '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': '3713071', 'n_tokens_mistral': 4089, 'n_tokens_neox': 3790, 'n_words': 1314}