/* Queries the key-tree of the parent context for "in_name" and, if found,
 * updates the "key" field of the current context. Otherwise, "key" is set to
 * NULL. */
static int cj_cb_map_key (void *ctx,
    unsigned char const *in_name, yajl_len_t in_name_len)
{
  cj_t *db = (cj_t *)ctx;
  c_avl_tree_t *tree;

  tree = db->state[db->depth-1].tree;

  if (tree != NULL)
  {
    cj_key_t *value = NULL;
    char *name;
    size_t name_len;

    /* Create a null-terminated version of the name. */
    name = db->state[db->depth].name;
    name_len = COUCH_MIN ((size_t) in_name_len,
        sizeof (db->state[db->depth].name) - 1);
    memcpy (name, in_name, name_len);
    name[name_len] = 0;

    if (c_avl_get (tree, name, (void *) &value) == 0)
      db->state[db->depth].key = value;
    else if (c_avl_get (tree, CJ_ANY, (void *) &value) == 0)
      db->state[db->depth].key = value;
    else
      db->state[db->depth].key = NULL;
  }

  return (CJ_CB_CONTINUE);
}

static int cj_cb_map_key (void *ctx, const unsigned char *val,
                            unsigned int len)
{
  cj_t *db = (cj_t *)ctx;
  c_avl_tree_t *tree;

  tree = db->state[db->depth-1].tree;

  if (tree != NULL)
  {
    cj_key_t *value;
    char *name;

    name = db->state[db->depth].name;
    len = COUCH_MIN(len, sizeof (db->state[db->depth].name)-1);
    sstrncpy (name, (char *)val, len+1);

    if (c_avl_get (tree, name, (void *) &value) == 0)
      db->state[db->depth].key = value;
    else if (c_avl_get (tree, CJ_ANY, (void *) &value) == 0)
      db->state[db->depth].key = value;
    else
      db->state[db->depth].key = NULL;
  }

  return (CJ_CB_CONTINUE);
}

int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step)
{
  char name[6 * DATA_MAX_NAME_LEN];
  cache_entry_t *ce = NULL;
  int ret = -1;

  if (FORMAT_VL (name, sizeof (name), vl) != 0)
  {
    ERROR ("uc_get_state: FORMAT_VL failed.");
    return (STATE_ERROR);
  }

  pthread_mutex_lock (&cache_lock);

  if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
  {
    assert (ce != NULL);
    ret = ce->hits;
    ce->hits = ret + step;
  }

  pthread_mutex_unlock (&cache_lock);

  return (ret);
} /* int uc_inc_hits */

static void ethstat_submit_value(const char *device, const char *type_instance,
                                 derive_t value) {
  static c_complain_t complain_no_map = C_COMPLAIN_INIT_STATIC;

  value_list_t vl = VALUE_LIST_INIT;
  value_map_t *map = NULL;

  if (value_map != NULL)
    c_avl_get(value_map, type_instance, (void *)&map);

  /* If the "MappedOnly" option is specified, ignore unmapped values. */
  if (collect_mapped_only && (map == NULL)) {
    if (value_map == NULL)
      c_complain(
          LOG_WARNING, &complain_no_map,
          "ethstat plugin: The \"MappedOnly\" option has been set to true, "
          "but no mapping has been configured. All values will be ignored!");
    return;
  }

  vl.values = &(value_t){.derive = value};
  vl.values_len = 1;

  sstrncpy(vl.plugin, "ethstat", sizeof(vl.plugin));
  sstrncpy(vl.plugin_instance, device, sizeof(vl.plugin_instance));
  if (map != NULL) {
    sstrncpy(vl.type, map->type, sizeof(vl.type));
    sstrncpy(vl.type_instance, map->type_instance, sizeof(vl.type_instance));
  } else {
    sstrncpy(vl.type, "derive", sizeof(vl.type));
    sstrncpy(vl.type_instance, type_instance, sizeof(vl.type_instance));
  }

  plugin_dispatch_values(&vl);
}

/* XXX: This function will acquire `cache_lock' but will not free it! */
static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */
{
  char name[6 * DATA_MAX_NAME_LEN];
  cache_entry_t *ce = NULL;
  int status;

  status = FORMAT_VL (name, sizeof (name), vl);
  if (status != 0)
  {
    ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed.");
    return (NULL);
  }

  pthread_mutex_lock (&cache_lock);

  status = c_avl_get (cache_tree, name, (void *) &ce);
  if (status != 0)
  {
    pthread_mutex_unlock (&cache_lock);
    return (NULL);
  }
  assert (ce != NULL);

  if (ce->meta == NULL)
    ce->meta = meta_data_create ();

  if (ce->meta == NULL)
    pthread_mutex_unlock (&cache_lock);

  return (ce->meta);
} /* }}} meta_data_t *uc_get_meta */

char *fbh_get (fbhash_t *h, const char *key) /* {{{ */
{
  char *value;
  char *value_copy;
  int status;

  if ((h == NULL) || (key == NULL))
    return (NULL);

  value = NULL;
  value_copy = NULL;

  pthread_mutex_lock (&h->lock);

  /* TODO: Checking this every time may be a bit much..? */
  fbh_check_file (h);

  status = c_avl_get (h->tree, key, (void *) &value);
  if (status == 0)
  {
    assert (value != NULL);
    value_copy = strdup (value);
  }

  pthread_mutex_unlock (&h->lock);

  return (value_copy);
} /* }}} char *fbh_get */

/* returns the number of successful calls to the callback function */
int lookup_search(lookup_t *obj, /* {{{ */
                  data_set_t const *ds, value_list_t const *vl) {
  by_type_entry_t *by_type = NULL;
  user_class_list_t *user_class_list = NULL;
  int retval = 0;
  int status;

  if ((obj == NULL) || (ds == NULL) || (vl == NULL))
    return (-EINVAL);

  by_type = lu_search_by_type(obj, vl->type, /* allocate = */ 0);
  if (by_type == NULL)
    return (0);

  status =
      c_avl_get(by_type->by_plugin_tree, vl->plugin, (void *)&user_class_list);
  if (status == 0) {
    status = lu_handle_user_class_list(obj, ds, vl, user_class_list);
    if (status < 0)
      return (status);
    retval += status;
  }

  if (by_type->wildcard_plugin_list != NULL) {
    status =
        lu_handle_user_class_list(obj, ds, vl, by_type->wildcard_plugin_list);
    if (status < 0)
      return (status);
    retval += status;
  }

  return (retval);
} /* }}} lookup_search */

/* Must hold metrics_lock when calling this function. */
static statsd_metric_t *statsd_metric_lookup_unsafe(char const *name, /* {{{ */
                                                    metric_type_t type) {
  char key[DATA_MAX_NAME_LEN + 2];
  char *key_copy;
  statsd_metric_t *metric;
  int status;

  switch (type) {
  case STATSD_COUNTER:
    key[0] = 'c';
    break;
  case STATSD_TIMER:
    key[0] = 't';
    break;
  case STATSD_GAUGE:
    key[0] = 'g';
    break;
  case STATSD_SET:
    key[0] = 's';
    break;
  default:
    return NULL;
  }

  key[1] = ':';
  sstrncpy(&key[2], name, sizeof(key) - 2);

  status = c_avl_get(metrics_tree, key, (void *)&metric);
  if (status == 0)
    return metric;

  key_copy = strdup(key);
  if (key_copy == NULL) {
    ERROR("statsd plugin: strdup failed.");
    return NULL;
  }

  metric = calloc(1, sizeof(*metric));
  if (metric == NULL) {
    ERROR("statsd plugin: calloc failed.");
    sfree(key_copy);
    return NULL;
  }

  metric->type = type;
  metric->latency = NULL;
  metric->set = NULL;

  status = c_avl_insert(metrics_tree, key_copy, metric);
  if (status != 0) {
    ERROR("statsd plugin: c_avl_insert failed.");
    sfree(key_copy);
    sfree(metric);
    return NULL;
  }

  return metric;
} /* }}} statsd_metric_lookup_unsafe */

static staging_entry_t *staging_entry_get (const char *host, /* {{{ */
    const char *name,
    const char *type, const char *type_instance,
    int values_len)
{
  char key[2 * DATA_MAX_NAME_LEN];
  staging_entry_t *se;
  int status;

  if (staging_tree == NULL)
    return (NULL);

  ssnprintf (key, sizeof (key), "%s/%s/%s", host, type,
      (type_instance != NULL) ? type_instance : "");

  se = NULL;
  status = c_avl_get (staging_tree, key, (void *) &se);
  if (status == 0)
    return (se);

  /* insert new entry */
  se = (staging_entry_t *) malloc (sizeof (*se));
  if (se == NULL)
    return (NULL);
  memset (se, 0, sizeof (*se));

  sstrncpy (se->key, key, sizeof (se->key));
  se->flags = 0;

  se->vl.values = (value_t *) calloc (values_len, sizeof (*se->vl.values));
  if (se->vl.values == NULL)
  {
    sfree (se);
    return (NULL);
  }
  se->vl.values_len = values_len;

  se->vl.time = 0;
  se->vl.interval = 0;
  sstrncpy (se->vl.host, host, sizeof (se->vl.host));
  sstrncpy (se->vl.plugin, "gmond", sizeof (se->vl.plugin));
  sstrncpy (se->vl.type, type, sizeof (se->vl.type));
  if (type_instance != NULL)
    sstrncpy (se->vl.type_instance, type_instance,
        sizeof (se->vl.type_instance));

  status = c_avl_insert (staging_tree, se->key, se);
  if (status != 0)
  {
    ERROR ("gmond plugin: c_avl_insert failed.");
    sfree (se->vl.values);
    sfree (se);
    return (NULL);
  }

  return (se);
} /* }}} staging_entry_t *staging_entry_get */

static int lu_add_by_plugin(by_type_entry_t *by_type, /* {{{ */
                            user_class_list_t *user_class_list) {
  user_class_list_t *ptr = NULL;
  identifier_match_t const *match = &user_class_list->entry.match;

  /* Lookup user_class_list from the per-plugin structure. If this is the first
   * user_class to be added, the block returns immediately. Otherwise they will
   * set "ptr" to non-NULL. */
  if (match->plugin.is_regex) {
    if (by_type->wildcard_plugin_list == NULL) {
      by_type->wildcard_plugin_list = user_class_list;
      return (0);
    }

    ptr = by_type->wildcard_plugin_list;
  }    /* if (plugin is wildcard) */
  else /* (plugin is not wildcard) */
  {
    int status;

    status =
        c_avl_get(by_type->by_plugin_tree, match->plugin.str, (void *)&ptr);

    if (status != 0) /* plugin not yet in tree */
    {
      char *plugin_copy = strdup(match->plugin.str);

      if (plugin_copy == NULL) {
        ERROR("utils_vl_lookup: strdup failed.");
        sfree(user_class_list);
        return (ENOMEM);
      }

      status =
          c_avl_insert(by_type->by_plugin_tree, plugin_copy, user_class_list);
      if (status != 0) {
        ERROR("utils_vl_lookup: c_avl_insert(\"%s\") failed with status %i.",
              plugin_copy, status);
        sfree(plugin_copy);
        sfree(user_class_list);
        return (status);
      } else {
        return (0);
      }
    } /* if (plugin not yet in tree) */
  }   /* if (plugin is not wildcard) */

  assert(ptr != NULL);

  while (ptr->next != NULL)
    ptr = ptr->next;
  ptr->next = user_class_list;

  return (0);
} /* }}} int lu_add_by_plugin */

static by_type_entry_t *lu_search_by_type (lookup_t *obj, /* {{{ */
    char const *type, _Bool allocate_if_missing)
{
  by_type_entry_t *by_type;
  char *type_copy;
  int status;

  status = c_avl_get (obj->by_type_tree, type, (void *) &by_type);
  if (status == 0)
    return (by_type);

  if (!allocate_if_missing)
    return (NULL);

  type_copy = strdup (type);
  if (type_copy == NULL)
  {
    ERROR ("utils_vl_lookup: strdup failed.");
    return (NULL);
  }

  by_type = malloc (sizeof (*by_type));
  if (by_type == NULL)
  {
    ERROR ("utils_vl_lookup: malloc failed.");
    sfree (type_copy);
    return (NULL);
  }
  memset (by_type, 0, sizeof (*by_type));
  by_type->wildcard_plugin_list = NULL;
  
  by_type->by_plugin_tree = c_avl_create ((void *) strcmp);
  if (by_type->by_plugin_tree == NULL)
  {
    ERROR ("utils_vl_lookup: c_avl_create failed.");
    sfree (by_type);
    sfree (type_copy);
    return (NULL);
  }

  status = c_avl_insert (obj->by_type_tree,
      /* key = */ type_copy, /* value = */ by_type);
  assert (status <= 0); /* >0 => entry exists => race condition. */
  if (status != 0)
  {
    ERROR ("utils_vl_lookup: c_avl_insert failed.");
    c_avl_destroy (by_type->by_plugin_tree);
    sfree (by_type);
    sfree (type_copy);
    return (NULL);
  }
  
  return (by_type);
} /* }}} by_type_entry_t *lu_search_by_type */

static int
basic_aggregator_config_aggregator_get_all_instances_of_type(char ***type_instances, char*type) {
	char **names = NULL;
	char *subtype;
	int l;
	int pos = 0;
	instances_list_t *v;
	int i;
	int typelen;
	*type_instances = NULL;

	for(l=0; type[l] && (pos < 2); l++) {
		if(type[l] == '/') pos++;
	}
	if(type[l] == '\0') {
		return(-1);
	}
	subtype = type+l;

	pthread_mutex_lock (&instances_of_types_mutex);
	if(0 != c_avl_get(instances_of_types_tree, subtype,(void*)&v)) {
		pthread_mutex_unlock (&instances_of_types_mutex);
		return(0);
	}
	for(i=0; v->instance[i]; i++); /* Count how many instances the type has */
	if(NULL == (names = malloc((i+1)*sizeof(*type_instances)))) {
		ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
		pthread_mutex_unlock (&instances_of_types_mutex);
		return(-1);
	}
	typelen = strlen(type);
	for(i=0; v->instance[i]; i++) {
		int li;
		li = strlen(v->instance[i]);
		if(NULL == (names[i] = malloc((typelen+2+li)*sizeof(**names)))) {
			int j;
			ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
			for(j=0; j<i; j++) free(names[j]);
			free(names);
			pthread_mutex_unlock (&instances_of_types_mutex);
			return(-1);
		}
		memcpy(names[i], type, typelen);
		names[i][typelen] = '-';
		memcpy(names[i]+typelen+1, v->instance[i],li+1);
	}
	names[i] = NULL;
	pthread_mutex_unlock (&instances_of_types_mutex);
	*type_instances = names;
		
	return(0);
}

int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_values_num)
{
  gauge_t *ret = NULL;
  size_t ret_num = 0;
  cache_entry_t *ce = NULL;
  int status = 0;

  pthread_mutex_lock (&cache_lock);

  if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
  {
    assert (ce != NULL);

    /* remove missing values from getval */
    if (ce->state == STATE_MISSING)
    {
      status = -1;
    }
    else
    {
      ret_num = ce->values_num;
      ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t));
      if (ret == NULL)
      {
        ERROR ("utils_cache: uc_get_rate_by_name: malloc failed.");
        status = -1;
      }
      else
      {
        memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t));
      }
    }
  }
  else
  {
    DEBUG ("utils_cache: uc_get_rate_by_name: No such value: %s", name);
    status = -1;
  }

  pthread_mutex_unlock (&cache_lock);

  if (status == 0)
  {
    *ret_values = ret;
    *ret_values_num = ret_num;
  }

  return (status);
} /* gauge_t *uc_get_rate_by_name */

/*
 * threshold_t *threshold_get
 *
 * Retrieve one specific threshold configuration. For looking up a threshold
 * matching a value_list_t, see "threshold_search" below. Returns NULL if the
 * specified threshold doesn't exist.
 */
threshold_t *threshold_get(const char *hostname, const char *plugin,
                           const char *plugin_instance, const char *type,
                           const char *type_instance) { /* {{{ */
    char name[6 * DATA_MAX_NAME_LEN];
    threshold_t *th = NULL;

    format_name(name, sizeof(name), (hostname == NULL) ? "" : hostname,
                (plugin == NULL) ? "" : plugin, plugin_instance,
                (type == NULL) ? "" : type, type_instance);
    name[sizeof(name) - 1] = '\0';

    if (c_avl_get(threshold_tree, name, (void *)&th) == 0)
        return (th);
    else
        return (NULL);
} /* }}} threshold_t *threshold_get */

static zone_stats_t *zone_find_stats(c_avl_tree_t *tree, zoneid_t zoneid) {
  zone_stats_t *ret = NULL;
  zoneid_t *key = NULL;

  if (c_avl_get(tree, (void **)&zoneid, (void **)&ret)) {
    if (!(ret = malloc(sizeof(*ret)))) {
      WARNING("zone plugin: no memory");
      return (NULL);
    }
    if (!(key = malloc(sizeof(*key)))) {
      WARNING("zone plugin: no memory");
      free(ret);
      return (NULL);
    }
    *key = zoneid;
    if (c_avl_insert(tree, key, ret)) {
      WARNING("zone plugin: error inserting into tree");
      return (NULL);
    }
  }
  return (ret);
}

static int
instances_of_types_tree_update (void) {
	char **names = NULL;
	cdtime_t *times = NULL;
	size_t number = 0;
	int i;
	int status = 0;

	if(NULL == instances_of_types_tree) {
		instances_of_types_tree = c_avl_create ((void *) strcmp);
	}

	status = uc_get_names (&names, &times, &number);
	if (status != 0)
	{
			size_t j; 
			DEBUG (OUTPUT_PREFIX_STRING "uc_get_names failed with status %i", status);
			for (j = 0; j < number; j++) { 
					sfree(names[j]); 
			} 
			sfree(names); 
			sfree(times); 
			return(status);
	}

	if(number == 0) {
		return(0);
	}
	pthread_mutex_lock (&instances_of_types_mutex);
	for(i=0; i<number; i++) {
		char *type;
		int l1,l2;
		int pos =0;
		instances_list_t *v;
		char *type_instance;
		int type_instance_is_missing = 1;

		for(l1=0; names[i][l1] && (pos < 2); l1++) {
			if(names[i][l1] == '/') pos++;
		}
		if(names[i][l1] == '\0') {
			sfree(names[i]);
			continue;
		}
		l2 = l1;
		while(names[i][l2] && (names[i][l2] != '-')) l2++;
		if(names[i][l2] == '\0') {
			sfree(names[i]);
			continue;
		}
		type = names[i]+l1;
		names[i][l2] = '\0';
		type_instance = names[i]+l2+1;


		if(0 == c_avl_get(instances_of_types_tree, type,(void*)&v)) {
			int i;
			for(i=0; v->instance[i]; i++) {
				if(0 == strcmp(v->instance[i], type_instance)) {
					type_instance_is_missing = 0;
					break;
				}
			}
			if(type_instance_is_missing) {
				if(i >= v->nb) {
					v->nb += 128;
					if(NULL == (v->instance = realloc(v->instance, v->nb*sizeof(*(v->instance))))) {
						ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
						pthread_mutex_unlock (&instances_of_types_mutex);
						return(-1);
					}
					}
				v->instance[i+1] = NULL;
				if(NULL == (v->instance[i] = sstrdup(type_instance))) {
					ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
					pthread_mutex_unlock (&instances_of_types_mutex);
					return(-1);
				}
			}
		} else {
			char *k;
			if(NULL == (v = malloc(sizeof(*v)))) {
				ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
				pthread_mutex_unlock (&instances_of_types_mutex);
				return(-1);
			}
			v->nb = 128;
			if(NULL == (v->instance = malloc(v->nb*sizeof(*(v->instance))))) {
				ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
				sfree(v);
				pthread_mutex_unlock (&instances_of_types_mutex);
				return(-1);
			}
			if(NULL == (k = sstrdup(type))) {
				ERROR(OUTPUT_PREFIX_STRING "Could not allocate memory");
				sfree(v);
				pthread_mutex_unlock (&instances_of_types_mutex);
				return(-1);
			}
			if(NULL == (v->instance[0] = sstrdup(type_instance))) {
//.........这里部分代码省略.........

int uc_update (const data_set_t *ds, const value_list_t *vl)
{
    char name[6 * DATA_MAX_NAME_LEN];
    cache_entry_t *ce = NULL;
    int status;
    size_t i;

    if (FORMAT_VL (name, sizeof (name), vl) != 0)
    {
        ERROR ("uc_update: FORMAT_VL failed.");
        return (-1);
    }

    pthread_mutex_lock (&cache_lock);

    status = c_avl_get (cache_tree, name, (void *) &ce);
    if (status != 0) /* entry does not yet exist */
    {
        status = uc_insert (ds, vl, name);
        pthread_mutex_unlock (&cache_lock);
        return (status);
    }

    assert (ce != NULL);
    assert (ce->values_num == ds->ds_num);

    if (ce->last_time >= vl->time)
    {
        pthread_mutex_unlock (&cache_lock);
        NOTICE ("uc_update: Value too old: name = %s; value time = %.3f; "
                "last cache update = %.3f;",
                name,
                CDTIME_T_TO_DOUBLE (vl->time),
                CDTIME_T_TO_DOUBLE (ce->last_time));
        return (-1);
    }

    for (i = 0; i < ds->ds_num; i++)
    {
        switch (ds->ds[i].type)
        {
        case DS_TYPE_COUNTER:
        {
            counter_t diff = counter_diff (ce->values_raw[i].counter, vl->values[i].counter);
            ce->values_gauge[i] = ((double) diff)
                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));
            ce->values_raw[i].counter = vl->values[i].counter;
        }
        break;

        case DS_TYPE_GAUGE:
            ce->values_raw[i].gauge = vl->values[i].gauge;
            ce->values_gauge[i] = vl->values[i].gauge;
            break;

        case DS_TYPE_DERIVE:
        {
            derive_t diff = vl->values[i].derive - ce->values_raw[i].derive;

            ce->values_gauge[i] = ((double) diff)
                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));
            ce->values_raw[i].derive = vl->values[i].derive;
        }
        break;

        case DS_TYPE_ABSOLUTE:
            ce->values_gauge[i] = ((double) vl->values[i].absolute)
                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));
            ce->values_raw[i].absolute = vl->values[i].absolute;
            break;

        default:
            /* This shouldn't happen. */
            pthread_mutex_unlock (&cache_lock);
            ERROR ("uc_update: Don't know how to handle data source type %i.",
                   ds->ds[i].type);
            return (-1);
        } /* switch (ds->ds[i].type) */

        DEBUG ("uc_update: %s: ds[%zu] = %lf", name, i, ce->values_gauge[i]);
    } /* for (i) */

    /* Update the history if it exists. */
    if (ce->history != NULL)
    {
        assert (ce->history_index < ce->history_length);
        for (i = 0; i < ce->values_num; i++)
        {
            size_t hist_idx = (ce->values_num * ce->history_index) + i;
            ce->history[hist_idx] = ce->values_gauge[i];
        }

        assert (ce->history_length > 0);
        ce->history_index = (ce->history_index + 1) % ce->history_length;
    }

    /* Prune invalid gauge data */
    uc_check_range (ds, ce);

    ce->last_time = vl->time;
//.........这里部分代码省略.........

static int basic_aggregator_submit_resultvalue (
	const aggregator_definition_t *agg,
	c_avl_tree_t *ds_data
	)
{
	value_t *values;
	value_list_t vl = VALUE_LIST_INIT;
	const data_set_t *ds;
	char *identifier;
	char *hostname;
	char *plugin;
	char *plugin_instance;
	char *type;
	char *type_instance;
	int  status;
	int i;
	int submit_ok;
	aggregator_operation_e operation;
	c_avl_iterator_t *iter;
	char *key_instance;
	c_avl_tree_t *ds_tree;

	identifier = sstrdup (agg->resultvalue);

	status = parse_identifier (identifier, &hostname,
					&plugin, &plugin_instance,
					&type, &type_instance);
	if (status != 0)
	{
		ERROR (OUTPUT_PREFIX_STRING "Cannot parse value `%s'.", agg->resultvalue);
		sfree (identifier);
		return (-1);
	}

	if((NULL == type) || (type[0] == '\0')) {
		ERROR (OUTPUT_PREFIX_STRING "parse_identifier() : type == NULL or is empty (identifier='%s')", identifier);
		sfree (identifier);
		return (-1);
	}

	ds = plugin_get_ds (type);
	if (ds == NULL)
	{
		ERROR (OUTPUT_PREFIX_STRING "plugin_get_ds (%s) == NULL", type);
		sfree (identifier);
		return (-1);
	}
	
	if(NULL == (values = malloc(ds->ds_num* sizeof(*values)))) {
		ERROR (OUTPUT_PREFIX_STRING "Could not allocate memory");
		sfree (identifier);
		return (-1);
	}
	
	iter = c_avl_get_iterator (ds_data);
	while (c_avl_iterator_next (iter, (void *) &key_instance, (void *) &ds_tree) == 0) {
	
		for (operation=0; operation < nb_enum_in_aggregator_operation_e; operation++) {
			if(0 == agg->operation[operation]) continue;
			memset(values, '\0', ds->ds_num*sizeof(*values));
			submit_ok = 1;
			for(i=0; i<ds->ds_num; i++) {
				value_and_nb_t *v;
				if(0 == c_avl_get(ds_tree, ds->ds[i].name,(void*)&v)) {
					switch(operation) {
						case operation_SUM : values[i].gauge = v->val; break;
						case operation_AVG : values[i].gauge = v->val/v->nb; break;
						default : assert (11 == 12); /* this should never happen */
					}
				} else {
						submit_ok = 0;
				}
	
			}
			if(submit_ok) {
				int l = 0;
				int pos = 0;
	
				vl.values = values;
				vl.values_len = ds->ds_num;
				
				if(plugin_instance) {
					l = strlen(plugin_instance);
				}
				if(l) {
					sstrncpy (vl.plugin_instance, plugin_instance, l+1);
					pos += l;
					sstrncpy (vl.plugin_instance + pos, "_", DATA_MAX_NAME_LEN - pos);
					pos += 1;
				}
				switch(operation) {
						case operation_SUM : sstrncpy (vl.plugin_instance + pos, "sum", DATA_MAX_NAME_LEN - pos); break;
						case operation_AVG : sstrncpy (vl.plugin_instance + pos, "avg", DATA_MAX_NAME_LEN - pos); break;
						default : assert (11 == 12); /* this should never happen */
				}
	
				sstrncpy (vl.host, hostname, sizeof (vl.host));
				sstrncpy (vl.plugin, plugin, sizeof (vl.plugin));
				sstrncpy (vl.type, type, sizeof (vl.type));
				if(key_instance[0]) sstrncpy (vl.type_instance, key_instance, sizeof (vl.type_instance));
//.........这里部分代码省略.........

int uc_get_history_by_name (const char *name,
    gauge_t *ret_history, size_t num_steps, size_t num_ds)
{
  cache_entry_t *ce = NULL;
  size_t i;
  int status = 0;

  pthread_mutex_lock (&cache_lock);

  status = c_avl_get (cache_tree, name, (void *) &ce);
  if (status != 0)
  {
    pthread_mutex_unlock (&cache_lock);
    return (-ENOENT);
  }

  if (((size_t) ce->values_num) != num_ds)
  {
    pthread_mutex_unlock (&cache_lock);
    return (-EINVAL);
  }

  /* Check if there are enough values available. If not, increase the buffer
   * size. */
  if (ce->history_length < num_steps)
  {
    gauge_t *tmp;
    size_t i;

    tmp = realloc (ce->history, sizeof (*ce->history)
	* num_steps * ce->values_num);
    if (tmp == NULL)
    {
      pthread_mutex_unlock (&cache_lock);
      return (-ENOMEM);
    }

    for (i = ce->history_length * ce->values_num;
	i < (num_steps * ce->values_num);
	i++)
      tmp[i] = NAN;

    ce->history = tmp;
    ce->history_length = num_steps;
  } /* if (ce->history_length < num_steps) */

  /* Copy the values to the output buffer. */
  for (i = 0; i < num_steps; i++)
  {
    size_t src_index;
    size_t dst_index;

    if (i < ce->history_index)
      src_index = ce->history_index - (i + 1);
    else
      src_index = ce->history_length + ce->history_index - (i + 1);
    src_index = src_index * num_ds;

    dst_index = i * num_ds;

    memcpy (ret_history + dst_index, ce->history + src_index,
	sizeof (*ret_history) * num_ds);
  }

  pthread_mutex_unlock (&cache_lock);

  return (0);
} /* int uc_get_history_by_name */

int uc_update (const data_set_t *ds, const value_list_t *vl)
{
  char name[6 * DATA_MAX_NAME_LEN];
  cache_entry_t *ce = NULL;
  int send_okay_notification = 0;
  time_t update_delay = 0;
  notification_t n;
  int status;
  int i;

  if (FORMAT_VL (name, sizeof (name), vl) != 0)
  {
    ERROR ("uc_update: FORMAT_VL failed.");
    return (-1);
  }

  pthread_mutex_lock (&cache_lock);

  status = c_avl_get (cache_tree, name, (void *) &ce);
  if (status != 0) /* entry does not yet exist */
  {
    status = uc_insert (ds, vl, name);
    pthread_mutex_unlock (&cache_lock);
    return (status);
  }

  assert (ce != NULL);
  assert (ce->values_num == ds->ds_num);

  if (ce->last_time >= vl->time)
  {
    pthread_mutex_unlock (&cache_lock);
    NOTICE ("uc_update: Value too old: name = %s; value time = %u; "
	"last cache update = %u;",
	name, (unsigned int) vl->time, (unsigned int) ce->last_time);
    return (-1);
  }

  /* Send a notification (after the lock has been released) if we switch the
   * state from something else to `okay'. */
  if (ce->state == STATE_MISSING)
  {
    send_okay_notification = 1;
    ce->state = STATE_OKAY;
    update_delay = time (NULL) - ce->last_update;
  }

  for (i = 0; i < ds->ds_num; i++)
  {
    switch (ds->ds[i].type)
    {
      case DS_TYPE_COUNTER:
	{
	  counter_t diff;

	  /* check if the counter has wrapped around */
	  if (vl->values[i].counter < ce->values_raw[i].counter)
	  {
	    if (ce->values_raw[i].counter <= 4294967295U)
	      diff = (4294967295U - ce->values_raw[i].counter)
		+ vl->values[i].counter;
	    else
	      diff = (18446744073709551615ULL - ce->values_raw[i].counter)
		+ vl->values[i].counter;
	  }
	  else /* counter has NOT wrapped around */
	  {
	    diff = vl->values[i].counter - ce->values_raw[i].counter;
	  }

	  ce->values_gauge[i] = ((double) diff)
	    / ((double) (vl->time - ce->last_time));
	  ce->values_raw[i].counter = vl->values[i].counter;
	}
	break;

      case DS_TYPE_GAUGE:
	ce->values_raw[i].gauge = vl->values[i].gauge;
	ce->values_gauge[i] = vl->values[i].gauge;
	break;

      case DS_TYPE_DERIVE:
	{
	  derive_t diff;

	  diff = vl->values[i].derive - ce->values_raw[i].derive;

	  ce->values_gauge[i] = ((double) diff)
	    / ((double) (vl->time - ce->last_time));
	  ce->values_raw[i].derive = vl->values[i].derive;
	}
	break;

      case DS_TYPE_ABSOLUTE:
	ce->values_gauge[i] = ((double) vl->values[i].absolute)
	  / ((double) (vl->time - ce->last_time));
	ce->values_raw[i].absolute = vl->values[i].absolute;
	break;

      default:
//.........这里部分代码省略.........