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

[xdoctest][task 232-235] reformat example code with google style in python/paddle/distributed/* #57591

Merged
merged 6 commits into from
Oct 10, 2023
Merged

[xdoctest][task 232-235] reformat example code with google style in python/paddle/distributed/* #57591

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d037416
[Doctest]fix No.232-235, test=docs_preview
jinyouzhi Sep 21, 2023
2453849
fix format
jinyouzhi Sep 21, 2023
d82c04c
add requires for rpc
jinyouzhi Sep 27, 2023
84810bf
fix typo
jinyouzhi Sep 27, 2023
2a44dfb
fix some
jinyouzhi Sep 28, 2023
d7c1bf4
fix upcase
jinyouzhi Sep 28, 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
107 changes: 60 additions & 47 deletions python/paddle/distributed/rpc/rpc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,13 @@ def init_rpc(name, rank=None, world_size=None, master_endpoint=None):
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc

rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8001")
rpc.shutdown()
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8001")

>>> rpc.shutdown()

"""
rank = int(os.environ["PADDLE_TRAINER_ID"]) if rank is None else rank
Expand Down Expand Up @@ -161,15 +163,17 @@ def rpc_sync(to, fn, args=None, kwargs=None, timeout=_DEFAULT_RPC_TIMEOUT):
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc

>>> def add(a, b):
... return a + b

def add(a, b):
return a + b
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8002")

rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8002")
ret = rpc.rpc_sync("worker0", add, args=(2, 3))
rpc.shutdown()
>>> ret = rpc.rpc_sync("worker0", add, args=(2, 3))
>>> rpc.shutdown()

"""
fut = _invoke_rpc(to, fn, args, kwargs, timeout)
Expand Down Expand Up @@ -201,16 +205,20 @@ def rpc_async(to, fn, args=None, kwargs=None, timeout=_DEFAULT_RPC_TIMEOUT):
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc

>>> def add(a, b):
... return a + b

def add(a, b):
return a + b
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8003")

rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8003")
fut = rpc.rpc_async("worker0", add, args=(2, 3))
print(fut.wait())
rpc.shutdown()
>>> fut = rpc.rpc_async("worker0", add, args=(2, 3))
>>> print(fut.wait())
5

>>> rpc.shutdown()

"""
return _invoke_rpc(to, fn, args, kwargs, timeout)
Expand Down Expand Up @@ -279,11 +287,13 @@ def shutdown():
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc

>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8004")

rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8004")
rpc.shutdown()
>>> rpc.shutdown()

"""
info = get_current_worker_info()
Expand All @@ -309,17 +319,18 @@ class `WorkerInfo` with attribute `name`, `rank`, `ip` and `port`.
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
import os
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc
>>> import os

os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9002"
rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8005")
>>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9002"
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8005")

print(rpc.get_worker_info("worker0"))
# {name: worker0, rank: 0, ip: 127.0.0.1, port: 9002}
>>> print(rpc.get_worker_info("worker0"))
{name: worker0, rank: 0, ip: 127.0.0.1, port: 9002}

rpc.shutdown()
>>> rpc.shutdown()

"""
return core.rpc_get_worker_info(name)
Expand All @@ -335,17 +346,18 @@ def get_all_worker_infos():
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
import os
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc
>>> import os

os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9003"
rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8006")
>>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9003"
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8006")

print(rpc.get_all_worker_infos())
# [{name: worker0, rank: 0, ip: 127.0.0.1, port: 9003}]
>>> print(rpc.get_all_worker_infos())
[{name: worker0, rank: 0, ip: 127.0.0.1, port: 9003}]

rpc.shutdown()
>>> rpc.shutdown()

"""
return core.rpc_get_all_worker_infos()
Expand All @@ -361,17 +373,18 @@ class `WorkerInfo` with attribute `name`, `rank`, `ip` and `port`.
Examples:
.. code-block:: python

import paddle.distributed.rpc as rpc
import os
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle.distributed.rpc as rpc
>>> import os

os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9004"
rpc.init_rpc("worker0", rank=0, world_size=1,
master_endpoint="127.0.0.1:8007")
>>> os.environ["PADDLE_WORKER_ENDPOINT"] = "127.0.0.1:9004"
>>> rpc.init_rpc("worker0", rank=0, world_size=1,
... master_endpoint="127.0.0.1:8007")

print(rpc.get_current_worker_info())
# {name: worker0, rank: 0, ip: 127.0.0.1, port: 9004}
>>> print(rpc.get_current_worker_info())
{name: worker0, rank: 0, ip: 127.0.0.1, port: 9004}

rpc.shutdown()
>>> rpc.shutdown()

"""
return core.rpc_get_current_worker_info()
86 changes: 44 additions & 42 deletions python/paddle/distributed/sharding/group_sharded.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,32 +77,33 @@ def group_sharded_parallel(
Examples:
.. code-block:: python

# required: distributed
import paddle
from paddle.nn import Linear
from paddle.distributed import fleet
from paddle.distributed.sharding import group_sharded_parallel
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle
>>> from paddle.nn import Linear
>>> from paddle.distributed import fleet
>>> from paddle.distributed.sharding import group_sharded_parallel

fleet.init(is_collective=True)
group = paddle.distributed.new_group([0, 1])
model = Linear(1000, 1000)
>>> fleet.init(is_collective=True)
>>> group = paddle.distributed.new_group([0, 1])
>>> model = Linear(1000, 1000)

clip = paddle.nn.ClipGradByGlobalNorm(clip_norm=1.0)
optimizer = paddle.optimizer.AdamW(learning_rate=0.001, parameters=model.parameters(), weight_decay=0.00001, grad_clip=clip)
>>> clip = paddle.nn.ClipGradByGlobalNorm(clip_norm=1.0)
>>> optimizer = paddle.optimizer.AdamW(learning_rate=0.001, parameters=model.parameters(), weight_decay=0.00001, grad_clip=clip)

# wrap sharding model, optimizer and scaler
model, optimizer, scaler = group_sharded_parallel(model, optimizer, "p_g", scaler=scaler)
>>> # wrap sharding model, optimizer and scaler
>>> model, optimizer, scaler = group_sharded_parallel(model, optimizer, "p_g", scaler=scaler)

img, label = data
label.stop_gradient = True
img.stop_gradient = True
>>> img, label = data
>>> label.stop_gradient = True
>>> img.stop_gradient = True

out = model(img)
loss = paddle.nn.functional.cross_entropy(input=out, label=label)
>>> out = model(img)
>>> loss = paddle.nn.functional.cross_entropy(input=out, label=label)

>>> loss.backward()
>>> optimizer.step()
>>> optimizer.clear_grad()

loss.backward()
optimizer.step()
optimizer.clear_grad()
"""

device = paddle.get_device().split(":")[0]
Expand Down Expand Up @@ -195,35 +196,36 @@ def save_group_sharded_model(model, output, optimizer=None):
Examples:
.. code-block:: python

# required: distributed
import paddle
from paddle.nn import Linear
from paddle.distributed import fleet
from paddle.distributed.sharding import group_sharded_parallel, save_group_sharded_model
>>> # doctest: +REQUIRES(env:DISTRIBUTED)
>>> import paddle
>>> from paddle.nn import Linear
>>> from paddle.distributed import fleet
>>> from paddle.distributed.sharding import group_sharded_parallel, save_group_sharded_model

>>> fleet.init(is_collective=True)
>>> group = paddle.distributed.new_group([0, 1])
>>> model = Linear(1000, 1000)

fleet.init(is_collective=True)
group = paddle.distributed.new_group([0, 1])
model = Linear(1000, 1000)
>>> clip = paddle.nn.ClipGradByGlobalNorm(clip_norm=1.0)
>>> optimizer = paddle.optimizer.AdamW(learning_rate=0.001, parameters=model.parameters(), weight_decay=0.00001, grad_clip=clip)

clip = paddle.nn.ClipGradByGlobalNorm(clip_norm=1.0)
optimizer = paddle.optimizer.AdamW(learning_rate=0.001, parameters=model.parameters(), weight_decay=0.00001, grad_clip=clip)
>>> # wrap sharding model, optimizer and scaler
>>> model, optimizer, scaler = group_sharded_parallel(model, optimizer, "p_g", scaler=scaler)

# wrap sharding model, optimizer and scaler
model, optimizer, scaler = group_sharded_parallel(model, optimizer, "p_g", scaler=scaler)
>>> img, label = data
>>> label.stop_gradient = True
>>> img.stop_gradient = True

img, label = data
label.stop_gradient = True
img.stop_gradient = True
>>> out = model(img)
>>> loss = paddle.nn.functional.cross_entropy(input=out, label=label)

out = model(img)
loss = paddle.nn.functional.cross_entropy(input=out, label=label)
>>> loss.backward()
>>> optimizer.step()
>>> optimizer.clear_grad()

loss.backward()
optimizer.step()
optimizer.clear_grad()
>>> # save model and optimizer state_dict
>>> save_group_sharded_model(model, optimizer, output=output_dir)

# save model and optimizer state_dict
save_group_sharded_model(model, optimizer, output=output_dir)
"""
logger_.info(
"==========Begin to save group sharded model and optimizer=========="
Expand Down
Loading