Gradual memory leak in at least dosing control / automations

[CamDavidsonPilon]

Running the following doing automation long term:

# -*- coding: utf-8 -*-
from __future__ import annotations

from typing import Optional

from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
from pioreactor.exc import CalibrationError
from pioreactor.utils import local_persistant_storage

class CyclingCulture(DosingAutomationJobContrib):
    """
    Measures the OD of the culture continuously. Once OD600 reaches the threshold_od, it replaces the contents of the
    chamber with the next media source until blank OD levels are read.
    The next media is the alternate media if the current is the standard media, vice-versa.
    """

    automation_name = "cycling_culture"

    def __init__(self,  **kwargs) -> None:
        super().__init__(**kwargs)

    def execute(self) -> Optional[events.DilutionEvent]:
        media_ml_moved = 0.0
        while True:
            results = self.execute_io_action(media_ml=3, waste_ml=3)
            media_ml_moved += results['media_ml']

Running on a Raspberry Pi 3 Model B. Starting, it consumed ~3-4% of memory (expected). After ~48h, it consumed 11% of total memory, and climbing.

[CamDavidsonPilon]

This is very likely fixed in develop branch. I used the following script to test what the source was:

# -*- coding: utf-8 -*-

import tracemalloc
tracemalloc.start()

from typing import Optional
import time
from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
from pioreactor.exc import CalibrationError
from pioreactor.utils import local_persistant_storage
from pioreactor.background_jobs.dosing_control import start_dosing_control

class CyclingCulture(DosingAutomationJobContrib):
    """
    Measures the OD of the culture continuously. Once OD600 reaches the threshold_od, it replaces the contents of the
    chamber with the next media source until blank OD levels are read.
    The next media is the alternate media if the current is the standard media, vice-versa.
    """

    automation_name = "cycling_culture"

    def __init__(self,  **kwargs) -> None:
        super().__init__(**kwargs)

    def execute(self) -> Optional[events.DilutionEvent]:
        media_ml_moved = 0.0
        while True:
            results = self.execute_io_action(media_ml=3, waste_ml=3)
            media_ml_moved += results['media_ml']

with start_dosing_control("cycling_culture", 0.5) as dc:
    time.sleep(10)
    snapshot1 = tracemalloc.take_snapshot()

    time.sleep(120)
    snapshot2 = tracemalloc.take_snapshot()

top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print("[ Top 10 differences ]")
for stat in top_stats[:10]:
    print(stat)

The output contained lots of mqtt files. Turns out each time we ran a pump it was creating at least one new MQTT client. We fixed it by passing in the mqtt_client into pumping when called from execute_io_action.