From 6aa1c5b1ebe3854ed08d4beb5df13d1a2ac9e6c3 Mon Sep 17 00:00:00 2001 From: John Date: Wed, 20 Feb 2019 16:07:11 +0800 Subject: [PATCH 1/3] ghttp updates --- g/g_setting.go | 16 ++++++++ g/net/ghttp/ghttp_server.go | 23 +++++++---- g/os/gfcache/gfcache_cache.go | 19 +-------- g/os/gproc/gproc.go | 6 ++- g/os/gproc/gproc_comm_receive.go | 70 ++++++++++++++++---------------- g/os/gproc/gproc_comm_send.go | 18 ++++---- 6 files changed, 81 insertions(+), 71 deletions(-) create mode 100644 g/g_setting.go diff --git a/g/g_setting.go b/g/g_setting.go new file mode 100644 index 000000000..c12f43b98 --- /dev/null +++ b/g/g_setting.go @@ -0,0 +1,16 @@ +// Copyright 2019 gf Author(https://github.com/gogf/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package g + +import "github.com/gogf/gf/g/net/ghttp" + +// SetServerGraceful enables/disables graceful reload feature of ghttp Web Server. +// +// 是否开启WebServer的平滑重启特性。 +func SetServerGraceful(enabled bool) { + ghttp.SetGraceful(enabled) +} diff --git a/g/net/ghttp/ghttp_server.go b/g/net/ghttp/ghttp_server.go index 3c5cf92ad..952b3ac01 100644 --- a/g/net/ghttp/ghttp_server.go +++ b/g/net/ghttp/ghttp_server.go @@ -140,17 +140,24 @@ var ( doneChan = make(chan struct{}, 1000) // 用于服务进程初始化，只能初始化一次，采用“懒初始化”(在server运行时才初始化) - serverProcInited = gtype.NewBool() + serverProcessInited = gtype.NewBool() + + // 是否开启WebServer平滑重启特性, 会开启额外的本地端口监听，用于进程管理通信 + gracefulEnabled = true ) +// 是否开启平滑重启特性 +func SetGraceful(enabled bool) { + gracefulEnabled = enabled +} // Web Server进程初始化. // 注意该方法不能放置于包初始化方法init中，不使用ghttp.Server的功能便不能初始化对应的协程goroutine逻辑. -func serverProcInit() { - if serverProcInited.Val() { +func serverProcessInit() { + if serverProcessInited.Val() { return } - serverProcInited.Set(true) + serverProcessInited.Set(true) // 如果是完整重启，那么需要等待主进程销毁后，才开始执行监听，防止端口冲突 if genv.Get(gADMIN_ACTION_RESTART_ENVKEY) != "" { if p, e := os.FindProcess(gproc.PPid()); e == nil { @@ -164,7 +171,9 @@ func serverProcInit() { // 信号量管理操作监听 go handleProcessSignal() // 异步监听进程间消息 - go handleProcessMessage() + if gracefulEnabled { + go handleProcessMessage() + } } // 获取/创建一个默认配置的HTTP Server(默认监听端口是80) @@ -207,7 +216,7 @@ func GetServer(name...interface{}) (*Server) { // 需要结合Wait方式一起使用 func (s *Server) Start() error { // 服务进程初始化，只会初始化一次 - serverProcInit() + serverProcessInit() // 当前Web Server状态判断 if s.Status() == SERVER_STATUS_RUNNING { @@ -253,7 +262,7 @@ func (s *Server) Start() error { if gproc.IsChild() { gtimer.SetTimeout(2*time.Second, func() { if err := gproc.Send(gproc.PPid(), []byte("exit"), gADMIN_GPROC_COMM_GROUP); err != nil { - panic(err) + glog.Error("ghttp server error in process communication:", err) } }) } diff --git a/g/os/gfcache/gfcache_cache.go b/g/os/gfcache/gfcache_cache.go index 315d0c6ac..092e93f7f 100644 --- a/g/os/gfcache/gfcache_cache.go +++ b/g/os/gfcache/gfcache_cache.go @@ -47,31 +47,16 @@ func (c *Cache) GetBinContents(path string) []byte { return b } -// 添加文件监控 +// 添加文件监控，一旦文件有变化立即清除缓存，下一次读取的时候再执行缓存。 func (c *Cache) addMonitor(path string) { // 防止多goroutine同时调用 if c.cache.Contains(path) { return } gfsnotify.Add(path, func(event *gfsnotify.Event) { - //glog.Debug("gfcache:", event) - length := 0 if r := c.cache.Get(path); r != nil { - length = len(r.([]byte)) - } - // 是否删除 - if event.IsRemove() { c.cache.Remove(path) - c.size.Add(-length) - return - } - // 更新缓存内容 - if c.cap.Val() == 0 || c.size.Val() < c.cap.Val() { - b := gfile.GetBinContents(path) - if len(b) > 0 { - c.size.Add(len(b) - length) - c.cache.Set(path, b) - } + c.size.Add(-len(r.([]byte))) } }) } \ No newline at end of file diff --git a/g/os/gproc/gproc.go b/g/os/gproc/gproc.go index c1bb2e27c..69dad15cf 100644 --- a/g/os/gproc/gproc.go +++ b/g/os/gproc/gproc.go @@ -27,8 +27,10 @@ const ( gPROC_TEMP_DIR_ENV_KEY = "GPROC_TEMP_DIR" ) -// 进程开始执行时间 -var processStartTime = time.Now() +var ( + // 进程开始执行时间 + processStartTime = time.Now() +) // 获取当前进程ID func Pid() int { diff --git a/g/os/gproc/gproc_comm_receive.go b/g/os/gproc/gproc_comm_receive.go index e8a7195ac..759faae95 100644 --- a/g/os/gproc/gproc_comm_receive.go +++ b/g/os/gproc/gproc_comm_receive.go @@ -3,6 +3,7 @@ // This Source Code Form is subject to the terms of the MIT License. // If a copy of the MIT was not distributed with this file, // You can obtain one at https://github.com/gogf/gf. + // "不要通过共享内存来通信，而应该通过通信来共享内存" @@ -26,16 +27,43 @@ const ( ) var ( - // 是否已开启TCP端口监听服务(使用int而非bool，以便于使用原子操作判断是否开启) - tcpListeningCount = gtype.NewInt() + // 是否已开启TCP端口监听服务 + tcpListened = gtype.NewBool() ) +// 获取其他进程传递到当前进程的消息包，阻塞执行。 +// 进程只有在执行该方法后才会打开请求端口，默认情况下不允许进程间通信。 +func Receive(group...string) *Msg { + // 一个进程只能开启一个监听goroutine + if tcpListened.Set(true) == false { + go startTcpListening() + } + queue := (*gqueue.Queue)(nil) + groupName := gPROC_COMM_DEAFULT_GRUOP_NAME + if len(group) > 0 { + groupName = group[0] + } + if v := commReceiveQueues.Get(groupName); v == nil { + commReceiveQueues.LockFunc(func(m map[string]interface{}) { + if v, ok := m[groupName]; ok { + queue = v.(*gqueue.Queue) + } else { + queue = gqueue.New(gPROC_MSG_QUEUE_MAX_LENGTH) + m[groupName] = queue + } + }) + } else { + queue = v.(*gqueue.Queue) + } + + if v := queue.Pop(); v != nil { + return v.(*Msg) + } + return nil +} + // 创建本地进程TCP通信服务 func startTcpListening() { - // 一个进程只能开启一个监听goroutine - if tcpListeningCount.Add(1) != 1 { - return - } var listen *net.TCPListener for i := gPROC_DEFAULT_TCP_PORT; ; i++ { addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("127.0.0.1:%d", i)) @@ -48,7 +76,6 @@ func startTcpListening() { } // 将监听的端口保存到通信文件中(字符串类型存放) gfile.PutContents(getCommFilePath(Pid()), gconv.String(i)) - //glog.Printfln("%d: gproc listening on [%s]", Pid(), addr) break } for { @@ -133,32 +160,3 @@ func bufferToMsgs(buffer []byte) []*Msg { return msgs } -// 获取其他进程传递到当前进程的消息包，阻塞执行。 -func Receive(group...string) *Msg { - // 开启接收协程时才会开启端口监听 - go startTcpListening() - - var queue *gqueue.Queue - groupName := gPROC_COMM_DEAFULT_GRUOP_NAME - if len(group) > 0 { - groupName = group[0] - } - - if v := commReceiveQueues.Get(groupName); v == nil { - commReceiveQueues.LockFunc(func(m map[string]interface{}) { - if v, ok := m[groupName]; ok { - queue = v.(*gqueue.Queue) - } else { - queue = gqueue.New(gPROC_MSG_QUEUE_MAX_LENGTH) - m[groupName] = queue - } - }) - } else { - queue = v.(*gqueue.Queue) - } - - if v := queue.Pop(); v != nil { - return v.(*Msg) - } - return nil -} \ No newline at end of file diff --git a/g/os/gproc/gproc_comm_send.go b/g/os/gproc/gproc_comm_send.go index c38efc6e0..bb7873200 100644 --- a/g/os/gproc/gproc_comm_send.go +++ b/g/os/gproc/gproc_comm_send.go @@ -7,16 +7,16 @@ package gproc import ( - "github.com/gogf/gf/g/net/gtcp" - "github.com/gogf/gf/g/os/gfile" - "github.com/gogf/gf/g/util/gconv" - "github.com/gogf/gf/g/encoding/gbinary" - "fmt" - "errors" - "time" "bytes" + "errors" + "fmt" + "github.com/gogf/gf/g/encoding/gbinary" + "github.com/gogf/gf/g/net/gtcp" + "github.com/gogf/gf/g/os/gfcache" "github.com/gogf/gf/g/os/glog" + "github.com/gogf/gf/g/util/gconv" "io" + "time" ) const ( @@ -26,7 +26,7 @@ const ( gPROC_COMM_DEAFULT_GRUOP_NAME = "" // 默认分组名称 ) -// 向指定gproc进程发送数据 +// 向指定gproc进程发送数据. // 数据格式：总长度(24bit)|发送进程PID(24bit)|接收进程PID(24bit)|分组长度(8bit)|分组名称(变长)|校验(32bit)|参数(变长) func Send(pid int, data []byte, group...string) error { groupName := gPROC_COMM_DEAFULT_GRUOP_NAME @@ -85,6 +85,6 @@ func getConnByPid(pid int) (*gtcp.Conn, error) { // 获取指定进程监听的端口号 func getPortByPid(pid int) int { path := getCommFilePath(pid) - content := gfile.GetContents(path) + content := gfcache.GetContents(path) return gconv.Int(content) } \ No newline at end of file From 3e0a975a883f7331e3a9ca01b8736a04cd1dd876 Mon Sep 17 00:00:00 2001 From: John Date: Wed, 20 Feb 2019 16:24:05 +0800 Subject: [PATCH 2/3] fix issue in gcron, allow special char '?' for day and week pattern --- g/os/gcron/gcron_schedule.go | 28 ++++++++++++++-------------- geg/os/gcron/gcron2.go | 15 ++++++++------- 2 files changed, 22 insertions(+), 21 deletions(-) diff --git a/g/os/gcron/gcron_schedule.go b/g/os/gcron/gcron_schedule.go index 029d8eb78..c9038e146 100644 --- a/g/os/gcron/gcron_schedule.go +++ b/g/os/gcron/gcron_schedule.go @@ -117,23 +117,23 @@ func newSchedule(pattern string) (*cronSchedule, error) { schedule.hour = m } // 天 - if m, err := parseItem(match[4], 1, 31, false); err != nil { + if m, err := parseItem(match[4], 1, 31, true); err != nil { return nil, err } else { schedule.day = m } - // 周 - if m, err := parseItem(match[5], 0, 6, false); err != nil { - return nil, err - } else { - schedule.week = m - } // 月 - if m, err := parseItem(match[6], 1, 12, false); err != nil { + if m, err := parseItem(match[5], 1, 12, false); err != nil { return nil, err } else { schedule.month = m } + // 周 + if m, err := parseItem(match[6], 0, 6, true); err != nil { + return nil, err + } else { + schedule.week = m + } return schedule, nil } else { return nil, errors.New(fmt.Sprintf(`invalid pattern: "%s"`, pattern)) @@ -200,14 +200,14 @@ func parseItemValue(value string, valueType byte) (int, error) { } else { // 英文字母 switch valueType { - case 'w': - if i, ok := weekMap[strings.ToLower(value)]; ok { - return int(i), nil - } case 'm': if i, ok := monthMap[strings.ToLower(value)]; ok { return int(i), nil } + case 'w': + if i, ok := weekMap[strings.ToLower(value)]; ok { + return int(i), nil + } } } return 0, errors.New(fmt.Sprintf(`invalid pattern value: "%s"`, value)) @@ -234,10 +234,10 @@ func (s *cronSchedule) meet(t time.Time) bool { if _, ok := s.day[t.Day()]; !ok { return false } - if _, ok := s.week[int(t.Weekday())]; !ok { + if _, ok := s.month[int(t.Month())]; !ok { return false } - if _, ok := s.month[int(t.Month())]; !ok { + if _, ok := s.week[int(t.Weekday())]; !ok { return false } return true diff --git a/geg/os/gcron/gcron2.go b/geg/os/gcron/gcron2.go index b388ccec7..17e9398ee 100644 --- a/geg/os/gcron/gcron2.go +++ b/geg/os/gcron/gcron2.go @@ -1,17 +1,18 @@ package main import ( + "fmt" "github.com/gogf/gf/g/os/gcron" - "github.com/gogf/gf/g/os/glog" "time" ) -func main() { - cron := gcron.New() - glog.Println("start") - cron.DelayAddOnce(1, "* * * * * *", func() { - glog.Println("run") - }) +func test() { +} + +func main() { + _, err := gcron.Add("*/10 * * * * ?", test) + fmt.Println(err) + fmt.Println(gcron.Entries()) time.Sleep(10*time.Second) } \ No newline at end of file From 3dd8b6ad335e5d95d422d2df43ffed83532a87c4 Mon Sep 17 00:00:00 2001 From: John Date: Wed, 20 Feb 2019 16:24:44 +0800 Subject: [PATCH 3/3] version updates --- version.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/version.go b/version.go index e45677ca5..0d0ed3eac 100644 --- a/version.go +++ b/version.go @@ -1,5 +1,5 @@ package gf -const VERSION = "v1.5.3" +const VERSION = "v1.5.4" const AUTHORS = "john"