changeset 347d108696ce in trytond:default
details: https://hg.tryton.org/trytond?cmd=changeset&node=347d108696ce
description:
Use default selectors instead of select
issue11268
review364881002
diffstat:
CHANGELOG | 1 +
trytond/bus.py | 10 +++++-----
trytond/cache.py | 11 +++++------
trytond/worker.py | 12 ++++++++----
4 files changed, 19 insertions(+), 15 deletions(-)
diffs (132 lines):
diff -r 4c5e46711589 -r 347d108696ce CHANGELOG
--- a/CHANGELOG Sun Mar 20 01:32:04 2022 +0100
+++ b/CHANGELOG Sun Mar 20 01:40:28 2022 +0100
@@ -1,3 +1,4 @@
+* Use default selectors instead of select
* Add batch option to push queue tasks
* Support ref value for reference field
* Enforce ref model in XML data
diff -r 4c5e46711589 -r 347d108696ce trytond/bus.py
--- a/trytond/bus.py Sun Mar 20 01:32:04 2022 +0100
+++ b/trytond/bus.py Sun Mar 20 01:40:28 2022 +0100
@@ -5,7 +5,7 @@
import json
import logging
import os
-import select
+import selectors
import threading
import time
import uuid
@@ -152,6 +152,7 @@
logger.info("listening on channel '%s'", cls._channel)
conn = db.get_connection(autocommit=True)
pid = os.getpid()
+ selector = selectors.DefaultSelector()
try:
cursor = conn.cursor()
cursor.execute('LISTEN "%s"' % cls._channel)
@@ -159,11 +160,9 @@
cls._messages[database] = messages = _MessageQueue(_cache_timeout)
now = time.time()
+ selector.register(conn, selectors.EVENT_READ)
while cls._queues[pid, database]['timeout'] > now:
- readable, _, _ = select.select([conn], [], [], _select_timeout)
- if not readable:
- continue
-
+ selector.select(timeout=_select_timeout)
conn.poll()
while conn.notifies:
notification = conn.notifies.pop()
@@ -189,6 +188,7 @@
del cls._queues[pid, database]
raise
finally:
+ selector.close()
db.put_connection(conn)
with cls._queues_lock[pid]:
diff -r 4c5e46711589 -r 347d108696ce trytond/cache.py
--- a/trytond/cache.py Sun Mar 20 01:32:04 2022 +0100
+++ b/trytond/cache.py Sun Mar 20 01:40:28 2022 +0100
@@ -4,7 +4,7 @@
import json
import logging
import os
-import select
+import selectors
import threading
import time
from collections import OrderedDict, defaultdict
@@ -355,17 +355,15 @@
logger.info("listening on channel '%s' of '%s'", cls._channel, dbname)
conn = database.get_connection(autocommit=True)
pid = os.getpid()
+ selector = selectors.DefaultSelector()
current_thread = threading.current_thread()
try:
cursor = conn.cursor()
cursor.execute('LISTEN "%s"' % cls._channel)
current_thread.listening = True
-
+ selector.register(conn, selectors.EVENT_READ)
while cls._listener.get((pid, dbname)) == current_thread:
- readable, _, _ = select.select([conn], [], [], 60)
- if not readable:
- continue
-
+ selector.select(timeout=60)
conn.poll()
while conn.notifies:
notification = conn.notifies.pop()
@@ -382,6 +380,7 @@
"cache listener on '%s' crashed", dbname, exc_info=True)
raise
finally:
+ selector.close()
database.put_connection(conn)
with cls._listener_lock[pid]:
if cls._listener.get((pid, dbname)) == current_thread:
diff -r 4c5e46711589 -r 347d108696ce trytond/worker.py
--- a/trytond/worker.py Sun Mar 20 01:32:04 2022 +0100
+++ b/trytond/worker.py Sun Mar 20 01:40:28 2022 +0100
@@ -3,7 +3,7 @@
import datetime as dt
import logging
import random
-import select
+import selectors
import signal
import time
from multiprocessing import Pool as MPool
@@ -61,6 +61,9 @@
tasks = TaskList()
timeout = options.timeout
+ selector = selectors.DefaultSelector()
+ for queue in queues:
+ selector.register(queue.connection, selectors.EVENT_READ)
try:
while True:
while len(tasks.filter()) >= processes:
@@ -73,14 +76,15 @@
tasks.append(queue.run(task_id))
break
else:
- connections = [q.connection for q in queues]
- connections, _, _ = select.select(connections, [], [], timeout)
- for connection in connections:
+ for key, _ in selector.select(timeout=timeout):
+ connection = key.fileobj
connection.poll()
while connection.notifies:
connection.notifies.pop(0)
except KeyboardInterrupt:
mpool.close()
+ finally:
+ selector.close()
def initializer(options, worker=True):
