<commit_before>package run

import (
	"log"
	"bufio"
	"io"
	"os/exec"
	"sync"
)

// Run runs the command and returns a channel of output lines, errors and result of cmd.Wait
func Run(cmd *exec.Cmd) (
	lines chan string, errors chan error, resultCh chan error) {
	lines = make(chan string)
	errors = make(chan error, 1)
	resultCh = make(chan error)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		errors <- err
		return
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		errors <- err
		return
	}

	err = cmd.Start()
	if err != nil {
		errors <- err
		return
	}

	var wg sync.WaitGroup
	go tailReader(bufio.NewReader(stdout), lines, errors, wg)
	go tailReader(bufio.NewReader(stderr), lines, errors, wg)

	go func() {
		wg.Wait()
		resultCh <- cmd.Wait()
	}()

	return
}

func tailReader(
	r *bufio.Reader, ch chan string, errCh chan error,
	wg sync.WaitGroup) {
	defer wg.Done()
	wg.Add(1)

	for {
		line, _, err := r.ReadLine()
		if err != nil {
			log.Printf("** %T | %#v | %v\n", err, err, err)

			if err != io.EOF {
				errCh <- err
			}
			break
		}
		ch <- string(line)
	}
}<commit_msg>WaitGroup.Add must be called from the main goroutine. also pass it by address. this magically gets rid of the EBADF error<commit_after>package run

import (
	"bufio"
	"io"
	"os/exec"
	"sync"
)

// Run runs the command and returns a channel of output lines, errors and result of cmd.Wait
func Run(cmd *exec.Cmd) (
	lines chan string, errors chan error, resultCh chan error) {
	lines = make(chan string)
	errors = make(chan error, 1)
	resultCh = make(chan error)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		errors <- err
		return
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		errors <- err
		return
	}

	err = cmd.Start()
	if err != nil {
		errors <- err
		return
	}

	go func() {
		var wg sync.WaitGroup
		wg.Add(2)
		go tailReader(bufio.NewReader(stdout), lines, errors, &wg)
		go tailReader(bufio.NewReader(stderr), lines, errors, &wg)
		wg.Wait()
		resultCh <- cmd.Wait()
	}()

	return
}

func tailReader(
	r *bufio.Reader, ch chan string, errCh chan error,
	wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		line, _, err := r.ReadLine()
		if err != nil {
			// log.Printf("** %T | %#v | %v\n", err, err, err)

			if err != io.EOF {
				errCh <- err
			}
			break
		}
		ch <- string(line)
	}
}
{'fraction_non_alphanumeric': 0.14267185473411154, 'fraction_numerical': 0.0017293558149589277, 'mean_word_length': 2.982788296041308, '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': '4161032', 'n_tokens_mistral': 974, 'n_tokens_neox': 880, 'n_words': 275}