fragmentation.c

GNUnet是一个安全的点对点网络框架

         fragment is redundant as it is fully enclosed in an earlier
         fragment */
      if (stats != NULL)
        stats->change (stat_defragmented, 1);
      return GNUNET_OK;         /* drop, there is a packet that spans our range! */
    }

  if ((before != NULL) &&
      (after != NULL) &&
      ((htons (before->frag->off) +
        FRAGSIZE (before)) >= htons (after->frag->off)))
    {
      /* this implies that the fragment that starts before us and the
         fragment that comes after this one leave no space in the middle
         or even overlap; thus we can drop this redundant piece */
      if (stats != NULL)
        stats->change (stat_defragmented, 1);
      return GNUNET_OK;
    }

  /* allocate pep */
  pep = GNUNET_malloc (sizeof (FC));
  pep->frag = GNUNET_malloc (ntohs (packet->header.size));
  memcpy (pep->frag, packet, ntohs (packet->header.size));
  pep->link = NULL;

  if (before == NULL)
    {
      pep->link = after;
      pos = entry->head;
      while (pos != after)
        {
          tmp = pos->link;
          GNUNET_free (pos->frag);
          GNUNET_free (pos);
          pos = tmp;
        }
      entry->head = pep;
      goto FINISH;
      /* end of insert first */
    }

  if (after == NULL)
    {
      /* insert last: find the end, free everything after it */
      freeFL (before->link, 1);
      before->link = pep;
      goto FINISH;
    }

  /* ok, we are filling the middle between two fragments; insert.  If
     there is anything else in the middle, it can be dropped as we're
     bigger & cover that area as well */
  /* free everything between before and after */
  pos = before->link;
  while (pos != after)
    {
      tmp = pos->link;
      GNUNET_free (pos->frag);
      GNUNET_free (pos);
      pos = tmp;
    }
  before->link = pep;
  pep->link = after;

FINISH:
  entry->ttl = GNUNET_get_time () + DEFRAGMENTATION_TIMEOUT;
  checkComplete (entry);
  return GNUNET_OK;
}

/**
 * Defragment the given fragment and pass to handler once
 * defragmentation is complete.
 *
 * @param frag the packet to defragment
 * @return GNUNET_SYSERR if the fragment is invalid
 */
static int
processFragment (const GNUNET_PeerIdentity * sender,
                 const GNUNET_MessageHeader * frag)
{
  unsigned int hash;
  FC *smf;

  if (ntohs (frag->size) < sizeof (P2P_fragmentation_MESSAGE))
    return GNUNET_SYSERR;

  GNUNET_mutex_lock (defragCacheLock);
  hash = sender->hashPubKey.bits[0] % DEFRAG_BUCKET_COUNT;
  smf = defragmentationCache[hash];
  while (smf != NULL)
    {
      if (GNUNET_OK ==
          tryJoin (smf, sender, (P2P_fragmentation_MESSAGE *) frag))
        {
          GNUNET_mutex_unlock (defragCacheLock);
          return GNUNET_OK;
        }
      if (0 == memcmp (sender, &smf->sender, sizeof (GNUNET_PeerIdentity)))
        {
          freeFL (smf->head, 1);
          break;
        }
      smf = smf->next;
    }
  if (smf == NULL)
    {
      smf = GNUNET_malloc (sizeof (FC));
      smf->next = defragmentationCache[hash];
      defragmentationCache[hash] = smf;
      smf->ttl = GNUNET_get_time () + DEFRAGMENTATION_TIMEOUT;
      smf->sender = *sender;
    }
  smf->id = ntohl (((P2P_fragmentation_MESSAGE *) frag)->id);
  smf->head = GNUNET_malloc (sizeof (FL));
  smf->head->link = NULL;
  smf->head->frag = GNUNET_malloc (ntohs (frag->size));
  memcpy (smf->head->frag, frag, ntohs (frag->size));

  GNUNET_mutex_unlock (defragCacheLock);
  return GNUNET_OK;
}

typedef struct
{
  GNUNET_PeerIdentity sender;
  /* maximums size of each fragment */
  unsigned short mtu;
  /** how long is this message part expected to be? */
  unsigned short len;
  /** when did we intend to transmit? */
  GNUNET_CronTime transmissionTime;
} FragmentBMC;

/**
 * Send a message that had to be fragmented (right now!).  First grabs
 * the first part of the message (obtained from ctx->se) and stores
 * that in a P2P_fragmentation_MESSAGE envelope.  The remaining fragments are
 * added to the send queue with GNUNET_EXTREME_PRIORITY (to ensure that they
 * will be transmitted next).  The logic here is that if the priority
 * for the first fragment was sufficiently high, the priority should
 * also have been sufficiently high for all of the other fragments (at
 * this time) since they have the same priority.  And we want to make
 * sure that we send all of them since just sending the first fragment
 * and then going to other messages of equal priority would not be
 * such a great idea (i.e. would just waste bandwidth).
 */
static int
fragmentBMC (void *buf, void *cls, unsigned short len)
{
  FragmentBMC *ctx = cls;
  static int idGen = 0;
  P2P_fragmentation_MESSAGE *frag;
  unsigned int pos;
  int id;
  unsigned short mlen;

  if ((len < ctx->mtu) || (buf == NULL))
    {
      GNUNET_free (ctx);
      return GNUNET_SYSERR;
    }
  if (stats != NULL)
    stats->change (stat_fragmented, 1);
  id = (idGen++) + GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK, 512);
  /* write first fragment to buf */
  frag = (P2P_fragmentation_MESSAGE *) buf;
  frag->header.size = htons (len);
  frag->header.type = htons (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
  frag->id = id;
  frag->off = htons (0);
  frag->len = htons (ctx->len);
  memcpy (&frag[1], &ctx[1], len - sizeof (P2P_fragmentation_MESSAGE));

  /* create remaining fragments, add to queue! */
  pos = len - sizeof (P2P_fragmentation_MESSAGE);
  frag = GNUNET_malloc (ctx->mtu);
  while (pos < ctx->len)
    {
      mlen = sizeof (P2P_fragmentation_MESSAGE) + ctx->len - pos;
      if (mlen > ctx->mtu)
        mlen = ctx->mtu;
      GNUNET_GE_ASSERT (NULL, mlen > sizeof (P2P_fragmentation_MESSAGE));
      frag->header.size = htons (mlen);
      frag->header.type = htons (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
      frag->id = id;
      frag->off = htons (pos);
      frag->len = htons (ctx->len);
      memcpy (&frag[1],
              &((char *) (&ctx[1]))[pos],
              mlen - sizeof (P2P_fragmentation_MESSAGE));
      coreAPI->ciphertext_send (&ctx->sender,
                                &frag->header,
                                GNUNET_EXTREME_PRIORITY,
                                ctx->transmissionTime - GNUNET_get_time ());
      pos += mlen - sizeof (P2P_fragmentation_MESSAGE);
    }
  GNUNET_GE_ASSERT (NULL, pos == ctx->len);
  GNUNET_free (frag);
  GNUNET_free (ctx);
  return GNUNET_OK;
}

/**
 * The given message must be fragmented.  Produce a placeholder that
 * corresponds to the first fragment.  Once that fragment is scheduled
 * for transmission, the placeholder should automatically add all of
 * the other fragments (with very high priority).
 */
void
fragment (const GNUNET_PeerIdentity * peer,
          unsigned int mtu,
          unsigned int prio,
          unsigned int targetTime,
          unsigned int len, GNUNET_BuildMessageCallback bmc, void *bmcClosure)
{
  FragmentBMC *fbmc;
  int xlen;

  GNUNET_GE_ASSERT (NULL, len > mtu);
  GNUNET_GE_ASSERT (NULL, mtu > sizeof (P2P_fragmentation_MESSAGE));
  fbmc = GNUNET_malloc (sizeof (FragmentBMC) + len);
  fbmc->mtu = mtu;
  fbmc->sender = *peer;
  fbmc->transmissionTime = targetTime;
  fbmc->len = len;
  if (bmc == NULL)
    {
      memcpy (&fbmc[1], bmcClosure, len);
      GNUNET_free (bmcClosure);
    }
  else
    {
      if (GNUNET_SYSERR == bmc (&fbmc[1], bmcClosure, len))
        {
          GNUNET_free (fbmc);
          return;
        }
    }
  xlen = mtu - sizeof (P2P_fragmentation_MESSAGE);
  coreAPI->ciphertext_send_with_callback (peer, &fragmentBMC, fbmc, mtu, prio * xlen / len,     /* compute new priority */
                                          targetTime);
}

/**
 * Initialize Fragmentation module.
 */
GNUNET_Fragmentation_ServiceAPI *
provide_module_fragmentation (GNUNET_CoreAPIForPlugins * capi)
{
  static GNUNET_Fragmentation_ServiceAPI ret;
  int i;

  coreAPI = capi;
  stats = coreAPI->service_request ("stats");
  if (stats != NULL)
    {
      stat_defragmented =
        stats->create (gettext_noop ("# messages defragmented"));
      stat_fragmented =
        stats->create (gettext_noop ("# messages fragmented"));
      stat_discarded = stats->create (gettext_noop ("# fragments discarded"));
    }
  for (i = 0; i < DEFRAG_BUCKET_COUNT; i++)
    defragmentationCache[i] = NULL;
  defragCacheLock = GNUNET_mutex_create (GNUNET_NO);
  GNUNET_cron_add_job (coreAPI->cron,
                       &defragmentationPurgeCron,
                       60 * GNUNET_CRON_SECONDS, 60 * GNUNET_CRON_SECONDS,
                       NULL);
  GNUNET_GE_LOG (capi->ectx,
                 GNUNET_GE_INFO | GNUNET_GE_USER | GNUNET_GE_REQUEST,
                 _("`%s' registering handler %d\n"), "fragmentation",
                 GNUNET_P2P_PROTO_MESSAGE_FRAGMENT);
  capi->p2p_ciphertext_handler_register (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT,
                                         &processFragment);

  ret.fragment = &fragment;
  return &ret;
}

/**
 * Shutdown fragmentation.
 */
void
release_module_fragmentation ()
{
  int i;

  coreAPI->p2p_ciphertext_handler_unregister
    (GNUNET_P2P_PROTO_MESSAGE_FRAGMENT, &processFragment);
  GNUNET_cron_del_job (coreAPI->cron, &defragmentationPurgeCron,
                       60 * GNUNET_CRON_SECONDS, NULL);
  for (i = 0; i < DEFRAG_BUCKET_COUNT; i++)
    {
      FC *pos = defragmentationCache[i];
      while (pos != NULL)
        {
          FC *next = pos->next;
          freeFL (pos->head, 1);
          GNUNET_free (pos);
          pos = next;
        }
    }
  if (stats != NULL)
    {
      coreAPI->service_release (stats);
      stats = NULL;
    }
  GNUNET_mutex_destroy (defragCacheLock);
  defragCacheLock = NULL;
  coreAPI = NULL;
}

/* end of fragmentation.c */