code: def flush_redis_unsafe(redis_client=None):
      
      if redis_client is None:
          ray.worker.global_worker.check_connected()
          redis_client = ray.worker.global_worker.redis_client
  
      # Delete the log files from the primary Redis shard.
      keys = redis_client.keys("LOGFILE:*")
      if len(keys) > 0:
          num_deleted = redis_client.delete(*keys)
      else:
          num_deleted = 0
      print("Deleted {} log files from Redis.".format(num_deleted))
  
      # Delete the event log from the primary Redis shard.
      keys = redis_client.keys("event_log:*")
      if len(keys) > 0:
          num_deleted = redis_client.delete(*keys)
      else:
          num_deleted = 0
      print("Deleted {} event logs from Redis.".format(num_deleted))
docs: This removes some non-critical state from the primary Redis shard.
  
      This removes the log files as well as the event log from Redis. This can
      be used to try to address out-of-memory errors caused by the accumulation
      of metadata in Redis. However, it will only partially address the issue as
      much of the data is in the task table (and object table), which are not
      flushed.
  
      Args:
        redis_client: optional, if not provided then ray.init() must have been
          called.