Pynrfjprog create dummy thread, how to avoid/delete them ?

[AntoineB]

Hi,

I'm devlopping python (v3.9.6) software to flash nRF52832 using pynrfjprog (v10.17.3) package.

The soft is a loop trying to erase and flash. But while debugging with VSCode, I see that pynrfjprog package  create "Dummy-X" threads and never release/kill them.

The point is, if the software run for a long time, more than 2000 thread can be running at the same time and take lot of ressources on the computer.

My question is, is there a way to erase/flash without creating this Dummy thread ? Or is there a way to delete them once the erase/flash action is over ?

Thanks in advance,

Antoine

[Einarh]

Hello

What software are you using to flash exactly? Is it a script you wrote yourself?

If so, please share your script.

Best regards,

Einar

[AntoineB]

Hi Einar,

It is in fact my own code. Please find bellow the script I'm using:

import logging
import pynrfjprog.HighLevel
import threading

class nRF528XXFlasher():
    api = None
    flashers = {}
    semaphore = threading.BoundedSemaphore(value=1)

    def __init__(self, snr=None, config={}, logger=None):
        self.config = config

        if logger is None:
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logger.getChild(__name__)

        self.logger.setLevel(self.config.get("log_level", "INFO"))

        try:
            nRF528XXFlasher.semaphore.acquire(timeout=3)

            if nRF528XXFlasher.api is None:
                nRF528XXFlasher.api = pynrfjprog.HighLevel.API(self.config.get("nRF528XX_flasher_enable_log", False))

            if not nRF528XXFlasher.api.is_open():
                nRF528XXFlasher.api.open()

            if snr is None:
                probes_snrs = nRF528XXFlasher.api.get_connected_probes()
                if len(probes_snrs) > 0:
                    self.logger.debug(f"nRF528XXFlasher possible serials : {probes_snrs}")
                    for probe_snr in probes_snrs:
                        logger.debug(probe_snr)
                        if probe_snr not in nRF528XXFlasher.flashers:
                            snr = probe_snr
                            break
        finally:
            nRF528XXFlasher.semaphore.release()

        self.snr = snr
        if self.snr is None:
            raise Exception(f"nRF528XX Flasher probe not found ({self.snr})")

        nRF528XXFlasher.flashers[snr] = self

        # Catch and redirect logs from the probe
        class ProbeLogFilter(logging.Filter):
            def __init__(self, flasher, name=""):
                super().__init__(name)
                self.flasher = flasher

            def filter(self, record):
                if self.flasher.config.get("nRF528XX_flasher_enable_log", False):
                    if str(self.flasher.snr) in record.name:
                        self.flasher.logger.handle(record)
                return False

        logging.getLogger(f"pynrfjprog.HighLevel.Probes.{self.snr}").addFilter(
            ProbeLogFilter(flasher=self))

    def __del__(self):
        for snr in nRF528XXFlasher.flashers:
            if nRF528XXFlasher.flashers[snr] == self:
                nRF528XXFlasher.flashers.pop(snr)
                break
        if len(nRF528XXFlasher.flashers) <= 0:
            if nRF528XXFlasher.api.is_open():
                nRF528XXFlasher.api.close()

    def erase_app(self):
        with pynrfjprog.HighLevel.DebugProbe(nRF528XXFlasher.api, self.snr, log=self.config.get("nRF528XX_flasher_enable_log", True)) as probe:
            probe.erase(erase_action=pynrfjprog.HighLevel.EraseAction.ERASE_ALL)
        return True

    def flash_app(self, hex_file):
        with pynrfjprog.HighLevel.DebugProbe(nRF528XXFlasher.api, self.snr, pynrfjprog.HighLevel.CoProcessor.CP_APPLICATION, log=self.config.get("nRF528XX_flasher_enable_log", True)) as probe:
            probe.program(hex_file, program_options=pynrfjprog.HighLevel.ProgramOptions(verify=pynrfjprog.HighLevel.VerifyAction.VERIFY_READ, erase_action=pynrfjprog.HighLevel.EraseAction.ERASE_ALL, reset=pynrfjprog.HighLevel.ResetAction.RESET_SYSTEM))
        return True

Best regards,

Antoine

[Einarh]

You can try to use LowLevel instead:

(you might have to set up your logfilter again)

import logging
from pynrfjprog import LowLevel
import threading

class nRF528XXFlasher:
    flashers = dict()
    semaphore = threading.BoundedSemaphore(value=1)
    
    def __init__(self, snr=None, config={}, logger=None):
        self.config = config
        
        if logger is None:
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logger.getChild(__name__)
        self.logger.setLevel(self.config.get("log_level", "INFO"))
        
        try:
            nRF528XXFlasher.semaphore.acquire(timeout=3)
            if snr is None:
                snr = self._get_random_snr()
            self.snr = snr
            nRF528XXFlasher.flashers[snr] = self
        finally:
            nRF528XXFlasher.semaphore.release()
            
        # Catch and redirect logs from the probe
        class ProbeLogFilter(logging.Filter):
            def __init__(self, flasher, name=""):
                super().__init__(name)
                self.flasher = flasher
                
            def filter(self, record):
                if self.flasher.config.get("nRF528XX_flasher_enable_log", False):
                    if str(self.flasher.snr) in record.name:
                        self.flasher.logger.handle(record)
                return False
                
        logging.getLogger(f"HighLevel.Probes.{self.snr}").addFilter(ProbeLogFilter(flasher=self))
        
    def _get_random_snr(self):
        with LowLevel.API(LowLevel.DeviceFamily.UNKNOWN) as api:
            probes_snrs = set(api.enum_emu_snr())
            self.logger.debug(f"nRF528XXFlasher possible serials : {probes_snrs}")
            available_snrs = probes_snrs.difference(nRF528XXFlasher.flashers.keys())
            if not available_snrs:
                raise AssertionError("No available snrs")
            return available_snrs.pop()
            
    def __del__(self):
        for snr in nRF528XXFlasher.flashers:
            if nRF528XXFlasher.flashers[snr] == self:
                nRF528XXFlasher.flashers.pop(snr)
                break
                
    def erase_app(self):
        with LowLevel.API(LowLevel.DeviceFamily.NRF52, log=self.config.get("nRF528XX_flasher_enable_log")) as api:
            api.connect_to_emu_with_snr(self.snr)
            api.erase_all()
        return True
        
    def flash_app(self, hex_file):
        with LowLevel.API(LowLevel.DeviceFamily.NRF52, log=self.config.get("nRF528XX_flasher_enable_log")) as api:
            api.connect_to_emu_with_snr(self.snr)
            api.program_file(hex_file)
            api.verify_file(hex_file)
            api.sys_reset()
        return True

-Einar

[AntoineB]

Hi Einarh,

Thanks for the quick reply. I've just tested the code you sent but with LowLevel api, Dummy threads are created when calling LowLevel.API to be used as api (line 57) but they are not deleted even with adding api.close()

Best regards,

Antoine

[AntoineB]

Hi again Einarh,

Do you have any clue on why this threads are created and never deleted ?

Thanks in advance for your help.

Best regard,

Antoine