fix: resolve data races in WindowOperator and GlobalServerManager

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

Author: ngjaying

internal/io/http/httpserver/data_server.go

@@ -46,6 +46,7 @@ type GlobalServerManager struct {
 }
 
 var manager *GlobalServerManager
+var managerLock sync.RWMutex
 
 func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 	r := mux.NewRouter()
@@ -66,6 +67,7 @@ func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 			return true
 		},
 	}
+	managerLock.Lock()
 	manager = &GlobalServerManager{
 		websocketEndpoint: map[string]*websocketEndpointContext{},
 		endpoint:          map[string]string{},
@@ -81,19 +83,36 @@ func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 			s.ListenAndServeTLS(conf.Config.Source.HttpServerTls.Certfile, conf.Config.Source.HttpServerTls.Keyfile)
 		}
 	}(manager)
+	managerLock.Unlock()
 	time.Sleep(500 * time.Millisecond)
 }
 
 func ShutDown() {
-	manager.Shutdown()
+	managerLock.RLock()
+	if manager != nil {
+		manager.Shutdown()
+	}
+	managerLock.RUnlock()
+	managerLock.Lock()
 	manager = nil
+	managerLock.Unlock()
 }
 
 func RegisterEndpoint(endpoint string, method string) (string, error) {
+	managerLock.RLock()
+	defer managerLock.RUnlock()
+	if manager == nil {
+		return "", fmt.Errorf("http server is not running")
+	}
 	return manager.RegisterEndpoint(endpoint, method)
 }
 
 func UnregisterEndpoint(endpoint, method string) {
+	managerLock.RLock()
+	defer managerLock.RUnlock()
+	if manager == nil {
+		return
+	}
 	manager.UnregisterEndpoint(endpoint, method)
 }
 

internal/topo/node/node_test/window_inc_agg_op_test.go

@@ -525,7 +525,7 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 }
 
 func waitExecute() {
-	time.Sleep(50 * time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
 }
 
 func TestIncHoppingWindow(t *testing.T) {

internal/topo/node/window_op.go

@@ -640,7 +640,7 @@ func (o *WindowOperator) handleInputs(ctx api.StreamContext, inputs []xsql.Event
 		// For hopping and sliding window, firstly check if the beginning tuples are expired and discard them
 		if o.isOverlapWindow && !allDiscarded {
 			if left.After(tuple.GetTimestamp()) {
-				log.Debugf("tuple %x emitted at %d expired", tuple, tuple.GetTimestamp().UnixMilli())
+				log.Debugf("tuple at %d expired", tuple.GetTimestamp().UnixMilli())
 				// Expired tuple, remove it by not adding back to inputs
 				continue
 			}
@@ -727,7 +727,7 @@ func (o *WindowOperator) scan(inputs []xsql.EventRow, triggerTime time.Time, ctx
 		results.WindowRange = xsql.NewWindowRange(windowStart, windowEnd.UnixMilli(), triggerTime.Add(-o.window.Delay).UnixMilli())
 	}
 	log.Debugf("window %s triggered for %d tuples", o.name, len(inputs))
-	log.Debugf("Sent: %v", results)
+
 	o.Broadcast(results)
 	o.onSend(ctx, results)