Bugfix second delay duplicate

example/broadcast/test_broadcast.go

@@ -26,7 +26,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ processor.BroadcastProcessor = &TestBroadcast{}
+var _ processor.BroadcastProcessor = (*TestBroadcast)(nil)
 
 type TestBroadcast struct{}
 

example/mapreduce/order_info.go

@@ -28,7 +28,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ processor.MapReduceJobProcessor = &TestMapReduceJob{}
+var _ processor.MapReduceJobProcessor = (*TestMapReduceJob)(nil)
 
 type OrderInfo struct {
 	Id    string `json:"id"`

example/standalone/helloworld.go

@@ -24,7 +24,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/jobcontext"
 )
 
-var _ processor.Processor = &HelloWorld{}
+var _ processor.Processor = (*HelloWorld)(nil)
 
 type HelloWorld struct{}
 

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/lithammer/shortuuid/v4 v4.0.0 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/orcaman/concurrent-map v1.0.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -42,6 +43,8 @@ require (
 	github.com/prometheus/procfs v0.10.1 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/twmb/murmur3 v1.1.6 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/otel v1.16.0 // indirect

go.sum

@@ -29,6 +29,7 @@ github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -39,6 +40,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw7k08o4c=
 github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y=
+github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
+github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/mattn/go-sqlite3 v1.14.24 h1:tpSp2G2KyMnnQu99ngJ47EIkWVmliIizyZBfPrBWDRM=
 github.com/mattn/go-sqlite3 v1.14.24/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
@@ -85,6 +88,10 @@ github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhso
 github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tinylib/msgp v1.1.5/go.mod h1:eQsjooMTnV42mHu917E26IogZ2930nFyBQdofk10Udg=
+github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
+github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
+github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
+github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
 github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31/go.mod h1:onvgF043R+lC5RZ8IT9rBXDaEDnpnw/Cl+HFiw+v/7Q=
 github.com/twmb/murmur3 v1.1.6 h1:mqrRot1BRxm+Yct+vavLMou2/iJt0tNVTTC0QoIjaZg=
 github.com/twmb/murmur3 v1.1.6/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
@@ -132,6 +139,8 @@ golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

internal/actor/container_actor.go

@@ -40,7 +40,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ actor.Actor = &containerActor{}
+var _ actor.Actor = (*containerActor)(nil)
 
 var defaultActorPool, _ = ants.NewPool(
 	math.MaxInt32,

internal/actor/heartbeat_actor.go

@@ -30,7 +30,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ actor.Actor = &heartbeatActor{}
+var _ actor.Actor = (*heartbeatActor)(nil)
 
 // heartbeatActor is the
 type heartbeatActor struct {

internal/actor/heartbeat_processor.go

@@ -32,7 +32,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ actor.Process = &heartbeatProcessor{}
+var _ actor.Process = (*heartbeatProcessor)(nil)
 
 type heartbeatProcessor struct {
 	connpool pool.ConnPool

internal/actor/job_instance_actor.go

@@ -39,7 +39,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ actor.Actor = &jobInstanceActor{}
+var _ actor.Actor = (*jobInstanceActor)(nil)
 
 type jobInstanceActor struct {
 	connpool       pool.ConnPool
@@ -178,7 +178,7 @@ func (a *jobInstanceActor) handleSubmitJobInstance(actorCtx actor.Context, msg *
 
 			if taskMaster != nil {
 				masterpool.GetTaskMasterPool().Put(jobInstanceInfo.GetJobInstanceId(), taskMaster)
-				if err := taskMaster.SubmitInstance(msg.Ctx, jobInstanceInfo); err != nil {
+				if err := taskMaster.SubmitInstance(jobInstanceInfo); err != nil {
 					return err
 				}
 				logger.Infof("Submit jobInstanceId=%d succeed", req.GetJobInstanceId())
@@ -230,34 +230,31 @@ func convert2JobInstanceData(datas []*schedulerx.UpstreamData) []*common.JobInst
 }
 
 func (a *jobInstanceActor) handleKillJobInstance(actorCtx actor.Context, msg *actorcomm.SchedulerWrappedMsg) error {
-	var (
-		taskMasterPool = masterpool.GetTaskMasterPool()
-		req            = msg.Msg.(*schedulerx.ServerKillJobInstanceRequest)
-	)
+	req := msg.Msg.(*schedulerx.ServerKillJobInstanceRequest)
+	taskMasterPool := masterpool.GetTaskMasterPool()
 	logger.Infof("handleKillJobInstance, jobInstanceId=%d ", req.GetJobInstanceId())
+
+	var resp *schedulerx.ServerKillJobInstanceResponse
 	if !taskMasterPool.Contains(req.GetJobInstanceId()) {
 		errMsg := fmt.Sprintf("%d is not exist", req.GetJobInstanceId())
 		logger.Infof(errMsg)
-		resp := &schedulerx.ServerKillJobInstanceResponse{
-			Success: proto.Bool(true),
+		resp = &schedulerx.ServerKillJobInstanceResponse{
+			Success: proto.Bool(false),
 			Message: proto.String(errMsg),
 		}
-		actorCtx.Send(actorcomm.SchedulerxServerPid(msg.Ctx), actorcomm.WrapSchedulerxMsg(msg.Ctx, resp, msg.SenderPath))
 	} else {
-		if taskMaster := masterpool.GetTaskMasterPool().Get(req.GetJobInstanceId()); taskMaster != nil {
+		if taskMaster := taskMasterPool.Get(req.GetJobInstanceId()); taskMaster != nil {
 			if err := taskMaster.KillInstance("killed from server"); err != nil {
 				logger.Infof("%d killed from server failed, err=%s", req.GetJobInstanceId(), err.Error())
 			}
 		}
-		errMsg := fmt.Sprintf("%d killed from server", req.GetJobInstanceId())
-		logger.Infof(errMsg)
-		resp := &schedulerx.ServerKillJobInstanceResponse{
-			Success: proto.Bool(false), // FIXME true or false
-			Message: proto.String(errMsg),
+		resp = &schedulerx.ServerKillJobInstanceResponse{
+			Success: proto.Bool(true),
+			Message: proto.String("killed from server"),
 		}
-		actorCtx.Send(actorcomm.SchedulerxServerPid(msg.Ctx), actorcomm.WrapSchedulerxMsg(msg.Ctx, resp, msg.SenderPath))
 		logger.Infof("Kill jobInstanceId=%d succeed", req.GetJobInstanceId())
 	}
+	actorCtx.Send(actorcomm.SchedulerxServerPid(msg.Ctx), actorcomm.WrapSchedulerxMsg(msg.Ctx, resp, msg.SenderPath))
 	return nil
 }
 

internal/actor/job_instance_processor.go

@@ -32,7 +32,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ actor.Process = &jobInstanceProcessor{}
+var _ actor.Process = (*jobInstanceProcessor)(nil)
 
 type jobInstanceProcessor struct {
 	connpool pool.ConnPool

internal/actor/task_actor.go

@@ -33,7 +33,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor"
 )
 
-var _ actor.Actor = &taskActor{}
+var _ actor.Actor = (*taskActor)(nil)
 
 // taskActor is the
 type taskActor struct {

internal/actor/task_actor_processor.go

@@ -31,7 +31,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ actor.Process = &taskProcessor{}
+var _ actor.Process = (*taskProcessor)(nil)
 
 type taskProcessor struct {
 	connpool pool.ConnPool

internal/batch/base_req_handler.go

@@ -28,7 +28,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ ReqHandler = &BaseReqHandler{}
+var _ ReqHandler = (*BaseReqHandler)(nil)
 
 var globalPool, _ = ants.NewPool(
 	math.MaxInt32,

internal/batch/container_status_req_handler.go

@@ -42,11 +42,6 @@ func (h *ContainerStatusReqHandler) GetTaskMasterAkkaPath() string {
 	return h.taskMasterAkkaPath
 }
 
-type Pair struct {
-	jobInstanceId int64
-	serialNum     int64
-}
-
 func (h *ContainerStatusReqHandler) Process(jobInstanceId int64, requests []interface{}, workerAddr string) {
 	reqs := make([]*schedulerx.ContainerReportTaskStatusRequest, 0, len(requests))
 	for _, req := range requests {

internal/batch/task_pull_req_handler.go

@@ -22,7 +22,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ TaskDispatchReqHandler = &TaskPullReqHandler{}
+var _ TaskDispatchReqHandler = (*TaskPullReqHandler)(nil)
 
 type TaskPullReqHandler struct {
 	*BaseTaskDispatchReqHandler

internal/batch/task_push_req_handler.go

@@ -24,7 +24,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ TaskDispatchReqHandler = &TaskPushReqHandler{}
+var _ TaskDispatchReqHandler = (*TaskPushReqHandler)(nil)
 
 type TaskPushReqHandler struct {
 	*BaseTaskDispatchReqHandler

internal/batch/tms_status_req_handler.go

@@ -24,7 +24,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/logger"
 )
 
-var _ ReqHandler = &TMStatusReqHandler{}
+var _ ReqHandler = (*TMStatusReqHandler)(nil)
 
 type TMStatusReqHandler struct {
 	*BaseReqHandler

internal/container/thread_container.go

@@ -37,7 +37,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ Container = &ThreadContainer{}
+var _ Container = (*ThreadContainer)(nil)
 
 type ThreadContainer struct {
 	jobCtx        *jobcontext.JobContext

internal/master/batch_task_master.go

@@ -28,7 +28,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/internal/utils"
 )
 
-var _ taskmaster.MapTaskMaster = &BatchTaskMaster{}
+var _ taskmaster.MapTaskMaster = (*BatchTaskMaster)(nil)
 
 type BatchTaskMaster struct {
 	*GridTaskMaster

internal/master/broadcast_task_master.go

@@ -17,7 +17,6 @@
 package master
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"strconv"
@@ -42,7 +41,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ taskmaster.TaskMaster = &BroadcastTaskMaster{}
+var _ taskmaster.TaskMaster = (*BroadcastTaskMaster)(nil)
 
 type BroadcastTaskMaster struct {
 	*TaskMaster
@@ -76,7 +75,7 @@ func NewBroadcastTaskMaster(jobInstanceInfo *common.JobInstanceInfo, actorCtx ac
 	return broadcastTaskMaster
 }
 
-func (m *BroadcastTaskMaster) SubmitInstance(ctx context.Context, jobInstanceInfo *common.JobInstanceInfo) error {
+func (m *BroadcastTaskMaster) SubmitInstance(jobInstanceInfo *common.JobInstanceInfo) error {
 	if err := m.preProcess(jobInstanceInfo); err != nil {
 		logger.Errorf("BroadcastTaskMaster.preProcess failed, jobInstanceId=%d, err=%s", jobInstanceInfo.GetJobInstanceId(), err.Error())
 		if e := m.TaskMaster.updateNewInstanceStatus(m.GetSerialNum(), m.jobInstanceInfo.GetJobInstanceId(), processor.InstanceStatusFailed, "Preprocess failed. "+err.Error()); e != nil {
@@ -185,7 +184,7 @@ func (m *BroadcastTaskMaster) dispatchTask(jobInstanceInfo *common.JobInstanceIn
 }
 
 func (m *BroadcastTaskMaster) KillInstance(reason string) error {
-	m.TaskMaster.KillInstance(reason)
+	_ = m.TaskMaster.KillInstance(reason)
 
 	for _, workerIdAddr := range m.allWorkers {
 		uniqueId, ok := m.worker2uniqueIdMap.Load(workerIdAddr)
@@ -362,7 +361,7 @@ func (m *BroadcastTaskMaster) checkWorkerAlive() {
 			m.aliveCheckWorkerSet.Add(worker)
 		}
 
-		for _, workerIdAddr := range m.aliveCheckWorkerSet.ToStringSlice() {
+		for _, workerIdAddr := range m.aliveCheckWorkerSet.Keys() {
 			req := &schedulerx.MasterCheckWorkerAliveRequest{
 				JobInstanceId: proto.Int64(m.jobInstanceInfo.GetJobInstanceId()),
 			}

internal/master/common_update_instance_status_handler.go

@@ -29,7 +29,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor"
 )
 
-var _ UpdateInstanceStatusHandler = &commonUpdateInstanceStatusHandler{}
+var _ UpdateInstanceStatusHandler = (*commonUpdateInstanceStatusHandler)(nil)
 
 type commonUpdateInstanceStatusHandler struct {
 	*baseUpdateInstanceStatusHandler

internal/master/grid_task_master.go

@@ -33,7 +33,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/jobcontext"
 )
 
-var _ taskmaster.MapTaskMaster = &GridTaskMaster{}
+var _ taskmaster.MapTaskMaster = (*GridTaskMaster)(nil)
 
 type GridTaskMaster struct {
 	*MapTaskMaster

internal/master/map_task_master.go

@@ -17,7 +17,6 @@
 package master
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -293,7 +292,7 @@ func (m *MapTaskMaster) checkWorkerAlive() {
 			m.taskPersistence.BatchUpdateTaskStatus(m.GetJobInstanceInfo().GetJobInstanceId(), taskstatus.TaskStatusFailed, "", "")
 			break
 		} else {
-			for _, workerIdAddr := range m.aliveCheckWorkerSet.ToStringSlice() {
+			for _, workerIdAddr := range m.aliveCheckWorkerSet.Keys() {
 				workerAddr := actorcomm.GetRealWorkerAddr(workerIdAddr)
 
 				times := 0
@@ -376,7 +375,7 @@ func (m *MapTaskMaster) notifyWorkerPull() {
 	}
 }
 
-func (m *MapTaskMaster) SubmitInstance(ctx context.Context, jobInstanceInfo *common.JobInstanceInfo) error {
+func (m *MapTaskMaster) SubmitInstance(jobInstanceInfo *common.JobInstanceInfo) error {
 	var err error
 	defer func() {
 		if err != nil {
@@ -790,7 +789,8 @@ func (m *MapTaskMaster) selectWorker() string {
 }
 
 func (m *MapTaskMaster) KillInstance(reason string) error {
-	m.TaskMaster.KillInstance(reason)
+	_ = m.TaskMaster.KillInstance(reason)
+
 	allWorkers := m.GetJobInstanceInfo().GetAllWorkers()
 	for _, workerIdAddr := range allWorkers {
 		request := &schedulerx.MasterKillContainerRequest{

internal/master/parallel_task_mater.go

@@ -40,7 +40,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ taskmaster.ParallelTaskMaster = &ParallelTaskMaster{}
+var _ taskmaster.ParallelTaskMaster = (*ParallelTaskMaster)(nil)
 
 const batchSize = 256
 

internal/master/persistence/server_task_persistence.go

@@ -35,7 +35,7 @@ import (
 	"github.com/alibaba/schedulerx-worker-go/processor/taskstatus"
 )
 
-var _ TaskPersistence = &ServerTaskPersistence{}
+var _ TaskPersistence = (*ServerTaskPersistence)(nil)
 
 type ServerTaskPersistence struct {
 	serverDiscovery *discovery.ServiceDiscover