eclipse-paho / eclipse-paho/paho.mqtt.python

Use Python selectors instead of select

Open
#697 14 comments 0 reactions 0 assignees View on GitHub
Status: Available Type: Enhancement
Dominant language
Python
Stars
2.4k
Forks
742
Avg merge
12d 48m
Merged PRs (30d)
1

Description

Using Python `select.select(..)` method limits the open connections to 340:

- https://github.com/eclipse/paho.mqtt.python/issues/183
- https://github.com/eclipse/paho.mqtt.python/issues/238
- https://github.com/eclipse/paho.mqtt.python/issues/499
- https://github.com/eclipse/paho.mqtt.python/issues/662

Python select [docs](https://docs.python.org/3/library/select.html "Python select documentation") encourages users to use [selectors](https://docs.python.org/3/library/selectors.html#module-selectors) instead of select.

Would it be possible to change the select usage to use selectors? The following implementation worked for us:

```python
import selectors

class Client:
[...]
def _loop(self, timeout: float = 1.0) -> int:
if timeout < 0.0:
raise ValueError("Invalid timeout.")

sel = selectors.DefaultSelector()

eventmask = selectors.EVENT_READ

with suppress(IndexError):
packet = self._out_packet.popleft()
self._out_packet.appendleft(packet)
eventmask = selectors.EVENT_WRITE | eventmask

if self._sockpairR is None:
sel.register(self._sock, eventmask)
else:
sel.register(self._sock, eventmask)
sel.register(self._sockpairR, selectors.EVENT_READ)

pending_bytes = 0
if hasattr(self._sock, "pending"):
pending_bytes = self._sock.pending()

if pending_bytes > 0:
timeout = 0.0

try:
events = sel.select(timeout)
except TypeError:
return int(MQTT_ERR_CONN_LOST)
except ValueError:
return int(MQTT_ERR_CONN_LOST)
except Exception:
return int(MQTT_ERR_UNKNOWN)

socklist: list[list] = [[], []]

for key, _event in events:
if key.events & selectors.EVENT_READ:
socklist[0].append(key.fileobj)

if key.events & selectors.EVENT_WRITE:
socklist[1].append(key.fileobj)

if self._sock in socklist[0] or pending_bytes > 0:
rc = self.loop_read()
if rc or self._sock is None:
return int(rc)

if self._sockpairR and self._sockpairR in socklist[0]:
socklist[1].insert(0, self._sock)

with suppress(BlockingIOError):
self._sockpairR.recv(10000)

if self._sock in socklist[1]:
rc = self.loop_write()
if rc or self._sock is None:
return int(rc)

sel.close()

return int(self.loop_misc())
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.