hans/Nominatim
1
0
Fork 1
mirror of https://github.com/osm-search/Nominatim.git synced 2026-03-11 21:34:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

use generator for thread choice

Browse Source
This commit is contained in:
Sarah Hoffmann
2020-01-19 14:51:15 +01:00
parent c11d1d78e9
commit 2a15b2522f
1 changed files with 21 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
nominatim/nominatim.py
View File

@@ -56,6 +56,7 @@ class IndexingThread(object):
self.wait() self.wait()
self.current_query = None self.current_query = None
self.current_params = None
def wait(self): def wait(self):
wait_select(self.conn) wait_select(self.conn)
@@ -63,8 +64,12 @@ class IndexingThread(object):
def perform(self, sql, args=None): def perform(self, sql, args=None):
self.current_query = sql self.current_query = sql
self.current_params = args
self.cursor.execute(sql, args) self.cursor.execute(sql, args)
def fileno(self):
return self.conn.fileno()
def is_done(self): def is_done(self):
if self.current_query is None: if self.current_query is None:
return True return True
@@ -78,7 +83,7 @@ class IndexingThread(object):
raise RuntimeError("Postgres exception has no error code") raise RuntimeError("Postgres exception has no error code")
if e.pgcode == '40P01': if e.pgcode == '40P01':
log.info("Deadlock detected, retry.") log.info("Deadlock detected, retry.")
self.cursor.execute(self.sql) self.cursor.execute(self.current_query, self.current_params)
else: else:
raise raise
@@ -95,8 +100,7 @@ class Indexer(object):
for i in range(options.threads): for i in range(options.threads):
t = IndexingThread(i, options) t = IndexingThread(i, options)
self.threads.append(t) self.threads.append(t)
self.poll.register(t.conn.fileno(), select.EPOLLIN) self.poll.register(t, select.EPOLLIN)
self.next_thread = 0
def run(self): def run(self):
log.info("Starting indexing rank ({} to {}) using {} threads".format( log.info("Starting indexing rank ({} to {}) using {} threads".format(
@@ -123,6 +127,7 @@ class Indexer(object):
cur.scroll(0, mode='absolute') cur.scroll(0, mode='absolute')
next_thread = self.find_free_thread()
done_tuples = 0 done_tuples = 0
rank_start_time = datetime.now() rank_start_time = datetime.now()
for r in cur: for r in cur:
@@ -141,7 +146,7 @@ class Indexer(object):
for place in pcur: for place in pcur:
place_id = place[0] place_id = place[0]
log.debug("Processing place {}".format(place_id)) log.debug("Processing place {}".format(place_id))
thread = self.find_free_thread() thread = next(next_thread)
thread.perform(obj.sql_index_place(), (place_id,)) thread.perform(obj.sql_index_place(), (place_id,))
done_tuples += 1 done_tuples += 1
@@ -164,12 +169,19 @@ class Indexer(object):
done_tuples/diff_seconds, obj.name())) done_tuples/diff_seconds, obj.name()))
def find_free_thread(self): def find_free_thread(self):
while True: thread_lookup = { t.fileno() : t for t in self.threads}
for t in self.threads:
if t.is_done():
return t
self.poll.poll() done_fids = [ t.fileno() for t in self.threads ]
while True:
for fid in done_fids:
thread = thread_lookup[fid]
if thread.is_done():
yield thread
else:
print("not good", fid)
done_fids = [ x[0] for x in self.poll.poll()]
assert(False, "Unreachable code") assert(False, "Unreachable code")