<|start_filename|>pkg/proto/apicode.go<|end_filename|>
package proto

const (
	Code_Success     = 0
	Code_FailedAuth  = 5001
	Code_Error       = 5002
	Code_NotFound    = 5004
	Code_NotAllowed  = 5005
	Code_JWTError    = 5006
	Code_RPCError    = 5007
	Code_ParamsError = 5008
	Code_DBError     = 5009
)

const (
	Msg_NotAllowed = "permission not allowed"
	Msg_JWTError   = "parse jwt token failed"
)


<|start_filename|>jiacrontab_admin/runtime.go<|end_filename|>
package admin

import (
	"jiacrontab/pkg/proto"
	"jiacrontab/pkg/version"
)

func SystemInfo(ctx *myctx) {
	var (
		err     error
		info    map[string]interface{}
		reqBody SystemInfoReqParams
	)

	if err = ctx.Valid(&reqBody); err != nil {
		ctx.respBasicError(err)
		return
	}

	if err = rpcCall(reqBody.Addr, "Srv.SystemInfo", proto.EmptyArgs{}, &info); err != nil {
		ctx.respRPCError(err)
		return
	}
	info["version"] = version.String("jiacrontab")
	ctx.respSucc("", info)
}


<|start_filename|>pkg/rpc/client_test.go<|end_filename|>
package rpc

import (
	"jiacrontab/pkg/proto"
	"log"
	"net/http"
	_ "net/http/pprof"
	"sync"
	"testing"
	"time"
)

type Logic struct {
}

func (l *Logic) Ping(args *proto.EmptyArgs, reply *proto.EmptyReply) error {
	return nil
}

func (p *Logic) Say(args string, reply *string) error {

	*reply = "hello boy"
	time.Sleep(100 * time.Second)
	return nil
}
func TestCall(t *testing.T) {
	done := make(chan struct{})
	go func() {

		done <- struct{}{}

		log.Println("start server")
		err := listen(":6478", &Logic{})
		if err != nil {
			t.Fatal("server error:", err)
		}
	}()
	<-done
	time.Sleep(5 * time.Second)
	// 等待server启动
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(i int) {

			defer wg.Done()
			var ret string
			// var args string
			err := Call(":6478", "Logic.Say", "", &ret)
			if err != nil {
				log.Println(i, "error:", err)
			}
			t.Log(i, ret)
		}(i)

	}

	go func() {
		t.Log("listen :6060")
		t.Log(http.ListenAndServe(":6060", nil))
	}()

	wg.Wait()
	log.Println("end")
	time.Sleep(2 * time.Minute)
}


<|start_filename|>pkg/rpc/clients.go<|end_filename|>
package rpc

import (
	"context"
	"net/rpc"
	"sync"

	"github.com/iwannay/log"
)

var (
	defaultClients  *clients
	PingService     = "Srv.Ping"
	RegisterService = "Srv.Register"
)

type clients struct {
	lock    sync.RWMutex
	clients map[string]*Client
}

func (c *clients) get(addr string) *Client {
	var (
		cli *Client
		ok  bool
		op  ClientOptions
	)

	c.lock.Lock()
	defer c.lock.Unlock()
	if cli, ok = c.clients[addr]; ok {
		return cli
	}
	op.Network = "tcp4"
	op.Addr = addr
	cli = Dial(op)
	c.clients[addr] = cli
	go cli.Ping(PingService)

	return cli
}

func (c *clients) del(addr string) {
	c.lock.Lock()
	defer c.lock.Unlock()
	if cli, ok := c.clients[addr]; ok {
		cli.Close()
	}
	delete(c.clients, addr)
}

func Call(addr string, serviceMethod string, args interface{}, reply interface{}) error {
	err := defaultClients.get(addr).Call(serviceMethod, context.TODO(), args, reply)
	if err == rpc.ErrShutdown {
		log.Debug("rpc remove", addr)
		Del(addr)
	}
	return err
}

func CallCtx(addr string, serviceMethod string, ctx context.Context, args interface{}, reply interface{}) error {
	err := defaultClients.get(addr).Call(serviceMethod, ctx, args, reply)
	if err == rpc.ErrShutdown {
		log.Debug("rpc remove", addr)
		Del(addr)
	}
	return err
}

func Del(addr string) {
	if defaultClients != nil {
		defaultClients.del(addr)
	}
}

func DelNode(addr string) {
	if defaultClients != nil {
		defaultClients.del(addr)
	}
}

func init() {
	defaultClients = &clients{
		clients: make(map[string]*Client),
	}
}


<|start_filename|>pkg/pprof/pprof_windows.go<|end_filename|>
package pprof

func listenSignal() {
}


<|start_filename|>pkg/util/fn.go<|end_filename|>
package util

import (
	"jiacrontab/pkg/file"
	"reflect"
	"strconv"

	"github.com/gofrs/uuid"

	"fmt"
	"io/ioutil"
	"math/rand"

	"github.com/iwannay/log"

	"flag"
	"os"
	"path/filepath"
	"runtime"
	"time"
)

func RandIntn(end int) int {
	return rand.Intn(end)
}

func CurrentTime(t int64) string {
	if t == 0 {
		return "0"
	}
	return time.Unix(t, 0).Format("2006-01-02 15:04:05")
}

func SystemInfo(startTime time.Time) map[string]interface{} {
	var afterLastGC string
	goNum := runtime.NumGoroutine()
	cpuNum := runtime.NumCPU()
	mstat := &runtime.MemStats{}
	runtime.ReadMemStats(mstat)
	costTime := int(time.Since(startTime).Seconds())

	if mstat.LastGC != 0 {
		afterLastGC = fmt.Sprintf("%.1fs", float64(time.Now().UnixNano()-int64(mstat.LastGC))/1000/1000/1000)
	} else {
		afterLastGC = "0"
	}

	return map[string]interface{}{
		"服务运行时间":      fmt.Sprintf("%d天%d小时%d分%d秒", costTime/(3600*24), costTime%(3600*24)/3600, costTime%3600/60, costTime%(60)),
		"goroutine数量": goNum,
		"cpu核心数":      cpuNum,

		"当前内存使用量":  file.FileSize(int64(mstat.Alloc)),
		"所有被分配的内存": file.FileSize(int64(mstat.TotalAlloc)),
		"内存占用量":    file.FileSize(int64(mstat.Sys)),
		"指针查找次数":   mstat.Lookups,
		"内存分配次数":   mstat.Mallocs,
		"内存释放次数":   mstat.Frees,
		"距离上次GC时间": afterLastGC,

		// "当前 Heap 内存使用量": file.FileSize(int64(mstat.HeapAlloc)),
		// "Heap 内存占用量":    file.FileSize(int64(mstat.HeapSys)),
		// "Heap 内存空闲量":    file.FileSize(int64(mstat.HeapIdle)),
		// "正在使用的 Heap 内存": file.FileSize(int64(mstat.HeapInuse)),
		// "被释放的 Heap 内存":  file.FileSize(int64(mstat.HeapReleased)),
		// "Heap 对象数量":     mstat.HeapObjects,

		"下次GC内存回收量": file.FileSize(int64(mstat.NextGC)),
		"GC暂停时间总量":  fmt.Sprintf("%.3fs", float64(mstat.PauseTotalNs)/1000/1000/1000),
		"上次GC暂停时间":  fmt.Sprintf("%.3fs", float64(mstat.PauseNs[(mstat.NumGC+255)%256])/1000/1000/1000),
	}
}

func TryOpen(path string, flag int) (*os.File, error) {
	fabs, err := filepath.Abs(path)
	if err != nil {
		log.Errorf("TryOpen:", err)
		return nil, err
	}

	f, err := os.OpenFile(fabs, flag, 0644)
	if os.IsNotExist(err) {
		err = os.MkdirAll(filepath.Dir(fabs), 0755)
		if err != nil {
			return nil, err
		}
		return os.OpenFile(fabs, flag, 0644)
	}
	return f, err
}

func CatFile(filepath string, limit int64, content *string) (isPath bool, err error) {
	f, err := os.Open(filepath)

	if err != nil {
		return false, err
	}
	fi, err := f.Stat()
	if err != nil {
		return false, err
	}

	if fi.Size() > limit {
		*content = filepath
		return true, nil
	}
	data, err := ioutil.ReadAll(f)
	if err != nil {
		return false, err
	}
	*content = string(data)
	return false, nil
}

func ParseInt(i string) int {
	v, _ := strconv.Atoi(i)
	return v
}

func ParseInt64(i string) int64 {
	v, _ := strconv.Atoi(i)
	return int64(v)
}

func InArray(val interface{}, arr interface{}) bool {
	t := reflect.TypeOf(arr)
	v := reflect.ValueOf(arr)

	if t.Kind() == reflect.Slice {
		for i := 0; i < v.Len(); i++ {
			if v.Index(i).Interface() == val {
				return true
			}
		}
	}

	return false
}

func UUID() string {

	uu, err := uuid.NewGen().NewV1()

	if err != nil {
		log.Error(err)
		return fmt.Sprint(time.Now().UnixNano())
	}

	return uu.String()
}

func GetHostname() string {
	hostname, err := os.Hostname()
	if err != nil {
		log.Error("GetHostname:", err)
	}
	return hostname
}

func HasFlagName(fs *flag.FlagSet, s string) bool {
	var found bool
	fs.Visit(func(flag *flag.Flag) {
		if flag.Name == s {
			found = true
		}
	})
	return found

}

func init() {
	rand.Seed(time.Now().UnixNano())
}


<|start_filename|>pkg/finder/reader.go<|end_filename|>
package finder

import (
	"io"
	"os"
)

type TailReader struct {
	f     *os.File
	curr  int64
	isEOF bool
}

func (t *TailReader) Read(b []byte) (n int, err error) {
	if t.isEOF {
		return 0, io.EOF
	}

	off := t.curr - int64(len(b))
	if off < 0 {
		off = 0
		n, err = t.f.ReadAt(b[0:t.curr], off)
	} else {
		t.curr = off
		n, err = t.f.ReadAt(b, off)
	}

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

	invert(b[0:n])

	if off == 0 {
		t.isEOF = true
	}

	return
}

func NewTailReader(f *os.File, offset int64) io.Reader {
	return &TailReader{
		f:    f,
		curr: offset,
	}

}

func invert(b []byte) {
	for i, j := 0, len(b)-1; i < j; i, j = i+1, j-1 {
		b[i], b[j] = b[j], b[i]
	}
}


<|start_filename|>jiacrontabd/dependencies.go<|end_filename|>
package jiacrontabd

import (
	"bytes"
	"context"
	"jiacrontab/pkg/proto"
	"time"

	"github.com/iwannay/log"
)

type depEntry struct {
	jobID       uint   // 定时任务id
	jobUniqueID string // job的唯一标志
	processID   int    // 当前依赖的父级任务（可能存在多个并发的task)
	id          string // depID uuid
	once        bool
	workDir     string
	user        string
	env         []string
	from        string
	commands    []string
	dest        string
	done        bool
	timeout     int64
	err         error
	ctx         context.Context
	name        string
	logPath     string
	logContent  []byte
}

func newDependencies(jd *Jiacrontabd) *dependencies {
	return &dependencies{
		jd:  jd,
		dep: make(chan *depEntry, 100),
	}
}

type dependencies struct {
	jd  *Jiacrontabd
	dep chan *depEntry
}

func (d *dependencies) add(t *depEntry) {
	select {
	case d.dep <- t:
	default:
		log.Warnf("discard %v", t)
	}

}

func (d *dependencies) run() {
	go func() {
		for {
			select {
			case t := <-d.dep:
				go d.exec(t)
			}
		}
	}()
}

// TODO: 主任务退出杀死依赖
func (d *dependencies) exec(task *depEntry) {

	var (
		reply bool
		err   error
	)

	if task.timeout == 0 {
		// 默认超时10分钟
		task.timeout = 600
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(task.timeout)*time.Second)
	defer cancel()
	myCmdUnit := cmdUint{
		args:          [][]string{task.commands},
		ctx:           ctx,
		dir:           task.workDir,
		user:          task.user,
		logPath:       task.logPath,
		ignoreFileLog: true,
		jd:            d.jd,
		exportLog:     true,
	}

	log.Infof("dep start exec %s->%v", task.name, task.commands)
	task.err = myCmdUnit.launch()
	task.logContent = bytes.TrimRight(myCmdUnit.content, "\x00")
	task.done = true
	log.Infof("exec %s %s cost %.4fs %v", task.name, task.commands, float64(myCmdUnit.costTime)/1000000000, err)

	task.dest, task.from = task.from, task.dest

	if !d.jd.SetDependDone(task) {
		err = d.jd.rpcCallCtx(ctx, "Srv.SetDependDone", proto.DepJob{
			Name:        task.name,
			Dest:        task.dest,
			From:        task.from,
			ID:          task.id,
			JobUniqueID: task.jobUniqueID,
			ProcessID:   task.processID,
			JobID:       task.jobID,
			Commands:    task.commands,
			LogContent:  task.logContent,
			Err:         err,
			Timeout:     task.timeout,
		}, &reply)

		if err != nil {
			log.Error("Srv.SetDependDone error:", err, "server addr:", d.jd.getOpts().AdminAddr)
		}

		if !reply {
			log.Errorf("task %s %v call Srv.SetDependDone failed! err:%v", task.name, task.commands, err)
		}
	}
}


<|start_filename|>pkg/util/time.go<|end_filename|>
package util

func CountDaysOfMonth(year int, month int) (days int) {
	if month != 2 {
		if month == 4 || month == 6 || month == 9 || month == 11 {
			days = 30
		} else {
			days = 31
		}
	} else {
		if ((year%4) == 0 && (year%100) != 0) || (year%400) == 0 {
			days = 29
		} else {
			days = 28
		}
	}
	return
}


<|start_filename|>pkg/rpc/client.go<|end_filename|>
package rpc

import (
	"context"
	"errors"
	"jiacrontab/pkg/proto"
	"net"
	"net/rpc"
	"time"

	"github.com/iwannay/log"
)

const (
	diaTimeout   = 5 * time.Second
	callTimeout  = 1 * time.Minute
	pingDuration = 3 * time.Second
)

var (
	ErrRpc        = errors.New("rpc is not available")
	ErrRpcTimeout = errors.New("rpc call timeout")
	ErrRpcCancel  = errors.New("rpc call cancel")
	ErrShutdown   = rpc.ErrShutdown
)

type ClientOptions struct {
	Network string
	Addr    string
}

type Client struct {
	*rpc.Client
	options ClientOptions
	quit    chan struct{}
	err     error
}

func Dial(options ClientOptions) (c *Client) {
	c = &Client{}
	c.options = options
	c.dial()
	c.quit = make(chan struct{}, 100)
	return c
}

func (c *Client) dial() (err error) {
	conn, err := net.DialTimeout(c.options.Network, c.options.Addr, diaTimeout)
	if err != nil {
		return err
	}
	c.Client = rpc.NewClient(conn)
	return nil
}

func (c *Client) Call(serviceMethod string, ctx context.Context, args interface{}, reply interface{}) error {
	if serviceMethod != PingService && serviceMethod != RegisterService {
		log.Info("rpc call", c.options.Addr, serviceMethod)
	}

	if c.Client == nil {
		return ErrRpc
	}
	select {
	case <-ctx.Done():
		return ErrRpcCancel
	case call := <-c.Client.Go(serviceMethod, args, reply, make(chan *rpc.Call, 1)).Done:
		return call.Error
	case <-time.After(callTimeout):
		return ErrRpcTimeout
	}
}

func (c *Client) Error() error {
	return c.err
}

func (c *Client) Close() {
	c.quit <- struct{}{}
}

func (c *Client) Ping(serviceMethod string) {
	var (
		err error
	)
	for {
		select {
		case <-c.quit:
			goto closed
		default:
		}
		if c.Client != nil && c.err == nil {
			if err = c.Call(serviceMethod, context.TODO(), &proto.EmptyArgs{}, &proto.EmptyReply{}); err != nil {
				c.err = err
				c.Client.Close()
				log.Infof("client.Call(%s, args, reply) error (%v) \n", serviceMethod, err)
			}
		} else {
			if err = c.dial(); err == nil {
				c.err = nil
				log.Info("client reconnet ", c.options.Addr)
			}
		}
		time.Sleep(pingDuration)
	}
closed:
	log.Info("rpc quited", c.options.Addr)
	if c.Client != nil {
		c.Client.Close()
	}
}


<|start_filename|>pkg/proto/resp.go<|end_filename|>
package proto

const (
	SuccessRespCode = 0
	ErrorRespCode   = -1
)

type Resp struct {
	Code    int         `json:"code"`
	Msg     string      `json:"msg"`
	Data    interface{} `json:"data,omitempty"`
	Sign    string      `json:"sign"`
	Version string      `json:"version"`
}


<|start_filename|>pkg/crontab/crontab_test.go<|end_filename|>
package crontab

import (
	"encoding/json"
	"fmt"
	"testing"
	"time"
)

func Test_crontab_Ready(t *testing.T) {
	var timeLayout = "2006-01-02 15:04:05"
	c := New()
	now := time.Now().Add(6 * time.Second)
	c.AddTask(&Task{
		Value:    "test1" + now.Format(timeLayout),
		Priority: now.UnixNano(),
	})

	now = time.Now().Add(1 * time.Second)

	c.AddTask(&Task{
		Value:    "test2" + now.Format(timeLayout),
		Priority: now.UnixNano(),
	})
	now = time.Now().Add(3 * time.Second)

	c.AddTask(&Task{
		Value:    "test3" + now.Format(timeLayout),
		Priority: now.UnixNano(),
	})

	now = time.Now().Add(4 * time.Second)
	c.AddTask(&Task{
		Value:    "test4" + now.Format(timeLayout),
		Priority: now.UnixNano(),
	})

	now = time.Now().Add(3 * time.Second)
	c.AddTask(&Task{
		Value:    "test5" + now.Format(timeLayout),
		Priority: now.UnixNano(),
	})

	bts, _ := json.MarshalIndent(c.GetAllTask(), "", "")
	fmt.Println(string(bts))

	go c.QueueScanWorker()

	go func() {
		for v := range c.Ready() {
			bts, _ := json.MarshalIndent(v, "", "")
			fmt.Println(string(bts))
		}
	}()

	time.Sleep(10 * time.Second)
}


<|start_filename|>pkg/crontab/parse.go<|end_filename|>
package crontab

import (
	"fmt"
	"math"
	"strconv"
	"strings"
)

func getRange(expr string, r bounds) (uint64, error) {
	var (
		start, end, step uint
		rangeAndStep     = strings.Split(expr, "/")
		lowAndHigh       = strings.Split(rangeAndStep[0], "-")
		singleDigit      = len(lowAndHigh) == 1
		err              error
	)

	var extra uint64
	if lowAndHigh[0] == "*" || lowAndHigh[0] == "?" {
		start = r.min
		end = r.max
		extra = starBit
	} else {
		if lowAndHigh[0] == "L" {
			return 0, nil
		}
		start, err = parseIntOrName(lowAndHigh[0], r.names)
		if err != nil {
			return 0, err
		}

		switch len(lowAndHigh) {
		case 1:
			end = start
		case 2:
			end, err = parseIntOrName(lowAndHigh[1], r.names)
			if err != nil {
				return 0, err
			}
		default:
			return 0, fmt.Errorf("Too many hyphens: %s", expr)
		}
	}

	switch len(rangeAndStep) {
	case 1:
		step = 1
	case 2:
		step, err = mustParseInt(rangeAndStep[1])
		if err != nil {
			return 0, err
		}

		// Special handling: "N/step" means "N-max/step".
		if singleDigit {
			end = r.max
		}
	default:
		return 0, fmt.Errorf("Too many slashes: %s", expr)
	}

	if start < r.min {
		return 0, fmt.Errorf("Beginning of range (%d) below minimum (%d): %s", start, r.min, expr)
	}
	if end > r.max {
		return 0, fmt.Errorf("End of range (%d) above maximum (%d): %s", end, r.max, expr)
	}
	if start > end {
		return 0, fmt.Errorf("Beginning of range (%d) beyond end of range (%d): %s", start, end, expr)
	}
	if step == 0 {
		return 0, fmt.Errorf("Step of range should be a positive number: %s", expr)
	}

	return getBits(start, end, step) | extra, nil
}

func parseIntOrName(expr string, names map[string]uint) (uint, error) {
	if names != nil {
		if namedInt, ok := names[strings.ToLower(expr)]; ok {
			return namedInt, nil
		}
	}
	return mustParseInt(expr)
}

// mustParseInt parses the given expression as an int or returns an error.
func mustParseInt(expr string) (uint, error) {
	num, err := strconv.Atoi(expr)
	if err != nil {
		return 0, fmt.Errorf("Failed to parse int from %s: %s", expr, err)
	}
	if num < 0 {
		return 0, fmt.Errorf("Negative number (%d) not allowed: %s", num, expr)
	}

	return uint(num), nil
}

// getBits sets all bits in the range [min, max], modulo the given step size.
func getBits(min, max, step uint) uint64 {
	var bits uint64

	// If step is 1, use shifts.
	if step == 1 {
		return ^(math.MaxUint64 << (max + 1)) & (math.MaxUint64 << min)
	}

	// Else, use a simple loop.
	for i := min; i <= max; i += step {
		bits |= 1 << i
	}
	return bits
}

func getField(field string, r bounds) (uint64, error) {
	var bits uint64
	ranges := strings.FieldsFunc(field, func(r rune) bool { return r == ',' })
	for _, expr := range ranges {
		bit, err := getRange(expr, r)
		if err != nil {
			return bits, err
		}
		bits |= bit
	}
	return bits, nil
}


<|start_filename|>pkg/rpc/server.go<|end_filename|>
package rpc

import (
	"github.com/iwannay/log"
	"net"
	"net/rpc"
)

// listen Start rpc server
func listen(addr string, srcvr ...interface{}) error {
	var err error
	for _, v := range srcvr {
		if err = rpc.Register(v); err != nil {
			return err
		}
	}

	l, err := net.Listen("tcp4", addr)
	if err != nil {
		return err
	}
	defer func() {
		log.Info("listen rpc", addr, "close")
		if err := l.Close(); err != nil {
			log.Infof("listen.Close() error(%v)", err)
		}
	}()

	rpc.Accept(l)
	return nil
}

// ListenAndServe  run rpc server
func ListenAndServe(addr string, srcvr ...interface{}) {
	log.Info("rpc server listen:", addr)
	err := listen(addr, srcvr...)
	if err != nil {
		panic(err)
	}

}


<|start_filename|>pkg/base/storage.go<|end_filename|>
package base

import (
	"sync"
)

type (
	Storage struct {
		sync.Map
	}
)

func NewStorage() *Storage {
	return &Storage{}
}

func (s *Storage) All() map[string]interface{} {
	data := make(map[string]interface{})
	s.Range(func(key, value interface{}) bool {

		data[key.(string)] = value
		return true
	})
	return data
}

func (s *Storage) Exists(key interface{}) bool {

	_, ok := s.Load(key)

	return ok
}

func (s *Storage) GetUint64(key interface{}) (uint64, bool) {
	val, ok := s.Load(key)
	if !ok {
		return 0, false
	}

	ret, ok := val.(uint64)
	return ret, ok
}

func (s *Storage) Len() uint {
	var count uint
	s.Range(func(key, val interface{}) bool {
		count++
		return true
	})

	return count

}


<|start_filename|>pkg/kproc/proc_posix.go<|end_filename|>
// +build !windows

package kproc

import (
	"context"
	"os"
	"os/exec"
	"os/user"
	"strconv"
	"syscall"

	"github.com/iwannay/log"
)

func CommandContext(ctx context.Context, name string, arg ...string) *KCmd {
	cmd := exec.CommandContext(ctx, name, arg...)
	cmd.SysProcAttr = &syscall.SysProcAttr{}
	cmd.SysProcAttr.Setsid = true
	return &KCmd{
		ctx:                ctx,
		Cmd:                cmd,
		isKillChildProcess: true,
		done:               make(chan struct{}),
	}
}

// SetUser 设置执行用户要保证root权限
func (k *KCmd) SetUser(username string) {
	if username == "" {
		return
	}
	u, err := user.Lookup(username)
	if err != nil {
		log.Error("setUser error:", err)
		return
	}

	log.Infof("KCmd set uid=%s,gid=%s", u.Uid, u.Gid)
	k.SysProcAttr = &syscall.SysProcAttr{}
	uid, _ := strconv.Atoi(u.Uid)
	gid, _ := strconv.Atoi(u.Gid)
	k.SysProcAttr.Credential = &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}

}

func (k *KCmd) KillAll() {

	select {
	case k.done <- struct{}{}:
	default:
	}

	if k.Process == nil {
		return
	}

	if k.isKillChildProcess == false {
		return
	}

	group, err := os.FindProcess(-k.Process.Pid)
	if err == nil {
		group.Signal(syscall.SIGKILL)
	}
}

func (k *KCmd) Wait() error {
	defer k.KillAll()
	go func() {
		select {
		case <-k.ctx.Done():
			k.KillAll()
		case <-k.done:
		}
	}()
	return k.Cmd.Wait()
}


<|start_filename|>pkg/util/arr.go<|end_filename|>
package util

func FilterEmptyEle(in []string) (out []string) {
	for _, v := range in {
		if v != "" {
			out = append(out, v)
		}
	}
	return
}


<|start_filename|>pkg/crontab/job_test.go<|end_filename|>
package crontab

import (
	"jiacrontab/pkg/test"
	"testing"
	"time"
)

func TestJob_NextExecutionTime(t *testing.T) {
	timeLayout := "2006-01-02 15:04:05"
	j := &Job{
		Second:  "48",
		Minute:  "3",
		Hour:    "12",
		Day:     "25",
		Weekday: "*",
		Month:   "1",
	}

	tt, err := j.NextExecutionTime(time.Now())
	test.Nil(t, err)
	test.Equal(t, "2020-01-25 12:03:48", tt.Format(timeLayout))

	tt, err = j.NextExecutionTime(tt)
	test.Nil(t, err)
	test.Equal(t, "2021-01-25 12:03:48", tt.Format(timeLayout))

	tt, err = j.NextExecutionTime(tt)
	test.Equal(t, "2022-01-25 12:03:48", tt.Format(timeLayout))

	j = &Job{
		Second:  "58",
		Minute:  "*/4",
		Hour:    "12",
		Day:     "4",
		Weekday: "*",
		Month:   "3",
	}
	tt, err = j.NextExecutionTime(time.Now())
	test.Nil(t, err)
	test.Equal(t, "2020-03-04 12:00:58", tt.Format(timeLayout))

	tt, err = j.NextExecutionTime(tt)
	test.Nil(t, err)
	test.Equal(t, "2020-03-04 12:04:58", tt.Format(timeLayout))

	tt, err = j.NextExecutionTime(tt)
	test.Nil(t, err)
	test.Equal(t, "2020-03-04 12:08:58", tt.Format(timeLayout))

	j = &Job{
		Second:  "0",
		Minute:  "*",
		Hour:    "*",
		Day:     "*",
		Weekday: "*",
		Month:   "*",
	}

	tt, err = j.NextExecutionTime(time.Now())
	test.Nil(t, err)
	t.Log(tt, j.GetLastExecTime())
	for i := 0; i < 1000; i++ {
		tt, err = j.NextExecutionTime(tt)
		test.Nil(t, err)
		t.Log(tt, j.GetLastExecTime())
	}

	t.Log("end")
}


<|start_filename|>pkg/kproc/proc.go<|end_filename|>
package kproc

import (
	"context"
	"jiacrontab/pkg/file"
	"os/exec"
)

type KCmd struct {
	ctx context.Context
	*exec.Cmd
	isKillChildProcess bool
	done               chan struct{}
}

// SetEnv 设置环境变量
func (k *KCmd) SetEnv(env []string) {
	if len(env) == 0 {
		return
	}
	k.Cmd.Env = env
}

// SetDir 设置工作目录
func (k *KCmd) SetDir(dir string) {
	if dir == "" {
		return
	}
	if file.Exist(dir) == false {
		return
	}
	k.Cmd.Dir = dir
}

// SetExitKillChildProcess 设置主进程退出时是否kill子进程,默认kill
func (k *KCmd) SetExitKillChildProcess(ok bool) {
	k.isKillChildProcess = ok
}


<|start_filename|>pkg/crontab/crontab.go<|end_filename|>
package crontab

import (
	"container/heap"
	"errors"
	"jiacrontab/pkg/pqueue"
	"sync"
	"time"
)

// Task 任务
type Task = pqueue.Item

type Crontab struct {
	pq    pqueue.PriorityQueue
	mux   sync.RWMutex
	ready chan *Task
}

func New() *Crontab {
	return &Crontab{
		pq:    pqueue.New(10000),
		ready: make(chan *Task, 10000),
	}
}

// AddJob 添加未经处理的job
func (c *Crontab) AddJob(j *Job) error {
	nt, err := j.NextExecutionTime(time.Now())
	if err != nil {
		return errors.New("Invalid execution time")
	}
	c.mux.Lock()
	heap.Push(&c.pq, &Task{
		Priority: nt.UnixNano(),
		Value:    j,
	})
	c.mux.Unlock()
	return nil
}

// AddJob 添加延时任务
func (c *Crontab) AddTask(t *Task) {
	c.mux.Lock()
	heap.Push(&c.pq, t)
	c.mux.Unlock()
}

func (c *Crontab) Len() int {
	c.mux.RLock()
	len := len(c.pq)
	c.mux.RUnlock()
	return len
}

func (c *Crontab) GetAllTask() []*Task {
	c.mux.Lock()
	list := c.pq
	c.mux.Unlock()
	return list
}

func (c *Crontab) Ready() <-chan *Task {
	return c.ready
}

func (c *Crontab) QueueScanWorker() {
	refreshTicker := time.NewTicker(20 * time.Millisecond)
	for {
		select {
		case <-refreshTicker.C:
			if len(c.pq) == 0 {
				continue
			}
		start:
			c.mux.Lock()
			now := time.Now().UnixNano()
			job, _ := c.pq.PeekAndShift(now)
			c.mux.Unlock()
			if job == nil {
				continue
			}
			c.ready <- job
			goto start

		}
	}
}


<|start_filename|>jiacrontab_admin/debug.go<|end_filename|>
package admin

import (
	"jiacrontab/pkg/base"
	"net/http/pprof"
)

func stat(ctx *myctx) {
	data := base.Stat.Collect()
	ctx.JSON(data)
}

func pprofHandler(ctx *myctx) {
	if h := pprof.Handler(ctx.Params().Get("key")); h != nil {
		h.ServeHTTP(ctx.ResponseWriter(), ctx.Request())
	}
}

func indexDebug(ctx *myctx) {
	pprof.Index(ctx.ResponseWriter(), ctx.Request())
}


<|start_filename|>jiacrontabd/config.go<|end_filename|>
package jiacrontabd

import (
	"jiacrontab/pkg/file"
	"jiacrontab/pkg/util"
	"net"
	"reflect"

	"github.com/iwannay/log"

	ini "gopkg.in/ini.v1"
)

const (
	appname = "jiacrontabd"
)

type Config struct {
	LogLevel            string `opt:"log_level"`
	VerboseJobLog       bool   `opt:"verbose_job_log"`
	ListenAddr          string `opt:"listen_addr"`
	AdminAddr           string `opt:"admin_addr"`
	LogPath             string `opt:"log_path"`
	AutoCleanTaskLog    bool   `opt:"auto_clean_task_log"`
	NodeName            string `opt:"node_name"`
	BoardcastAddr       string `opt:"boardcast_addr"`
	ClientAliveInterval int    `opt:"client_alive_interval"`
	CfgPath             string
	Debug               bool `opt:"debug"`
	iniFile             *ini.File
	DriverName          string `opt:"driver_name"`
	DSN                 string `opt:"dsn"`
}

func (c *Config) Resolve() error {
	c.iniFile = c.loadConfig(c.CfgPath)

	val := reflect.ValueOf(c).Elem()
	typ := val.Type()

	for i := 0; i < typ.NumField(); i++ {
		field := typ.Field(i)
		opt := field.Tag.Get("opt")
		if opt == "" {
			continue
		}
		sec := c.iniFile.Section("jiacrontabd")

		if !sec.HasKey(opt) {
			continue
		}

		key := sec.Key(opt)
		switch field.Type.Kind() {
		case reflect.Bool:
			v, err := key.Bool()
			if err != nil {
				log.Errorf("cannot resolve ini field %s err(%v)", opt, err)
			}
			val.Field(i).SetBool(v)
		case reflect.String:
			val.Field(i).SetString(key.String())
		case reflect.Int, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Int8:
			if v, err := key.Int64(); err != nil {
				log.Errorf("cannot convert to int64 type (%s)", err)
			} else {
				val.Field(i).SetInt(v)
			}
		}
	}

	if c.BoardcastAddr == "" {
		_, port, _ := net.SplitHostPort(c.ListenAddr)
		c.BoardcastAddr = util.InternalIP() + ":" + port
	}

	return nil
}

func NewConfig() *Config {
	return &Config{
		LogLevel:            "warn",
		VerboseJobLog:       true,
		ListenAddr:          "127.0.0.1:20001",
		AdminAddr:           "127.0.0.1:20003",
		LogPath:             "./logs",
		AutoCleanTaskLog:    true,
		NodeName:            util.GetHostname(),
		CfgPath:             "./jiacrontabd.ini",
		DriverName:          "sqlite3",
		DSN:                 "data/jiacrontabd.db",
		ClientAliveInterval: 30,
	}
}

func (c *Config) loadConfig(path string) *ini.File {
	if !file.Exist(path) {
		f, err := file.CreateFile(path)
		if err != nil {
			panic(err)
		}
		f.Close()
	}

	iniFile, err := ini.Load(path)
	if err != nil {
		panic(err)
	}
	return iniFile
}


<|start_filename|>pkg/proto/daemon.go<|end_filename|>
package proto

import (
	"jiacrontab/models"
)

type EditDaemonJobArgs struct {
	Job     models.DaemonJob
	GroupID uint
	UserID  uint
	Root    bool
}
