Hard latency restriction for the PulseAudio backend.

README.rst

@@ -112,6 +112,13 @@ In order to request lower latencies, pass a ``blocksize`` to ``player`` or
 try to honor your request as best it can. On Windows/WASAPI, setting
 ``exclusive_mode=True`` might help, too (this is currently experimental).
 
+In Linux, it is possible to restrict the latency by setting the optional
+parameter ``maxlatency``, which takes an integer number of samples. The setting
+of this parameter limits the buffer size of the PulseAudio backend. If your
+algorithm cannot keep up with the playback/recording, buffer underflows or overflows
+will occur. Underflow and overflow events can be displayed by setting the optional
+argument ``report_under_overflow`` to ``True``.
+
 Another source of latency is in the ``record`` function, which buffers output up
 to the requested ``numframes``. In general, for optimal latency, you should use
 a ``numframes`` significantly lower than the ``blocksize`` above, maybe by a

soundcard/pulseaudio.py

@@ -18,7 +18,7 @@
 except OSError:
     # Try explicit file name, if the general does not work (e.g. on nixos)
     _pa = _ffi.dlopen('libpulse.so')
-    
+
 # First, we need to define a global _PulseAudio proxy for interacting
 # with the C API:
 
@@ -286,6 +286,10 @@ def __exit__(self_, exc_type, exc_value, traceback):
     _pa_stream_writable_size = _lock(_pa.pa_stream_writable_size)
     _pa_stream_write = _lock(_pa.pa_stream_write)
     _pa_stream_set_read_callback = _pa.pa_stream_set_read_callback
+    _pa_stream_set_overflow_callback = _pa.pa_stream_set_overflow_callback
+    _pa_stream_set_underflow_callback = _pa.pa_stream_set_underflow_callback
+    _pa_stream_get_underflow_index = _pa.pa_stream_get_underflow_index
+
 
 _pulse = _PulseAudio()
 atexit.register(_pulse._shutdown)
@@ -501,7 +505,7 @@ class _Speaker(_SoundCard):
     def __repr__(self):
         return '<Speaker {} ({} channels)>'.format(self.name, self.channels)
 
-    def player(self, samplerate, channels=None, blocksize=None):
+    def player(self, samplerate, channels=None, blocksize=None, maxlatency=None, report_under_overflow=False):
         """Create Player for playing audio.
 
         Parameters
@@ -517,20 +521,21 @@ def player(self, samplerate, channels=None, blocksize=None):
         blocksize : int
             Will play this many samples at a time. Choose a lower
             block size for lower latency and more CPU usage.
-        exclusive_mode : bool, optional
-            Windows only: open sound card in exclusive mode, which
-            might be necessary for short block lengths or high
-            sample rates or optimal performance. Default is ``False``.
+        maxlatency : int
+            Linux only: restrict latency to maxlatency sample frames. If set, buffer underflows or overflows will occur when
+            processing cannot keep up.
+        report_under_overflow : bool, optional
+            Linux only: print debug information to terminal, whenever buffer underflows or overflows occur.
 
         Returns
         -------
         player : _Player
         """
         if channels is None:
             channels = self.channels
-        return _Player(self._id, samplerate, channels, blocksize)
+        return _Player(self._id, samplerate, channels, blocksize, maxlatency, report_under_overflow)
 
-    def play(self, data, samplerate, channels=None, blocksize=None):
+    def play(self, data, samplerate, channels=None, blocksize=None, maxlatency=None, report_under_overflow=False):
         """Play some audio data.
 
         Parameters
@@ -548,10 +553,15 @@ def play(self, data, samplerate, channels=None, blocksize=None):
         blocksize : int
             Will play this many samples at a time. Choose a lower
             block size for lower latency and more CPU usage.
+        maxlatency : int
+            Linux only: restrict latency to maxlatency sample frames. If set, buffer underflows or overflows will occur,
+            when the processing cannot keep up.
+        report_under_overflow : bool, optional
+            Linux only: print debug information to terminal, whenever buffer underflows or overflows occur.
         """
         if channels is None:
             channels = self.channels
-        with _Player(self._id, samplerate, channels, blocksize) as s:
+        with _Player(self._id, samplerate, channels, blocksize, maxlatency, report_under_overflow) as s:
             s.play(data)
 
     def _get_info(self):
@@ -582,7 +592,7 @@ def isloopback(self):
         """bool : Whether this microphone is recording a speaker."""
         return self._get_info()['device.class'] == 'monitor'
 
-    def recorder(self, samplerate, channels=None, blocksize=None):
+    def recorder(self, samplerate, channels=None, blocksize=None, maxlatency=None):
         """Create Recorder for recording audio.
 
         Parameters
@@ -598,6 +608,9 @@ def recorder(self, samplerate, channels=None, blocksize=None):
         blocksize : int
             Will record this many samples at a time. Choose a lower
             block size for lower latency and more CPU usage.
+        maxlatency : int
+            Linux only: restrict latency to maxlatency sample frames. If set, buffer underflows or overflows will occur,
+            when the processing cannot keep up.
         exclusive_mode : bool, optional
             Windows only: open sound card in exclusive mode, which
             might be necessary for short block lengths or high
@@ -609,9 +622,9 @@ def recorder(self, samplerate, channels=None, blocksize=None):
         """
         if channels is None:
             channels = self.channels
-        return _Recorder(self._id, samplerate, channels, blocksize)
+        return _Recorder(self._id, samplerate, channels, blocksize, maxlatency)
 
-    def record(self, numframes, samplerate, channels=None, blocksize=None):
+    def record(self, numframes, samplerate, channels=None, blocksize=None, maxlatency=None):
         """Record some audio data.
 
         Parameters
@@ -629,6 +642,10 @@ def record(self, numframes, samplerate, channels=None, blocksize=None):
         blocksize : int
             Will record this many samples at a time. Choose a lower
             block size for lower latency and more CPU usage.
+        maxlatency : int
+            Linux only: restrict latency to maxlatency sample frames. If set, buffer underflows or overflows will occur,
+            when the processing cannot keep up.
+
 
         Returns
         -------
@@ -637,7 +654,7 @@ def record(self, numframes, samplerate, channels=None, blocksize=None):
         """
         if channels is None:
             channels = self.channels
-        with _Recorder(self._id, samplerate, channels, blocksize) as r:
+        with _Recorder(self._id, samplerate, channels, blocksize, maxlatency) as r:
             return r.record(numframes)
 
 
@@ -653,12 +670,15 @@ class _Stream:
 
     """
 
-    def __init__(self, id, samplerate, channels, blocksize=None, name='outputstream'):
+    def __init__(self, id, samplerate, channels, blocksize=None, maxlatency=None, report_under_overflow=False,
+                 name='outputstream'):
         self._id = id
         self._samplerate = samplerate
         self._name = name
         self._blocksize = blocksize
         self.channels = channels
+        self._maxlatency = maxlatency
+        self._report_under_overflow = report_under_overflow
 
     def __enter__(self):
         samplespec = _ffi.new("pa_sample_spec*")
@@ -693,12 +713,14 @@ def __enter__(self):
             errno = _pulse._pa_context_errno(_pulse.context)
             raise RuntimeError("stream creation failed with error ", errno)
         bufattr = _ffi.new("pa_buffer_attr*")
-        bufattr.maxlength = 2**32-1 # max buffer length
-        numchannels = self.channels if isinstance(self.channels, int) else len(self.channels)
-        bufattr.fragsize = self._blocksize*numchannels*4 if self._blocksize else 2**32-1 # recording block sys.getsizeof()
-        bufattr.minreq = 2**32-1 # start requesting more data at this bytes
-        bufattr.prebuf = 2**32-1 # start playback after this bytes are available
-        bufattr.tlength = self._blocksize*numchannels*4 if self._blocksize else 2**32-1 # buffer length in bytes on server
+        numchannels = samplespec.channels
+        bytes_per_sample = 4  # for _pa.PA_SAMPLE_FLOAT32LE
+        bufattr.maxlength = self._maxlatency * numchannels * bytes_per_sample if self._maxlatency else 2 ** 32 - 1
+        bufattr.fragsize = self._blocksize * numchannels * bytes_per_sample if self._blocksize else 2 ** 32 - 1
+        bufattr.minreq = 2 ** 32 - 1  # start requesting more data at this bytes
+        bufattr.prebuf = 2 ** 32 - 1  # start playback after prebuf bytes are available
+        bufattr.tlength = self._blocksize * numchannels * bytes_per_sample if self._blocksize else 2 ** 32 - 1  # buffer length in bytes on server
+
         self._connect_stream(bufattr)
         while _pulse._pa_stream_get_state(self.stream) not in [_pa.PA_STREAM_READY, _pa.PA_STREAM_FAILED]:
             time.sleep(0.01)
@@ -740,8 +762,24 @@ class _Player(_Stream):
     """
 
     def _connect_stream(self, bufattr):
+        if self._report_under_overflow:
+            @_ffi.callback("pa_stream_notify_cb_t")
+            def overflow_callback(stream, userdata):
+                print('Overflow detected.')
+
+            self._overflow_callback = overflow_callback
+            _pulse._pa_stream_set_overflow_callback(self.stream, overflow_callback, _ffi.NULL)
+
+            @_ffi.callback("pa_stream_notify_cb_t")
+            def underflow_callback(stream, userdata):
+                time_underflow = _pulse._pa_stream_get_underflow_index(stream)
+                print('Underflow detected at position ' + str(time_underflow))
+
+            self._underflow_callback = underflow_callback
+            _pulse._pa_stream_set_underflow_callback(self.stream, underflow_callback, _ffi.NULL)
+
         _pulse._pa_stream_connect_playback(self.stream, self._id.encode(), bufattr, _pa.PA_STREAM_ADJUST_LATENCY,
-                                                _ffi.NULL, _ffi.NULL)
+                                           _ffi.NULL, _ffi.NULL)
 
     def play(self, data):
         """Play some audio data.

soundcard/pulseaudio.py.h

@@ -415,5 +415,9 @@ pa_stream_state_t pa_stream_get_state(pa_stream *p);
 
 typedef void(*pa_stream_request_cb_t)(pa_stream *p, size_t nbytes, void *userdata);
 void pa_stream_set_read_callback(pa_stream *p, pa_stream_request_cb_t cb, void *userdata);
+typedef void (*pa_stream_notify_cb_t)(pa_stream *p, void *userdata);
+void pa_stream_set_overflow_callback(pa_stream *p, pa_stream_notify_cb_t cb, void *userdata);
+void pa_stream_set_underflow_callback(pa_stream *p, pa_stream_notify_cb_t cb, void *userdata);
+int64_t pa_stream_get_underflow_index(const pa_stream *p);
 
 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata);