Fixed all the comments

cmd/controllers/app/server.go

@@ -90,7 +90,7 @@ func Run(opt *options.ServerOption) error {
 	garbageCollector := garbagecollector.New(vkClient)
 
 	run := func(ctx context.Context) {
-		go jobController.Run(opt.WorkerThreads, ctx.Done())
+		go jobController.Run(ctx.Done())
 		go queueController.Run(ctx.Done())
 		go garbageCollector.Run(ctx.Done())
 		<-ctx.Done()

pkg/controllers/job/helpers/helpers.go

@@ -22,6 +22,7 @@ import (
 	"math/rand"
 	"strings"
 	"time"
+	"volcano.sh/volcano/pkg/controllers/apis"
 )
 
 const (
@@ -61,3 +62,8 @@ func genRandomStr(l int) string {
 func MakeVolumeClaimName(jobName string) string {
 	return fmt.Sprintf(VolumeClaimFmt, jobName, genRandomStr(12))
 }
+
+// GetJobKeyByReq gets the key for the job request
+func GetJobKeyByReq(req *apis.Request) string {
+	return fmt.Sprintf("%s/%s", req.Namespace, req.JobName)
+}

pkg/controllers/job/job_controller.go

@@ -18,13 +18,13 @@ package job
 
 import (
 	"fmt"
-	"sync"
-
-	"github.com/golang/glog"
 	"hash"
 	"hash/fnv"
+	"sync"
 	"time"
 
+	"github.com/golang/glog"
+
 	"k8s.io/api/core/v1"
 	"k8s.io/api/scheduling/v1beta1"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -107,10 +107,7 @@ type Controller struct {
 
 	sync.Mutex
 	errTasks workqueue.RateLimitingInterface
-
-	// To protect the queue list by processing multiple workers
-	lock    sync.RWMutex
-	workers uint32
+	workers  uint32
 }
 
 // NewJobController create new Job Controller
@@ -119,7 +116,6 @@ func NewJobController(
 	kbClient kbver.Interface,
 	vkClient vkver.Interface,
 	workers uint32,
-
 ) *Controller {
 
 	//Initialize event client
@@ -205,7 +201,7 @@ func NewJobController(
 }
 
 // Run start JobController
-func (cc *Controller) Run(workers uint32, stopCh <-chan struct{}) {
+func (cc *Controller) Run(stopCh <-chan struct{}) {
 
 	go cc.sharedInformers.Start(stopCh)
 	go cc.jobInformer.Informer().Run(stopCh)
@@ -221,7 +217,7 @@ func (cc *Controller) Run(workers uint32, stopCh <-chan struct{}) {
 
 	go wait.Until(cc.handleCommands, 0, stopCh)
 	var i uint32
-	for i = 0; i < workers; i++ {
+	for i = 0; i < cc.workers; i++ {
 		go func(num uint32) {
 			wait.Until(
 				func() {
@@ -230,7 +226,6 @@ func (cc *Controller) Run(workers uint32, stopCh <-chan struct{}) {
 				time.Second,
 				stopCh)
 		}(i)
-
 	}
 
 	go cc.cache.Run(stopCh)
@@ -242,14 +237,12 @@ func (cc *Controller) Run(workers uint32, stopCh <-chan struct{}) {
 }
 
 func (cc *Controller) worker(i uint32) {
-
 	glog.Infof("worker %d start ...... ", i)
 
 	for cc.processNextReq(i) {
 	}
 }
 
-// TODO we may need to make this sharding more proper if required
 func (cc *Controller) belongsToThisRoutine(key string, count uint32) bool {
 	var hashVal hash.Hash32
 	var val uint32
@@ -266,8 +259,7 @@ func (cc *Controller) belongsToThisRoutine(key string, count uint32) bool {
 	return false
 }
 
-// TODO we may need to make this sharding more proper if required
-func (cc *Controller) getWorkerID(key string) workqueue.RateLimitingInterface {
+func (cc *Controller) getWorkerQueue(key string) workqueue.RateLimitingInterface {
 	var hashVal hash.Hash32
 	var val uint32
 
@@ -276,17 +268,13 @@ func (cc *Controller) getWorkerID(key string) workqueue.RateLimitingInterface {
 
 	val = hashVal.Sum32()
 
-	cc.lock.Lock()
 	queue := cc.queueList[val%cc.workers]
-	cc.lock.Unlock()
 
 	return queue
 }
 
 func (cc *Controller) processNextReq(count uint32) bool {
-	cc.lock.Lock()
 	queue := cc.queueList[count]
-	cc.lock.Unlock()
 	obj, shutdown := queue.Get()
 	if shutdown {
 		glog.Errorf("Fail to pop item from queue")
@@ -297,10 +285,9 @@ func (cc *Controller) processNextReq(count uint32) bool {
 	defer queue.Done(req)
 
 	key := jobcache.JobKeyByReq(&req)
-	// Later we can remove this code if we want
 	if !cc.belongsToThisRoutine(key, count) {
 		glog.Errorf("should not occur The job does not belongs to this routine key:%s, worker:%d...... ", key, count)
-		queueLocal := cc.getWorkerID(key)
+		queueLocal := cc.getWorkerQueue(key)
 		queueLocal.Add(req)
 		return true
 	}

pkg/controllers/job/job_controller_handler.go

@@ -34,6 +34,7 @@ import (
 
 	"volcano.sh/volcano/pkg/controllers/apis"
 	vkcache "volcano.sh/volcano/pkg/controllers/cache"
+	vkjobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers"
 )
 
 func (cc *Controller) addCommand(obj interface{}) {
@@ -65,8 +66,8 @@ func (cc *Controller) addJob(obj interface{}) {
 		glog.Errorf("Failed to add job <%s/%s>: %v in cache",
 			job.Namespace, job.Name, err)
 	}
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 }
 
@@ -102,8 +103,8 @@ func (cc *Controller) updateJob(oldObj, newObj interface{}) {
 		Event: vkbatchv1.OutOfSyncEvent,
 	}
 
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 }
 
@@ -169,8 +170,8 @@ func (cc *Controller) addPod(obj interface{}) {
 		glog.Errorf("Failed to add Pod <%s/%s>: %v to cache",
 			pod.Namespace, pod.Name, err)
 	}
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 }
 
@@ -247,8 +248,8 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 		JobVersion: int32(dVersion),
 	}
 
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 }
 
@@ -310,8 +311,8 @@ func (cc *Controller) deletePod(obj interface{}) {
 			pod.Namespace, pod.Name, err)
 	}
 
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 }
 
@@ -357,8 +358,8 @@ func (cc *Controller) processNextCommand() bool {
 		Action:    vkbatchv1.Action(cmd.Action),
 	}
 
-	key := vkcache.JobKeyByReq(&req)
-	queue := cc.getWorkerID(key)
+	key := vkjobhelpers.GetJobKeyByReq(&req)
+	queue := cc.getWorkerQueue(key)
 	queue.Add(req)
 
 	return true
@@ -394,8 +395,8 @@ func (cc *Controller) updatePodGroup(oldObj, newObj interface{}) {
 		case kbtype.PodGroupInqueue:
 			req.Action = vkbatchv1.EnqueueAction
 		}
-		key := vkcache.JobKeyByReq(&req)
-		queue := cc.getWorkerID(key)
+		key := vkjobhelpers.GetJobKeyByReq(&req)
+		queue := cc.getWorkerQueue(key)
 		queue.Add(req)
 	}
 }

pkg/controllers/job/job_controller_handler_test.go

@@ -59,8 +59,7 @@ func newController() *Controller {
 	}
 
 	vkclient := vkclientset.NewForConfigOrDie(config)
-	controller := NewJobController(kubeClientSet, kubeBatchClientSet, vkclient, 1)
-	controller.workers = 1
+	controller := NewJobController(kubeClientSet, kubeBatchClientSet, vkclient, 3)
 
 	return controller
 }
@@ -161,7 +160,8 @@ func TestJobAddFunc(t *testing.T) {
 		if job == nil || err != nil {
 			t.Errorf("Error while Adding Job in case %d with error %s", i, err)
 		}
-		len := controller.queueList[0].Len()
+		queue := controller.getWorkerQueue(key)
+		len := queue.Len()
 		if testcase.ExpectValue != len {
 			t.Errorf("case %d (%s): expected: %v, got %v ", i, testcase.Name, testcase.ExpectValue, len)
 		}
@@ -512,7 +512,9 @@ func TestUpdatePodGroupFunc(t *testing.T) {
 	for i, testcase := range testCases {
 		controller := newController()
 		controller.updatePodGroup(testcase.oldPodGroup, testcase.newPodGroup)
-		len := controller.queueList[0].Len()
+		key := fmt.Sprintf("%s/%s", testcase.oldPodGroup.Namespace, testcase.oldPodGroup.Name)
+		queue := controller.getWorkerQueue(key)
+		len := queue.Len()
 		if testcase.ExpectValue != len {
 			t.Errorf("case %d (%s): expected: %v, got %v ", i, testcase.Name, testcase.ExpectValue, len)
 		}

pkg/controllers/job/job_controller_plugins_test.go

@@ -35,7 +35,7 @@ func newFakeController() *Controller {
 	VolcanoClientSet := volcanoclient.NewSimpleClientset()
 	KubeClientSet := kubeclient.NewSimpleClientset()
 
-	controller := NewJobController(KubeClientSet, KubeBatchClientSet, VolcanoClientSet, 1)
+	controller := NewJobController(KubeClientSet, KubeBatchClientSet, VolcanoClientSet, 3)
 	return controller
 }