def flush_task_and_object_metadata_unsafe()

in forge/blade/lib/ray.py [0:0]

• 21 lines of code
• 6 McCabe index (conditional complexity)

def flush_task_and_object_metadata_unsafe():
   """This removes some critical state from the Redis shards.

   This removes all of the object and task metadata. This can be used to try
   to address out-of-memory errors caused by the accumulation of metadata in
   Redis. However, after running this command, fault tolerance will most
   likely not work.
   """
   if not hasattr(ray.worker.global_worker, "redis_client"):
       raise Exception("ray.experimental.flush_redis_unsafe cannot be called "
                       "before ray.init() has been called.")

   def flush_shard(redis_client):
       num_task_keys_deleted = 0
       for key in redis_client.scan_iter(match=TASK_TABLE_PREFIX + b"*"):
           num_task_keys_deleted += redis_client.delete(key)
       print("Deleted {} task keys from Redis.".format(num_task_keys_deleted))

       num_object_keys_deleted = 0
       for key in redis_client.scan_iter(match=OBJECT_INFO_PREFIX + b"*"):
           num_object_keys_deleted += redis_client.delete(key)
       print("Deleted {} object info keys from Redis.".format(
                 num_object_keys_deleted))

       num_object_location_keys_deleted = 0
       for key in redis_client.scan_iter(match=OBJECT_LOCATION_PREFIX + b"*"):
           num_object_location_keys_deleted += redis_client.delete(key)
       print("Deleted {} object location keys from Redis.".format(
                 num_object_location_keys_deleted))

   for redis_client in ray.worker.global_state.redis_clients:
       flush_shard(redis_client)