def __init__(self, bot: Bot):
        super().__init__()

        self.bot = bot

        # Categories
        self.available_category: discord.CategoryChannel = None
        self.in_use_category: discord.CategoryChannel = None
        self.dormant_category: discord.CategoryChannel = None

        # Queues
        self.channel_queue: asyncio.Queue[discord.TextChannel] = None
        self.name_queue: t.Deque[str] = None

        self.name_positions = self.get_names()
        self.last_notification: t.Optional[datetime] = None

        # Asyncio stuff
        self.queue_tasks: t.List[asyncio.Task] = []
        self.ready = asyncio.Event()
        self.on_message_lock = asyncio.Lock()
        self.init_task = self.bot.loop.create_task(self.init_cog()) 

def start(self) -> None:
        """(async) Start the PT executable and wait until it's ready.

        "Ready" means that all transports have finished initializing.
        """
        self._check_not_started()
        await self._pre_start()
        env = self._build_env()
        self._logger.debug('PT environment variables: %r', env)
        self._process = await asyncio.create_subprocess_exec(
            *self._pt_args,
            env=env,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=None,
        )
        self._logger.debug('Started PT subprocess: %r', self._process)
        self._stdout_task = asyncio.create_task(self._process_stdout())
        try:
            await self._ready
        except Exception:
            await self.stop()
            raise 

def test_increment_lock(self):
        """Test that we can't produce a race condition in .increment."""
        await self.cog.redis.set("test_key", 0)
        tasks = []

        # Increment this a lot in different tasks
        for _ in range(100):
            task = asyncio.create_task(
                self.cog.redis.increment("test_key", 1)
            )
            tasks.append(task)
        await asyncio.gather(*tasks)

        # Confirm that the value has been incremented the exact right number of times.
        value = await self.cog.redis.get("test_key")
        self.assertEqual(value, 100) 

def ensure_valid_reminder(
        self,
        reminder: dict,
        cancel_task: bool = True
    ) -> t.Tuple[bool, discord.User, discord.TextChannel]:
        """Ensure reminder author and channel can be fetched otherwise delete the reminder."""
        user = self.bot.get_user(reminder['author'])
        channel = self.bot.get_channel(reminder['channel_id'])
        is_valid = True
        if not user or not channel:
            is_valid = False
            log.info(
                f"Reminder {reminder['id']} invalid: "
                f"User {reminder['author']}={user}, Channel {reminder['channel_id']}={channel}."
            )
            asyncio.create_task(self._delete_reminder(reminder['id'], cancel_task))

        return is_valid, user, channel 

def schedule_task(self, task_id: t.Hashable, task_data: t.Any) -> None:
        """
        Schedules a task.

        `task_data` is passed to the `Scheduler._scheduled_task()` coroutine.
        """
        log.trace(f"{self.cog_name}: scheduling task #{task_id}...")

        if task_id in self._scheduled_tasks:
            log.debug(
                f"{self.cog_name}: did not schedule task #{task_id}; task was already scheduled."
            )
            return

        task = asyncio.create_task(self._scheduled_task(task_data))
        task.add_done_callback(partial(self._task_done_callback, task_id))

        self._scheduled_tasks[task_id] = task
        log.debug(f"{self.cog_name}: scheduled task #{task_id} {id(task)}.") 

def init(app: web.Application) -> None:
    event_dispatcher = app['event_dispatcher']
    event_dispatcher.consume('kernel_preparing', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_pulling', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_creating', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_started', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_terminating', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_terminated', app, handle_kernel_lifecycle)
    event_dispatcher.consume('kernel_success', app, handle_batch_result)
    event_dispatcher.consume('kernel_failure', app, handle_batch_result)
    event_dispatcher.consume('kernel_stat_sync', app, handle_kernel_stat_sync)
    event_dispatcher.consume('kernel_log', app, handle_kernel_log)
    event_dispatcher.consume('instance_started', app, handle_instance_lifecycle)
    event_dispatcher.consume('instance_terminated', app, handle_instance_lifecycle)
    event_dispatcher.consume('instance_heartbeat', app, handle_instance_heartbeat)
    event_dispatcher.consume('instance_stats', app, handle_instance_stats)

    app['pending_waits'] = set()

    # Scan ALIVE agents
    app['agent_lost_checker'] = aiotools.create_timer(
        functools.partial(check_agent_lost, app), 1.0)
    app['stats_task'] = asyncio.create_task(stats_report_timer(app)) 

def test_wait_before_create(zk, path):
    """await barrier.wait() should finish immediately if the barrier does not
    exist. Because it is semantically right: No barrier, no blocking.
    """
    wait_finished = False

    async def start_worker():
        barrier = zk.recipes.Barrier(path)
        await barrier.wait()
        nonlocal wait_finished
        wait_finished = True

    task = asyncio.create_task(start_worker())

    try:
        await asyncio.wait_for(task, timeout=2)
    except asyncio.TimeoutError:
        pass

    assert wait_finished 

def test_election_early_wait_for_leadership(zk, path):
    elec = zk.recipes.LeaderElection(path)

    early_wait_success = asyncio.Event()

    async def wait_early():
        await elec.wait_for_leadership()
        assert elec.has_leadership
        early_wait_success.set()

    asyncio.create_task(wait_early())
    await asyncio.sleep(0.5)
    assert not elec.has_leadership

    await elec.volunteer()

    # NO WAIT
    await asyncio.wait_for(early_wait_success.wait(), timeout=0.5)

    await elec.resign()

    assert not elec.has_leadership

    await zk.delete(path) 

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.label = lv.label(self)
        self._text = "Text"
        self.label.set_long_mode(lv.label.LONG.BREAK)
        self.label.set_align(lv.label.ALIGN.CENTER)

        self.note = lv.label(self)
        style = lv.style_t()
        lv.style_copy(style, qr_style)
        style.text.font = lv.font_roboto_16
        style.text.color = lv.color_hex(0x192432)
        self.note.set_style(0, style)
        self.note.set_text("")
        self.note.set_align(lv.label.ALIGN.CENTER)

        self.set_text(self._text)
        self.task = asyncio.create_task(self.animate())
        self.set_event_cb(self.cb) 

def stop(self):
        try:
            srv = self._http_proto_server
            if srv is not None:
                self._http_proto_server = None
                srv.close()
                await srv.wait_closed()
        finally:
            try:
                async with taskgroup.TaskGroup() as g:
                    for cmp in self._compilers_list:
                        g.create_task(cmp.close())
                    self._compilers_list.clear()

                for pgcon in self._pgcons_list:
                    pgcon.terminate()
                self._pgcons_list.clear()
                if self._http_request_logger is not None:
                    self._http_request_logger.cancel()
                    await self._http_request_logger
            finally:
                await super().stop() 

def asyncTaskAutoCancel(func):
    """Wraps an async method into a regular method, that will schedule
    the async function as a task. If this task has previously been scheduled
    and has not yet run, it will be cancelled. So a newly scheduled task
    overrides an older one.
    """
    taskAttributeName = f"_{func.__name__}_autoCancelTask"
    @functools.wraps(func)
    def createFuncTask(self, *args, **kwargs):
        oldTask = getattr(self, taskAttributeName, None)
        if oldTask is not None:
            oldTask.cancel()
        coro = func(self, *args, **kwargs)
        task = asyncio.create_task(coro)
        task.add_done_callback(_done_callback)
        setattr(self, taskAttributeName, task)
        return task
    return createFuncTask 

def __init__(self, update_statuses):
        self._update_statuses = update_statuses
        self._process_provider = ProcessProvider()
        self._process = None
        self._exe = self._find_exe()
        self._games_provider = LocalGames()

        self.database_parser = None
        self.config_parser = ConfigParser(None)
        self.uninstaller = None
        self.installed_games_cache = self.get_installed_games()

        loop = asyncio.get_event_loop()
        loop.create_task(self._register_local_data_watcher())
        loop.create_task(self._register_classic_games_updater())
        self.classic_games_parsing_task = None 

def _run(self):
        self._stopping.clear()

        self._redis = await utils.create_aredis_for_stream()
        self._stream_selector = StreamSelector(self.worker_count, self._redis)

        if "stream" in self.enabled_services:
            for i in range(self.worker_count):
                self._worker_tasks.append(
                    asyncio.create_task(self.stream_worker_task(i))
                )

        if "stream-monitoring" in self.enabled_services:
            self._stream_monitoring_task = asyncio.create_task(self.monitoring_task())

        if "smart-queue" in self.enabled_services:
            self._smart_queue_task = asyncio.create_task(
                self.smart_queue_processing_task()
            )

        LOG.debug("%d workers spawned", self.worker_count) 

def test_create_task_without_context(self):
            async def run_later(msg="Hello"):
                # print("run_later: %s" % async_tracer.active_span.operation_name)
                async with aiohttp.ClientSession() as session:
                    return await self.fetch(session, testenv["wsgi_server"] + "/")

            async def test():
                with async_tracer.start_active_span('test'):
                    asyncio.create_task(run_later("Hello"))
                await asyncio.sleep(0.5)

            self.loop.run_until_complete(test())

            spans = self.recorder.queued_spans()

            self.assertEqual(2, len(spans))
            self.assertEqual("sdk", spans[0].n)
            self.assertEqual("wsgi", spans[1].n)

            # Without the context propagated, we should get two separate traces
            self.assertNotEqual(spans[0].t, spans[1].t) 

def restart_MQTT(self):
        
        if not (bumper.mqtt_server.broker.transitions.state == "stopped" or bumper.mqtt_server.broker.transitions.state == "not_started"):
            # close session writers - this was required so bots would reconnect properly after restarting
            for sess in list(bumper.mqtt_server.broker._sessions):                
                sessobj = bumper.mqtt_server.broker._sessions[sess][1]
                if sessobj.session.transitions.state == "connected":
                    await sessobj.writer.close()

            #await bumper.mqtt_server.broker.shutdown()
            aloop = asyncio.get_event_loop()
            aloop.call_later(
            0.1, lambda: asyncio.create_task(bumper.mqtt_server.broker.shutdown())
            )  # In .1 seconds shutdown broker

        
        aloop = asyncio.get_event_loop()
        aloop.call_later(
           1.5, lambda: asyncio.create_task(bumper.mqtt_server.broker_coro())
        )  # In 1.5 seconds start broker 

def handle_RestartService(self, request):
        try:
            service = request.match_info.get("service", "")
            if service == "Helperbot":
                await self.restart_Helper()
                return web.json_response({"status": "complete"})
            elif service == "MQTTServer":
                asyncio.create_task(self.restart_MQTT())
                aloop = asyncio.get_event_loop()
                aloop.call_later(
                    5, lambda: asyncio.create_task(self.restart_Helper())
                )  # In 5 seconds restart Helperbot
                
                return web.json_response({"status": "complete"})
            elif service == "XMPPServer":
                await self.restart_XMPP()
                return web.json_response({"status": "complete"})
            else:
                return web.json_response({"status": "invalid service"})

        except Exception as e:
            confserverlog.exception("{}".format(e))
            pass 

def broker_coro(self):

        mqttserverlog.info(
            "Starting MQTT Server at {}:{}".format(self.address[0], self.address[1])
        )        

        try:
            await self.broker.start()

        except hbmqtt.broker.BrokerException as e:
            mqttserverlog.exception(e)
            #asyncio.create_task(bumper.shutdown())
            pass

        except Exception as e:
            mqttserverlog.exception("{}".format(e))
            #asyncio.create_task(bumper.shutdown())
            pass 

def test_start_stop():
    with LogCapture() as l:
        if os.path.exists("tests/tmp.db"):
            os.remove("tests/tmp.db")  # Remove existing db

        b = bumper
        b.db = "tests/tmp.db"  # Set db location for testing
        b.conf1_listen_address = "127.0.0.1"
        b.conf1_listen_port = 444
        asyncio.create_task(b.start())
        await asyncio.sleep(0.1)
        l.check_present(("bumper", "INFO", "Starting Bumper"))
        l.clear()

        asyncio.create_task(b.shutdown())
        await asyncio.sleep(0.1)
        l.check_present(
            ("bumper", "INFO", "Shutting down"), ("bumper", "INFO", "Shutdown complete")
        )
        assert b.shutting_down == True 

def test_start_stop_debug():
    with LogCapture() as l:
        if os.path.exists("tests/tmp.db"):
            os.remove("tests/tmp.db")  # Remove existing db

        b = bumper
        b.db = "tests/tmp.db"  # Set db location for testing
        b.bumper_listen = "0.0.0.0"
        b.bumper_debug = True
        asyncio.create_task(b.start())

        await asyncio.sleep(0.1)
        asyncio.create_task(b.shutdown())
        l.check_present(("bumper", "INFO", "Starting Bumper"))
        l.clear()
        await asyncio.sleep(0.1)
        l.check_present(
            ("bumper", "INFO", "Shutting down"), ("bumper", "INFO", "Shutdown complete")
        )
        assert b.shutting_down == True 

def __init__(self, loop=None):
        self._loop = loop or asyncio.get_event_loop()
        # prefer asyncio.create_task starting from Python 3.7
        if hasattr(asyncio, 'create_task'):
            self._ensure_future = asyncio.create_task
        else:
            self._ensure_future = asyncio.ensure_future 

def __init__(self, bot: Bot):
        self.bot = bot

        self.webhook = None
        self.access_token = None
        self.client_auth = BasicAuth(RedditConfig.client_id, RedditConfig.secret)

        bot.loop.create_task(self.init_reddit_ready())
        self.auto_poster_loop.start() 

def cog_unload(self) -> None:
        """Stop the loop task and revoke the access token when the cog is unloaded."""
        self.auto_poster_loop.cancel()
        if self.access_token and self.access_token.expires_at > datetime.utcnow():
            asyncio.create_task(self.revoke_access_token()) 

def wait_for_dormant_channel(self) -> discord.TextChannel:
        """Wait for a dormant channel to become available in the queue and return it."""
        log.trace("Waiting for a dormant channel.")

        task = asyncio.create_task(self.channel_queue.get())
        self.queue_tasks.append(task)
        channel = await task

        log.trace(f"Channel #{channel} ({channel.id}) finally retrieved from the queue.")
        self.queue_tasks.remove(task)

        return channel 

def __init__(self, bot: Bot):
        super().__init__()
        self.bot = bot
        self.muted_channels = set()
        self._get_instance_vars_task = self.bot.loop.create_task(self._get_instance_vars())
        self._get_instance_vars_event = asyncio.Event() 

def cog_unload(self) -> None:
        """Send alert with silenced channels on unload."""
        if self.muted_channels:
            channels_string = ''.join(channel.mention for channel in self.muted_channels)
            message = f"<@&{Roles.moderators}> channels left silenced on cog unload: {channels_string}"
            asyncio.create_task(self._mod_alerts_channel.send(message))

    # This cannot be static (must have a __func__ attribute). 

def __ainit__(self) -> None:
        log.info('Session scheduler started')
        self.tick_task = asyncio.create_task(self.generate_scheduling_tick())
        self.registry.event_dispatcher.consume('kernel_enqueued', None, self.schedule)
        self.registry.event_dispatcher.consume('kernel_terminated', None, self.schedule)
        self.registry.event_dispatcher.consume('instance_started', None, self.schedule)
        self.registry.event_dispatcher.consume('do_schedule', None, self.schedule)
        # TODO: add events for resource configuration changes and subscribe them here.
        self.lock_manager = aioredlock.Aioredlock([
            {'host': str(self.config['redis']['addr'][0]),
             'port': self.config['redis']['addr'][1],
             'password': self.config['redis']['password'] if self.config['redis']['password'] else None,
             'db': REDIS_LIVE_DB},
        ]) 

def start(
        self,
        func: BackgroundTask,
        name: str = None,
        *,
        sched: Scheduler = None,
    ) -> uuid.UUID:
        task_id = uuid.uuid4()
        redis_producer = self.event_dispatcher.redis_producer

        def _pipe_builder():
            pipe = redis_producer.pipeline()
            tracker_key = f'bgtask.{task_id}'
            now = str(time.time())
            pipe.hmset_dict(tracker_key, {
                'status': 'started',
                'current': '0',
                'total': '0',
                'msg': '',
                'started_at': now,
                'last_update': now,
            })
            pipe.expire(tracker_key, MAX_BGTASK_ARCHIVE_PERIOD)
            return pipe

        await redis.execute_with_retries(_pipe_builder)

        if sched:
            # aiojobs' Scheduler doesn't support add_done_callback yet
            raise NotImplementedError
        else:
            task = asyncio.create_task(self._wrapper_task(func, task_id, name))
            self.ongoing_tasks.add(task)
            task.add_done_callback(self.ongoing_tasks.remove)
        return task_id 

def init(app: web.Application) -> None:
    app['status_watch_task'] = asyncio.create_task(detect_status_update(app)) 

def upload_files(request: web.Request) -> web.Response:
    loop = asyncio.get_event_loop()
    reader = await request.multipart()
    registry = request.app['registry']
    session_name = request.match_info['session_name']
    requester_access_key, owner_access_key = await get_access_key_scopes(request)
    log.info('UPLOAD_FILE (ak:{0}/{1}, s:{2})',
             requester_access_key, owner_access_key, session_name)
    try:
        await registry.increment_session_usage(session_name, owner_access_key)
        file_count = 0
        upload_tasks = []
        async for file in aiotools.aiter(reader.next, None):
            if file_count == 20:
                raise InvalidAPIParameters('Too many files')
            file_count += 1
            # This API handles only small files, so let's read it at once.
            chunks = []
            recv_size = 0
            while True:
                chunk = await file.read_chunk(size=1048576)
                if not chunk:
                    break
                chunk_size = len(chunk)
                if recv_size + chunk_size >= 1048576:
                    raise InvalidAPIParameters('Too large file')
                chunks.append(chunk)
                recv_size += chunk_size
            data = file.decode(b''.join(chunks))
            log.debug('received file: {0} ({1:,} bytes)', file.filename, recv_size)
            t = loop.create_task(
                registry.upload_file(session_name, owner_access_key,
                                     file.filename, data))
            upload_tasks.append(t)
        await asyncio.gather(*upload_tasks)
    except BackendError:
        log.exception('UPLOAD_FILES: exception')
        raise
    return web.Response(status=204) 

def _run(node):
    out = Collect(node)  # noqa F405
    graph = out._construct_graph()
    nodes = graph.getNodes()

    value, last = None, None

    while True:
        for level in nodes:
            await asyncio.gather(*(asyncio.create_task(n()) for n in level))
        value, last = out.value(), value
        if isinstance(value, StreamEnd):
            break
    return last