src/fragmentation/defragmentation.c

   1 /*
   2      This file is part of GNUnet
   3      Copyright (C) 2009, 2011 GNUnet e.V.
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20 /**
  21  * @file src/fragmentation/defragmentation.c
  22  * @brief library to help defragment messages
  23  * @author Christian Grothoff
  24  */
  25 #include "platform.h"
  26 #include "gnunet_fragmentation_lib.h"
  27 #include "fragmentation.h"
  28
  29 /**
  30  * Timestamps for fragments.
  31  */
  32 struct FragTimes
  33 {
  34   /**
  35    * The time the fragment was received.
  36    */
  37   struct GNUNET_TIME_Absolute time;
  38
  39   /**
  40    * Number of the bit for the fragment (in [0,..,63]).
  41    */
  42   unsigned int bit;
  43 };
  44
  45
  46 /**
  47  * Information we keep for one message that is being assembled.  Note
  48  * that we keep the context around even after the assembly is done to
  49  * handle 'stray' messages that are received 'late'.  A message
  50  * context is ONLY discarded when the queue gets too big.
  51  */
  52 struct MessageContext
  53 {
  54   /**
  55    * This is a DLL.
  56    */
  57   struct MessageContext *next;
  58
  59   /**
  60    * This is a DLL.
  61    */
  62   struct MessageContext *prev;
  63
  64   /**
  65    * Associated defragmentation context.
  66    */
  67   struct GNUNET_DEFRAGMENT_Context *dc;
  68
  69   /**
  70    * Pointer to the assembled message, allocated at the
  71    * end of this struct.
  72    */
  73   const struct GNUNET_MessageHeader *msg;
  74
  75   /**
  76    * Last time we received any update for this message
  77    * (least-recently updated message will be discarded
  78    * if we hit the queue size).
  79    */
  80   struct GNUNET_TIME_Absolute last_update;
  81
  82   /**
  83    * Task scheduled for transmitting the next ACK to the
  84    * other peer.
  85    */
  86   struct GNUNET_SCHEDULER_Task * ack_task;
  87
  88   /**
  89    * When did we receive which fragment? Used to calculate
  90    * the time we should send the ACK.
  91    */
  92   struct FragTimes frag_times[64];
  93
  94   /**
  95    * Which fragments have we gotten yet? bits that are 1
  96    * indicate missing fragments.
  97    */
  98   uint64_t bits;
  99
 100   /**
 101    * Unique ID for this message.
 102    */
 103   uint32_t fragment_id;
 104
 105   /**
 106    * Which 'bit' did the last fragment we received correspond to?
 107    */
 108   unsigned int last_bit;
 109
 110   /**
 111    * For the current ACK round, which is the first relevant
 112    * offset in @e frag_times?
 113    */
 114   unsigned int frag_times_start_offset;
 115
 116   /**
 117    * Which offset whould we write the next frag value into
 118    * in the @e frag_times array? All smaller entries are valid.
 119    */
 120   unsigned int frag_times_write_offset;
 121
 122   /**
 123    * Total size of the message that we are assembling.
 124    */
 125   uint16_t total_size;
 126
 127   /**
 128    * Was the last fragment we got a duplicate?
 129    */
 130   int16_t last_duplicate;
 131
 132 };
 133
 134
 135 /**
 136  * Defragmentation context (one per connection).
 137  */
 138 struct GNUNET_DEFRAGMENT_Context
 139 {
 140
 141   /**
 142    * For statistics.
 143    */
 144   struct GNUNET_STATISTICS_Handle *stats;
 145
 146   /**
 147    * Head of list of messages we're defragmenting.
 148    */
 149   struct MessageContext *head;
 150
 151   /**
 152    * Tail of list of messages we're defragmenting.
 153    */
 154   struct MessageContext *tail;
 155
 156   /**
 157    * Closure for @e proc and @e ackp.
 158    */
 159   void *cls;
 160
 161   /**
 162    * Function to call with defragmented messages.
 163    */
 164   GNUNET_FRAGMENT_MessageProcessor proc;
 165
 166   /**
 167    * Function to call with acknowledgements.
 168    */
 169   GNUNET_DEFRAGMENT_AckProcessor ackp;
 170
 171   /**
 172    * Running average of the latency (delay between messages) for this
 173    * connection.
 174    */
 175   struct GNUNET_TIME_Relative latency;
 176
 177   /**
 178    * num_msgs how many fragmented messages
 179    * to we defragment at most at the same time?
 180    */
 181   unsigned int num_msgs;
 182
 183   /**
 184    * Current number of messages in the 'struct MessageContext'
 185    * DLL (smaller or equal to 'num_msgs').
 186    */
 187   unsigned int list_size;
 188
 189   /**
 190    * Maximum message size for each fragment.
 191    */
 192   uint16_t mtu;
 193
 194 };
 195
 196
 197 /**
 198  * Create a defragmentation context.
 199  *
 200  * @param stats statistics context
 201  * @param mtu the maximum message size for each fragment
 202  * @param num_msgs how many fragmented messages
 203  *                 to we defragment at most at the same time?
 204  * @param cls closure for @a proc and @a ackp
 205  * @param proc function to call with defragmented messages
 206  * @param ackp function to call with acknowledgements (to send
 207  *             back to the other side)
 208  * @return the defragmentation context
 209  */
 210 struct GNUNET_DEFRAGMENT_Context *
 211 GNUNET_DEFRAGMENT_context_create (struct GNUNET_STATISTICS_Handle *stats,
 212                                   uint16_t mtu, unsigned int num_msgs,
 213                                   void *cls,
 214                                   GNUNET_FRAGMENT_MessageProcessor proc,
 215                                   GNUNET_DEFRAGMENT_AckProcessor ackp)
 216 {
 217   struct GNUNET_DEFRAGMENT_Context *dc;
 218
 219   dc = GNUNET_new (struct GNUNET_DEFRAGMENT_Context);
 220   dc->stats = stats;
 221   dc->cls = cls;
 222   dc->proc = proc;
 223   dc->ackp = ackp;
 224   dc->num_msgs = num_msgs;
 225   dc->mtu = mtu;
 226   dc->latency = GNUNET_TIME_UNIT_SECONDS;       /* start with likely overestimate */
 227   return dc;
 228 }
 229
 230
 231 /**
 232  * Destroy the given defragmentation context.
 233  *
 234  * @param dc defragmentation context
 235  */
 236 void
 237 GNUNET_DEFRAGMENT_context_destroy (struct GNUNET_DEFRAGMENT_Context *dc)
 238 {
 239   struct MessageContext *mc;
 240
 241   while (NULL != (mc = dc->head))
 242   {
 243     GNUNET_CONTAINER_DLL_remove (dc->head, dc->tail, mc);
 244     dc->list_size--;
 245     if (NULL != mc->ack_task)
 246     {
 247       GNUNET_SCHEDULER_cancel (mc->ack_task);
 248       mc->ack_task = NULL;
 249     }
 250     GNUNET_free (mc);
 251   }
 252   GNUNET_assert (0 == dc->list_size);
 253   GNUNET_free (dc);
 254 }
 255
 256
 257 /**
 258  * Send acknowledgement to the other peer now.
 259  *
 260  * @param cls the message context
 261  */
 262 static void
 263 send_ack (void *cls)
 264 {
 265   struct MessageContext *mc = cls;
 266   struct GNUNET_DEFRAGMENT_Context *dc = mc->dc;
 267   struct FragmentAcknowledgement fa;
 268
 269   mc->ack_task = NULL;
 270   fa.header.size = htons (sizeof (struct FragmentAcknowledgement));
 271   fa.header.type = htons (GNUNET_MESSAGE_TYPE_FRAGMENT_ACK);
 272   fa.fragment_id = htonl (mc->fragment_id);
 273   fa.bits = GNUNET_htonll (mc->bits);
 274   GNUNET_STATISTICS_update (mc->dc->stats,
 275                             _("# acknowledgements sent for fragment"),
 276                             1,
 277                             GNUNET_NO);
 278   mc->last_duplicate = GNUNET_NO; /* clear flag */
 279   dc->ackp (dc->cls,
 280             mc->fragment_id,
 281             &fa.header);
 282 }
 283
 284
 285 /**
 286  * This function is from the GNU Scientific Library, linear/fit.c,
 287  * Copyright (C) 2000 Brian Gough
 288  */
 289 static void
 290 gsl_fit_mul (const double *x, const size_t xstride, const double *y,
 291              const size_t ystride, const size_t n, double *c1, double *cov_11,
 292              double *sumsq)
 293 {
 294   double m_x = 0, m_y = 0, m_dx2 = 0, m_dxdy = 0;
 295
 296   size_t i;
 297
 298   for (i = 0; i < n; i++)
 299   {
 300     m_x += (x[i * xstride] - m_x) / (i + 1.0);
 301     m_y += (y[i * ystride] - m_y) / (i + 1.0);
 302   }
 303
 304   for (i = 0; i < n; i++)
 305   {
 306     const double dx = x[i * xstride] - m_x;
 307     const double dy = y[i * ystride] - m_y;
 308
 309     m_dx2 += (dx * dx - m_dx2) / (i + 1.0);
 310     m_dxdy += (dx * dy - m_dxdy) / (i + 1.0);
 311   }
 312
 313   /* In terms of y =  b x */
 314
 315   {
 316     double s2 = 0, d2 = 0;
 317     double b = (m_x * m_y + m_dxdy) / (m_x * m_x + m_dx2);
 318
 319     *c1 = b;
 320
 321     /* Compute chi^2 = \sum (y_i -  b * x_i)^2 */
 322
 323     for (i = 0; i < n; i++)
 324     {
 325       const double dx = x[i * xstride] - m_x;
 326       const double dy = y[i * ystride] - m_y;
 327       const double d = (m_y - b * m_x) + dy - b * dx;
 328
 329       d2 += d * d;
 330     }
 331
 332     s2 = d2 / (n - 1.0);        /* chisq per degree of freedom */
 333
 334     *cov_11 = s2 * 1.0 / (n * (m_x * m_x + m_dx2));
 335
 336     *sumsq = d2;
 337   }
 338 }
 339
 340
 341 /**
 342  * Estimate the latency between messages based on the most recent
 343  * message time stamps.
 344  *
 345  * @param mc context with time stamps
 346  * @return average delay between time stamps (based on least-squares fit)
 347  */
 348 static struct GNUNET_TIME_Relative
 349 estimate_latency (struct MessageContext *mc)
 350 {
 351   struct FragTimes *first;
 352   size_t total = mc->frag_times_write_offset - mc->frag_times_start_offset;
 353   double x[total];
 354   double y[total];
 355   size_t i;
 356   double c1;
 357   double cov11;
 358   double sumsq;
 359   struct GNUNET_TIME_Relative ret;
 360
 361   first = &mc->frag_times[mc->frag_times_start_offset];
 362   GNUNET_assert (total > 1);
 363   for (i = 0; i < total; i++)
 364   {
 365     x[i] = (double) i;
 366     y[i] = (double) (first[i].time.abs_value_us - first[0].time.abs_value_us);
 367   }
 368   gsl_fit_mul (x, 1, y, 1, total, &c1, &cov11, &sumsq);
 369   c1 += sqrt (sumsq);           /* add 1 std dev */
 370   ret.rel_value_us = (uint64_t) c1;
 371   if (0 == ret.rel_value_us)
 372     ret = GNUNET_TIME_UNIT_MICROSECONDS;        /* always at least 1 */
 373   return ret;
 374 }
 375
 376
 377 /**
 378  * Discard the message context that was inactive for the longest time.
 379  *
 380  * @param dc defragmentation context
 381  */
 382 static void
 383 discard_oldest_mc (struct GNUNET_DEFRAGMENT_Context *dc)
 384 {
 385   struct MessageContext *old;
 386   struct MessageContext *pos;
 387
 388   old = NULL;
 389   pos = dc->head;
 390   while (NULL != pos)
 391   {
 392     if ((old == NULL) ||
 393         (old->last_update.abs_value_us > pos->last_update.abs_value_us))
 394       old = pos;
 395     pos = pos->next;
 396   }
 397   GNUNET_assert (NULL != old);
 398   GNUNET_CONTAINER_DLL_remove (dc->head, dc->tail, old);
 399   dc->list_size--;
 400   if (NULL != old->ack_task)
 401   {
 402     GNUNET_SCHEDULER_cancel (old->ack_task);
 403     old->ack_task = NULL;
 404   }
 405   GNUNET_free (old);
 406 }
 407
 408
 409 /**
 410  * We have received a fragment.  Process it.
 411  *
 412  * @param dc the context
 413  * @param msg the message that was received
 414  * @return #GNUNET_OK on success,
 415  *         #GNUNET_NO if this was a duplicate,
 416  *         #GNUNET_SYSERR on error
 417  */
 418 int
 419 GNUNET_DEFRAGMENT_process_fragment (struct GNUNET_DEFRAGMENT_Context *dc,
 420                                     const struct GNUNET_MessageHeader *msg)
 421 {
 422   struct MessageContext *mc;
 423   const struct FragmentHeader *fh;
 424   uint16_t msize;
 425   uint16_t foff;
 426   uint32_t fid;
 427   char *mbuf;
 428   unsigned int bit;
 429   struct GNUNET_TIME_Absolute now;
 430   struct GNUNET_TIME_Relative delay;
 431   unsigned int bc;
 432   unsigned int b;
 433   unsigned int n;
 434   unsigned int num_fragments;
 435   int duplicate;
 436   int last;
 437
 438   if (ntohs (msg->size) < sizeof (struct FragmentHeader))
 439   {
 440     GNUNET_break_op (0);
 441     return GNUNET_SYSERR;
 442   }
 443   if (ntohs (msg->size) > dc->mtu)
 444   {
 445     GNUNET_break_op (0);
 446     return GNUNET_SYSERR;
 447   }
 448   fh = (const struct FragmentHeader *) msg;
 449   msize = ntohs (fh->total_size);
 450   if (msize < sizeof (struct GNUNET_MessageHeader))
 451   {
 452     GNUNET_break_op (0);
 453     return GNUNET_SYSERR;
 454   }
 455   fid = ntohl (fh->fragment_id);
 456   foff = ntohs (fh->offset);
 457   if (foff >= msize)
 458   {
 459     GNUNET_break_op (0);
 460     return GNUNET_SYSERR;
 461   }
 462   if (0 != (foff % (dc->mtu - sizeof (struct FragmentHeader))))
 463   {
 464     GNUNET_break_op (0);
 465     return GNUNET_SYSERR;
 466   }
 467   GNUNET_STATISTICS_update (dc->stats,
 468                             _("# fragments received"),
 469                             1,
 470                             GNUNET_NO);
 471   num_fragments = (ntohs (msg->size) + dc->mtu - sizeof (struct FragmentHeader)-1) / (dc->mtu - sizeof (struct FragmentHeader));
 472   last = 0;
 473   for (mc = dc->head; NULL != mc; mc = mc->next)
 474     if (mc->fragment_id > fid)
 475       last++;
 476
 477   mc = dc->head;
 478   while ((NULL != mc) && (fid != mc->fragment_id))
 479     mc = mc->next;
 480   bit = foff / (dc->mtu - sizeof (struct FragmentHeader));
 481   if (bit * (dc->mtu - sizeof (struct FragmentHeader)) + ntohs (msg->size) -
 482       sizeof (struct FragmentHeader) > msize)
 483   {
 484     /* payload extends past total message size */
 485     GNUNET_break_op (0);
 486     return GNUNET_SYSERR;
 487   }
 488   if ((NULL != mc) && (msize != mc->total_size))
 489   {
 490     /* inconsistent message size */
 491     GNUNET_break_op (0);
 492     return GNUNET_SYSERR;
 493   }
 494   now = GNUNET_TIME_absolute_get ();
 495   if (NULL == mc)
 496   {
 497     mc = GNUNET_malloc (sizeof (struct MessageContext) + msize);
 498     mc->msg = (const struct GNUNET_MessageHeader *) &mc[1];
 499     mc->dc = dc;
 500     mc->total_size = msize;
 501     mc->fragment_id = fid;
 502     mc->last_update = now;
 503     n = (msize + dc->mtu - sizeof (struct FragmentHeader) - 1) / (dc->mtu -
 504                                                                   sizeof (struct
 505                                                                           FragmentHeader));
 506     if (n == 64)
 507       mc->bits = UINT64_MAX;    /* set all 64 bit */
 508     else
 509       mc->bits = (1LLU << n) - 1;        /* set lowest 'bits' bit */
 510     if (dc->list_size >= dc->num_msgs)
 511       discard_oldest_mc (dc);
 512     GNUNET_CONTAINER_DLL_insert (dc->head,
 513                                  dc->tail,
 514                                  mc);
 515     dc->list_size++;
 516   }
 517
 518   /* copy data to 'mc' */
 519   if (0 != (mc->bits & (1LLU << bit)))
 520   {
 521     mc->bits -= 1LLU << bit;
 522     mbuf = (char *) &mc[1];
 523     GNUNET_memcpy (&mbuf[bit * (dc->mtu - sizeof (struct FragmentHeader))], &fh[1],
 524             ntohs (msg->size) - sizeof (struct FragmentHeader));
 525     mc->last_update = now;
 526     if (bit < mc->last_bit)
 527       mc->frag_times_start_offset = mc->frag_times_write_offset;
 528     mc->last_bit = bit;
 529     mc->frag_times[mc->frag_times_write_offset].time = now;
 530     mc->frag_times[mc->frag_times_write_offset].bit = bit;
 531     mc->frag_times_write_offset++;
 532     duplicate = GNUNET_NO;
 533   }
 534   else
 535   {
 536     duplicate = GNUNET_YES;
 537     GNUNET_STATISTICS_update (dc->stats,
 538                               _("# duplicate fragments received"),
 539                               1,
 540                               GNUNET_NO);
 541   }
 542
 543   /* count number of missing fragments after the current one */
 544   bc = 0;
 545   for (b = bit; b < 64; b++)
 546     if (0 != (mc->bits & (1LLU << b)))
 547       bc++;
 548     else
 549       bc = 0;
 550
 551   /* notify about complete message */
 552   if ( (GNUNET_NO == duplicate) &&
 553        (0 == mc->bits) )
 554   {
 555     GNUNET_STATISTICS_update (dc->stats,
 556                               _("# messages defragmented"),
 557                               1,
 558                               GNUNET_NO);
 559     /* message complete, notify! */
 560     dc->proc (dc->cls, mc->msg);
 561   }
 562   /* send ACK */
 563   if (mc->frag_times_write_offset - mc->frag_times_start_offset > 1)
 564   {
 565     dc->latency = estimate_latency (mc);
 566   }
 567   delay = GNUNET_TIME_relative_saturating_multiply (dc->latency,
 568                                                     bc + 1);
 569   if ( (last + fid == num_fragments) ||
 570        (0 == mc->bits) ||
 571        (GNUNET_YES == duplicate) )
 572   {
 573     /* message complete or duplicate or last missing fragment in
 574        linear sequence; ACK now! */
 575     delay = GNUNET_TIME_UNIT_ZERO;
 576   }
 577   if (NULL != mc->ack_task)
 578     GNUNET_SCHEDULER_cancel (mc->ack_task);
 579   mc->ack_task = GNUNET_SCHEDULER_add_delayed (delay,
 580                                                &send_ack,
 581                                                mc);
 582   if (GNUNET_YES == duplicate)
 583   {
 584     mc->last_duplicate = GNUNET_YES;
 585     return GNUNET_NO;
 586   }
 587   return GNUNET_YES;
 588 }
 589
 590 /* end of defragmentation.c */