class_docstr: Controls an object on a remote worker
  
  An actor allows remote control of a stateful object living on a remote
  worker.  Method calls on this object trigger operations on the remote
  object and return BaseActorFutures on which we can block to get results.
  
  Examples
  --------
  >>> class Counter:
  ...    def __init__(self):
  ...        self.n = 0
  ...    def increment(self):
  ...        self.n += 1
  ...        return self.n
  
  >>> from dask.distributed import Client
  >>> client = Client()
  
  You can create an actor by submitting a class with the keyword
  ``actor=True``.
  
  >>> future = client.submit(Counter, actor=True)
  >>> counter = future.result()
  >>> counter
  <Actor: Counter, key=Counter-1234abcd>
  
  Calling methods on this object immediately returns deferred ``BaseActorFuture``
  objects.  You can call ``.result()`` on these objects to block and get the
  result of the function call.
  
  >>> future = counter.increment()
  >>> future.result()
  1
  >>> future = counter.increment()
  >>> future.result()
  2
class_docstr_tok: ["Controls","an","object","on","a","remote","worker","An","actor","allows","remote","control","of","a","stateful","object","living","on","a","remote","worker",".","Method","calls","on","this","object","trigger","operations","on","the","remote","object","and","return","BaseActorFutures","on","which","we","can","block","to","get","results",".","Examples","--","--","--","--",">",">",">","class","Counter",":","...","def","__init__","(","self",")",":","...","self.n","=","0","...","def","increment","(","self",")",":","...","self.n","+=","1","...","return","self.n",">",">",">","from","dask.distributed","import","Client",">",">",">","client","=","Client","(",")","You","can","create","an","actor","by","submitting","a","class","with","the","keyword","``","actor=True","``",".",">",">",">","future","=","client.submit","(","Counter",",","actor=True",")",">",">",">","counter","=","future.result","(",")",">",">",">","counter","<","Actor",":","Counter",",","key=Counter-1234abcd",">","Calling","methods","on","this","object","immediately","returns","deferred","``","BaseActorFuture","``","objects",".","You","can","call","``",".result","(",")","``","on","these","objects","to","block","and","get","the","result","of","the","function","call",".",">",">",">","future","=","counter.increment","(",")",">",">",">","future.result","(",")","1",">",">",">","future","=","counter.increment","(",")",">",">",">","future.result","(",")","2"]
code: def __getattr__(self, key):
      if self._future and self._future.status not in (
          "finished",
          "pending",
      ):
          raise ValueError(
              "Worker holding Actor was lost.  Status: "
              + self._future.status
          )
      self._try_bind_worker_client()
      if (
          self._worker
          and self._worker.address == self._address
          and getattr(thread_state, "actor", False)
      ):
          # actor calls actor on same worker
          actor = self._worker.actors[self.key]
          attr = getattr(actor, key)
  
          if iscoroutinefunction(attr):
              return attr
  
          elif callable(attr):
              return lambda *args, **kwargs: EagerActorFuture(
                  attr(*args, **kwargs)
              )
          else:
              return attr
  
      attr = getattr(self._cls, key)
  
      if callable(attr):
  
          @functools.wraps(attr)
          def func(*args, **kwargs):
              async def run_actor_function_on_worker():
                  try:
                      result = await self._worker_rpc.actor_execute(
                          function=key,
                          actor=self.key,
                          args=[to_serialize(arg) for arg in args],
                          kwargs={
                              k: to_serialize(v)
                              for k, v in kwargs.items()
                          },
                      )
                  except OSError:
                      if self._future and not self._future.done():
                          await self._future
                          return await run_actor_function_on_worker()
                      else:
                          exc = OSError(
                              "Unable to contact Actor's worker"
                          )
                          return _Error(exc)
                  if result["status"] == "OK":
                      return _OK(result["result"])
                  return _Error(result["exception"])
  
              actor_future = ActorFuture(io_loop=self._io_loop)
  
              async def wait_then_set_result():
                  actor_future._set_result(
                      await run_actor_function_on_worker()
                  )
  
              self._io_loop.add_callback(wait_then_set_result)
              return actor_future
  
          return func
  
      else:
  
          async def get_actor_attribute_from_worker():
              x = await self._worker_rpc.actor_attribute(
                  attribute=key, actor=self.key
              )
              if x["status"] == "OK":
                  return x["result"]
              else:
                  raise x["exception"]
  
          return self._sync(get_actor_attribute_from_worker)
  
imports: from __future__ import annotations
  import abc
  import functools
  import threading
  from collections.abc import Awaitable, Generator
  from dataclasses import dataclass
  from datetime import timedelta
  from typing import Generic, Literal, NoReturn, TypeVar
  from tornado.ioloop import IOLoop
  from distributed.client import Future
  from distributed.protocol import to_serialize
  from distributed.utils import LateLoopEvent, iscoroutinefunction, sync, thread_state
  from distributed.utils_comm import WrappedKey
  from distributed.worker import get_client, get_worker
label_desc: Use image node_id 11 for calling the Actor obj's underlying member method code with example usage: obj.__getattr__(key) and returns: func, self, attr, actor_future, function, attr, _Error, x, _OK, unknown, _Error