tensorflow_io/core/kernels/gsmemcachedfs/memcached_file_system.cc
// A helper function to build a FileBlockCache for MemcachedGcsFileSystem.
A helper function to build a FileBlockCache for MemcachedGcsFileSystem.
A helper function to build a FileBlockCache for MemcachedGcsFileSystem .
std::unique_ptr<FileBlockCache> MemcachedGcsFileSystem::MakeFileBlockCache(
    size_t block_size, size_t max_bytes, uint64 max_staleness) {
  StringPiece client_cache_type;
  if (!GetEnvVar(kClientCacheType, StringPieceIdentity, &client_cache_type) ||
      !make_memcached_gcs_fs_cache_) {
    client_cache_type = kRamFileBlockCache;
  }
  VLOG(1) << "client_cache_type = " << client_cache_type;
  auto block_fetcher = [this](const string& fname, size_t offset, size_t n,
                              char* buffer, size_t* bytes_transferred) {
    return LoadBufferFromGCS(fname, offset, n, buffer, bytes_transferred);
  };
  VLOG(1) << "Creating " << client_cache_type;
  if (client_cache_type == kMemcachedFileBlockCache) {
    size_t client_pool_size = kDefaultMemcachedClientPoolSize;
    uint64 value;
    if (GetEnvVar(kMemcachedClientPoolSize, strings::safe_strtou64, &value)) {
      client_pool_size = value;
    }
    VLOG(1) << "Memcached client pool with " << client_pool_size << " clients.";
    memcached_clients_ =
        absl::make_unique<std::vector<MemcachedDaoInterface*>>();
    memcached_daos_ = absl::make_unique<
        std::vector<std::unique_ptr<MemcachedDaoInterface>>>();
    for (int i = 0; i < client_pool_size; ++i) {
      memcached_daos_->emplace_back(absl::make_unique<MemcachedDao>());
      memcached_clients_->push_back((*memcached_daos_)[i].get());
    }
    std::vector<string> servers;
    Status status = server_list_provider_->GetServerList(&servers);
    if (!status.ok()) {
      LOG(ERROR) << "Not able to get memcached server list. Will use a RAM "
                    "cache instead";
      std::unique_ptr<FileBlockCache> file_block_cache(new RamFileBlockCache(
          block_size, max_bytes, max_staleness, block_fetcher));
      return file_block_cache;
    }
    std::vector<string> options;
    if (GetEnvVar(kMemcachedOptions, SplitByCommaToVector, &options)) {
      VLOG(1) << "Override of memcached options";
    }
    size_t local_cache_size = 0;
    if (GetEnvVar(kMemcachedLocalCachesize, strings::safe_strtou64, &value)) {
      local_cache_size = value * 1024 * 1024 * 1024;
      VLOG(1) << "Distributed cache client has mini-reads cache of size = "
              << local_cache_size;
    }
    std::unique_ptr<FileBlockCache> file_block_cache(
        new MemcachedFileBlockCache(*memcached_clients_, block_size, max_bytes,
                                    max_staleness, local_cache_size, servers,
                                    options, block_fetcher));
    return file_block_cache;
  }
  if (client_cache_type == kNoneFileBlockCache) {
    std::unique_ptr<FileBlockCache> file_block_cache(
        new RamFileBlockCache(0, 0, max_staleness, block_fetcher));
    return file_block_cache;
  }
  if (client_cache_type != kRamFileBlockCache) {
    LOG(WARNING) << kClientCacheType << " set to unknown value \""
                 << client_cache_type << "\"; defaulting to "
                 << kRamFileBlockCache;
  }
  std::unique_ptr<FileBlockCache> file_block_cache(new RamFileBlockCache(
      block_size, max_bytes, max_staleness, block_fetcher));
  return file_block_cache;
}
std :: unique_ptr < FileBlockCache > MemcachedGcsFileSystem :: MakeFileBlockCache ( size_t block_size , size_t max_bytes , uint64 max_staleness ) { StringPiece client_cache_type ; if ( ! GetEnvVar ( kClientCacheType , StringPieceIdentity , & client_cache_type ) || ! make_memcached_gcs_fs_cache_ ) { client_cache_type = kRamFileBlockCache ; } VLOG ( 1 ) << " " << client_cache_type ; auto block_fetcher = [ this ] ( const string & fname , size_t offset , size_t n , char * buffer , size_t * bytes_transferred ) { return LoadBufferFromGCS ( fname , offset , n , buffer , bytes_transferred ) ; } ; VLOG ( 1 ) << " " << client_cache_type ; if ( client_cache_type == kMemcachedFileBlockCache ) { size_t client_pool_size = kDefaultMemcachedClientPoolSize ; uint64 value ; if ( GetEnvVar ( kMemcachedClientPoolSize , strings :: safe_strtou64 , & value ) ) { client_pool_size = value ; } VLOG ( 1 ) << " " << client_pool_size << " " ; memcached_clients_ = absl :: make_unique < std :: vector < MemcachedDaoInterface * > > ( ) ; memcached_daos_ = absl :: make_unique < std :: vector < std :: unique_ptr < MemcachedDaoInterface > > > ( ) ; for ( int i = 0 ; i < client_pool_size ; ++ i ) { memcached_daos_ -> emplace_back ( absl :: make_unique < MemcachedDao > ( ) ) ; memcached_clients_ -> push_back ( ( * memcached_daos_ ) [ i ] . get ( ) ) ; } std :: vector < string > servers ; Status status = server_list_provider_ -> GetServerList ( & servers ) ; if ( ! status . ok ( ) ) { LOG ( ERROR ) << " " " " ; std :: unique_ptr < FileBlockCache > file_block_cache ( new RamFileBlockCache ( block_size , max_bytes , max_staleness , block_fetcher ) ) ; return file_block_cache ; } std :: vector < string > options ; if ( GetEnvVar ( kMemcachedOptions , SplitByCommaToVector , & options ) ) { VLOG ( 1 ) << " " ; } size_t local_cache_size = 0 ; if ( GetEnvVar ( kMemcachedLocalCachesize , strings :: safe_strtou64 , & value ) ) { local_cache_size = value * 1024 * 1024 * 1024 ; VLOG ( 1 ) << " " << local_cache_size ; } std :: unique_ptr < FileBlockCache > file_block_cache ( new MemcachedFileBlockCache ( * memcached_clients_ , block_size , max_bytes , max_staleness , local_cache_size , servers , options , block_fetcher ) ) ; return file_block_cache ; } if ( client_cache_type == kNoneFileBlockCache ) { std :: unique_ptr < FileBlockCache > file_block_cache ( new RamFileBlockCache ( 0 , 0 , max_staleness , block_fetcher ) ) ; return file_block_cache ; } if ( client_cache_type != kRamFileBlockCache ) { LOG ( WARNING ) << kClientCacheType << " \" " << client_cache_type << " \" " << kRamFileBlockCache ; } std :: unique_ptr < FileBlockCache > file_block_cache ( new RamFileBlockCache ( block_size , max_bytes , max_staleness , block_fetcher ) ) ; return file_block_cache ; }
A helper function to build a FileBlockCache for MemcachedGcsFileSystem.
A helper function to build a FileBlockCache for MemcachedGcsFileSystem .
// If no cache type is specified, the RamFileBlockCache will be used // by default. // At this point, if the client_cache_type is something other than // the RamFileBlockCache, it means that the user has explicitly requested // an unsupported type.  Warn and continue with the default.