<commit_before>package main

import (
	"io"
	"io/ioutil"
	"os"
	"runtime"
	"sync"
)

// FilePipe is a tailable buffer backed by a file.
type FilePipe struct {
	f *os.File

	mu      sync.Mutex
	cond    sync.Cond
	size    int64
	closing bool
}

// Create a new FilePipe backed by a temporary file.
func NewFilePipe() (*FilePipe, error) {
	f, err := ioutil.TempFile("", "filepipe")
	if err != nil {
		return nil, err
	}
	fp := &FilePipe{f: f}
	fp.cond.L = &fp.mu
	runtime.SetFinalizer(fp, (*FilePipe).Release)
	return fp, nil
}

// Release the resources associated with the filepipe. In particular,
// remove the backing file. After this call has been made no readers
// of this filepipe should be used.
func (fp *FilePipe) Release() error {
	fp.Close()
	filename := fp.f.Name()
	err := fp.f.Close()
	if err1 := os.Remove(filename); err == nil {
		err = err1
	}
	return err
}

// Create a new reader that starts reading the filepipe's contents from
// the very beginning.
func (fp *FilePipe) Reader() io.Reader {
	return &filePipeReader{fp: fp, pos: 0}
}

func (fp *FilePipe) Write(buf []byte) (int, error) {
	n, err := fp.f.Write(buf)
	fp.mu.Lock()
	fp.size += int64(n)
	fp.cond.Broadcast()
	fp.mu.Unlock()
	return n, err
}

func (fp *FilePipe) Close() error {
	fp.mu.Lock()
	fp.closing = true
	fp.cond.Broadcast()
	fp.mu.Unlock()
	return nil
}

type filePipeReader struct {
	fp  *FilePipe
	pos int64
}

func (fpr *filePipeReader) Read(buf []byte) (int, error) {
	for {
		n, err := fpr.fp.f.ReadAt(buf, fpr.pos)
		fpr.pos += int64(n)
		if err == io.EOF {
			err = nil
		}
		if err != nil || n > 0 {
			return n, err
		}
		fpr.fp.mu.Lock()
		for fpr.pos >= fpr.fp.size && !fpr.fp.closing {
			fpr.fp.cond.Wait()
		}
		closing := fpr.fp.closing
		fpr.fp.mu.Unlock()
		if closing {
			return 0, io.EOF
		}
	}
}<commit_msg>fix a premature EOF bug in filepipe<commit_after>package main

import (
	"io"
	"io/ioutil"
	"os"
	"runtime"
	"sync"
)

// FilePipe is a tailable buffer backed by a file.
type FilePipe struct {
	f *os.File

	mu      sync.Mutex
	cond    sync.Cond
	size    int64
	closing bool
}

// Create a new FilePipe backed by a temporary file.
func NewFilePipe() (*FilePipe, error) {
	f, err := ioutil.TempFile("", "filepipe")
	if err != nil {
		return nil, err
	}
	fp := &FilePipe{f: f}
	fp.cond.L = &fp.mu
	runtime.SetFinalizer(fp, (*FilePipe).Release)
	return fp, nil
}

// Release the resources associated with the filepipe. In particular,
// remove the backing file. After this call has been made no readers
// of this filepipe should be used.
func (fp *FilePipe) Release() error {
	fp.Close()
	filename := fp.f.Name()
	err := fp.f.Close()
	if err1 := os.Remove(filename); err == nil {
		err = err1
	}
	return err
}

// Create a new reader that starts reading the filepipe's contents from
// the very beginning.
func (fp *FilePipe) Reader() io.Reader {
	return &filePipeReader{fp: fp, pos: 0}
}

func (fp *FilePipe) Write(buf []byte) (int, error) {
	n, err := fp.f.Write(buf)
	fp.mu.Lock()
	fp.size += int64(n)
	fp.cond.Broadcast()
	fp.mu.Unlock()
	return n, err
}

func (fp *FilePipe) Close() error {
	fp.mu.Lock()
	fp.closing = true
	fp.cond.Broadcast()
	fp.mu.Unlock()
	return nil
}

type filePipeReader struct {
	fp  *FilePipe
	pos int64
}

func (fpr *filePipeReader) Read(buf []byte) (int, error) {
	for {
		n, err := fpr.fp.f.ReadAt(buf, fpr.pos)
		fpr.pos += int64(n)
		if err == io.EOF {
			err = nil
		}
		if err != nil || n > 0 {
			return n, err
		}
		fpr.fp.mu.Lock()
		for fpr.pos >= fpr.fp.size && !fpr.fp.closing {
			fpr.fp.cond.Wait()
		}
		closing := fpr.pos >= fpr.fp.size && fpr.fp.closing // TODO: make a test for the bug that was here
		fpr.fp.mu.Unlock()
		if closing {
			return 0, io.EOF
		}
	}
}
{'fraction_non_alphanumeric': 0.14963797264682221, 'fraction_numerical': 0.006972378653794583, 'mean_word_length': 3.023732470334412, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 8, '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': '444466', 'n_tokens_mistral': 1564, 'n_tokens_neox': 1487, 'n_words': 440}