disttask: add operator abstraction #45927

ywqzzy · 2023-08-09T02:59:22Z

What problem does this PR solve?

Issue Number: ref #41495

Problem Summary:

What is changed and how it works?

Operator

Operator can run business logic with multiple workers.
Note that we use workerPool to manage all workers.
Workers will handle task when there is a task send to it.

Data

Abstract DataSource and DataSink to handle the dataflow.

Pipeline

Pipeline is a chain of Operators.

The workflow is:

add task to the first operator, then this operator start to handle tasks.
When the task is handled, the operator send task to the next operator.

Moreover, we can read dataSource to add task. It will be implemented soon.

Check List

Tests

Unit test
Integration test
Manual test (add detailed scripts or steps below)
No code

Side effects

Performance regression: Consumes more CPU
Performance regression: Consumes more Memory
Breaking backward compatibility

Documentation

Release note

Please refer to Release Notes Language Style Guide to write a quality release note.

None

ti-chi-bot · 2023-08-09T02:59:28Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign zanmato1984 for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Needs approval from an approver in each of these files:

OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

tiprow · 2023-08-09T02:59:51Z

Hi @ywqzzy. Thanks for your PR.

PRs from untrusted users cannot be marked as trusted with /ok-to-test in this repo meaning untrusted PR authors can never trigger tests themselves. Collaborators can still trigger tests on the PR using /test all.

I understand the commands that are listed here.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

disttask/framework/operator/asyncPipeline.go

purelind · 2023-08-09T03:43:44Z

/retest

tiprow · 2023-08-09T03:44:14Z

@purelind: Cannot trigger testing until a trusted user reviews the PR and leaves an /ok-to-test message.

In response to this:

/retest

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

codecov · 2023-08-09T03:47:02Z

Codecov Report

Merging #45927 (c17dc02) into master (a0cbff2) will decrease coverage by 0.6708%.
Report is 59 commits behind head on master.
The diff coverage is 90.5511%.

Additional details and impacted files

@@               Coverage Diff                @@
##             master     #45927        +/-   ##
================================================
- Coverage   73.3499%   72.6792%   -0.6708%     
================================================
  Files          1277       1303        +26     
  Lines        393392     399993      +6601     
================================================
+ Hits         288553     290712      +2159     
- Misses        86434      90810      +4376     
- Partials      18405      18471        +66

Flag	Coverage Δ
integration	`25.5599% <0.0000%> (?)`
unit	`73.3556% <90.5511%> (+0.0056%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Components	Coverage Δ
dumpling	`54.0444% <ø> (ø)`
parser	`85.0548% <ø> (-0.0055%)`	⬇️
br	`47.6072% <ø> (-4.4240%)`	⬇️

…rator

D3Hunter · 2023-08-10T02:15:57Z

disttask/framework/operator/async.go

+	source DataSource
+	sink   DataSink
+	pool   *workerpool.WorkerPool[T] // workers running on pool


make them public, there's no need to have those getter/setter methods, those fields shouldn't be changed after initialized

this pool shouldn't be part of a operator, the pipeline that execute those operators should use it.

this pool shouldn't be part of a operator, the pipeline that execute those operators should use it.

Ok, pipeline hold the pools will be cleaner.

this pool shouldn't be part of a operator, the pipeline that execute those operators should use it.

Ok, pipeline hold the pools will be cleaner.

I will do it when we support better way to build pipeline with plan.

disttask/framework/operator/async.go

disttask/framework/operator/data.go

disttask/framework/operator/async_pipeline.go

disttask/framework/operator/async.go

D3Hunter · 2023-08-10T06:03:20Z

disttask/framework/operator/async_test.go

+	source0 := impl0.getSource()
+	source1 := impl1.getSource()
+	source2 := impl2.getSource()
+	sink := &simpleAsyncDataSink{0, 0, sync.Mutex{}}
+
+	impl0.setSink(source1.(DataSink))
+	impl1.setSink(source2.(DataSink))
+	impl2.setSink(sink)


if we're sure pipeline is a chain, better provide a method in pipeline to connect them in an easy way

if we're sure pipeline is a chain, better provide a method in pipeline to connect them in an easy way

Yep, I will try.

if we're sure pipeline is a chain, better provide a method in pipeline to connect them in an easy way

I will do it when plan is supported.

hawkingrei · 2023-08-15T03:10:58Z

/ok-to-test

ti-chi-bot · 2023-08-15T03:15:06Z

@ywqzzy: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
idc-jenkins-ci-tidb/build	`c17dc02`	link	true	`/test build`

Full PR test history. Your PR dashboard.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.

tiprow · 2023-08-15T03:26:02Z

@ywqzzy: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
tiprow_fast_test	`c17dc02`	link	true	`/test tiprow_fast_test`

Full PR test history. Your PR dashboard.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.

tangenta · 2023-08-16T05:49:50Z

disttask/framework/operator/data.go

+// Operator read data from DataSource, then process the read data.
+type DataSource[T any] interface {
+	Start() error
+	Next() (T, error)


How about renaming it to Read()? It should be complementary to DataSink.Write().

How about renaming it to Read()? It should be complementary to DataSink.Write().

Ok.

ywqzzy added 2 commits August 9, 2023 10:58

update

9616f27

update

9a836e7

ti-chi-bot bot added do-not-merge/needs-linked-issue release-note-none Denotes a PR that doesn't merit a release note. labels Aug 9, 2023

ti-chi-bot bot added the size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. label Aug 9, 2023

refine

a31c045

hawkingrei reviewed Aug 9, 2023

View reviewed changes

disttask/framework/operator/asyncPipeline.go Outdated Show resolved Hide resolved

ywqzzy added 3 commits August 9, 2023 11:11

fix

ff41503

refine

f906379

fix

fd035cd

ywqzzy added 4 commits August 9, 2023 11:57

fix

729dbc5

fix

f1af317

Merge branch 'master' of https://github.com/pingcap/tidb into new_ope…

d99061f

…rator

fix

99d518c

D3Hunter reviewed Aug 10, 2023

View reviewed changes

ywqzzy added 10 commits August 12, 2023 14:37

refine

b791c01

refine

75e0a3f

fix

faddcaa

fix

27285bb

refine

1e32c9c

update

236a573

refine comments.

b1f3f4f

refine

e9b9aac

refine

92b127b

refine

83d89ab

more cases, refine datasource and sink.

c17dc02

ti-chi-bot bot added the ok-to-test Indicates a PR is ready to be tested. label Aug 15, 2023

ti-chi-bot bot removed the do-not-merge/needs-linked-issue label Aug 16, 2023

ywqzzy requested a review from D3Hunter August 16, 2023 02:11

tangenta reviewed Aug 16, 2023

View reviewed changes

ywqzzy closed this Aug 21, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

disttask: add operator abstraction #45927

disttask: add operator abstraction #45927

ywqzzy commented Aug 9, 2023 •

edited

Loading

ti-chi-bot bot commented Aug 9, 2023

tiprow bot commented Aug 9, 2023

purelind commented Aug 9, 2023

tiprow bot commented Aug 9, 2023

codecov bot commented Aug 9, 2023 •

edited

Loading

D3Hunter Aug 10, 2023

D3Hunter Aug 10, 2023

ywqzzy Aug 11, 2023

ywqzzy Aug 15, 2023

D3Hunter Aug 10, 2023

ywqzzy Aug 11, 2023

ywqzzy Aug 15, 2023

hawkingrei commented Aug 15, 2023

ti-chi-bot bot commented Aug 15, 2023

tiprow bot commented Aug 15, 2023

tangenta Aug 16, 2023

ywqzzy Aug 16, 2023

disttask: add operator abstraction #45927

disttask: add operator abstraction #45927

Conversation

ywqzzy commented Aug 9, 2023 • edited Loading

What problem does this PR solve?

What is changed and how it works?

Operator

Data

Pipeline

Check List

Release note

ti-chi-bot bot commented Aug 9, 2023

tiprow bot commented Aug 9, 2023

purelind commented Aug 9, 2023

tiprow bot commented Aug 9, 2023

codecov bot commented Aug 9, 2023 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hawkingrei commented Aug 15, 2023

ti-chi-bot bot commented Aug 15, 2023

tiprow bot commented Aug 15, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ywqzzy commented Aug 9, 2023 •

edited

Loading

codecov bot commented Aug 9, 2023 •

edited

Loading