src/sensor/gnunet-service-sensor-reporting.c

   1 /*
   2      This file is part of GNUnet.
   3      (C)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file sensor/gnunet-service-sensor-reporting.c
  23  * @brief sensor service reporting functionality
  24  * @author Omar Tarabai
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "sensor.h"
  29 #include "gnunet_peerstore_service.h"
  30 #include "gnunet_cadet_service.h"
  31 #include "gnunet_applications.h"
  32
  33 #define LOG(kind,...) GNUNET_log_from (kind, "sensor-reporting",__VA_ARGS__)
  34
  35 /**
  36  * Context of reporting operations
  37  */
  38 struct ReportingContext
  39 {
  40
  41   /**
  42    * DLL
  43    */
  44   struct ReportingContext *prev;
  45
  46   /**
  47    * DLL
  48    */
  49   struct ReportingContext *next;
  50
  51   /**
  52    * Sensor information
  53    */
  54   struct SensorInfo *sensor;
  55
  56   /**
  57    * Collection point reporting task
  58    * (OR GNUNET_SCHEDULER_NO_TASK)
  59    */
  60   GNUNET_SCHEDULER_TaskIdentifier cp_task;
  61
  62   /**
  63    * Watcher of sensor values
  64    */
  65   struct GNUNET_PEERSTORE_WatchContext *wc;
  66
  67   /**
  68    * Last value read from sensor
  69    */
  70   void *last_value;
  71
  72   /**
  73    * Size of @last_value
  74    */
  75   size_t last_value_size;
  76
  77   /**
  78    * Timestamp of last value reading
  79    */
  80   uint64_t timestamp;
  81
  82 };
  83
  84 /**
  85  * Context of a created CADET channel
  86  */
  87 struct CadetChannelContext
  88 {
  89
  90   /**
  91    * DLL
  92    */
  93   struct CadetChannelContext *prev;
  94
  95   /**
  96    * DLL
  97    */
  98   struct CadetChannelContext *next;
  99
 100   /**
 101    * Peer Id of
 102    */
 103   struct GNUNET_PeerIdentity pid;
 104
 105   /**
 106    * CADET channel handle
 107    */
 108   struct GNUNET_CADET_Channel *c;
 109
 110   /**
 111    * Are we sending data on this channel?
 112    * #GNUNET_YES / #GNUNET_NO
 113    */
 114   int sending;
 115
 116   /**
 117    * Pointer to a pending message to be sent over the channel
 118    */
 119   void *pending_msg;
 120
 121   /**
 122    * Size of @pending_msg
 123    */
 124   size_t pending_msg_size;
 125
 126   /**
 127    * Handle to CADET tranmission request in case we are sending
 128    * (sending == GNUNET_YES)
 129    */
 130   struct GNUNET_CADET_TransmitHandle *th;
 131
 132 };
 133
 134 /**
 135  * Our configuration.
 136  */
 137 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 138
 139 /**
 140  * Handle to peerstore service
 141  */
 142 static struct GNUNET_PEERSTORE_Handle *peerstore;
 143
 144 /**
 145  * My peer id
 146  */
 147 static struct GNUNET_PeerIdentity mypeerid;
 148
 149 /**
 150  * Handle to CADET service
 151  */
 152 static struct GNUNET_CADET_Handle *cadet;
 153
 154 /**
 155  * Head of DLL of all reporting contexts
 156  */
 157 struct ReportingContext *rc_head;
 158
 159 /**
 160  * Tail of DLL of all reporting contexts
 161  */
 162 struct ReportingContext *rc_tail;
 163
 164 /**
 165  * Head of DLL of all cadet channels
 166  */
 167 struct CadetChannelContext *cc_head;
 168
 169 /**
 170  * Tail of DLL of all cadet channels
 171  */
 172 struct CadetChannelContext *cc_tail;
 173
 174
 175 /**
 176  * Destroy a reporting context structure
 177  */
 178 static void
 179 destroy_reporting_context (struct ReportingContext *rc)
 180 {
 181   if (NULL != rc->wc)
 182   {
 183     GNUNET_PEERSTORE_watch_cancel (rc->wc);
 184     rc->wc = NULL;
 185   }
 186   if (GNUNET_SCHEDULER_NO_TASK != rc->cp_task)
 187   {
 188     GNUNET_SCHEDULER_cancel(rc->cp_task);
 189     rc->cp_task = GNUNET_SCHEDULER_NO_TASK;
 190   }
 191   if (NULL != rc->last_value)
 192   {
 193     GNUNET_free (rc->last_value);
 194     rc->last_value_size = 0;
 195   }
 196   GNUNET_free(rc);
 197 }
 198
 199 /**
 200  * Destroy a CADET channel context struct
 201  */
 202 static void
 203 destroy_cadet_channel_context (struct CadetChannelContext *cc)
 204 {
 205   if (NULL != cc->th)
 206   {
 207     GNUNET_CADET_notify_transmit_ready_cancel (cc->th);
 208     cc->th = NULL;
 209   }
 210   if (NULL != cc->pending_msg)
 211   {
 212     GNUNET_free (cc->pending_msg);
 213     cc->pending_msg = NULL;
 214   }
 215   if (NULL != cc->c)
 216   {
 217     GNUNET_CADET_channel_destroy (cc->c);
 218     cc->c = NULL;
 219   }
 220   GNUNET_free (cc);
 221 }
 222
 223 /**
 224  * Stop sensor reporting module
 225  */
 226 void SENSOR_reporting_stop ()
 227 {
 228   struct ReportingContext *rc;
 229   struct CadetChannelContext *cc;
 230
 231   LOG (GNUNET_ERROR_TYPE_DEBUG, "Stopping sensor reporting module.\n");
 232   while (NULL != cc_head)
 233   {
 234     cc = cc_head;
 235     GNUNET_CONTAINER_DLL_remove (cc_head, cc_tail, cc);
 236     destroy_cadet_channel_context (cc);
 237   }
 238   while (NULL != rc_head)
 239   {
 240     rc = rc_head;
 241     GNUNET_CONTAINER_DLL_remove (rc_head, rc_tail, rc);
 242     destroy_reporting_context (rc);
 243   }
 244   if (NULL != peerstore)
 245   {
 246     GNUNET_PEERSTORE_disconnect (peerstore);
 247     peerstore = NULL;
 248   }
 249   if (NULL != cadet)
 250   {
 251     GNUNET_CADET_disconnect (cadet);
 252     cadet = NULL;
 253   }
 254 }
 255
 256 /**
 257  * Returns CADET channel established to given peer
 258  * or creates a new one
 259  *
 260  * @param pid Peer Identity
 261  * @return Context of established cadet channel
 262  */
 263 static struct CadetChannelContext *
 264 get_cadet_channel (struct GNUNET_PeerIdentity pid)
 265 {
 266   struct CadetChannelContext *cc;
 267
 268   cc = cc_head;
 269   while (NULL != cc)
 270   {
 271     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&pid, &cc->pid))
 272       return cc;
 273     cc = cc->next;
 274   }
 275   cc = GNUNET_new (struct CadetChannelContext);
 276   cc->c = GNUNET_CADET_channel_create(cadet,
 277       cc,
 278       &pid,
 279       GNUNET_APPLICATION_TYPE_SENSORDASHBOARD,
 280       GNUNET_CADET_OPTION_DEFAULT);
 281   cc->pid = pid;
 282   cc->sending = GNUNET_NO;
 283   GNUNET_CONTAINER_DLL_insert (cc_head, cc_tail, cc);
 284   return cc;
 285 }
 286
 287 /**
 288  * Construct a reading message ready to be sent over CADET channel
 289  *
 290  * @param rc reporting context to read data from
 291  * @param msg used to return the created message structure
 292  * @return size of created message
 293  */
 294 static size_t
 295 construct_reading_message (struct ReportingContext *rc,
 296     struct GNUNET_SENSOR_Reading **msg)
 297 {
 298   struct GNUNET_SENSOR_Reading *ret;
 299   size_t sensorname_size;
 300   size_t total_size;
 301   void *dummy;
 302
 303   sensorname_size = strlen (rc->sensor->name) + 1;
 304   total_size = sizeof(struct GNUNET_SENSOR_Reading) +
 305       sensorname_size +
 306       rc->last_value_size;
 307   ret = GNUNET_malloc (total_size);
 308   ret->sensorname_size = sensorname_size;
 309   ret->sensorversion_major = rc->sensor->version_major;
 310   ret->sensorversion_minor = rc->sensor->version_minor;
 311   ret->timestamp = rc->timestamp;
 312   ret->value_size = rc->last_value_size;
 313   dummy = &ret[1];
 314   memcpy (dummy, rc->sensor->name, sensorname_size);
 315   dummy += sensorname_size;
 316   memcpy (dummy, rc->last_value, rc->last_value_size);
 317   *msg = ret;
 318   return total_size;
 319 }
 320
 321 /**
 322  * Function called to notify a client about the connection begin ready
 323  * to queue more data.  @a buf will be NULL and @a size zero if the
 324  * connection was closed for writing in the meantime.
 325  *
 326  * @param cls closure
 327  * @param size number of bytes available in @a buf
 328  * @param buf where the callee should write the message
 329  * @return number of bytes written to @a buf
 330  */
 331 static size_t
 332 do_report_collection_point (void *cls, size_t size, void *buf)
 333 {
 334   struct CadetChannelContext *cc = cls;
 335   size_t written = 0;
 336
 337   LOG (GNUNET_ERROR_TYPE_DEBUG, "Copying to CADET transmit buffer.\n");
 338   cc->sending = GNUNET_NO;
 339   if (NULL == buf || size != cc->pending_msg_size)
 340   {
 341     LOG (GNUNET_ERROR_TYPE_WARNING,
 342         "CADET failed to transmit message to collection point, discarding.");
 343   }
 344   else
 345   {
 346     memcpy (buf, cc->pending_msg, cc->pending_msg_size);
 347     written = cc->pending_msg_size;
 348   }
 349   GNUNET_free (cc->pending_msg);
 350   cc->pending_msg_size = 0;
 351   return written;
 352 }
 353
 354 /**
 355  * Task scheduled to send values to collection point
 356  *
 357  * @param cls closure, a 'struct CollectionReportingContext *'
 358  * @param tc unused
 359  */
 360 static void report_collection_point
 361 (void *cls, const struct GNUNET_SCHEDULER_TaskContext* tc)
 362 {
 363   struct ReportingContext *rc = cls;
 364   struct SensorInfo *sensor = rc->sensor;
 365   struct CadetChannelContext *cc;
 366   struct GNUNET_SENSOR_Reading *msg;
 367   size_t msg_size;
 368
 369   rc->cp_task = GNUNET_SCHEDULER_NO_TASK;
 370   if (0 == rc->last_value_size) /* Did not receive a sensor value yet */
 371   {
 372     LOG (GNUNET_ERROR_TYPE_WARNING, "Did not receive a value from `%s' "
 373         "to report yet.\n", rc->sensor->name);
 374     rc->cp_task = GNUNET_SCHEDULER_add_delayed (sensor->collection_interval,
 375             &report_collection_point, rc);
 376     return;
 377   }
 378   LOG (GNUNET_ERROR_TYPE_DEBUG, "Now trying to report last seen value of `%s' "
 379       "to collection point.\n", rc->sensor->name);
 380   GNUNET_assert (NULL != sensor->collection_point);
 381   cc = get_cadet_channel (*sensor->collection_point);
 382   if (GNUNET_YES == cc->sending)
 383   {
 384     LOG (GNUNET_ERROR_TYPE_DEBUG,
 385         "Cadet channel to collection point busy, "
 386         "trying again for sensor `%s' on next interval.\n", rc->sensor->name);
 387     rc->cp_task = GNUNET_SCHEDULER_add_delayed (sensor->collection_interval,
 388             &report_collection_point, rc);
 389     return;
 390   }
 391   msg_size = construct_reading_message (rc, &msg);
 392   cc->sending = GNUNET_YES;
 393   cc->pending_msg = msg;
 394   cc->pending_msg_size = msg_size;
 395   cc->th = GNUNET_CADET_notify_transmit_ready (cc->c,
 396       GNUNET_YES,
 397       sensor->collection_interval,
 398       msg_size,
 399       &do_report_collection_point,
 400       cc);
 401   rc->cp_task = GNUNET_SCHEDULER_add_delayed (sensor->collection_interval,
 402       &report_collection_point, rc);
 403 }
 404
 405 /*
 406  * Sensor value watch callback
 407  */
 408 static int
 409 sensor_watch_cb (void *cls,
 410     struct GNUNET_PEERSTORE_Record *record,
 411     char *emsg)
 412 {
 413   struct ReportingContext *rc = cls;
 414
 415   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a sensor `%s' watch value, "
 416       "updating notification last_value.\n", rc->sensor->name);
 417   if (NULL != emsg)
 418     return GNUNET_YES;
 419   if (NULL != rc->last_value)
 420   {
 421     GNUNET_free (rc->last_value);
 422     rc->last_value_size = 0;
 423   }
 424   rc->last_value = GNUNET_malloc(record->value_size);
 425   memcpy (rc->last_value, record->value, record->value_size);
 426   rc->last_value_size = record->value_size;
 427   rc->timestamp = GNUNET_TIME_absolute_get().abs_value_us;
 428   return GNUNET_YES;
 429 }
 430
 431 /**
 432  * Iterator for defined sensors
 433  * Watches sensors for readings to report
 434  *
 435  * @param cls unused
 436  * @param key unused
 437  * @param value a 'struct SensorInfo *' with sensor information
 438  * @return #GNUNET_YES to continue iterations
 439  */
 440 static int
 441 init_sensor_reporting (void *cls,
 442     const struct GNUNET_HashCode *key,
 443     void *value)
 444 {
 445   struct SensorInfo *sensor = value;
 446   struct ReportingContext *rc;
 447
 448   if (NULL == sensor->collection_point &&
 449       GNUNET_NO == sensor->p2p_report)
 450     return GNUNET_YES;
 451   rc = GNUNET_new (struct ReportingContext);
 452   rc->sensor = sensor;
 453   rc->last_value = NULL;
 454   rc->last_value_size = 0;
 455   rc->wc = GNUNET_PEERSTORE_watch(peerstore,
 456       "sensor",
 457       &mypeerid,
 458       sensor->name,
 459       &sensor_watch_cb,
 460       rc);
 461   if (NULL != sensor->collection_point)
 462   {
 463     LOG (GNUNET_ERROR_TYPE_INFO,
 464         "Will start reporting sensor `%s' values to "
 465         "collection point `%s' every %s.\n",
 466         sensor->name, GNUNET_i2s_full(sensor->collection_point),
 467         GNUNET_STRINGS_relative_time_to_string(sensor->collection_interval,
 468             GNUNET_YES));
 469     rc->cp_task =
 470         GNUNET_SCHEDULER_add_delayed (sensor->collection_interval,
 471             &report_collection_point,
 472             rc);
 473   }
 474   if (GNUNET_YES == sensor->p2p_report)
 475   {
 476     LOG (GNUNET_ERROR_TYPE_INFO,
 477         "Will start reporting sensor `%s' values to p2p network every %s.\n",
 478         sensor->name,
 479         GNUNET_STRINGS_relative_time_to_string(sensor->p2p_interval,
 480             GNUNET_YES));
 481   }
 482   GNUNET_CONTAINER_DLL_insert (rc_head, rc_tail, rc);
 483   return GNUNET_YES;
 484 }
 485
 486 /**
 487  * Function called whenever a channel is destroyed.  Should clean up
 488  * any associated state.
 489  *
 490  * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
 491  *
 492  * @param cls closure (set from #GNUNET_CADET_connect)
 493  * @param channel connection to the other end (henceforth invalid)
 494  * @param channel_ctx place where local state associated
 495  *                   with the channel is stored
 496  */
 497 static void cadet_channel_destroyed (void *cls,
 498     const struct GNUNET_CADET_Channel *channel,
 499     void *channel_ctx)
 500 {
 501   struct CadetChannelContext *cc = channel_ctx;
 502
 503   LOG (GNUNET_ERROR_TYPE_DEBUG,
 504       "Received a `channel destroyed' notification from CADET, "
 505       "cleaning up.\n");
 506   GNUNET_CONTAINER_DLL_remove (cc_head, cc_tail, cc);
 507   cc->c = NULL;
 508   destroy_cadet_channel_context (cc);
 509 }
 510
 511 /**
 512  * Start the sensor reporting module
 513  *
 514  * @param c our service configuration
 515  * @param sensors multihashmap of loaded sensors
 516  * @return #GNUNET_OK if started successfully, #GNUNET_SYSERR otherwise
 517  */
 518 int
 519 SENSOR_reporting_start (const struct GNUNET_CONFIGURATION_Handle *c,
 520     struct GNUNET_CONTAINER_MultiHashMap *sensors)
 521 {
 522   static struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
 523       {NULL, 0, 0}
 524   };
 525
 526   LOG (GNUNET_ERROR_TYPE_DEBUG, "Starting sensor reporting module.\n");
 527   GNUNET_assert(NULL != sensors);
 528   cfg = c;
 529   peerstore = GNUNET_PEERSTORE_connect(cfg);
 530   if (NULL == peerstore)
 531   {
 532     LOG (GNUNET_ERROR_TYPE_ERROR,
 533         _("Failed to connect to peerstore service.\n"));
 534     SENSOR_reporting_stop ();
 535     return GNUNET_SYSERR;
 536   }
 537   cadet = GNUNET_CADET_connect(cfg,
 538       NULL,
 539       NULL,
 540       &cadet_channel_destroyed,
 541       cadet_handlers,
 542       NULL);
 543   if (NULL == cadet)
 544   {
 545     LOG (GNUNET_ERROR_TYPE_ERROR,
 546         _("Failed to connect to CADET service.\n"));
 547     SENSOR_reporting_stop ();
 548     return GNUNET_SYSERR;
 549   }
 550   GNUNET_CRYPTO_get_peer_identity(cfg, &mypeerid);
 551   GNUNET_CONTAINER_multihashmap_iterate(sensors, &init_sensor_reporting, NULL);
 552
 553   return GNUNET_OK;
 554 }
 555
 556 /* end of gnunet-service-sensor-reporting.c */