statistics writer clubbed with insert readings method
@@ -223,9 +223,6 @@ class Ingest(object):
'to %s', cls._readings_buffer_size,
cls._readings_list_size * cls._max_concurrent_readings_inserts)
- # Start asyncio tasks
- cls._write_statistics_task = asyncio.ensure_future(cls._write_statistics())
-
cls._last_insert_time = 0
cls._insert_readings_wait_tasks = []
@@ -272,17 +269,6 @@ class Ingest(object):
cls._readings_list_not_empty = None
cls._readings_lists_not_full = None
- # Write statistics
- if cls._write_statistics_sleep_task is not None:
- cls._write_statistics_sleep_task.cancel()
- cls._write_statistics_sleep_task = None
-
- try:
- await cls._write_statistics_task
- cls._write_statistics_task = None
- except Exception:
- _LOGGER.exception('An exception was raised by Ingest._write_statistics')
-
cls._started = False
@classmethod
@@ -395,6 +381,8 @@ class Ingest(object):
_LOGGER.warning('Insert failed: Queue index: %s Batch size: %s', list_index, batch_size)
break
+ await cls._update_statistics()
+
del readings_list[:batch_size]
if not lists_not_full.is_set():
@@ -403,9 +391,8 @@ class Ingest(object):
_LOGGER.info('Insert readings loop stopped')
@classmethod
- async def _write_statistics(cls):
+ async def _update_statistics(cls):
"""Periodically commits collected readings statistics"""
- _LOGGER.info('South statistics writer started')
stats = await statistics.create_statistics(cls.storage_async)
@@ -415,26 +402,11 @@ class Ingest(object):
'discarded before being placed in the buffer. This may be due to some '
'error in the readings themselves.')
- while not cls._stop:
- # stop() calls _write_statistics_sleep_task.cancel().
- # Tracking _write_statistics_sleep_task separately is cleaner than canceling
- # this entire coroutine because allowing storage activity to be
- # interrupted will result in strange behavior.
- cls._write_statistics_sleep_task = asyncio.ensure_future(
- asyncio.sleep(cls._write_statistics_frequency_seconds))
-
- try:
- await cls._write_statistics_sleep_task
- except asyncio.CancelledError:
- pass
- finally:
- cls._write_statistics_sleep_task = None
-
readings = cls._readings_stats
cls._readings_stats = 0
try:
- asyncio.ensure_future(stats.update('READINGS', readings))
+ await stats.update('READINGS', readings)
except Exception as ex:
cls._readings_stats += readings
_LOGGER.exception('An error occurred while writing readings statistics, %s', str(ex))
@@ -443,7 +415,7 @@ class Ingest(object):
cls._discarded_readings_stats = 0
try:
- asyncio.ensure_future(stats.update('DISCARDED', readings))
+ await stats.update('DISCARDED', readings)
except Exception as ex:
cls._discarded_readings_stats += readings
_LOGGER.exception('An error occurred while writing discarded statistics, Error: %s', str(ex))
@@ -453,13 +425,11 @@ class Ingest(object):
description = 'Readings received by FogLAMP since startup for sensor {}'.format(key)
await stats.register(key, description)
try:
- asyncio.ensure_future(stats.add_update(cls._sensor_stats))
+ await stats.add_update(cls._sensor_stats)
cls._sensor_stats = {}
except Exception as ex:
_LOGGER.exception('An error occurred while writing sensor statistics, Error: %s', str(ex))
- _LOGGER.info('South statistics writer stopped')
-
@classmethod
def is_available(cls) -> bool:
"""Indicates whether all lists are currently full
