example/core/thread/main.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <iostream>
  10
  11 #include <anna/core/mt/ThreadManager.hpp>
  12 #include <anna/core/util/CommandLine.hpp>
  13 #include <anna/core/mt/Mutex.hpp>
  14 #include <anna/core/tracing/Logger.hpp>
  15 #include <anna/core/mt/Runnable.hpp>
  16 #include <anna/core/mt/Thread.hpp>
  17 #include <anna/core/tracing/TraceWriter.hpp>
  18 #include <anna/core/util/SortedVector.hpp>
  19
  20 #include <anna/io/functions.hpp>
  21
  22 using namespace std;
  23 using namespace anna;
  24
  25 class Client : public Runnable {
  26 public:
  27    Client (const int id, const int value) : Runnable (), a_id (id), a_value (value) {;}
  28
  29 private:
  30    const int a_id;
  31    const int a_value;
  32
  33    void run () throw (RuntimeException);
  34
  35    anna_complete_runnable (Client);
  36 };
  37
  38 class SOAP {
  39 public:
  40    static int theMethod (const int id, const int value) throw (RuntimeException);
  41 };
  42
  43 class Receiver;
  44
  45 class Waiter {
  46 public:
  47    int getId () const throw () { return a_id; }
  48    int getValue () const throw () { return a_value; }
  49
  50    void setId (const int id) throw () { a_id = id; }
  51    void setValue (const int value) throw () { a_value = value; }
  52
  53    void sendRequest () throw (RuntimeException);
  54    int waitResponse () throw (RuntimeException);
  55    void notifyResponse (const int result) throw (RuntimeException);
  56
  57    void initialize () throw (RuntimeException);
  58
  59 private:
  60    static const int ReadChannel = 0;
  61    static const int WriteChannel = 1;
  62
  63    static ThreadManager a_threadManager;
  64
  65    int a_pipe [2];
  66    int a_id;
  67    int a_value;
  68    int a_result;
  69    Receiver* a_receiver;
  70 };
  71
  72 class Context : public SafeRecycler <Waiter> {
  73 public:
  74    Waiter* createWaiter (const int id, const int value) throw (RuntimeException);
  75    void destroyWaiter (Waiter* waiter) throw (RuntimeException);
  76    Waiter* findWaiter (const int id) throw () { return a_waiters.find (id); }
  77
  78 private:
  79    struct SortByID {
  80       static int value (const Waiter* waiter) throw () { return waiter->getId (); }
  81    };
  82
  83    typedef SortedVector <Waiter, SortByID, int> waiter_container;
  84
  85    waiter_container a_waiters;
  86 };
  87
  88 class Receiver : public Runnable {
  89 public:
  90    Receiver (const int id, const int value) : a_id (id), a_value (value) {;}
  91
  92 private:
  93    int a_id;
  94    int a_value;
  95
  96    void run () throw (RuntimeException);
  97
  98    anna_complete_runnable (Waiter);
  99 };
 100
 101 Context* g_context = NULL;
 102 ThreadManager Waiter::a_threadManager ("WaiterThreadManager");
 103
 104 int main (const int argc, const char** argv)
 105 {
 106    CommandLine& ccll = CommandLine::instantiate ();
 107    ThreadManager* threadManager;
 108
 109    g_context = new Context;
 110
 111    try {
 112       ccll.add ("maxthr", CommandLine::Argument::Mandatory, "Maximum number of threads");
 113       ccll.add ("mode", CommandLine::Argument::Mandatory, "Manager instance mode");
 114       ccll.add ("nthr", CommandLine::Argument::Mandatory, "Number of threads to create");
 115       ccll.add ("trace", CommandLine::Argument::Optional, "Trace level");
 116
 117       ccll.initialize (argv, argc);
 118       ccll.verify ();
 119
 120       Logger::initialize ("thread", new TraceWriter ("file.trace", 2048000));
 121
 122       if (ccll.exists ("trace"))
 123          Logger::setLevel (Logger::asLevel (ccll.getValue ("trace")));
 124
 125       ThreadManager::Mode::_v mode = ThreadManager::Mode::asEnum (ccll.getValue ("mode"));
 126
 127       if (mode == ThreadManager::Mode::None) {
 128          string msg (ccll.getValue ("mode"));
 129          msg += " mode not valid (";
 130          msg +=  ThreadManager::Mode::asList ();
 131          msg += ")";
 132          throw RuntimeException (msg, ANNA_FILE_LOCATION);
 133       }
 134
 135       const int maxThread = ccll.getIntegerValue ("maxthr");
 136       int nthr = max (2, ccll.getIntegerValue ("nthr"));
 137
 138       threadManager = new ThreadManager ("MyManager", mode, maxThread, Thread::Flag::Joinable);
 139       Thread* thread = NULL;
 140
 141       // Primero genera los clientes que generan la primera petición de cáloulo. Todos
 142       // terminan llamando al método SOAP::theMethod, que será el que monte la  espera
 143       // y "visualize la respuesta"
 144       for (int ii = 0, min = 0, max = 10; ii < nthr; ii ++) {
 145
 146          Client* client = new Client (ii, min + rand () % max);
 147          // Crear los threads Tx
 148          thread = threadManager->createThread ();
 149          thread->start (*client);
 150
 151          min = max;
 152          max += 10;
 153       }
 154
 155       threadManager->join ();
 156    }
 157    catch (RuntimeException& ex) {
 158       cout << ex.asString () << endl << endl;
 159       ex.trace ();
 160    }
 161 }
 162
 163 // Esto simula el proceso cliente que envía la petición al proceso gSOAP
 164 /* Se ejecuta desde el thread Tx */
 165 void Client::run ()
 166    throw (RuntimeException)
 167 {
 168    int result = SOAP::theMethod (a_id, a_value);
 169
 170    LOGDEBUG (
 171       string msg (asString ());
 172       msg += functions::asText (" | Result: ", result);
 173       Logger::debug (msg, ANNA_FILE_LOCATION);
 174    );
 175 }
 176
 177 // La implementación del método SOAP debe devolver un valor, pero no puede
 178 // calcular por sí misma la respuesta, sino que tiene que apoyarse en un tercero
 179 // que será el que le dé la respuesta "real" en algún momento.
 180 /* Tx */
 181 int SOAP::theMethod (const int id, const int value)
 182    throw (RuntimeException)
 183 {
 184    int result = -1;
 185
 186    Waiter* waiter = NULL;
 187
 188    try {
 189       waiter = g_context->createWaiter (id, value);
 190       waiter->initialize ();
 191       waiter->sendRequest ();
 192       result = waiter->waitResponse ();
 193
 194       cout << "Id: " << id << " | Value: " << value << " | Result: " << result << endl << endl;
 195    }
 196    catch (RuntimeException& ex) {
 197       cout << "Id: " << id << " | Value: " << value << " | " << ex.asString () << endl << endl;
 198       ex.trace ();
 199    }
 200
 201    g_context->destroyWaiter (waiter);  // Sólo lo libera cuando ya no necesita usar sus datos.
 202
 203    return result;
 204 }
 205
 206 /* Tx */
 207 void Waiter::initialize ()
 208    throw (RuntimeException)
 209 {
 210    if (pipe (a_pipe) == -1) {
 211       string msg (functions::asText ("Waiter::initialize | Id: ", a_id));
 212       throw RuntimeException (msg, errno, ANNA_FILE_LOCATION);
 213    }
 214 }
 215
 216 /* Tx */
 217 void Waiter::sendRequest ()
 218    throw (RuntimeException)
 219 {
 220    try {
 221       /* Ojo que esto generará memory-leaks, pero no es el objeto de éste ejemplo la buena gestión de la memoria */
 222       Receiver* receiver = new Receiver (a_id, a_value);
 223       a_threadManager.createThread ()->start (*receiver);
 224    }
 225    catch (RuntimeException& ex) {
 226       string msg ("Waiter::sendRequest | ");
 227       msg += ex.asString ();
 228       throw RuntimeException (msg, ANNA_FILE_LOCATION);
 229    }
 230 }
 231
 232 /* Tx */
 233 int Waiter::waitResponse ()
 234    throw (RuntimeException)
 235 {
 236    bool receive = io::functions::waitInput (a_pipe [ReadChannel], (Millisecond)500);
 237
 238    if (receive == false) {
 239       string msg (anna::functions::asText ("Id: ", a_id));
 240       msg += anna::functions::asText (" | Value: ", a_value);
 241       msg += " | Timeout";
 242       throw RuntimeException (msg, ANNA_FILE_LOCATION);
 243    }
 244
 245    // Esta variable se ha establecido en el notifyResponse (Ty)
 246    return a_result;
 247 }
 248
 249 /* Ty */
 250 void Waiter::notifyResponse (const int result)
 251    throw (RuntimeException)
 252 {
 253    int byte = 0;
 254
 255    a_result = result;
 256
 257    ssize_t r = write (a_pipe [WriteChannel], &byte, sizeof (byte));
 258 }
 259
 260 // Modela al thread que recibe la respuesta correspondiente a la petición por la
 261 // que debería haber un Waiter esperando.
 262 /* Ty */
 263 void Receiver::run ()
 264    throw (RuntimeException)
 265 {
 266    anna::functions::sleep ((Millisecond)(rand () % 1000));
 267
 268    int result = a_value + a_value;
 269
 270    // Protege el contexto para evitar condiciones de carrera, como que en el "mismo" instante
 271    // en que el receptor decide responder, el waiter en Tx decida que ha terminado de esperar
 272    // la respuesta.
 273    Guard guard (g_context, "Receiver::run");
 274
 275    Waiter* waiter = g_context->findWaiter (a_id);
 276
 277    // Si el Waiter se cansó de esperar la respuesta ...
 278    if (waiter == NULL)
 279       return;
 280
 281    waiter->notifyResponse (result);
 282 }
 283
 284 Waiter* Context::createWaiter (const int id, const int value)
 285    throw (RuntimeException)
 286 {
 287    Guard guard (*this, "Context::createWaiter");
 288
 289    Waiter* result = a_waiters.find (id);
 290
 291    if (result != NULL)
 292       throw RuntimeException ("Waiter already used", ANNA_FILE_LOCATION);
 293
 294    result = Recycler <Waiter>::create ();
 295
 296    result->setId (id);
 297    result->setValue (value);
 298
 299    a_waiters.add (result);
 300
 301    return result;
 302 }
 303
 304 void Context::destroyWaiter (Waiter* waiter)
 305    throw (RuntimeException)
 306 {
 307    Guard guard (this, "Context::destroyWaiter");
 308
 309    // Lo elimina de la lista ordenada
 310    a_waiters.erase (waiter);
 311
 312    // Lo marca como disponible para usar
 313    Recycler <Waiter>::release (waiter);
 314 }
 315