| 82 | } |
| 83 | |
| 84 | func (r *resolver) ResolveFilePaths(input io.Reader, output io.Writer) { |
| 85 | reader := csv.NewReader(input) |
| 86 | writer := csv.NewWriter(output) |
| 87 | // Start the workers going. |
| 88 | numWorkers := 100 |
| 89 | workQueue := make(chan []string, numWorkers) |
| 90 | outputQueue := make(chan []string, 1000) |
| 91 | wg := sync.WaitGroup{} |
| 92 | for range numWorkers { |
| 93 | wg.Add(1) |
| 94 | go func() { |
| 95 | defer wg.Done() |
| 96 | for { |
| 97 | sample, ok := <-workQueue |
| 98 | if !ok { |
| 99 | return |
| 100 | } |
| 101 | filename := sample[0] |
| 102 | owner, err := strconv.ParseUint(sample[1], 0, 63) |
| 103 | if err != nil { |
| 104 | panic(fmt.Errorf("Couldn't parse owner inode (%v): %v", sample[1], err)) |
| 105 | } |
| 106 | path, err := r.Resolve(msgs.InodeId(owner), filename) |
| 107 | if err != nil { |
| 108 | r.logger.ErrorNoAlert("Failed to resolve file path: %v", err) |
| 109 | } else { |
| 110 | sample[0] = path // overwrite filename with resolved path |
| 111 | } |
| 112 | outputQueue <- sample |
| 113 | } |
| 114 | }() |
| 115 | } |
| 116 | go func() { |
| 117 | for { |
| 118 | sample, err := reader.Read() |
| 119 | if err == io.EOF { |
| 120 | break |
| 121 | } |
| 122 | if err != nil { |
| 123 | panic(fmt.Errorf("error reading csv: %v", err)) |
| 124 | } |
| 125 | workQueue <- sample |
| 126 | } |
| 127 | r.logger.Info("Finished reading input") |
| 128 | close(workQueue) |
| 129 | wg.Wait() |
| 130 | close(outputQueue) |
| 131 | }() |
| 132 | processed := 0 |
| 133 | for sample := range outputQueue { |
| 134 | err := writer.Write(sample) |
| 135 | if err != nil { |
| 136 | panic(fmt.Errorf("error writing csv: %v", err)) |
| 137 | } |
| 138 | processed += 1 |
| 139 | if processed%10000 == 0 { |
| 140 | r.logger.Info("%d samples processed", processed) |
| 141 | } |