src/transport/transport_api_monitor_plugins.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2014, 2016 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 /**
  20  * @file transport/transport_api_monitor_plugins.c
  21  * @brief montoring api for transport plugin session status
  22  * @author Christian Grothoff
  23  */
  24 #include "platform.h"
  25 #include "gnunet_util_lib.h"
  26 #include "gnunet_arm_service.h"
  27 #include "gnunet_hello_lib.h"
  28 #include "gnunet_protocols.h"
  29 #include "gnunet_transport_service.h"
  30 #include "transport.h"
  31
  32
  33 /**
  34  * Handle for a plugin session state monitor.
  35  */
  36 struct GNUNET_TRANSPORT_PluginMonitor
  37 {
  38
  39   /**
  40    * Connection to the service.
  41    */
  42   struct GNUNET_MQ_Handle *mq;
  43
  44   /**
  45    * Our configuration.
  46    */
  47   const struct GNUNET_CONFIGURATION_Handle *cfg;
  48
  49   /**
  50    * Callback to call.
  51    */
  52   GNUNET_TRANSPORT_SessionMonitorCallback cb;
  53
  54   /**
  55    * Closure for @e cb
  56    */
  57   void *cb_cls;
  58
  59   /**
  60    * Map of session_ids (reduced to 32-bits) to
  61    * `struct GNUNET_TRANSPORT_PluginSession` objects.
  62    */
  63   struct GNUNET_CONTAINER_MultiHashMap32 *sessions;
  64
  65   /**
  66    * Backoff for reconnect.
  67    */
  68   struct GNUNET_TIME_Relative backoff;
  69
  70   /**
  71    * Task ID for reconnect.
  72    */
  73   struct GNUNET_SCHEDULER_Task *reconnect_task;
  74
  75 };
  76
  77
  78 /**
  79  * Abstract representation of a plugin's session.
  80  * Corresponds to the `struct GNUNET_ATS_Session` within the TRANSPORT service.
  81  */
  82 struct GNUNET_TRANSPORT_PluginSession
  83 {
  84   /**
  85    * Unique session identifier.
  86    */
  87   uint64_t session_id;
  88
  89   /**
  90    * Location for the client to store "data".
  91    */
  92   void *client_ctx;
  93 };
  94
  95
  96
  97 /**
  98  * Task run to re-establish the connection.
  99  *
 100  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
 101  */
 102 static void
 103 do_plugin_connect (void *cls);
 104
 105
 106 /**
 107  * Free the session entry and notify the callback about its demise.
 108  *
 109  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor`
 110  * @param key key of the session in the map
 111  * @param value the session to free
 112  * @return #GNUNET_OK (continue to iterate)
 113  */
 114 static int
 115 free_entry (void *cls,
 116             uint32_t key,
 117             void *value)
 118 {
 119   struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
 120   struct GNUNET_TRANSPORT_PluginSession *ps = value;
 121
 122   pm->cb (pm->cb_cls,
 123           ps,
 124           &ps->client_ctx,
 125           NULL);
 126   GNUNET_break (GNUNET_YES ==
 127                 GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
 128                                                         key,
 129                                                         ps));
 130   GNUNET_break (NULL == ps->client_ctx);
 131   GNUNET_free (ps);
 132   return GNUNET_OK;
 133 }
 134
 135
 136 /**
 137  * Cut the existing connection and reconnect.
 138  *
 139  * @param pm our context
 140  */
 141 static void
 142 reconnect_plugin_ctx (struct GNUNET_TRANSPORT_PluginMonitor *pm)
 143 {
 144   GNUNET_MQ_destroy (pm->mq);
 145   pm->mq = NULL;
 146   GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
 147                                            &free_entry,
 148                                            pm);
 149   pm->backoff = GNUNET_TIME_STD_BACKOFF (pm->backoff);
 150   pm->reconnect_task = GNUNET_SCHEDULER_add_delayed (pm->backoff,
 151                                                      &do_plugin_connect,
 152                                                      pm);
 153 }
 154
 155
 156 /**
 157  * Convert 64-bit session ID to 32-bit index for hash map.
 158  *
 159  * @param id 64-bit session ID
 160  * @return 32-bit hash map index
 161  */
 162 static uint32_t
 163 wrap_id (uint64_t id)
 164 {
 165   return ((uint32_t) id) ^ ((uint32_t) (id >> 32));
 166 }
 167
 168
 169 /**
 170  * Context for #locate_by_id().
 171  */
 172 struct SearchContext
 173 {
 174
 175   /**
 176    * Result.
 177    */
 178   struct GNUNET_TRANSPORT_PluginSession *ps;
 179
 180   /**
 181    * ID to locate.
 182    */
 183   uint64_t session_id;
 184
 185 };
 186
 187
 188 /**
 189  * Locate a session entry.
 190  *
 191  * @param cls our `struct SearchContext`
 192  * @param key key of the session in the map
 193  * @param value a session
 194  * @return #GNUNET_OK (continue to iterate), or #GNUNET_SYSERR (match found)
 195  */
 196 static int
 197 locate_by_id (void *cls,
 198               uint32_t key,
 199               void *value)
 200 {
 201   struct SearchContext *sc = cls;
 202   struct GNUNET_TRANSPORT_PluginSession *ps = value;
 203
 204   if (sc->session_id == ps->session_id)
 205   {
 206     sc->ps = ps;
 207     return GNUNET_SYSERR;
 208   }
 209   return GNUNET_OK;
 210 }
 211
 212
 213 /**
 214  * Function called with responses from the service.
 215  *
 216  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
 217  * @paramm tpmm message with event data
 218  * @return #GNUNET_Ok if message is well-formed
 219  */
 220 static int
 221 check_event (void *cls,
 222              const struct TransportPluginMonitorMessage *tpmm)
 223 {
 224   const char *pname;
 225   size_t pname_len;
 226   size_t paddr_len;
 227
 228   pname = (const char *) &tpmm[1];
 229   pname_len = ntohs (tpmm->plugin_name_len);
 230   paddr_len = ntohs (tpmm->plugin_address_len);
 231   if ( (pname_len +
 232         paddr_len +
 233         sizeof (struct TransportPluginMonitorMessage) != ntohs (tpmm->header.size)) ||
 234        ( (0 != pname_len) &&
 235          ('\0' != pname[pname_len - 1]) ) )
 236   {
 237     GNUNET_break (0);
 238     return GNUNET_SYSERR;
 239   }
 240   return GNUNET_OK;
 241 }
 242
 243
 244 /**
 245  * Function called with responses from the service.
 246  *
 247  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
 248  * @paramm tpmm message with event data
 249  */
 250 static void
 251 handle_event (void *cls,
 252               const struct TransportPluginMonitorMessage *tpmm)
 253 {
 254   struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
 255   struct GNUNET_TRANSPORT_PluginSession *ps;
 256   const char *pname;
 257   const void *paddr;
 258   enum GNUNET_TRANSPORT_SessionState ss;
 259   size_t pname_len;
 260   size_t paddr_len;
 261   struct GNUNET_TRANSPORT_SessionInfo info;
 262   struct GNUNET_HELLO_Address addr;
 263   struct SearchContext rv;
 264
 265   pname = (const char *) &tpmm[1];
 266   pname_len = ntohs (tpmm->plugin_name_len);
 267   paddr_len = ntohs (tpmm->plugin_address_len);
 268   paddr = &pname[pname_len];
 269   ps = NULL;
 270   ss = (enum GNUNET_TRANSPORT_SessionState) ntohs (tpmm->session_state);
 271   if (GNUNET_TRANSPORT_SS_INIT == ss)
 272   {
 273     ps = GNUNET_new (struct GNUNET_TRANSPORT_PluginSession);
 274     ps->session_id = tpmm->session_id;
 275     (void) GNUNET_CONTAINER_multihashmap32_put (pm->sessions,
 276                                                 wrap_id (tpmm->session_id),
 277                                                 ps,
 278                                                 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 279
 280   }
 281   else
 282   {
 283     rv.session_id = tpmm->session_id;
 284     rv.ps = NULL;
 285     (void) GNUNET_CONTAINER_multihashmap32_get_multiple (pm->sessions,
 286                                                          wrap_id (tpmm->session_id),
 287                                                          &locate_by_id,
 288                                                          &rv);
 289     ps = rv.ps;
 290     if (NULL == ps)
 291     {
 292       GNUNET_break (0);
 293       reconnect_plugin_ctx (pm);
 294       return;
 295     }
 296   }
 297   info.state = ss;
 298   info.is_inbound = (int16_t) ntohs (tpmm->is_inbound);
 299   info.num_msg_pending = ntohl (tpmm->msgs_pending);
 300   info.num_bytes_pending = ntohl (tpmm->bytes_pending);
 301   info.receive_delay = GNUNET_TIME_absolute_ntoh (tpmm->delay);
 302   info.session_timeout = GNUNET_TIME_absolute_ntoh (tpmm->timeout);
 303   info.address = &addr;
 304   addr.peer = tpmm->peer;
 305   addr.address = (0 == paddr_len) ? NULL : paddr;
 306   addr.address_length = paddr_len;
 307   addr.transport_name = (0 == pname_len) ? NULL : pname;
 308   addr.local_info = GNUNET_HELLO_ADDRESS_INFO_NONE;
 309   pm->cb (pm->cb_cls,
 310           ps,
 311           &ps->client_ctx,
 312           &info);
 313
 314   if (GNUNET_TRANSPORT_SS_DONE == ss)
 315   {
 316     GNUNET_break (NULL == ps->client_ctx);
 317     GNUNET_assert (GNUNET_YES ==
 318                    GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
 319                                                            wrap_id (tpmm->session_id),
 320                                                            ps));
 321     GNUNET_free (ps);
 322   }
 323 }
 324
 325
 326 /**
 327  * Function called with sync responses from the service.
 328  *
 329  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
 330  * @param msg message from the service
 331  */
 332 static void
 333 handle_sync (void *cls,
 334              const struct GNUNET_MessageHeader *msg)
 335 {
 336   struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
 337
 338   /* we are in sync, notify callback */
 339   pm->cb (pm->cb_cls,
 340           NULL,
 341           NULL,
 342           NULL);
 343 }
 344
 345
 346 /**
 347  * Generic error handler, called with the appropriate
 348  * error code and the same closure specified at the creation of
 349  * the message queue.
 350  * Not every message queue implementation supports an error handler.
 351  *
 352  * @param cls closure with the `struct GNUNET_NSE_Handle *`
 353  * @param error error code
 354  */
 355 static void
 356 mq_error_handler (void *cls,
 357                   enum GNUNET_MQ_Error error)
 358 {
 359   struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
 360
 361   reconnect_plugin_ctx (pm);
 362 }
 363
 364
 365 /**
 366  * Task run to re-establish the connection.
 367  *
 368  * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
 369  */
 370 static void
 371 do_plugin_connect (void *cls)
 372 {
 373   struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
 374   struct GNUNET_MQ_MessageHandler handlers[] = {
 375     GNUNET_MQ_hd_var_size (event,
 376                            GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_EVENT,
 377                            struct TransportPluginMonitorMessage,
 378                            pm),
 379     GNUNET_MQ_hd_fixed_size (sync,
 380                              GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_SYNC,
 381                              struct GNUNET_MessageHeader,
 382                              pm),
 383     GNUNET_MQ_handler_end ()
 384   };
 385   struct GNUNET_MessageHeader *msg;
 386   struct GNUNET_MQ_Envelope *env;
 387
 388   pm->reconnect_task = NULL;
 389   pm->mq = GNUNET_CLIENT_connect (pm->cfg,
 390                                   "transport",
 391                                   handlers,
 392                                   &mq_error_handler,
 393                                   pm);
 394   if (NULL == pm->mq)
 395     return;
 396   env = GNUNET_MQ_msg (msg,
 397                        GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_START);
 398   GNUNET_MQ_send (pm->mq,
 399                   env);
 400 }
 401
 402
 403 /**
 404  * Install a plugin session state monitor callback.  The callback
 405  * will be notified whenever the session changes.
 406  *
 407  * @param cfg configuration to use
 408  * @param cb callback to invoke on events
 409  * @param cb_cls closure for @a cb
 410  * @return NULL on error, otherwise handle for cancellation
 411  */
 412 struct GNUNET_TRANSPORT_PluginMonitor *
 413 GNUNET_TRANSPORT_monitor_plugins (const struct GNUNET_CONFIGURATION_Handle *cfg,
 414                                   GNUNET_TRANSPORT_SessionMonitorCallback cb,
 415                                   void *cb_cls)
 416 {
 417   struct GNUNET_TRANSPORT_PluginMonitor *pm;
 418
 419   pm = GNUNET_new (struct GNUNET_TRANSPORT_PluginMonitor);
 420   pm->cb = cb;
 421   pm->cb_cls = cb_cls;
 422   pm->cfg = cfg;
 423   do_plugin_connect (pm);
 424   if (NULL == pm->mq)
 425   {
 426     GNUNET_free (pm);
 427     return NULL;
 428   }
 429   pm->sessions = GNUNET_CONTAINER_multihashmap32_create (128);
 430   return pm;
 431 }
 432
 433
 434 /**
 435  * Cancel monitoring the plugin session state.  The callback will
 436  * be called once for each session that is up with the information
 437  * #GNUNET_TRANSPORT_SS_FINI (even though the session may stay up;
 438  * this is just to enable client-side cleanup).
 439  *
 440  * @param pm handle of the request that is to be cancelled
 441  */
 442 void
 443 GNUNET_TRANSPORT_monitor_plugins_cancel (struct GNUNET_TRANSPORT_PluginMonitor *pm)
 444 {
 445   if (NULL != pm->mq)
 446   {
 447     GNUNET_MQ_destroy (pm->mq);
 448     pm->mq = NULL;
 449   }
 450   if (NULL != pm->reconnect_task)
 451   {
 452     GNUNET_SCHEDULER_cancel (pm->reconnect_task);
 453     pm->reconnect_task = NULL;
 454   }
 455   GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
 456                                            &free_entry,
 457                                            pm);
 458   GNUNET_CONTAINER_multihashmap32_destroy (pm->sessions);
 459   GNUNET_free (pm);
 460 }
 461
 462
 463 /* end of transport_api_monitor_plugins.c */