diff options
| -rwxr-xr-x | tools/merge_fetch.py | 12 |
1 files changed, 7 insertions, 5 deletions
diff --git a/tools/merge_fetch.py b/tools/merge_fetch.py index e0a6243..d6c8cb3 100755 --- a/tools/merge_fetch.py +++ b/tools/merge_fetch.py @@ -176,7 +176,8 @@ def merge_fetch_parallel(args, config, localconfig): procs = {} for storagenode in storagenodes: name = storagenode['name'] - procs[name] = newworker(name, [args, localconfig, storagenode]) + procs[name] = newworker(name, + [args, localconfig, storagenode]) + (storagenode,) currentsizefilecontent = "" # Entries in log, kept in both a set and a list. @@ -194,11 +195,12 @@ def merge_fetch_parallel(args, config, localconfig): ## Poll worker processes and handle messages. assert not fetch_set newentry = [] - for name, pipe, p in procs.values(): + for name, pipe, p, storagenode in procs.values(): if not p.is_alive(): logging.warning("%s is gone, restarting", name) - procs[name] = newworker(name, [args, localconfig, - storagenodes[name]]) + procs[name] = \ + newworker(name, + [args, localconfig, storagenode]) + (storagenode,) continue logging.info("polling %s", name) while pipe.poll(): @@ -241,7 +243,7 @@ def merge_fetch_parallel(args, config, localconfig): l = list(procs.values()) rand.shuffle(l) fetch_dict[e] = cycle(l) - name, pipe, _ = fetch_dict[e].next() + name, pipe, _, _ = fetch_dict[e].next() logging.info("asking %s to FETCH %s", name, hexencode(e)) pipe.send(('FETCH', e)) |