From 59ad1a9b003b5899f8a5df6ed5b7e78e04256c7b Mon Sep 17 00:00:00 2001 From: John Date: Sun, 5 May 2019 22:57:13 +0800 Subject: [PATCH] fix issue in incorrect every running logics of gcron.AddOnce --- TODO.MD | 1 + g/os/gcron/gcron.go | 52 +++++++++++++---------- g/os/gcron/gcron_cron.go | 65 +++++++++++++++++------------ g/os/gcron/gcron_entry.go | 73 +++++++++++++++++++++------------ g/os/gcron/gcron_unit_1_test.go | 16 +++++++- g/os/gtimer/gtimer.go | 2 +- g/os/gtimer/gtimer_entry.go | 5 ++- geg/os/gcron/gcron2.go | 7 ++-- 8 files changed, 139 insertions(+), 82 deletions(-) diff --git a/TODO.MD b/TODO.MD index 6ac97bca3..1a4d09dc1 100644 --- a/TODO.MD +++ b/TODO.MD @@ -45,6 +45,7 @@ 1. gredis增加cluster支持； 1. gset.Add/Remove/Contains方法增加批量操作支持； 1. gmlock增加手动清理机制：当内存锁不再使用时，由调用端决定是否清理内存锁； +1. gtimer增加DelayAdd*方法返回Entry对象，以便DelayAdd*的定时任务也能进行状态控制；gcron同理需要改进； # DONE 1. gconv完善针对不同类型的判断，例如：尽量减少sprintf("%v", xxx)来执行string类型的转换； diff --git a/g/os/gcron/gcron.go b/g/os/gcron/gcron.go index 03c07f410..4369b2245 100644 --- a/g/os/gcron/gcron.go +++ b/g/os/gcron/gcron.go @@ -5,8 +5,6 @@ // You can obtain one at https://github.com/gogf/gf. // Package gcron implements a cron pattern parser and job runner. -// -// 定时任务. package gcron import ( @@ -25,96 +23,108 @@ const ( ) var ( - // 默认的cron管理对象 + // Default cron object. defaultCron = New() ) -// 设置日志输出路径 +// SetLogPath sets the logging folder path for default cron object. func SetLogPath(path string) { defaultCron.SetLogPath(path) } -// 获取设置的日志输出路径 +// GetLogPath returns the logging folder path of default cron object. func GetLogPath() string { return defaultCron.GetLogPath() } -// 设置日志输出等级。 +// SetLogLevel sets the logging level for default cron object. func SetLogLevel(level int) { defaultCron.SetLogLevel(level) } -// 获取日志输出等级。 +// GetLogLevel returns the logging level for default cron object. func GetLogLevel() int { return defaultCron.GetLogLevel() } -// 添加定时任务，可以给定名字，以便于后续执行删除 +// Add adds a timed task to default cron object. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func Add(pattern string, job func(), name ... string) (*Entry, error) { return defaultCron.Add(pattern, job, name...) } -// 添加单例运行定时任务 +// AddSingleton adds a singleton timed task, to default cron object. +// A singleton timed task is that can only be running one single instance at the same time. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func AddSingleton(pattern string, job func(), name ... string) (*Entry, error) { return defaultCron.AddSingleton(pattern, job, name...) } -// 添加只运行一次的定时任务 +// AddOnce adds a timed task which can be run only once, to default cron object. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func AddOnce(pattern string, job func(), name ... string) (*Entry, error) { return defaultCron.AddOnce(pattern, job, name...) } -// 添加运行指定次数的定时任务 +// AddTimes adds a timed task which can be run specified times, to default cron object. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func AddTimes(pattern string, times int, job func(), name ... string) (*Entry, error) { return defaultCron.AddTimes(pattern, times, job, name...) } -// 延迟添加定时任务 +// DelayAdd adds a timed task to default cron object after time. func DelayAdd(delay time.Duration, pattern string, job func(), name ... string) { defaultCron.DelayAdd(delay, pattern, job, name...) } -// 延迟添加单例定时任务，delay参数单位为秒 +// DelayAddSingleton adds a singleton timed task after time to default cron object. func DelayAddSingleton(delay time.Duration, pattern string, job func(), name ... string) { defaultCron.DelayAddSingleton(delay, pattern, job, name...) } -// 延迟添加只运行一次的定时任务，delay参数单位为秒 +// DelayAddOnce adds a timed task after time to default cron object. +// This timed task can be run only once. func DelayAddOnce(delay time.Duration, pattern string, job func(), name ... string) { defaultCron.DelayAddOnce(delay, pattern, job, name...) } -// 延迟添加运行指定次数的定时任务，delay参数单位为秒 +// DelayAddTimes adds a timed task after time to default cron object. +// This timed task can be run specified times. func DelayAddTimes(delay time.Duration, pattern string, times int, job func(), name ... string) { defaultCron.DelayAddTimes(delay, pattern, times, job, name...) } -// 检索指定名称的定时任务 +// Search returns a scheduled task with the specified . +// It returns nil if no found. func Search(name string) *Entry { return defaultCron.Search(name) } -// 根据指定名称删除定时任务 +// Remove deletes scheduled task which named . func Remove(name string) { defaultCron.Remove(name) } -// 获取所有已注册的定时任务数量 +// Size returns the size of the timed tasks of default cron. func Size() int { return defaultCron.Size() } -// 获取所有已注册的定时任务项 +// Entries return all timed tasks as slice. func Entries() []*Entry { return defaultCron.Entries() } -// 启动指定的定时任务 +// Start starts running the specified timed task named . func Start(name string) { defaultCron.Start(name) } -// 停止指定的定时任务 +// Stop stops running the specified timed task named . func Stop(name string) { defaultCron.Stop(name) } diff --git a/g/os/gcron/gcron_cron.go b/g/os/gcron/gcron_cron.go index bb7eb2a06..b1a21aa88 100644 --- a/g/os/gcron/gcron_cron.go +++ b/g/os/gcron/gcron_cron.go @@ -17,16 +17,15 @@ import ( "time" ) -// 定时任务管理对象 type Cron struct { - idGen *gtype.Int64 // 用于唯一名称生成 - status *gtype.Int // 定时任务状态(0: 未执行; 1: 运行中; 2: 已停止; -1:删除关闭) - entries *gmap.StringInterfaceMap // 所有的定时任务项 - logPath *gtype.String // 日志文件输出目录 - logLevel *gtype.Int // 日志输出等级 + idGen *gtype.Int64 // Used for unique name generation. + status *gtype.Int // Timed task status(0: Not Start; 1: Running; 2: Stopped; -1: Closed) + entries *gmap.StringInterfaceMap // All timed task entries. + logPath *gtype.String // Logging path(folder). + logLevel *gtype.Int // Logging level. } -// 创建自定义的定时任务管理对象 +// New returns a new Cron object with default settings. func New() *Cron { return &Cron { idGen : gtype.NewInt64(), @@ -37,37 +36,42 @@ func New() *Cron { } } -// 设置日志输出路径 +// SetLogPath sets the logging folder path. func (c *Cron) SetLogPath(path string) { c.logPath.Set(path) } -// 获取设置的日志输出路径 +// GetLogPath return the logging folder path. func (c *Cron) GetLogPath() string { return c.logPath.Val() } -// 设置日志输出等级。 +// SetLogLevel sets the logging level. func (c *Cron) SetLogLevel(level int) { c.logLevel.Set(level) } -// 获取日志输出等级。 +// GetLogLevel returns the logging level. func (c *Cron) GetLogLevel() int { return c.logLevel.Val() } -// 添加定时任务 +// Add adds a timed task. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func (c *Cron) Add(pattern string, job func(), name ... string) (*Entry, error) { if len(name) > 0 { if c.Search(name[0]) != nil { return nil, errors.New(fmt.Sprintf(`cron job "%s" already exists`, name[0])) } } - return c.addEntry(pattern, job, false, gDEFAULT_TIMES, name...) + return c.addEntry(pattern, job, false, name...) } -// 添加单例运行定时任务 +// AddSingleton adds a singleton timed task. +// A singleton timed task is that can only be running one single instance at the same time. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func (c *Cron) AddSingleton(pattern string, job func(), name ... string) (*Entry, error) { if entry, err := c.Add(pattern, job, name ...); err != nil { return nil, err @@ -77,7 +81,9 @@ func (c *Cron) AddSingleton(pattern string, job func(), name ... string) (*Entry } } -// 添加只运行一次的定时任务 +// AddOnce adds a timed task which can be run only once. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func (c *Cron) AddOnce(pattern string, job func(), name ... string) (*Entry, error) { if entry, err := c.Add(pattern, job, name ...); err != nil { return nil, err @@ -87,7 +93,9 @@ func (c *Cron) AddOnce(pattern string, job func(), name ... string) (*Entry, err } } -// 添加运行指定次数的定时任务 +// AddTimes adds a timed task which can be run specified times. +// A unique can be bound with the timed task. +// It returns and error if the is already used. func (c *Cron) AddTimes(pattern string, times int, job func(), name ... string) (*Entry, error) { if entry, err := c.Add(pattern, job, name ...); err != nil { return nil, err @@ -97,7 +105,7 @@ func (c *Cron) AddTimes(pattern string, times int, job func(), name ... string) } } -// 延迟添加定时任务 +// DelayAdd adds a timed task after time. func (c *Cron) DelayAdd(delay time.Duration, pattern string, job func(), name ... string) { gtimer.AddOnce(delay, func() { if _, err := c.Add(pattern, job, name ...); err != nil { @@ -106,7 +114,7 @@ func (c *Cron) DelayAdd(delay time.Duration, pattern string, job func(), name .. }) } -// 延迟添加单例定时任务 +// DelayAddSingleton adds a singleton timed task after time. func (c *Cron) DelayAddSingleton(delay time.Duration, pattern string, job func(), name ... string) { gtimer.AddOnce(delay, func() { if _, err := c.AddSingleton(pattern, job, name ...); err != nil { @@ -115,7 +123,8 @@ func (c *Cron) DelayAddSingleton(delay time.Duration, pattern string, job func() }) } -// 延迟添加运行指定次数的定时任务 +// DelayAddOnce adds a timed task after time. +// This timed task can be run only once. func (c *Cron) DelayAddOnce(delay time.Duration, pattern string, job func(), name ... string) { gtimer.AddOnce(delay, func() { if _, err := c.AddOnce(pattern, job, name ...); err != nil { @@ -124,7 +133,8 @@ func (c *Cron) DelayAddOnce(delay time.Duration, pattern string, job func(), nam }) } -// 延迟添加只运行一次的定时任务 +// DelayAddTimes adds a timed task after time. +// This timed task can be run specified times. func (c *Cron) DelayAddTimes(delay time.Duration, pattern string, times int, job func(), name ... string) { gtimer.AddOnce(delay, func() { if _, err := c.AddTimes(pattern, times, job, name ...); err != nil { @@ -133,7 +143,8 @@ func (c *Cron) DelayAddTimes(delay time.Duration, pattern string, times int, job }) } -// 检索指定名称的定时任务 +// Search returns a scheduled task with the specified . +// It returns nil if no found. func (c *Cron) Search(name string) *Entry { if v := c.entries.Get(name); v != nil { return v.(*Entry) @@ -141,7 +152,7 @@ func (c *Cron) Search(name string) *Entry { return nil } -// 开启定时任务执行(可以指定特定名称的一个或若干个定时任务) +// Start starts running the specified timed task named . func (c *Cron) Start(name...string) { if len(name) > 0 { for _, v := range name { @@ -154,7 +165,7 @@ func (c *Cron) Start(name...string) { } } -// 停止定时任务执行(可以指定特定名称的一个或若干个定时任务) +// Stop stops running the specified timed task named . func (c *Cron) Stop(name...string) { if len(name) > 0 { for _, v := range name { @@ -167,24 +178,24 @@ func (c *Cron) Stop(name...string) { } } -// 根据指定名称删除定时任务。 +// Remove deletes scheduled task which named . func (c *Cron) Remove(name string) { if v := c.entries.Get(name); v != nil { v.(*Entry).Close() } } -// 关闭定时任务 +// Close stops and closes current cron. func (c *Cron) Close() { c.status.Set(STATUS_CLOSED) } -// 获取所有已注册的定时任务数量 +// Size returns the size of the timed tasks. func (c *Cron) Size() int { return c.entries.Size() } -// 获取所有已注册的定时任务项(按照注册时间从小到大进行排序) +// Entries return all timed tasks as slice(order by registered time asc). func (c *Cron) Entries() []*Entry { array := garray.NewSortedArraySize(c.entries.Size(), func(v1, v2 interface{}) int { entry1 := v1.(*Entry) diff --git a/g/os/gcron/gcron_entry.go b/g/os/gcron/gcron_entry.go index b4802eba0..ecc880089 100644 --- a/g/os/gcron/gcron_entry.go +++ b/g/os/gcron/gcron_entry.go @@ -7,12 +7,13 @@ package gcron import ( - "github.com/gogf/gf/g/os/glog" - "github.com/gogf/gf/g/os/gtimer" - "github.com/gogf/gf/g/util/gconv" - "reflect" - "runtime" - "time" + "github.com/gogf/gf/g/container/gtype" + "github.com/gogf/gf/g/os/glog" + "github.com/gogf/gf/g/os/gtimer" + "github.com/gogf/gf/g/util/gconv" + "reflect" + "runtime" + "time" ) // Timed task entry. @@ -21,6 +22,7 @@ type Entry struct { entry *gtimer.Entry // Associated gtimer.Entry. schedule *cronSchedule // Timed schedule object. jobName string // Callback function name(address info). + times *gtype.Int // Running times limit. Name string // Entry name. Job func() `json:"-"` // Callback function. Time time.Time // Registered time. @@ -29,17 +31,18 @@ type Entry struct { // addEntry creates and returns a new Entry object. // Param is the callback function for timed task execution. // Param specifies whether timed task executing in singleton mode. -// Param limits the times for timed task executing. // Param names this entry for manual control. -func (c *Cron) addEntry(pattern string, job func(), singleton bool, times int, name ... string) (*Entry, error) { +func (c *Cron) addEntry(pattern string, job func(), singleton bool, name ... string) (*Entry, error) { schedule, err := newSchedule(pattern) if err != nil { return nil, err } + // No limit for , for gtimer checking scheduling every second. entry := &Entry { cron : c, schedule : schedule, jobName : runtime.FuncForPC(reflect.ValueOf(job).Pointer()).Name(), + times : gtype.NewInt(gDEFAULT_TIMES), Job : job, Time : time.Now(), } @@ -48,82 +51,98 @@ func (c *Cron) addEntry(pattern string, job func(), singleton bool, times int, n } else { entry.Name = "gcron-" + gconv.String(c.idGen.Add(1)) } - entry.entry = gtimer.AddEntry(time.Second, entry.check, singleton, times, gtimer.STATUS_STOPPED) - entry.entry.Start() + // When you add a scheduled task, you cannot allow it to run. + // It cannot start running when added to gtimer. + // It should start running after the entry is added to the entries map, + // to avoid the task from running during adding where the entries + // does not have the entry information, which might cause panic. + entry.entry = gtimer.AddEntry(time.Second, entry.check, singleton, -1, gtimer.STATUS_STOPPED) c.entries.Set(entry.Name, entry) + entry.entry.Start() return entry, nil } -// 是否单例运行 +// IsSingleton return whether this entry is a singleton timed task. func (entry *Entry) IsSingleton() bool { return entry.entry.IsSingleton() } -// 设置单例运行 +// SetSingleton sets the entry running in singleton mode. func (entry *Entry) SetSingleton(enabled bool) { entry.entry.SetSingleton(true) } -// 设置任务的运行次数 +// SetTimes sets the times which the entry can run. func (entry *Entry) SetTimes(times int) { - entry.entry.SetTimes(times) + entry.times.Set(times) } -// 定时任务状态 +// Status returns the status of entry. func (entry *Entry) Status() int { return entry.entry.Status() } -// 设置定时任务状态, 返回设置之前的状态 +// SetStatus sets the status of the entry. func (entry *Entry) SetStatus(status int) int { return entry.entry.SetStatus(status) } -// 启动定时任务 +// Start starts running the entry. func (entry *Entry) Start() { entry.entry.Start() } -// 停止定时任务 +// Stop stops running the entry. func (entry *Entry) Stop() { entry.entry.Stop() } -// 关闭定时任务 +// Close stops and removes the entry from cron. func (entry *Entry) Close() { entry.cron.entries.Remove(entry.Name) entry.entry.Close() } -// 定时任务检查执行 +// Timed task check execution. +// The running times limits feature is implemented by gcron.Entry and cannot be implemented by gtimer.Entry. +// gcron.Entry relies on gtimer to implement a scheduled task check for gcron.Entry per second. func (entry *Entry) check() { if entry.schedule.meet(time.Now()) { path := entry.cron.GetLogPath() level := entry.cron.GetLogLevel() - // 检查定时任务对象状态(非任务状态) switch entry.cron.status.Val() { case STATUS_STOPPED: return case STATUS_CLOSED: - entry.cron.Remove(entry.Name) - glog.Path(path).Level(level).Debugfln("[gcron] %s(%s) %s remove", entry.Name, entry.schedule.pattern, entry.jobName) - gtimer.Exit() + glog.Path(path).Level(level).Debugfln("[gcron] %s(%s) %s removed", entry.Name, entry.schedule.pattern, entry.jobName) + entry.Close() case STATUS_READY: fallthrough case STATUS_RUNNING: + // Running times check. + times := entry.times.Add(-1) + if times <= 0 { + if entry.entry.SetStatus(STATUS_CLOSED) == STATUS_CLOSED || times < 0 { + return + } + } + if times < 2000000000 && times > 1000000000 { + entry.times.Set(gDEFAULT_TIMES) + } glog.Path(path).Level(level).Debugfln("[gcron] %s(%s) %s start", entry.Name, entry.schedule.pattern, entry.jobName) defer func() { - if entry.entry.Status() == STATUS_CLOSED { - entry.cron.Remove(entry.Name) - } if err := recover(); err != nil { glog.Path(path).Level(level).Errorfln("[gcron] %s(%s) %s end with error: %v", entry.Name, entry.schedule.pattern, entry.jobName, err) } else { glog.Path(path).Level(level).Debugfln("[gcron] %s(%s) %s end", entry.Name, entry.schedule.pattern, entry.jobName) } + if entry.entry.Status() == STATUS_CLOSED { + entry.Close() + } }() entry.Job() + } } } diff --git a/g/os/gcron/gcron_unit_1_test.go b/g/os/gcron/gcron_unit_1_test.go index 4fb93494d..9a378ca47 100644 --- a/g/os/gcron/gcron_unit_1_test.go +++ b/g/os/gcron/gcron_unit_1_test.go @@ -122,7 +122,7 @@ func TestCron_AddSingleton(t *testing.T) { } -func TestCron_AddOnce(t *testing.T) { +func TestCron_AddOnce1(t *testing.T) { gtest.Case(t, func() { cron := gcron.New() array := garray.New() @@ -139,6 +139,20 @@ func TestCron_AddOnce(t *testing.T) { }) } +func TestCron_AddOnce2(t *testing.T) { + gtest.Case(t, func() { + cron := gcron.New() + array := garray.New() + cron.AddOnce("@every 2s", func() { + array.Append(1) + }) + gtest.Assert(cron.Size(), 1) + time.Sleep(2500*time.Millisecond) + gtest.Assert(array.Len(), 1) + gtest.Assert(cron.Size(), 0) + }) +} + func TestCron_AddTimes(t *testing.T) { gtest.Case(t, func() { cron := gcron.New() diff --git a/g/os/gtimer/gtimer.go b/g/os/gtimer/gtimer.go index 7c5418062..9454a4c48 100644 --- a/g/os/gtimer/gtimer.go +++ b/g/os/gtimer/gtimer.go @@ -99,7 +99,7 @@ func DelayAddTimes(delay time.Duration, interval time.Duration, times int, job J defaultTimer.DelayAddTimes(delay, interval, times, job) } -// 在Job方法中调用，停止当前运行的任务。 +// 在Job方法中调用，停止并删除当前运行的任务。 func Exit() { panic(gPANIC_EXIT) } diff --git a/g/os/gtimer/gtimer_entry.go b/g/os/gtimer/gtimer_entry.go index 0b6c9ba62..4174805c2 100644 --- a/g/os/gtimer/gtimer_entry.go +++ b/g/os/gtimer/gtimer_entry.go @@ -29,7 +29,11 @@ type Entry struct { type JobFunc = func() // 创建定时任务。 +// 如果times参数<=0，表示不限制运行次数。 func (w *wheel) addEntry(interval time.Duration, job JobFunc, singleton bool, times int, status int) *Entry { + if times <= 0 { + times = gDEFAULT_TIMES + } ms := interval.Nanoseconds()/1e6 num := ms/w.intervalMs if num == 0 { @@ -172,7 +176,6 @@ func (entry *Entry) check(nowTicks int64, nowMs int64) (runnable, addable bool) } // 是否不限制运行次数 if times < 2000000000 && times > 1000000000 { - times = gDEFAULT_TIMES entry.times.Set(gDEFAULT_TIMES) } return true, true diff --git a/geg/os/gcron/gcron2.go b/geg/os/gcron/gcron2.go index 7390efd68..931317e40 100644 --- a/geg/os/gcron/gcron2.go +++ b/geg/os/gcron/gcron2.go @@ -1,20 +1,19 @@ package main import ( - "github.com/gogf/gf/g" "github.com/gogf/gf/g/os/gcron" + "github.com/gogf/gf/g/os/glog" "time" ) func test() { - + glog.Println(111) } func main() { - _, err := gcron.Add("*/10 * * * * ?", test) + _, err := gcron.AddOnce("@every 2s", test) if err != nil { panic(err) } - g.Dump(gcron.Entries()) time.Sleep(10 * time.Second) }