feat: pex support parent task

client/config/headers.go

@@ -37,4 +37,6 @@ const (
 	HeaderDragonflyObjectMetaStorageClass = "X-Dragonfly-Object-Meta-Storage-Class"
 	// HeaderDragonflyObjectOperation is used for object storage operation.
 	HeaderDragonflyObjectOperation = "X-Dragonfly-Object-Operation"
+	// HeaderDragonflyForwardedFor is used to mark http request forwarded from other peers
+	HeaderDragonflyForwardedFor = "X-Dragonfly-Forwarded-For"
 )

client/daemon/peer/peertask_stream.go

@@ -56,6 +56,14 @@ func (req *StreamTaskRequest) TaskID() string {
 	return req.taskID
 }
 
+func (req *StreamTaskRequest) HasParentTask() bool {
+	return req.Range != nil
+}
+
+func (req *StreamTaskRequest) ParentTaskID() string {
+	return idgen.ParentTaskIDV1(req.URL, req.URLMeta)
+}
+
 // StreamTask represents a peer task with stream io for reading directly without once more disk io
 type StreamTask interface {
 	// Start starts the special peer task, return an io.Reader for stream io

client/daemon/transport/transport.go

@@ -178,13 +178,11 @@ func peerProxyCacheLoaderFunc(c *ttlcache.Cache[string, *http.Transport], hostPo
 	}
 	roundTripper := &http.Transport{
 		Proxy: http.ProxyURL(proxyURL),
-		DialContext: func(dialer *net.Dialer) func(context.Context, string, string) (net.Conn, error) {
-			return dialer.DialContext
-		}(&net.Dialer{
+		DialContext: (&net.Dialer{
 			Timeout:   30 * time.Second,
 			KeepAlive: 30 * time.Second,
-		}),
-		MaxIdleConns:          100,
+		}).DialContext,
+		MaxIdleConns:          1000,
 		IdleConnTimeout:       90 * time.Second,
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
@@ -310,6 +308,7 @@ func (rt *transport) download(ctx context.Context, req *http.Request) (*http.Res
 	filter := nethttp.PickHeader(req.Header, config.HeaderDragonflyFilter, rt.defaultFilter)
 	tag := nethttp.PickHeader(req.Header, config.HeaderDragonflyTag, rt.defaultTag)
 	application := nethttp.PickHeader(req.Header, config.HeaderDragonflyApplication, rt.defaultApplication)
+	forwarded := nethttp.PickHeader(req.Header, config.HeaderDragonflyForwardedFor, "")
 	var priority = rt.defaultPriority
 	priorityString := nethttp.PickHeader(req.Header, config.HeaderDragonflyPriority, fmt.Sprintf("%d", rt.defaultPriority))
 	priorityInt, err := strconv.ParseInt(priorityString, 10, 32)
@@ -336,30 +335,16 @@ func (rt *transport) download(ctx context.Context, req *http.Request) (*http.Res
 	log := logger.With(append(logKV, "task", taskID)...)
 
 	log.Infof("start download with url: %s", reqURL)
-	log.Debugf("request url: %s, with header: %#v", reqURL, req.Header)
-
-	if rt.peerSearcher != nil {
-		searchPeerResult := rt.peerSearcher.SearchPeer(taskID)
-		switch searchPeerResult.Type {
-		case pex.SearchPeerResultTypeLocal:
-			log.Debugf("local peer exists")
-			goto local
-		case pex.SearchPeerResultTypeReplica:
-			log.Debugf("make replica from other peers")
-			goto local
-		case pex.SearchPeerResultTypeNotFound:
-			log.Debugf("no available peer after search peer")
-			goto local
-		case pex.SearchPeerResultTypeRemote:
-			resp, err := rt.proxyToPeers(log, req, searchPeerResult.Peers)
-			if err == nil {
-				return resp, nil
-			}
-			log.Warnf("proxy to other peers error: %s, fallback to local", err)
+	log.Debugf("request url: %s, with header: %#v, forwarded from: %q", reqURL, req.Header, forwarded)
+
+	// check whether the request is forwarded to avoid infinity forward
+	if rt.peerSearcher != nil && forwarded == "" {
+		resp := rt.tryProxyToPeers(log, req, streamTaskRequest)
+		if resp != nil {
+			return resp, nil
 		}
 	}
 
-local:
 	body, attr, err := rt.peerTaskManager.StartStreamTask(ctx, streamTaskRequest)
 	if err != nil {
 		log.Errorf("start stream task error: %v", err)
@@ -426,6 +411,50 @@ local:
 	return resp, nil
 }
 
+func (rt *transport) tryProxyToPeers(
+	log *logger.SugaredLoggerOnWith,
+	request *http.Request,
+	taskRequest *peer.StreamTaskRequest) (resp *http.Response) {
+	// search normal peers
+	resp = rt.searchPeers(log, request, taskRequest.TaskID(), taskRequest.PeerID)
+	if resp != nil {
+		return resp
+	}
+	// search parent peers
+	if taskRequest.HasParentTask() {
+		parentTaskID := taskRequest.ParentTaskID()
+		resp = rt.searchPeers(log, request, parentTaskID, taskRequest.PeerID)
+		if resp != nil {
+			log.Debugf("proxy to parent task %s done", parentTaskID)
+			return resp
+		}
+	}
+	return nil
+}
+
+func (rt *transport) searchPeers(log *logger.SugaredLoggerOnWith, request *http.Request, taskID, sourcePeerID string) *http.Response {
+	searchPeerResult := rt.peerSearcher.SearchPeer(taskID)
+	switch searchPeerResult.Type {
+	case pex.SearchPeerResultTypeLocal:
+		log.Debugf("local peer exists")
+	case pex.SearchPeerResultTypeReplica:
+		log.Debugf("make replica from other peers")
+	case pex.SearchPeerResultTypeNotFound:
+		log.Debugf("no available peer after search peer")
+	case pex.SearchPeerResultTypeRemote:
+		// mark request is already forwarded to avoid infinity forward
+		request.Header.Set(config.HeaderDragonflyForwardedFor, sourcePeerID)
+		resp, err := rt.proxyToPeers(log, request, searchPeerResult.Peers)
+		if err == nil {
+			return resp
+		}
+		// forward failed, remove marked header
+		request.Header.Del(config.HeaderDragonflyForwardedFor)
+		log.Warnf("proxy to other peers error: %s, fallback to local", err)
+	}
+	return nil
+}
+
 func (rt *transport) proxyToPeers(log *logger.SugaredLoggerOnWith, req *http.Request, peers []*pex.DestPeer) (*http.Response, error) {
 	// shuffle peers
 	if len(peers) > 1 {