Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
counter += 1
await asyncio.sleep(2)
periodic = PeriodicCallback(task)
periodic.start(0.1, event_loop, shield=True)
await asyncio.sleep(0)
with pytest.raises(asyncio.CancelledError):
await periodic.stop()
assert counter == 1
# No shield
counter = 0
periodic = PeriodicCallback(task)
periodic.start(0.1, event_loop, shield=False)
await asyncio.sleep(0)
with pytest.raises(asyncio.CancelledError):
await periodic.stop()
assert counter == 0
async def test_shield(event_loop):
counter = 0
async def task():
nonlocal counter
counter += 1
await asyncio.sleep(2)
periodic = PeriodicCallback(task)
periodic.start(0.1, event_loop, shield=True)
await asyncio.sleep(0)
with pytest.raises(asyncio.CancelledError):
await periodic.stop()
assert counter == 1
# No shield
counter = 0
periodic = PeriodicCallback(task)
periodic.start(0.1, event_loop, shield=False)
await asyncio.sleep(0)
async def start(self):
self.logger = self.log.getChild(str(id(self)))
self.profiler = cProfile.Profile()
self.periodic = PeriodicCallback(self.save_stats)
self.profiler.enable()
self.periodic.start(self.interval)
async def start(self):
log.warning("Start memory tracer")
tracemalloc.start()
self._tracer = PeriodicCallback(self.show_stats)
self._log = log.getChild(str(id(self)))
self._snapshot_on_start = self.take_snapshot()
self._tracer.start(self.interval)
async def start(self):
namespace = ".".join(
strip_carbon_ns(item) for item in self.namespace
)
client = PROTOCOLS[self.protocol](
self.host,
self.port,
namespace=namespace,
storage=self.storage(),
loop=self.loop
)
set_client(client)
self._handle = PeriodicCallback(client.send)
self._handle.start(self.send_interval, loop=self.loop)
log.info(
'Periodic carbon metrics sender started. Send to %s://%s:%d with '
'interval %rs', self.protocol, self.host, self.port,
self.send_interval)