Repository: argoproj/argo (go)

Issue:
semaphore limit leads to far lower number of conccurent workflows than expected
## Summary 

Using a semaphore to limit the concurrent workflow executions leads to a far lower number of concurrent executions. I started 100 instances of a workflow limited to 6 concurrent executions using a semaphore. This was more-or-less taken from the argo examples. Interestingly in the beginning argo executed 6 Worflows in parallel, but after that dropped the number of concurrent executions of the workflow.

I logged the executions using a script and gnuplot and got this graph of concurrent executions over time:

![output](https://user-images.githubusercontent.com/140716/97180967-0bc8a280-179b-11eb-9da8-31e608ac030f.png)

The repetition of this test resulted in similar chart.

## Diagnostics

What Kubernetes provider are you using? 

```
$ k version
Client Version: version.Info{Major:"1", Minor:"18", GitVersion:"v1.18.0", GitCommit:"9e991415386e4cf155a24b1da15becaa390438d8", GitTreeState:"clean", BuildDate:"2020-03-25T14:58:59Z", GoVersion:"go1.13.8", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"18", GitVersion:"v1.18.0", GitCommit:"9e991415386e4cf155a24b1da15becaa390438d8", GitTreeState:"clean", BuildDate:"2020-03-25T14:50:46Z", GoVersion:"go1.13.8", Compiler:"gc", Platform:"linux/amd64"}
```

What version of Argo Workflows are you running? 

argo 2.11.5

Configmap used for semaphore:
```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-config
data:
  workflow: "6"
```

Workflow:

```yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: synchronization-wf-level-
spec:
  entrypoint: whalesay
  synchronization:
    semaphore:
      configMapKeyRef:
        name: my-config
        key: workflow
  templates:
    - name: whalesay
      container:
        image: docker/whalesay:latest
        command: [cowsay]
        args: ["hello world"]
        imagePullPolicy: "IfNotPresent"

```

Logs  - these are just the last few lines, as I got 1.2Mb of logs for this. In case you need more to investigate this, I can provide more

```
time="2020-10-26T13:42:31Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:31Z" level=info msg="Updating node synchronization-wf-level-hpg6g message: ContainerCreating"
time="2020-10-26T13:42:31Z" level=info msg="Skipped pod synchronization-wf-level-hpg6g (synchronization-wf-level-hpg6g) creation: already exists" namespace=gzs-default podPhase=Pending workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:31Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Running resourceVersion=52456084 workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:32Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:32Z" level=info msg="Updating node synchronization-wf-level-hpg6g status Pending -> Running"
time="2020-10-26T13:42:32Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Running resourceVersion=52456099 workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Setting node synchronization-wf-level-hpg6g outputs"
time="2020-10-26T13:42:33Z" level=info msg="Updating node synchronization-wf-level-hpg6g status Running -> Succeeded"
time="2020-10-26T13:42:33Z" level=info msg="Updated phase Running -> Succeeded" namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Marking workflow completed" namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Checking daemoned children of " namespace=gzs-default workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Lock has been released by gzs-default/synchronization-wf-level-hpg6g. Available locks: 6" semaphore=gzs-default/ConfigMap/my-config/workflow
time="2020-10-26T13:42:33Z" level=info msg="gzs-default/synchronization-wf-level-hpg6g released a lock from gzs-default/ConfigMap/my-config/workflow"
time="2020-10-26T13:42:33Z" level=info msg="Released all acquired locks" key=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Succeeded resourceVersion=52456104 workflow=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:33Z" level=info msg="gzs-default/synchronization-wf-level-hpg6g released a lock from gzs-default/ConfigMap/my-config/workflow"
time="2020-10-26T13:42:33Z" level=info msg="Found Workflow gzs-default/synchronization-wf-level-hpg6g set expire at 2020-10-26 13:52:33 +0000 UTC (9m59.791764994s from now)"
time="2020-10-26T13:42:33Z" level=info msg="Queueing workflow gzs-default/synchronization-wf-level-hpg6g for delete in 9m59.791764994s"
time="2020-10-26T13:42:33Z" level=info msg="Found Workflow gzs-default/synchronization-wf-level-hpg6g set expire at 2020-10-26 13:52:33 +0000 UTC (9m59.791189794s from now)"
time="2020-10-26T13:42:33Z" level=info msg="Queueing workflow gzs-default/synchronization-wf-level-hpg6g for delete in 9m59.791189794s"
time="2020-10-26T13:42:34Z" level=info msg="gzs-default/ConfigMap/my-config/workflow acquired by gzs-default/synchronization-wf-level-rfv88 " semaphore=gzs-default/ConfigMap/my-config/workflow
time="2020-10-26T13:42:34Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:34Z" level=info msg="Updated phase  -> Running" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:34Z" level=info msg="Pod node synchronization-wf-level-rfv88 initialized Pending" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:34Z" level=info msg="Created pod: synchronization-wf-level-rfv88 (synchronization-wf-level-rfv88)" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:34Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Running resourceVersion=52456118 workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:34Z" level=info msg="Delete pod for gc successfully" namespace=gzs-default pod=synchronization-wf-level-hpg6g
time="2020-10-26T13:42:35Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:35Z" level=info msg="Updating node synchronization-wf-level-rfv88 message: ContainerCreating"
time="2020-10-26T13:42:35Z" level=info msg="Skipped pod synchronization-wf-level-rfv88 (synchronization-wf-level-rfv88) creation: already exists" namespace=gzs-default podPhase=Pending workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:35Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Running resourceVersion=52456128 workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:36Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:36Z" level=info msg="Updating node synchronization-wf-level-rfv88 status Pending -> Running"
time="2020-10-26T13:42:36Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Running resourceVersion=52456142 workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Processing workflow" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Setting node synchronization-wf-level-rfv88 outputs"
time="2020-10-26T13:42:37Z" level=info msg="Updating node synchronization-wf-level-rfv88 status Running -> Succeeded"
time="2020-10-26T13:42:37Z" level=info msg="Updated phase Running -> Succeeded" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Marking workflow completed" namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Checking daemoned children of " namespace=gzs-default workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Lock has been released by gzs-default/synchronization-wf-level-rfv88. Available locks: 6" semaphore=gzs-default/ConfigMap/my-config/workflow
time="2020-10-26T13:42:37Z" level=info msg="gzs-default/synchronization-wf-level-rfv88 released a lock from gzs-default/ConfigMap/my-config/workflow"
time="2020-10-26T13:42:37Z" level=info msg="Released all acquired locks" key=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Workflow update successful" namespace=gzs-default phase=Succeeded resourceVersion=52456147 workflow=synchronization-wf-level-rfv88
time="2020-10-26T13:42:37Z" level=info msg="Found Workflow gzs-default/synchronization-wf-level-rfv88 set expire at 2020-10-26 13:52:37 +0000 UTC (9m59.698888772s from now)"
time="2020-10-26T13:42:37Z" level=info msg="Queueing workflow gzs-default/synchronization-wf-level-rfv88 for delete in 9m59.698888772s"
time="2020-10-26T13:42:37Z" level=info msg="gzs-default/synchronization-wf-level-rfv88 released a lock from gzs-default/ConfigMap/my-config/workflow"
time="2020-10-26T13:42:37Z" level=info msg="Found Workflow gzs-default/synchronization-wf-level-rfv88 set expire at 2020-10-26 13:52:37 +0000 UTC (9m59.698308465s from now)"
time="2020-10-26T13:42:37Z" level=info msg="Queueing workflow gzs-default/synchronization-wf-level-rfv88 for delete in 9m59.698308465s"
time="2020-10-26T13:42:38Z" level=info msg="Delete pod for gc successfully" namespace=gzs-default pod=synchronization-wf-level-rfv88

```

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


Pull request description:
fix(controller): Trigger no of workflows based on available lock
Signed-off-by: Saravanan Balasubramanian <sarabala1979@gmail.com>

Checklist:

* [ ] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [ ] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).
* [ ] I've signed the CLA.
* [ ] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [ ] My builds are green. Try syncing with master if they are not. 
Fixes #4377


Patch:
diff --git a/workflow/sync/semaphore.go b/workflow/sync/semaphore.go
index a6458d0a1..65cd854ba 100644
--- a/workflow/sync/semaphore.go
+++ b/workflow/sync/semaphore.go
@@ -86,18 +86,24 @@ func (s *PrioritySemaphore) release(key string) bool {
 		}
 
 		s.semaphore.Release(1)
-
-		s.log.Infof("Lock has been released by %s. Available locks: %d", key, s.limit-len(s.lockHolder))
+		availableLocks := s.limit - len(s.lockHolder)
+		s.log.Infof("Lock has been released by %s. Available locks: %d", key, availableLocks)
 		if s.pending.Len() > 0 {
-			item := s.pending.peek()
-			keyStr := fmt.Sprint(item.key)
-			items := strings.Split(keyStr, "/")
-			workflowKey := keyStr
-			if len(items) == 3 {
-				workflowKey = fmt.Sprintf("%s/%s", items[0], items[1])
+			triggerCount := availableLocks
+			if s.pending.Len() < triggerCount {
+				triggerCount = s.pending.Len()
+			}
+			for idx := 0; idx < triggerCount; idx++ {
+				item := s.pending.items[idx]
+				keyStr := fmt.Sprint(item.key)
+				items := strings.Split(keyStr, "/")
+				workflowKey := keyStr
+				if len(items) == 3 {
+					workflowKey = fmt.Sprintf("%s/%s", items[0], items[1])
+				}
+				s.log.Debugf("Enqueue the workflow %s", workflowKey)
+				s.nextWorkflow(workflowKey)
 			}
-			s.log.Debugf("Enqueue the workflow %s", workflowKey)
-			s.nextWorkflow(workflowKey)
 		}
 	}
 	return true


Test patch:
diff --git a/workflow/sync/sync_manager_test.go b/workflow/sync/sync_manager_test.go
index fdf416600..5a12c594d 100644
--- a/workflow/sync/sync_manager_test.go
+++ b/workflow/sync/sync_manager_test.go
@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"testing"
@@ -545,3 +546,46 @@ func TestSemaphoreTmplLevel(t *testing.T) {
 
 	})
 }
+
+func TestTriggerWFWithAvailableLock(t *testing.T) {
+	assert := assert.New(t)
+	kube := fake.NewSimpleClientset()
+	var cm v1.ConfigMap
+	err := yaml.Unmarshal([]byte(configMap), &cm)
+	cm.Data["workflow"] = "3"
+	assert.NoError(err)
+	_, err = kube.CoreV1().ConfigMaps("default").Create(&cm)
+	assert.NoError(err)
+	syncLimitFunc := GetSyncLimitFunc(kube)
+	t.Run("TriggerWfsWithAvailableLocks", func(t *testing.T) {
+		triggerCount := 0
+		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
+			triggerCount++
+		})
+		var wfs []wfv1.Workflow
+		for i := 0; i < 3; i++ {
+			wf := unmarshalWF(wfWithSemaphore)
+			wf.Name = fmt.Sprintf("%s-%d", "acquired", i)
+			status, wfUpdate, msg, err := concurrenyMgr.TryAcquire(wf, "", wf.Spec.Synchronization)
+			assert.NoError(err)
+			assert.Empty(msg)
+			assert.True(status)
+			assert.True(wfUpdate)
+			wfs = append(wfs, *wf)
+
+		}
+		for i := 0; i < 3; i++ {
+			wf := unmarshalWF(wfWithSemaphore)
+			wf.Name = fmt.Sprintf("%s-%d", "wait", i)
+			status, wfUpdate, msg, err := concurrenyMgr.TryAcquire(wf, "", wf.Spec.Synchronization)
+			assert.NoError(err)
+			assert.NotEmpty(msg)
+			assert.False(status)
+			assert.True(wfUpdate)
+		}
+		concurrenyMgr.Release(&wfs[0], "", wfs[0].Spec.Synchronization)
+		triggerCount = 0
+		concurrenyMgr.Release(&wfs[1], "", wfs[1].Spec.Synchronization)
+		assert.Equal(2, triggerCount)
+	})
+}