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

Jump to bottom

Move cache_rpc_response from GrpcWorkerServiceOptions to ConfigProto.… #29379

Closed
wants to merge 1 commit into from
Closed

Move cache_rpc_response from GrpcWorkerServiceOptions to ConfigProto.… #29379

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
11 changes: 5 additions & 6 deletions tensorflow/core/distributed_runtime/rpc/grpc_worker_service.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,11 +370,6 @@ class GrpcWorkerService : public AsyncServiceInterface {
GrpcWorkerServiceOptions options)
: is_shutdown_(false) {
builder->RegisterService(&worker_service_);
// TODO(jingdong): it would be cleaner to move this option to GrpcWorker
// since the cache is maintained by GrpcWorker now.
if (options.cache_rpc_response) {
worker->EnableResponseCache();
}

for (int i = 0; i < options.num_serving_threads; i++) {
threads_.emplace_back(
Expand Down Expand Up @@ -428,7 +423,11 @@ GrpcWorker::GrpcWorker(WorkerEnv* worker_env, const ConfigProto& config)
recv_buf_max_chunk_(
config.experimental().recv_buf_max_chunk() > 0
? config.experimental().recv_buf_max_chunk()
: (config.experimental().recv_buf_max_chunk() < 0 ? 0 : 4096)) {}
: (config.experimental().recv_buf_max_chunk() < 0 ? 0 : 4096)) {
if (config.rpc_options().cache_rpc_response()) {
EnableResponseCache();
}
}

void GrpcWorker::EnableResponseCache() {
VLOG(1) << "Enabling gRPC tensor response cache.";
Expand Down
8 changes: 0 additions & 8 deletions tensorflow/core/distributed_runtime/rpc/grpc_worker_service.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,14 +75,6 @@ struct GrpcWorkerServiceOptions {
// default queue depth for a method.
std::unordered_map<int, int> queue_depth;
int num_serving_threads = 8;

// Setting cache_rpc_response to true will enable sender side caching of
// response for RecvTensorAsync and RecvBufAsync to allow receiver to retry
// requests . This is only necessary when the network fabric is experiencing a
// significant error rate. Without it we'll fail a step on an network error,
// while with it we'll be able to complete long steps (like complex
// initializations) in the face of some network errors during RecvTensor.
bool cache_rpc_response = false;
};

// Returns an implementation of WorkerService rpc service.
Expand Down
8 changes: 8 additions & 0 deletions tensorflow/core/protobuf/config.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,6 +327,14 @@ message RPCOptions {
// If compression_algorithm is set, the compression level to be used.
// From 0 (no compression), up to 3.
int32 compression_level = 3;

// Setting cache_rpc_response to true will enable sender side caching of
// response for RecvTensorAsync and RecvBufAsync to allow receiver to retry
// requests . This is only necessary when the network fabric is experiencing a
// significant error rate. Without it we'll fail a step on an network error,
// while with it we'll be able to complete long steps (like complex
// initializations) in the face of some network errors during RecvTensor.
bool cache_rpc_response = 4;
}

// Session configuration parameters.
Expand Down