include/anna/comm/CongestionController.hpp

   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 #ifndef anna_comm_CongestionController_hpp
  10 #define anna_comm_CongestionController_hpp
  11
  12 #include <utility>
  13
  14 #include <anna/config/defines.hpp>
  15 #include <anna/core/Singleton.hpp>
  16 #include <anna/core/mt/NRMutex.hpp>
  17 #include <anna/core/util/Average.hpp>
  18
  19 namespace anna {
  20
  21 namespace xml {
  22 class Node;
  23 }
  24
  25 namespace comm {
  26
  27 class ClientSocket;
  28
  29 namespace handler {
  30 class LocalConnection;
  31 }
  32
  33 /**
  34    Gestor de congestion autonomo.
  35
  36 Su funcionamiento a grandes rasgos es el siguiente:
  37
  38    -# El campo \em getsockopt (...SO_RCVBUF...) nos da el tamaño maximo del buffer de recepcion
  39    para un socket dado. Segun hemos visto puede varias desde 32K hasta 80 K, dependiendo del sistema
  40    operativo y la configuracion. Independientemente del valor particular este sera nuestro limite
  41    maximo (100%), ya que si la longitud de la cola de espera alcanza este valor, el socket dejara de
  42    estar disponible para todos los clientes.
  43    -# El metodo \em ioctl (....FIONREAD...) nos da el numero de bytes pendientes de tratar en el socket.
  44    -# La carga del socket estara definida por (2) * 100 / (1).
  45    -# Con esta clase podemos limitar la carga maxima soportada por los sockets de nuestra
  46    aplicacion. Por defecto sera 60% del valor (1), un valor mas conservador podria ser 40%.
  47    -# La zona sin congestion estara entre 0% y el valor (4). La zona de congestion estara entre (valor (4) + 1)%  y el 100%.
  48    -# La zona de congestion se divide, a su vez, en 4 zonas logicas: Si la carga del socket esta en el
  49    primer 25% se descartan mensajes con una probabilidad del 25%. Si la carga esta en el segundo 25% se
  50    descartan mensajes con una probabilidad del 50%. Si la carga esta entre el 51% y el 85% se descartan
  51    paquetes con una probabilidad del 85% y si esta entre el 86% y el 100% se descartan los paquetes con
  52    una probabilidad del 99%.
  53
  54 Ventajas:
  55
  56    \li Trata cada socket de forma independiente, con lo que ajustamos mucho mejor el rendimiento de
  57    nuestra aplicacion.
  58    \li No hay que imaginar los 4 niveles de a8_controlniveles. Solo tendremos que establecer el nivel
  59    de carga maximo y el resto lo hace el proceso por si mismo. Sea como sea, es mucho mas facil ajustar
  60    un unico valor que los 8 o 10 necesarios con el sistema anterior.
  61    \li No requiere ninguna configuracion en base de datos.
  62 */
  63 class CongestionController : public Singleton <CongestionController> {
  64 public:
  65   /**
  66    * Nivel máximo de congestión.
  67    */
  68   static const int MaxLevel = 4;
  69
  70   /**
  71      Posibles resultados de CongestionController::getAdvice.
  72   */
  73   struct Advice {
  74     enum _v {
  75       None, /**< No hay datos para tomar una decision.  */
  76       Process, /**< Se puede procesar el mensaje sin contribuir a la congestion */
  77       Discard /**< Se aconseja descartar para limitar la congestion */
  78     };
  79   };
  80
  81   /**
  82      Posibles formas de calcular la carga que hara aconsejar el descarte o aceptacion de un mensaje.
  83   */
  84   struct Mode {
  85     enum _v {
  86       Auto, /**< Escoge la forma de calculo que mejor se adapte al modo de compilacion. Normalmente
  87                 El calculo de tipo Local sera mas util en modo MT, mientras que el Global puede ser
  88                 mas aconsejable en modo ST.
  89                 */
  90       Local, /**< El calculo de la carga se calcula de forma independiente para cada Socket */
  91       Global /**< El calculo de la carga se calcula de forma conjunta para todos los Sockets */
  92     };
  93   };
  94
  95   typedef std::pair <int, int> Workload;
  96
  97   static const int DefaultLimit = 60;
  98
  99   /**
 100    * Valor máximo que se puede asignar al control de bytes pendientes de tratar.
 101    */
 102   static const int MaxPendingBytes;
 103
 104   /**
 105      Devuelve el modo que estamos usando para calcular la carga.
 106      \return El modo que estamos usando para calcular la carga.
 107   */
 108   Mode::_v getMode() const throw() { return a_mode; }
 109
 110   /**
 111    * Devuelve el número total de mensajes recibididos.
 112    * \return el número total de mensajes recibididos.
 113    */
 114   int getMessageCounter() const throw() { return a_messageCounter; }
 115
 116   /**
 117    * Devuelve el número de mensajes tratados correctamente.
 118    * \return el número de mensajes tratados correctamente.
 119    */
 120   int getSuccessCounter() const throw() { return a_messageCounter - a_discardedCounter; }
 121
 122   /**
 123      Establece el limite de carga que vamos a imponer.
 124      \param limit Limite de carga. Debe ser un valor entre 0 y 100.
 125   */
 126   void setLimit(const int limit) throw();
 127
 128   /**
 129      Establece el modo en que vamos a calcular la carga.
 130      \param mode Modo usado para calcular la carga.
 131   */
 132   void setMode(const Mode::_v mode) throw() {
 133     if((a_mode = mode) == Mode::Auto) {
 134       WHEN_MULTITHREAD(a_effectiveMode = Mode::Local);
 135       WHEN_SINGLETHREAD(a_effectiveMode = Mode::Global);
 136     } else
 137       a_effectiveMode = mode;
 138   }
 139
 140   /**
 141      Devuelve \em true si esta instancia ha recibido datos o \em false en otro caso.
 142      \return \em true si esta instancia ha recibido datos o \em false en otro caso.
 143   */
 144   bool isEmpty() const throw() { return a_avgWorkload.isEmpty(); }
 145
 146   /**
 147    *  Establece el nº máximo de bytes que deberían tener los socket en la cola de entrada.
 148    * \param maxPendingBytes Número máximo de bytes que debería tener los socket en la cola de entrada.
 149    * \warning El valor indicado debe estar dentro del siguiente ámbito:
 150    * \code
 151    * [#comm::Communicator::getReadingChunkSize, min (#comm::Communicator::getReadingChunkSize * 4, #MaxPendingBytes)]
 152    * \endcode
 153    */
 154   void setMaxPendingBytes(const int maxPendingBytes) throw(RuntimeException);
 155
 156   /**
 157      Devuelve el consejo sobre si debemos tratar/o no el ultimo mensaje recibido
 158      por el ClientSocket recibido como parametro.
 159      \param clientSocket Socket cliente por el que hemos recibido el ultimo mensaje.
 160      \return Advice::Process para indicar que debemos procesar el mensaje
 161      , Advice::None para indicar que todavia no tiene datos para tomar una
 162      decision clara o Advice::Discard para indicar que debemos descartar.
 163   */
 164   Advice::_v getAdvice(const ClientSocket& clientSocket) throw();
 165
 166   /**
 167      Devuelve el consejo sobre si debemos tratar/o no el ultimo mensaje recibido
 168      por el ClientSocket recibido como parametro.
 169      \param clientSocket Socket cliente por el que hemos recibido el ultimo mensaje.
 170      \return Advice::Process para indicar que debemos procesar el mensaje
 171      , Advice::None para indicar que todavia no tiene datos para tomar una
 172      decision clara o Advice::Discard para indicar que debemos descartar.
 173   */
 174   Advice::_v getAdvice(const ClientSocket* clientSocket) throw() {
 175     return (clientSocket == NULL) ? Advice::Process : getAdvice(*clientSocket);
 176   }
 177
 178   /**
 179    * Devuelve información sobre las estadísticas de carga en las que se basó para hacer
 180    * los cálculos.
 181    *
 182    * \warning Sólo debería invocarse a este método después de invocar a #getAdvice.
 183    * \warning Este método no es MT-estricto, por lo que en un entorno MT los valores
 184    * sobre los que se calculó el último #getAdvice y los datos obtenidos con este método
 185    * pueden haber variado ligeramente.
 186    *
 187    * \see #getLoad
 188    * \see #getLevel
 189    */
 190   Workload getAccumulatedWorkload() const throw();
 191
 192   /**
 193    * Devuelve información sobre las estadísticas de carga del socket recibido como parámetro.
 194    *
 195    *
 196    * \param clientSocket Socket del que se quiere obtener el estado actual de ocupación.
 197    *
 198    * \warning Sólo debería invocarse a este método después de invocar a #getAdvice.
 199    * \warning Este método no es MT-estricto, por lo que en un entorno MT los valores
 200    * sobre los que se calculó el último #getAdvice y los datos obtenidos con este método
 201    * pueden haber variado ligeramente.
 202    *
 203    * \see #getLoad
 204    * \see #getLevel
 205    */
 206   Workload getCurrentWorkload(const ClientSocket& clientSocket) const throw();
 207
 208   /**
 209      Devuelve un documento XML con la informacion relevante sobre esta clase.
 210      \return un documento XML con la informacion relevante sobre esta clase.
 211   */
 212   xml::Node* asXML(xml::Node* parent) const throw();
 213
 214   /**
 215    * Extrae la carga media que soporta es proceso. La carga es un número en [0, 100].
 216    * \return la carga media que soporta es proceso. La carga es un número en [0, 100].
 217    */
 218   static int getLoad(const Workload& workload) throw() { return workload.second; }
 219
 220   /**
 221    * Extrae el nivel de carga en la que está el proceso. El nivel de carga es un número en [0, 4].
 222    * \return el nivel de carga en la que está el proceso. El nivel de carga es un número en [0, 4].
 223    */
 224   static int getLevel(const Workload& workload) throw() { return workload.first; }
 225
 226 private:
 227   static const Millisecond DelayTrace;
 228   static const int UnusedPendingBytes = -1;
 229   static const Millisecond HeartBeat;
 230
 231   //----------------------------------------------------------------------------------
 232   // a_limit: % de ocupacion maxima al que vamos a limitar los canales.
 233   // a_discardLevel: Valor maximo del % de cada nivel de descarte. 0 = 25, 1 = 50,
 234   // 2 = 85, 3 = 99
 235   //----------------------------------------------------------------------------------
 236   int a_limit;
 237   int a_discardLevel [MaxLevel];
 238   int a_percentage [MaxLevel];
 239   NRMutex a_mutex;
 240   Mode::_v a_mode;
 241   Mode::_v a_effectiveMode;
 242   Average <unsigned int> a_avgWorkload;
 243   unsigned int a_messageCounter;
 244   unsigned int a_discardedCounter;
 245   mutable Millisecond a_timeTrace;
 246   int a_maxPendingBytes;
 247   int a_incomingSocketCounter;
 248   Millisecond a_tickTime;
 249
 250   void incrementIncomingSocket() throw(RuntimeException);
 251   void decrementIncomingSocket() throw(RuntimeException);
 252
 253   CongestionController();
 254   CongestionController(const CongestionController&);
 255
 256   int calculeWorkload(const ClientSocket&) const throw();
 257
 258   friend class Singleton <CongestionController>;
 259
 260   friend class ClientSocket;
 261   // getUpdatePeriod
 262
 263   friend class handler::LocalConnection;
 264   // incrementIncomingSocket, decrementIncomingSocket
 265 };
 266
 267 }
 268 }
 269
 270 #endif
 271