Repository: dragonflyoss/Dragonfly2
Pull request #1000 (2022-01-17 10:38:24)

feat: task mutex replace sync kmutex
<!--- Provide a general summary of your changes in the Title above -->

## Description
- Task mutex replace sync kmutex.
<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate):

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1000](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (47fc8c6) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/97f94cfa4fdcc0d94e4d6756584336fd38814734?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (97f94cf) will **increase** coverage by `45.66%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff            @@
##           main    #1000       +/-   ##
=========================================
+ Coverage      0   45.66%   +45.66%     
=========================================
  Files         0      128      +128     
  Lines         0     9094     +9094     
=========================================
+ Hits          0     4153     +4153     
- Misses        0     4473     +4473     
- Partials      0      468      +468     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `45.66% <ø> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [scheduler/service/service.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NlcnZpY2Uvc2VydmljZS5nbw==) | `92.68% <ø> (ø)` | |
| [scheduler/config/config.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL2NvbmZpZy9jb25maWcuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/ratelimiter/ratelimiter/ratelimiter.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3JhdGVsaW1pdGVyL3JhdGVsaW1pdGVyL3JhdGVsaW1pdGVyLmdv) | `96.61% <0.00%> (ø)` | |
| [client/daemon/transport/transport.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi90cmFuc3BvcnQvdHJhbnNwb3J0Lmdv) | `59.48% <0.00%> (ø)` | |
| [cdn/plugins/plugin\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3BsdWdpbnMvcGx1Z2luX21hbmFnZXIuZ28=) | `96.07% <0.00%> (ø)` | |
| [client/daemon/proxy/proxy\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wcm94eS9wcm94eV9tYW5hZ2VyLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/source/httpprotocol/http\_source\_client.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9odHRwcHJvdG9jb2wvaHR0cF9zb3VyY2VfY2xpZW50Lmdv) | `63.00% <0.00%> (ø)` | |
| [cdn/supervisor/cdn/downloader.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvY2RuL2Rvd25sb2FkZXIuZ28=) | `48.38% <0.00%> (ø)` | |
| [client/config/peerhost.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9wZWVyaG9zdC5nbw==) | `35.29% <0.00%> (ø)` | |
| [pkg/reachable/reachable.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3JlYWNoYWJsZS9yZWFjaGFibGUuZ28=) | `100.00% <0.00%> (ø)` | |
| ... and [119 more](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1000/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | |


Patch:
diff --git a/deploy/docker-compose/config/scheduler.yaml b/deploy/docker-compose/config/scheduler.yaml
index 08bc1a29..dc0c7a18 100644
--- a/deploy/docker-compose/config/scheduler.yaml
+++ b/deploy/docker-compose/config/scheduler.yaml
@@ -9,5 +9,11 @@ worker:
   senderNum: 10
   senderJobPoolSize: 10000
 
+manager:
+  enable: false
+
+job:
+  enable: false
+
 dynconfig:
-  cdnDirPath: /opt/dragonfly/scheduler-cdn
+  cdnDir: /opt/dragonfly/scheduler-cdn
diff --git a/pkg/sync/kmutex.go b/pkg/sync/kmutex.go
deleted file mode 100644
index 833bc570..00000000
--- a/pkg/sync/kmutex.go
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- *     Copyright 2020 The Dragonfly Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package sync
-
-import (
-	"sync"
-)
-
-type Kmutex struct {
-	m *sync.Map
-}
-
-func NewKmutex() *Kmutex {
-	m := sync.Map{}
-	return &Kmutex{&m}
-}
-
-func (k *Kmutex) Lock(key interface{}) {
-	m := sync.Mutex{}
-	rm, _ := k.m.LoadOrStore(key, &m)
-	mu := rm.(*sync.Mutex)
-	mu.Lock()
-}
-
-func (k *Kmutex) Unlock(key interface{}) {
-	rm, ok := k.m.Load(key)
-	if !ok {
-		return
-	}
-
-	mu := rm.(*sync.Mutex)
-	k.m.Delete(key)
-	mu.Unlock()
-}
diff --git a/pkg/sync/krwmutex.go b/pkg/sync/krwmutex.go
deleted file mode 100644
index fe3b5743..00000000
--- a/pkg/sync/krwmutex.go
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- *     Copyright 2020 The Dragonfly Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package sync
-
-import (
-	"sync"
-)
-
-type Krwmutex struct {
-	m *sync.Map
-}
-
-func NewKrwmutex() *Krwmutex {
-	m := sync.Map{}
-	return &Krwmutex{&m}
-}
-
-func (k *Krwmutex) Lock(key interface{}) {
-	m := sync.RWMutex{}
-	rm, _ := k.m.LoadOrStore(key, &m)
-	mu := rm.(*sync.RWMutex)
-	mu.Lock()
-}
-
-func (k *Krwmutex) Unlock(key interface{}) {
-	rm, ok := k.m.Load(key)
-	if !ok {
-		return
-	}
-
-	mu := rm.(*sync.RWMutex)
-	k.m.Delete(key)
-	mu.Unlock()
-}
-
-func (k *Krwmutex) RLock(key interface{}) {
-	m := sync.RWMutex{}
-	rm, _ := k.m.LoadOrStore(key, &m)
-	mu := rm.(*sync.RWMutex)
-	mu.RLock()
-}
-
-func (k *Krwmutex) RUnlock(key interface{}) {
-	rm, ok := k.m.Load(key)
-	if !ok {
-		return
-	}
-
-	mu := rm.(*sync.RWMutex)
-	k.m.Delete(key)
-	mu.RUnlock()
-}
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 0bcb15ce..6f9170c7 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -24,7 +24,6 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
-	pkgsync "d7y.io/dragonfly/v2/pkg/sync"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	"d7y.io/dragonfly/v2/scheduler/metrics"
 	"d7y.io/dragonfly/v2/scheduler/resource"
@@ -58,9 +57,6 @@ type service struct {
 
 	// Dynamic config
 	dynconfig config.DynconfigInterface
-
-	// Key map mutex
-	kmu *pkgsync.Krwmutex
 }
 
 func New(
@@ -78,7 +74,6 @@ func New(
 		callback:  callback,
 		config:    cfg,
 		dynconfig: dynconfig,
-		kmu:       pkgsync.NewKrwmutex(),
 	}
 }
 
@@ -92,9 +87,6 @@ func (s *service) CDN() resource.CDN {
 
 func (s *service) RegisterTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*resource.Task, error) {
 	task := resource.NewTask(idgen.TaskID(req.Url, req.UrlMeta), req.Url, s.config.Scheduler.BackSourceCount, req.UrlMeta)
-
-	s.kmu.Lock(task.ID)
-	defer s.kmu.Unlock(task.ID)
 	task, ok := s.resource.TaskManager().LoadOrStore(task)
 	if ok && (task.FSM.Is(resource.TaskStateRunning) || task.FSM.Is(resource.TaskStateSucceeded)) {
 		// Task is healthy and can be reused


Test patch:
diff --git a/pkg/sync/kmutex_test.go b/pkg/sync/kmutex_test.go
deleted file mode 100644
index 6154fce3..00000000
--- a/pkg/sync/kmutex_test.go
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- *     Copyright 2020 The Dragonfly Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package sync
-
-import (
-	"testing"
-	"time"
-)
-
-func TestKmutex_Lock(t *testing.T) {
-	k := NewKmutex()
-	done := time.After(time.Second)
-
-	k.Lock("foo")
-	foo := make(chan struct{})
-	go func() {
-		k.Lock("foo")
-		close(foo)
-	}()
-
-	k.Lock("bar")
-	bar := make(chan struct{})
-	go func() {
-		k.Lock("bar")
-		close(bar)
-	}()
-
-	k.Lock("baz")
-	baz := make(chan struct{})
-	go func() {
-		k.Lock("baz")
-		close(baz)
-	}()
-
-	select {
-	case <-bar:
-	case <-baz:
-	case <-foo:
-	case <-done:
-		t.Fatal("kmutex is not unlocked")
-	default:
-	}
-
-	k.Unlock("foo")
-	select {
-	case <-bar:
-	case <-baz:
-	case <-done:
-		t.Fatal("bar and baz are not unlocked")
-	case <-foo:
-		k.Unlock("foo")
-	}
-
-	k.Unlock("bar")
-	select {
-	case <-foo:
-	case <-baz:
-	case <-done:
-		t.Fatal("foo and baz are not unlocked")
-	case <-bar:
-		k.Unlock("bar")
-	}
-
-	k.Unlock("baz")
-	select {
-	case <-foo:
-	case <-bar:
-	case <-done:
-		t.Fatal("foo and bar are not unlocked")
-	case <-baz:
-		k.Unlock("baz")
-	}
-}
diff --git a/pkg/sync/krwmutex_test.go b/pkg/sync/krwmutex_test.go
deleted file mode 100644
index 615f51bf..00000000
--- a/pkg/sync/krwmutex_test.go
+++ /dev/null
@@ -1,152 +0,0 @@
-/*
- *     Copyright 2020 The Dragonfly Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package sync
-
-import (
-	"testing"
-	"time"
-)
-
-func TestKrwmutex_Lock(t *testing.T) {
-	k := NewKrwmutex()
-	done := time.After(time.Second)
-
-	k.Lock("foo")
-	foo := make(chan struct{})
-	go func() {
-		k.Lock("foo")
-		close(foo)
-	}()
-
-	k.Lock("bar")
-	bar := make(chan struct{})
-	go func() {
-		k.Lock("bar")
-		close(bar)
-	}()
-
-	k.Lock("baz")
-	baz := make(chan struct{})
-	go func() {
-		k.Lock("baz")
-		close(baz)
-	}()
-
-	select {
-	case <-bar:
-	case <-baz:
-	case <-foo:
-	case <-done:
-		t.Fatal("kmutex is not unlocked")
-	default:
-	}
-
-	k.Unlock("foo")
-	select {
-	case <-bar:
-	case <-baz:
-	case <-done:
-		t.Fatal("bar and baz are not unlocked")
-	case <-foo:
-		k.Unlock("foo")
-	}
-
-	k.Unlock("bar")
-	select {
-	case <-foo:
-	case <-baz:
-	case <-done:
-		t.Fatal("foo and baz are not unlocked")
-	case <-bar:
-		k.Unlock("bar")
-	}
-
-	k.Unlock("baz")
-	select {
-	case <-foo:
-	case <-bar:
-	case <-done:
-		t.Fatal("foo and bar are not unlocked")
-	case <-baz:
-		k.Unlock("baz")
-	}
-}
-
-func TestKrwmutex_RLock(t *testing.T) {
-	k := NewKrwmutex()
-	done := time.After(time.Second)
-
-	k.Lock("foo")
-	foo := make(chan struct{})
-	go func() {
-		k.RLock("foo")
-		close(foo)
-	}()
-
-	k.Lock("bar")
-	bar := make(chan struct{})
-	go func() {
-		k.RLock("bar")
-		close(bar)
-	}()
-
-	k.Lock("baz")
-	baz := make(chan struct{})
-	go func() {
-		k.RLock("baz")
-		close(baz)
-	}()
-
-	select {
-	case <-bar:
-	case <-baz:
-	case <-foo:
-	case <-done:
-		t.Fatal("kmutex is not unlocked")
-	default:
-	}
-
-	k.Unlock("foo")
-	select {
-	case <-bar:
-	case <-baz:
-	case <-done:
-		t.Fatal("bar and baz are not unlocked")
-	case <-foo:
-		k.RUnlock("foo")
-	}
-
-	k.Unlock("bar")
-	select {
-	case <-foo:
-	case <-baz:
-	case <-done:
-		t.Fatal("foo and baz are not unlocked")
-	case <-bar:
-		k.RUnlock("bar")
-	}
-
-	k.Unlock("baz")
-	select {
-	case <-foo:
-	case <-bar:
-	case <-done:
-		t.Fatal("foo and bar are not unlocked")
-	case <-baz:
-		k.RUnlock("baz")
-	}
-}