<commit_before>package donut

import (
	"bytes"
	"errors"
	"io"
	"strconv"
	"strings"
	"time"

	"crypto/md5"
	"encoding/hex"

	"github.com/minio-io/iodine"
	"github.com/minio-io/minio/pkg/encoding/erasure"
	"github.com/minio-io/minio/pkg/utils/split"
	"hash"
)

// getErasureTechnique - convert technique string into Technique type
func getErasureTechnique(technique string) (erasure.Technique, error) {
	switch true {
	case technique == "Cauchy":
		return erasure.Cauchy, nil
	case technique == "Vandermonde":
		return erasure.Cauchy, nil
	default:
		return erasure.None, iodine.New(errors.New("Invalid erasure technique: "+technique), nil)
	}
}

// erasureReader - returns aligned streaming reads over a PipeWriter
func erasureReader(readers []io.ReadCloser, donutMetadata map[string]string, writer *io.PipeWriter) {
	totalChunks, err := strconv.Atoi(donutMetadata["chunkCount"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	totalLeft, err := strconv.ParseInt(donutMetadata["size"], 10, 64)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	blockSize, err := strconv.Atoi(donutMetadata["blockSize"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	parsedk, err := strconv.ParseUint(donutMetadata["erasureK"], 10, 8)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	k := uint8(parsedk)
	parsedm, err := strconv.ParseUint(donutMetadata["erasureM"], 10, 8)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	m := uint8(parsedm)
	expectedMd5sum, err := hex.DecodeString(donutMetadata["md5"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	technique, err := getErasureTechnique(donutMetadata["erasureTechnique"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	hasher := md5.New()
	params, err := erasure.ParseEncoderParams(k, m, technique)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
	}
	encoder := erasure.NewEncoder(params)
	for i := 0; i < totalChunks; i++ {
		totalLeft, err = decodeChunk(writer, readers, encoder, hasher, k, m, totalLeft, blockSize)
		if err != nil {
			errParams := map[string]string{
				"totalLeft": strconv.FormatInt(totalLeft, 10),
			}
			for k, v := range donutMetadata {
				errParams[k] = v
			}
			writer.CloseWithError(iodine.New(err, errParams))
		}
	}
	actualMd5sum := hasher.Sum(nil)
	if bytes.Compare(expectedMd5sum, actualMd5sum) != 0 {
		writer.CloseWithError(iodine.New(errors.New("decoded md5sum did not match. expected: "+string(expectedMd5sum)+" actual: "+string(actualMd5sum)), donutMetadata))
		return
	}
	writer.Close()
	return
}

func decodeChunk(writer *io.PipeWriter, readers []io.ReadCloser, encoder *erasure.Encoder, hasher hash.Hash, k, m uint8, totalLeft int64, blockSize int) (int64, error) {
	curBlockSize := 0
	if int64(blockSize) < totalLeft {
		curBlockSize = blockSize
	} else {
		curBlockSize = int(totalLeft) // cast is safe, blockSize in if protects
	}

	curChunkSize := erasure.GetEncodedBlockLen(curBlockSize, uint8(k))
	encodedBytes := make([][]byte, 16)
	for i, reader := range readers {
		defer reader.Close()
		var bytesBuffer bytes.Buffer
		written, err := io.CopyN(&bytesBuffer, reader, int64(curChunkSize))
		if err != nil {
			errParams := map[string]string{}
			errParams["part"] = strconv.FormatInt(written, 10)
			errParams["block.written"] = strconv.FormatInt(written, 10)
			errParams["block.length"] = strconv.Itoa(curChunkSize)
			return totalLeft, iodine.New(err, errParams)
		}
		encodedBytes[i] = bytesBuffer.Bytes()
	}
	decodedData, err := encoder.Decode(encodedBytes, curBlockSize)
	if err != nil {
		errParams := map[string]string{}
		errParams["block.length"] = strconv.Itoa(curChunkSize)
		return totalLeft, iodine.New(err, errParams)
	}
	_, err = hasher.Write(decodedData) // not expecting errors from hash, will also catch further down on .Sum mismatch in parent
	if err != nil {
		errParams := map[string]string{}
		errParams["block.length"] = strconv.Itoa(curChunkSize)
		return totalLeft, iodine.New(err, errParams)
	}
	_, err = io.Copy(writer, bytes.NewBuffer(decodedData))
	if err != nil {<commit_msg>Fix for readers closing in erasure early<commit_after>package donut

import (
	"bytes"
	"errors"
	"io"
	"strconv"
	"strings"
	"time"

	"crypto/md5"
	"encoding/hex"

	"github.com/minio-io/iodine"
	"github.com/minio-io/minio/pkg/encoding/erasure"
	"github.com/minio-io/minio/pkg/utils/split"
	"hash"
	"log"
)

// getErasureTechnique - convert technique string into Technique type
func getErasureTechnique(technique string) (erasure.Technique, error) {
	switch true {
	case technique == "Cauchy":
		return erasure.Cauchy, nil
	case technique == "Vandermonde":
		return erasure.Cauchy, nil
	default:
		return erasure.None, iodine.New(errors.New("Invalid erasure technique: "+technique), nil)
	}
}

// erasureReader - returns aligned streaming reads over a PipeWriter
func erasureReader(readers []io.ReadCloser, donutMetadata map[string]string, writer *io.PipeWriter) {
	totalChunks, err := strconv.Atoi(donutMetadata["chunkCount"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	totalLeft, err := strconv.ParseInt(donutMetadata["size"], 10, 64)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	blockSize, err := strconv.Atoi(donutMetadata["blockSize"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	parsedk, err := strconv.ParseUint(donutMetadata["erasureK"], 10, 8)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	k := uint8(parsedk)
	parsedm, err := strconv.ParseUint(donutMetadata["erasureM"], 10, 8)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	m := uint8(parsedm)
	expectedMd5sum, err := hex.DecodeString(donutMetadata["md5"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	technique, err := getErasureTechnique(donutMetadata["erasureTechnique"])
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
		return
	}
	hasher := md5.New()
	params, err := erasure.ParseEncoderParams(k, m, technique)
	if err != nil {
		writer.CloseWithError(iodine.New(err, donutMetadata))
	}
	encoder := erasure.NewEncoder(params)
	for i := 0; i < totalChunks; i++ {
		log.Println(i)
		totalLeft, err = decodeChunk(writer, readers, encoder, hasher, k, m, totalLeft, blockSize)
		if err != nil {
			errParams := map[string]string{
				"totalLeft": strconv.FormatInt(totalLeft, 10),
			}
			for k, v := range donutMetadata {
				errParams[k] = v
			}
			writer.CloseWithError(iodine.New(err, errParams))
		}
	}
	actualMd5sum := hasher.Sum(nil)
	if bytes.Compare(expectedMd5sum, actualMd5sum) != 0 {
		writer.CloseWithError(iodine.New(errors.New("decoded md5sum did not match. expected: "+string(expectedMd5sum)+" actual: "+string(actualMd5sum)), donutMetadata))
		return
	}
	writer.Close()
	return
}

func decodeChunk(writer *io.PipeWriter, readers []io.ReadCloser, encoder *erasure.Encoder, hasher hash.Hash, k, m uint8, totalLeft int64, blockSize int) (int64, error) {
	curBlockSize := 0
	if int64(blockSize) < totalLeft {
		curBlockSize = blockSize
	} else {
		curBlockSize = int(totalLeft) // cast is safe, blockSize in if protects
	}

	curChunkSize := erasure.GetEncodedBlockLen(curBlockSize, uint8(k))
	encodedBytes := make([][]byte, 16)
	for i, reader := range readers {
		var bytesBuffer bytes.Buffer
		written, err := io.CopyN(&bytesBuffer, reader, int64(curChunkSize))
		if err != nil {
			errParams := map[string]string{}
			errParams["part"] = strconv.FormatInt(written, 10)
			errParams["block.written"] = strconv.FormatInt(written, 10)
			errParams["block.length"] = strconv.Itoa(curChunkSize)
			return totalLeft, iodine.New(err, errParams)
		}
		encodedBytes[i] = bytesBuffer.Bytes()
	}
	decodedData, err := encoder.Decode(encodedBytes, curBlockSize)
	if err != nil {
		errParams := map[string]string{}
		errParams["block.length"] = strconv.Itoa(curChunkSize)
		return totalLeft, iodine.New(err, errParams)
	}
	_, err = hasher.Write(decodedData) // not expecting errors from hash, will also catch further down on .Sum mismatch in parent
	if err != nil {
		errParams := map[string]string{}
		errParams["block.length"] = strconv.Itoa(curChunkSize)
		return totalLeft, iodine.New(err, errParams)
	}
	_, err = io.Copy(writer, bytes.NewBuffer(decodedData))
	if err != nil {
{'fraction_non_alphanumeric': 0.13674418604651162, 'fraction_numerical': 0.01, 'mean_word_length': 4.9235537190082646, 'pattern_counts': {'":': 6, '<': 7, '<?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': '4012079', 'n_tokens_mistral': 3253, 'n_tokens_neox': 3036, 'n_words': 660}