src/fragmentation/fragmentation.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2004, 2006, 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 2, 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  * @file fragmentation/fragmentation.c
  22  * @brief fragmentation and defragmentation, this code allows
  23  *        sending and receiving messages that are larger than
  24  *        the MTU of the transport.  Messages are still limited
  25  *        to a maximum size of 65535 bytes, which is a good
  26  *        idea because otherwise we may need ungainly fragmentation
  27  *        buffers.  Each connected peer can have at most one
  28  *        fragmented packet at any given point in time (prevents
  29  *        DoS attacks).  Fragmented messages that have not been
  30  *        completed after a certain amount of time are discarded.
  31  * @author Christian Grothoff
  32  */
  33
  34 #include "platform.h"
  35 #include "gnunet_fragmentation_lib.h"
  36 #include "gnunet_protocols.h"
  37 #include "gnunet_util_lib.h"
  38 /**
  39  * Message fragment.  This header is followed
  40  * by the actual data of the fragment.
  41  */
  42 struct Fragment
  43 {
  44
  45   struct GNUNET_MessageHeader header;
  46
  47   /**
  48    * Fragment offset.
  49    */
  50   uint32_t off GNUNET_PACKED;
  51
  52   /**
  53    * "unique" id for the fragment
  54    */
  55   uint64_t id GNUNET_PACKED;
  56
  57   size_t mtu;
  58   uint32_t totalNum;
  59
  60 };
  61
  62 struct GNUNET_FRAGEMENT_Ctxbuffer{
  63         uint64_t id;
  64     uint16_t size;
  65     char * buff;
  66     int counter;
  67     struct GNUNET_TIME_Absolute receivedTime;
  68     struct GNUNET_PeerIdentity *peerID;
  69         struct GNUNET_FRAGEMENT_Ctxbuffer *next;
  70         int * num;
  71 };
  72
  73
  74 /**
  75  * Defragmentation context.
  76  */
  77 struct GNUNET_FRAGMENT_Context
  78 {
  79         uint32_t maxNum;
  80         struct GNUNET_FRAGEMENT_Ctxbuffer *buffer;
  81 };
  82
  83
  84 /**
  85  * Fragment an over-sized message.
  86  *
  87  * @param msg the message to fragment
  88  * @param mtu the maximum message size
  89  * @param proc function to call for each fragment
  90  * @param proc_cls closure for proc
  91  */
  92 void
  93 GNUNET_FRAGMENT_fragment (const struct GNUNET_MessageHeader *msg,
  94                           uint16_t mtu,
  95                           GNUNET_FRAGMENT_MessageProcessor proc,
  96                           void *proc_cls)
  97 {
  98         uint32_t id = GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK, 256);
  99         size_t size = sizeof(struct Fragment);
 100         if(msg->size > mtu){
 101                 uint16_t lastSize = (msg->size) % (mtu-size);
 102                 int num = ceil(msg->size / mtu - size);
 103                 int i;
 104                 for(i = 0; i<num; i++){
 105                         struct Fragment *frag = (struct Fragment *)GNUNET_malloc(size);
 106                         frag->header.type = htons(GNUNET_MESSAGE_TYPE_FRAGMENT);
 107                         frag->id = htonl(id);
 108                         frag->off = htons(mtu*i);
 109                         frag->mtu = htons(mtu);
 110                         if(lastSize!=0){
 111                                 frag->totalNum = htons(num+1);
 112                         }
 113                         else{
 114                                 frag->totalNum = htons(num);
 115                         }
 116             if(i!=num-1){
 117                 frag->header.size = htons(mtu - size);
 118                 memcpy((char*)&frag[1], (char *)&msg[1]+frag->off, mtu - size);
 119             }
 120             else{
 121                 frag->header.size = htons(lastSize);
 122                 memcpy((char*)&frag[1], (char *)&msg[1]+frag->off, lastSize);
 123             }
 124             proc(proc_cls, &frag->header);
 125             free(frag);
 126                 }
 127         }
 128 }
 129
 130
 131 /**
 132  * Create a defragmentation context.
 133  *
 134  * @param stats statistics context
 135  * @param proc function to call with defragmented messages
 136  * @param proc_cls closure for proc
 137  * @return the defragmentation context
 138  */
 139 struct GNUNET_FRAGMENT_Context *
 140 GNUNET_FRAGMENT_context_create (struct GNUNET_STATISTICS_Handle *stats,
 141                                 GNUNET_FRAGMENT_MessageProcessor proc,
 142                                 void *proc_cls)
 143 {
 144         struct GNUNET_FRAGMENT_Context *ctx = (struct GNUNET_FRAGMENT_Context*)GNUNET_malloc(sizeof(struct GNUNET_FRAGMENT_Context));
 145         ctx->maxNum = 100;
 146         ctx->buffer = NULL;
 147         return ctx;
 148 }
 149
 150
 151 /**
 152  * Destroy the given defragmentation context.
 153  */
 154 void
 155 GNUNET_FRAGMENT_context_destroy (struct GNUNET_FRAGMENT_Context *ctx)
 156 {
 157         struct GNUNET_FRAGEMENT_Ctxbuffer *buffer;
 158         for(buffer = ctx->buffer; buffer!=NULL; buffer = buffer->next){
 159                 GNUNET_free(buffer->num);
 160                 GNUNET_free(buffer);
 161         }
 162         GNUNET_free(ctx);
 163         GNUNET_assert (0);
 164 }
 165
 166
 167 /**
 168  * We have received a fragment.  Process it.
 169  *
 170  * @param ctx the context
 171  * @param sender who transmitted the fragment
 172  * @param msg the message that was received
 173  */
 174 void
 175 GNUNET_FRAGMENT_process (struct GNUNET_FRAGMENT_Context *ctx,
 176                          const struct GNUNET_PeerIdentity *sender,
 177                          const struct GNUNET_MessageHeader *msg)
 178 {
 179            uint16_t type = ntohs(msg->type);
 180            int exited = 0, received = 0;
 181        if(type!=GNUNET_MESSAGE_TYPE_FRAGMENT){
 182            return;
 183        }
 184        struct Fragment *frag = (struct Fragment *)msg;
 185        struct GNUNET_FRAGEMENT_Ctxbuffer* buffer;
 186        for(buffer = ctx->buffer; buffer!= NULL; buffer = buffer->next){
 187            if(ctx->buffer->counter == ntohs(frag->totalNum)){return;}
 188            if(buffer->id == ntohl(frag->id)&&(buffer->peerID==sender)){
 189                    exited = 1;
 190                    int i;
 191                    for(i = 0; i<ntohs(frag->totalNum); i++){
 192                      if(buffer->num[i]==ntohs(frag->off)/ntohs(frag->mtu)){
 193                          received = 1;
 194                          break;
 195                      }
 196                }
 197                    if(!received){
 198                            buffer->num[buffer->counter++]=ntohs(frag->off)/ntohs(frag->mtu);
 199                              }
 200                    buffer->receivedTime = GNUNET_TIME_absolute_get ();
 201                    uint16_t size = ntohs(frag->header.size);
 202                    memcpy(&buffer->buff[ntohs(frag->off)], &frag[1], size);
 203                    break;
 204            }
 205        }
 206        if(!exited){
 207            buffer = (struct GNUNET_FRAGEMENT_Ctxbuffer* )GNUNET_malloc(sizeof(struct GNUNET_FRAGEMENT_Ctxbuffer));
 208            buffer->num = (int*)GNUNET_malloc(ntohs(frag->totalNum)*sizeof(int));
 209            buffer->num[buffer->counter++]=ntohs(frag->off)/ntohs(frag->mtu);
 210            memcpy(buffer->peerID,sender,sizeof(struct GNUNET_PeerIdentity));
 211            buffer->receivedTime = GNUNET_TIME_absolute_get ();
 212                    uint16_t size = ntohs(frag->header.size);
 213                    memcpy(&buffer->buff[ntohs(frag->off)], &frag[1], size);
 214        }
 215
 216 }
 217
 218
 219
 220 #if 0
 221
 222 /**
 223  * How many buckets does the fragment hash table
 224  * have?
 225  */
 226 #define DEFRAG_BUCKET_COUNT 16
 227
 228 /**
 229  * After how long do fragments time out?
 230  */
 231 #ifndef DEFRAGMENTATION_TIMEOUT
 232 #define DEFRAGMENTATION_TIMEOUT (3 * GNUNET_CRON_MINUTES)
 233 #endif
 234
 235 /**
 236  * Entry in the linked list of fragments.
 237  */
 238 typedef struct FL
 239 {
 240   struct FL *link;
 241   P2P_fragmentation_MESSAGE *frag;
 242 } FL;
 243
 244 /**
 245  * Entry in the GNUNET_hash table of fragments.
 246  */
 247 typedef struct FC
 248 {
 249   struct FC *next;
 250   FL *head;
 251   GNUNET_PeerIdentity sender;
 252   int id;
 253   GNUNET_CronTime ttl;
 254 } FC;
 255
 256 #define FRAGSIZE(fl) ((ntohs(fl->frag->header.size)-sizeof(P2P_fragmentation_MESSAGE)))
 257
 258 static GNUNET_CoreAPIForPlugins *coreAPI;
 259
 260 static GNUNET_Stats_ServiceAPI *stats;
 261
 262 static int stat_defragmented;
 263
 264 static int stat_fragmented;
 265
 266 static int stat_discarded;
 267
 268 /**
 269  * Hashtable *with* collision management!
 270  */
 271 static FC *defragmentationCache[DEFRAG_BUCKET_COUNT];
 272
 273 /**
 274  * Lock for the defragmentation cache.
 275  */
 276 static struct GNUNET_Mutex *defragCacheLock;
 277
 278 static void
 279 freeFL (FL * fl, int c)
 280 {
 281   while (fl != NULL)
 282     {
 283       FL *link = fl->link;
 284       if (stats != NULL)
 285         stats->change (stat_discarded, c);
 286       GNUNET_free (fl->frag);
 287       GNUNET_free (fl);
 288       fl = link;
 289     }
 290 }
 291
 292 /**
 293  * This cron job ensures that we purge buffers of fragments
 294  * that have timed out.  It can run in much longer intervals
 295  * than the defragmentationCron, e.g. every 60s.
 296  * <p>
 297  * This method goes through the hashtable, finds entries that
 298  * have timed out and removes them (and all the fragments that
 299  * belong to the entry).  It's a bit more complicated as the
 300  * collision list is also collapsed.
 301  */
 302 static void
 303 defragmentationPurgeCron (void *unused)
 304 {
 305   int i;
 306   FC *smf;
 307   FC *next;
 308   FC *last;
 309
 310   GNUNET_mutex_lock (defragCacheLock);
 311   for (i = 0; i < DEFRAG_BUCKET_COUNT; i++)
 312     {
 313       last = NULL;
 314       smf = defragmentationCache[i];
 315       while (smf != NULL)
 316         {
 317           if (smf->ttl < GNUNET_get_time ())
 318             {
 319               /* free linked list of fragments */
 320               freeFL (smf->head, 1);
 321               next = smf->next;
 322               GNUNET_free (smf);
 323               if (last == NULL)
 324                 defragmentationCache[i] = next;
 325               else
 326                 last->next = next;
 327               smf = next;
 328             }
 329           else
 330             {
 331               last = smf;
 332               smf = smf->next;
 333             }
 334         }                       /* while smf != NULL */
 335     }                           /* for all buckets */
 336   GNUNET_mutex_unlock (defragCacheLock);
 337 }
 338
 339 /**
 340  * Check if this fragment-list is complete.  If yes, put it together,
 341  * process and free all buffers.  Does not free the pep
 342  * itself (but sets the TTL to 0 to have the cron free it
 343  * in the next iteration).
 344  *
 345  * @param pep the entry in the GNUNET_hash table
 346  */
 347 static void
 348 checkComplete (FC * pep)
 349 {
 350   FL *pos;
 351   unsigned short off;
 352   unsigned short len;
 353   char *msg;
 354
 355   GNUNET_GE_ASSERT (NULL, pep != NULL);
 356   pos = pep->head;
 357   if (pos == NULL)
 358     return;
 359   len = ntohs (pos->frag->len);
 360   if (len == 0)
 361     goto CLEANUP;               /* really bad error! */
 362   off = 0;
 363   while ((pos != NULL) && (ntohs (pos->frag->off) <= off))
 364     {
 365       if (off >= off + FRAGSIZE (pos))
 366         goto CLEANUP;           /* error! */
 367       if (ntohs (pos->frag->off) + FRAGSIZE (pos) > off)
 368         off = ntohs (pos->frag->off) + FRAGSIZE (pos);
 369       else
 370         goto CLEANUP;           /* error! */
 371       pos = pos->link;
 372     }
 373   if (off < len)
 374     return;                     /* some fragment is still missing */
 375
 376   msg = GNUNET_malloc (len);
 377   pos = pep->head;
 378   while (pos != NULL)
 379     {
 380       memcpy (&msg[ntohs (pos->frag->off)], &pos->frag[1], FRAGSIZE (pos));
 381       pos = pos->link;
 382     }
 383   if (stats != NULL)
 384     stats->change (stat_defragmented, 1);
 385 #if 0
 386   printf ("Finished defragmentation!\n");
 387 #endif
 388   /* handle message! */
 389   coreAPI->loopback_send (&pep->sender, msg, len, GNUNET_YES, NULL);
 390   GNUNET_free (msg);
 391 CLEANUP:
 392   /* free fragment buffers */
 393   freeFL (pep->head, 0);
 394   pep->head = NULL;
 395   pep->ttl = 0;
 396 }
 397
 398 /**
 399  * See if the new fragment is a part of this entry and join them if
 400  * yes.  Return GNUNET_SYSERR if the fragments do not match.  Return GNUNET_OK if
 401  * the fragments do match and the fragment has been processed.  The
 402  * defragCacheLock is already acquired by the caller whenever this
 403  * method is called.<p>
 404  *
 405  * @param entry the entry in the cache
 406  * @param pep the new entry
 407  * @param packet the ip part in the new entry
 408  */
 409 static int
 410 tryJoin (FC * entry,
 411          const GNUNET_PeerIdentity * sender,
 412          const P2P_fragmentation_MESSAGE * packet)
 413 {
 414   /* frame before ours; may end in the middle of
 415      our frame or before it starts; NULL if we are
 416      the earliest position we have received so far */
 417   FL *before;
 418   /* frame after ours; may start in the middle of
 419      our frame or after it; NULL if we are the last
 420      fragment we have received so far */
 421   FL *after;
 422   /* current position in the frame-list */
 423   FL *pos;
 424   /* the new entry that we're inserting */
 425   FL *pep;
 426   FL *tmp;
 427   unsigned short end;
 428
 429   GNUNET_GE_ASSERT (NULL, entry != NULL);
 430   if (0 != memcmp (sender, &entry->sender, sizeof (GNUNET_PeerIdentity)))
 431     return GNUNET_SYSERR;       /* wrong fragment list, try another! */
 432   if (ntohl (packet->id) != entry->id)
 433     return GNUNET_SYSERR;       /* wrong fragment list, try another! */
 434 #if 0
 435   printf ("Received fragment %u from %u to %u\n",
 436           ntohl (packet->id),
 437           ntohs (packet->off),
 438           ntohs (packet->off) + ntohs (packet->header.size) -
 439           sizeof (P2P_fragmentation_MESSAGE));
 440 #endif
 441   pos = entry->head;
 442   if ((pos != NULL) && (packet->len != pos->frag->len))
 443     return GNUNET_SYSERR;       /* wrong fragment size */
 444
 445   before = NULL;
 446   /* find the before-frame */
 447   while ((pos != NULL) && (ntohs (pos->frag->off) < ntohs (packet->off)))
 448     {
 449       before = pos;
 450       pos = pos->link;
 451     }
 452
 453   /* find the after-frame */
 454   end =
 455     ntohs (packet->off) + ntohs (packet->header.size) -
 456     sizeof (P2P_fragmentation_MESSAGE);
 457   if (end <= ntohs (packet->off))
 458     {
 459       GNUNET_GE_LOG (NULL,
 460                      GNUNET_GE_DEVELOPER | GNUNET_GE_DEBUG | GNUNET_GE_BULK,
 461                      "Received invalid fragment at %s:%d\n", __FILE__,
 462                      __LINE__);
 463       return GNUNET_SYSERR;     /* yuck! integer overflow! */
 464     }
 465
 466   if (before != NULL)
 467     after = before;
 468   else
 469     after = entry->head;
 470   while ((after != NULL) && (ntohs (after->frag->off) < end))
 471     after = after->link;
 472
 473   if ((before != NULL) && (before == after))
 474     {
 475       /* this implies after or before != NULL and thereby the new
 476          fragment is redundant as it is fully enclosed in an earlier
 477          fragment */
 478       if (stats != NULL)
 479         stats->change (stat_defragmented, 1);
 480       return GNUNET_OK;         /* drop, there is a packet that spans our range! */
 481     }
 482
 483   if ((before != NULL) &&
 484       (after != NULL) &&
 485       ((htons (before->frag->off) +
 486         FRAGSIZE (before)) >= htons (after->frag->off)))
 487     {
 488       /* this implies that the fragment that starts before us and the
 489          fragment that comes after this one leave no space in the middle
 490          or even overlap; thus we can drop this redundant piece */
 491       if (stats != NULL)
 492         stats->change (stat_defragmented, 1);
 493       return GNUNET_OK;
 494     }
 495
 496   /* allocate pep */
 497   pep = GNUNET_malloc (sizeof (FC));
 498   pep->frag = GNUNET_malloc (ntohs (packet->header.size));
 499   memcpy (pep->frag, packet, ntohs (packet->header.size));
 500   pep->link = NULL;
 501
 502   if (before == NULL)
 503     {
 504       pep->link = after;
 505       pos = entry->head;
 506       while (pos != after)
 507         {
 508           tmp = pos->link;
 509           GNUNET_free (pos->frag);
 510           GNUNET_free (pos);
 511           pos = tmp;
 512         }
 513       entry->head = pep;
 514       goto FINISH;
 515       /* end of insert first */
 516     }
 517
 518   if (after == NULL)
 519     {
 520       /* insert last: find the end, free everything after it */
 521       freeFL (before->link, 1);
 522       before->link = pep;
 523       goto FINISH;
 524     }
 525
 526   /* ok, we are filling the middle between two fragments; insert.  If
 527      there is anything else in the middle, it can be dropped as we're
 528      bigger & cover that area as well */
 529   /* free everything between before and after */
 530   pos = before->link;
 531   while (pos != after)
 532     {
 533       tmp = pos->link;
 534       GNUNET_free (pos->frag);
 535       GNUNET_free (pos);
 536       pos = tmp;
 537     }
 538   before->link = pep;
 539   pep->link = after;
 540
 541 FINISH:
 542   entry->ttl = GNUNET_get_time () + DEFRAGMENTATION_TIMEOUT;
 543   checkComplete (entry);
 544   return GNUNET_OK;
 545 }
 546
 547 /**
 548  * Defragment the given fragment and pass to handler once
 549  * defragmentation is complete.
 550  *
 551  * @param frag the packet to defragment
 552  * @return GNUNET_SYSERR if the fragment is invalid
 553  */
 554 static int
 555 processFragment (const GNUNET_PeerIdentity * sender,
 556                  const GNUNET_MessageHeader * frag)
 557 {
 558   unsigned int hash;
 559   FC *smf;
 560
 561   if (ntohs (frag->size) < sizeof (P2P_fragmentation_MESSAGE))
 562     return GNUNET_SYSERR;
 563
 564   GNUNET_mutex_lock (defragCacheLock);
 565   hash = sender->hashPubKey.bits[0] % DEFRAG_BUCKET_COUNT;
 566   smf = defragmentationCache[hash];
 567   while (smf != NULL)
 568     {
 569       if (GNUNET_OK ==
 570           tryJoin (smf, sender, (P2P_fragmentation_MESSAGE *) frag))
 571         {
 572           GNUNET_mutex_unlock (defragCacheLock);
 573           return GNUNET_OK;
 574         }
 575       if (0 == memcmp (sender, &smf->sender, sizeof (GNUNET_PeerIdentity)))
 576         {
 577           freeFL (smf->head, 1);
 578           break;
 579         }
 580       smf = smf->next;
 581     }
 582   if (smf == NULL)
 583     {
 584       smf = GNUNET_malloc (sizeof (FC));
 585       smf->next = defragmentationCache[hash];
 586       defragmentationCache[hash] = smf;
 587       smf->ttl = GNUNET_get_time () + DEFRAGMENTATION_TIMEOUT;
 588       smf->sender = *sender;
 589     }
 590   smf->id = ntohl (((P2P_fragmentation_MESSAGE *) frag)->id);
 591   smf->head = GNUNET_malloc (sizeof (FL));
 592   smf->head->link = NULL;
 593   smf->head->frag = GNUNET_malloc (ntohs (frag->size));
 594   memcpy (smf->head->frag, frag, ntohs (frag->size));
 595
 596   GNUNET_mutex_unlock (defragCacheLock);
 597   return GNUNET_OK;
 598 }
 599
 600 typedef struct
 601 {
 602   GNUNET_PeerIdentity sender;
 603   /* maximums size of each fragment */
 604   unsigned short mtu;
 605   /** how long is this message part expected to be? */
 606   unsigned short len;
 607   /** when did we intend to transmit? */
 608   GNUNET_CronTime transmissionTime;
 609 } FragmentBMC;
 610
 611 /**
 612  * Send a message that had to be fragmented (right now!).  First grabs
 613  * the first part of the message (obtained from ctx->se) and stores
 614  * that in a P2P_fragmentation_MESSAGE envelope.  The remaining fragments are
 615  * added to the send queue with GNUNET_EXTREME_PRIORITY (to ensure that they
 616  * will be transmitted next).  The logic here is that if the priority
 617  * for the first fragment was sufficiently high, the priority should
 618  * also have been sufficiently high for all of the other fragments (at
 619  * this time) since they have the same priority.  And we want to make
 620  * sure that we send all of them since just sending the first fragment
 621  * and then going to other messages of equal priority would not be
 622  * such a great idea (i.e. would just waste bandwidth).
 623  */
 624 static int
 625 fragmentBMC (void *buf, void *cls, unsigned short len)
 626 {
 627   FragmentBMC *ctx = cls;
 628   static int idGen = 0;
 629   P2P_fragmentation_MESSAGE *frag;
 630   unsigned int pos;
 631   int id;
 632   unsigned short mlen;
 633
 634   if ((len < ctx->mtu) || (buf == NULL))
 635     {
 636       GNUNET_free (ctx);
 637       return GNUNET_SYSERR;
 638     }
 639   if (stats != NULL)
 640     stats->change (stat_fragmented, 1);
 641   id = (idGen++) + GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK, 512);
 642   /* write first fragment to buf */
 643   frag = (P2P_fragmentation_MESSAGE *) buf;
 644   frag->header.size = htons (len);
 645   frag->header.type = htons (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
 646   frag->id = id;
 647   frag->off = htons (0);
 648   frag->len = htons (ctx->len);
 649   memcpy (&frag[1], &ctx[1], len - sizeof (P2P_fragmentation_MESSAGE));
 650
 651   /* create remaining fragments, add to queue! */
 652   pos = len - sizeof (P2P_fragmentation_MESSAGE);
 653   frag = GNUNET_malloc (ctx->mtu);
 654   while (pos < ctx->len)
 655     {
 656       mlen = sizeof (P2P_fragmentation_MESSAGE) + ctx->len - pos;
 657       if (mlen > ctx->mtu)
 658         mlen = ctx->mtu;
 659       GNUNET_GE_ASSERT (NULL, mlen > sizeof (P2P_fragmentation_MESSAGE));
 660       frag->header.size = htons (mlen);
 661       frag->header.type = htons (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
 662       frag->id = id;
 663       frag->off = htons (pos);
 664       frag->len = htons (ctx->len);
 665       memcpy (&frag[1],
 666               &((char *) (&ctx[1]))[pos],
 667               mlen - sizeof (P2P_fragmentation_MESSAGE));
 668       coreAPI->ciphertext_send (&ctx->sender,
 669                                 &frag->header,
 670                                 GNUNET_EXTREME_PRIORITY,
 671                                 ctx->transmissionTime - GNUNET_get_time ());
 672       pos += mlen - sizeof (P2P_fragmentation_MESSAGE);
 673     }
 674   GNUNET_GE_ASSERT (NULL, pos == ctx->len);
 675   GNUNET_free (frag);
 676   GNUNET_free (ctx);
 677   return GNUNET_OK;
 678 }
 679
 680 /**
 681  * The given message must be fragmented.  Produce a placeholder that
 682  * corresponds to the first fragment.  Once that fragment is scheduled
 683  * for transmission, the placeholder should automatically add all of
 684  * the other fragments (with very high priority).
 685  */
 686 void
 687 fragment (const GNUNET_PeerIdentity * peer,
 688           unsigned int mtu,
 689           unsigned int prio,
 690           unsigned int targetTime,
 691           unsigned int len, GNUNET_BuildMessageCallback bmc, void *bmcClosure)
 692 {
 693   FragmentBMC *fbmc;
 694   int xlen;
 695
 696   GNUNET_GE_ASSERT (NULL, len > mtu);
 697   GNUNET_GE_ASSERT (NULL, mtu > sizeof (P2P_fragmentation_MESSAGE));
 698   fbmc = GNUNET_malloc (sizeof (FragmentBMC) + len);
 699   fbmc->mtu = mtu;
 700   fbmc->sender = *peer;
 701   fbmc->transmissionTime = targetTime;
 702   fbmc->len = len;
 703   if (bmc == NULL)
 704     {
 705       memcpy (&fbmc[1], bmcClosure, len);
 706       GNUNET_free (bmcClosure);
 707     }
 708   else
 709     {
 710       if (GNUNET_SYSERR == bmc (&fbmc[1], bmcClosure, len))
 711         {
 712           GNUNET_free (fbmc);
 713           return;
 714         }
 715     }
 716   xlen = mtu - sizeof (P2P_fragmentation_MESSAGE);
 717   coreAPI->ciphertext_send_with_callback (peer, &fragmentBMC, fbmc, mtu, prio * xlen / len,     /* compute new priority */
 718                                           targetTime);
 719 }
 720
 721 /**
 722  * Initialize Fragmentation module.
 723  */
 724 GNUNET_Fragmentation_ServiceAPI *
 725 provide_module_fragmentation (GNUNET_CoreAPIForPlugins * capi)
 726 {
 727   static GNUNET_Fragmentation_ServiceAPI ret;
 728   int i;
 729
 730   coreAPI = capi;
 731   stats = coreAPI->service_request ("stats");
 732   if (stats != NULL)
 733     {
 734       stat_defragmented =
 735         stats->create (gettext_noop ("# messages defragmented"));
 736       stat_fragmented =
 737         stats->create (gettext_noop ("# messages fragmented"));
 738       stat_discarded = stats->create (gettext_noop ("# fragments discarded"));
 739     }
 740   for (i = 0; i < DEFRAG_BUCKET_COUNT; i++)
 741     defragmentationCache[i] = NULL;
 742   defragCacheLock = GNUNET_mutex_create (GNUNET_NO);
 743   GNUNET_cron_add_job (coreAPI->cron,
 744                        &defragmentationPurgeCron,
 745                        60 * GNUNET_CRON_SECONDS, 60 * GNUNET_CRON_SECONDS,
 746                        NULL);
 747   GNUNET_GE_LOG (capi->ectx,
 748                  GNUNET_GE_INFO | GNUNET_GE_USER | GNUNET_GE_REQUEST,
 749                  _("`%s' registering handler %d\n"), "fragmentation",
 750                  GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
 751   capi->p2p_ciphertext_handler_register (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT,
 752                                          &processFragment);
 753
 754   ret.fragment = &fragment;
 755   return &ret;
 756 }
 757
 758 /**
 759  * Shutdown fragmentation.
 760  */
 761 void
 762 release_module_fragmentation ()
 763 {
 764   int i;
 765
 766   coreAPI->p2p_ciphertext_handler_unregister
 767     (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT, &processFragment);
 768   GNUNET_cron_del_job (coreAPI->cron, &defragmentationPurgeCron,
 769                        60 * GNUNET_CRON_SECONDS, NULL);
 770   for (i = 0; i < DEFRAG_BUCKET_COUNT; i++)
 771     {
 772       FC *pos = defragmentationCache[i];
 773       while (pos != NULL)
 774         {
 775           FC *next = pos->next;
 776           freeFL (pos->head, 1);
 777           GNUNET_free (pos);
 778           pos = next;
 779         }
 780     }
 781   if (stats != NULL)
 782     {
 783       coreAPI->service_release (stats);
 784       stats = NULL;
 785     }
 786   GNUNET_mutex_destroy (defragCacheLock);
 787   defragCacheLock = NULL;
 788   coreAPI = NULL;
 789 }
 790
 791 #endif
 792
 793 /* end of fragmentation.c */