traffic.c

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

static int
trafficQueryHandler (struct GNUNET_ClientHandle *sock,
                     const GNUNET_MessageHeader * message)
{
  const CS_traffic_request_MESSAGE *msg;
  CS_traffic_info_MESSAGE *reply;
  int ret;

  if (sizeof (CS_traffic_request_MESSAGE) != ntohs (message->size))
    return GNUNET_SYSERR;
  msg = (const CS_traffic_request_MESSAGE *) message;
  reply = buildReply (ntohl (msg->timePeriod));
  ret = coreAPI->cs_send_message (sock, &reply->header, GNUNET_YES);
  GNUNET_free (reply);
  return ret;
}


/**
 * Get statistics over the number of messages that
 * were received or send of a given type.
 *
 * @param messageType the type of the message
 * @param sendReceive GNUNET_TRAFFIC_TYPE_SENT for sending, GNUNET_TRAFFIC_TYPE_RECEIVED for receiving
 * @param timePeriod how many TRAFFIC_TIME_UNITs to take
 *        into consideration (limited by HISTORY_SIZE)
 * @param avgMessageSize average size of the messages (set)
 * @param messageCount number of messages (set)
 * @param peerCount number of peers engaged (set)
 * @param timeDistribution bit-vector giving times of interactions,
 *        highest bit is current time-unit, bit 1 is 32 time-units ago (set)
 * @return GNUNET_OK on success, GNUNET_SYSERR on error
 */
static int
getTrafficStats (unsigned int timePeriod,
                 unsigned short messageType,
                 unsigned short sendReceive,
                 unsigned int *messageCount,
                 unsigned int *peerCount,
                 unsigned int *avgMessageSize, unsigned int *timeDistribution)
{
  DirectedTrafficCounter *dtc;
  unsigned int i;
  unsigned int nowUnit;
  double totSize;

  if (timePeriod > HISTORY_SIZE)
    timePeriod = HISTORY_SIZE;
  GNUNET_mutex_lock (lock);
  if ((messageType >= max_message_type) || (counters[messageType] == NULL))
    {
      *avgMessageSize = 0;
      *messageCount = 0;
      *peerCount = 0;
      *timeDistribution = 0;
      GNUNET_mutex_unlock (lock);
      return GNUNET_OK;
    }

  if (sendReceive == GNUNET_TRAFFIC_TYPE_SENT)
    dtc = &counters[messageType]->send;
  else
    dtc = &counters[messageType]->receive;
  updateUse (dtc, 0, 0, GNUNET_YES);

  nowUnit = GNUNET_get_time () / GNUNET_TRAFFIC_TIME_UNIT;
  *peerCount = 0;
  *messageCount = 0;
  totSize = 0;
  for (i = 0; i < MAX_PEER_IDs; i++)
    if (dtc->peers[i].time > nowUnit - timePeriod)
      (*peerCount)++;
  for (i = 0; i < timePeriod; i++)
    {
      unsigned int slot;

      slot = HS_SLOT (nowUnit - i);
      (*messageCount) += dtc->count[slot];
      totSize += dtc->count[slot] * dtc->avgSize[slot];
    }
  if (*messageCount > 0)
    *avgMessageSize = (unsigned short) (totSize / (*messageCount));
  else
    *avgMessageSize = 0;
  *timeDistribution = dtc->slots;
  GNUNET_mutex_unlock (lock);
  return GNUNET_OK;
}


/**
 * Ensure that the counters array has the appropriate
 * size and a valid traffic counter allocated for the
 * given port.
 */
static void
checkPort (unsigned short port)
{
  if (port >= max_message_type)
    GNUNET_array_grow (counters, max_message_type, port + 1);
  if (counters[port] == NULL)
    {
      counters[port] = GNUNET_malloc (sizeof (TrafficCounter));
      memset (counters[port], 0, sizeof (TrafficCounter));
    }
}

static void
updateTrafficSendCounter (unsigned short ptyp, unsigned short plen)
{
  if (ptyp >= GNUNET_P2P_PROTO_MAX_USED)
    return;                     /* not tracked */
  if (0 == stat_traffic_transmitted_by_type[ptyp])
    {
      char *s;
      s = GNUNET_malloc (256);
      GNUNET_snprintf (s, 256, _("# bytes transmitted of type %d"), ptyp);
      stat_traffic_transmitted_by_type[ptyp] = stats->create (s);
      GNUNET_free (s);
    }
  stats->change (stat_traffic_transmitted_by_type[ptyp], plen);
}

static void
updateTrafficReceiveCounter (unsigned short ptyp, unsigned short plen)
{
  if (ptyp < GNUNET_P2P_PROTO_MAX_USED)
    {
      if (0 == stat_traffic_received_by_type[ptyp])
        {
          char *s;
          s = GNUNET_malloc (256);
          GNUNET_snprintf (s, 256, _("# bytes received of type %d"), ptyp);
          stat_traffic_received_by_type[ptyp] = stats->create (s);
          GNUNET_free (s);
        }
      stats->change (stat_traffic_received_by_type[ptyp], plen);
    }
}

static void
updatePlaintextTrafficReceiveCounter (unsigned short ptyp,
                                      unsigned short plen)
{
  if (ptyp < GNUNET_P2P_PROTO_MAX_USED)
    {
      if (0 == stat_pt_traffic_received_by_type[ptyp])
        {
          char *s;
          s = GNUNET_malloc (256);
          GNUNET_snprintf (s,
                           256, _("# bytes received in plaintext of type %d"),
                           ptyp);
          stat_pt_traffic_received_by_type[ptyp] = stats->create (s);
          GNUNET_free (s);
        }
      stats->change (stat_pt_traffic_received_by_type[ptyp], plen);
    }
}

/**
 * A message was received.  Update traffic stats.
 *
 * @param header the header of the message
 * @param sender the identity of the sender
 */
static int
trafficReceive (const GNUNET_PeerIdentity * sender,
                const GNUNET_MessageHeader * header)
{
  unsigned short port;

  if (sender == NULL)
    return GNUNET_OK;
  port = ntohs (header->type);
  updateTrafficReceiveCounter (port, ntohs (header->size));
  GNUNET_mutex_lock (lock);
  checkPort (port);
  updateUse (&counters[port]->receive,
             ntohs (header->size), sender->hashPubKey.bits[0], GNUNET_NO);
  GNUNET_mutex_unlock (lock);
  return GNUNET_OK;
}


/**
 * A message is send.  Update traffic stats.
 *
 * @param header the header of the message
 * @param receiver the identity of the receiver
 */
static int
trafficSend (const GNUNET_PeerIdentity * receiver,
             const GNUNET_MessageHeader * header)
{
  unsigned short port;

  port = ntohs (MAKE_UNALIGNED (header->type));
  updateTrafficSendCounter (port, ntohs (MAKE_UNALIGNED (header->size)));
  GNUNET_mutex_lock (lock);
  checkPort (port);
  updateUse (&counters[port]->send,
             ntohs (MAKE_UNALIGNED (header->size)),
             receiver->hashPubKey.bits[0], GNUNET_NO);
  GNUNET_mutex_unlock (lock);
  return GNUNET_OK;
}

/**
 * A message is send.  Update traffic stats.
 *
 * @param header the header of the message
 * @param receiver the identity of the receiver
 */
static int
plaintextReceive (const GNUNET_PeerIdentity * receiver,
                  const GNUNET_MessageHeader * header,
                  GNUNET_TSession * session)
{
  unsigned short port;

  port = ntohs (MAKE_UNALIGNED (header->type));
  updatePlaintextTrafficReceiveCounter (port,
                                        ntohs (MAKE_UNALIGNED
                                               (header->size)));
  return GNUNET_OK;
}


/**
 * Initialize the traffic module.
 */
GNUNET_Traffic_ServiceAPI *
provide_module_traffic (GNUNET_CoreAPIForPlugins * capi)
{
  static GNUNET_Traffic_ServiceAPI api;
  int i;

  coreAPI = capi;
#if DEBUG
  GNUNET_GC_get_configuration_value_number (capi->cfg,
                                            "NETWORK",
                                            "PORT", 0, 65536, 2087,
                                            &server_port);
#endif
  api.get = &getTrafficStats;
  for (i = 0; i < GNUNET_P2P_PROTO_MAX_USED; i++)
    stat_traffic_transmitted_by_type[i] = 0;
  coreAPI->peer_send_notification_register (&trafficSend);
  for (i = 0; i < GNUNET_P2P_PROTO_MAX_USED; i++)
    {
      stat_traffic_received_by_type[i] = 0;
      coreAPI->p2p_ciphertext_handler_register (i, &trafficReceive);
      coreAPI->p2p_plaintext_handler_register (i, &plaintextReceive);
    }

  GNUNET_GE_ASSERT (coreAPI->ectx, counters == NULL);
  lock = GNUNET_mutex_create (GNUNET_NO);
  stats = capi->service_request ("stats");
  return &api;
}

/**
 * Shutdown the traffic module.
 */
void
release_module_traffic ()
{
  unsigned int i;

  for (i = 0; i < GNUNET_P2P_PROTO_MAX_USED; i++)
    {
      coreAPI->p2p_ciphertext_handler_unregister (i, &trafficReceive);
      coreAPI->p2p_plaintext_handler_unregister (i, &plaintextReceive);
    }
  coreAPI->peer_send_notification_unregister (&trafficSend);
  coreAPI->service_release (stats);
  stats = NULL;
  for (i = 0; i < max_message_type; i++)
    GNUNET_free_non_null (counters[i]);
  GNUNET_array_grow (counters, max_message_type, 0);
  GNUNET_mutex_destroy (lock);
  lock = NULL;
  coreAPI = NULL;
}



static GNUNET_Traffic_ServiceAPI *myApi;
static GNUNET_CoreAPIForPlugins *myCoreAPI;

/**
 * Initialize the traffic module.
 */
int
initialize_module_traffic (GNUNET_CoreAPIForPlugins * capi)
{
  GNUNET_GE_ASSERT (capi->ectx, myCoreAPI == NULL);
  myCoreAPI = capi;
  myApi = capi->service_request ("traffic");
  if (myApi == NULL)
    {
      GNUNET_GE_BREAK (capi->ectx, 0);
      myCoreAPI = NULL;
      return GNUNET_SYSERR;
    }
  capi->cs_handler_register (GNUNET_CS_PROTO_TRAFFIC_QUERY,
                             &trafficQueryHandler);
  GNUNET_GE_ASSERT (capi->ectx,
                    0 == GNUNET_GC_set_configuration_value_string (capi->cfg,
                                                                   capi->ectx,
                                                                   "ABOUT",
                                                                   "traffic",
                                                                   gettext_noop
                                                                   ("tracks bandwidth utilization by gnunetd")));
  return GNUNET_OK;
}

/**
 * Shutdown the traffic module.
 */
void
done_module_traffic ()
{
  GNUNET_GE_ASSERT (NULL, myCoreAPI != NULL);
  GNUNET_GE_ASSERT (myCoreAPI->ectx,
                    GNUNET_SYSERR !=
                    myCoreAPI->cs_handler_unregister
                    (GNUNET_CS_PROTO_TRAFFIC_QUERY, &trafficQueryHandler));
  myCoreAPI->service_release (myApi);
  myApi = NULL;
  myCoreAPI = NULL;
}


/* end of traffic.c */