lint

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>

Author: alexeykudinkin

python/ray/data/_internal/execution/resource_manager.py

@@ -781,8 +781,8 @@ def max_task_output_bytes_to_read(
             return None
         res = self._op_budgets[op].object_store_memory
         # Add the remaining of `_reserved_for_op_outputs`.
-        op_outputs_usage = self._resource_manager.get_op_outputs_object_store_usage_with_downstream(
-            op
+        op_outputs_usage = (
+            self._resource_manager.get_op_outputs_object_store_usage_with_downstream(op)
         )
 
         res += max(self._reserved_for_op_outputs[op] - op_outputs_usage, 0)
@@ -824,10 +824,14 @@ def update_budgets(
             op_mem_usage = 0
             # Add the memory usage of the operator itself,
             # excluding `_reserved_for_op_outputs`.
-            op_mem_usage += self._resource_manager.get_op_internal_object_store_usage(op)
+            op_mem_usage += self._resource_manager.get_op_internal_object_store_usage(
+                op
+            )
             # Add the portion of op outputs usage that has
             # exceeded `_reserved_for_op_outputs`.
-            op_outputs_usage = self._resource_manager.get_op_outputs_object_store_usage_with_downstream(op)
+            op_outputs_usage = self._resource_manager.get_op_outputs_object_store_usage_with_downstream(
+                op
+            )
             op_mem_usage += max(op_outputs_usage - self._reserved_for_op_outputs[op], 0)
 
             op_usage = self._resource_manager.get_op_usage(op).copy(

python/ray/data/tests/test_autoscaler.py

@@ -68,7 +68,9 @@ def test_actor_pool_scaling():
         input_dependencies=[MagicMock()],
         internal_queue_size=MagicMock(return_value=1),
     )
-    op_state = OpState(op, inqueues=[MagicMock(__len__=MagicMock(return_value=10), num_blocks=10)])
+    op_state = OpState(
+        op, inqueues=[MagicMock(__len__=MagicMock(return_value=10), num_blocks=10)]
+    )
     op_state._scheduling_status = MagicMock(under_resource_limits=True)
 
     @contextmanager

python/ray/data/tests/test_resource_manager.py

@@ -604,9 +604,6 @@ def test_reserve_incremental_resource_usage(self, restore_data_context):
         allocator = resource_manager._op_resource_allocator
         assert isinstance(allocator, ReservationOpResourceAllocator)
 
-        op_usages = {op: ExecutionResources.zero() for op in [o1, o2, o3, o4, o5]}
-        op_internal_usage = dict.fromkeys([o1, o2, o3, o4, o5], 0)
-        op_outputs_usages = dict.fromkeys([o1, o2, o3, o4, o5], 0)
         allocator.update_budgets(
             limits=global_limits,
         )
@@ -663,9 +660,6 @@ def test_reserve_min_resources_for_gpu_ops(
         allocator = resource_manager._op_resource_allocator
         assert isinstance(allocator, ReservationOpResourceAllocator)
 
-        op_usages = {op: ExecutionResources.zero() for op in [o1, o2]}
-        op_internal_usage = dict.fromkeys([o1, o2], 0)
-        op_outputs_usages = dict.fromkeys([o1, o2], 0)
         allocator.update_budgets(
             limits=global_limits,
         )
@@ -698,9 +692,6 @@ def test_does_not_reserve_more_than_max_resource_usage(self):
         )
         allocator = resource_manager._op_resource_allocator
 
-        op_usages = {op: ExecutionResources.zero() for op in [o1, o2]}
-        op_internal_usage = dict.fromkeys([o1, o2], 0)
-        op_outputs_usages = dict.fromkeys([o1, o2], 0)
         global_limits = resource_manager.get_global_limits()
         allocator.update_budgets(
             limits=global_limits,
@@ -736,9 +727,6 @@ def test_only_handle_eligible_ops(self, restore_data_context):
         allocator = resource_manager._op_resource_allocator
         assert isinstance(allocator, ReservationOpResourceAllocator)
 
-        op_usages = {op: ExecutionResources.zero() for op in [o1, o2, o3]}
-        op_internal_usage = dict.fromkeys([o1, o2, o3], 0)
-        op_outputs_usages = dict.fromkeys([o1, o2, o3], 0)
         global_limits = resource_manager.get_global_limits()
         allocator.update_budgets(
             limits=global_limits,

python/ray/data/tests/test_stats.py

@@ -239,10 +239,7 @@ def gen_expected_metrics(
                 "'block_completion_time': "
                 f"{gen_histogram_values(histogram_buckets_s, 'N')}"
             ),
-            (
-                "'task_completion_time_s': "
-                f"{'N' if task_backpressure else 'Z'}"
-            ),
+            ("'task_completion_time_s': " f"{'N' if task_backpressure else 'Z'}"),
             (
                 "'task_completion_time_excl_backpressure_s': "
                 f"{'N' if task_backpressure else 'Z'}"