@@ -6,9 +6,10 @@ import (
66
77 "github.com/rs/zerolog"
88
9- "github.com/onflow/flow-go/engine"
109 "github.com/onflow/flow-go/engine/execution"
1110 "github.com/onflow/flow-go/module"
11+ "github.com/onflow/flow-go/module/component"
12+ "github.com/onflow/flow-go/module/irrecoverable"
1213 "github.com/onflow/flow-go/utils/logging"
1314
1415 "github.com/sethvargo/go-retry"
@@ -26,74 +27,102 @@ func NewAsyncUploader(uploader Uploader,
2627 maxRetryNumber uint64 ,
2728 log zerolog.Logger ,
2829 metrics module.ExecutionMetrics ) * AsyncUploader {
29- return & AsyncUploader {
30- unit : engine .NewUnit (),
30+ a := & AsyncUploader {
3131 uploader : uploader ,
3232 log : log .With ().Str ("component" , "block_data_uploader" ).Logger (),
3333 metrics : metrics ,
3434 retryInitialTimeout : retryInitialTimeout ,
3535 maxRetryNumber : maxRetryNumber ,
36+ // we use a channel rather than a Fifoqueue here because a Fifoqueue might drop items when full,
37+ // but it is not acceptable to skip uploading an execution result
38+ queue : make (chan * execution.ComputationResult , 20000 ),
3639 }
40+ builder := component .NewComponentManagerBuilder ()
41+ for i := 0 ; i < 10 ; i ++ {
42+ builder .AddWorker (a .UploadWorker )
43+ }
44+ a .cm = builder .Build ()
45+ a .Component = a .cm
46+ return a
3747}
3848
3949// AsyncUploader wraps up another Uploader instance and make its upload asynchronous
4050type AsyncUploader struct {
41- module.ReadyDoneAware
42- unit * engine.Unit
4351 uploader Uploader
4452 log zerolog.Logger
4553 metrics module.ExecutionMetrics
4654 retryInitialTimeout time.Duration
4755 maxRetryNumber uint64
4856 onComplete OnCompleteFunc // callback function called after Upload is completed
57+ queue chan * execution.ComputationResult
58+ cm * component.ComponentManager
59+ component.Component
4960}
5061
51- func (a * AsyncUploader ) Ready () <- chan struct {} {
52- return a .unit .Ready ()
53- }
54-
55- func (a * AsyncUploader ) Done () <- chan struct {} {
56- return a .unit .Done ()
62+ // UploadWorker implements a component worker which asynchronously uploads computation results
63+ // from the execution node (after a block is executed) to storage such as a GCP bucket or S3 bucket.
64+ func (a * AsyncUploader ) UploadWorker (ctx irrecoverable.SignalerContext , ready component.ReadyFunc ) {
65+ ready ()
66+
67+ done := ctx .Done ()
68+ for {
69+ select {
70+ case <- done :
71+ return
72+ case computationResult := <- a .queue :
73+ a .UploadTask (ctx , computationResult )
74+ }
75+ }
5776}
5877
5978func (a * AsyncUploader ) SetOnCompleteCallback (onComplete OnCompleteFunc ) {
6079 a .onComplete = onComplete
6180}
6281
82+ // Upload adds the computation result to a queue to be processed asynchronously by workers,
83+ // ensuring that multiple uploads can be run in parallel.
84+ // No errors expected during normal operation.
6385func (a * AsyncUploader ) Upload (computationResult * execution.ComputationResult ) error {
86+ a .queue <- computationResult
87+ return nil
88+ }
6489
90+ // UploadTask implements retrying for uploading computation results.
91+ // When the upload is complete, the callback will be called with the result (for example,
92+ // to record that the upload was successful) and any error.
93+ // No errors expected during normal operation.
94+ func (a * AsyncUploader ) UploadTask (ctx context.Context , computationResult * execution.ComputationResult ) {
6595 backoff := retry .NewFibonacci (a .retryInitialTimeout )
6696 backoff = retry .WithMaxRetries (a .maxRetryNumber , backoff )
6797
68- a .unit .Launch (func () {
69- a .metrics .ExecutionBlockDataUploadStarted ()
70- start := time .Now ()
71-
72- a .log .Debug ().Msgf ("computation result of block %s is being uploaded" ,
73- computationResult .ExecutableBlock .ID ().String ())
98+ a .metrics .ExecutionBlockDataUploadStarted ()
99+ start := time .Now ()
74100
75- err := retry .Do (a .unit .Ctx (), backoff , func (ctx context.Context ) error {
76- err := a .uploader .Upload (computationResult )
77- if err != nil {
78- a .log .Warn ().Err (err ).Msg ("error while uploading block data, retrying" )
79- }
80- return retry .RetryableError (err )
81- })
101+ a .log .Debug ().Msgf ("computation result of block %s is being uploaded" ,
102+ computationResult .ExecutableBlock .ID ().String ())
82103
104+ err := retry .Do (ctx , backoff , func (ctx context.Context ) error {
105+ err := a .uploader .Upload (computationResult )
83106 if err != nil {
84- a .log .Error ().Err (err ).
85- Hex ("block_id" , logging .Entity (computationResult .ExecutableBlock )).
86- Msg ("failed to upload block data" )
87- } else {
88- a .log .Debug ().Msgf ("computation result of block %s was successfully uploaded" ,
89- computationResult .ExecutableBlock .ID ().String ())
107+ a .log .Warn ().Err (err ).Msg ("error while uploading block data, retrying" )
90108 }
109+ return retry .RetryableError (err )
110+ })
91111
92- a .metrics .ExecutionBlockDataUploadFinished (time .Since (start ))
112+ // We only log upload errors here because the errors originate from an external cloud provider
113+ // and the upload success is not critical to correct continued operation of the node
114+ if err != nil {
115+ a .log .Error ().Err (err ).
116+ Hex ("block_id" , logging .Entity (computationResult .ExecutableBlock )).
117+ Msg ("failed to upload block data" )
118+ } else {
119+ a .log .Debug ().Msgf ("computation result of block %s was successfully uploaded" ,
120+ computationResult .ExecutableBlock .ID ().String ())
121+ }
93122
94- if a . onComplete != nil {
95- a . onComplete ( computationResult , err )
96- }
97- } )
98- return nil
123+ a . metrics . ExecutionBlockDataUploadFinished ( time . Since ( start ))
124+
125+ if a . onComplete != nil {
126+ a . onComplete ( computationResult , err )
127+ }
99128}
0 commit comments