<commit_before>// Ottily executes a javascript snippet on each line of an input file in parallel.
//
// Noop:
//
//     $ ottily datasets/simple.ldj
//     {"name": "ottily", "language": "Golang"}
//
// Inline script with -e:
//
//     $ ottily -e 'output=input.length' datasets/simple.ldj
//	   40
//
//     $ ottily -e 'o=JSON.parse(input); o["language"] = "Go"; output=JSON.stringify(o);' datasets/simple.ldj
//     {"language":"Go","name":"ottily"}
//
// Pass a script file:
//
//     $ ottily -s scripts/classified.js datasets/simple.ldj
//     CLASSIFIED
//
package main

import (
	"bufio"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"runtime"
	"runtime/pprof"
	"strings"
	"sync"

	"github.com/robertkrimen/otto"
)

const NOOP_SCRIPT = "output = input"
const VERSION = "0.1.2"

func Worker(lines, out chan *string, script, preload string, wg *sync.WaitGroup) {
	defer wg.Done()
	vm := otto.New()

	if preload != "" {
		_, err := vm.Run(preload)
		if err != nil {
			log.Fatal(err)
		}
	}

	compiled, err := vm.Compile("", script)
	if err != nil {
		log.Fatal(err)
	}

	for line := range lines {
		vm.Set("input", *line)
		_, err := vm.Run(compiled)
		if err != nil {
			log.Fatal(err)
		}
		result, err := vm.Get("output")
		if result == otto.NullValue() {
			continue
		}
		if err != nil {
			log.Fatal(err)
		}
		r := result.String()
		out <- &r
	}
}

// FanInWriter writes the channel content to the writer
func FanInWriter(writer io.Writer, in chan *string, done chan bool) {
	for s := range in {
		writer.Write([]byte(*s))
		writer.Write([]byte("\n"))
	}
	done <- true
}

func main() {
	preload := flag.String("p", "", "include this js file before processing")
	script := flag.String("s", "", "script to execute on each line of input")
	execute := flag.String("e", "", "execute argument on each line of input")
	numWorkers := flag.Int("w", runtime.NumCPU(), "number of workers")
	version := flag.Bool("v", false, "prints current program version")
	cpuprofile := flag.String("cpuprofile", "", "write cpu profile to file")
	limit := flag.Int("l", -1, "maximum number of docs to convert")

	flag.Parse()

	if *cpuprofile != "" {
		f, err := os.Create(*cpuprofile)
		if err != nil {
			log.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	if *version {
		fmt.Println(VERSION)
		os.Exit(0)
	}

	if flag.NArg() < 1 {
		log.Fatal("input file required")
	}

	preloadContent := ""

	if *preload != "" {
		ff, err := os.Open(*preload)
		if err != nil {
			log.Fatal(err)
		}
		b, err := ioutil.ReadAll(ff)
		if err != nil {
			log.Fatal(err)
		}
		preloadContent = string(b)
	}

	content := NOOP_SCRIPT

	if *script != "" {
		ff, err := os.Open(*script)
		if err != nil {
			log.Fatal(err)
		}
		b, err := ioutil.ReadAll(ff)
		if err != nil {
			log.Fatal(err)
		}
		content = string(b)
	}

	if *execute != "" {
		content = *execute
	}

	ff, err := os.Open(flag.Arg(0))
	if err != nil {
		log.Fatal(err)
	}
	defer ff.Close()
	reader := bufio.NewReader(ff)

	if *numWorkers > 0 {
		runtime.GOMAXPROCS(*numWorkers)
	}

	queue := make(chan *string)
	out := make(chan *string)
	done := make(chan bool)
	var wg sync.WaitGroup

	writer := bufio.NewWriter(os.Stdout)
	defer writer.Flush()
	go FanInWriter(writer, out, done)

	for i := 0; i < *numWorkers; i++ {
		wg.Add(1)
		go Worker(queue, out, content, preloadContent, &wg)
	}

	counter := 0
	for {
		if *limit != -1 && counter == *limit {
			break
		}
		line, err := reader.ReadString('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatal(err)
		}
		line = strings.TrimSpace(line)
		queue <- &line
		counter++
	}
	close(queue)
	wg.Wait()
	close(out)
	<-done
}<commit_msg>use 0 as default for no limit<commit_after>// Ottily executes a javascript snippet on each line of an input file in parallel.
//
// Noop:
//
//     $ ottily datasets/simple.ldj
//     {"name": "ottily", "language": "Golang"}
//
// Inline script with -e:
//
//     $ ottily -e 'output=input.length' datasets/simple.ldj
//	   40
//
//     $ ottily -e 'o=JSON.parse(input); o["language"] = "Go"; output=JSON.stringify(o);' datasets/simple.ldj
//     {"language":"Go","name":"ottily"}
//
// Pass a script file:
//
//     $ ottily -s scripts/classified.js datasets/simple.ldj
//     CLASSIFIED
//
package main

import (
	"bufio"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"runtime"
	"runtime/pprof"
	"strings"
	"sync"

	"github.com/robertkrimen/otto"
)

const NOOP_SCRIPT = "output = input"
const VERSION = "0.1.3"

func Worker(lines, out chan *string, script, preload string, wg *sync.WaitGroup) {
	defer wg.Done()
	vm := otto.New()

	if preload != "" {
		_, err := vm.Run(preload)
		if err != nil {
			log.Fatal(err)
		}
	}

	compiled, err := vm.Compile("", script)
	if err != nil {
		log.Fatal(err)
	}

	for line := range lines {
		vm.Set("input", *line)
		_, err := vm.Run(compiled)
		if err != nil {
			log.Fatal(err)
		}
		result, err := vm.Get("output")
		if result == otto.NullValue() {
			continue
		}
		if err != nil {
			log.Fatal(err)
		}
		r := result.String()
		out <- &r
	}
}

// FanInWriter writes the channel content to the writer
func FanInWriter(writer io.Writer, in chan *string, done chan bool) {
	for s := range in {
		writer.Write([]byte(*s))
		writer.Write([]byte("\n"))
	}
	done <- true
}

func main() {
	preload := flag.String("p", "", "include this js file before processing")
	script := flag.String("s", "", "script to execute on each line of input")
	execute := flag.String("e", "", "execute argument on each line of input")
	numWorkers := flag.Int("w", runtime.NumCPU(), "number of workers")
	version := flag.Bool("v", false, "prints current program version")
	cpuprofile := flag.String("cpuprofile", "", "write cpu profile to file")
	limit := flag.Int("l", 0, "maximum number of docs to convert (0 no limit)")

	flag.Parse()

	if *cpuprofile != "" {
		f, err := os.Create(*cpuprofile)
		if err != nil {
			log.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	if *version {
		fmt.Println(VERSION)
		os.Exit(0)
	}

	if flag.NArg() < 1 {
		log.Fatal("input file required")
	}

	preloadContent := ""

	if *preload != "" {
		ff, err := os.Open(*preload)
		if err != nil {
			log.Fatal(err)
		}
		b, err := ioutil.ReadAll(ff)
		if err != nil {
			log.Fatal(err)
		}
		preloadContent = string(b)
	}

	content := NOOP_SCRIPT

	if *script != "" {
		ff, err := os.Open(*script)
		if err != nil {
			log.Fatal(err)
		}
		b, err := ioutil.ReadAll(ff)
		if err != nil {
			log.Fatal(err)
		}
		content = string(b)
	}

	if *execute != "" {
		content = *execute
	}

	ff, err := os.Open(flag.Arg(0))
	if err != nil {
		log.Fatal(err)
	}
	defer ff.Close()
	reader := bufio.NewReader(ff)

	if *numWorkers > 0 {
		runtime.GOMAXPROCS(*numWorkers)
	}

	queue := make(chan *string)
	out := make(chan *string)
	done := make(chan bool)
	var wg sync.WaitGroup

	writer := bufio.NewWriter(os.Stdout)
	defer writer.Flush()
	go FanInWriter(writer, out, done)

	for i := 0; i < *numWorkers; i++ {
		wg.Add(1)
		go Worker(queue, out, content, preloadContent, &wg)
	}

	counter := 0
	for {
		if *limit != 0 && counter == *limit {
			break
		}
		line, err := reader.ReadString('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatal(err)
		}
		line = strings.TrimSpace(line)
		queue <- &line
		counter++
	}
	close(queue)
	wg.Wait()
	close(out)
	<-done
}
{'fraction_non_alphanumeric': 0.180040733197556, 'fraction_numerical': 0.004073319755600814, 'mean_word_length': 3.0076169749727963, 'pattern_counts': {'":': 8, '<': 15, '<?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': '4074355', 'n_tokens_mistral': 3121, 'n_tokens_neox': 2852, 'n_words': 715}