Sample Pushed when Streaming is closing

[matpadalino87]

Dear all,

I'm writing you to ask about an issue that I have when I'm closing my stream. I'm pushing one sample with a marker stream type and when I stop the execution of my script Labrecorder saves a marker equal to 0 or another number randomly if the sent markers are more than 100. I also used and not used the command "del" to destroy the outlet object but in both cases the randomic marker is saved on Labrecorder... Can you help me? Can you give me some information about this last marker automatically generated?

Have nice day!

[cboulay]

This is pretty interesting, though I'm not certain that I understand the problem exactly.

Can you please provide minimal scripts to reproduce the problem (one for the inlet, one for the outlet)?

[matpadalino87]

If you record this outlet with labrecorder app you can see that at the end of pushing a sample equal to 0 or a int32 value choosed in the range of the pushed sample values is sent to LSL env. Idem if you receive with the inlet this markers streaming a sample is printed on console after having destroyed the outlet.

In my code outlet and inlet run in 2 different processes (2 different spyder routines), to simplify the procedure the outlet is a for loop but for while loop of inlet must be stopped manually.

Thank you for the support!!! Have nice day

Outlet:

info = StreamInfo(name='phase', type='Markers', channel_count=1,
                  channel_format='int32', source_id='example_stream_001')
outlet = StreamOutlet(info)  # Broadcast the stream.
time.sleep(10)
lista_stim=[]
num_mark = 200
for i in range(0,num_mark):
    last_time = local_clock()
    #Send the calculated time stamp to LSL
    ph_2stim = [random.randint(10,100)]
    outlet.push_sample(ph_2stim, last_time)
    print(ph_2stim)
    lista_stim.append(ph_2stim)
    time.sleep(2)
del outlet

###################################################################
Inlet:

from pylsl import resolve_stream,StreamInlet, local_clock
streams = resolve_stream('source_id','example_stream_001') # make connection
inlet0 = StreamInlet(streams[0])
inlet0.open_stream()

while True:
        sample,timestamp = inlet0.pull_sample()
        print(sample)

[agricolab]

What version of pylsl and liblsl are you using on which operating system?

[matpadalino87]

I use a Window 10 sys and pylsl 1.16.1 ( I work on anaconda env with Spyder but from terminal I don't see liblsl package version with conda list command).

[mscheltienne]

Chiming on this old issue, you can reproduce easily by reducing the delays, for instance I just tried it with the mne-lsl and numpy version below. It comes down to the del outlet call, if it's included then the inlet side receives an ERR| Stream transmission broke off (Input stream error.) and this additional sample as the same time. If the outlet is not deleted before the inlet is closed then you don't get the extra sample.

You can for instance wait for the end of the outlet loop; then query as much as you want the inlet with pull_sample(), then delete the outlet; and you will get the extra sample in the next pull_sample() query.

@cboulay Question is, is this a feature? Is LSL sending a code/something when the outlet is destroyed through lib.lsl_destroy_outlet(obj)?

Outlet

import time

import numpy as np
from mne_lsl.lsl import StreamInfo, StreamOutlet, local_clock

rng = np.random.default_rng()
info = StreamInfo(
    name="phase",
    stype="Markers",
    n_channels=1,
    dtype=np.int32,
    source_id="test",
    sfreq=0,
)
outlet = StreamOutlet(info)

time.sleep(4)
lista_stim = []

num_mark = 200
for _ in range(num_mark):
    ph_2stim = rng.integers(10, 100, size=1)
    outlet.push_sample(ph_2stim, local_clock())
    print(ph_2stim)
    lista_stim.append(ph_2stim[0])
    time.sleep(0.1)

del outlet

Inlet

from mne_lsl.lsl import StreamInlet, resolve_streams

streams = resolve_streams(source_id="test")
inlet = StreamInlet(streams[0])
inlet.open_stream()

samples = []
while True:
    sample, timestamp = inlet.pull_sample()
    if sample.size == 0:
        continue
    samples.extend(list(sample))

[JesseLivezey]

A few more data points:

I'm on ubuntu 24.04

If I run @mscheltienne's code (in 2 notebooks simultaneously)

• I see the extra sample whether I use push_sample() or push_chunk() with single samples
• Pushing more samples (higher num_mark) makes it easier to get an extra sample, but it seems like it is probably possible for any num_mark >= 1 if you make the sleep in the loop short enough
• For int64/float64, you see extra samples sooner (lower num_mark) than int32/64, the opposite seems to be true for int16/8
• If I add a longer sleep after the output loop but before del (1 second), I have to increase num_mark to see the extra sample. For num_mark = 200, I can remove the sleeps in the outlet loop and just sleep once before del and I don't see the extra sample. But if num_mark = 1000, I see it again.
• The extra sample does seem to have a timestamp that corresponds to del not a push

And maybe this is expected but

• if the del happens fast enough after the push_samples(), and num_mark is small enough, sometimes the inlet receives no samples (or only some of the first samples)

[JesseLivezey]

Another observation: if I open an xdf file that has an extra sample for an irregularly sampled stream, there is an internal inconsistency in how many samples it has

streams[2]["footer"]["info"]["sample_count"] # 200, correct
len(streams[2]['time_series']) # 201, 1 extra sample

The footer seems to have a correct count, but there is an extra sample in the time_series and time_stamps data.

It also seems like the streams with these extra samples can occasionally have bad values in the last clock_offsets and this causes problems for syncing timestamps in pyxdf.

Clock offset time/value for a irregularly sampled stream that contains an extra sample. The (last) clock_offset values look reasonable and the synced timestamps make sense.

Another irregularly sampled stream from the same xdf file that also contains an extra sample. The last entry in the clock offsets has seemingly very bad values. When I load this stream with pyxdf and synchronize_clocks=True, I get synced timestamps that don't make any sense. I'm guessing the final clock_offset entry may be causing problems for pyxdf.

[sappelhoff]

I think this can be closed after the fixes by @mscheltienne