code: def test_worker_actor_handle_is_weakref_from_compute_sync(client):
      workers = list(client.run(lambda: None))
  
      with dask.annotate(workers=workers[0]):
          counter = dask.delayed(Counter)()
      with dask.annotate(workers=workers[1]):
          intermediate = dask.delayed(lambda c: None)(counter)
      with dask.annotate(workers=workers[0]):
          final = dask.delayed(lambda x, c: x)(intermediate, counter)
  
      final.compute(actors=counter, optimize_graph=False)
  
      def worker_tasks_running(dask_worker):
          return len(dask_worker.data) + len(dask_worker.actors)
  
      start = time()
      while any(client.run(worker_tasks_running).values()):
          sleep(0.01)
          assert time() < start + 30
  
imports: from __future__ import annotations
  import asyncio
  import operator
  from time import sleep
  import pytest
  import dask
  from distributed import Actor, BaseActorFuture, Client, Future, Nanny, as_completed, get_client, wait, worker_client
  from distributed.metrics import time
  from distributed.utils import LateLoopEvent
  from distributed.utils_test import cluster, double, gen_cluster, inc
  from distributed.worker import get_worker
label_desc: Use image node_id 5 for calling a global function with example usage: test_worker_actor_handle_is_weakref_from_compute_sync(client) and returns: unknown