moses-smt/mosesdecoder
1
1
Fork 0
mirror of https://github.com/moses-smt/mosesdecoder.git synced 2024-09-11 11:25:40 +03:00
Code Issues Packages Projects Releases Wiki Activity

Extend multi_moses.py to support multi-threaded moses instances

Browse Source
This commit is contained in:
Michael Denkowski 2015-10-08 11:50:34 -04:00
parent 15b4aa91b0
commit 923f512be0
1 changed files with 131 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

198
scripts/generic/multi_moses.py
View File

@ -7,22 +7,29 @@
'''Parallelize decoding with multiple instances of moses on a local machine
Parallel decoding works well with memory mapped models:
To use with mert-moses.pl, activate --multi-moses and set the number of moses
instances and threads per instance with --decoder-flags='--threads P:T:E'
Compact phrase table and reordering (Marcin Junczys-Dowmunt)
http://www.statmt.org/moses/?n=Advanced.RuleTables#ntoc3
http://www.statmt.org/moses/?n=Advanced.RuleTables#ntoc4
This script runs a specified number of moses instances, each using one or more
threads. The highest speed is generally seen with many single-threaded
instances while the lowest memory usage is seen with a single many-threaded
instance. It is recommended to use the maximum number of instances that will
fit into memory (up to the number of available CPUs) and distribute CPUs across
them equally. For example, a machine with 32 CPUs that can fit 3 copies of
moses into memory would use --threads 2:11:10 for 2 instances with 11 threads
each and an extra instance with 10 threads (3 instances total using all CPUs).
Dynamic phrase table (Ulrich Germann)
Memory mapped models can be shared by multiple processes and increase the number
of instances that can fit into memory:
Mmaped phrase tables (Ulrich Germann)
http://www.statmt.org/moses/?n=Advanced.Incremental#ntoc3
Language model (Kenneth Heafield)
Mmaped mapped language models (Kenneth Heafield)
http://www.statmt.org/moses/?n=FactoredTraining.BuildingLanguageModel#ntoc19
To use with mert-moses.pl, activate --multi-moses and set the number of moses
instances with --decoder-flags='-threads N'
'''
import collections
import os
import Queue
import signal
@ -33,22 +40,37 @@ import threading
HELP = '''Multiple process decoding with Moses
Usage:
{} moses -config moses.ini [options] [decoder flags]
{} moses --config moses.ini [options] [decoder flags]
Options:
-threads N: number of decoders to run in parallel
(default 1, overrides moses.ini)
-n-best-list nbest.out N [distinct]: location and size of N-best list
-show-weights: for mert-moses.pl, just call moses and exit
--threads P:T:E
P: Number of parallel instances to run
T: Number of threads per instance
E: Number of threads in optional extra instance
(default 1:1:0, overrides [threads] in moses.ini. Specifying T
and E is optional, e.g. --threads 16 starts 16 single-threaded
instances)
--n-best-list nbest.out N [distinct]: location and size of N-best list
--show-weights: for mert-moses.pl, just call moses and exit
Other options (decoder flags) are passed through to moses instances
'''
# Defaults
INPUT = sys.stdin
PROCS = 1
THREADS = 1
EXTRA = 0
DONE = threading.Event()
PID = os.getpid()
# A very long time, used as Queue operation timeout even though we don't
# actually want a timeout but we do want interruptibility
# (https://bugs.python.org/issue1360)
NEVER = 60 * 60 * 24 * 365 * 1000
# Single unit of computation: decode a line, output result, signal done
Task = collections.namedtuple('Task', ['id', 'line', 'out', 'event'])
def kill_main(msg):
'''kill -9 the main thread to stop everything immediately'''
@ -61,62 +83,93 @@ def gzopen(f):
return gzip.open(f, 'rb') if f.endswith('.gz') else open(f, 'r')
def run_instance(cmd, tasks, n_best=False):
def run_instance(cmd_base, threads, tasks, n_best=False):
'''Run an instance of moses that processes tasks (input lines) from a
queue'''
# Main loop: take tasks as they are available, complete them, and mark them
# as done. Stop when DONE encountered.
queue using a specified number of threads'''
cmd = cmd_base[:]
cmd.append('--threads')
cmd.append(str(threads))
try:
# Queue of tasks instance is currently working on, limited to the number of
# threads. The queue should be kept full for optimal CPU usage.
work = Queue.Queue(maxsize=threads)
# Multi-threaded instance
moses = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
# Read and handle instance output as available
def handle_output():
while True:
# Output line triggers task completion
line = moses.stdout.readline()
# End of output (instance finished)
if not line:
break
task = work.get(timeout=NEVER)
if n_best:
# Read and copy lines until sentinel line, copy real line id
# id ||| hypothesis words ||| feature scores ||| total score
(first_i, rest) = line.split(' ||| ', 1)
task.out.append(' ||| '.join((task.id, rest)))
while True:
line = moses.stdout.readline()
(i, rest) = line.split(' ||| ', 1)
# Sentinel
if i != first_i:
break
task.out.append(' ||| '.join((task.id, rest)))
else:
task.out.append(line)
# Signal task done
task.event.set()
# Output thread
handler = threading.Thread(target=handle_output, args=())
# Daemon: guaranteed to finish before non-daemons
handler.setDaemon(True)
handler.start()
# Input thread: take tasks as they are available and add them to work
# queue. Stop when DONE encountered.
while True:
(id, line, out, event) = tasks.get()
if event == DONE:
task = tasks.get(timeout=NEVER)
work.put(task, timeout=NEVER)
if task.event == DONE:
break
if n_best:
# Input line followed by blank line (sentinel)
moses.stdin.write(line)
moses.stdin.write(task.line)
moses.stdin.write('\n')
# Read and copy lines until sentinel line, copy real line id
# id ||| hypothesis words ||| feature scores ||| total score
line = moses.stdout.readline()
(first_i, rest) = line.split(' ||| ', 1)
out.append(' ||| '.join((id, rest)))
while True:
line = moses.stdout.readline()
(i, rest) = line.split(' ||| ', 1)
if i != first_i:
break
out.append(' ||| '.join((id, rest)))
else:
moses.stdin.write(line)
out.append(moses.stdout.readline())
event.set()
moses.stdin.write(task.line)
# Cleanup
moses.stdin.close()
moses.wait()
handler.join()
except:
kill_main('Error communicating with moses instance: see stderr')
kill_main('Error with moses instance: see stderr')
def write_results(results, n_best=False, n_best_out=None):
'''Write out results (output lines) from a queue as they are populated'''
while True:
(id, line, out, event) = results.get()
if event == DONE:
task = results.get(timeout=NEVER)
if task.event == DONE:
break
event.wait()
task.event.wait()
if n_best:
# Write top-best and N-best
# id ||| hypothesis words ||| feature scores ||| total score
top_best = out[0].split(' ||| ', 2)[1]
top_best = task.out[0].split(' ||| ', 2)[1]
# Except don't write top-best if writing N-best to stdout "-"
if n_best_out != sys.stdout:
sys.stdout.write('{}\n'.format(top_best))
sys.stdout.flush()
for line in out:
for line in task.out:
n_best_out.write(line)
n_best_out.flush()
else:
sys.stdout.write(out[0])
sys.stdout.write(task.out[0])
sys.stdout.flush()
@ -124,7 +177,9 @@ def main(argv):
# Defaults
moses_ini = None
input = INPUT
procs = PROCS
threads = THREADS
extra = EXTRA
n_best = False
n_best_file = None
n_best_size = None
@ -138,18 +193,23 @@ def main(argv):
# Parse special options and remove from cmd
i = 1
while i < len(cmd):
if cmd[i] in ('-f', '-config'):
if cmd[i] in ('-f', '-config', '--config'):
moses_ini = cmd[i + 1]
# Do not remove from cmd
i += 2
elif cmd[i] in ('-i', '-input-file'):
elif cmd[i] in ('-i', '-input-file', '--input-file'):
input = gzopen(cmd[i + 1])
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] in ('-th', '-threads'):
threads_found = True
threads = int(cmd[i + 1])
elif cmd[i] in ('-th', '-threads', '--threads'):
# P:T:E
args = cmd[i + 1].split(':')
procs = int(args[0])
if len(args) > 1:
threads = int(args[1])
if len(args) > 2:
extra = int(args[2])
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] == '-n-best-list':
elif cmd[i] in ('-n-best-list', '--n-best-list'):
n_best = True
n_best_file = cmd[i + 1]
n_best_size = cmd[i + 2]
@ -160,15 +220,12 @@ def main(argv):
else:
cmd = cmd[:i] + cmd[i + 3:]
# Handled specially for mert-moses.pl
elif cmd[i] == '-show-weights':
elif cmd[i] in ('-show-weights', '--show-weights'):
show_weights = True
# Do not remove from cmd
i += 1
else:
i += 1
# Override thread specification in moses.ini
cmd.append('-threads')
cmd.append('1')
# If mert-moses.pl passes -show-weights, just call moses
if show_weights:
@ -178,25 +235,27 @@ def main(argv):
# Check inputs
if not (len(cmd) > 0 and moses_ini):
sys.stderr.write(HELP.format(argv[0]))
sys.stderr.write(HELP.format(os.path.basename(argv[0])))
sys.exit(2)
if not (os.path.isfile(cmd[0]) and os.access(cmd[0], os.X_OK)):
raise ProgramFailure(
'Error: moses "{}" is not executable\n'.format(cmd[0]))
raise Exception('moses "{}" is not executable\n'.format(cmd[0]))
# Report settings
sys.stderr.write('Moses flags: {}\n'.format(' '.join('\'{}\''.format(s) if ' ' in s else s for s in cmd[1:])))
sys.stderr.write('Instances: {}\n'.format(threads))
sys.stderr.write('Instances: {}\n'.format(procs))
sys.stderr.write('Threads per: {}\n'.format(threads))
if extra:
sys.stderr.write('Extra: {}\n'.format(extra))
if n_best:
sys.stderr.write('N-best list: {} ({}{})\n'.format(n_best_file, n_best_size, ', distinct' if n_best_distinct else ''))
# Task and result queues (buffer 8 * threads input lines)
tasks = Queue.Queue(maxsize=(8 * threads))
# Task and result queues (buffer 8 * total threads input lines)
tasks = Queue.Queue(maxsize=(8 * ((procs * threads) + extra)))
results = Queue.Queue()
# N-best capture
if n_best:
cmd.append('-n-best-list')
cmd.append('--n-best-list')
cmd.append('-')
cmd.append(n_best_size)
if n_best_distinct:
@ -208,9 +267,11 @@ def main(argv):
# Start instances
instances = []
for _ in range(threads):
t = threading.Thread(target=run_instance, args=(cmd, tasks, n_best))
for i in range(procs + (1 if extra else 0)):
t = threading.Thread(target=run_instance, args=(cmd, (threads if i < procs else extra), tasks, n_best))
instances.append(t)
# Daemon: guaranteed to finish before non-daemons
t.setDaemon(True)
t.start()
# Start results writer
@ -224,19 +285,19 @@ def main(argv):
if not line:
break
# (input, out lines, err lines, "done" event)
task = (str(id), line, [], threading.Event())
results.put(task)
tasks.put(task)
task = Task(str(id), line, [], threading.Event())
results.put(task, timeout=NEVER)
tasks.put(task, timeout=NEVER)
id += 1
# Tell instances to exit
for t in instances:
tasks.put((None, None, None, DONE))
tasks.put(Task(None, None, None, DONE), timeout=NEVER)
for t in instances:
t.join()
# Stop results writer
results.put((None, None, None, DONE))
results.put(Task(None, None, None, DONE), timeout=NEVER)
writer.join()
# Cleanup
@ -245,4 +306,7 @@ def main(argv):
if __name__ == '__main__':
main(sys.argv)
try:
main(sys.argv)
except:
kill_main('Error with main I/O: see stderr')