6061247d199c381dc2bce53f810be383c8ad8867
[anna.git] / example / core / thread / main.cpp
1 // ANNA - Anna is Not 'N' Anymore
2 //
3 // (c) Copyright 2005-2014 Eduardo Ramos Testillano & Francisco Ruiz Rayo
4 //
5 // https://bitbucket.org/testillano/anna
6 //
7 // Redistribution and use in source and binary forms, with or without
8 // modification, are permitted provided that the following conditions
9 // are met:
10 //
11 //     * Redistributions of source code must retain the above copyright
12 // notice, this list of conditions and the following disclaimer.
13 //     * Redistributions in binary form must reproduce the above
14 // copyright notice, this list of conditions and the following disclaimer
15 // in the documentation and/or other materials provided with the
16 // distribution.
17 //     * Neither the name of Google Inc. nor the names of its
18 // contributors may be used to endorse or promote products derived from
19 // this software without specific prior written permission.
20 //
21 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22 // "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23 // LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24 // A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25 // OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27 // LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28 // DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29 // THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30 // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32 //
33 // Authors: eduardo.ramos.testillano@gmail.com
34 //          cisco.tierra@gmail.com
35
36
37 #include <iostream>
38
39 #include <anna/core/mt/ThreadManager.hpp>
40 #include <anna/core/util/CommandLine.hpp>
41 #include <anna/core/mt/Mutex.hpp>
42 #include <anna/core/tracing/Logger.hpp>
43 #include <anna/core/mt/Runnable.hpp>
44 #include <anna/core/mt/Thread.hpp>
45 #include <anna/core/tracing/TraceWriter.hpp>
46 #include <anna/core/util/SortedVector.hpp>
47
48 #include <anna/io/functions.hpp>
49
50 using namespace std;
51 using namespace anna;
52
53 class Client : public Runnable {
54 public:
55    Client (const int id, const int value) : Runnable (), a_id (id), a_value (value) {;}
56
57 private:
58    const int a_id;
59    const int a_value;
60
61    void run () throw (RuntimeException);
62
63    anna_complete_runnable (Client);
64 };
65
66 class SOAP {
67 public:
68    static int theMethod (const int id, const int value) throw (RuntimeException);
69 };
70
71 class Receiver;
72
73 class Waiter {
74 public:
75    int getId () const throw () { return a_id; }
76    int getValue () const throw () { return a_value; }
77
78    void setId (const int id) throw () { a_id = id; }
79    void setValue (const int value) throw () { a_value = value; }
80
81    void sendRequest () throw (RuntimeException);
82    int waitResponse () throw (RuntimeException);
83    void notifyResponse (const int result) throw (RuntimeException);   
84
85    void initialize () throw (RuntimeException);
86    
87 private:
88    static const int ReadChannel = 0;
89    static const int WriteChannel = 1;
90    
91    static ThreadManager a_threadManager;
92
93    int a_pipe [2];
94    int a_id;
95    int a_value;
96    int a_result;
97    Receiver* a_receiver;
98 };
99
100 class Context : public SafeRecycler <Waiter> {
101 public:  
102    Waiter* createWaiter (const int id, const int value) throw (RuntimeException);
103    void destroyWaiter (Waiter* waiter) throw (RuntimeException);
104    Waiter* findWaiter (const int id) throw () { return a_waiters.find (id); }
105
106 private:
107    struct SortByID {
108       static int value (const Waiter* waiter) throw () { return waiter->getId (); }
109    };
110
111    typedef SortedVector <Waiter, SortByID, int> waiter_container;
112
113    waiter_container a_waiters;
114 };
115
116 class Receiver : public Runnable {
117 public:
118    Receiver (const int id, const int value) : a_id (id), a_value (value) {;}
119
120 private:
121    int a_id;
122    int a_value;
123
124    void run () throw (RuntimeException);
125
126    anna_complete_runnable (Waiter);
127 };
128
129 Context* g_context = NULL;
130 ThreadManager Waiter::a_threadManager ("WaiterThreadManager");
131
132 int main (const int argc, const char** argv)
133 {
134    CommandLine& ccll = CommandLine::instantiate ();
135    ThreadManager* threadManager;
136    
137    g_context = new Context;
138
139    try {
140       ccll.add ("maxthr", CommandLine::Argument::Mandatory, "Maximum number of threads");
141       ccll.add ("mode", CommandLine::Argument::Mandatory, "Manager instance mode");
142       ccll.add ("nthr", CommandLine::Argument::Mandatory, "Number of threads to create");
143       ccll.add ("trace", CommandLine::Argument::Optional, "Trace level");
144
145       ccll.initialize (argv, argc);
146       ccll.verify ();
147
148       Logger::initialize ("thread", new TraceWriter ("file.trace", 2048000));
149
150       if (ccll.exists ("trace"))
151          Logger::setLevel (Logger::asLevel (ccll.getValue ("trace")));
152          
153       ThreadManager::Mode::_v mode = ThreadManager::Mode::asEnum (ccll.getValue ("mode"));
154
155       if (mode == ThreadManager::Mode::None) {
156          string msg (ccll.getValue ("mode"));
157          msg += " mode not valid (";
158          msg +=  ThreadManager::Mode::asList ();
159          msg += ")";
160          throw RuntimeException (msg, ANNA_FILE_LOCATION);
161       }
162
163       const int maxThread = ccll.getIntegerValue ("maxthr");
164       int nthr = max (2, ccll.getIntegerValue ("nthr"));
165      
166       threadManager = new ThreadManager ("MyManager", mode, maxThread, Thread::Flag::Joinable);
167       Thread* thread = NULL;
168
169       // Primero genera los clientes que generan la primera petición de cáloulo. Todos
170       // terminan llamando al método SOAP::theMethod, que será el que monte la  espera
171       // y "visualize la respuesta"
172       for (int ii = 0, min = 0, max = 10; ii < nthr; ii ++) {
173           
174          Client* client = new Client (ii, min + rand () % max);
175          // Crear los threads Tx 
176          thread = threadManager->createThread ();
177          thread->start (*client);
178
179          min = max;
180          max += 10;
181       }
182       
183       threadManager->join ();
184    }
185    catch (RuntimeException& ex) {
186       cout << ex.asString () << endl << endl;
187       ex.trace ();
188    }
189 }
190
191 // Esto simula el proceso cliente que envía la petición al proceso gSOAP
192 /* Se ejecuta desde el thread Tx */
193 void Client::run () 
194    throw (RuntimeException)
195 {
196    int result = SOAP::theMethod (a_id, a_value);
197
198    LOGDEBUG (
199       string msg (asString ());
200       msg += functions::asText (" | Result: ", result);
201       Logger::debug (msg, ANNA_FILE_LOCATION);
202    );
203 }
204
205 // La implementación del método SOAP debe devolver un valor, pero no puede 
206 // calcular por sí misma la respuesta, sino que tiene que apoyarse en un tercero
207 // que será el que le dé la respuesta "real" en algún momento.
208 /* Tx */
209 int SOAP::theMethod (const int id, const int value)
210    throw (RuntimeException)
211 {
212    int result = -1;
213
214    Waiter* waiter = NULL;
215
216    try {
217       waiter = g_context->createWaiter (id, value);
218       waiter->initialize ();
219       waiter->sendRequest (); 
220       result = waiter->waitResponse ();
221       
222       cout << "Id: " << id << " | Value: " << value << " | Result: " << result << endl << endl;      
223    }
224    catch (RuntimeException& ex) {
225       cout << "Id: " << id << " | Value: " << value << " | " << ex.asString () << endl << endl;
226       ex.trace ();
227    }  
228
229    g_context->destroyWaiter (waiter);  // Sólo lo libera cuando ya no necesita usar sus datos.
230    
231    return result;
232 }
233
234 /* Tx */
235 void Waiter::initialize () 
236    throw (RuntimeException)
237 {
238    if (pipe (a_pipe) == -1) {
239       string msg (functions::asText ("Waiter::initialize | Id: ", a_id));
240       throw RuntimeException (msg, errno, ANNA_FILE_LOCATION); 
241    }     
242 }
243
244 /* Tx */
245 void Waiter::sendRequest () 
246    throw (RuntimeException)
247 {
248    try {
249       /* Ojo que esto generará memory-leaks, pero no es el objeto de éste ejemplo la buena gestión de la memoria */
250       Receiver* receiver = new Receiver (a_id, a_value);   
251       a_threadManager.createThread ()->start (*receiver);         
252    }
253    catch (RuntimeException& ex) {
254       string msg ("Waiter::sendRequest | ");
255       msg += ex.asString ();
256       throw RuntimeException (msg, ANNA_FILE_LOCATION);
257    }
258 }
259
260 /* Tx */
261 int Waiter::waitResponse () 
262    throw (RuntimeException)
263 {
264    bool receive = io::functions::waitInput (a_pipe [ReadChannel], (Millisecond)500);
265    
266    if (receive == false) {      
267       string msg (anna::functions::asText ("Id: ", a_id));
268       msg += anna::functions::asText (" | Value: ", a_value);
269       msg += " | Timeout";            
270       throw RuntimeException (msg, ANNA_FILE_LOCATION);     
271    }
272    
273    // Esta variable se ha establecido en el notifyResponse (Ty)
274    return a_result;
275 }
276
277 /* Ty */
278 void Waiter::notifyResponse (const int result) 
279    throw (RuntimeException)
280 {
281    int byte = 0;
282    
283    a_result = result;
284    
285    write (a_pipe [WriteChannel], &byte, sizeof (byte));
286 }
287
288 // Modela al thread que recibe la respuesta correspondiente a la petición por la 
289 // que debería haber un Waiter esperando.
290 /* Ty */
291 void Receiver::run () 
292    throw (RuntimeException)
293 {
294    anna::functions::sleep ((Millisecond)(rand () % 1000));
295
296    int result = a_value + a_value;
297
298    // Protege el contexto para evitar condiciones de carrera, como que en el "mismo" instante
299    // en que el receptor decide responder, el waiter en Tx decida que ha terminado de esperar
300    // la respuesta.
301    Guard guard (g_context, "Receiver::run");
302    
303    Waiter* waiter = g_context->findWaiter (a_id);
304
305    // Si el Waiter se cansó de esperar la respuesta ...
306    if (waiter == NULL)
307       return;
308
309    waiter->notifyResponse (result);
310 }
311
312 Waiter* Context::createWaiter (const int id, const int value) 
313    throw (RuntimeException)
314 {
315    Guard guard (*this, "Context::createWaiter");
316    
317    Waiter* result = a_waiters.find (id);
318    
319    if (result != NULL) 
320       throw RuntimeException ("Waiter already used", ANNA_FILE_LOCATION);
321    
322    result = Recycler <Waiter>::create ();
323    
324    result->setId (id);
325    result->setValue (value);
326    
327    a_waiters.add (result);
328    
329    return result;   
330 }
331
332 void Context::destroyWaiter (Waiter* waiter) 
333    throw (RuntimeException)
334 {
335    Guard guard (this, "Context::destroyWaiter");
336
337    // Lo elimina de la lista ordenada
338    a_waiters.erase (waiter);   
339    
340    // Lo marca como disponible para usar
341    Recycler <Waiter>::release (waiter);   
342 }
343