Skip to content
Snippets Groups Projects
Commit 347d1086 authored by Cédric Krier's avatar Cédric Krier :atom:
Browse files

Use default selectors instead of select

issue11268
review364881002
parent 4c5e4671
No related branches found
No related tags found
1 merge request!114Return an empty list when the treeview has no selection
* 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
......
......@@ -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,4 +160,5 @@
cls._messages[database] = messages = _MessageQueue(_cache_timeout)
now = time.time()
selector.register(conn, selectors.EVENT_READ)
while cls._queues[pid, database]['timeout'] > now:
......@@ -162,8 +164,5 @@
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]:
......
......@@ -4,7 +4,7 @@
import json
import logging
import os
import select
import selectors
import threading
import time
from collections import OrderedDict, defaultdict
......@@ -355,8 +355,9 @@
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
......@@ -358,7 +359,7 @@
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:
......@@ -364,8 +365,5 @@
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:
......
......@@ -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,11 +76,10 @@
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()
......@@ -79,8 +81,10 @@
connection.poll()
while connection.notifies:
connection.notifies.pop(0)
except KeyboardInterrupt:
mpool.close()
finally:
selector.close()
def initializer(options, worker=True):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment