handle frame.err in flush
Vladimir Bauer
3 years ago
| 291 | 291 | return s.flush(cw, height) |
| 292 | 292 | } |
| 293 | 293 | |
| 294 | func (s *pState) flush(cw *cwriter.Writer, height int) error { | |
| 294 | func (s *pState) flush(cw *cwriter.Writer, height int) (err error) { | |
| 295 | 295 | var wg sync.WaitGroup |
| 296 | 296 | var popCount int |
| 297 | 297 | |
| 298 | 298 | for s.bHeap.Len() > 0 { |
| 299 | var usedRows int | |
| 300 | 299 | b := heap.Pop(&s.bHeap).(*Bar) |
| 301 | 300 | frame := <-b.frameCh |
| 302 | if frame.recovered { | |
| 303 | s.heapUpdated = true | |
| 301 | if frame.err != nil { | |
| 302 | if err == nil { | |
| 303 | err = frame.err | |
| 304 | } | |
| 304 | 305 | continue |
| 305 | 306 | } |
| 307 | var usedRows int | |
| 306 | 308 | for i := len(frame.rows) - 1; i >= 0; i-- { |
| 307 | 309 | if row := frame.rows[i]; len(s.rows) < height { |
| 308 | 310 | s.rows = append(s.rows, row) |
| 340 | 342 | s.pool = append(s.pool, b) |
| 341 | 343 | } |
| 342 | 344 | |
| 345 | if err != nil { | |
| 346 | return err | |
| 347 | } | |
| 348 | ||
| 343 | 349 | wg.Add(1) |
| 344 | 350 | go func() { |
| 345 | 351 | for _, b := range s.pool { |
| 348 | 354 | wg.Done() |
| 349 | 355 | }() |
| 350 | 356 | |
| 351 | readRows := len(s.rows) | |
| 352 | for i := readRows - 1; i >= 0; i-- { | |
| 357 | for i := len(s.rows) - 1; i >= 0; i-- { | |
| 353 | 358 | _, err := cw.ReadFrom(s.rows[i]) |
| 354 | 359 | if err != nil { |
| 355 | if s.debugOut != nil { | |
| 356 | fmt.Fprintf(s.debugOut, "cw.ReadFrom: %s\n", err.Error()) | |
| 357 | } | |
| 358 | readRows-- | |
| 359 | } | |
| 360 | } | |
| 361 | ||
| 362 | err := cw.Flush(readRows - popCount) | |
| 360 | wg.Wait() | |
| 361 | return err | |
| 362 | } | |
| 363 | } | |
| 364 | ||
| 365 | err = cw.Flush(len(s.rows) - popCount) | |
| 363 | 366 | wg.Wait() |
| 364 | 367 | s.rows = s.rows[:0] |
| 365 | 368 | s.pool = s.pool[:0] |