code: def _maybe_flush_gcs(self):
          
          if not self.issue_gcs_flushes:
              return
          if self.gcs_flush_policy is None:
              serialized = self.redis.get("gcs_flushing_policy")
              if serialized is None:
                  # Client has not set any policy; by default flushing is off.
                  return
              self.gcs_flush_policy = pickle.loads(serialized)
  
          if not self.gcs_flush_policy.should_flush(self.redis_shard):
              return
  
          max_entries_to_flush = self.gcs_flush_policy.num_entries_to_flush()
          num_flushed = self.redis_shard.execute_command(
              "HEAD.FLUSH {}".format(max_entries_to_flush))
          logger.info("Monitor: num_flushed {}".format(num_flushed))
  
          # This flushes event log and log files.
          ray.experimental.flush_redis_unsafe(self.redis)
  
          self.gcs_flush_policy.record_flush()
docs: Experimental: issue a flush request to the GCS.
  
          The purpose of this feature is to control GCS memory usage.
  
          To activate this feature, Ray must be compiled with the flag
          RAY_USE_NEW_GCS set, and Ray must be started at run time with the flag
          as well.