Skip to content

Commit

Permalink
Merge pull request #92 from comptonqc/patch-pending_summary
Browse files Browse the repository at this point in the history 
Improve metric summarization performance under heavy load
  • Loading branch information
@vvuksan
vvuksan committed Apr 30, 2013
2 parents e077126 + 0705a5d commit 8ff4bb3
Show file tree
Hide file tree
Showing 3 changed files with 41 additions and 25 deletions.
8 changes: 5 additions & 3 deletions gmetad/cleanup.c
View file
Expand Up @@ -129,19 +129,21 @@ cleanup_source ( datum_t *key, datum_t *val, void *arg )
#if 0
unsigned int born;
#endif
hash_t *metric_summary;

/* Currently we never delete a source. */

/* Clean up metric summaries */
cleanup.tv = tv;
cleanup.key = 0;
cleanup.hashval = 0;
while (hash_walkfrom(source->metric_summary, cleanup.hashval,
metric_summary = source->metric_summary; /* Just look at the current one */
while (hash_walkfrom(metric_summary, cleanup.hashval,
cleanup_metric, (void*) &cleanup)) {

if (cleanup.key) {
cleanup.hashval = hashval(cleanup.key, source->metric_summary);
rv=hash_delete(cleanup.key, source->metric_summary);
cleanup.hashval = hashval(cleanup.key, metric_summary);
rv=hash_delete(cleanup.key, metric_summary);
if (rv) datum_free(rv);
cleanup.key=0;
}
Expand Down
1 change: 1 addition & 0 deletions gmetad/gmetad.h
View file
Expand Up @@ -168,6 +168,7 @@ typedef struct
hash_t *authority; /* Null for a grid. */
short int authority_ptr; /* An authority URL. */
hash_t *metric_summary;
hash_t *metric_summary_pending;
pthread_mutex_t *sum_finished; /* A lock held during summarization. */
data_source_list_t *ds;
uint32_t hosts_up;
Expand Down
57 changes: 35 additions & 22 deletions gmetad/process_xml.c
View file
Expand Up @@ -204,30 +204,31 @@ startElement_GRID(void *data, const char *el, const char **attr)
name);
return 1;
}

source->metric_summary_pending = hash_create(DEFAULT_METRICSIZE);
if (!source->metric_summary_pending)
{
err_msg("Could not create pending summary hash for cluster %s",
name);
return 1;
}

source->ds = xmldata->ds;

/* Initialize the partial sum lock */
source->sum_finished = (pthread_mutex_t *)
malloc(sizeof(pthread_mutex_t));
pthread_mutex_init(source->sum_finished, NULL);

/* Grab the "partial sum" mutex until we are finished
* summarizing. */
pthread_mutex_lock(source->sum_finished);
}
else
{ /* Found Cluster. Put into our Source buffer in xmldata. */
memcpy(source, hash_datum->data, hash_datum->size);
datum_free(hash_datum);

/* Grab the "partial sum" mutex until we are finished
* summarizing. Needs to be done asap.*/
pthread_mutex_lock(source->sum_finished);

source->hosts_up = 0;
source->hosts_down = 0;

hash_foreach(source->metric_summary, zero_out_summary, NULL);
hash_foreach(source->metric_summary_pending, zero_out_summary, NULL);
}

/* Edge has the same invariant as in fillmetric(). */
Expand Down Expand Up @@ -323,28 +324,30 @@ startElement_CLUSTER(void *data, const char *el, const char **attr)
err_msg("Could not create summary hash for cluster %s", name);
return 1;
}

source->metric_summary_pending = hash_create(DEFAULT_METRICSIZE);
if (!source->metric_summary_pending)
{
err_msg("Could not create pending summary hash for cluster %s", name);
return 1;
}

source->ds = xmldata->ds;

/* Initialize the partial sum lock */
source->sum_finished = (pthread_mutex_t *)
malloc(sizeof(pthread_mutex_t));
pthread_mutex_init(source->sum_finished, NULL);

/* Grab the "partial sum" mutex until we are finished summarizing. */
pthread_mutex_lock(source->sum_finished);
}
else
{
memcpy(source, hash_datum->data, hash_datum->size);
datum_free(hash_datum);

/* We need this lock before zeroing metric sums. */
pthread_mutex_lock(source->sum_finished);

source->hosts_up = 0;
source->hosts_down = 0;

hash_foreach(source->metric_summary, zero_out_summary, NULL);
hash_foreach(source->metric_summary_pending, zero_out_summary, NULL);
}

/* Edge has the same invariant as in fillmetric(). */
Expand Down Expand Up @@ -681,7 +684,7 @@ startElement_METRIC(void *data, const char *el, const char **attr)
/* Always update summary for numeric metrics. */
if (do_summary)
{
summary = xmldata->source.metric_summary;
summary = xmldata->source.metric_summary_pending;
hash_datum = hash_lookup(&hashkey, summary);
if (!hash_datum)
{
Expand Down Expand Up @@ -921,7 +924,7 @@ startElement_METRICS(void *data, const char *el, const char **attr)
}
}

summary = xmldata->source.metric_summary;
summary = xmldata->source.metric_summary_pending;
hash_datum = hash_lookup(&hashkey, summary);
if (!hash_datum)
{
Expand Down Expand Up @@ -1138,9 +1141,14 @@ endElement_GRID(void *data, const char *el)
if (authority_mode(xmldata))
{
source = &xmldata->source;
summary = xmldata->source.metric_summary;

/* Release the partial sum mutex */
/* Swap the metric_summary and metric_summary_pending over */
pthread_mutex_lock(source->sum_finished);
{
summary = xmldata->source.metric_summary_pending;
xmldata->source.metric_summary_pending = xmldata->source.metric_summary;
xmldata->source.metric_summary = summary;
}
pthread_mutex_unlock(source->sum_finished);

hashkey.data = (void*) xmldata->sourcename;
Expand Down Expand Up @@ -1176,9 +1184,14 @@ endElement_CLUSTER(void *data, const char *el)
if (authority_mode(xmldata))
{
source = &xmldata->source;
summary = xmldata->source.metric_summary;

/* Release the partial sum mutex */
/* Swap the metric_summary and metric_summary_pending over */
pthread_mutex_lock(source->sum_finished);
{
summary = xmldata->source.metric_summary_pending;
xmldata->source.metric_summary_pending = xmldata->source.metric_summary;
xmldata->source.metric_summary = summary;
}
pthread_mutex_unlock(source->sum_finished);

hashkey.data = (void*) xmldata->sourcename;
Expand Down

0 comments on commit 8ff4bb3

Please sign in to comment.