Merge pull request #37 from alibaba/develop

v1.0.7
alibaba · Nov 7, 2024 · 6dac403 · 6dac403
2 parents 6619423 + 8e36fb1
commit 6dac403
Show file tree

Hide file tree

Showing 13 changed files with 94 additions and 205 deletions.
diff --git a/client.go b/client.go
@@ -164,7 +164,7 @@ func newClient(cfg *Config, opts ...Option) (*Client, error) {
 	}
 
 	stopChan := make(chan os.Signal, 1)
-	signal.Notify(stopChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1, syscall.SIGUSR2)
+	signal.Notify(stopChan, syscall.SIGINT, syscall.SIGTERM)
 
 	// Keep heartbeat, and receive message
 	// KeepHeartbeat must after init actors, so that can get actorSystemPort from actorSystem

diff --git a/example/broadcast/main.go b/example/broadcast/main.go
@@ -44,7 +44,7 @@ func main() {
 
 	// wait for the stop signal
 	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1, syscall.SIGUSR2)
+	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
 	<-c
 	time.Sleep(time.Second * 5)
 }
diff --git a/example/mapreduce/main.go b/example/mapreduce/main.go
@@ -47,7 +47,7 @@ func main() {
 
 	// wait for the stop signal
 	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1, syscall.SIGUSR2)
+	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
 	<-c
 	time.Sleep(time.Second * 5)
 }
diff --git a/example/standalone/main.go b/example/standalone/main.go
@@ -46,7 +46,7 @@ func main() {
 
 	// wait for the stop signal
 	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1, syscall.SIGUSR2)
+	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
 	<-c
 	time.Sleep(time.Second * 5)
 }
diff --git a/go.mod b/go.mod
@@ -1,6 +1,6 @@
 module github.com/alibaba/schedulerx-worker-go
 
-go 1.18
+go 1.20
 
 require (
 	github.com/asynkron/protoactor-go v0.0.0-20230916135836-b14bb1f51af6

diff --git a/internal/actor/common/actor_system.go b/internal/actor/common/actor_system.go
@@ -27,7 +27,6 @@ var (
 	actorSystem *actor.ActorSystem
 )
 
-// GetActorSystem must be executed before InitActorSystem, otherwise it returns nil
 func GetActorSystem() *actor.ActorSystem {
 	once.Do(func() {
 		actorSystem = actor.NewActorSystem()

diff --git a/internal/batch/base_req_handler.go b/internal/batch/base_req_handler.go
@@ -173,13 +173,12 @@ func (rcvr *BaseReqHandler) Stop() {
 	}
 	if rcvr.batchProcessSvc != nil {
 		rcvr.batchProcessSvc.Release()
-		rcvr.batchProcessSvc = nil
 	}
 	if rcvr.reqsQueue != nil {
 		rcvr.reqsQueue.Clear()
 	}
 	if rcvr.activeRunnableNum != nil {
-		rcvr.activeRunnableNum = nil
+		rcvr.activeRunnableNum.Store(0)
 	}
 }
 

diff --git a/internal/master/map_task_master.go b/internal/master/map_task_master.go
@@ -421,8 +421,9 @@ func (m *MapTaskMaster) BatchUpdateTaskStatues(requests []*schedulerx.ContainerR
 			taskName   = request.GetTaskName()
 		)
 
-		logger.Debugf("report task status:%s from worker:%s, uniqueId:%d", taskStatus.Descriptor(), workerAddr,
+		logger.Debugf("report task status:%s from worker:%s, uniqueId:%s", taskStatus.Descriptor(), workerAddr,
 			utils.GetUniqueId(request.GetJobId(), request.GetJobInstanceId(), request.GetTaskId()))
+
 		m.taskProgressMap.LoadOrStore(taskName, common.NewTaskProgressCounter(taskName))
 		if _, ok := m.workerProgressMap.Load(workerAddr); workerAddr != "" && !ok {
 			m.workerProgressMap.LoadOrStore(workerAddr, common.NewWorkerProgressCounter(workerAddr))
@@ -488,7 +489,10 @@ func (m *MapTaskMaster) BatchUpdateTaskStatues(requests []*schedulerx.ContainerR
 		break
 	}
 	if !updateSuccess {
-		m.UpdateNewInstanceStatus(m.GetSerialNum(), processor.InstanceStatusFailed, "persistent batch update TaskStatus error up to 3 times")
+		err := m.UpdateNewInstanceStatus(m.GetSerialNum(), processor.InstanceStatusFailed, "persistent batch update TaskStatus error up to 3 times")
+		if err != nil {
+			logger.Errorf("jobInstanceId=%d, UpdateNewInstanceStatus failed, err=%s", m.GetJobInstanceInfo().GetJobInstanceId(), err.Error())
+		}
 	}
 	logger.Debugf("jobInstanceId=%d batch update status db cost %dms", m.GetJobInstanceInfo().GetJobInstanceId(), time.Since(startTime).Milliseconds())
 }

diff --git a/internal/master/persistence/h2_connection_pool.go b/internal/master/persistence/h2_connection_pool.go
@@ -22,11 +22,9 @@ import (
 	"path/filepath"
 
 	_ "github.com/mattn/go-sqlite3"
-
-	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-const memoryModeDataSourceName = ":memory:"
+const memoryModeDataSourceName = "file::memory:?cache=shared"
 
 type H2ConnectionPool struct {
 	*sql.DB
@@ -50,39 +48,24 @@ func NewH2ConnectionPool(opts ...Option) (*H2ConnectionPool, error) {
 		opt(options)
 	}
 
-	dataSourceName := "sqlite3.db"
-	if options.dataSourceName != "" {
-		dataSourceName = options.dataSourceName
-	}
-
 	files, err := filepath.Glob("schedulerx2_*_sqlite3.db")
 	if err != nil {
 		return nil, err
 	}
 	for _, file := range files {
-		if err := os.Remove(file); err != nil {
+		if err = os.Remove(file); err != nil {
 			return nil, err
 		}
 	}
 
-	// memory mode no need creating local file
-	if dataSourceName != memoryModeDataSourceName {
-		logger.Infof("Creating %s sqlite3...", dataSourceName)
-		file, err := os.Create(dataSourceName)
-		if err != nil {
-			return nil, err
-		}
-		defer file.Close()
-
-		logger.Infof("sqlite3 DB=%s created", dataSourceName)
+	dataSourceName := "sqlite3.db"
+	if options.dataSourceName != "" {
+		dataSourceName = options.dataSourceName
 	}
-
-	sqliteDatabase, err := sql.Open("sqlite3", dataSourceName)
+	db, err := sql.Open("sqlite3", dataSourceName)
 	if err != nil {
 		return nil, err
 	}
-
-	return &H2ConnectionPool{
-		sqliteDatabase,
-	}, nil
+	db.SetMaxIdleConns(5) // default 2
+	return &H2ConnectionPool{DB: db}, nil
 }
diff --git a/internal/master/persistence/server_task_persistence.go b/internal/master/persistence/server_task_persistence.go
@@ -256,7 +256,7 @@ func (rcvr *ServerTaskPersistence) UpdateTaskStatues(taskStatusInfos []*schedule
 	}
 	info := taskStatusInfos[0]
 	status2WorkIdAddr2TaskIds := getTaskStatusMap(taskStatusInfos)
-	var batchTaskStatuesReq *schedulerx.WorkerBatchReportTaskStatuesRequest
+	var batchTaskStatuesReq schedulerx.WorkerBatchReportTaskStatuesRequest
 	for status, workerAddr2TaskIds := range status2WorkIdAddr2TaskIds {
 		for workerIdAddr, taskIds := range workerAddr2TaskIds {
 			workerIdAddrParts := strings.Split(workerIdAddr, "@")
@@ -273,7 +273,7 @@ func (rcvr *ServerTaskPersistence) UpdateTaskStatues(taskStatusInfos []*schedule
 	batchTaskStatuesReq.JobInstanceId = proto.Int64(info.GetJobInstanceId())
 
 	actorcomm.AtLeastOnceDeliveryMsgReceiver() <- &actorcomm.SchedulerWrappedMsg{
-		Msg: batchTaskStatuesReq,
+		Msg: &batchTaskStatuesReq,
 	}
 
 	return nil