torch/lib/THD/base/data_channels/Store.cpp
/*
 * query communicates with the worker. The format
 * of the query is as follows:
 * type of query | size of arg1 | arg1 | size of arg2 | arg2 | ...
 * or, in the case of wait
 * type of query | number of args | size of arg1 | arg1 | ...
 */
query communicates with the worker.
query communicates with the worker .
void Store::StoreDeamon::query(rank_type rank) {
  int socket = _sockets[rank];
  QueryType qt;
  recv_bytes<QueryType>(socket, &qt, 1);
  if (qt == QueryType::SET) {
    std::string key = recv_string(socket);
    _store[key] = recv_vector<char>(socket);
    auto to_wake = _waiting.find(key);
    if (to_wake != _waiting.end()) {
      for (int proc : to_wake->second) {
        if (--_keys_awaited[proc] == 0)
          send_value<QueryType>(_sockets[proc], QueryType::STOP_WAITING);
      }
      _waiting.erase(to_wake);
    }
  } else if (qt == QueryType::GET) {
    std::string key = recv_string(socket);
    std::vector<char> data = _store.at(key);
    send_vector(socket, data);
  } else if (qt == QueryType::WAIT) {
    size_type nargs;
    recv_bytes<size_type>(socket, &nargs, 1);
    std::vector<std::string> keys(nargs);
    for (size_t i = 0; i < nargs; i++) {
      keys[i] = recv_string(socket);
    }
    if (checkAndUpdate(keys)) {
      send_value<QueryType>(socket, QueryType::STOP_WAITING);
    } else {
      for (auto& key : keys) {
        _waiting[key].push_back(rank);
      }
      _keys_awaited[rank] = keys.size();
    }
  } else {
    throw std::runtime_error("expected a query type");
  }
}
void Store :: StoreDeamon :: query ( rank_type rank ) { int socket = _sockets [ rank ] ; QueryType qt ; recv_bytes < QueryType > ( socket , & qt , 1 ) ; if ( qt == QueryType :: SET ) { std :: string key = recv_string ( socket ) ; _store [ key ] = recv_vector < char > ( socket ) ; auto to_wake = _waiting . find ( key ) ; if ( to_wake != _waiting . end ( ) ) { for ( int proc : to_wake -> second ) { if ( -- _keys_awaited [ proc ] == 0 ) send_value < QueryType > ( _sockets [ proc ] , QueryType :: STOP_WAITING ) ; } _waiting . erase ( to_wake ) ; } } else if ( qt == QueryType :: GET ) { std :: string key = recv_string ( socket ) ; std :: vector < char > data = _store . at ( key ) ; send_vector ( socket , data ) ; } else if ( qt == QueryType :: WAIT ) { size_type nargs ; recv_bytes < size_type > ( socket , & nargs , 1 ) ; std :: vector < std :: string > keys ( nargs ) ; for ( size_t i = 0 ; i < nargs ; i ++ ) { keys [ i ] = recv_string ( socket ) ; } if ( checkAndUpdate ( keys ) ) { send_value < QueryType > ( socket , QueryType :: STOP_WAITING ) ; } else { for ( auto & key : keys ) { _waiting [ key ] . push_back ( rank ) ; } _keys_awaited [ rank ] = keys . size ( ) ; } } else { throw std :: runtime_error ( " " ) ; } }
query communicates with the worker.
query communicates with the worker .
// On "set", wake up all of the processes that wait // for keys already in the store