Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: optimize stream task id #2983

Merged
merged 3 commits into from
Dec 29, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions client/daemon/peer/peertask_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -341,6 +341,7 @@ func (ptm *peerTaskManager) StartStreamTask(ctx context.Context, req *StreamTask
IsMigrating: false,
}

taskID := idgen.TaskIDV1(req.URL, req.URLMeta)
if ptm.Multiplex {
// try breakpoint resume for task has range header
if req.Range != nil && !ptm.SplitRunningTasks {
Expand All @@ -357,14 +358,14 @@ func (ptm *peerTaskManager) StartStreamTask(ctx context.Context, req *StreamTask
}

// reuse by completed task
r, attr, ok := ptm.tryReuseStreamPeerTask(ctx, req)
r, attr, ok := ptm.tryReuseStreamPeerTask(ctx, taskID, req)
if ok {
metrics.PeerTaskCacheHitCount.Add(1)
return r, attr, nil
}
}

pt, err := ptm.newStreamTask(ctx, peerTaskRequest, req.Range)
pt, err := ptm.newStreamTask(ctx, taskID, peerTaskRequest, req.Range)
if err != nil {
return nil, nil, err
}
Expand Down
1 change: 1 addition & 0 deletions client/daemon/peer/peertask_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -910,6 +910,7 @@ func (ts *testSpec) runConductorTest(assert *testifyassert.Assertions, require *

// test reuse stream task
rc, _, ok := ptm.tryReuseStreamPeerTask(context.Background(),
taskID,
&StreamTaskRequest{
URL: ts.url,
URLMeta: urlMeta,
Expand Down
3 changes: 1 addition & 2 deletions client/daemon/peer/peertask_reuse.go
Original file line number Diff line number Diff line change
Expand Up @@ -225,9 +225,8 @@ func (ptm *peerTaskManager) storePartialFile(ctx context.Context, request *FileT
return nil
}

func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context, taskID string,
request *StreamTaskRequest) (io.ReadCloser, map[string]string, bool) {
taskID := idgen.TaskIDV1(request.URL, request.URLMeta)
var (
reuse *storage.ReusePeerTask
reuseRange *http.Range // the range of parent peer task data to read
Expand Down
5 changes: 4 additions & 1 deletion client/daemon/peer/peertask_reuse_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import (
"d7y.io/dragonfly/v2/client/daemon/storage"
"d7y.io/dragonfly/v2/client/daemon/storage/mocks"
"d7y.io/dragonfly/v2/client/daemon/test"
"d7y.io/dragonfly/v2/pkg/idgen"
"d7y.io/dragonfly/v2/pkg/net/http"
)

Expand Down Expand Up @@ -713,7 +714,9 @@ func TestReuseStreamPeerTask(t *testing.T) {
},
},
}
tc.verify(ptm.tryReuseStreamPeerTask(context.Background(), tc.request))

taskID := idgen.TaskIDV1(tc.request.URL, tc.request.URLMeta)
tc.verify(ptm.tryReuseStreamPeerTask(context.Background(), taskID, tc.request))
})
}
}
3 changes: 1 addition & 2 deletions client/daemon/peer/peertask_stream.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@ import (
"d7y.io/dragonfly/v2/client/daemon/storage"
logger "d7y.io/dragonfly/v2/internal/dflog"
"d7y.io/dragonfly/v2/internal/util"
"d7y.io/dragonfly/v2/pkg/idgen"
"d7y.io/dragonfly/v2/pkg/net/http"
)

Expand Down Expand Up @@ -73,6 +72,7 @@ type resumeStreamTask struct {

func (ptm *peerTaskManager) newStreamTask(
ctx context.Context,
taskID string,
request *schedulerv1.PeerTaskRequest,
rg *http.Range) (*streamTask, error) {
metrics.StreamTaskCount.Add(1)
Expand All @@ -87,7 +87,6 @@ func (ptm *peerTaskManager) newStreamTask(
parent = ptm.prefetchParentTask(request, "")
}

taskID := idgen.TaskIDV1(request.Url, request.UrlMeta)
ptc, err := ptm.getPeerTaskConductor(ctx, taskID, request, limit, parent, rg, "", false)
if err != nil {
return nil, err
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -317,7 +317,7 @@ func TestStreamPeerTask_BackSource_Partial_WithContentLength(t *testing.T) {
PeerHost: &schedulerv1.PeerHost{},
}
ctx := context.Background()
pt, err := ptm.newStreamTask(ctx, req, nil)
pt, err := ptm.newStreamTask(ctx, taskID, req, nil)
assert.Nil(err, "new stream peer task")

rc, _, err := pt.Start(ctx)
Expand Down
3 changes: 2 additions & 1 deletion client/daemon/peer/peertask_stream_resume_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,8 @@ func TestStreamPeerTask_Resume(t *testing.T) {

// set up parent task
wg.Add(1)
pt, err := ptm.newStreamTask(ctx, req, nil)

pt, err := ptm.newStreamTask(ctx, taskID, req, nil)
assert.Nil(err, "new parent stream peer task")

rc, _, err := pt.Start(ctx)
Expand Down
Loading