code: def test_cancel_with_dependencies_in_memory(
      ws, release_dep, done_ev_cls
  ):
      """Cancel an executing task y with an in-memory dependency x; then simulate that x
      did not have any further dependents, so cancel x as well.
  
      Test that x immediately transitions to released state and is forgotten as soon as
      y finishes computing.
  
      Read: https://github.com/dask/distributed/issues/6893"""
      ws.handle_stimulus(
          UpdateDataEvent(data={"x": 1}, stimulus_id="s1"),
          ComputeTaskEvent.dummy(
              "y", who_has={"x": [ws.address]}, stimulus_id="s2"
          ),
      )
      assert ws.tasks["x"].state == "memory"
      assert ws.tasks["y"].state == "executing"
  
      ws.handle_stimulus(FreeKeysEvent(keys=["y"], stimulus_id="s3"))
      assert ws.tasks["x"].state == "memory"
      assert ws.tasks["y"].state == "cancelled"
  
      if release_dep:
          # This will happen iff x has no dependents or waiters on the scheduler
          ws.handle_stimulus(
              FreeKeysEvent(keys=["x"], stimulus_id="s4")
          )
          assert ws.tasks["x"].state == "released"
          assert ws.tasks["y"].state == "cancelled"
  
          ws.handle_stimulus(done_ev_cls.dummy("y", stimulus_id="s5"))
          assert "y" not in ws.tasks
          assert "x" not in ws.tasks
      else:
          ws.handle_stimulus(done_ev_cls.dummy("y", stimulus_id="s5"))
          assert "y" not in ws.tasks
          assert ws.tasks["x"].state == "memory"
  
imports: from __future__ import annotations
  import asyncio
  import pytest
  import distributed
  from distributed import Event, Lock, Worker
  from distributed.client import wait
  from distributed.utils_test import BlockedExecute, BlockedGatherDep, BlockedGetData, _LockedCommPool, assert_story, async_poll_for, freeze_batched_send, gen_cluster, inc, lock_inc, wait_for_state, wait_for_stimulus
  from distributed.worker_state_machine import AddKeysMsg, ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, GatherDep, GatherDepFailureEvent, GatherDepNetworkFailureEvent, GatherDepSuccessEvent, LongRunningMsg, ReleaseWorkerDataMsg, RemoveReplicasEvent, RescheduleEvent, SecedeEvent, TaskFinishedMsg, UpdateDataEvent
label_desc: Use image node_id 12 for calling a global function with example usage: test_cancel_with_dependencies_in_memory(ws, release_dep, done_ev_cls) without return types