



\section{
\label{code-guide-threading.tex}
Multi-Threading}


Modern server machines have multiple CPUs, each with multiple cores. Utilizing all these cores requires either running multiple processes on the same machine or writing programs that use multiple threads.



Since many aspects of a machine translation system (training, tuning, using) lend themselves very easily to parallel processing, Moses increasingly uses multi-threading in its components. At this point, the following components allow for parallel execution when the switch "-{}-threads NUM" is added with an appropriate maximum number of threads executed at the same time:

\begin{compactitem}
\item the decoder binary {\tt moses}

\item the minimum error rate training tuner {\tt mert}

\item the hierarchical rule extractor {\tt extract-rules}

\end{compactitem}

Multi-threading in Moses is based on the C++ Boost libraries, and two Moses helper libraries that make the type of multi-threading that is typical for Moses more convenient: {\tt ThreadPool} and {\tt OutputCollector}.



We will explain the implementation of multi-threaded processing on hand of a simple example.



\subsection{
Tasks}


The part of the program that is to be run in parallel threads is called a {\bf task}, and it needs to be placed into a class of its own.






\begin{SaveVerbatim}{myverb}
class ExampleTask : public Moses::Task
{
public:
ExampleTask() {}
~ExampleTask() {}

void Run() {
std::cout << "Hello World." << endl;
}
}
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

Such a class now allows to be instantiated and run:






\begin{SaveVerbatim}{myverb}
ExampleTask *task = new ExampleTask()
new->Run();
delete(new);
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

This will print {\tt "Hello World."}, and is otherwise not very exciting.



Let's make the task a bit more interesting. Our new tasks waits for a random amount of time and then prints out a message:






\begin{SaveVerbatim}{myverb}
ExampleTask(string message):m_message(message) {}

void Run() {
// length of pause
int r = rand()%10;

// pause
int j = 0;
for(int i=0; i<1e8*r; i++) { j+=i; }

// write message (and length of pause)
std::cout << m_message << " (" << r << ")" << endl;
}
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

We can now create multiple instances of this task, and execute each:






\begin{SaveVerbatim}{myverb}
// set up tasks
srand(time(NULL));
ExampleTask *task0 = new ExampleTask("zero");
ExampleTask *task1 = new ExampleTask("one");
ExampleTask *task2 = new ExampleTask("two");

// serial execution
task0->Run();
task1->Run();
task2->Run();
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

This will print out three lines (the random numbers in parenthesis will vary):






\begin{SaveVerbatim}{myverb}
zero (2)
one (4)
two (5)
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

Okay, where is the multi-threading? Here it comes. 



\subsection{
ThreadPool}


Instead of simply running one of the tasks after the other, we assign them to a thread pool. Once assigned, they are spawned off to a thread and will be executed in parallel to the running main process.






\begin{SaveVerbatim}{myverb}
// set up thread pool
int thread_count = 10;
Moses::ThreadPool pool(thread_count);

// submit tasks
pool.Submit(task0);
pool.Submit(task1);
pool.Submit(task2);

// wait for all threads to finish
pool.Stop(true);
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

That's all too easy to be true, right? Yes, it is. 



Since the three threads are running in parallel, there is no telling when they print out their message. Not only could the lines be printed in a different order then the tasks were scheduled, the threads may even write all over each other.



This is the catch with multi-threading: any interaction with non-local data structures must be handled very carefully. Ideally, threads only change local data (defined in the class), and once they are done (after {\tt pool.Stop(true)}), results can be read out. This is in fact what happens in multi-threaded {\tt mert}\footnote{\sf http://www.statmt.org/moses/html/d6/d7d/mert\_8cpp\_source.html}.



In our case, as in the decoder, we want to output text line by line (the decoder outputs translation, and possibly additional information such as n-best lists). 



\subsection{
OutputCollector}


The Moses code offers the class {\tt OutputCollector} to buffer up the output until it is safe to print out. In the simplest case, it prints to {\tt STDOUT}, but it can also write to a file, and indeed it offers both regular output (default {\tt STDOUT}) and debugging output (default {\tt STDERR}), which both can be redirected to different files.






\begin{SaveVerbatim}{myverb}
Moses::OutputCollector* outputCollector = new Moses::OutputCollector();
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

A task can then send its output to the output collector with the function {\tt Write}, for example:






\begin{SaveVerbatim}{myverb}
m_collector->Write(id, "Hello World!");
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

The {\tt id} is the sequential number of the sentence, starting at 0. This helps the output collector to keep track of what can be written out and what needs to be buffered. The output collector will not write output for sentence 1, if it has not yet received output for sentence 0.



\subsection{
Not Deleting Threads after Execution}


By default, the {\tt Task} objects are deleted after execution. However, you may want to keep the objects around. This happens for instance in {\tt mert}, where each {\tt Task} finds an optimized weight setting, which is to be processed afterwards. In this case, you have to add the following lines to your {\tt Task} definition:






\begin{SaveVerbatim}{myverb}
virtual bool DeleteAfterExecution() {
return false;
}
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

\subsection{
Limit the Size of the Thread Queue}


By default, when a thread is submitted to the {\tt ThreadPool} by calling its {\tt Submit()} function, it is added to an internal queue, and the main process immediately resumes. That means, if a million threads are scheduled, the thread queue is filled with a million instances of the {\tt Task}, which may consume a lot of memory.



If you want to restrict the number of threads in the queue, you can call, say, {\tt pool.SetQueueLimit(1000)} to limit it to 1000 queued {\tt Task} instances. When the queue is full, {\tt Submit()} blocks.



\subsection{
Example}


Below now the complete example.



Note:

\begin{compactitem}
\item The task class has now two more class variables which are set upon instantiation: the sequence id {\tt m\_id} (a sequential number starting at 0), and a pointer to the output collector {\tt m\_collector}.

\item Always implement a fallback to non-threaded compilation ({\tt \#ifdef WITH\_THREADS} .. {\tt \#else} .. {\tt \#endif})

\item Output is placed into a file named {\tt output-file.txt} (lines 43-45) instead of {\tt STDOUT}.

\end{compactitem}




\begin{SaveVerbatim}{myverb}
01: #include <iostream>
02: #include <fstream>
03: #include <ostream>
04: #include <cstdlib>
05: #include <sstream>
06: #include "ThreadPool.h"
07: #include "OutputCollector.h"
08: 
09: using namespace std;
10:
11: class ExampleTask : public Moses::Task
12: {
13: private:
14:   unsigned int m_id;
15:   string m_message;
16:   Moses::OutputCollector* m_collector;
17: public:
18:   ExampleTask(unsigned int id, string message, Moses::OutputCollector* collector):
19:     m_id(id),
20:     m_message(message),
21:     m_collector(collector) {}
22: 
23:   ~ExampleTask() {}
24: 
25:   void Run() {
26:     // length of pause
27:     int r = rand()%10;
28: 
29:     // pause
30:     int j = 0;
31:     for(int i=0; i<1e8*r; i++) { j+=i; }
32: 
33:     // write message (and length of pause)
34:     ostringstream out;
35:     out << m_message << " (" << r << ")" << endl;
36:     m_collector->Write(m_id, out.str());
37:   }
38: };
39: 
40: int main ()
41: {
42:   // output into file
43:   string outfile = "output-file.txt";
44:   std::ofstream *outputStream = new ofstream(outfile.c_str());
45:   Moses::OutputCollector* outputCollector = new Moses::OutputCollector(outputStream);
46: 
47:   // set up tasks
48:   srand(time(NULL));
49:   ExampleTask *task0 = new ExampleTask(0,"zero",outputCollector);
50:   ExampleTask *task1 = new ExampleTask(1,"one",outputCollector);
51:   ExampleTask *task2 = new ExampleTask(2,"two",outputCollector);
52: 
53: #ifdef WITH_THREADS
54:   // set up thread pool
55:   int thread_count = 10;
56:   Moses::ThreadPool pool(thread_count);
57: 
58:   // submit tasks
59:   pool.Submit(task0);
60:   pool.Submit(task1);
61:   pool.Submit(task2);
62: 
63:   // wait for all threads to finish
64:   pool.Stop(true);
65: #else
66:   // fallback: serial execution
67:   task0->Run();
68:   task1->Run();
69:   task2->Run();
70: #endif
71: }
\end{SaveVerbatim}
\begin{shaded}
\scriptsize
\UseVerbatim{myverb}
\end{shaded}

To compile this, you need to copy {\tt ThreadPool.h}, {\tt ThreadPool.cpp}, and {\tt OutputCollector.h} into you code directory or add paths so that they point to the {\tt moses/src} directory and compile as follows:






\begin{SaveVerbatim}{myverb}
g++ -c ThreadPool.cpp -DWITH_THREADS -DBOOST_HAS_PTHREADS
g++ -c test.cpp -DWITH_THREADS -DBOOST_HAS_PTHREADS
g++ -o test test.o ThreadPool.o -pthread -lboost_thread-mt
\end{SaveVerbatim}
\begin{shaded}
\small
\UseVerbatim{myverb}
\end{shaded}

Make sure that the Boost libraries are in you compile paths.



When you run this example you will notice that, whatever the lengths of the pauses, the output always appears in the correct order (i.e. zero, one, two).







\begin{center}\textit{\small Subsection last  modified on April 17, 2012, at 11:49 PM}\end{center}
