fix(os/gcron): unit testing case of package gcron occasionally failed (gogf#4419)

fix gogf#3999

1. fix  jobWaiter   sync.WaitGroup  data race
2. fix logger      glog.ILogger data race

---------

Co-authored-by: hailaz <739476267@qq.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

os/gcron/gcron.go

@@ -125,3 +125,9 @@ func Stop(name ...string) {
 func StopGracefully() {
 	defaultCron.StopGracefully()
 }
+
+// StopGracefullyNonBlocking stops all running tasks gracefully without blocking,
+// returning a context that callers can use to wait for completion.
+func StopGracefullyNonBlocking() context.Context {
+	return defaultCron.StopGracefullyNonBlocking()
+}

os/gcron/gcron_cron.go

@@ -20,11 +20,14 @@ import (
 
 // Cron stores all the cron job entries.
 type Cron struct {
-	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.StrAnyMap // All timed task entries.
-	logger    glog.ILogger    // Logger, it is nil in default.
-	jobWaiter sync.WaitGroup  // Graceful shutdown when cron jobs are stopped.
+	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.StrAnyMap // All timed task entries.
+	logger      glog.ILogger    // Logger, it is nil in default.
+	loggerMu    sync.RWMutex
+	jobWaiter   sync.WaitGroup // Graceful shutdown when cron jobs are stopped.
+	running     bool
+	runningLock sync.Mutex
 }
 
 // New returns a new Cron object with default settings.
@@ -33,16 +36,21 @@ func New() *Cron {
 		idGen:   gtype.NewInt64(),
 		status:  gtype.NewInt(StatusRunning),
 		entries: gmap.NewStrAnyMap(true),
+		running: true,
 	}
 }
 
 // SetLogger sets the logger for cron.
 func (c *Cron) SetLogger(logger glog.ILogger) {
+	c.loggerMu.Lock()
+	defer c.loggerMu.Unlock()
 	c.logger = logger
 }
 
 // GetLogger returns the logger in the cron.
 func (c *Cron) GetLogger() glog.ILogger {
+	c.loggerMu.RLock()
+	defer c.loggerMu.RUnlock()
 	return c.logger
 }
 
@@ -171,7 +179,10 @@ func (c *Cron) Start(name ...string) {
 			}
 		}
 	} else {
+		c.runningLock.Lock()
 		c.status.Set(StatusReady)
+		c.running = true
+		c.runningLock.Unlock()
 	}
 }
 
@@ -185,14 +196,32 @@ func (c *Cron) Stop(name ...string) {
 			}
 		}
 	} else {
+		c.runningLock.Lock()
 		c.status.Set(StatusStopped)
+		c.running = false
+		c.runningLock.Unlock()
 	}
 }
 
 // StopGracefully Blocks and waits all current running jobs done.
 func (c *Cron) StopGracefully() {
-	c.status.Set(StatusStopped)
-	c.jobWaiter.Wait()
+	ctx := c.StopGracefullyNonBlocking()
+	<-ctx.Done()
+}
+
+// StopGracefullyNonBlocking stops all running tasks gracefully without blocking,
+// returning a context that callers can use to wait for completion.
+func (c *Cron) StopGracefullyNonBlocking() context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		c.runningLock.Lock()
+		defer c.runningLock.Unlock()
+		c.status.Set(StatusStopped)
+		c.running = false
+		c.jobWaiter.Wait()
+		cancel()
+	}()
+	return ctx
 }
 
 // Remove deletes scheduled task which named `name`.
@@ -204,7 +233,10 @@ func (c *Cron) Remove(name string) {
 
 // Close stops and closes current cron.
 func (c *Cron) Close() {
+	c.runningLock.Lock()
+	defer c.runningLock.Unlock()
 	c.status.Set(StatusClosed)
+	c.running = false
 }
 
 // Size returns the size of the timed tasks.

os/gcron/gcron_entry.go

@@ -152,7 +152,15 @@ func (e *Entry) checkAndRun(ctx context.Context) {
 		e.Close()
 
 	case StatusReady, StatusRunning:
-		e.cron.jobWaiter.Add(1)
+		e.cron.runningLock.Lock()
+		if e.cron.running {
+			e.cron.jobWaiter.Add(1)
+		} else {
+			e.cron.runningLock.Unlock()
+			e.logDebugf(ctx, `cron job "%s" stoped or closed`, e.getJobNameWithPattern())
+			return
+		}
+		e.cron.runningLock.Unlock()
 		defer func() {
 			e.cron.jobWaiter.Done()
 			if exception := recover(); exception != nil {

os/gcron/gcron_z_example_1_test.go

@@ -31,7 +31,7 @@ func ExampleCron_gracefulShutdown() {
 		g.Log().Debug(ctx, "Every 2s job start")
 		time.Sleep(5 * time.Second)
 		g.Log().Debug(ctx, "Every 2s job after 5 second end")
-	}, "MyCronJob")
+	}, "MyCronJob1")
 	if err != nil {
 		panic(err)
 	}
@@ -46,3 +46,25 @@ func ExampleCron_gracefulShutdown() {
 	gcron.StopGracefully()
 	glog.Print(ctx, "All cron jobs completed")
 }
+
+func ExampleCron_StopGracefullyNonBlocking() {
+	_, err := gcron.Add(ctx, "*/2 * * * * *", func(ctx context.Context) {
+		g.Log().Debug(ctx, "Every 2s job start")
+		time.Sleep(5 * time.Second)
+		g.Log().Debug(ctx, "Every 2s job after 5 second end")
+	}, "MyCronJob2")
+	if err != nil {
+		panic(err)
+	}
+
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+
+	sig := <-quit
+	glog.Printf(ctx, "Signal received: %s, stopping cron", sig)
+
+	glog.Print(ctx, "Waiting for all cron jobs to complete...")
+	ctx := gcron.StopGracefullyNonBlocking()
+	<-ctx.Done()
+	glog.Print(ctx, "All cron jobs completed")
+}

os/gcron/gcron_z_unit_entry_test.go

@@ -22,10 +22,12 @@ func TestCron_Entry_Operations(t *testing.T) {
 			cron  = gcron.New()
 			array = garray.New(true)
 		)
-		cron.DelayAddTimes(ctx, 500*time.Millisecond, "* * * * * *", 2, func(ctx context.Context) {
-			array.Append(1)
-		})
-		t.Assert(cron.Size(), 0)
+		go func() {
+			cron.DelayAddTimes(ctx, 500*time.Millisecond, "* * * * * *", 2, func(ctx context.Context) {
+				array.Append(1)
+			})
+			t.Assert(cron.Size(), 0)
+		}()
 		time.Sleep(800 * time.Millisecond)
 		t.Assert(array.Len(), 0)
 		t.Assert(cron.Size(), 1)
@@ -39,12 +41,16 @@ func TestCron_Entry_Operations(t *testing.T) {
 			cron  = gcron.New()
 			array = garray.New(true)
 		)
-		entry, err1 := cron.Add(ctx, "* * * * * *", func(ctx context.Context) {
-			array.Append(1)
-		})
-		t.Assert(err1, nil)
-		t.Assert(array.Len(), 0)
-		t.Assert(cron.Size(), 1)
+		var entry *gcron.Entry
+		go func() {
+			var err error
+			entry, err = cron.Add(ctx, "* * * * * *", func(ctx context.Context) {
+				array.Append(1)
+			})
+			t.Assert(err, nil)
+			t.Assert(array.Len(), 0)
+			t.Assert(cron.Size(), 1)
+		}()
 		time.Sleep(1300 * time.Millisecond)
 		t.Assert(array.Len(), 1)
 		t.Assert(cron.Size(), 1)