@cryptotaxi247 / netdata-1 / commits / c8ded37b2

Fix race condition in dbengine (#7565)

* fix_db_race_condition: unit test Adjust unit test for dbengine * fix_db_race_condition: page cache Fix database * fix_db_race_condition: Missing function call This commit brings the correct function call inside rrdengine.c

thiagoftsm committed Dec 18, 2019 at 16:37 UTC c8ded37b25237ec4aed8604888bd4281c33087a6
8 files changed +121 -73
daemon/main.c
+9 -4
@@ -340,10 +340,12 @@ int help(int exitcode) {
340 " -W unittest Run internal unittests and exit.\n\n"
341 #ifdef ENABLE_DBENGINE
342 " -W createdataset=N Create a DB engine dataset of N seconds and exit.\n\n"
343 - " -W stresstest=A,B,C,D,E Run a DB engine stress test for A seconds,\n"
343 + " -W stresstest=A,B,C,D,E,F\n"
344 + " Run a DB engine stress test for A seconds,\n"
345 " with B writers and C readers, with a ramp up\n"
346 " time of D seconds for writers, a page cache\n"
346 - " size of E MiB, and exit.\n\n"
347 + " size of E MiB, an optional disk space limit"
348 + " of F MiB and exit.\n\n"
349 #endif
350 " -W set section option value\n"
351 " set netdata.conf option from the command line.\n\n"
@@ -956,7 +958,7 @@ int main(int argc, char **argv) {
958 else if(strncmp(optarg, stresstest_string, strlen(stresstest_string)) == 0) {
959 char *endptr;
960 unsigned test_duration_sec = 0, dset_charts = 0, query_threads = 0, ramp_up_seconds = 0,
959 - page_cache_mb = 0;
961 + page_cache_mb = 0, disk_space_mb = 0;
962
963 optarg += strlen(stresstest_string);
964 test_duration_sec = (unsigned)strtoul(optarg, &endptr, 0);
@@ -968,8 +970,11 @@ int main(int argc, char **argv) {
970 ramp_up_seconds = (unsigned)strtoul(endptr + 1, &endptr, 0);
971 if (',' == *endptr)
972 page_cache_mb = (unsigned)strtoul(endptr + 1, &endptr, 0);
973 + if (',' == *endptr)
974 + disk_space_mb = (unsigned)strtoul(endptr + 1, &endptr, 0);
975 +
976 dbengine_stress_test(test_duration_sec, dset_charts, query_threads, ramp_up_seconds,
972 - page_cache_mb);
977 + page_cache_mb, disk_space_mb);
978 return 0;
979 }
980 #endif
daemon/unit_test.c
+56 -27
@@ -1723,7 +1723,7 @@ int test_dbengine(void)
1723
1724 default_rrd_memory_mode = RRD_MEMORY_MODE_DBENGINE;
1725
1726 - debug(D_RRDHOST, "Initializing localhost with hostname 'unittest-dbengine'");
1726 + fprintf(stderr, "Initializing localhost with hostname 'unittest-dbengine'");
1727 host = dbengine_rrdhost_find_or_create("unittest-dbengine");
1728 if (NULL == host)
1729 return 1;
@@ -1915,6 +1915,9 @@ static void generate_dbengine_chart(void *arg)
1915 rrdset_done(st);
1916 thread_info->time_max = time_current;
1917 }
1918 + for (j = 0; j < DSET_DIMS; ++j) {
1919 + rrdeng_store_metric_finalize(rd[j]);
1920 + }
1921 }
1922
1923 void generate_dbengine_dataset(unsigned history_seconds)
@@ -1935,7 +1938,7 @@ void generate_dbengine_dataset(unsigned history_seconds)
1938 default_rrdeng_disk_quota_mb -= default_rrdeng_disk_quota_mb * EXPECTED_COMPRESSION_RATIO / 100;
1939
1940 error_log_limit_unlimited();
1938 - debug(D_RRDHOST, "Initializing localhost with hostname 'dbengine-dataset'");
1941 + fprintf(stderr, "Initializing localhost with hostname 'dbengine-dataset'");
1942
1943 host = dbengine_rrdhost_find_or_create("dbengine-dataset");
1944 if (NULL == host)
@@ -1986,6 +1989,7 @@ struct dbengine_query_thread {
1989 unsigned history_seconds; /* how far back in the past to go */
1990 volatile long done; /* initialize to 0, set to 1 to stop thread */
1991 unsigned long errors, queries_nr, queried_metrics_nr; /* statistics */
1992 + uint8_t delete_old_data; /* if non zero then data are deleted when disk space is exhausted */
1993
1994 struct dbengine_chart_thread *chart_threads[]; /* dset_charts elements */
1995 };
@@ -1995,7 +1999,7 @@ static void query_dbengine_chart(void *arg)
1999 struct dbengine_query_thread *thread_info = (struct dbengine_query_thread *)arg;
2000 const int DSET_CHARTS = thread_info->dset_charts;
2001 const int DSET_DIMS = thread_info->dset_dims;
1998 - time_t time_after, time_before, time_min, time_max, duration;
2002 + time_t time_after, time_before, time_min, time_approx_min, time_max, duration;
2003 int i, j, update_every = 1;
2004 RRDSET *st;
2005 RRDDIM *rd;
@@ -2015,6 +2019,13 @@ static void query_dbengine_chart(void *arg)
2019
2020 time_min = thread_info->time_present - thread_info->history_seconds + 1;
2021 time_max = thread_info->chart_threads[i]->time_max;
2022 +
2023 + if (thread_info->delete_old_data) {
2024 + /* A time window of twice the disk space is sufficient for compression space savings of up to 50% */
2025 + time_approx_min = time_max - (default_rrdeng_disk_quota_mb * 2 * 1024 * 1024) /
2026 + (((uint64_t) DSET_DIMS * DSET_CHARTS) * sizeof(storage_number));
2027 + time_min = MAX(time_min, time_approx_min);
2028 + }
2029 if (!time_max) {
2030 time_before = time_after = time_min;
2031 } else {
@@ -2030,18 +2041,22 @@ static void query_dbengine_chart(void *arg)
2041 expected = unpack_storage_number(pack_storage_number((calculated_number) generatedv, SN_EXISTS));
2042
2043 if (unlikely(rd->state->query_ops.is_finished(&handle))) {
2033 - fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2034 - CALCULATED_NUMBER_FORMAT ", found data gap, ### E R R O R ###\n",
2035 - st->name, rd->name, (unsigned long) time_now, expected);
2036 - ++thread_info->errors;
2044 + if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
2045 + fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2046 + CALCULATED_NUMBER_FORMAT ", found data gap, ### E R R O R ###\n",
2047 + st->name, rd->name, (unsigned long) time_now, expected);
2048 + ++thread_info->errors;
2049 + }
2050 break;
2051 }
2052 n = rd->state->query_ops.next_metric(&handle, &time_retrieved);
2053 if (SN_EMPTY_SLOT == n) {
2041 - fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2042 - CALCULATED_NUMBER_FORMAT ", found data gap, ### E R R O R ###\n",
2043 - st->name, rd->name, (unsigned long) time_now, expected);
2044 - ++thread_info->errors;
2054 + if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
2055 + fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2056 + CALCULATED_NUMBER_FORMAT ", found data gap, ### E R R O R ###\n",
2057 + st->name, rd->name, (unsigned long) time_now, expected);
2058 + ++thread_info->errors;
2059 + }
2060 break;
2061 }
2062 ++thread_info->queried_metrics_nr;
@@ -2049,15 +2064,21 @@ static void query_dbengine_chart(void *arg)
2064
2065 same = (calculated_number_round(value) == calculated_number_round(expected)) ? 1 : 0;
2066 if (!same) {
2052 - fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2053 - CALCULATED_NUMBER_FORMAT ", found " CALCULATED_NUMBER_FORMAT ", ### E R R O R ###\n",
2054 - st->name, rd->name, (unsigned long) time_now, expected, value);
2055 - ++thread_info->errors;
2067 + if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
2068 + fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value "
2069 + CALCULATED_NUMBER_FORMAT ", found " CALCULATED_NUMBER_FORMAT
2070 + ", ### E R R O R ###\n",
2071 + st->name, rd->name, (unsigned long) time_now, expected, value);
2072 + ++thread_info->errors;
2073 + }
2074 }
2075 if (time_retrieved != time_now) {
2058 - fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, found timestamp %lu ### E R R O R ###\n",
2059 - st->name, rd->name, (unsigned long) time_now, (unsigned long) time_retrieved);
2060 - ++thread_info->errors;
2076 + if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
2077 + fprintf(stderr,
2078 + " DB-engine stresstest %s/%s: at %lu secs, found timestamp %lu ### E R R O R ###\n",
2079 + st->name, rd->name, (unsigned long) time_now, (unsigned long) time_retrieved);
2080 + ++thread_info->errors;
2081 + }
2082 }
2083 }
2084 rd->state->query_ops.finalize(&handle);
@@ -2065,17 +2086,19 @@ static void query_dbengine_chart(void *arg)
2086 }
2087
2088 void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsigned QUERY_THREADS,
2068 - unsigned RAMP_UP_SECONDS, unsigned PAGE_CACHE_MB)
2089 + unsigned RAMP_UP_SECONDS, unsigned PAGE_CACHE_MB, unsigned DISK_SPACE_MB)
2090 {
2091 const unsigned DSET_DIMS = 128;
2092 const uint64_t EXPECTED_COMPRESSION_RATIO = 20;
2072 - const unsigned HISTORY_SECONDS = 3600 * 24 * 365; /* 1 year of history */
2093 + const unsigned HISTORY_SECONDS = 3600 * 24 * 365 * 50; /* 50 year of history */
2094 RRDHOST *host = NULL;
2095 struct dbengine_chart_thread **chart_threads;
2096 struct dbengine_query_thread **query_threads;
2097 unsigned i, j;
2098 time_t time_start, time_end;
2099
2100 + error_log_limit_unlimited();
2101 +
2102 if (!TEST_DURATION_SEC)
2103 TEST_DURATION_SEC = 10;
2104 if (!DSET_CHARTS)
@@ -2087,13 +2110,18 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi
2110
2111 default_rrd_memory_mode = RRD_MEMORY_MODE_DBENGINE;
2112 default_rrdeng_page_cache_mb = PAGE_CACHE_MB;
2090 - // Worst case for uncompressible data
2091 - default_rrdeng_disk_quota_mb = (((uint64_t)DSET_DIMS * DSET_CHARTS) * sizeof(storage_number) * HISTORY_SECONDS) /
2092 - (1024 * 1024);
2093 - default_rrdeng_disk_quota_mb -= default_rrdeng_disk_quota_mb * EXPECTED_COMPRESSION_RATIO / 100;
2113 + if (DISK_SPACE_MB) {
2114 + fprintf(stderr, "By setting disk space limit data are allowed to be deleted. "
2115 + "Data validation is turned off for this run.\n");
2116 + default_rrdeng_disk_quota_mb = DISK_SPACE_MB;
2117 + } else {
2118 + // Worst case for uncompressible data
2119 + default_rrdeng_disk_quota_mb =
2120 + (((uint64_t) DSET_DIMS * DSET_CHARTS) * sizeof(storage_number) * HISTORY_SECONDS) / (1024 * 1024);
2121 + default_rrdeng_disk_quota_mb -= default_rrdeng_disk_quota_mb * EXPECTED_COMPRESSION_RATIO / 100;
2122 + }
2123
2095 - error_log_limit_unlimited();
2096 - debug(D_RRDHOST, "Initializing localhost with hostname 'dbengine-stress-test'");
2124 + fprintf(stderr, "Initializing localhost with hostname 'dbengine-stress-test'\n");
2125
2126 host = dbengine_rrdhost_find_or_create("dbengine-stress-test");
2127 if (NULL == host)
@@ -2112,7 +2140,7 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi
2140 "%u MiB of page cache.\n",
2141 RAMP_UP_SECONDS, TEST_DURATION_SEC, DSET_CHARTS, QUERY_THREADS, PAGE_CACHE_MB);
2142
2115 - time_start = now_realtime_sec();
2143 + time_start = now_realtime_sec() + HISTORY_SECONDS; /* move history to the future */
2144 for (i = 0 ; i < DSET_CHARTS ; ++i) {
2145 chart_threads[i]->host = host;
2146 chart_threads[i]->chartname = "random";
@@ -2146,6 +2174,7 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi
2174 for (j = 0 ; j < DSET_CHARTS ; ++j) {
2175 query_threads[i]->chart_threads[j] = chart_threads[j];
2176 }
2177 + query_threads[i]->delete_old_data = DISK_SPACE_MB ? 1 : 0;
2178 assert(0 == uv_thread_create(&query_threads[i]->thread, query_dbengine_chart, query_threads[i]));
2179 }
2180 sleep(TEST_DURATION_SEC);
daemon/unit_test.h
+1 -1
@@ -12,7 +12,7 @@ extern int unit_test_buffer(void);
12 extern int test_dbengine(void);
13 extern void generate_dbengine_dataset(unsigned history_seconds);
14 extern void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsigned QUERY_THREADS,
15 - unsigned RAMP_UP_SECONDS, unsigned PAGE_CACHE_MB);
15 + unsigned RAMP_UP_SECONDS, unsigned PAGE_CACHE_MB, unsigned DISK_SPACE_MB);
16
17 #endif
18
database/engine/pagecache.c
+24 -14
@@ -101,6 +101,13 @@ void pg_cache_wake_up_waiters_unsafe(struct rrdeng_page_descr *descr)
101 uv_cond_broadcast(&pg_cache_descr->cond);
102 }
103
104 +void pg_cache_wake_up_waiters(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr)
105 +{
106 + rrdeng_page_descr_mutex_lock(ctx, descr);
107 + pg_cache_wake_up_waiters_unsafe(descr);
108 + rrdeng_page_descr_mutex_unlock(ctx, descr);
109 +}
110 +
111 /*
112 * The caller must hold page descriptor lock.
113 * The lock will be released and re-acquired. The descriptor is not guaranteed
@@ -135,10 +142,8 @@ unsigned long pg_cache_wait_event(struct rrdengine_instance *ctx, struct rrdeng_
142
143 /*
144 * The caller must hold page descriptor lock.
138 - * Gets a reference to the page descriptor.
139 - * Returns 1 on success and 0 on failure.
145 */
141 -int pg_cache_try_get_unsafe(struct rrdeng_page_descr *descr, int exclusive_access)
146 +int pg_cache_can_get_unsafe(struct rrdeng_page_descr *descr, int exclusive_access)
147 {
148 struct page_cache_descr *pg_cache_descr = descr->pg_cache_descr;
149
@@ -146,25 +151,25 @@ int pg_cache_try_get_unsafe(struct rrdeng_page_descr *descr, int exclusive_acces
151 (exclusive_access && pg_cache_descr->refcnt)) {
152 return 0;
153 }
149 - if (exclusive_access)
150 - pg_cache_descr->flags |= RRD_PAGE_LOCKED;
151 - ++pg_cache_descr->refcnt;
154
155 return 1;
156 }
157
158 /*
159 * The caller must hold page descriptor lock.
158 - * Same return values as pg_cache_try_get_unsafe() without doing anything.
160 + * Gets a reference to the page descriptor.
161 + * Returns 1 on success and 0 on failure.
162 */
160 -int pg_cache_can_get_unsafe(struct rrdeng_page_descr *descr, int exclusive_access)
163 +int pg_cache_try_get_unsafe(struct rrdeng_page_descr *descr, int exclusive_access)
164 {
165 struct page_cache_descr *pg_cache_descr = descr->pg_cache_descr;
166
164 - if ((pg_cache_descr->flags & (RRD_PAGE_LOCKED | RRD_PAGE_READ_PENDING)) ||
165 - (exclusive_access && pg_cache_descr->refcnt)) {
167 + if (!pg_cache_can_get_unsafe(descr, exclusive_access))
168 return 0;
167 - }
169 +
170 + if (exclusive_access)
171 + pg_cache_descr->flags |= RRD_PAGE_LOCKED;
172 + ++pg_cache_descr->refcnt;
173
174 return 1;
175 }
@@ -409,7 +414,9 @@ void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_desc
414 print_page_cache_descr(descr);
415 pg_cache_wait_event_unsafe(descr);
416 }
412 - if (!remove_dirty) {
417 + if (remove_dirty) {
418 + pg_cache_descr->flags &= ~RRD_PAGE_DIRTY;
419 + } else {
420 /* even a locked page could be dirty */
421 while (unlikely(pg_cache_descr->flags & RRD_PAGE_DIRTY)) {
422 debug(D_RRDENGINE, "%s: Found dirty page, waiting for it to be flushed:", __func__);
@@ -429,8 +436,11 @@ void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_desc
436 uv_rwlock_wrunlock(&pg_cache->pg_cache_rwlock);
437 }
438 pg_cache_put(ctx, descr);
432 - if (descr->pg_cache_descr_state & PG_CACHE_DESCR_ALLOCATED)
433 - rrdeng_try_deallocate_pg_cache_descr(ctx, descr);
439 + rrdeng_try_deallocate_pg_cache_descr(ctx, descr);
440 + while (descr->pg_cache_descr_state & PG_CACHE_DESCR_ALLOCATED) {
441 + rrdeng_try_deallocate_pg_cache_descr(ctx, descr); /* spin */
442 + (void)sleep_usec(1000); /* 1 msec */
443 + }
444 destroy:
445 freez(descr);
446 pg_cache_update_metric_times(page_index);
database/engine/pagecache.h
+1
@@ -148,6 +148,7 @@ struct page_cache { /* TODO: add statistics */
148 };
149
150 extern void pg_cache_wake_up_waiters_unsafe(struct rrdeng_page_descr *descr);
151 +extern void pg_cache_wake_up_waiters(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr);
152 extern void pg_cache_wait_event_unsafe(struct rrdeng_page_descr *descr);
153 extern unsigned long pg_cache_wait_event(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr);
154 extern void pg_cache_replaceQ_insert(struct rrdengine_instance *ctx,
database/engine/rrdengine.c
+5 -5
@@ -121,19 +121,19 @@ after_crc_check:
121 } else {
122 (void) memcpy(page, uncompressed_buf + page_offset, descr->page_length);
123 }
124 - pg_cache_replaceQ_insert(ctx, descr);
124 rrdeng_page_descr_mutex_lock(ctx, descr);
125 pg_cache_descr = descr->pg_cache_descr;
126 pg_cache_descr->page = page;
127 pg_cache_descr->flags |= RRD_PAGE_POPULATED;
128 pg_cache_descr->flags &= ~RRD_PAGE_READ_PENDING;
130 - debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);
129 + rrdeng_page_descr_mutex_unlock(ctx, descr);
130 + pg_cache_replaceQ_insert(ctx, descr);
131 if (xt_io_descr->release_descr) {
132 - pg_cache_put_unsafe(descr);
132 + pg_cache_put(ctx, descr);
133 } else {
134 - pg_cache_wake_up_waiters_unsafe(descr);
134 + debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);
135 + pg_cache_wake_up_waiters(ctx, descr);
136 }
136 - rrdeng_page_descr_mutex_unlock(ctx, descr);
137 }
138 if (!have_read_error && RRD_NO_COMPRESSION != header->compression_algorithm) {
139 freez(uncompressed_buf);
database/engine/rrdengineapi.c
+6 -5
@@ -473,17 +473,18 @@ storage_number rrdeng_load_metric_next(struct rrddim_query_handle *rrdimm_handle
473 /* We need to get a new page */
474 if (descr) {
475 /* Drop old page's reference */
476 - handle->next_page_time = (page_end_time / USEC_PER_SEC) + 1;
477 - if (unlikely(handle->next_page_time > rrdimm_handle->end_time)) {
478 - goto no_more_metrics;
479 - }
480 - next_page_time = handle->next_page_time * USEC_PER_SEC;
476 #ifdef NETDATA_INTERNAL_CHECKS
477 rrd_stat_atomic_add(&ctx->stats.metric_API_consumers, -1);
478 #endif
479 pg_cache_put(ctx, descr);
480 handle->descr = NULL;
481 + handle->next_page_time = (page_end_time / USEC_PER_SEC) + 1;
482 + if (unlikely(handle->next_page_time > rrdimm_handle->end_time)) {
483 + goto no_more_metrics;
484 + }
485 + next_page_time = handle->next_page_time * USEC_PER_SEC;
486 }
487 +
488 descr = pg_cache_lookup_next(ctx, handle->page_index, &handle->page_index->id,
489 next_page_time, rrdimm_handle->end_time * USEC_PER_SEC);
490 if (NULL == descr) {
database/engine/rrdenglocking.c
+19 -17
@@ -20,12 +20,7 @@ struct page_cache_descr *rrdeng_create_pg_cache_descr(struct rrdengine_instance
20
21 void rrdeng_destroy_pg_cache_descr(struct rrdengine_instance *ctx, struct page_cache_descr *pg_cache_descr)
22 {
23 - /* Flush any lock and condition variable users */
24 - uv_mutex_lock(&pg_cache_descr->mutex);
25 - uv_mutex_unlock(&pg_cache_descr->mutex);
26 -
23 uv_cond_destroy(&pg_cache_descr->cond);
28 -
24 uv_mutex_destroy(&pg_cache_descr->mutex);
25 freez(pg_cache_descr);
26 rrd_stat_atomic_add(&ctx->stats.page_cache_descriptors, -1);
@@ -100,7 +95,7 @@ void rrdeng_page_descr_mutex_lock(struct rrdengine_instance *ctx, struct rrdeng_
95 void rrdeng_page_descr_mutex_unlock(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr)
96 {
97 unsigned long old_state, new_state, ret_state, old_users;
103 - struct page_cache_descr *pg_cache_descr;
98 + struct page_cache_descr *pg_cache_descr, *delete_pg_cache_descr = NULL;
99 uint8_t we_locked;
100
101 uv_mutex_unlock(&descr->pg_cache_descr->mutex);
@@ -116,7 +111,8 @@ void rrdeng_page_descr_mutex_unlock(struct rrdengine_instance *ctx, struct rrden
111 ret_state = ulong_compare_and_swap(&descr->pg_cache_descr_state, old_state, 0);
112 if (old_state == ret_state) {
113 /* success */
119 - break;
114 + rrdeng_destroy_pg_cache_descr(ctx, delete_pg_cache_descr);
115 + return;
116 }
117 continue; /* spin */
118 }
@@ -128,12 +124,15 @@ void rrdeng_page_descr_mutex_unlock(struct rrdengine_instance *ctx, struct rrden
124 pg_cache_descr = descr->pg_cache_descr;
125 /* caller is the only page cache descriptor user and there are no pending references on the page */
126 if ((old_state & PG_CACHE_DESCR_DESTROY) && (1 == old_users) &&
131 - !pg_cache_descr->flags && !pg_cache_descr->refcnt && !pg_cache_descr->waiters) {
127 + !pg_cache_descr->flags && !pg_cache_descr->refcnt) {
128 + assert(!pg_cache_descr->waiters);
129 +
130 new_state = PG_CACHE_DESCR_LOCKED;
131 ret_state = ulong_compare_and_swap(&descr->pg_cache_descr_state, old_state, new_state);
132 if (old_state == ret_state) {
133 we_locked = 1;
136 - rrdeng_destroy_pg_cache_descr(ctx, pg_cache_descr);
134 + delete_pg_cache_descr = pg_cache_descr;
135 + descr->pg_cache_descr = NULL;
136 /* retry */
137 continue;
138 }
@@ -150,7 +149,6 @@ void rrdeng_page_descr_mutex_unlock(struct rrdengine_instance *ctx, struct rrden
149 }
150 /* spin */
151 }
153 -
152 }
153
154 /*
@@ -161,11 +159,11 @@ void rrdeng_page_descr_mutex_unlock(struct rrdengine_instance *ctx, struct rrden
159 void rrdeng_try_deallocate_pg_cache_descr(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr)
160 {
161 unsigned long old_state, new_state, ret_state, old_users;
164 - struct page_cache_descr *pg_cache_descr;
165 - uint8_t just_locked, we_freed, must_unlock;
162 + struct page_cache_descr *pg_cache_descr = NULL;
163 + uint8_t just_locked, can_free, must_unlock;
164
165 just_locked = 0;
168 - we_freed = 0;
166 + can_free = 0;
167 must_unlock = 0;
168 while (1) { /* spin */
169 old_state = descr->pg_cache_descr_state;
@@ -177,9 +175,11 @@ void rrdeng_try_deallocate_pg_cache_descr(struct rrdengine_instance *ctx, struct
175 must_unlock = 1;
176 just_locked = 0;
177 /* Try deallocate if there are no pending references on the page */
180 - if (!pg_cache_descr->flags && !pg_cache_descr->refcnt && !pg_cache_descr->waiters) {
181 - rrdeng_destroy_pg_cache_descr(ctx, pg_cache_descr);
182 - we_freed = 1;
178 + if (!pg_cache_descr->flags && !pg_cache_descr->refcnt) {
179 + assert(!pg_cache_descr->waiters);
180 +
181 + descr->pg_cache_descr = NULL;
182 + can_free = 1;
183 /* success */
184 continue;
185 }
@@ -188,7 +188,7 @@ void rrdeng_try_deallocate_pg_cache_descr(struct rrdengine_instance *ctx, struct
188 if (unlikely(must_unlock)) {
189 assert(0 == old_users);
190
191 - if (we_freed) {
191 + if (can_free) {
192 /* success */
193 new_state = 0;
194 } else {
@@ -198,6 +198,8 @@ void rrdeng_try_deallocate_pg_cache_descr(struct rrdengine_instance *ctx, struct
198 ret_state = ulong_compare_and_swap(&descr->pg_cache_descr_state, old_state, new_state);
199 if (old_state == ret_state) {
200 /* unlocked */
201 + if (can_free)
202 + rrdeng_destroy_pg_cache_descr(ctx, pg_cache_descr);
203 return;
204 }
205 continue; /* spin */