Repository: dragonflyoss/Dragonfly2
Pull request #1467 (2022-07-15 08:14:29)

fix: adjust the configuration check sequence
<!--- Provide a general summary of your changes in the Title above -->

## Description

Adjust the configuration check sequence.
Before the service starts to initialize, verify the configuration content first, which can ensure a smoother subsequent process.

<!--- 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)
- [ ] 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! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


gaius-qi: Please help to fix this problem in the manager, dfget and dfcache modules together. Thx.
codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1467](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (b190457) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/7439b0e6c57d810b51ad0ba575b217ea88c78402?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (7439b0e) will **decrease** coverage by `0.09%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/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/1467?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1467      +/-   ##
==========================================
- Coverage   43.74%   43.65%   -0.10%     
==========================================
  Files         118      118              
  Lines       10723    10735      +12     
==========================================
- Hits         4691     4686       -5     
- Misses       5564     5579      +15     
- Partials      468      470       +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `43.65% <ø> (-0.10%)` | :arrow_down: |

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/1467?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/dfget/dfget.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RmZ2V0L2RmZ2V0Lmdv) | `10.30% <0.00%> (-2.46%)` | :arrow_down: |
| [client/daemon/peer/peertask\_file.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2ZpbGUuZ28=) | `57.25% <0.00%> (-1.53%)` | :arrow_down: |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `47.61% <0.00%> (-1.30%)` | :arrow_down: |
| [client/daemon/storage/local\_storage\_subtask.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9zdG9yYWdlL2xvY2FsX3N0b3JhZ2Vfc3VidGFzay5nbw==) | `27.35% <0.00%> (-0.38%)` | :arrow_down: |
| [client/daemon/storage/local\_storage.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9zdG9yYWdlL2xvY2FsX3N0b3JhZ2UuZ28=) | `30.10% <0.00%> (-0.24%)` | :arrow_down: |
| [pkg/source/source\_client.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9zb3VyY2VfY2xpZW50Lmdv) | `1.61% <0.00%> (-0.04%)` | :arrow_down: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `57.11% <0.00%> (+0.29%)` | :arrow_up: |
| [client/daemon/upload/upload\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi91cGxvYWQvdXBsb2FkX21hbmFnZXIuZ28=) | `59.03% <0.00%> (+0.49%)` | :arrow_up: |
| [client/config/dfget.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1467/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9kZmdldC5nbw==) | `29.76% <0.00%> (+1.35%)` | :arrow_up: |

zc2638: I just started looking at the code so I ignored the other modules.
Done.
@gaius-qi 

Patch:
diff --git a/client/config/constants.go b/client/config/constants.go
index 7c1eab65..0bbcdacd 100644
--- a/client/config/constants.go
+++ b/client/config/constants.go
@@ -92,4 +92,5 @@ const (
 	DefaultPeerStartPort          = 65000
 	DefaultUploadStartPort        = 65002
 	DefaultObjectStorageStartPort = 65004
+	DefaultHealthyStartPort       = 40901
 )
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index b881240e..74d0f4ac 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -157,6 +157,22 @@ var peerHostConfig = func() *DaemonOption {
 			Multiplex:              false,
 			DiskGCThresholdPercent: 95,
 		},
+		Health: &HealthOption{
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultHealthyStartPort,
+						End:   DefaultEndPort,
+					},
+				},
+			},
+			Path: "/server/ping",
+		},
 		Reload: ReloadOption{
 			Interval: util.Duration{
 				Duration: time.Minute,
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index 22713e92..be720f58 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -156,6 +156,22 @@ var peerHostConfig = func() *DaemonOption {
 			Multiplex:              false,
 			DiskGCThresholdPercent: 95,
 		},
+		Health: &HealthOption{
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultHealthyStartPort,
+						End:   DefaultEndPort,
+					},
+				},
+			},
+			Path: "/server/ping",
+		},
 		Reload: ReloadOption{
 			Interval: util.Duration{
 				Duration: time.Minute,
diff --git a/client/daemon/proxy/mocks/proxy_manager_mock.go b/client/daemon/proxy/mocks/proxy_manager_mock.go
index 63acd19c..b62d445f 100644
--- a/client/daemon/proxy/mocks/proxy_manager_mock.go
+++ b/client/daemon/proxy/mocks/proxy_manager_mock.go
@@ -8,6 +8,7 @@ import (
 	net "net"
 	reflect "reflect"
 
+	config "d7y.io/dragonfly/v2/client/config"
 	gomock "github.com/golang/mock/gomock"
 )
 
@@ -89,3 +90,50 @@ func (mr *MockManagerMockRecorder) Stop() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockManager)(nil).Stop))
 }
+
+// Watch mocks base method.
+func (m *MockManager) Watch(arg0 *config.ProxyOption) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "Watch", arg0)
+}
+
+// Watch indicates an expected call of Watch.
+func (mr *MockManagerMockRecorder) Watch(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockManager)(nil).Watch), arg0)
+}
+
+// MockConfigWatcher is a mock of ConfigWatcher interface.
+type MockConfigWatcher struct {
+	ctrl     *gomock.Controller
+	recorder *MockConfigWatcherMockRecorder
+}
+
+// MockConfigWatcherMockRecorder is the mock recorder for MockConfigWatcher.
+type MockConfigWatcherMockRecorder struct {
+	mock *MockConfigWatcher
+}
+
+// NewMockConfigWatcher creates a new mock instance.
+func NewMockConfigWatcher(ctrl *gomock.Controller) *MockConfigWatcher {
+	mock := &MockConfigWatcher{ctrl: ctrl}
+	mock.recorder = &MockConfigWatcherMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockConfigWatcher) EXPECT() *MockConfigWatcherMockRecorder {
+	return m.recorder
+}
+
+// Watch mocks base method.
+func (m *MockConfigWatcher) Watch(arg0 *config.ProxyOption) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "Watch", arg0)
+}
+
+// Watch indicates an expected call of Watch.
+func (mr *MockConfigWatcherMockRecorder) Watch(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockConfigWatcher)(nil).Watch), arg0)
+}
diff --git a/client/daemon/upload/upload_manager.go b/client/daemon/upload/upload_manager.go
index c92edb11..7068f576 100644
--- a/client/daemon/upload/upload_manager.go
+++ b/client/daemon/upload/upload_manager.go
@@ -192,8 +192,6 @@ func (um *uploadManager) getDownload(ctx *gin.Context) {
 		return
 	}
 
-	// Add header "Content-Length" to avoid chunked body in http client.
-	ctx.Header(headers.ContentLength, fmt.Sprintf("%d", rg[0].Length))
 	reader, closer, err := um.storageManager.ReadPiece(ctx,
 		&storage.ReadPieceRequest{
 			PeerTaskMetadata: storage.PeerTaskMetadata{
@@ -212,8 +210,12 @@ func (um *uploadManager) getDownload(ctx *gin.Context) {
 	}
 	defer closer.Close()
 
+	// Add header "Content-Length" to avoid chunked body in http client.
+	ctx.Header(headers.ContentLength, fmt.Sprintf("%d", rg[0].Length))
+
 	// write header immediately, prevent client disconnecting after limiter.Wait() due to response header timeout
 	ctx.Writer.WriteHeaderNow()
+	ctx.Writer.Flush()
 
 	if um.Limiter != nil {
 		if err = um.Limiter.WaitN(ctx, int(rg[0].Length)); err != nil {
diff --git a/cmd/dfcache/cmd/root.go b/cmd/dfcache/cmd/root.go
index 57c3036c..36e839de 100644
--- a/cmd/dfcache/cmd/root.go
+++ b/cmd/dfcache/cmd/root.go
@@ -112,6 +112,15 @@ func initDfcacheDfpath(cfg *config.CacheOption) (dfpath.Dfpath, error) {
 
 // runDfcache does some init operations and starts to download.
 func runDfcacheSubcmd(cmdName string, args []string) error {
+	// Convert config
+	if err := dfcacheConfig.Convert(cmdName, args); err != nil {
+		return err
+	}
+	// Validate config
+	if err := dfcacheConfig.Validate(cmdName); err != nil {
+		return err
+	}
+
 	var (
 		daemonClient client.DaemonClient
 		err          error
@@ -129,16 +138,6 @@ func runDfcacheSubcmd(cmdName string, args []string) error {
 	}
 	logger.Infof("Version:\n%s", version.Version())
 
-	// Convert config
-	if err := dfcacheConfig.Convert(cmdName, args); err != nil {
-		return err
-	}
-
-	// Validate config
-	if err := dfcacheConfig.Validate(cmdName); err != nil {
-		return err
-	}
-
 	// Dfcache config values
 	s, _ := yaml.Marshal(dfcacheConfig)
 	logger.Infof("client dfcache configuration:\n%s", string(s))
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index 836bfe91..65bf06f2 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -55,6 +55,15 @@ it supports container engine, wget and other downloading tools through proxy fun
 	SilenceUsage:       true,
 	FParseErrWhitelist: cobra.FParseErrWhitelist{UnknownFlags: true},
 	RunE: func(cmd *cobra.Command, args []string) error {
+		// Convert config
+		if err := cfg.Convert(); err != nil {
+			return err
+		}
+		// Validate config
+		if err := cfg.Validate(); err != nil {
+			return err
+		}
+
 		// Initialize daemon dfpath
 		d, err := initDaemonDfpath(cfg)
 		if err != nil {
@@ -67,16 +76,6 @@ it supports container engine, wget and other downloading tools through proxy fun
 		}
 		logger.RedirectStdoutAndStderr(cfg.Console, path.Join(d.LogDir(), "daemon"))
 
-		// Convert config
-		if err := cfg.Convert(); err != nil {
-			return err
-		}
-
-		// Validate config
-		if err := cfg.Validate(); err != nil {
-			return err
-		}
-
 		return runDaemon(d)
 	},
 }
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index e062910d..8c6a02d6 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -69,6 +69,15 @@ var rootCmd = &cobra.Command{
 	RunE: func(cmd *cobra.Command, args []string) error {
 		start := time.Now()
 
+		// Convert config
+		if err := dfgetConfig.Convert(args); err != nil {
+			return err
+		}
+		// Validate config
+		if err := dfgetConfig.Validate(); err != nil {
+			return err
+		}
+
 		// Initialize daemon dfpath
 		d, err := initDfgetDfpath(dfgetConfig)
 		if err != nil {
@@ -83,16 +92,6 @@ var rootCmd = &cobra.Command{
 		// update plugin directory
 		source.UpdatePluginDir(d.PluginDir())
 
-		// Convert config
-		if err := dfgetConfig.Convert(args); err != nil {
-			return err
-		}
-
-		// Validate config
-		if err := dfgetConfig.Validate(); err != nil {
-			return err
-		}
-
 		fmt.Printf("--%s--  %s\n", start.Format("2006-01-02 15:04:05"), dfgetConfig.URL)
 		fmt.Printf("dfget version: %s\n", version.GitVersion)
 		fmt.Printf("current user: %s, default peer ip: %s\n", basic.Username, ip.IPv4)
diff --git a/cmd/manager/cmd/root.go b/cmd/manager/cmd/root.go
index 5aced092..91825d18 100644
--- a/cmd/manager/cmd/root.go
+++ b/cmd/manager/cmd/root.go
@@ -46,6 +46,11 @@ for managing schedulers and seed peers, offering http apis and portal, etc.`,
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	RunE: func(cmd *cobra.Command, args []string) error {
+		// Validate config
+		if err := cfg.Validate(); err != nil {
+			return err
+		}
+
 		// Initialize dfpath
 		d, err := initDfpath(cfg.Server)
 		if err != nil {
@@ -58,11 +63,6 @@ for managing schedulers and seed peers, offering http apis and portal, etc.`,
 		}
 		logger.RedirectStdoutAndStderr(cfg.Console, path.Join(d.LogDir(), "manager"))
 
-		// Validate config
-		if err := cfg.Validate(); err != nil {
-			return err
-		}
-
 		return runManager(d)
 	},
 }
diff --git a/cmd/scheduler/cmd/root.go b/cmd/scheduler/cmd/root.go
index e6a0c970..6ddaa390 100644
--- a/cmd/scheduler/cmd/root.go
+++ b/cmd/scheduler/cmd/root.go
@@ -47,6 +47,11 @@ generate and maintain a P2P network during the download process, and push suitab
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	RunE: func(cmd *cobra.Command, args []string) error {
+		// Validate config
+		if err := cfg.Validate(); err != nil {
+			return err
+		}
+
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
@@ -62,11 +67,6 @@ generate and maintain a P2P network during the download process, and push suitab
 		}
 		logger.RedirectStdoutAndStderr(cfg.Console, path.Join(d.LogDir(), "scheduler"))
 
-		// Validate config
-		if err := cfg.Validate(); err != nil {
-			return err
-		}
-
 		return runScheduler(ctx, d)
 	},
 }
diff --git a/pkg/dag/dag.go b/pkg/dag/dag.go
new file mode 100644
index 00000000..96abd37e
--- /dev/null
+++ b/pkg/dag/dag.go
@@ -0,0 +1,249 @@
+/*
+ *     Copyright 2022 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.
+ */
+
+//go:generate mockgen -destination mocks/dag_mock.go -source dag.go -package mocks
+
+package dag
+
+import (
+	"errors"
+	"sync"
+)
+
+var (
+	// ErrVertexNotFound represents vertex not found.
+	ErrVertexNotFound = errors.New("vertex not found")
+
+	// ErrVertexAlreadyExists represents vertex already exists.
+	ErrVertexAlreadyExists = errors.New("vertex already exists")
+
+	// ErrParnetAlreadyExists represents parent of vertex already exists.
+	ErrParnetAlreadyExists = errors.New("parent of vertex already exists")
+
+	// ErrChildAlreadyExists represents child of vertex already exists.
+	ErrChildAlreadyExists = errors.New("child of vertex already exists")
+
+	// ErrCycleBetweenVertices represents cycle between vertices.
+	ErrCycleBetweenVertices = errors.New("cycle between vertices")
+)
+
+// DAG is the interface used for directed acyclic graph.
+type DAG interface {
+	// AddVertex adds vertex to graph.
+	AddVertex(id string, value any) error
+
+	// DeleteVertex deletes vertex graph.
+	DeleteVertex(id string)
+
+	// GetVertex gets vertex from graph.
+	GetVertex(id string) (*Vertex, error)
+
+	// LenVertex returns length of vertices.
+	LenVertex() int
+
+	// RangeVertex calls f sequentially for each key and value present in the vertices.
+	// If f returns false, range stops the iteration.
+	RangeVertex(fn func(key string, value *Vertex) bool)
+
+	// AddEdge adds edge between two vertices.
+	AddEdge(fromVertexID, toVertexID string) error
+
+	// DeleteEdge deletes edge between two vertices.
+	DeleteEdge(fromVertexID, toVertexID string) error
+}
+
+// dag provides directed acyclic graph function.
+type dag struct {
+	mu       sync.RWMutex
+	vertices map[string]*Vertex
+}
+
+// New returns a new DAG interface.
+func NewDAG() DAG {
+	return &dag{
+		vertices: make(map[string]*Vertex),
+	}
+}
+
+// AddVertex adds vertex to graph.
+func (d *dag) AddVertex(id string, value any) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if _, ok := d.vertices[id]; ok {
+		return ErrVertexAlreadyExists
+	}
+
+	d.vertices[id] = NewVertex(id, value)
+	return nil
+}
+
+// DeleteVertex deletes vertex graph.
+func (d *dag) DeleteVertex(id string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	vertex, ok := d.vertices[id]
+	if !ok {
+		return
+	}
+
+	vertex.Parents.Range(func(item any) bool {
+		parent, ok := item.(*Vertex)
+		if !ok {
+			return true
+		}
+
+		parent.Children.Delete(vertex)
+		return true
+	})
+
+	vertex.Children.Range(func(item any) bool {
+		child, ok := item.(*Vertex)
+		if !ok {
+			return true
+		}
+
+		child.Parents.Delete(vertex)
+		return true
+	})
+
+	delete(d.vertices, id)
+}
+
+// GetVertex gets vertex from graph.
+func (d *dag) GetVertex(id string) (*Vertex, error) {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+
+	vertex, ok := d.vertices[id]
+	if !ok {
+		return nil, ErrVertexNotFound
+	}
+
+	return vertex, nil
+}
+
+// LenVertex returns length of vertices.
+func (d *dag) LenVertex() int {
+	return len(d.vertices)
+}
+
+// RangeVertex calls f sequentially for each key and value present in the vertices.
+// If f returns false, range stops the iteration.
+func (d *dag) RangeVertex(fn func(key string, value *Vertex) bool) {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+
+	for k, v := range d.vertices {
+		if !fn(k, v) {
+			break
+		}
+	}
+}
+
+// AddEdge adds edge between two vertices.
+func (d *dag) AddEdge(fromVertexID, toVertexID string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if fromVertexID == toVertexID {
+		return ErrCycleBetweenVertices
+	}
+
+	fromVertex, ok := d.vertices[fromVertexID]
+	if !ok {
+		return ErrVertexNotFound
+	}
+
+	toVertex, ok := d.vertices[toVertexID]
+	if !ok {
+		return ErrVertexNotFound
+	}
+
+	for _, child := range fromVertex.Children.Values() {
+		vertex, ok := child.(*Vertex)
+		if !ok {
+			continue
+		}
+
+		if vertex.ID == toVertexID {
+			return ErrCycleBetweenVertices
+		}
+	}
+
+	if d.depthFirstSearch(toVertexID, fromVertexID) {
+		return ErrCycleBetweenVertices
+	}
+
+	if ok := fromVertex.Children.Add(toVertex); !ok {
+		return ErrChildAlreadyExists
+	}
+
+	if ok := toVertex.Parents.Add(fromVertex); !ok {
+		return ErrParnetAlreadyExists
+	}
+
+	return nil
+}
+
+// DeleteEdge deletes edge between two vertices.
+func (d *dag) DeleteEdge(fromVertexID, toVertexID string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	fromVertex, ok := d.vertices[fromVertexID]
+	if !ok {
+		return ErrVertexNotFound
+	}
+
+	toVertex, ok := d.vertices[toVertexID]
+	if !ok {
+		return ErrVertexNotFound
+	}
+
+	fromVertex.Children.Delete(toVertex)
+	toVertex.Parents.Delete(fromVertex)
+	return nil
+}
+
+// depthFirstSearch is a depth-first search of the directed acyclic graph.
+func (d *dag) depthFirstSearch(fromVertexID, toVertexID string) bool {
+	successors := make(map[string]struct{})
+	d.search(fromVertexID, successors)
+	_, ok := successors[toVertexID]
+	return ok
+}
+
+// depthFirstSearch finds successors of vertex.
+func (d *dag) search(vertexID string, successors map[string]struct{}) {
+	vertex, ok := d.vertices[vertexID]
+	if !ok {
+		return
+	}
+
+	for _, child := range vertex.Children.Values() {
+		vertex, ok := child.(*Vertex)
+		if !ok {
+			continue
+		}
+
+		if _, ok := successors[vertex.ID]; !ok {
+			successors[vertex.ID] = struct{}{}
+			d.search(vertex.ID, successors)
+		}
+	}
+}
diff --git a/pkg/dag/mocks/dag_mock.go b/pkg/dag/mocks/dag_mock.go
new file mode 100644
index 00000000..9cd3c2fd
--- /dev/null
+++ b/pkg/dag/mocks/dag_mock.go
@@ -0,0 +1,130 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: dag.go
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	reflect "reflect"
+
+	dag "d7y.io/dragonfly/v2/pkg/dag"
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockDAG is a mock of DAG interface.
+type MockDAG struct {
+	ctrl     *gomock.Controller
+	recorder *MockDAGMockRecorder
+}
+
+// MockDAGMockRecorder is the mock recorder for MockDAG.
+type MockDAGMockRecorder struct {
+	mock *MockDAG
+}
+
+// NewMockDAG creates a new mock instance.
+func NewMockDAG(ctrl *gomock.Controller) *MockDAG {
+	mock := &MockDAG{ctrl: ctrl}
+	mock.recorder = &MockDAGMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockDAG) EXPECT() *MockDAGMockRecorder {
+	return m.recorder
+}
+
+// AddEdge mocks base method.
+func (m *MockDAG) AddEdge(fromVertexID, toVertexID string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AddEdge", fromVertexID, toVertexID)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AddEdge indicates an expected call of AddEdge.
+func (mr *MockDAGMockRecorder) AddEdge(fromVertexID, toVertexID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddEdge", reflect.TypeOf((*MockDAG)(nil).AddEdge), fromVertexID, toVertexID)
+}
+
+// AddVertex mocks base method.
+func (m *MockDAG) AddVertex(id string, value any) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AddVertex", id, value)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AddVertex indicates an expected call of AddVertex.
+func (mr *MockDAGMockRecorder) AddVertex(id, value interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddVertex", reflect.TypeOf((*MockDAG)(nil).AddVertex), id, value)
+}
+
+// DeleteEdge mocks base method.
+func (m *MockDAG) DeleteEdge(fromVertexID, toVertexID string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteEdge", fromVertexID, toVertexID)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteEdge indicates an expected call of DeleteEdge.
+func (mr *MockDAGMockRecorder) DeleteEdge(fromVertexID, toVertexID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteEdge", reflect.TypeOf((*MockDAG)(nil).DeleteEdge), fromVertexID, toVertexID)
+}
+
+// DeleteVertex mocks base method.
+func (m *MockDAG) DeleteVertex(id string) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "DeleteVertex", id)
+}
+
+// DeleteVertex indicates an expected call of DeleteVertex.
+func (mr *MockDAGMockRecorder) DeleteVertex(id interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteVertex", reflect.TypeOf((*MockDAG)(nil).DeleteVertex), id)
+}
+
+// GetVertex mocks base method.
+func (m *MockDAG) GetVertex(id string) (*dag.Vertex, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetVertex", id)
+	ret0, _ := ret[0].(*dag.Vertex)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetVertex indicates an expected call of GetVertex.
+func (mr *MockDAGMockRecorder) GetVertex(id interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetVertex", reflect.TypeOf((*MockDAG)(nil).GetVertex), id)
+}
+
+// LenVertex mocks base method.
+func (m *MockDAG) LenVertex() int {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "LenVertex")
+	ret0, _ := ret[0].(int)
+	return ret0
+}
+
+// LenVertex indicates an expected call of LenVertex.
+func (mr *MockDAGMockRecorder) LenVertex() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LenVertex", reflect.TypeOf((*MockDAG)(nil).LenVertex))
+}
+
+// RangeVertex mocks base method.
+func (m *MockDAG) RangeVertex(fn func(string, *dag.Vertex) bool) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "RangeVertex", fn)
+}
+
+// RangeVertex indicates an expected call of RangeVertex.
+func (mr *MockDAGMockRecorder) RangeVertex(fn interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RangeVertex", reflect.TypeOf((*MockDAG)(nil).RangeVertex), fn)
+}
diff --git a/pkg/dag/vertex.go b/pkg/dag/vertex.go
new file mode 100644
index 00000000..99ff1442
--- /dev/null
+++ b/pkg/dag/vertex.go
@@ -0,0 +1,52 @@
+/*
+ *     Copyright 2022 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 dag
+
+import "d7y.io/dragonfly/v2/pkg/container/set"
+
+// Vertex is a vertex of the directed acyclic graph.
+type Vertex struct {
+	ID       string
+	Value    any
+	Parents  set.SafeSet
+	Children set.SafeSet
+}
+
+// New returns a new Vertex instance.
+func NewVertex(id string, value any) *Vertex {
+	return &Vertex{
+		ID:       id,
+		Value:    value,
+		Parents:  set.NewSafeSet(),
+		Children: set.NewSafeSet(),
+	}
+}
+
+// Degree returns the degree of vertex.
+func (v *Vertex) Degree() int {
+	return int(v.Parents.Len() + v.Children.Len())
+}
+
+// InDegree returns the indegree of vertex.
+func (v *Vertex) InDegree() int {
+	return int(v.Parents.Len())
+}
+
+// OutDegree returns the outdegree of vertex.
+func (v *Vertex) OutDegree() int {
+	return int(v.Children.Len())
+}
diff --git a/pkg/rpc/scheduler/mocks/scheduler_mock.go b/pkg/rpc/scheduler/mocks/scheduler_mock.go
index 88baadb1..1b89941d 100644
--- a/pkg/rpc/scheduler/mocks/scheduler_mock.go
+++ b/pkg/rpc/scheduler/mocks/scheduler_mock.go
@@ -50,6 +50,76 @@ func (mr *MockisRegisterResult_DirectPieceMockRecorder) isRegisterResult_DirectP
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "isRegisterResult_DirectPiece", reflect.TypeOf((*MockisRegisterResult_DirectPiece)(nil).isRegisterResult_DirectPiece))
 }
 
+// MockisPeerPacket_ErrorDetail is a mock of isPeerPacket_ErrorDetail interface.
+type MockisPeerPacket_ErrorDetail struct {
+	ctrl     *gomock.Controller
+	recorder *MockisPeerPacket_ErrorDetailMockRecorder
+}
+
+// MockisPeerPacket_ErrorDetailMockRecorder is the mock recorder for MockisPeerPacket_ErrorDetail.
+type MockisPeerPacket_ErrorDetailMockRecorder struct {
+	mock *MockisPeerPacket_ErrorDetail
+}
+
+// NewMockisPeerPacket_ErrorDetail creates a new mock instance.
+func NewMockisPeerPacket_ErrorDetail(ctrl *gomock.Controller) *MockisPeerPacket_ErrorDetail {
+	mock := &MockisPeerPacket_ErrorDetail{ctrl: ctrl}
+	mock.recorder = &MockisPeerPacket_ErrorDetailMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockisPeerPacket_ErrorDetail) EXPECT() *MockisPeerPacket_ErrorDetailMockRecorder {
+	return m.recorder
+}
+
+// isPeerPacket_ErrorDetail mocks base method.
+func (m *MockisPeerPacket_ErrorDetail) isPeerPacket_ErrorDetail() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "isPeerPacket_ErrorDetail")
+}
+
+// isPeerPacket_ErrorDetail indicates an expected call of isPeerPacket_ErrorDetail.
+func (mr *MockisPeerPacket_ErrorDetailMockRecorder) isPeerPacket_ErrorDetail() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "isPeerPacket_ErrorDetail", reflect.TypeOf((*MockisPeerPacket_ErrorDetail)(nil).isPeerPacket_ErrorDetail))
+}
+
+// MockisPeerResult_ErrorDetail is a mock of isPeerResult_ErrorDetail interface.
+type MockisPeerResult_ErrorDetail struct {
+	ctrl     *gomock.Controller
+	recorder *MockisPeerResult_ErrorDetailMockRecorder
+}
+
+// MockisPeerResult_ErrorDetailMockRecorder is the mock recorder for MockisPeerResult_ErrorDetail.
+type MockisPeerResult_ErrorDetailMockRecorder struct {
+	mock *MockisPeerResult_ErrorDetail
+}
+
+// NewMockisPeerResult_ErrorDetail creates a new mock instance.
+func NewMockisPeerResult_ErrorDetail(ctrl *gomock.Controller) *MockisPeerResult_ErrorDetail {
+	mock := &MockisPeerResult_ErrorDetail{ctrl: ctrl}
+	mock.recorder = &MockisPeerResult_ErrorDetailMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockisPeerResult_ErrorDetail) EXPECT() *MockisPeerResult_ErrorDetailMockRecorder {
+	return m.recorder
+}
+
+// isPeerResult_ErrorDetail mocks base method.
+func (m *MockisPeerResult_ErrorDetail) isPeerResult_ErrorDetail() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "isPeerResult_ErrorDetail")
+}
+
+// isPeerResult_ErrorDetail indicates an expected call of isPeerResult_ErrorDetail.
+func (mr *MockisPeerResult_ErrorDetailMockRecorder) isPeerResult_ErrorDetail() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "isPeerResult_ErrorDetail", reflect.TypeOf((*MockisPeerResult_ErrorDetail)(nil).isPeerResult_ErrorDetail))
+}
+
 // MockSchedulerClient is a mock of SchedulerClient interface.
 type MockSchedulerClient struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/slices/slices.go b/pkg/slices/slices.go
index f744da7e..47488c59 100644
--- a/pkg/slices/slices.go
+++ b/pkg/slices/slices.go
@@ -26,3 +26,18 @@ func Contains[T comparable](s []T, e T) bool {
 
 	return false
 }
+
+// FindDuplicate returns duplicate element in a collection.
+func FindDuplicate[T comparable](s []T) (T, bool) {
+	visited := make(map[T]struct{})
+	for _, v := range s {
+		if _, ok := visited[v]; ok {
+			return v, true
+		}
+
+		visited[v] = struct{}{}
+	}
+
+	var zero T
+	return zero, false
+}
diff --git a/pkg/source/mocks/mock_source_client.go b/pkg/source/mocks/mock_source_client.go
index 6cddad6c..b2a7d888 100644
--- a/pkg/source/mocks/mock_source_client.go
+++ b/pkg/source/mocks/mock_source_client.go
@@ -5,7 +5,6 @@
 package mocks
 
 import (
-	url "net/url"
 	reflect "reflect"
 
 	source "d7y.io/dragonfly/v2/pkg/source"
@@ -172,10 +171,10 @@ func (m *MockResourceLister) EXPECT() *MockResourceListerMockRecorder {
 }
 
 // List mocks base method.
-func (m *MockResourceLister) List(request *source.Request) ([]*url.URL, error) {
+func (m *MockResourceLister) List(request *source.Request) ([]source.URLEntry, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "List", request)
-	ret0, _ := ret[0].([]*url.URL)
+	ret0, _ := ret[0].([]source.URLEntry)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index dfa04716..32ec3887 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -33,6 +33,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/container/set"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/slices"
 )
 
 const (
@@ -373,11 +374,11 @@ func (p *Peer) Depth() int {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	node := p
-	var depth int
+	var (
+		node      = p
+		ancestors = []string{p.ID}
+	)
 	for node != nil {
-		depth++
-
 		if node.Host.Type != HostTypeNormal {
 			break
 		}
@@ -387,8 +388,10 @@ func (p *Peer) Depth() int {
 			break
 		}
 
+		ancestors = append(ancestors, parent.ID)
+
 		// Prevent traversal tree from infinite loop.
-		if p.ID == parent.ID {
+		if _, found := slices.FindDuplicate(ancestors); found {
 			p.Log.Error("tree structure produces an infinite loop")
 			break
 		}
@@ -396,7 +399,7 @@ func (p *Peer) Depth() int {
 		node = parent
 	}
 
-	return depth
+	return len(ancestors)
 }
 
 // Ancestors returns peer's ancestors.
@@ -404,22 +407,25 @@ func (p *Peer) Ancestors() []string {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	var ancestors []string
-	node := p
+	var (
+		node      = p
+		ancestors = []string{p.ID}
+	)
 	for node != nil {
 		parent, ok := node.LoadParent()
 		if !ok {
 			return ancestors
 		}
 
+		ancestors = append(ancestors, parent.ID)
+
 		// Prevent traversal tree from infinite loop.
-		if parent.ID == node.ID {
+		if _, found := slices.FindDuplicate(ancestors); found {
 			p.Log.Error("tree structure produces an infinite loop")
-			return ancestors
+			break
 		}
 
 		node = parent
-		ancestors = append(ancestors, node.ID)
 	}
 
 	return ancestors
@@ -427,34 +433,39 @@ func (p *Peer) Ancestors() []string {
 
 // IsDescendant determines whether it is ancestor of peer.
 func (p *Peer) IsDescendant(ancestor *Peer) bool {
-	return p.isDescendant(ancestor, p)
+	return ancestor.isDescendant(p)
 }
 
 // IsAncestor determines whether it is descendant of peer.
 func (p *Peer) IsAncestor(descendant *Peer) bool {
-	return p.isDescendant(p, descendant)
+	return p.isDescendant(descendant)
 }
 
 // isDescendant determines whether it is ancestor of peer.
-func (p *Peer) isDescendant(ancestor, descendant *Peer) bool {
+func (p *Peer) isDescendant(descendant *Peer) bool {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	node := descendant
+	var (
+		node      = descendant
+		ancestors = []string{descendant.ID}
+	)
 	for node != nil {
 		parent, ok := node.LoadParent()
 		if !ok {
 			return false
 		}
 
-		if parent.ID == ancestor.ID {
+		if parent.ID == p.ID {
 			return true
 		}
 
+		ancestors = append(ancestors, parent.ID)
+
 		// Prevent traversal tree from infinite loop.
-		if parent.ID == descendant.ID {
+		if _, found := slices.FindDuplicate(ancestors); found {
 			p.Log.Error("tree structure produces an infinite loop")
-			return true
+			break
 		}
 
 		node = parent


Test patch:
diff --git a/pkg/dag/dag_test.go b/pkg/dag/dag_test.go
new file mode 100644
index 00000000..9590e6c9
--- /dev/null
+++ b/pkg/dag/dag_test.go
@@ -0,0 +1,649 @@
+/*
+ *     Copyright 2022 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 dag
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewDAG(t *testing.T) {
+	d := NewDAG()
+	assert := assert.New(t)
+	assert.Equal(reflect.TypeOf(d).Elem().Name(), "dag")
+}
+
+func TestDAGAddVertex(t *testing.T) {
+	tests := []struct {
+		name   string
+		id     string
+		value  any
+		expect func(t *testing.T, d DAG, err error)
+	}{
+		{
+			name:  "add vertex",
+			id:    mockVertexID,
+			value: mockVertexValue,
+			expect: func(t *testing.T, d DAG, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+			},
+		},
+		{
+			name:  "vertex already exists",
+			id:    mockVertexID,
+			value: mockVertexValue,
+			expect: func(t *testing.T, d DAG, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+
+				assert.EqualError(d.AddVertex(mockVertexID, mockVertexValue), ErrVertexAlreadyExists.Error())
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d, d.AddVertex(tc.id, tc.name))
+		})
+	}
+}
+
+func TestDAGDeleteVertex(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "delete vertex",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				if err := d.AddVertex(mockVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+				d.DeleteVertex(mockVertexID)
+
+				_, err := d.GetVertex(mockVertexID)
+				assert.EqualError(err, ErrVertexNotFound.Error())
+			},
+		},
+		{
+			name: "delete vertex with edges",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+
+				var (
+					mockToVertexID = "baz"
+				)
+				if err := d.AddVertex(mockToVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexID, mockToVertexID); err != nil {
+					assert.NoError(err)
+				}
+
+				d.DeleteVertex(mockVertexID)
+
+				_, err := d.GetVertex(mockVertexID)
+				assert.EqualError(err, ErrVertexNotFound.Error())
+
+				vertex, err := d.GetVertex(mockToVertexID)
+				assert.NoError(err)
+				assert.Equal(vertex.Parents.Len(), uint(0))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func TestDAGGetVertex(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "get vertex",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				if err := d.AddVertex(mockVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				vertex, err := d.GetVertex(mockVertexID)
+				assert.NoError(err)
+				assert.Equal(vertex.ID, mockVertexID)
+				assert.Equal(vertex.Value, mockVertexValue)
+				assert.Equal(vertex.Parents.Len(), uint(0))
+				assert.Equal(vertex.Children.Len(), uint(0))
+			},
+		},
+		{
+			name: "vertex not found",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				_, err := d.GetVertex(mockVertexID)
+				assert.EqualError(err, ErrVertexNotFound.Error())
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func TestDAGVertexLenVertex(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "get length of vertex",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				if err := d.AddVertex(mockVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				d.LenVertex()
+				assert.Equal(d.LenVertex(), 1)
+
+				d.DeleteVertex(mockVertexID)
+				assert.Equal(d.LenVertex(), 0)
+			},
+		},
+		{
+			name: "empty dag",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				assert.Equal(d.LenVertex(), 0)
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func TestDAGRange(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "range vertices",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				if err := d.AddVertex(mockVertexID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				d.RangeVertex(func(k string, v *Vertex) bool {
+					assert.Equal(k, mockVertexID)
+					return true
+				})
+			},
+		},
+		{
+			name: "range failed",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+				)
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexFID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				d.RangeVertex(func(k string, v *Vertex) bool {
+					assert.Contains([]string{mockVertexEID, mockVertexFID}, k)
+					return false
+				})
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func TestDAGAddEdge(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "add edge",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+					mockVertexGID = "bag"
+					mockVertexHID = "bah"
+					mockVertexIID = "bai"
+				)
+
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexFID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexGID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexHID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexIID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexGID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexHID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexGID, mockVertexIID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexIID, mockVertexHID); err != nil {
+					assert.NoError(err)
+				}
+			},
+		},
+		{
+			name: "cycle between vertices",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+					mockVertexGID = "bag"
+					mockVertexHID = "bah"
+					mockVertexIID = "bai"
+				)
+
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexFID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexGID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexHID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexIID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrCycleBetweenVertices.Error())
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrCycleBetweenVertices.Error())
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexGID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexGID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrCycleBetweenVertices.Error())
+				}
+
+				if err := d.AddEdge(mockVertexGID, mockVertexHID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexHID, mockVertexIID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexIID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrCycleBetweenVertices.Error())
+				}
+			},
+		},
+		{
+			name: "vertex not found",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+				)
+
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.EqualError(err, ErrVertexNotFound.Error())
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrVertexNotFound.Error())
+				}
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func TestDAGDeleteEdge(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, d DAG)
+	}{
+		{
+			name: "delete edge",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+				)
+
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddVertex(mockVertexFID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.DeleteEdge(mockVertexFID, mockVertexEID); err != nil {
+					assert.NoError(err)
+				}
+
+				vf, err := d.GetVertex(mockVertexFID)
+				assert.NoError(err)
+				assert.Equal(vf.Parents.Len(), uint(1))
+				ve, err := d.GetVertex(mockVertexEID)
+				assert.NoError(err)
+				assert.Equal(ve.Children.Len(), uint(1))
+
+				if err := d.DeleteEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.NoError(err)
+				}
+
+				vf, err = d.GetVertex(mockVertexFID)
+				assert.NoError(err)
+				assert.Equal(vf.Parents.Len(), uint(0))
+				ve, err = d.GetVertex(mockVertexEID)
+				assert.NoError(err)
+				assert.Equal(ve.Children.Len(), uint(0))
+			},
+		},
+		{
+			name: "vertex not found",
+			expect: func(t *testing.T, d DAG) {
+				assert := assert.New(t)
+				var (
+					mockVertexEID = "bae"
+					mockVertexFID = "baf"
+				)
+
+				if err := d.AddVertex(mockVertexEID, mockVertexValue); err != nil {
+					assert.NoError(err)
+				}
+
+				if err := d.AddEdge(mockVertexEID, mockVertexFID); err != nil {
+					assert.EqualError(err, ErrVertexNotFound.Error())
+				}
+
+				if err := d.AddEdge(mockVertexFID, mockVertexEID); err != nil {
+					assert.EqualError(err, ErrVertexNotFound.Error())
+				}
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := NewDAG()
+			tc.expect(t, d)
+		})
+	}
+}
+
+func BenchmarkDAGAddVertex(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		ids = append(ids, fmt.Sprint(n))
+	}
+
+	b.ResetTimer()
+	for _, id := range ids {
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkDAGDeleteVertex(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		id := fmt.Sprint(n)
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+
+		ids = append(ids, id)
+	}
+
+	b.ResetTimer()
+	for _, id := range ids {
+		d.DeleteVertex(id)
+	}
+}
+
+func BenchmarkDAGDeleteVertexWithMultiEdges(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		id := fmt.Sprint(n)
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+
+		ids = append(ids, id)
+	}
+
+	edgeCount := 5
+	for index, id := range ids {
+		if index+edgeCount > len(ids)-1 {
+			break
+		}
+
+		for n := 1; n < edgeCount; n++ {
+			if err := d.AddEdge(id, ids[index+n]); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+
+	b.ResetTimer()
+	for _, id := range ids {
+		d.DeleteVertex(id)
+	}
+}
+
+func BenchmarkDAGAddEdge(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		id := fmt.Sprint(n)
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+
+		ids = append(ids, id)
+	}
+
+	b.ResetTimer()
+	for index, id := range ids {
+		if index < 1 {
+			continue
+		}
+
+		if err := d.AddEdge(id, ids[index-1]); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkDAGAddEdgeWithMultiEdges(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		id := fmt.Sprint(n)
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+
+		ids = append(ids, id)
+	}
+
+	edgeCount := 5
+	for index, id := range ids {
+		if index+edgeCount > len(ids)-1 {
+			break
+		}
+
+		for n := 1; n < edgeCount; n++ {
+			if err := d.AddEdge(id, ids[index+n]); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+
+	b.ResetTimer()
+	for index, id := range ids {
+		if index+edgeCount+1 > len(ids)-1 {
+			break
+		}
+
+		if err := d.AddEdge(id, ids[index+edgeCount+1]); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkDAGDeleteEdge(b *testing.B) {
+	var ids []string
+	d := NewDAG()
+	for n := 0; n < b.N; n++ {
+		id := fmt.Sprint(n)
+		if err := d.AddVertex(id, nil); err != nil {
+			b.Fatal(err)
+		}
+
+		ids = append(ids, id)
+	}
+
+	for index, id := range ids {
+		if index < 1 {
+			continue
+		}
+
+		if err := d.AddEdge(id, ids[index-1]); err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	b.ResetTimer()
+	for index, id := range ids {
+		if index < 1 {
+			continue
+		}
+
+		if err := d.DeleteEdge(id, ids[index-1]); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
diff --git a/pkg/dag/vertex_test.go b/pkg/dag/vertex_test.go
new file mode 100644
index 00000000..21b1a10c
--- /dev/null
+++ b/pkg/dag/vertex_test.go
@@ -0,0 +1,97 @@
+/*
+ *     Copyright 2022 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 dag
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	mockVertexID    = "foo"
+	mockVertexValue = "var"
+)
+
+func TestNewVertex(t *testing.T) {
+	v := NewVertex(mockVertexID, mockVertexValue)
+	assert := assert.New(t)
+	assert.Equal(v.ID, mockVertexID)
+	assert.Equal(v.Value, mockVertexValue)
+	assert.Equal(v.Parents.Len(), uint(0))
+	assert.Equal(v.Children.Len(), uint(0))
+}
+
+func TestVertexDegree(t *testing.T) {
+	v := NewVertex(mockVertexID, mockVertexValue)
+	assert := assert.New(t)
+	assert.Equal(v.ID, mockVertexID)
+	assert.Equal(v.Value, mockVertexValue)
+	assert.Equal(v.Degree(), 0)
+
+	v.Parents.Add(mockVertexID)
+	assert.Equal(v.Degree(), 1)
+
+	v.Children.Add(mockVertexID)
+	assert.Equal(v.Degree(), 2)
+
+	v.Parents.Delete(mockVertexID)
+	assert.Equal(v.Degree(), 1)
+
+	v.Children.Delete(mockVertexID)
+	assert.Equal(v.Degree(), 0)
+}
+
+func TestVertexInDegree(t *testing.T) {
+	v := NewVertex(mockVertexID, mockVertexValue)
+	assert := assert.New(t)
+	assert.Equal(v.ID, mockVertexID)
+	assert.Equal(v.Value, mockVertexValue)
+	assert.Equal(v.InDegree(), 0)
+
+	v.Parents.Add(mockVertexID)
+	assert.Equal(v.InDegree(), 1)
+
+	v.Children.Add(mockVertexID)
+	assert.Equal(v.InDegree(), 1)
+
+	v.Parents.Delete(mockVertexID)
+	assert.Equal(v.InDegree(), 0)
+
+	v.Children.Delete(mockVertexID)
+	assert.Equal(v.InDegree(), 0)
+}
+
+func TestVertexOutDegree(t *testing.T) {
+	v := NewVertex(mockVertexID, mockVertexValue)
+	assert := assert.New(t)
+	assert.Equal(v.ID, mockVertexID)
+	assert.Equal(v.Value, mockVertexValue)
+	assert.Equal(v.OutDegree(), 0)
+
+	v.Parents.Add(mockVertexID)
+	assert.Equal(v.OutDegree(), 0)
+
+	v.Children.Add(mockVertexID)
+	assert.Equal(v.OutDegree(), 1)
+
+	v.Parents.Delete(mockVertexID)
+	assert.Equal(v.OutDegree(), 1)
+
+	v.Children.Delete(mockVertexID)
+	assert.Equal(v.OutDegree(), 0)
+}
diff --git a/pkg/slices/slices_test.go b/pkg/slices/slices_test.go
index 9aefd352..7a0751a0 100644
--- a/pkg/slices/slices_test.go
+++ b/pkg/slices/slices_test.go
@@ -29,3 +29,25 @@ func TestContains(t *testing.T) {
 	assert.True(Contains([]string{"a", "b", "c"}, "a"))
 	assert.False(Contains([]string{"a", "b", "c"}, "d"))
 }
+
+func TestFindDuplicate(t *testing.T) {
+	assert := assert.New(t)
+	var (
+		dupi  int
+		dups  string
+		found bool
+	)
+	dupi, found = FindDuplicate([]int{1, 2, 1, 3})
+	assert.True(found)
+	assert.Equal(dupi, 1)
+
+	_, found = FindDuplicate([]int{1, 2, 3})
+	assert.False(found)
+
+	dups, found = FindDuplicate([]string{"a", "b", "c", "b"})
+	assert.True(found)
+	assert.Equal(dups, "b")
+
+	_, found = FindDuplicate([]string{"a", "b", "c"})
+	assert.False(found)
+}
diff --git a/scheduler/resource/peer_test.go b/scheduler/resource/peer_test.go
index 964eb145..9d8e2b0e 100644
--- a/scheduler/resource/peer_test.go
+++ b/scheduler/resource/peer_test.go
@@ -526,7 +526,7 @@ func TestPeer_Depth(t *testing.T) {
 				peer.StoreParent(peer)
 
 				assert := assert.New(t)
-				assert.Equal(peer.Depth(), 1)
+				assert.Equal(peer.Depth(), 2)
 			},
 		},
 	}
@@ -557,8 +557,8 @@ func TestPeer_Ancestors(t *testing.T) {
 			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
 				assert := assert.New(t)
 				peer.StoreChild(mockChildPeer)
-				assert.Equal(len(mockChildPeer.Ancestors()), 1)
-				assert.EqualValues(mockChildPeer.Ancestors(), []string{peer.ID})
+				assert.Equal(len(mockChildPeer.Ancestors()), 2)
+				assert.EqualValues(mockChildPeer.Ancestors(), []string{mockChildPeer.ID, peer.ID})
 			},
 		},
 		{
@@ -566,8 +566,8 @@ func TestPeer_Ancestors(t *testing.T) {
 			childID: idgen.PeerID("127.0.0.1"),
 			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
 				assert := assert.New(t)
-				assert.Equal(len(mockChildPeer.Ancestors()), 0)
-				assert.EqualValues(mockChildPeer.Ancestors(), []string(nil))
+				assert.Equal(len(mockChildPeer.Ancestors()), 1)
+				assert.EqualValues(mockChildPeer.Ancestors(), []string{mockChildPeer.ID})
 			},
 		},
 		{
@@ -576,8 +576,8 @@ func TestPeer_Ancestors(t *testing.T) {
 			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
 				assert := assert.New(t)
 				peer.StoreChild(peer)
-				assert.Equal(len(peer.Ancestors()), 0)
-				assert.Equal(peer.Ancestors(), []string(nil))
+				assert.Equal(len(peer.Ancestors()), 2)
+				assert.Equal(peer.Ancestors(), []string{peer.ID, peer.ID})
 			},
 		},
 	}