src/experimentation/gnunet-daemon-experimentation_scheduler.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009 Christian Grothoff (and other contributing authors)
   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 experimentation/gnunet-daemon-experimentation_scheduler.c
  23  * @brief experimentation daemon: execute experiments
  24  * @author Christian Grothoff
  25  * @author Matthias Wachs
  26  */
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_core_service.h"
  30 #include "gnunet_statistics_service.h"
  31 #include "gnunet-daemon-experimentation.h"
  32
  33 /**
  34  * An experiment is added during startup as not running NOT_RUNNING
  35  *
  36  * The scheduler then decides to schedule it and sends a request to the
  37  * remote peer, if core cannot send since it is busy we wait for some time
  38  * and change state to BUSY, if we can send we change to REQUESTED and wait
  39  * for remote peers ACK.
  40  *
  41  * When we receive an ACK we change to STARTED and when scheduler decides that
  42  * the experiment is finished we change to STOPPED.
  43  */
  44
  45 enum ExperimentState
  46 {
  47         /* Experiment is added and waiting to be executed */
  48         NOT_RUNNING,
  49         /* Cannot send request to remote peer, core is busy*/
  50         BUSY,
  51         /* We requested experiment and wait for remote peer to ACK */
  52         REQUESTED,
  53         /* Experiment is running */
  54         STARTED,
  55         /* Experiment is done */
  56         STOPPED
  57 };
  58
  59 struct ScheduledExperiment {
  60         struct ScheduledExperiment *next;
  61         struct ScheduledExperiment *prev;
  62
  63         struct Experiment *e;
  64         struct Node *n;
  65         int state;
  66         int outbound;
  67         GNUNET_SCHEDULER_TaskIdentifier task;
  68 };
  69
  70 struct ScheduledExperiment *waiting_in_head;
  71 struct ScheduledExperiment *waiting_in_tail;
  72
  73 struct ScheduledExperiment *running_in_head;
  74 struct ScheduledExperiment *running_in_tail;
  75
  76 struct ScheduledExperiment *waiting_out_head;
  77 struct ScheduledExperiment *waiting_out_tail;
  78
  79 struct ScheduledExperiment *running_out_head;
  80 struct ScheduledExperiment *running_out_tail;
  81
  82
  83 static unsigned int experiments_scheduled;
  84 static unsigned int experiments_outbound_running;
  85 static unsigned int experiments_inbound_running;
  86 static unsigned int experiments_requested;
  87
  88
  89 static struct ScheduledExperiment *
  90 find_experiment (struct ScheduledExperiment *head, struct ScheduledExperiment *tail,
  91                                                                  struct Node *n, struct Experiment *e, int outbound)
  92 {
  93         struct ScheduledExperiment *cur;
  94         for (cur = head; NULL != cur; cur = cur->next)
  95         {
  96                 if ((cur->n == n) && (cur->e == e) && (cur->outbound == outbound)) /* Node and experiment are equal */
  97                         break;
  98         }
  99         return cur;
 100 }
 101
 102 static void
 103 request_timeout (void *cls,const struct GNUNET_SCHEDULER_TaskContext* tc)
 104 {
 105         struct ScheduledExperiment *se = cls;
 106         se->task = GNUNET_SCHEDULER_NO_TASK;
 107
 108         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Peer `%s' did not respond to request for experiment `%s'\n"),
 109                         GNUNET_i2s (&se->n->id), se->e->name);
 110
 111         GNUNET_CONTAINER_DLL_remove (waiting_out_head, waiting_out_tail, se);
 112         GNUNET_free (se);
 113
 114         /* Remove experiment */
 115         GNUNET_assert (experiments_requested > 0);
 116         experiments_requested --;
 117         GNUNET_STATISTICS_set (GED_stats, "# experiments requested", experiments_requested, GNUNET_NO);
 118 }
 119
 120 static void run_experiment_inbound (void *cls,const struct GNUNET_SCHEDULER_TaskContext* tc)
 121 {
 122         struct ScheduledExperiment *se = cls;
 123         struct GNUNET_TIME_Relative start;
 124         struct GNUNET_TIME_Relative end;
 125
 126         se->task = GNUNET_SCHEDULER_NO_TASK;
 127
 128         switch (se->state) {
 129                 case NOT_RUNNING:
 130                         /* Send START_ACK message */
 131                         GED_nodes_send_start_ack (se->n, se->e);
 132                         se->state = REQUESTED;
 133                         /* Schedule to run */
 134                         start = GNUNET_TIME_absolute_get_remaining(se->e->start);
 135                         if (0 == start.rel_value_us)
 136                                         se->task = GNUNET_SCHEDULER_add_now (&run_experiment_inbound, se);
 137                         else
 138                                         se->task = GNUNET_SCHEDULER_add_delayed (start, &run_experiment_inbound, se);
 139                         break;
 140                 case REQUESTED:
 141                         experiments_inbound_running ++;
 142                         GNUNET_STATISTICS_set (GED_stats, "# experiments inbound running", experiments_inbound_running, GNUNET_NO);
 143                         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Starting inbound experiment `%s' with peer `%s'\n"),
 144                                         se->e->name, GNUNET_i2s (&se->n->id));
 145                         se->state = STARTED;
 146                         se->task = GNUNET_SCHEDULER_add_now (&run_experiment_inbound, se);
 147                         break;
 148                 case STARTED:
 149                         /* Experiment is running */
 150                         GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running %s experiment `%s' peer for `%s'\n",
 151                                         "inbound", GNUNET_i2s (&se->n->id), se->e->name);
 152
 153                         /* do work here */
 154
 155                         /* Reschedule */
 156                         end = GNUNET_TIME_absolute_get_remaining(GNUNET_TIME_absolute_add (se->e->stop, se->e->frequency));
 157                         if (0 == end.rel_value_us)
 158                         {
 159                                 se->state = STOPPED;
 160                                 return; /* End of experiment is reached */
 161                         }
 162                         /* Reschedule */
 163                         se->task = GNUNET_SCHEDULER_add_delayed (se->e->frequency, &run_experiment_inbound, se);
 164                         break;
 165                 case STOPPED:
 166                         /* Experiment expired */
 167                         break;
 168                 default:
 169                         break;
 170         }
 171
 172 }
 173
 174 static void run_experiment_outbound (void *cls,const struct GNUNET_SCHEDULER_TaskContext* tc)
 175 {
 176         struct ScheduledExperiment *se = cls;
 177         struct GNUNET_TIME_Relative end;
 178
 179         se->task = GNUNET_SCHEDULER_NO_TASK;
 180
 181         switch (se->state) {
 182                 case NOT_RUNNING:
 183                         /* Send START message */
 184                         GED_nodes_send_start (se->n, se->e);
 185                         se->state = REQUESTED;
 186                         se->task = GNUNET_SCHEDULER_add_delayed (EXP_RESPONSE_TIMEOUT, &request_timeout, se);
 187                         experiments_requested ++;
 188                         GNUNET_STATISTICS_set (GED_stats, "# experiments requested", experiments_requested, GNUNET_NO);
 189                         break;
 190                 case REQUESTED:
 191                         /* Expecting START_ACK */
 192                         GNUNET_break (0);
 193                         break;
 194                 case STARTED:
 195                         /* Experiment is running */
 196                         GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running %s experiment `%s' peer for `%s'\n",
 197                                         "outbound", GNUNET_i2s (&se->n->id), se->e->name);
 198
 199                         /* do work here */
 200
 201                         /* Reschedule */
 202                         end = GNUNET_TIME_absolute_get_remaining(GNUNET_TIME_absolute_add (se->e->stop, se->e->frequency));
 203                         if (0 == end.rel_value_us)
 204                         {
 205                                 se->state = STOPPED;
 206                                 return; /* End of experiment is reached */
 207                         }
 208                         /* Reschedule */
 209                 se->task = GNUNET_SCHEDULER_add_delayed (se->e->frequency, &run_experiment_outbound, se);
 210                         break;
 211                 case STOPPED:
 212                         /* Experiment expired */
 213                         break;
 214                 default:
 215                         break;
 216         }
 217 }
 218
 219
 220 /**
 221  * Handle a START message from a remote node
 222  *
 223  * @param n the node
 224  * @param e the experiment
 225  */
 226 void
 227 GED_scheduler_handle_start (struct Node *n, struct Experiment *e)
 228 {
 229   if ((NULL != find_experiment (waiting_in_head, waiting_in_tail, n, e, GNUNET_NO)) ||
 230       (NULL != find_experiment (running_in_head, running_in_tail, n, e, GNUNET_NO)))
 231   {
 232     GNUNET_break_op (0);
 233     return;
 234   }
 235
 236   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 237               "Received %s message from peer %s for experiment `%s'\n",
 238               "START", GNUNET_i2s (&n->id), e->name);
 239   GED_scheduler_add (n, e, GNUNET_NO);
 240 }
 241
 242
 243 /**
 244  * Handle a START_ACK message from a remote node
 245  *
 246  * @param n the node
 247  * @param e the experiment
 248  */
 249 void
 250 GED_scheduler_handle_start_ack (struct Node *n, struct Experiment *e)
 251 {
 252         struct ScheduledExperiment *se;
 253
 254         if (NULL == (se = find_experiment (waiting_out_head, waiting_out_tail, n, e, GNUNET_YES)))
 255         {
 256                 GNUNET_break (0);
 257                 return;
 258         }
 259
 260         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s message from peer %s for requested experiment `%s'\n",
 261                         "START_ACK", GNUNET_i2s (&n->id), e->name);
 262
 263         if (GNUNET_SCHEDULER_NO_TASK != se->task)
 264         {
 265                 GNUNET_SCHEDULER_cancel (se->task); /* *Canceling timeout task */
 266                 se->task = GNUNET_SCHEDULER_NO_TASK;
 267         }
 268
 269         /* Remove from waiting list, add to running list */
 270         GNUNET_CONTAINER_DLL_remove (waiting_out_head, waiting_out_tail, se);
 271         GNUNET_CONTAINER_DLL_insert (running_out_head, running_out_tail, se);
 272
 273         /* Change state and schedule to run */
 274         experiments_outbound_running ++;
 275         GNUNET_STATISTICS_set (GED_stats, "# experiments outbound running", experiments_outbound_running, GNUNET_NO);
 276         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Starting outbound experiment `%s' with peer `%s'\n"),
 277                         e->name, GNUNET_i2s (&n->id));
 278         se->state = STARTED;
 279         se->task = GNUNET_SCHEDULER_add_now (&run_experiment_outbound, se);
 280 }
 281
 282
 283 /**
 284  * Handle a STOP message from a remote node
 285  *
 286  * @param n the node
 287  * @param e the experiment
 288  */
 289 void
 290 GED_scheduler_handle_stop (struct Node *n, struct Experiment *e)
 291 {
 292         struct ScheduledExperiment *se;
 293
 294         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, _("Received %s message from peer %s for experiment `%s'\n"),
 295                         "STOP", GNUNET_i2s (&n->id), e->name);
 296
 297         if (NULL != (se = find_experiment (waiting_in_head, waiting_in_tail, n, e, GNUNET_NO)))
 298         {
 299                 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s message from peer %s for waiting experiment `%s'\n",
 300                                 "STOP", GNUNET_i2s (&n->id), e->name);
 301         }
 302
 303         if (NULL != (se = find_experiment (running_in_head, running_in_tail, n, e, GNUNET_NO)))
 304         {
 305                 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s message from peer %s for running experiment `%s'\n",
 306                                 "STOP", GNUNET_i2s (&n->id), e->name);
 307         }
 308
 309 }
 310
 311 /**
 312  * Add a new experiment for a node
 313  *
 314  * @param n the node
 315  * @param e the experiment
 316  * @param outbound are we initiator (GNUNET_YES) or client (GNUNET_NO)?
 317  */
 318 void
 319 GED_scheduler_add (struct Node *n, struct Experiment *e, int outbound)
 320 {
 321         struct ScheduledExperiment *se;
 322         struct GNUNET_TIME_Relative start;
 323         struct GNUNET_TIME_Relative end;
 324
 325         GNUNET_assert ((GNUNET_YES == outbound) || (GNUNET_NO == outbound));
 326
 327         start = GNUNET_TIME_absolute_get_remaining(e->start);
 328         end = GNUNET_TIME_absolute_get_remaining(e->stop);
 329         if (0 == end.rel_value_us)
 330                         return; /* End of experiment is reached */
 331
 332         /* Add additional checks here if required */
 333         se = GNUNET_malloc (sizeof (struct ScheduledExperiment));
 334         se->state = NOT_RUNNING;
 335         se->outbound = outbound;
 336         se->e = e;
 337         se->n = n;
 338
 339         if (GNUNET_YES == outbound)
 340         {
 341           if (0 == start.rel_value_us)
 342                                 se->task = GNUNET_SCHEDULER_add_now (&run_experiment_outbound, se);
 343                 else
 344                                 se->task = GNUNET_SCHEDULER_add_delayed (start, &run_experiment_outbound, se);
 345                 GNUNET_CONTAINER_DLL_insert (waiting_out_head, waiting_out_tail, se);
 346         }
 347         else
 348         {
 349                 if (0 == start.rel_value_us)
 350                                 se->task = GNUNET_SCHEDULER_add_now (&run_experiment_inbound, se);
 351                 else
 352                                 se->task = GNUNET_SCHEDULER_add_delayed (start, &run_experiment_inbound, se);
 353                 GNUNET_CONTAINER_DLL_insert (waiting_in_head, waiting_in_tail, se);
 354         }
 355
 356         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Added %s experiment `%s' for node to be scheduled\n",
 357                         (GNUNET_YES == outbound) ? "outbound" : "inbound", e->name, GNUNET_i2s(&se->n->id));
 358         experiments_scheduled ++;
 359         GNUNET_STATISTICS_set (GED_stats, "# experiments scheduled", experiments_scheduled, GNUNET_NO);
 360
 361 }
 362
 363 /**
 364  * Start the scheduler component
 365  */
 366 void
 367 GED_scheduler_start ()
 368 {
 369         experiments_requested = 0;
 370         experiments_scheduled = 0;
 371 }
 372
 373
 374 /**
 375  * Stop the scheduler component
 376  */
 377 void
 378 GED_scheduler_stop ()
 379 {
 380         struct ScheduledExperiment *cur;
 381         struct ScheduledExperiment *next;
 382
 383         next = waiting_in_head;
 384         while (NULL != (cur = next))
 385         {
 386                         next = cur->next;
 387                         GNUNET_CONTAINER_DLL_remove (waiting_in_head, waiting_in_tail, cur);
 388                         if (GNUNET_SCHEDULER_NO_TASK != cur->task)
 389                         {
 390                                         GNUNET_SCHEDULER_cancel (cur->task);
 391                                         cur->task = GNUNET_SCHEDULER_NO_TASK;
 392                         }
 393                         GNUNET_free (cur);
 394                         GNUNET_assert (experiments_scheduled > 0);
 395                         experiments_scheduled --;
 396                         GNUNET_STATISTICS_set (GED_stats, "# experiments scheduled", experiments_scheduled, GNUNET_NO);
 397         }
 398
 399         next = running_in_head;
 400         while (NULL != (cur = next))
 401         {
 402                         next = cur->next;
 403                         GNUNET_CONTAINER_DLL_remove (running_in_head, running_in_tail, cur);
 404                         if (GNUNET_SCHEDULER_NO_TASK != cur->task)
 405                         {
 406                                         GNUNET_SCHEDULER_cancel (cur->task);
 407                                         cur->task = GNUNET_SCHEDULER_NO_TASK;
 408                         }
 409                         GNUNET_free (cur);
 410                         GNUNET_assert (experiments_outbound_running > 0);
 411                         experiments_inbound_running --;
 412                         GNUNET_STATISTICS_set (GED_stats, "# experiments inbound running", experiments_inbound_running, GNUNET_NO);
 413         }
 414
 415         next = waiting_out_head;
 416         while (NULL != (cur = next))
 417         {
 418                         next = cur->next;
 419                         GNUNET_CONTAINER_DLL_remove (waiting_out_head, waiting_out_tail, cur);
 420                         if (GNUNET_SCHEDULER_NO_TASK != cur->task)
 421                         {
 422                                         GNUNET_SCHEDULER_cancel (cur->task);
 423                                         cur->task = GNUNET_SCHEDULER_NO_TASK;
 424                         }
 425                         GNUNET_free (cur);
 426                         GNUNET_assert (experiments_scheduled > 0);
 427                         experiments_scheduled --;
 428                         GNUNET_STATISTICS_set (GED_stats, "# experiments scheduled", experiments_scheduled, GNUNET_NO);
 429         }
 430
 431         next = running_out_head;
 432         while (NULL != (cur = next))
 433         {
 434                         next = cur->next;
 435                         GNUNET_CONTAINER_DLL_remove (running_out_head, running_out_tail, cur);
 436                         if (GNUNET_SCHEDULER_NO_TASK != cur->task)
 437                         {
 438                                         GNUNET_SCHEDULER_cancel (cur->task);
 439                                         cur->task = GNUNET_SCHEDULER_NO_TASK;
 440                         }
 441                         GNUNET_free (cur);
 442                         GNUNET_assert (experiments_outbound_running > 0);
 443                         experiments_outbound_running --;
 444                         GNUNET_STATISTICS_set (GED_stats, "# experiments outbound running", experiments_outbound_running, GNUNET_NO);
 445         }
 446 }
 447
 448 /* end of gnunet-daemon-experimentation_scheduler.c */