AydarZaynutdinov commented on a change in pull request #16683:
URL: https://github.com/apache/beam/pull/16683#discussion_r798386814
##########
File path: playground/backend/internal/preparers/java_preparers.go
##########
@@ -88,12 +97,35 @@ func (builder *JavaPreparersBuilder) WithFileNameChanger()
*JavaPreparersBuilder
return builder
}
+func (builder *JavaPreparersBuilder) WithGraphExtractor()
*JavaPreparersBuilder {
Review comment:
Please add a comment for this method.
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -112,6 +114,7 @@ func runStep(ctx context.Context, cacheService cache.Cache,
paths *fs_tool.LifeC
var runError bytes.Buffer
runOutput := streaming.RunOutputWriter{Ctx: pipelineLifeCycleCtx,
CacheService: cacheService, PipelineId: pipelineId}
go readLogFile(pipelineLifeCycleCtx, ctx, cacheService,
paths.AbsoluteLogFilePath, pipelineId, stopReadLogsChannel,
finishReadLogsChannel)
+ go readGraphFile(pipelineLifeCycleCtx, ctx, cacheService,
filepath.Join(paths.AbsoluteBaseFolderPath, preparers.GraphFileName),
pipelineId)
Review comment:
`LifeCyclePaths` contains value `AbsoluteLogFilePath`. Maybe we can do
the same for the graph file instead of
`filepath.Join(paths.AbsoluteBaseFolderPath, preparers.GraphFileName)`?
##########
File path: playground/backend/internal/preparers/python_preparers_test.go
##########
@@ -35,13 +35,13 @@ func TestGetPythonPreparers(t *testing.T) {
// As a result, want to receive slice of preparers with
len = 1
Review comment:
ditto.
##########
File path: playground/backend/internal/utils/preparators_utils.go
##########
@@ -41,7 +41,7 @@ func GetPreparers(sdk pb.Sdk, filepath string, valResults
*sync.Map) (*[]prepare
case pb.Sdk_SDK_GO:
Review comment:
ditto.
##########
File path: playground/backend/internal/preparers/python_preparers.go
##########
@@ -18,19 +18,32 @@ package preparers
import (
Review comment:
Is it required for this PR?
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -406,6 +409,32 @@ func cancelCheck(ctx context.Context, pipelineId
uuid.UUID, cancelChannel chan b
}
}
+// readGraphFile reads graph from the log file and keeps it to the cache.
+// If context is done it means that the code processing was finished
(successfully/with error/timeout). Write no graph to the cache.
+// If <-startReadGraphChannel it means that the graph written to the file and
can be read.
+// In other case each pauseDuration checks that graph file exists or not.
+func readGraphFile(pipelineLifeCycleCtx, backgroundCtx context.Context,
cacheService cache.Cache, graphFilePath string, pipelineId uuid.UUID) {
+ startReadGraphChannel := make(chan bool, 1)
+ ticker := time.NewTicker(pauseDuration)
+ for {
+ select {
+ // in case of timeout or cancel
+ case <-pipelineLifeCycleCtx.Done():
+ ticker.Stop()
+ return
+ // in case of graph file exists and can be read
+ case <-startReadGraphChannel:
+ utils.ReadAndSetToCacheGraph(backgroundCtx,
cacheService, pipelineId, graphFilePath)
+ return
+ // waiting when graph file appears
+ case <-ticker.C:
+ if _, err := os.Stat(graphFilePath); err == nil {
+ startReadGraphChannel <- true
+ }
Review comment:
we can combine these 2 cases:
```suggestion
// waiting when graph file appears
case <-ticker.C:
if _, err := os.Stat(graphFilePath); err == nil {
utils.ReadAndSetToCacheGraph(backgroundCtx,
cacheService, pipelineId, graphFilePath)
return
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
