Increase parallelism by putting more things in goroutines

drive-demuxer.go

@@ -190,6 +190,10 @@ func (s *Step) RemoveInputDirs(child string) {
 }
 
 func (s *Step) ListChildren() []string {
+	s.OpenHandles <- 1
+	defer func() {
+		<-s.OpenHandles
+	}()
 	leftInPath := s.InputPath("", LEFT)
 	rightInPath := s.InputPath("", RIGHT)
 	results := make(map[string]bool)
@@ -215,6 +219,10 @@ func (s *Step) ListChildren() []string {
 }
 
 func (s *Step) AreFilesIdentical(child string) bool {
+	s.OpenHandles <- 1
+	defer func() {
+		<-s.OpenHandles
+	}()
 	leftInPath := s.InputPath(child, LEFT)
 	rightInPath := s.InputPath(child, RIGHT)
 	leftInfo, err := os.Stat(leftInPath)
@@ -254,8 +262,9 @@ func (s *Step) AreFilesIdentical(child string) bool {
 		}
 	}()
 
-	var leftData [4096]byte
+	// 4 MB buffers
-	var rightData [4096]byte
+	var leftData [4194304]byte
+	var rightData [4194304]byte
 	for {
 		leftRead, err := leftFile.Read(leftData[:])
 		if err != nil && err != io.EOF {
@@ -270,7 +279,7 @@ func (s *Step) AreFilesIdentical(child string) bool {
 		if leftRead != rightRead {
 			return false
 		}
-		if !bytes.Equal(leftData[:], rightData[:]) {
+		if !bytes.Equal(leftData[:leftRead], rightData[:rightRead]) {
 			return false
 		}
 		if err == io.EOF {
@@ -282,51 +291,92 @@ func (s *Step) AreFilesIdentical(child string) bool {
 func (s *Step) Walk() {
 	wg := sync.WaitGroup{}
 
-	s.OpenHandles <- 1
 	children := s.ListChildren()
 	s.Bar.AddTotal(int64(len(children)))
 	for _, child := range children {
 		rightPath := s.InputPath(child, RIGHT)
 		rightState, err := s.CheckState(rightPath)
 		if err != nil {
-			log.Printf("Error statting path %s: %s\n", rightPath, err)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Error statting path %s: %s\n", rightPath, err)
+				s.Bar.Increment()
+			}()
 			continue
 		} else if rightState == UNKNOWN {
-			log.Printf("Unknown stat value for path %s\n", rightPath)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Unknown stat value for path %s\n", rightPath)
+				s.Bar.Increment()
+			}()
 			continue
 		} else if rightState == MISSING {
-			s.SeparateLeft(child)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				s.SeparateLeft(child)
+				s.Bar.Increment()
+			}()
 			continue
 		}
 
 		leftPath := s.InputPath(child, LEFT)
 		leftState, err := s.CheckState(leftPath)
 		if err != nil {
-			log.Printf("Error statting path %s: %s\n", leftPath, err)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Error statting path %s: %s\n", leftPath, err)
+				s.Bar.Increment()
+			}()
 			continue
 		} else if leftState == UNKNOWN {
-			log.Printf("Unknown stat value for path %s\n", leftPath)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Unknown stat value for path %s\n", leftPath)
+				s.Bar.Increment()
+			}()
 			continue
 		} else if leftState == MISSING {
-			s.SeparateRight(child)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func() {
+				defer func() {
+					wg.Done()
+				}()
+				s.SeparateRight(child)
+				s.Bar.Increment()
+			}()
 			continue
 		}
 
 		switch rightState {
 		case FILE:
-			if leftState == FILE && s.AreFilesIdentical(child) {
+			wg.Add(1)
-				s.Combine(child)
+			go func() {
-				s.Bar.Increment()
+				defer func() {
-			} else {
+					wg.Done()
-				s.Separate(child)
+				}()
-				s.Bar.Increment()
+				if leftState == FILE && s.AreFilesIdentical(child) {
-			}
+					s.Combine(child)
+					s.Bar.Increment()
+				} else {
+					s.Separate(child)
+					s.Bar.Increment()
+				}
+			}()
 		case DIRECTORY:
 			if leftState == DIRECTORY {
 				s.MakeCombinedDir(child)
@@ -352,7 +402,6 @@ func (s *Step) Walk() {
 		}
 	}
 
-	<-s.OpenHandles
 	wg.Wait()
 }