Feat/fetch p2p network info dfdaemon

apis/swagger.yml

@@ -1441,6 +1441,31 @@ definitions:
         items:
           $ref: "#/definitions/Node"
 
+  HeartBeatRequest:
+    type: "object"
+    description: ""
+    properties:
+      IP:
+        type: "string"
+        description: "IP address which peer client carries"
+        format: "string"
+      port:
+        type: "integer"
+        description: |
+          when registering, dfget will setup one uploader process.
+          This one acts as a server for peer pulling tasks.
+          This port is which this server listens on.
+        format: "int32"
+        minimum: 15000
+        maximum: 65000
+      cID:
+        type: "string"
+        description: |
+          CID means the client ID. It maps to the specific dfget process.
+          When user wishes to download an image/file, user would start a dfget process to do this.
+          This dfget is treated a client and carries a client ID.
+          Thus, multiple dfget processes on the same peer have different CIDs.
+
   ErrorResponse:
     type: "object"
     description: |

cmd/dfdaemon/app/root.go

@@ -18,6 +18,7 @@ package app
 
 import (
 	"encoding/json"
+	"github.com/dragonflyoss/Dragonfly/dfdaemon/localManager"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -68,6 +69,9 @@ var rootCmd = &cobra.Command{
 		cfgJSON, _ := json.Marshal(cfg)
 		logrus.Infof("using config: %s", cfgJSON)
 
+
+		_ = localManager.NewLocalManager(cfg.DFGetConfig())
+
 		s, err := dfdaemon.NewFromConfig(*cfg)
 		if err != nil {
 			return errors.Wrap(err, "create dfdaemon from config")

dfdaemon/downloader/p2p/dfclient.go

@@ -107,14 +107,14 @@ func (c *DFClient) doDownload(ctx context.Context, url string, header map[string
 	runtimeConfig.URL = url
 	runtimeConfig.RV.TaskURL = url
 	runtimeConfig.RV.TaskFileName = getTaskFileName(destPath, c.dfGetConfig.Sign)
-	runtimeConfig.Header = flattenHeader(header)
+	runtimeConfig.Header = FlattenHeader(header)
 	runtimeConfig.Output = destPath
 	runtimeConfig.RV.RealTarget = destPath
 	runtimeConfig.RV.TargetDir = filepath.Dir(destPath)
 
 	hr := http.Header(header)
 	if digestHeaderStr := hr.Get(dfgetcfg.StrDigest); digestHeaderStr != "" {
-		ds, err := getDigest(digestHeaderStr)
+		ds, err := GetDigestFromHeader(digestHeaderStr)
 		if err != nil {
 			return nil, err
 		}
@@ -139,7 +139,7 @@ func getCid(localIP string, sign string) string {
 	return localIP + "-" + sign
 }
 
-func flattenHeader(header map[string][]string) []string {
+func FlattenHeader(header map[string][]string) []string {
 	var res []string
 	for key, value := range header {
 		// discard HTTP host header for backing to source successfully
@@ -157,7 +157,7 @@ func flattenHeader(header map[string][]string) []string {
 	return res
 }
 
-func getDigest(digestHeaderStr string) ([]*DigestStruct, error) {
+func GetDigestFromHeader(digestHeaderStr string) ([]*DigestStruct, error) {
 	var (
 		digest   string
 		rangeStr string

dfdaemon/downloader/p2p/dfclient_test.go

@@ -63,7 +63,7 @@ func (tc *testcase) WithValue(key string, value ...string) *testcase {
 func (tc *testcase) Test(t *testing.T) {
 	a := assert.New(t)
 
-	actualResult := flattenHeader(tc.header)
+	actualResult := FlattenHeader(tc.header)
 
 	for _, r := range actualResult {
 		if _, ok := tc.result[r]; !ok {

dfdaemon/localManager/cache.go

@@ -0,0 +1,12 @@
+package localManager
+
+import "github.com/dragonflyoss/Dragonfly/pkg/syncmap"
+
+type cache struct {
+
+}
+
+type cacheManager struct {
+	// key is taskID, value is the
+	taskContainer	*syncmap.SyncMap
+}

dfdaemon/localManager/finite_queue.go

@@ -0,0 +1,148 @@
+package localManager
+
+import (
+	"github.com/dragonflyoss/Dragonfly/pkg/errortypes"
+	"sync"
+)
+
+// finiteQueue provides a circle queue with capacity, and it will weed out the earlier item if full
+type finiteQueue struct {
+	sync.Mutex
+	capacity  int
+	head 	  *itemNode
+	tail      *itemNode
+
+	itemMap   map[string]*itemNode
+}
+
+type itemNode struct {
+	prv  *itemNode
+	next *itemNode
+	data  interface{}
+	key   string
+}
+
+func newFiniteQueue(capacity int) *finiteQueue {
+	return &finiteQueue{
+		capacity: capacity,
+		head: nil,
+		tail: nil,
+		itemMap: make(map[string]*itemNode, capacity),
+	}
+}
+
+// put item to front
+func (q *finiteQueue) putFront(key string, data interface{}) {
+	q.Lock()
+	defer q.Unlock()
+
+	if i, ok := q.itemMap[key]; ok {
+		//todo: update data
+		i.data = data
+		q.internalPutFront(i)
+		return
+	}
+
+	if len(q.itemMap) >= q.capacity {
+		// remove the earliest item
+		i := q.internalRemoveTail()
+		if i != nil {
+			delete(q.itemMap, i.key)
+		}
+	}
+
+	node := &itemNode{
+		key: key,
+		data: data,
+	}
+
+	q.itemMap[key] = node
+	q.internalPutFront(node)
+}
+
+// getFront will get several item from front and not poll out them.
+func (q *finiteQueue) getFront(count int) []interface{} {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.head == nil {
+		return nil
+	}
+
+	result := make([]interface{}, count)
+	item := q.head
+	index := 0
+	for{
+		result[index] = item.data
+		index ++
+		if index >= count {
+			break
+		}
+		if item == q.tail {
+			break
+		}
+		item = item.next
+	}
+
+	return result[:index]
+}
+
+func (q *finiteQueue) getItemByKey(key string) (interface{}, error) {
+	q.Lock()
+	defer q.Unlock()
+
+	if data, exist := q.itemMap[key]; exist {
+		return data.data, nil
+	}
+
+	return nil, errortypes.ErrDataNotFound
+}
+
+func (q *finiteQueue) internalPutFront(i *itemNode) {
+	if q.head == i {
+		return
+	}
+
+	if q.head == nil {
+		q.head = i
+		q.tail = i
+		i.prv = i
+		i.next = i
+		return
+	}
+
+	if q.tail == i {
+		q.tail = i.prv
+	}
+
+	if i.prv != nil && i.next != nil {
+		i.prv.next = i.next
+		i.next.prv = i.prv
+	}
+
+	q.tail.next = i
+	q.head.prv = i
+	i.prv = q.tail
+	i.next = q.head
+
+	q.head = i
+}
+
+func (q *finiteQueue) internalRemoveTail() *itemNode {
+	if q.tail == nil {
+		return nil
+	}
+
+	if q.head == q.tail {
+		q.head = nil
+		q.tail = nil
+		return q.tail
+	}
+
+	result := q.tail
+	q.tail.prv.next = q.head
+	q.head.prv = q.tail.prv
+	q.tail = q.tail.prv
+
+	return result
+}