Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Optimize SubtaskGraph generation #3342

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
+19 −8
Open

Optimize SubtaskGraph generation #3342

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8b763e2
Optimize SubtaskGraph generation
zhongchun Apr 21, 2023
ef47ad4
Initialize `dtypes`, `index_value`, and `columns_value` for `BaseData…
zhongchun Apr 24, 2023
40373a8
Initialize the lazy fields of ChunkData explicitly
zhongchun Apr 25, 2023
8fb4545
Fix chunk creation of subtask graph generation
zhongchun Apr 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 19 additions & 8 deletions mars/services/task/analyzer/analyzer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,11 +273,13 @@ def _gen_subtask_info(
chunk_priority = chunk.op.priority
# process input chunks
inp_chunks = []
input_changed = False
build_fetch_index_to_chunks = dict()
for i, inp_chunk in enumerate(chunk.inputs):
if inp_chunk in chunks_set:
inp_chunks.append(chunk_to_copied[inp_chunk])
else:
input_changed = True
build_fetch_index_to_chunks[i] = inp_chunk
inp_chunks.append(None)
if not isinstance(inp_chunk.op, Fetch):
Expand All @@ -287,14 +289,23 @@ def _gen_subtask_info(
)
for i, fetch_chunk in zip(build_fetch_index_to_chunks, fetch_chunks):
inp_chunks[i] = fetch_chunk
copied_op = chunk.op.copy()
copied_op._key = chunk.op.key
out_chunks = [
c.data
for c in copied_op.new_chunks(
inp_chunks, kws=[c.params.copy() for c in chunk.op.outputs]
)
]

if input_changed:
copied_op = chunk.op.copy()
copied_op._key = chunk.op.key
out_chunks = [
c.data
for c in copied_op.new_chunks(
inp_chunks, kws=[c.params.copy() for c in chunk.op.outputs]
)
]
else:
out_chunks = chunk.op.outputs
# Note: `dtypes`, `index_value`, and `columns_value` are lazily
# initialized, so we should call property `params` to initialize
# these fields.
[o.params for o in out_chunks]
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's wired, what would happen without these codes?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There will no columns_value, index_value which are used in MainPool.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These field are lazily initialized, but field a or b are lazily initialized by params. Can you make a initialized by a, b initialized by b? Then we can lazily initialize them in Worker Main Pool.


for src_chunk, out_chunk in zip(chunk.op.outputs, out_chunks):
processed.add(src_chunk)
out_chunk._key = src_chunk.key
Expand Down