| 43 | } |
| 44 | |
| 45 | func (v *vertex) execute(ctx context.Context) error { |
| 46 | if v.logger != nil { |
| 47 | v.logger.Log("msg", logging.Sprintf("started to execute vertex %s", v.name)) |
| 48 | defer v.logger.Log("msg", logging.Sprintf("finished executing vertex %s", v.name)) |
| 49 | } |
| 50 | for i := range v.parents { |
| 51 | select { |
| 52 | case <-v.parents[i].done: |
| 53 | case <-ctx.Done(): |
| 54 | return ctx.Err() |
| 55 | } |
| 56 | } |
| 57 | if err := v.work(ctx); err != nil { |
| 58 | return err |
| 59 | } |
| 60 | close(v.done) |
| 61 | |
| 62 | errGroup, ctx := errgroup.WithContext(ctx) |
| 63 | for _, c := range v.children { |
| 64 | c := c |
| 65 | c.once.Do(func() { |
| 66 | errGroup.Go(func() error { |
| 67 | return c.execute(ctx) |
| 68 | }) |
| 69 | }) |
| 70 | } |
| 71 | return errGroup.Wait() |
| 72 | } |
| 73 | |
| 74 | func (v *vertex) addChild(child *vertex) { |
| 75 | v.children = append(v.children, child) |