listen_and_serv_op support async update #10042
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -27,6 +27,38 @@ void RunServer(std::shared_ptr<detail::AsyncGRPCServer> service) { | |
| VLOG(4) << "RunServer thread end"; | ||
| } | ||
|
|
||
| static void split(const std::string &str, char sep, | ||
| std::vector<std::string> *pieces) { | ||
| pieces->clear(); | ||
| if (str.empty()) { | ||
| return; | ||
| } | ||
| size_t pos = 0; | ||
| size_t next = str.find(sep, pos); | ||
| while (next != std::string::npos) { | ||
| pieces->push_back(str.substr(pos, next - pos)); | ||
| pos = next + 1; | ||
| next = str.find(sep, pos); | ||
| } | ||
| if (!str.substr(pos).empty()) { | ||
| pieces->push_back(str.substr(pos)); | ||
| } | ||
| } | ||
|
|
||
| static void AsyncExecuteBlock(framework::Executor *executor, | ||
| framework::ExecutorPrepareContext *prepared, | ||
| framework::Scope *scope) { | ||
| std::future<void> future = framework::Async([&executor, &prepared, &scope]() { | ||
| try { | ||
| executor->RunPreparedContext(prepared, scope, false, false); | ||
| } catch (std::exception &e) { | ||
| LOG(ERROR) << "run sub program error " << e.what(); | ||
| } | ||
| }); | ||
| // TODO(qiao) maybe we can remove this | ||
|
There was a problem hiding this comment. removing this means more "async" mode, trainer even doesn't know whether the sent gradient is updated to the server side weights before it gets the latest weights. Or do you mean by letting updates to different weights become parallel? There was a problem hiding this comment. The current implementation will update gradients in sequence if we keep this wait. This may influence the effect, I will do some test on it. There was a problem hiding this comment. After discussing with @typhoonzero , we think that each gradient should be put to an independent block queue to ensure that they are updated without conflict. There was a problem hiding this comment.
Do you mean each gradient of one parameter, such as grad_w1(trainer0), grad_w1(trainer1), grad_w2(trainer0), we put grad_w1(trainer0) and grad_w1(trainer1) into a queue, and grad_w2(trainer0) into another one? According to the design doc, maybe we need multiple BlockingQueues so that each parameter can own one of them to implement a lock of updating parameter. There was a problem hiding this comment. yes, we need multiple block queue, each will store gradients for on parameters, but we do not need to add a lock, because the queue will block until the optimize block is finished. |
||
| future.wait(); | ||
| } | ||
|
|
||
| static void ParallelExecuteBlocks( | ||
| const std::vector<size_t> ¶llel_blkids, framework::Executor *executor, | ||
| const std::vector<std::shared_ptr<framework::ExecutorPrepareContext>> | ||
|
|
@@ -169,15 +201,85 @@ void ListenAndServOp::RunSyncLoop(framework::Executor *executor, | |
| } // while(true) | ||
| } | ||
|
|
||
| void ListenAndServOp::RunAsyncLoop(framework::Executor *executor, | ||
| framework::ProgramDesc *program, | ||
| framework::Scope *recv_scope, | ||
| framework::BlockDesc *prefetch_block) const { | ||
| VLOG(3) << "RunAsyncLoop in"; | ||
| // grad name to block id | ||
| std::unordered_map<std::string, int32_t> grad_to_id; | ||
| std::unordered_map<int32_t, std::string> id_to_grad; | ||
|
|
||
| auto grad_to_id_str = Attr<std::vector<std::string>>("grad_to_id"); | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. I think grad_to_id_str should be created in Python by transpiler because the transpile logic know how to split the operator and block, listen_and_serv_op just use the result is fine, or it has to understand the detailed logic of transpiler. There was a problem hiding this comment. Firstly we want to make In that case, for Async Execution, There was a problem hiding this comment. After discussing with @typhoonzero, I get the point, I totally agree with the idea that listen_and_serv_op should be a general operator! We will find a better way to implement async update in the future PRs. |
||
| for (auto &grad_and_id : grad_to_id_str) { | ||
| std::vector<std::string> pieces; | ||
| split(grad_and_id, ':', &pieces); | ||
| VLOG(3) << "after split, grad = " << pieces[0] << ", id=" << pieces[1]; | ||
| PADDLE_ENFORCE_EQ(pieces.size(), 2); | ||
| PADDLE_ENFORCE_EQ(grad_to_id.count(pieces[0]), 0); | ||
| int block_id = std::stoi(pieces[1]); | ||
| grad_to_id[pieces[0]] = block_id; | ||
| id_to_grad[block_id] = pieces[0]; | ||
| } | ||
| size_t num_blocks = program->Size(); | ||
| PADDLE_ENFORCE_GE(num_blocks, 2, | ||
| "server program should have at least 2 blocks"); | ||
|
|
||
| std::vector<int> block_list; | ||
| for (size_t blkid = 1; blkid < num_blocks; ++blkid) { | ||
| block_list.push_back(blkid); | ||
| } | ||
| auto optimize_prepared = executor->Prepare(*program, block_list); | ||
| std::unordered_map<std::string, | ||
| std::shared_ptr<framework::ExecutorPrepareContext>> | ||
| grad_to_prepared; | ||
|
There was a problem hiding this comment. grad_to_prepared_block |
||
| for (size_t i = 0; i < block_list.size(); ++i) { | ||
| grad_to_prepared[id_to_grad[block_list[i]]] = optimize_prepared[i]; | ||
| } | ||
|
|
||
| VLOG(3) << "RunAsyncLoop into while"; | ||
| bool exit_flag = false; | ||
| while (!exit_flag) { | ||
| const detail::ReceivedMessage v = rpc_service_->Get(); | ||
| auto recv_var_name = v.first; | ||
| if (recv_var_name == LISTEN_TERMINATE_MESSAGE) { | ||
| LOG(INFO) << "received terminate message and exit"; | ||
| exit_flag = true; | ||
| break; | ||
| } else { | ||
| VLOG(3) << "received grad: " << recv_var_name; | ||
| auto var = v.second->GetVar(); | ||
| if (var == nullptr) { | ||
| LOG(ERROR) << "Can not find server side var: " << recv_var_name; | ||
| PADDLE_THROW("Can not find server side var"); | ||
| } | ||
| AsyncExecuteBlock(executor, grad_to_prepared[recv_var_name].get(), | ||
| &(v.second->GetLocalScope())); | ||
| // TODO(qiao): explain why | ||
| if (var->IsType<framework::SelectedRows>()) { | ||
|
There was a problem hiding this comment. Maybe we don't need to clear the rows, because of each gradient var is in a new scope. There was a problem hiding this comment. Great suggestion! removed. |
||
| var->GetMutable<framework::SelectedRows>()->mutable_rows()->clear(); | ||
| } | ||
| } | ||
|
|
||
| if (exit_flag) { | ||
| rpc_service_->ShutDown(); | ||
| break; | ||
| } | ||
| } // while(true) | ||
| } | ||
|
|
||
| void ListenAndServOp::RunImpl(const framework::Scope &scope, | ||
| const platform::Place &dev_place) const { | ||
| platform::DeviceContextPool &pool = platform::DeviceContextPool::Instance(); | ||
| auto &dev_ctx = *pool.Get(dev_place); | ||
| framework::Scope &recv_scope = scope.NewScope(); | ||
|
|
||
| bool sync_mode = Attr<bool>("sync_mode"); | ||
|
|
||
| PADDLE_ENFORCE(!rpc_service_); | ||
| std::string endpoint = Attr<std::string>("endpoint"); | ||
|
|
||
| rpc_service_.reset(new detail::AsyncGRPCServer(endpoint, sync_mode)); | ||
|
|
||
| auto *optimize_block = Attr<framework::BlockDesc *>(kOptimizeBlock); | ||
| auto *prefetch_block = Attr<framework::BlockDesc *>(kPrefetchBlock); | ||
|
|
@@ -202,7 +304,11 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope, | |
| sleep(5); | ||
| // Write to a file of server selected port for python use. | ||
| SavePort(rpc_service_); | ||
| if (sync_mode) { | ||
| RunSyncLoop(&executor, program, &recv_scope, prefetch_block); | ||
| } else { | ||
| RunAsyncLoop(&executor, program, &recv_scope, prefetch_block); | ||
| } | ||
| } | ||
|
|
||
| class ListenAndServOpMaker : public framework::OpProtoAndCheckerMaker { | ||
|
|
@@ -221,6 +327,12 @@ from send_op and send back variables to recv_op. | |
| "IP address to listen on.") | ||
| .SetDefault("127.0.0.1:6164") | ||
| .AddCustomChecker([](const std::string &ip) { return !ip.empty(); }); | ||
| AddAttr<std::vector<std::string>>( | ||
| "grad_to_id", | ||
|
|
||
| "['param1@GRAD.block0:1', 'param2@GRAD.blockn:2'] " | ||
| "a map from grad name to it's optimize block id") | ||
| .SetDefault({}); | ||
| AddAttr<bool>("sync_mode", "if works at sync_mode or not").SetDefault(true); | ||
| AddAttr<framework::BlockDesc *>(kOptimizeBlock, | ||
| "BlockID to run on server side."); | ||
| AddAttr<framework::BlockDesc *>(kPrefetchBlock, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
maybe
is_syncor justsynccan tell the meaning?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think
sync_modemeans it works in a mode, butis_syncmeans itself is async. So I thinksync_modeis better.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I see.