Feat/fetch p2p network info dfdaemon - #5
Open
wangforthinker wants to merge 39 commits into
Open
wangforthinker wants to merge 39 commits into
wangforthinker wants to merge 39 commits into
Conversation
…work if need update. fix bug
add some log for test.
fix not report to uploader.
fix requestManager get wrong front urls.
Feat/support local task reload
add heart beat for peer
wangforthinker
force-pushed
the
feat/fetch-p2p-network-info-dfdaemon
branch
from
March 12, 2020 02:31
79ef05b to
d92072a
Compare
wangforthinker
force-pushed
the
feat/fetch-p2p-network-info-dfdaemon
branch
2 times, most recently
from
March 12, 2020 07:06
e8ae41e to
f241659
Compare
lowzj
reviewed
Mar 13, 2020
| "sync" | ||
| ) | ||
|
|
||
| // finiteQueue provides a circle queue with capacity, and it will weed out the earlier item if full |
There was a problem hiding this comment.
pkg/queue 下面有个有限队列的实现，跟这个区别在于队列满时会阻塞。可否考虑将本实现也放到pkg/queue下面。
另外我觉得应该叫循环队列，但是我从实现上看更像个cache。
Author
There was a problem hiding this comment.
嗯 这是一个有限长度的循环队列。已放在pkg/queue.
| type finiteQueue struct { | ||
| sync.Mutex | ||
| capacity int | ||
| head *itemNode |
| uploaderAPI api.UploaderAPI | ||
|
|
||
| // postNotifyUploader should be called after notify the local uploader finish | ||
| postNotifyUploader func(req *api.FinishTaskRequest) |
Author
There was a problem hiding this comment.
| ticker := time.NewTicker(5 * time.Millisecond) | ||
| defer ticker.Stop() | ||
| timeout := time.After(233 * time.Millisecond) | ||
| timeout := time.After(10 * time.Second) |
| logrus.Infof("in downloadPiece by returnSrc, url: %s, header: %v, err: %d", pc.pieceTask.Url, header, err) | ||
| }else{ | ||
| downloadRequest := pc.createDownloadRequest() | ||
| downloadRequest.PieceRange = fmt.Sprintf("0-%d", downloadRequest.PieceSize + config.PieceMetaSize) |
| buf := make([]byte, 128) | ||
| binary.BigEndian.PutUint32(buf, uint32((pieceSize)|(pieceSize)<<4)) | ||
| content.Write(buf[:config.PieceHeadSize]) | ||
| defer content.Write([]byte{config.PieceTailChar}) |
There was a problem hiding this comment.
直接添加分片头尾信息有问题，因为分片的md5值是包含了：piece head + piece data + piece tail。
这种情况下考虑使用 NewLimitReaderWithLimiterAndMD5Sum ，下载前后将分片头尾信息放到md5计算中。
| func (csw *ClientStreamWriter) Read(p []byte) (n int, err error) { | ||
| n, err = csw.limitReader.Read(p) | ||
| csw.alreadyReadSize += int64(n) | ||
| if csw.alreadyReadSize == csw.expectReadSize { |
There was a problem hiding this comment.
这里是为了做什么？csw.alreadyReadSize += int64(n) 结果可能也大于 csw. expectReadSize
Author
There was a problem hiding this comment.
当读到的内容跟预计的一样多时，主动close(pipewriter)。 不然请求方不会收到EOF。
| // fix resource release | ||
| func (roundTripper *DFRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { | ||
| if req.Header.Get("x-nydus-proxy-healthcheck") != "" { | ||
| if roundTripper.config.Extreme.SpecKeyOfDirectRet != "" { |
There was a problem hiding this comment.
这一段改造可否独立个函数出来，因为跟之前的逻辑基本都不一样，可在这里判断进入不同的处理函数中。
wangforthinker
force-pushed
the
feat/fetch-p2p-network-info-dfdaemon
branch
3 times, most recently
from
March 16, 2020 08:03
0364ad6 to
4d1a730
Compare
lowzj
reviewed
Mar 18, 2020
| // remove the earliest item | ||
| i := q.internalRemoveTail() | ||
| if i != nil { | ||
| delete(q.itemMap, key) |
There was a problem hiding this comment.
The deleted element is not the tail of the list, it may delete nothing actually.
The stored value in list.Element may be like this:
type elementData struct {
key string
data interface{}
}So delete the value from map can be written:
delete(q.itemMap, i.Value.(*elementData).key)|
|
||
| // if pieceRange == "" means all Pieces of file | ||
| func (sm *SchedulerManager) SchedulerByTaskID(ctx context.Context, taskID string, srcCid string, pieceRange string, pieceSize int32) ([]*Result, error) { | ||
| sm.Lock() |
There was a problem hiding this comment.
这个锁的粒度似乎太大了，所有任务都需要串行调度，而且对于peer负载的更新也依赖于此锁
Author
There was a problem hiding this comment.
实际测试下来，调度所占耗时不到1%，暂时不影响。
| PeerInfo: node.Basic, | ||
| Generation: sm.generation, | ||
| StartDownload: func(peerID string, generation int64) { | ||
| sm.downloadStartCh <- notifySt{peerID: peerID, generation: generation} |
| WithRegistryMirror(c.RegistryMirror), | ||
| WithStreamDownloaderFactory(func() downloader.Stream { | ||
| // dfget.NewGetter(c.DFGetConfig()) | ||
| dfget.NewGetter(c.DFGetConfig()) |
wangforthinker
force-pushed
the
feat/fetch-p2p-network-info-dfdaemon
branch
from
March 19, 2020 02:27
4d1a730 to
5b9b471
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Ⅰ. Describe what this PR did
Ⅱ. Does this pull request fix one issue?
Ⅲ. Why don't you add test cases (unit test/integration test)? (你真的觉得不需要加测试吗？)
Ⅳ. Describe how to verify it
Ⅴ. Special notes for reviews