class_docstr: Tests for StreamOutputBackpressurePolicy.
class_docstr_tok: ["Tests","for","StreamOutputBackpressurePolicy","."]
docstr: Test no deadlock in case of resource contention from
  non-Data code.
docstr_tok: ["Test","no","deadlock","in","case","of","resource","contention","from","non-Data","code","."]
code: def test_no_deadlock_for_resource_contention(self):
      """Test no deadlock in case of resource contention from
      non-Data code."""
      # Create a non-Data actor that uses 4 CPUs, only 1 CPU
      # is left for Data. Currently Data StreamExecutor still
      # incorrectly assumes it has all the 5 CPUs.
      # Check that we don't deadlock in this case.
  
      @ray.remote(num_cpus=4)
      class DummyActor:
          def foo(self):
              return None
  
      dummy_actor = DummyActor.remote()
      ray.get(dummy_actor.foo.remote())
  
      producer_timestamps, consumer_timestamps = self._run_dataset(
          producer_num_cpus=1, consumer_num_cpus=0.9
      )
      assert producer_timestamps[-1] < consumer_timestamps[0], (
          producer_timestamps,
          consumer_timestamps,
      )
  
imports: import functools
  import time
  import unittest
  from collections import defaultdict
  from contextlib import contextmanager
  from unittest.mock import MagicMock, patch
  import numpy
  import pytest
  import ray
  from ray.data._internal.execution.backpressure_policy import ENABLED_BACKPRESSURE_POLICIES_CONFIG_KEY, ConcurrencyCapBackpressurePolicy, StreamingOutputBackpressurePolicy
  from ray.data.tests.conftest import restore_data_context
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import shutdown_only
label_desc: Use image node_id 7 for calling the TestStreamOutputBackpressurePolicy obj's underlying member method code with example usage: obj.test_no_deadlock_for_resource_contention() and returns: None