Skip to content

Monitor unit test non-deterministically fails in CI/CD #109

@mathisrichter

Description

@mathisrichter

@PhilippPlank: This error happens at the GitHub checks occasionally. It never happened when I run my unit tests and the solution is to re-run the tests on GitHub. What is the problem with this test and does it connected with the division by 0 runtime warning?

Error: Test failed: proc.monitor.test_monitors.Monitors.test_monitor_collects_voltage_and_spike_data_from_population_lif
69 Traceback (most recent call last):
70 File "/opt/hostedtoolcache/Python/3.9.9/x64/lib/python3.9/unittest/case.py", line 59, in testPartExecutor
71 yield
72 File "/opt/hostedtoolcache/Python/3.9.9/x64/lib/python3.9/unittest/case.py", line 592, in run
73 self._callTestMethod(testMethod)
74 File "/opt/hostedtoolcache/Python/3.9.9/x64/lib/python3.9/unittest/case.py", line 550, in _callTestMethod
75 method()
76 File "/home/runner/work/lava/lava/tests/lava/proc/monitor/test_monitors.py", line 236, in test_monitor_collects_voltage_and_spike_data_from_population_lif
77 self.assertTrue(np.all(spike_data == np.array([[0, 0, 0, 1, 0, 0],
78 File "/opt/hostedtoolcache/Python/3.9.9/x64/lib/python3.9/unittest/case.py", line 688, in assertTrue
79 raise self.failureException(msg)
80 AssertionError: False is not true

@mathisrichter: I believe this is the same non-deterministic behavior we were seeing with our SpikeGenerator. The Monitor probes OutPorts by setting up an InPort that receives from it. But it does not use recv() right away - instead it uses probe() first. Since the order in which LIF and Monitor get executed within one time step is not deterministic, a spike is sometimes picked up in one time step, sometimes in the next. The unit test tests for it arriving in a particular time step.

def run_spk(self):
    """
    During this phase, InPorts of Monitor process collects data from
    monitored OutPorts
    """
    # Check if this Monitor Process instance has been assigned to monitor
    # any OutPort by checking proc_params["n_in_ports"], if so loop over
    # those InPorts; readout their values to correspoinding data-stroing
    # Var
    for i in range(self.proc_params["n_in_ports"]):
        in_port_name = self.proc_params["InPorts"][i]
        out_read_name = self.proc_params["VarsData2"][i]
        # Check if buffer is non-empty, otherwise add zero to the list
        if getattr(self, in_port_name).csp_ports[0].probe():
            getattr(self, out_read_name)[:, self.current_ts - 1] = \
                np.squeeze(np.array(getattr(self, in_port_name).recv()))
        else:
            getattr(self, out_read_name)[:, self.current_ts - 1] = 0

Metadata

Metadata

Assignees

Labels

1-bugSomething isn't working

Type

No type

Projects

No projects

Relationships

None yet

Development

No branches or pull requests

Issue actions