new task scheduler

cmd/version/version.go

@@ -1,3 +1,3 @@
 package version
 
-const Name = "0.2.3"
+const Name = "0.2.4"

config/config.go

@@ -77,8 +77,7 @@ type MasterConfig struct {
 	Host        string `toml:"host"`         // master host
 	HttpPort    int    `toml:"http_port"`    // HTTP port
 	HttpHost    string `toml:"http_host"`    // HTTP host
-	SearchJobs  int    `toml:"search_jobs"`  // number of working jobs to search model
-	FitJobs     int    `toml:"fit_jobs"`     // number of working jobs to fit model
+	NumJobs     int    `toml:"n_jobs"`       // number of working jobs
 	MetaTimeout int    `toml:"meta_timeout"` // cluster meta timeout (second)
 }
 
@@ -90,8 +89,7 @@ func (config *MasterConfig) LoadDefaultIfNil() *MasterConfig {
 			Host:        "127.0.0.1",
 			HttpPort:    8088,
 			HttpHost:    "127.0.0.1",
-			SearchJobs:  1,
-			FitJobs:     1,
+			NumJobs:     1,
 			MetaTimeout: 60,
 		}
 	}
@@ -172,11 +170,8 @@ func (config *Config) FillDefault(meta toml.MetaData) {
 	if !meta.IsDefined("master", "http_host") {
 		config.Master.HttpHost = defaultMasterConfig.HttpHost
 	}
-	if !meta.IsDefined("master", "fit_jobs") {
-		config.Master.FitJobs = defaultMasterConfig.FitJobs
-	}
-	if !meta.IsDefined("master", "search_jobs") {
-		config.Master.SearchJobs = defaultMasterConfig.SearchJobs
+	if !meta.IsDefined("master", "n_jobs") {
+		config.Master.NumJobs = defaultMasterConfig.NumJobs
 	}
 	if !meta.IsDefined("master", "meta_timeout") {
 		config.Master.MetaTimeout = defaultMasterConfig.MetaTimeout

config/config.toml.template

@@ -3,14 +3,14 @@
 
 # The database for caching, support Redis only:
 #   redis://<user>:<password>@<host>:<port>/<db_number>
-cache_store = "redis://localhost:6379"
+cache_store = "redis://localhost:6379/0"
 
 # The database for persist data, support MySQL, Postgres, ClickHouse and MongoDB:
 #   mysql://[username[:password]@][protocol[(address)]]/dbname[?param1=value1&...&paramN=valueN]
 #   postgres://bob:[email protected]:5432/mydb?sslmode=verify-full
 #   clickhouse://user:password@host[:port]/database?param1=value1&...&paramN=valueN
 #   mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]
-data_store = "mysql://root@tcp(localhost:3306)/gorse?parseTime=true"
+data_store = "mysql://root:password@tcp(localhost:3306)/gorse?parseTime=true"
 
 # The cache size for recommended/popular/latest items. The default value is 100.
 cache_size = 200
@@ -42,8 +42,7 @@ port = 8086                     # master port
 host = "0.0.0.0"                # master host
 http_port = 8088                # HTTP API port
 http_host = "0.0.0.0"           # HTTP API host
-search_jobs = 2                 # number of jobs for model search
-fit_jobs = 2                    # number of jobs for model fitting
+n_jobs = 4                      # number of working jobs
 meta_timeout = 10               # cluster meta timeout (second)
 
 # This section declares settings for the server node.

config/config_test.go

@@ -25,8 +25,8 @@ func TestLoadConfig(t *testing.T) {
 	assert.NoError(t, err)
 
 	// database configuration
-	assert.Equal(t, "redis://localhost:6379", config.Database.CacheStore)
-	assert.Equal(t, "mysql://root@tcp(localhost:3306)/gorse?parseTime=true", config.Database.DataStore)
+	assert.Equal(t, "redis://localhost:6379/0", config.Database.CacheStore)
+	assert.Equal(t, "mysql://root:password@tcp(localhost:3306)/gorse?parseTime=true", config.Database.DataStore)
 	assert.Equal(t, true, config.Database.AutoInsertUser)
 	assert.Equal(t, false, config.Database.AutoInsertItem)
 	assert.Equal(t, 200, config.Database.CacheSize)
@@ -41,8 +41,7 @@ func TestLoadConfig(t *testing.T) {
 	assert.Equal(t, "0.0.0.0", config.Master.Host)
 	assert.Equal(t, 8088, config.Master.HttpPort)
 	assert.Equal(t, "0.0.0.0", config.Master.HttpHost)
-	assert.Equal(t, 2, config.Master.SearchJobs)
-	assert.Equal(t, 2, config.Master.FitJobs)
+	assert.Equal(t, 4, config.Master.NumJobs)
 	assert.Equal(t, 10, config.Master.MetaTimeout)
 
 	// server configuration

docker/config.toml

@@ -42,8 +42,7 @@ port = 8086                     # master port
 host = "0.0.0.0"                # master host
 http_port = 8088                # HTTP API port
 http_host = "0.0.0.0"           # HTTP API host
-search_jobs = 2                 # number of jobs for model search
-fit_jobs = 2                    # number of jobs for model fitting
+n_jobs = 4                      # number of working jobs
 meta_timeout = 10               # cluster meta timeout (second)
 
 # This section declares settings for the server node.

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.3.0 // indirect
-	github.com/gorse-io/dashboard v0.0.0-20210808135039-daf299fb1808
+	github.com/gorse-io/dashboard v0.0.0-20210815043718-3ceb24c81758
 	github.com/haxii/go-swagger-ui v3.19.4+incompatible
 	github.com/json-iterator/go v1.1.10
 	github.com/juju/errors v0.0.0-20200330140219-3fe23663418f

go.sum

@@ -195,7 +195,6 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -205,8 +204,8 @@ github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2z
 github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
-github.com/gorse-io/dashboard v0.0.0-20210808135039-daf299fb1808 h1:RguWnzYAceMiLKUVhfkciFfyqNvqgc+OB5ssgWU2uIc=
-github.com/gorse-io/dashboard v0.0.0-20210808135039-daf299fb1808/go.mod h1:14j0KVMmOAAN0xvsa2wxFoV/X8Aa/gCeNFRLjtYP6+M=
+github.com/gorse-io/dashboard v0.0.0-20210815043718-3ceb24c81758 h1:4rUiFGaPS07hCuYZBTL2AdmgS5LmXTOxPPiLwt++1eo=
+github.com/gorse-io/dashboard v0.0.0-20210815043718-3ceb24c81758/go.mod h1:14j0KVMmOAAN0xvsa2wxFoV/X8Aa/gCeNFRLjtYP6+M=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -294,12 +293,10 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxv
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
@@ -468,6 +465,7 @@ github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
@@ -529,7 +527,6 @@ golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a h1:vclmkQCjlDX5OydZ9wv8rBCcS0QyQY66Mpf/7BZbInM=
 golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -571,7 +568,6 @@ golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/
 golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200904194848-62affa334b73/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb h1:eBmm0M9fYhWpKZLjQUUKka/LtIxf46G4fxeEz5KJr9U=
 golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110 h1:qWPm9rbaAMKs8Bq/9LRpbMqxWRVUAQwMI9fVrssnTfw=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
@@ -702,7 +698,6 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLks
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20160105164936-4f90aeace3a2/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

master/master.go

@@ -44,7 +44,8 @@ type Master struct {
 	protocol.UnimplementedMasterServer
 	server.RestServer
 
-	taskMonitor *TaskMonitor
+	taskMonitor   *TaskMonitor
+	taskScheduler *TaskScheduler
 
 	// cluster meta cache
 	ttlCache       *ttlcache.Cache
@@ -94,9 +95,17 @@ type Master struct {
 // NewMaster creates a master node.
 func NewMaster(cfg *config.Config) *Master {
 	rand.Seed(time.Now().UnixNano())
+	// create task monitor
+	taskMonitor := NewTaskMonitor()
+	for _, taskName := range []string{TaskFindLatest, TaskFindPopular, TaskFindNeighbor, TaskFitRankingModel,
+		TaskFitClickModel, TaskAnalyze, TaskSearchRankingModel, TaskSearchClickModel} {
+		taskMonitor.Pending(taskName)
+	}
 	return &Master{
-		nodesInfo:   make(map[string]*Node),
-		taskMonitor: NewTaskMonitor(),
+		nodesInfo: make(map[string]*Node),
+		// create task monitor
+		taskMonitor:   taskMonitor,
+		taskScheduler: NewTaskScheduler(),
 		// init versions
 		rankingModelVersion: rand.Int63(),
 		clickModelVersion:   rand.Int63(),
@@ -107,13 +116,13 @@ func NewMaster(cfg *config.Config) *Master {
 		rankingModelSearcher: ranking.NewModelSearcher(
 			cfg.Recommend.SearchEpoch,
 			cfg.Recommend.SearchTrials,
-			cfg.Master.SearchJobs),
+			cfg.Master.NumJobs),
 		// default click model
 		clickModel: click.NewFM(click.FMClassification, nil),
 		clickModelSearcher: click.NewModelSearcher(
 			cfg.Recommend.SearchEpoch,
 			cfg.Recommend.SearchTrials,
-			cfg.Master.SearchJobs,
+			cfg.Master.NumJobs,
 		),
 		RestServer: server.RestServer{
 			GorseConfig: cfg,
@@ -196,12 +205,10 @@ func (m *Master) Serve() {
 	}
 
 	go m.StartHttpServer()
-	go m.FitLoop()
+	go m.RunPrivilegedTasksLoop()
 	base.Logger().Info("start model fit", zap.Int("period", m.GorseConfig.Recommend.FitPeriod))
-	go m.SearchLoop()
+	go m.RunRagtagTasksLoop()
 	base.Logger().Info("start model searcher", zap.Int("period", m.GorseConfig.Recommend.SearchPeriod))
-	go m.AnalyzeLoop()
-	base.Logger().Info("start analyze")
 
 	// start rpc server
 	base.Logger().Info("start rpc server",
@@ -219,7 +226,7 @@ func (m *Master) Serve() {
 	}
 }
 
-func (m *Master) FitLoop() {
+func (m *Master) RunPrivilegedTasksLoop() {
 	defer base.CheckPanic()
 	var (
 		lastNumRankingUsers    int
@@ -258,27 +265,38 @@ func (m *Master) FitLoop() {
 			continue
 		}
 
+		// pre-lock tasks
+		tasksNames := []string{TaskFindLatest, TaskFindPopular, TaskFindNeighbor, TaskFitRankingModel, TaskFitClickModel}
+		for _, taskName := range tasksNames {
+			m.taskScheduler.PreLock(taskName)
+		}
+
 		// fit ranking model
 		lastNumRankingUsers, lastNumRankingItems, lastNumRankingFeedback, err =
-			m.fitRankingModelAndNonPersonalized(lastNumRankingUsers, lastNumRankingItems, lastNumRankingFeedback)
+			m.runRankingRelatedTasks(lastNumRankingUsers, lastNumRankingItems, lastNumRankingFeedback)
 		if err != nil {
 			base.Logger().Error("failed to fit ranking model", zap.Error(err))
 			continue
 		}
 
 		// fit click model
 		lastNumClickUsers, lastNumClickItems, lastNumClickFeedback, err =
-			m.fitClickModel(lastNumClickUsers, lastNumClickItems, lastNumClickFeedback)
+			m.runFitClickModelTask(lastNumClickUsers, lastNumClickItems, lastNumClickFeedback)
 		if err != nil {
 			base.Logger().Error("failed to fit click model", zap.Error(err))
 			continue
 		}
+
+		// release locks
+		for _, taskName := range tasksNames {
+			m.taskScheduler.UnLock(taskName)
+		}
 	}
 }
 
-// SearchLoop searches optimal recommendation model in background. It never modifies variables other than
+// RunRagtagTasksLoop searches optimal recommendation model in background. It never modifies variables other than
 // rankingModelSearcher, clickSearchedModel and clickSearchedScore.
-func (m *Master) SearchLoop() {
+func (m *Master) RunRagtagTasksLoop() {
 	defer base.CheckPanic()
 	var (
 		lastNumRankingUsers     int
@@ -290,13 +308,19 @@ func (m *Master) SearchLoop() {
 		err                     error
 	)
 	for {
+		// analyze click-through-rate
+		if err := m.runAnalyzeTask(); err != nil {
+			base.Logger().Error("failed to analyze", zap.Error(err))
+		}
+		// search optimal ranking model
 		lastNumRankingUsers, lastNumRankingItems, lastNumRankingFeedbacks, err =
 			m.runSearchRankingModelTask(lastNumRankingUsers, lastNumRankingItems, lastNumRankingFeedbacks)
 		if err != nil {
 			base.Logger().Error("failed to search ranking model", zap.Error(err))
 			time.Sleep(time.Minute)
 			continue
 		}
+		// search optimal click model
 		lastNumClickUsers, lastNumClickItems, lastNumClickFeedbacks, err =
 			m.runSearchClickModelTask(lastNumClickUsers, lastNumClickItems, lastNumClickFeedbacks)
 		if err != nil {
@@ -308,15 +332,6 @@ func (m *Master) SearchLoop() {
 	}
 }
 
-func (m *Master) AnalyzeLoop() {
-	for {
-		if err := m.runAnalyzeTask(); err != nil {
-			base.Logger().Error("failed to analyze", zap.Error(err))
-		}
-		time.Sleep(time.Hour)
-	}
-}
-
 func (m *Master) hasFeedbackInserted() bool {
 	numInserted, err := m.CacheClient.GetInt(cache.GlobalMeta, cache.NumInserted)
 	if err != nil {