<commit_before>package pygo

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"os/exec"
)

const (
	binary = "python2.7"
	//code   = "import pygo; pygo.wrap('%s')"
	code = "import os; f = os.fdopen(4, 'w').write('%s')"
)

type pygoImpl struct {
	binPath string
	module  string
	ps      *os.Process

	chanin  *os.File
	chanout *os.File
	chanerr *os.File

	//only filled if process exited.
	stderr string
	state  *os.ProcessState
}

func NewPy(module string) (Pygo, error) {
	path, err := exec.LookPath(binary)
	if err != nil {
		return nil, err
	}

	py := &pygoImpl{
		binPath: path,
		module:  module,
	}

	err = py.init()
	if err != nil {
		return nil, err
	}

	go py.wait()

	return py, nil
}

func (py *pygoImpl) wait() {
	data, err := ioutil.ReadAll(py.chanerr)
	if err != nil {
		log.Println(err)
	}

	py.stderr = string(data)

	state, _ := py.ps.Wait()
	py.state = state
}

//init opes the pipes and start the python process.
func (py *pygoImpl) init() error {
	stderrReader, stderrWriter, err := os.Pipe()

	if err != nil {
		return err
	}

	pyIn, goOut, err := os.Pipe()
	if err != nil {
		return err
	}

	goIn, pyOut, err := os.Pipe()
	if err != nil {
		return err
	}

	attr := &os.ProcAttr{
		Files: []*os.File{nil, nil, stderrWriter, pyIn, pyOut},
	}

	ps, err := os.StartProcess(py.binPath, []string{
		binary,
		"-c",
		fmt.Sprintf(code, py.module)},
		attr)

	if err != nil {
		return err
	}

	py.ps = ps
	py.chanin = goOut
	py.chanout = goIn
	py.chanerr = stderrReader

	return nil
}

func (py *pygoImpl) Error() string {
	return py.stderr
}

func (py *pygoImpl) Do(fnc string, args map[string]interface{}) (interface{}, error) {
	if py.state != nil {
		return nil, fmt.Errorf("Can't execute python code, python process has exited", py.stderr)
	}

	data := map[string]interface{}{
		"func": fnc,
		"args": args,
	}

	bytes, err := json.Marshal(data)
	if err != nil {
		return nil, err
	}

	n, err := py.chanin.Write(bytes)
	if err != nil {
		return nil, err
	}

	//read response.
	buffer := make([]byte, 1000)
	n, err = py.chanout.Read(buffer)

	if err != nil && err != io.EOF {
		return nil, err
	}
	//log.Println(n, buffer)

	return string(buffer[:n]), nil
}<commit_msg>Process calls in synced matter<commit_after>package pygo

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"os/exec"
)

const (
	binary = "python2.7"
	//code   = "import pygo; pygo.wrap('%s')"
	code = "import os; f = os.fdopen(4, 'w').write('%s')"
)

type response struct {
	value interface{}
	err   error
}

type call struct {
	function string
	args     map[string]interface{}
	response chan *response
}

type pygoImpl struct {
	binPath string
	module  string
	ps      *os.Process

	chanin  *os.File
	chanout *os.File
	chanerr *os.File

	//only filled if process exited.
	stderr string
	state  *os.ProcessState

	channel chan *call
}

func NewPy(module string) (Pygo, error) {
	path, err := exec.LookPath(binary)
	if err != nil {
		return nil, err
	}

	py := &pygoImpl{
		binPath: path,
		module:  module,
		channel: make(chan *call),
	}

	err = py.init()
	if err != nil {
		return nil, err
	}

	go py.wait()
	go py.process()

	return py, nil
}

func (py *pygoImpl) wait() {
	data, err := ioutil.ReadAll(py.chanerr)
	if err != nil {
		log.Println(err)
	}

	py.stderr = string(data)

	state, _ := py.ps.Wait()
	py.state = state
}

//init opes the pipes and start the python process.
func (py *pygoImpl) init() error {
	stderrReader, stderrWriter, err := os.Pipe()

	if err != nil {
		return err
	}

	pyIn, goOut, err := os.Pipe()
	if err != nil {
		return err
	}

	goIn, pyOut, err := os.Pipe()
	if err != nil {
		return err
	}

	attr := &os.ProcAttr{
		Files: []*os.File{nil, nil, stderrWriter, pyIn, pyOut},
	}

	ps, err := os.StartProcess(py.binPath, []string{
		binary,
		"-c",
		fmt.Sprintf(code, py.module)},
		attr)

	if err != nil {
		return err
	}

	py.ps = ps
	py.chanin = goOut
	py.chanout = goIn
	py.chanerr = stderrReader

	return nil
}

func (py *pygoImpl) Error() string {
	return py.stderr
}

func (py *pygoImpl) processSingle() {
	c := <-py.channel

	var response response

	defer func() {
		c.response <- &response
	}()

	data := map[string]interface{}{
		"function": c.function,
		"args":     c.args,
	}

	bytes, err := json.Marshal(data)
	if err != nil {
		response.err = err
		return
	}

	n, err := py.chanin.Write(bytes)
	if err != nil {
		response.err = err
		return
	}

	//read response.
	buffer := make([]byte, 1000)
	n, err = py.chanout.Read(buffer)

	if err != nil && err != io.EOF {
		response.err = err
		return
	}

	response.value = buffer[:n]
}

func (py *pygoImpl) process() {
	for {
		py.processSingle()
	}
}

func (py *pygoImpl) Do(function string, args map[string]interface{}) (interface{}, error) {
	if py.state != nil {
		return nil, fmt.Errorf("Can't execute python code, python process has exited", py.stderr)
	}

	responseChan := make(chan *response)
	call := call{
		function: function,
		args:     args,
		response: responseChan,
	}
	py.channel <- &call
	response := <-responseChan
	return response.value, response.err
}
{'fraction_non_alphanumeric': 0.1566004328152666, 'fraction_numerical': 0.0027542789691127286, 'mean_word_length': 3.1232765612327658, 'pattern_counts': {'":': 4, '<': 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': '1123893', 'n_tokens_mistral': 2123, 'n_tokens_neox': 1997, 'n_words': 553}