Fix parallelism

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,77 +291,123 @@ 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(rightPath string) {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Error statting path %s: %s\n", rightPath, err)
+				s.Bar.Increment()
+			}(rightPath)
 			continue
 		} else if rightState == UNKNOWN {
-			log.Printf("Unknown stat value for path %s\n", rightPath)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func(rightPath string) {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Unknown stat value for path %s\n", rightPath)
+				s.Bar.Increment()
+			}(rightPath)
 			continue
 		} else if rightState == MISSING {
-			s.SeparateLeft(child)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func(child string) {
+				defer func() {
+					wg.Done()
+				}()
+				s.SeparateLeft(child)
+				s.Bar.Increment()
+			}(child)
 			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(leftPath string) {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Error statting path %s: %s\n", leftPath, err)
+				s.Bar.Increment()
+			}(leftPath)
 			continue
 		} else if leftState == UNKNOWN {
-			log.Printf("Unknown stat value for path %s\n", leftPath)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func(leftPath string) {
+				defer func() {
+					wg.Done()
+				}()
+				log.Printf("Unknown stat value for path %s\n", leftPath)
+				s.Bar.Increment()
+			}(leftPath)
 			continue
 		} else if leftState == MISSING {
-			s.SeparateRight(child)
+			wg.Add(1)
-			s.Bar.Increment()
+			go func(child string) {
+				defer func() {
+					wg.Done()
+				}()
+				s.SeparateRight(child)
+				s.Bar.Increment()
+			}(child)
 			continue
 		}
 
 		switch rightState {
 		case FILE:
-			if leftState == FILE && s.AreFilesIdentical(child) {
+			wg.Add(1)
-				s.Combine(child)
+			go func(child string) {
-				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()
+				}
+			}(child)
 		case DIRECTORY:
 			if leftState == DIRECTORY {
-				s.MakeCombinedDir(child)
 				substep := Step{
 					TopLevel: s.TopLevel,
 					Subpath:  path.Join(s.Subpath, child),
 				}
 				wg.Add(1)
-				go func() {
+				go func(child string, substep *Step) {
 					defer func() {
 						wg.Done()
 					}()
+					s.MakeCombinedDir(child)
 					substep.Walk()
 					s.RemoveInputDirs(child)
 					s.Bar.Increment()
-				}()
+				}(child, &substep)
 			} else {
-				s.Separate(child)
+				wg.Add(1)
-				s.Bar.Increment()
+				go func(child string) {
+					defer func() {
+						wg.Done()
+					}()
+					s.Separate(child)
+					s.Bar.Increment()
+				}(child)
 			}
 		default:
 			panic("Unexpected state")
 		}
 	}
 
-	<-s.OpenHandles
 	wg.Wait()
 }