def _create_cluster(self):
        if not self._fargate_scheduler or not self._fargate_workers:
            raise RuntimeError("You must specify a cluster when not using Fargate.")
        if self._worker_gpu:
            raise RuntimeError(
                "It is not possible to use GPUs with Fargate. "
                "Please provide an existing cluster with GPU instances available. "
            )
        self.cluster_name = dask.config.expand_environment_variables(
            self._cluster_name_template
        )
        self.cluster_name = self.cluster_name.format(uuid=str(uuid.uuid4())[:10])
        async with self._client("ecs") as ecs:
            response = await ecs.create_cluster(
                clusterName=self.cluster_name, tags=dict_to_aws(self.tags)
            )
        weakref.finalize(self, self.sync, self._delete_cluster)
        return response["cluster"]["clusterArn"] 

def __init__(self, suffix="", prefix=None, dir=None):
        if "RAM_DISK" in os.environ:
            import uuid

            name = uuid.uuid4().hex
            dir_name = os.path.join(os.environ["RAM_DISK"].strip(), name)
            os.mkdir(dir_name)
            self.name = dir_name
        else:
            suffix = suffix if suffix else ""
            if not prefix:
                self.name = mkdtemp(suffix=suffix, dir=dir)
            else:
                self.name = mkdtemp(suffix, prefix, dir)
        self._finalizer = finalize(
            self,
            self._cleanup,
            self.name,
            warn_message="Implicitly cleaning up {!r}".format(self),
        ) 

def __init__(self, value_ref: executor_pb2.ValueRef, type_spec, executor):
    """Creates the value.

    Args:
      value_ref: An instance of `executor_pb2.ValueRef` returned by the remote
        executor service.
      type_spec: An instance of `computation_types.Type`.
      executor: The executor that created this value.
    """
    py_typecheck.check_type(value_ref, executor_pb2.ValueRef)
    py_typecheck.check_type(type_spec, computation_types.Type)
    py_typecheck.check_type(executor, RemoteExecutor)
    self._value_ref = value_ref
    self._type_signature = type_spec
    self._executor = executor

    # Clean up the value and the memory associated with it on the remote
    # worker when no references to it remain.
    def finalizer(value_ref, executor):
      executor._dispose(value_ref)  # pylint: disable=protected-access

    weakref.finalize(self, finalizer, value_ref, executor) 

def test_all_freed(self):
        # we want a weakrefable subclass of weakref.finalize
        class MyFinalizer(weakref.finalize):
            pass

        a = self.A()
        res = []
        def callback():
            res.append(123)
        f = MyFinalizer(a, callback)

        wr_callback = weakref.ref(callback)
        wr_f = weakref.ref(f)
        del callback, f

        self.assertIsNotNone(wr_callback())
        self.assertIsNotNone(wr_f())

        del a
        self._collect_if_necessary()

        self.assertIsNone(wr_callback())
        self.assertIsNone(wr_f())
        self.assertEqual(res, [123]) 

def run_in_child(cls):
        def error():
            # Create an atexit finalizer from inside a finalizer called
            # at exit.  This should be the next to be run.
            g1 = weakref.finalize(cls, print, 'g1')
            print('f3 error')
            1/0

        # cls should stay alive till atexit callbacks run
        f1 = weakref.finalize(cls, print, 'f1', _global_var)
        f2 = weakref.finalize(cls, print, 'f2', _global_var)
        f3 = weakref.finalize(cls, error)
        f4 = weakref.finalize(cls, print, 'f4', _global_var)

        assert f1.atexit == True
        f2.atexit = False
        assert f3.atexit == True
        assert f4.atexit == True 

def __init__(self):
        super().__init__()

        self._state = MediaState.Null
        self._elements = []
        self._old_pipe = ''
        self._loop_count = 0

        self._gst_pipe = Gst.Pipeline()
        self._gst_state = Gst.State.NULL
        self._time_query = Gst.Query.new_position(Gst.Format.TIME)

        bus = self._gst_pipe.get_bus()
        bus.add_signal_watch()

        # Use a weakref instead of the method or the object will not be
        # garbage-collected
        on_message = weakref.WeakMethod(self.__on_message)
        handler = bus.connect('message', lambda *args: on_message()(*args))
        weakref.finalize(self, self.__finalizer, self._gst_pipe, handler,
                         self._elements)

        self.changed('loop').connect(self.__prepare_loops)
        self.changed('pipe').connect(self.__prepare_pipe) 

def test_all_freed(self):
        # we want a weakrefable subclass of weakref.finalize
        class MyFinalizer(weakref.finalize):
            pass

        a = self.A()
        res = []
        def callback():
            res.append(123)
        f = MyFinalizer(a, callback)

        wr_callback = weakref.ref(callback)
        wr_f = weakref.ref(f)
        del callback, f

        self.assertIsNotNone(wr_callback())
        self.assertIsNotNone(wr_f())

        del a
        self._collect_if_necessary()

        self.assertIsNone(wr_callback())
        self.assertIsNone(wr_f())
        self.assertEqual(res, [123]) 

def run_in_child(cls):
        def error():
            # Create an atexit finalizer from inside a finalizer called
            # at exit.  This should be the next to be run.
            g1 = weakref.finalize(cls, print, 'g1')
            print('f3 error')
            1/0

        # cls should stay alive till atexit callbacks run
        f1 = weakref.finalize(cls, print, 'f1', _global_var)
        f2 = weakref.finalize(cls, print, 'f2', _global_var)
        f3 = weakref.finalize(cls, error)
        f4 = weakref.finalize(cls, print, 'f4', _global_var)

        assert f1.atexit == True
        f2.atexit = False
        assert f3.atexit == True
        assert f4.atexit == True 

def test_no_linger(self):
        """Test that deleted animations are garbage-collected"""
        anim1_alivelist = ['animation 1 alive']
        anim2_alivelist = ['animation 2 alive']
        obj_alivelist = ['object alive']
        # cannot use SimpleNamespace because it doesn't support weakref
        obj = TestObject()
        obj.attribute = 0
        anim1 = animate(obj, attribute=1, duration=5)
        anim2 = animate(obj, attribute=2, duration=1)
        weakref.finalize(anim1, anim1_alivelist.clear)
        weakref.finalize(anim2, anim2_alivelist.clear)
        weakref.finalize(obj, obj_alivelist.clear)

        del anim1
        del anim2
        del obj
        gc.collect()
        self.assertEqual(anim1_alivelist, [])

        clock.tick(3)
        gc.collect()
        self.assertEqual(anim2_alivelist, [])
        self.assertEqual(obj_alivelist, []) 

def __init__(self, session=None, soup_config={'features': 'lxml'},
                 requests_adapters=None,
                 raise_on_404=False, user_agent=None):

        self.raise_on_404 = raise_on_404
        self.session = session or requests.Session()

        if hasattr(weakref, 'finalize'):
            self._finalize = weakref.finalize(self.session, self.close)
        else:   # pragma: no cover
            # Python < 3 does not have weakref.finalize, but these
            # versions accept calling session.close() within __del__
            self._finalize = self.close

        self.set_user_agent(user_agent)

        if requests_adapters is not None:
            for adaptee, adapter in requests_adapters.items():
                self.session.mount(adaptee, adapter)

        self.soup_config = soup_config or dict() 

def test_all_freed(self):
        # we want a weakrefable subclass of weakref.finalize
        class MyFinalizer(weakref.finalize):
            pass

        a = self.A()
        res = []
        def callback():
            res.append(123)
        f = MyFinalizer(a, callback)

        wr_callback = weakref.ref(callback)
        wr_f = weakref.ref(f)
        del callback, f

        self.assertIsNotNone(wr_callback())
        self.assertIsNotNone(wr_f())

        del a
        self._collect_if_necessary()

        self.assertIsNone(wr_callback())
        self.assertIsNone(wr_f())
        self.assertEqual(res, [123]) 

def run_in_child(cls):
        def error():
            # Create an atexit finalizer from inside a finalizer called
            # at exit.  This should be the next to be run.
            g1 = weakref.finalize(cls, print, 'g1')
            print('f3 error')
            1/0

        # cls should stay alive till atexit callbacks run
        f1 = weakref.finalize(cls, print, 'f1', _global_var)
        f2 = weakref.finalize(cls, print, 'f2', _global_var)
        f3 = weakref.finalize(cls, error)
        f4 = weakref.finalize(cls, print, 'f4', _global_var)

        assert f1.atexit == True
        f2.atexit = False
        assert f3.atexit == True
        assert f4.atexit == True 

def __init__(self, suffix="", prefix=None, dir=None):
        if "RAM_DISK" in os.environ:
            import uuid

            name = uuid.uuid4().hex
            dir_name = os.path.join(os.environ["RAM_DISK"].strip(), name)
            os.mkdir(dir_name)
            self.name = dir_name
        else:
            suffix = suffix if suffix else ""
            if not prefix:
                self.name = mkdtemp(suffix=suffix, dir=dir)
            else:
                self.name = mkdtemp(suffix, prefix, dir)
        self._finalizer = finalize(
            self,
            self._cleanup,
            self.name,
            warn_message="Implicitly cleaning up {!r}".format(self),
        ) 

def run(self, **options):
        self.shutdown()
        import threading
        options = combine_dicts(self.run_options, options)
        memo = os.environ.get("WERKZEUG_RUN_MAIN")
        try:
            os.environ["WERKZEUG_RUN_MAIN"] = "true"
            threading.Thread(
                target=run_server,
                args=(self.app(), self.get_port(**options))
            ).start()
            # noinspection PyArgumentList
            self.shutdown = weakref.finalize(self, self.shutdown_site, self.url)
            self.wait_server()
        finally:
            if memo is None:
                os.environ.pop("WERKZEUG_RUN_MAIN")
            else:
                os.environ["WERKZEUG_RUN_MAIN"] = memo

        return self 

def __init__(self, session=None, soup_config={'features': 'lxml'},
                 requests_adapters=None,
                 raise_on_404=False, user_agent=None):

        self.raise_on_404 = raise_on_404
        self.session = session or requests.Session()

        if hasattr(weakref, 'finalize'):
            self._finalize = weakref.finalize(self.session, self.close)
        else:   # pragma: no cover
            # Python < 3 does not have weakref.finalize, but these
            # versions accept calling session.close() within __del__
            self._finalize = self.close

        self.set_user_agent(user_agent)

        if requests_adapters is not None:
            for adaptee, adapter in requests_adapters.items():
                self.session.mount(adaptee, adapter)

        self.soup_config = soup_config or dict() 

def assertCalled(test, times=None):
    def decorator(fn):
        fn._called = 0

        def finalize(fn):
            if times is not None:
                test.assertEqual(fn._called, times, "Function '{}' was not called the correct number of times".format(fn.__name__))
            else:
                test.assertTrue(fn._called, "Function '{}' was never called".format(fn.__name__))

        weakref.finalize(fn, finalize, fn)

        @wraps(fn)
        def wrapper(*args, **kwargs):
            fn._called += 1
            fn(*args, **kwargs)

        return wrapper
    return decorator 

def __init__(self, url, engine_kwargs=None, skip_compatibility_check=False):
        # type: (str, Optional[Dict[str, Any]], bool) -> None

        self.engine_kwargs = engine_kwargs or {}
        self.url = self._fill_storage_url_template(url)
        self.skip_compatibility_check = skip_compatibility_check

        self._set_default_engine_kwargs_for_mysql(url, self.engine_kwargs)

        try:
            self.engine = create_engine(self.url, **self.engine_kwargs)
        except ImportError as e:
            raise ImportError(
                "Failed to import DB access module for the specified storage URL. "
                "Please install appropriate one. (The actual import error is: " + str(e) + ".)"
            )

        self.scoped_session = orm.scoped_session(orm.sessionmaker(bind=self.engine))
        models.BaseModel.metadata.create_all(self.engine)

        self._version_manager = _VersionManager(self.url, self.engine, self.scoped_session)
        if not skip_compatibility_check:
            self._version_manager.check_table_schema_compatibility()

        weakref.finalize(self, self._finalize) 

def __new__(cls, dir=False, **kwargs):
        """
        Create a tempfile, return pathlib.Path reference to it.
        """
        if dir:
            name = tempfile.mkdtemp(**kwargs)
        else:
            fd, name = tempfile.mkstemp(**kwargs)
            # fd is now assigned to our process table, but we don't need to do
            # anything with the file. We will call `open` on the `name` later
            # producing a different file descriptor, so close this one to
            # prevent a resource leak.
            os.close(fd)
        obj = super().__new__(cls, name)
        obj._destructor = weakref.finalize(obj, cls._destruct, str(obj))
        return obj 

def test_gc(ax):
    def inner():
        img = ax.imshow([[0, 1], [2, 3]])
        cursor = mplcursors.cursor(img)
        f_img = weakref.finalize(img, lambda: None)
        f_cursor = weakref.finalize(cursor, lambda: None)
        img.remove()
        return f_img, f_cursor
    f_img, f_cursor = inner()
    gc.collect()
    assert not f_img.alive
    assert not f_cursor.alive 

def __init__(self, suffix=None, prefix=None, dir=None):
        self.name = mkdtemp(suffix, prefix, dir)
        self._finalizer = _weakref.finalize(
            self, self._cleanup, self.name,
            warn_message="Implicitly cleaning up {!r}".format(self)) 

def _create_execution_role(self):
        async with self._client("iam") as iam:
            response = await iam.create_role(
                RoleName=self._execution_role_name,
                AssumeRolePolicyDocument="""{
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                        "Effect": "Allow",
                        "Principal": {
                            "Service": "ecs-tasks.amazonaws.com"
                        },
                        "Action": "sts:AssumeRole"
                        }
                    ]
                    }""",
                Description="A role for ECS to use when executing",
                Tags=dict_to_aws(self.tags, upper=True),
            )
            await iam.attach_role_policy(
                RoleName=self._execution_role_name,
                PolicyArn="arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly",
            )
            await iam.attach_role_policy(
                RoleName=self._execution_role_name,
                PolicyArn="arn:aws:iam::aws:policy/CloudWatchLogsFullAccess",
            )
            await iam.attach_role_policy(
                RoleName=self._execution_role_name,
                PolicyArn="arn:aws:iam::aws:policy/service-role/AmazonEC2ContainerServiceRole",
            )
            weakref.finalize(
                self, self.sync, self._delete_role, self._execution_role_name
            )
            return response["Role"]["Arn"] 

def _create_task_role(self):
        async with self._client("iam") as iam:
            response = await iam.create_role(
                RoleName=self._task_role_name,
                AssumeRolePolicyDocument="""{
                "Version": "2012-10-17",
                "Statement": [
                    {
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "ecs-tasks.amazonaws.com"
                    },
                    "Action": "sts:AssumeRole"
                    }
                ]
                }""",
                Description="A role for dask tasks to use when executing",
                Tags=dict_to_aws(self.tags, upper=True),
            )

            for policy in self._task_role_policies:
                await iam.attach_role_policy(
                    RoleName=self._task_role_name, PolicyArn=policy
                )

            weakref.finalize(self, self.sync, self._delete_role, self._task_role_name)
            return response["Role"]["Arn"] 

def _create_security_groups(self):
        async with self._client("ec2") as ec2:
            response = await ec2.create_security_group(
                Description="A security group for dask-ecs",
                GroupName=self.cluster_name,
                VpcId=self._vpc,
                DryRun=False,
            )

            await ec2.authorize_security_group_ingress(
                GroupId=response["GroupId"],
                IpPermissions=[
                    {
                        "IpProtocol": "TCP",
                        "FromPort": 8786,
                        "ToPort": 8787,
                        "IpRanges": [
                            {"CidrIp": "0.0.0.0/0", "Description": "Anywhere"}
                        ],
                        "Ipv6Ranges": [{"CidrIpv6": "::/0", "Description": "Anywhere"}],
                    },
                    {
                        "IpProtocol": "TCP",
                        "FromPort": 0,
                        "ToPort": 65535,
                        "UserIdGroupPairs": [{"GroupId": response["GroupId"]}],
                    },
                ],
                DryRun=False,
            )
            # await ec2.create_tags(
            #     Resources=[response["GroupId"]], Tags=dict_to_aws(self.tags, upper=True)
            #  )

            weakref.finalize(self, self.sync, self._delete_security_groups)
            return [response["GroupId"]] 

def reset_state(self):
        self._current_indent = 0
        self._finalize_buffer()
        self._code_buf = StringIO()
        self._code = None
        self._finalizer = finalize(self, self._finalize_buffer) 

def __init__(self, defer_atexit=False):
        """Constructor.

        :param bool defer_atexit: cleanup() to atexit instead of after garbage collection.
        """
        self.name = tempfile.mkdtemp('sphinxcontrib_versioning')
        if defer_atexit:
            atexit.register(shutil.rmtree, self.name, True)
            return
        try:
            weakref.finalize(self, shutil.rmtree, self.name, True)
        except AttributeError:
            weakref.proxy(self, functools.partial(shutil.rmtree, self.name, True)) 

def __init__(self, executor, *args, **kwargs):
    py_typecheck.check_type(executor, executor_base.Executor)
    super().__init__(*args, **kwargs)
    self._executor = executor
    self._lock = threading.Lock()

    # The keys in this dictionary are value ids (the same as what we return
    # in the gRPC responses), and the values are `concurrent.futures.Future`
    # instances (this may, and probably will change as we flesh out the rest
    # of this implementation).
    self._values = {}

    def run_loop(loop):
      loop.run_forever()
      loop.close()

    self._event_loop = asyncio.new_event_loop()
    self._event_loop.set_task_factory(
        tracing.propagate_trace_context_task_factory)
    self._thread = threading.Thread(
        target=functools.partial(run_loop, self._event_loop), daemon=True)
    self._thread.start()

    def finalize(loop, thread):
      loop.call_soon_threadsafe(loop.stop)
      thread.join()

    weakref.finalize(self, finalize, self._event_loop, self._thread) 

def __init__(self, target_executor: eb.Executor):
    """Creates a concurrent executor backed by a target executor.

    Args:
      target_executor: The executor that does all the work.
    """
    py_typecheck.check_type(target_executor, eb.Executor)
    self._target_executor = target_executor
    self._event_loop = asyncio.new_event_loop()
    self._event_loop.set_task_factory(
        tracing.propagate_trace_context_task_factory)

    def run_loop(loop):
      loop.run_forever()
      loop.close()

    self._thread = threading.Thread(
        target=functools.partial(run_loop, self._event_loop), daemon=True)
    self._thread.start()

    def finalizer(loop, thread):
      logging.debug('Finalizing, joining thread.')
      loop.call_soon_threadsafe(loop.stop)
      thread.join()
      logging.debug('Thread joined.')

    weakref.finalize(self, finalizer, self._event_loop, self._thread) 

def finalize(self):
        """Finalizes the object if not already done.

        Returns: None
        """
        # this is the "public" finalize method
        raise NotImplementedError(
            "finalize() must be implemented by AutoFinalizedObject."
        ) 

def __del__(self):
        self.finalize() 

def _do_finalize_object_ref(obj_ref):
        """Helper function for weakref.finalize() that dereferences a weakref
        to an object and calls its _do_finalize_object() method if the object
        is still alive. Does nothing otherwise.

        Returns: None (implicit)

        Arguments:
        * obj_ref -- weakref to an object
        """
        obj = obj_ref()
        if obj is not None:
            # else object disappeared
            obj._do_finalize_object()