<commit_before>package subcmd


import (
    "net"
    "strings"
    "github.com/Lupino/periodic/driver"
    "github.com/Lupino/periodic/protocol"
    "fmt"
    "log"
    "bytes"
    "errors"
    "io"
    "os"
    "os/exec"
    "strconv"
    "time"
)


func Run(entryPoint, Func, cmd string) {
    parts := strings.SplitN(entryPoint, "://", 2)
    for {
        c, err := net.Dial(parts[0], parts[1])
        if err != nil {
            if err != io.EOF {
                log.Printf("Error: %s\n", err.Error())
            }
            log.Printf("Wait 5 second to reconnecting")
            time.Sleep(5 * time.Second)
            continue
        }
        conn := protocol.Conn{Conn: c}
        err = handleWorker(conn, Func, cmd)
        if err != nil {
            if err != io.EOF {
                log.Printf("Error: %s\n", err.Error())
            }
        }
        conn.Close()
    }
}


func handleWorker(conn protocol.Conn, Func, cmd string) (err error) {
    err = conn.Send(protocol.TYPE_WORKER.Bytes())
    if err != nil {
        return
    }
    var msgId = []byte("100")
    buf := bytes.NewBuffer(nil)
    buf.Write(msgId)
    buf.Write(protocol.NULL_CHAR)
    buf.WriteByte(byte(protocol.CAN_DO))
    buf.Write(protocol.NULL_CHAR)
    buf.WriteString(Func)
    err = conn.Send(buf.Bytes())
    if err != nil {
        return
    }

    var payload []byte
    var job driver.Job
    var jobHandle []byte
    for {
        buf = bytes.NewBuffer(nil)
        buf.Write(msgId)
        buf.Write(protocol.NULL_CHAR)
        buf.Write(protocol.GRAB_JOB.Bytes())
        err = conn.Send(buf.Bytes())
        if err != nil {
            return
        }
        payload, err = conn.Receive()
        if err != nil {
            return
        }
        job, jobHandle, err = extraJob(payload)
        realCmd := strings.Split(cmd, " ")
        realCmd = append(realCmd, job.Name)
        c := exec.Command(realCmd[0], realCmd[1:]...)
        c.Stdin = strings.NewReader(job.Args)
        var out bytes.Buffer
        c.Stdout = &out
        c.Stderr = os.Stderr
        err = c.Run()
        var schedLater int
        var fail = false
        for {
            line, err := out.ReadString([]byte("\n")[0])
            if err != nil {
                break
            }
            if strings.HasPrefix(line, "SCHED_LATER") {
                parts := strings.SplitN(line[:len(line) - 1], " ", 2)
                later := strings.Trim(parts[1], " ")
                schedLater, _ = strconv.Atoi(later)
            } else if strings.HasPrefix(line, "FAIL") {
                fail = true
            } else {
                fmt.Print(line)
            }
        }
        buf = bytes.NewBuffer(nil)
        buf.Write(msgId)
        buf.Write(protocol.NULL_CHAR)
        if err != nil || fail {
            buf.WriteByte(byte(protocol.JOB_FAIL))
        } else if schedLater > 0 {
            buf.WriteByte(byte(protocol.SCHED_LATER))
        } else {
            buf.WriteByte(byte(protocol.JOB_DONE))
        }
        buf.Write(protocol.NULL_CHAR)
        buf.Write(jobHandle)
        if schedLater > 0 {
            buf.Write(protocol.NULL_CHAR)
            buf.WriteString(strconv.Itoa(schedLater))
        }
        err = conn.Send(buf.Bytes())
        if err != nil {
            return
        }
    }
}


func extraJob(payload []byte) (job driver.Job, jobHandle []byte, err error) {
    parts := bytes.SplitN(payload, protocol.NULL_CHAR, 3)<commit_msg>Fix. the protocol name<commit_after>package subcmd


import (
    "net"
    "strings"
    "github.com/Lupino/periodic/driver"
    "github.com/Lupino/periodic/protocol"
    "fmt"
    "log"
    "bytes"
    "errors"
    "io"
    "os"
    "os/exec"
    "strconv"
    "time"
)


func Run(entryPoint, Func, cmd string) {
    parts := strings.SplitN(entryPoint, "://", 2)
    for {
        c, err := net.Dial(parts[0], parts[1])
        if err != nil {
            if err != io.EOF {
                log.Printf("Error: %s\n", err.Error())
            }
            log.Printf("Wait 5 second to reconnecting")
            time.Sleep(5 * time.Second)
            continue
        }
        conn := protocol.Conn{Conn: c}
        err = handleWorker(conn, Func, cmd)
        if err != nil {
            if err != io.EOF {
                log.Printf("Error: %s\n", err.Error())
            }
        }
        conn.Close()
    }
}


func handleWorker(conn protocol.Conn, Func, cmd string) (err error) {
    err = conn.Send(protocol.TYPE_WORKER.Bytes())
    if err != nil {
        return
    }
    var msgId = []byte("100")
    buf := bytes.NewBuffer(nil)
    buf.Write(msgId)
    buf.Write(protocol.NULL_CHAR)
    buf.WriteByte(byte(protocol.CAN_DO))
    buf.Write(protocol.NULL_CHAR)
    buf.WriteString(Func)
    err = conn.Send(buf.Bytes())
    if err != nil {
        return
    }

    var payload []byte
    var job driver.Job
    var jobHandle []byte
    for {
        buf = bytes.NewBuffer(nil)
        buf.Write(msgId)
        buf.Write(protocol.NULL_CHAR)
        buf.Write(protocol.GRAB_JOB.Bytes())
        err = conn.Send(buf.Bytes())
        if err != nil {
            return
        }
        payload, err = conn.Receive()
        if err != nil {
            return
        }
        job, jobHandle, err = extraJob(payload)
        realCmd := strings.Split(cmd, " ")
        realCmd = append(realCmd, job.Name)
        c := exec.Command(realCmd[0], realCmd[1:]...)
        c.Stdin = strings.NewReader(job.Args)
        var out bytes.Buffer
        c.Stdout = &out
        c.Stderr = os.Stderr
        err = c.Run()
        var schedLater int
        var fail = false
        for {
            line, err := out.ReadString([]byte("\n")[0])
            if err != nil {
                break
            }
            if strings.HasPrefix(line, "SCHED_LATER") {
                parts := strings.SplitN(line[:len(line) - 1], " ", 2)
                later := strings.Trim(parts[1], " ")
                schedLater, _ = strconv.Atoi(later)
            } else if strings.HasPrefix(line, "FAIL") {
                fail = true
            } else {
                fmt.Print(line)
            }
        }
        buf = bytes.NewBuffer(nil)
        buf.Write(msgId)
        buf.Write(protocol.NULL_CHAR)
        if err != nil || fail {
            buf.WriteByte(byte(protocol.WORK_FAIL))
        } else if schedLater > 0 {
            buf.WriteByte(byte(protocol.SCHED_LATER))
        } else {
            buf.WriteByte(byte(protocol.WORK_DONE))
        }
        buf.Write(protocol.NULL_CHAR)
        buf.Write(jobHandle)
        if schedLater > 0 {
            buf.Write(protocol.NULL_CHAR)
            buf.WriteString(strconv.Itoa(schedLater))
        }
        err = conn.Send(buf.Bytes())
        if err != nil {
            return
        }
    }
}


func extraJob(payload []byte) (job driver.Job, jobHandle []byte, err error) {
    parts := bytes.SplitN(payload, protocol.NULL_CHAR, 3)
{'fraction_non_alphanumeric': 0.13238464855541182, 'fraction_numerical': 0.004887164007474486, 'mean_word_length': 1.6089238845144358, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 7, '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': '742678', 'n_tokens_mistral': 2259, 'n_tokens_neox': 2191, 'n_words': 440}