Merge pull request #11 from alibaba/develop

fix bug of mapreduce
alibaba · Mar 8, 2024 · a4ecae5 · a4ecae5
2 parents f0c9954 + e6b5cbc
commit a4ecae5
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 23 deletions.
diff --git a/internal/batch/base_req_handler.go b/internal/batch/base_req_handler.go
@@ -140,6 +140,7 @@ func (rcvr *BaseReqHandler) Start(h ReqHandler) error {
 	}
 	rcvr.batchProcessSvc = gopool
 
+	rcvr.stopBatchRetrieveCh = make(chan struct{})
 	rcvr.batchRetrieveFunc = func() {
 		for {
 			select {

diff --git a/internal/master/common_update_instance_status_handler.go b/internal/master/common_update_instance_status_handler.go
@@ -17,9 +17,6 @@
 package master
 
 import (
-	"github.com/asynkron/protoactor-go/actor"
-	"google.golang.org/protobuf/proto"
-
 	"github.com/alibaba/schedulerx-worker-go/config"
 	"github.com/alibaba/schedulerx-worker-go/internal/actor/common"
 	"github.com/alibaba/schedulerx-worker-go/internal/common"
@@ -28,6 +25,8 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/internal/utils"
 	"github.com/alibaba/schedulerx-worker-go/logger"
 	"github.com/alibaba/schedulerx-worker-go/processor"
+	"github.com/asynkron/protoactor-go/actor"
+	"google.golang.org/protobuf/proto"
 )
 
 var _ UpdateInstanceStatusHandler = &commonUpdateInstanceStatusHandler{}

diff --git a/internal/master/map_task_master.go b/internal/master/map_task_master.go
@@ -51,8 +51,7 @@ import (
 )
 
 var (
-	_    taskmaster.MapTaskMaster = &MapTaskMaster{}
-	once sync.Once
+	_ taskmaster.MapTaskMaster = &MapTaskMaster{}
 )
 
 type MapTaskMaster struct {
@@ -84,6 +83,7 @@ type MapTaskMaster struct {
 	xAttrs              *common.MapTaskXAttrs
 	taskCounter         *atomic.Int64
 	localTaskRouterPath string
+	once                sync.Once
 }
 
 func NewMapTaskMaster(jobInstanceInfo *common.JobInstanceInfo, actorCtx actor.Context) taskmaster.TaskMaster {
@@ -139,7 +139,7 @@ func NewMapTaskMaster(jobInstanceInfo *common.JobInstanceInfo, actorCtx actor.Co
 }
 
 func (m *MapTaskMaster) init() {
-	once.Do(func() {
+	m.once.Do(func() {
 		m.TaskMaster.Init()
 		jobIdAndInstanceId := strconv.FormatInt(m.GetJobInstanceInfo().GetJobId(), 10) + "_" + strconv.FormatInt(m.GetJobInstanceInfo().GetJobInstanceId(), 10)
 		logger.Infof("jobInstanceId=%d, map master config, pageSize:%d, queueSize:%d, dispatcherSize:%d, workerSize:%d",
@@ -160,9 +160,9 @@ func (m *MapTaskMaster) init() {
 		go m.checkWorkerAlive()
 
 		// PULL_MODEL specially
-		if m.xAttrs != nil && m.xAttrs.GetTaskDispatchMode() == string(common.TaskDispatchModePull) {
-			go m.notifyWorkerPull()
-		}
+		//		if m.xAttrs != nil && m.xAttrs.GetTaskDispatchMode() == string(common.TaskDispatchModePull) {
+		//			go m.notifyWorkerPull()
+		//		}
 	})
 }
 

diff --git a/internal/master/task_master.go b/internal/master/task_master.go
@@ -80,8 +80,8 @@ func NewTaskMaster(actorCtx actor.Context, jobInstanceInfo *common.JobInstanceIn
 }
 
 func (m *TaskMaster) Init() {
-	m.lock.Lock()
-	defer m.lock.Unlock()
+	//	m.lock.Lock()
+	//	defer m.lock.Unlock()
 	if !m.inited {
 		m.inited = true
 	}
@@ -159,7 +159,10 @@ func (m *TaskMaster) UpdateTaskStatus(req *schedulerx.ContainerReportTaskStatusR
 	return nil
 }
 
-func (m *TaskMaster) updateNewInstanceStatus(serialNum, jobInstanceId int64, newStatus processor.InstanceStatus, result string) error {
+func (m *TaskMaster) updateNewInstanceStatus(serialNum int64, jobInstanceId int64, newStatus processor.InstanceStatus, result string) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	//	fmt.Printf("serialNum=%d, jobInstanceId=%d, status=%s\n", serialNum, jobInstanceId, newStatus)
 	if err := m.statusHandler.Handle(serialNum, newStatus, result); err != nil {
 		return fmt.Errorf("update status failed, err=%s", err.Error())
 	}
@@ -229,9 +232,9 @@ func (m *TaskMaster) BatchUpdateTaskStatus(taskMaster taskmaster.TaskMaster, req
 }
 
 func (m *TaskMaster) KillInstance(reason string) error {
-	m.lock.Lock()
+	//	m.lock.Lock()
 	m.killed = true
-	m.lock.Unlock()
+	//	m.lock.Unlock()
 
 	GetTimeScheduler().remove(m.jobInstanceInfo.GetJobInstanceId())
 	return nil
@@ -289,20 +292,20 @@ func (m *TaskMaster) PostFinish(jobInstanceId int64) *processor.ProcessResult {
 }
 
 func (m *TaskMaster) GetInstanceStatus() processor.InstanceStatus {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
+	//	m.lock.RLock()
+	//	defer m.lock.RUnlock()
 	return m.instanceStatus
 }
 
 func (m *TaskMaster) SetInstanceStatus(instanceStatus processor.InstanceStatus) {
-	m.lock.Lock()
+	//	m.lock.Lock()
 	m.instanceStatus = instanceStatus
-	m.lock.Unlock()
+	//	m.lock.Unlock()
 }
 
 func (m *TaskMaster) IsKilled() bool {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
+	//	m.lock.RLock()
+	//	defer m.lock.RUnlock()
 	return m.killed
 }
 
@@ -316,8 +319,8 @@ func (m *TaskMaster) GetAliveCheckWorkerSet() *utils.ConcurrentSet {
 }
 
 func (m *TaskMaster) IsInited() bool {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
+	//	m.lock.RLock()
+	//	defer m.lock.RUnlock()
 	return m.inited
 }
 

diff --git a/internal/version/version.go b/internal/version/version.go
@@ -18,5 +18,5 @@ package version
 
 // Version used for statistics and debug
 func Version() string {
-	return "v0.0.2"
+	return "v0.0.4"
 }