@cryptotaxi247 / kubo / commits / 17e789203

enhance(cmd/verify): add goroutine count to improve verify speed

License: MIT Signed-off-by: chenminjian <727180553@qq.com>

chenminjian committed Nov 2, 2018 at 16:27 UTC 17e7892036a696b9a349eb79b42cccbc4a6a48f1
1 file changed +50 -6
core/commands/repo.go
+50 -6
@@ -2,12 +2,15 @@ package commands
2
3 import (
4 "bytes"
5 + "context"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 + "runtime"
12 "strings"
13 + "sync"
14 "text/tabwriter"
15
16 oldcmds "github.com/ipfs/go-ipfs/commands"
@@ -276,6 +279,48 @@ type VerifyProgress struct {
279 Progress int
280 }
281
282 +func verifyWorkerRun(ctx context.Context, wg *sync.WaitGroup, keys <-chan cid.Cid, results chan<- string, bs bstore.Blockstore) {
283 + defer wg.Done()
284 +
285 + for k := range keys {
286 + _, err := bs.Get(k)
287 + if err != nil {
288 + select {
289 + case results <- fmt.Sprintf("block %s was corrupt (%s)", k, err):
290 + case <-ctx.Done():
291 + return
292 + }
293 +
294 + continue
295 + }
296 +
297 + select {
298 + case results <- "":
299 + case <-ctx.Done():
300 + return
301 + }
302 + }
303 +}
304 +
305 +func verifyResultChan(ctx context.Context, keys <-chan cid.Cid, bs bstore.Blockstore) <-chan string {
306 + results := make(chan string)
307 +
308 + go func() {
309 + defer close(results)
310 +
311 + var wg sync.WaitGroup
312 +
313 + for i := 0; i < runtime.NumCPU()*2; i++ {
314 + wg.Add(1)
315 + go verifyWorkerRun(ctx, &wg, keys, results, bs)
316 + }
317 +
318 + wg.Wait()
319 + }()
320 +
321 + return results
322 +}
323 +
324 var repoVerifyCmd = &oldcmds.Command{
325 Helptext: cmdkit.HelpText{
326 Tagline: "Verify all blocks in repo are not corrupted.",
@@ -300,15 +345,14 @@ var repoVerifyCmd = &oldcmds.Command{
345 return
346 }
347
348 + results := verifyResultChan(req.Context(), keys, bs)
349 +
350 var fails int
351 var i int
305 - for k := range keys {
306 - _, err := bs.Get(k)
307 - if err != nil {
352 + for msg := range results {
353 + if msg != "" {
354 select {
309 - case out <- &VerifyProgress{
310 - Msg: fmt.Sprintf("block %s was corrupt (%s)", k, err),
311 - }:
355 + case out <- &VerifyProgress{Msg: msg}:
356 case <-req.Context().Done():
357 return
358 }