<commit_before>package tty

import (
	"bufio"
	"io"
	"os"
	"syscall"

	"github.com/elves/elvish/sys"
)

const (
	asyncReaderChanSize int = 128
)

// AsyncReader delivers a Unix fd stream to a channel of runes.
type AsyncReader struct {
	rd           *os.File
	bufrd        *bufio.Reader
	rCtrl, wCtrl *os.File
	ctrlCh       chan struct{}
	ch           chan rune
	errCh        chan error
}

// NewAsyncReader creates a new AsyncReader from a file.
func NewAsyncReader(rd *os.File) *AsyncReader {
	rCtrl, wCtrl, err := os.Pipe()
	if err != nil {
		panic(err)
	}
	return &AsyncReader{
		rd,
		bufio.NewReaderSize(rd, 0),
		rCtrl, wCtrl,
		make(chan struct{}),
		make(chan rune, asyncReaderChanSize),
		make(chan error),
	}
}

// Chan returns a channel onto which the AsyncReader writes the runes it reads.
func (ar *AsyncReader) Chan() <-chan rune {
	return ar.ch
}

// ErrorChan returns a channel onto which the AsyncReader writes the errors it
// encounters.
func (ar *AsyncReader) ErrorChan() <-chan error {
	return ar.errCh
}

// Run runs the AsyncReader. It blocks until Quit is called and should be
// called in a separate goroutine.
func (ar *AsyncReader) Run() {
	fd := ar.rd.Fd()
	cfd := ar.rCtrl.Fd()
	poller := sys.Poller{}
	var cBuf [1]byte

	// unix-specific, no equavilent on windows
	if nonblock, _ := sys.GetNonblock(int(fd)); !nonblock {
		sys.SetNonblock(int(fd), true)
		defer sys.SetNonblock(int(fd), false)
	}

	if err := poller.Init([]uintptr{fd, cfd}, []uintptr{}); err != nil {
		// fatal error, unable to initialize poller
		// TODO show erorr
		return
	}

	for {
		rfds, _, err := poller.Poll(nil)
		if err != nil {
			switch err {
			case syscall.EINTR:
				continue
			default:
				ar.errCh <- err
				return
			}
		}
		for _, rfd := range *rfds {
			if rfd == cfd {
				// Consume the written byte
				ar.rCtrl.Read(cBuf[:])
				<-ar.ctrlCh
				return
			}
		}
	ReadRune:
		for {
			r, _, err := ar.bufrd.ReadRune()
			switch err {
			case nil:
				// Logger.Printf("read rune: %q", r)
				select {
				case ar.ch <- r:
				case <-ar.ctrlCh:
					ar.rCtrl.Read(cBuf[:])
					return
				}
			case io.EOF:
				return
			default:
				// BUG(xiaq): AsyncReader relies on the undocumented fact
				// that (*os.File).Read returns an *os.File.PathError
				patherr, ok := err.(*os.PathError) //.Err
				if ok && patherr.Err == syscall.EWOULDBLOCK || patherr.Err == syscall.EAGAIN {
					break ReadRune
				} else {
					select {
					case ar.errCh <- err:
					case <-ar.ctrlCh:
						ar.rCtrl.Read(cBuf[:])
						return
					}
				}
			}
		}
	}
}

// Quit terminates the loop of Run.
func (ar *AsyncReader) Quit() {
	_, err := ar.wCtrl.Write([]byte{'q'})
	if err != nil {
		panic(err)
	}
	ar.ctrlCh <- struct{}{}
}

// Close releases files and channels associated with the AsyncReader. It does
// not close the file used to create it.
func (ar *AsyncReader) Close() {
	ar.rCtrl.Close()
	ar.wCtrl.Close()
	close(ar.ctrlCh)
	close(ar.ch)
}

func max(a, b int) int {
	if a >= b {
		return a
	}<commit_msg>Fix AsyncReader for go 1.9.<commit_after>package tty

import (
	"os"
	"syscall"

	"github.com/elves/elvish/sys"
)

const (
	asyncReaderChanSize int = 128
)

// AsyncReader delivers a Unix fd stream to a channel of runes.
type AsyncReader struct {
	rd           *os.File
	rCtrl, wCtrl *os.File
	ctrlCh       chan struct{}
	ch           chan rune
	errCh        chan error
}

// NewAsyncReader creates a new AsyncReader from a file.
func NewAsyncReader(rd *os.File) *AsyncReader {
	rCtrl, wCtrl, err := os.Pipe()
	if err != nil {
		panic(err)
	}
	return &AsyncReader{
		rd,
		rCtrl, wCtrl,
		make(chan struct{}),
		make(chan rune, asyncReaderChanSize),
		make(chan error),
	}
}

// Chan returns a channel onto which the AsyncReader writes the runes it reads.
func (ar *AsyncReader) Chan() <-chan rune {
	return ar.ch
}

// ErrorChan returns a channel onto which the AsyncReader writes the errors it
// encounters.
func (ar *AsyncReader) ErrorChan() <-chan error {
	return ar.errCh
}

// Run runs the AsyncReader. It blocks until Quit is called and should be
// called in a separate goroutine.
func (ar *AsyncReader) Run() {
	fd := ar.rd.Fd()
	cfd := ar.rCtrl.Fd()
	poller := sys.Poller{}
	var cBuf [1]byte

	// unix-specific, no equavilent on windows
	if nonblock, _ := sys.GetNonblock(int(fd)); !nonblock {
		sys.SetNonblock(int(fd), true)
		defer sys.SetNonblock(int(fd), false)
	}

	if err := poller.Init([]uintptr{fd, cfd}, []uintptr{}); err != nil {
		// fatal error, unable to initialize poller
		ar.waitForQuit(err)
		return
	}

	for {
		rfds, _, err := poller.Poll(nil)
		if err != nil {
			switch err {
			case syscall.EINTR:
				continue
			default:
				ar.waitForQuit(err)
				return
			}
		}
		for _, rfd := range *rfds {
			if rfd == cfd {
				// Consume the written byte
				ar.rCtrl.Read(cBuf[:])
				<-ar.ctrlCh
				return
			}
		}

		bytes := make([]byte, 0, 32)
	ReadRunes:
		for {
			buf := make([]byte, 32)
			nr, err := syscall.Read(int(fd), buf[:])

			if err == nil {
				bytes = append(bytes, buf[:nr]...)
			} else {
				if err == syscall.EAGAIN || err == syscall.EWOULDBLOCK {
					// All input read, break the loop.
					break ReadRunes
				}
				// Write error to errCh, unless termination is requested.
				select {
				case ar.errCh <- err:
				case <-ar.ctrlCh:
					ar.rCtrl.Read(cBuf[:])
					return
				}
			}
		}
		// TODO(xiaq): Invalid UTF-8 will result in a bunch of \ufffd, which is
		// not helpful for debugging.
		for _, r := range string(bytes) {
			// Write error to ch, unless termination is requested.
			select {
			case ar.ch <- r:
			case <-ar.ctrlCh:
				ar.rCtrl.Read(cBuf[:])
				return
			}
		}
	}
}

func (ar *AsyncReader) waitForQuit(err error) {
	var cBuf [1]byte

	select {
	case ar.errCh <- err:
	case <-ar.ctrlCh:
		ar.rCtrl.Read(cBuf[:])
		return
	}
	<-ar.ctrlCh
	ar.rCtrl.Read(cBuf[:])
}

// Quit terminates the loop of Run.
func (ar *AsyncReader) Quit() {
	_, err := ar.wCtrl.Write([]byte{'q'})
	if err != nil {
		panic(err)
	}
	ar.ctrlCh <- struct{}{}
}

// Close releases files and channels associated with the AsyncReader. It does
// not close the file used to create it.
func (ar *AsyncReader) Close() {
	ar.rCtrl.Close()
	ar.wCtrl.Close()
	close(ar.ctrlCh)
	close(ar.ch)
}

func max(a, b int) int {
	if a >= b {
		return a
	}
{'fraction_non_alphanumeric': 0.13767427122940432, 'fraction_numerical': 0.0028517110266159697, 'mean_word_length': 2.6875, 'pattern_counts': {'":': 0, '<': 23, '<?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': '5182783', 'n_tokens_mistral': 2686, 'n_tokens_neox': 2315, 'n_words': 686}