Fuzzer now deduplicates on the analysis side instead of the download side

fuzzer/go/aggregator/aggregator.go

 package aggregator
 
 import (
         "bytes"
         "crypto/sha1"
         "fmt"
         "io"
         "io/ioutil"
         "os"
         "path/filepath"
         "sort"
         "strings"
         "sync"
         "sync/atomic"
         "time"
 
         "github.com/skia-dev/glog"
         "go.skia.org/infra/fuzzer/go/common"
         "go.skia.org/infra/fuzzer/go/config"
-        "go.skia.org/infra/fuzzer/go/frontend/data"
+        "go.skia.org/infra/fuzzer/go/data"
+        "go.skia.org/infra/fuzzer/go/deduplicator"
         "go.skia.org/infra/go/exec"
         "go.skia.org/infra/go/fileutil"
         "go.skia.org/infra/go/metrics2"
         "go.skia.org/infra/go/util"
         "golang.org/x/net/context"
         "google.golang.org/cloud/storage"
 )
 
 // Aggregator is a key part of the fuzzing operation
 // (see https://skia.googlesource.com/buildbot/+/master/fuzzer/DESIGN.md).
@@ skipping 18 matching lines @@
         storageClient  *storage.Client
         fuzzPath       string
         executablePath string
         // For passing the paths of new binaries that should be scanned.
         forAnalysis chan string
         // For passing the file names of analyzed fuzzes that should be uploaded from where they rest on
         // disk in `fuzzPath`
         forUpload chan uploadPackage
 
         forBugReporting chan bugReportingPackage
+
+        deduplicator *deduplicator.Deduplicator
+
         // The shutdown channels are used to signal shutdowns.  There are two groups, to
         // allow for a softer, cleaner shutdown w/ minimal lost work.
         // Group A (monitoring) includes the scanning and the monitoring routine.
         // Group B (aggregation) include the analysis and upload routines and the bug reporting routine.
         monitoringShutdown   chan bool
         monitoringWaitGroup  *sync.WaitGroup
         aggregationShutdown  chan bool
         aggregationWaitGroup *sync.WaitGroup
         // These three counts are used to determine if there is any pending work.
         // There is no pending work if all three of these values are equal and the
         // work queues are empty.
         analysisCount  int64
         uploadCount    int64
         bugReportCount int64
+
+        greyNames []string
+        badNames  []string
 }
 
 const (
-        BAD_FUZZ  = "bad"
-        GREY_FUZZ = "grey"
+        BAD_FUZZ       = "bad"
+        GREY_FUZZ      = "grey"
+        HANG_THRESHOLD = 100
 )
 
 var (
         CLANG_DEBUG   = common.TEST_HARNESS_NAME + "_clang_debug"
         CLANG_RELEASE = common.TEST_HARNESS_NAME + "_clang_release"
         ASAN_DEBUG    = common.TEST_HARNESS_NAME + "_asan_debug"
         ASAN_RELEASE  = common.TEST_HARNESS_NAME + "_asan_release"
 )
 
 // uploadPackage is a struct containing all the pieces of a fuzz that need to be uploaded to GCS
 type uploadPackage struct {
         Data     data.GCSPackage
         FilePath string
         // Must be BAD_FUZZ or GREY_FUZZ
         FuzzType string
 }
 
 // bugReportingPackage is a struct containing the pieces of a fuzz that may need to have
 // a bug filed or updated.
 type bugReportingPackage struct {
         FuzzName   string
         CommitHash string
         IsBadFuzz  bool
 }
 
 // StartAggregator creates and starts a Aggregator.
 // If there is a problem starting up, an error is returned.  Other errors will be logged.
-func StartAggregator(s *storage.Client, category string) (*Aggregator, error) {
+func StartAggregator(s *storage.Client, category string, startingReports <-chan data.FuzzReport) (*Aggregator, error) {
         b := Aggregator{
                 Category:           category,
                 storageClient:      s,
                 fuzzPath:           filepath.Join(config.Aggregator.FuzzPath, category),
                 executablePath:     filepath.Join(config.Aggregator.ExecutablePath, category),
                 forAnalysis:        make(chan string, 10000),
                 forUpload:          make(chan uploadPackage, 100),
                 forBugReporting:    make(chan bugReportingPackage, 100),
                 MakeBugOnBadFuzz:   false,
                 UploadGreyFuzzes:   false,
+                deduplicator:       deduplicator.New(),
                 monitoringShutdown: make(chan bool, 2),
                 // aggregationShutdown needs to be created with a calculated capacity in start
         }
 
+        // preload the duplicator
+        for report := range startingReports {
+                b.deduplicator.IsUnique(report)
+        }
+
         return &b, b.start()
 }
 
 // start starts up the Aggregator.  It refreshes all status it needs and builds a debug and a
 // release version of Skia for use in analysis.  It then spawns the aggregation pipeline and a
 // monitoring thread.
 func (agg *Aggregator) start() error {
         // Set the wait groups to fresh
         agg.monitoringWaitGroup = &sync.WaitGroup{}
         agg.aggregationWaitGroup = &sync.WaitGroup{}
@@ skipping 210 matching lines @@
         if err := agg.setupAnalysis(executableDir); err != nil {
                 glog.Errorf("[%s] Analyzer %d terminated due to error: %s", agg.Category, identifier, err)
                 return
         }
         for {
                 select {
                 case badFuzzPath := <-agg.forAnalysis:
                         atomic.AddInt64(&agg.analysisCount, int64(1))
                         err := agg.analysisHelper(executableDir, badFuzzPath)
                         if err != nil {
+                                atomic.AddInt64(&agg.analysisCount, int64(-1))
                                 glog.Errorf("[%s] Analyzer %d terminated due to error: %s", agg.Category, identifier, err)
                                 return
                         }
                 case <-agg.aggregationShutdown:
                         glog.Infof("[%s] Analyzer %d recieved shutdown signal", agg.Category, identifier)
                         return
                 }
         }
 }
 
@@ skipping 149 matching lines @@
 func (agg *Aggregator) waitForUploads(identifier int) {
         defer agg.aggregationWaitGroup.Done()
         defer metrics2.NewCounter("upload-process-count", map[string]string{"category": agg.Category}).Dec(int64(1))
         glog.Infof("[%s] Spawning uploader %d", agg.Category, identifier)
         for {
                 select {
                 case p := <-agg.forUpload:
                         atomic.AddInt64(&agg.uploadCount, int64(1))
                         if !agg.UploadGreyFuzzes && p.FuzzType == GREY_FUZZ {
                                 glog.Infof("[%s] Skipping upload of grey fuzz %s", agg.Category, p.Data.Name)
+                                // We are skipping the bugReport, so increment the counts.
+                                atomic.AddInt64(&agg.bugReportCount, int64(1))
+                                continue
+                        }
+                        if p.FuzzType != GREY_FUZZ && !agg.deduplicator.IsUnique(data.ParseReport(p.Data)) {
+                                glog.Infof("[%s] Skipping upload of duplicate fuzz %s", agg.Category, p.Data.Name)
+                                // We are skipping the bugReport, so increment the counts.
+                                atomic.AddInt64(&agg.bugReportCount, int64(1))
                                 continue
                         }
                         if err := agg.upload(p); err != nil {
                                 glog.Errorf("[%s] Uploader %d terminated due to error: %s", agg.Category, identifier, err)
+                                // We are skipping the bugReport, so increment the counts.
+                                atomic.AddInt64(&agg.bugReportCount, int64(1))
                                 return
                         }
                         agg.forBugReporting <- bugReportingPackage{
                                 FuzzName:   p.Data.Name,
                                 CommitHash: config.Generator.SkiaVersion.Hash,
                                 IsBadFuzz:  p.FuzzType == BAD_FUZZ,
                         }
                 case <-agg.aggregationShutdown:
                         glog.Infof("[%s] Uploader %d recieved shutdown signal", agg.Category, identifier)
                         return
                 }
         }
 }
 
 // upload breaks apart the uploadPackage into its constituant parts and uploads them to GCS using
 // some helper methods.
 func (agg *Aggregator) upload(p uploadPackage) error {
         glog.Infof("[%s] uploading %s with file %s and analysis bytes %d;%d;%d|%d;%d;%d", agg.Category, p.Data.Name, p.FilePath, len(p.Data.Debug.Asan), len(p.Data.Debug.Dump), len(p.Data.Debug.StdErr), len(p.Data.Release.Asan), len(p.Data.Release.Dump), len(p.Data.Release.StdErr))
+        if p.FuzzType == GREY_FUZZ {
+                agg.greyNames = append(agg.greyNames, p.Data.Name)
+        } else {
+                agg.badNames = append(agg.badNames, p.Data.Name)
+        }
 
         if err := agg.uploadBinaryFromDisk(p, p.Data.Name, p.FilePath); err != nil {
                 return err
         }
         if err := agg.uploadString(p, p.Data.Name+"_debug.asan", p.Data.Debug.Asan); err != nil {
                 return err
         }
         if err := agg.uploadString(p, p.Data.Name+"_debug.dump", p.Data.Debug.Dump); err != nil {
                 return err
         }
@@ skipping 111 matching lines @@
         // aggregationShutdown channel.
         for i := len(agg.aggregationShutdown); i < cap(agg.aggregationShutdown); i++ {
                 agg.aggregationShutdown <- true
         }
         agg.aggregationWaitGroup.Wait()
 }
 
 // RestartAnalysis restarts the shut down aggregator.  Anything that is in the scanning directory
 // should be cleared out, lest it be rescanned/analyzed.
 func (agg *Aggregator) RestartAnalysis() error {
+        agg.deduplicator.Clear()
         return agg.start()
 }
 
 // WaitForEmptyQueues will return once there is nothing more in the analysis-upload-report
 // pipeline, waiting in increments of config.Aggregator.StatusPeriod until it is done.
 func (agg *Aggregator) WaitForEmptyQueues() {
         a := len(agg.forAnalysis)
         u := len(agg.forUpload)
         b := len(agg.forBugReporting)
         if a == 0 && u == 0 && b == 0 && agg.analysisCount == agg.uploadCount && agg.uploadCount == agg.bugReportCount {
                 glog.Infof("[%s] Queues were already empty", agg.Category)
                 return
         }
         t := time.Tick(config.Aggregator.StatusPeriod)
         glog.Infof("[%s] Waiting %s for the aggregator's queues to be empty", agg.Category, config.Aggregator.StatusPeriod)
+        hangCount := 0
         for _ = range t {
                 a = len(agg.forAnalysis)
                 u = len(agg.forUpload)
                 b = len(agg.forBugReporting)
                 glog.Infof("[%s] AnalysisQueue: %d, UploadQueue: %d, BugReportingQueue: %d", agg.Category, a, u, b)
                 glog.Infof("[%s] AnalysisTotal: %d, UploadTotal: %d, BugReportingTotal: %d", agg.Category, agg.analysisCount, agg.uploadCount, agg.bugReportCount)
                 if a == 0 && u == 0 && b == 0 && agg.analysisCount == agg.uploadCount && agg.uploadCount == agg.bugReportCount {
                         break
                 }
+                // This prevents waiting forever if an upload crashes, aborts or otherwise hangs.
+                hangCount++
+                if hangCount >= HANG_THRESHOLD {
+                        glog.Warningf("Was waiting for %d rounds and still wasn't done.  Quitting anyway.", hangCount)
+                }
+
                 glog.Infof("[%s] Waiting %s for the aggregator's queues to be empty", agg.Category, config.Aggregator.StatusPeriod)
-
         }
 }
 
 // ForceAnalysis directly adds the given path to the analysis queue, where it will be analyzed,
 // uploaded and possibly bug reported.
 func (agg *Aggregator) ForceAnalysis(path string) {
         agg.forAnalysis <- path
 }
+
+func (agg *Aggregator) ClearUploadedFuzzNames() {
+        agg.greyNames = []string{}
+        agg.badNames = []string{}
+}
+
+func (agg *Aggregator) UploadedFuzzNames() (bad, grey []string) {
+        return agg.badNames, agg.greyNames
+}