mirrors/pgloader
1
0
Fork 0
mirror of https://github.com/dimitri/pgloader.git synced 2026-01-22 07:31:04 +01:00
Code Issues Projects Releases Wiki Activity

FIX round-robin reader case when not all configured threads are needed to load the file

Browse Source
This commit is contained in:
dim 2008-02-16 19:17:03 +00:00
parent a27a801f62
commit 84738febfd
2 changed files with 18 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/parallel.conf
View File

@ -12,7 +12,7 @@ lc_messages = C
;client_encoding = 'utf-8'
client_encoding = 'latin1'
copy_every = 50
copy_every = 5000
commit_every = 5
#copy_delimiter = %
@ -44,7 +44,7 @@ split_file_reading = True
[rrr_csv]
format = csv
use_template = ptmpl
section_threads = 2
section_threads = 3
split_file_reading = False
rrqueue_size = -1

18
pgloader/pgloader.py
View File

@ -1009,6 +1009,10 @@ class PGLoader(threading.Thread):
p = c
c = (n / self.rrqueue_size) % self.section_threads
# save this for later reference
last_p = p
last_c = c
# we could have some locks to release here
self.log.debug("p=%d c=%d n=%d (n/rrqueue_size)%%%d=%d " \
% (p, c, n,
@ -1022,12 +1026,22 @@ class PGLoader(threading.Thread):
self.log.debug("locks[%d].release" % p)
locks[p].release()
# we could have read all the data and not needed all workers,
# log it when it's the case, then set .done = True without
# taking again a lock which we already have.
if n < (self.section_threads * self.rrqueue_size):
self.log.info("processed all data with only %d workers" % (c+1))
# mark all worker threads has done
k = threads.keys()
for c in range(self.section_threads):
self.log.debug("locks[%d].acquire to set %s.done = True" \
# we don't need any lock.acquire if we didn't use the worker
if n > (self.section_threads * self.rrqueue_size) \
or c <= last_c:
self.log.debug("locks[%d].acquire to set %s.done = True" \
% (c, k[c]))
locks[c].acquire()
locks[c].acquire()
threads[k[c]].done = True