Home | History | Annotate | Download | only in gdr 
      1 /* Copyright 2017 The TensorFlow Authors. All Rights Reserved.
      2 
      3 Licensed under the Apache License, Version 2.0 (the "License");
      4 you may not use this file except in compliance with the License.
      5 You may obtain a copy of the License at
      6 
      7     http://www.apache.org/licenses/LICENSE-2.0
      8 
      9 Unless required by applicable law or agreed to in writing, software
     10 distributed under the License is distributed on an "AS IS" BASIS,
     11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     12 See the License for the specific language governing permissions and
     13 limitations under the License.
     14 ==============================================================================*/
     15 
     16 #include "tensorflow/contrib/gdr/gdr_server_lib.h"
     17 
     18 #include "grpc/support/alloc.h"
     19 #include "tensorflow/contrib/gdr/gdr_memory_manager.h"
     20 #include "tensorflow/contrib/gdr/gdr_rendezvous_mgr.h"
     21 #include "tensorflow/contrib/gdr/gdr_worker.h"
     22 
     23 #include "grpc/support/alloc.h"
     24 
     25 namespace tensorflow {
     26 
     27 GdrServer::GdrServer(const ServerDef& server_def, Env* env)
     28     : GrpcServer(server_def, env) {
     29   string host;
     30   string port;
     31   for (const auto& job : server_def.cluster().job()) {
     32     if (job.name() == server_def.job_name()) {
     33       auto iter = job.tasks().find(server_def.task_index());
     34       if (iter != job.tasks().end()) {
     35         const std::vector<string> hostname_port =
     36             str_util::Split(iter->second, ':');
     37         if (hostname_port.size() == 2) {
     38           host = hostname_port[0];
     39           port = hostname_port[1];
     40         }
     41       }
     42     }
     43   }
     44   remote_memory_manager_ = std::unique_ptr<RemoteMemoryManager>(
     45       CreateRemoteMemoryManager(host, port));
     46 }
     47 
     48 GdrServer::~GdrServer() {}
     49 
     50 Status GdrServer::Init() {
     51   RendezvousMgrCreationFunction rendezvous_mgr_func =
     52       [this](const WorkerEnv* env) {
     53         return new GdrRendezvousMgr(env, remote_memory_manager_.get());
     54       };
     55   WorkerCreationFunction worker_func = [this](WorkerEnv* env) {
     56     return std::unique_ptr<GdrWorker>(
     57         new GdrWorker(env, remote_memory_manager_.get()));
     58   };
     59   TF_RETURN_IF_ERROR(
     60       GrpcServer::Init(nullptr, rendezvous_mgr_func, worker_func));
     61 
     62   return remote_memory_manager_->Init();
     63 }
     64 
     65 Status GdrServer::Start() {
     66   {
     67     mutex_lock l(mu_);
     68     gdr_thread_.reset(worker_env()->env->StartThread(
     69         ThreadOptions(), "TF_gdr_service",
     70         [this] { remote_memory_manager_->Run(); }));
     71   }
     72   return GrpcServer::Start();
     73 }
     74 
     75 Status GdrServer::Stop() {
     76   TF_RETURN_IF_ERROR(GrpcServer::Stop());
     77   remote_memory_manager_->Stop();
     78   return Status::OK();
     79 }
     80 
     81 Status GdrServer::Join() {
     82   {
     83     mutex_lock l(mu_);
     84     gdr_thread_.reset();
     85   }
     86   return GrpcServer::Join();
     87 }
     88 
     89 /* static */
     90 Status GdrServer::Create(const ServerDef& server_def, Env* env,
     91                          std::unique_ptr<ServerInterface>* out_server) {
     92   std::unique_ptr<GdrServer> ret(
     93       new GdrServer(server_def, env == nullptr ? Env::Default() : env));
     94   TF_RETURN_IF_ERROR(ret->Init());
     95   *out_server = std::move(ret);
     96   return Status::OK();
     97 }
     98 
     99 namespace {
    100 
    101 class GdrServerFactory : public ServerFactory {
    102  public:
    103   bool AcceptsOptions(const ServerDef& server_def) override {
    104     return server_def.protocol() == "grpc+gdr";
    105   }
    106 
    107   Status NewServer(const ServerDef& server_def,
    108                    std::unique_ptr<ServerInterface>* out_server) override {
    109     return GdrServer::Create(server_def, Env::Default(), out_server);
    110   }
    111 };
    112 
    113 // Registers a `ServerFactory` for `GdrServer` instances.
    114 class GdrServerRegistrar {
    115  public:
    116   GdrServerRegistrar() {
    117     gpr_allocation_functions alloc_fns;
    118     memset(&alloc_fns, 0, sizeof(alloc_fns));
    119     alloc_fns.malloc_fn = port::Malloc;
    120     alloc_fns.realloc_fn = port::Realloc;
    121     alloc_fns.free_fn = port::Free;
    122     gpr_set_allocation_functions(alloc_fns);
    123     ServerFactory::Register("GDR_SERVER", new GdrServerFactory());
    124   }
    125 };
    126 static GdrServerRegistrar registrar;
    127 
    128 }  // namespace
    129 }  // namespace tensorflow
    130