@cryptotaxi247 / netdata-1 / commits / 00f4b111f

free strings judy arrays to show leaked strings (#19827)

* free strings judy arrays to show leaked strings * fix cleanup of rrd functions * free ae->summary when freeing alert transitions * fix leak in dyncfg json parsing of alert prototypes * fix leak in chart names * fix leak in query leaking units * fix plugins.d leaking strings * delay stream connector exit * log startup/shutdown step in sanitizer mode

Costa Tsaousis committed Mar 12, 2025 at 16:08 UTC 00f4b111f41fdedb9f381035f39a5554c526d434

15 files changed +174 -89

src/daemon/daemon-service.h

+6 -5

@@ -16,11 +16,12 @@ typedef enum {

16 SERVICE_ACLK = (1 << 7),

17 SERVICE_HEALTH = (1 << 8),

18 SERVICE_STREAMING = (1 << 9),

19 - SERVICE_CONTEXT = (1 << 10),

20 - SERVICE_ANALYTICS = (1 << 11),

21 - SERVICE_EXPORTERS = (1 << 12),

22 - SERVICE_HTTPD = (1 << 13),

23 - SERVICE_SYSTEMD = (1 << 14),

19 + SERVICE_STREAMING_CONNECTOR = (1 << 10),

20 + SERVICE_CONTEXT = (1 << 11),

21 + SERVICE_ANALYTICS = (1 << 12),

22 + SERVICE_EXPORTERS = (1 << 13),

23 + SERVICE_HTTPD = (1 << 14),

24 + SERVICE_SYSTEMD = (1 << 15),

25 } SERVICE_TYPE;

27 typedef enum {

src/daemon/daemon-shutdown-watcher.c

+18

@@ -34,6 +34,12 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

34 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

35 watcher_steps[step_id].msg);

37 +#if defined(FSANITIZE_ADDRESS)

38 + fprintf(stderr, " > shutdown step: [%d/%d] - {at %s} started '%s'...\n",

39 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

40 + watcher_steps[step_id].msg);

41 +#endif

42 +

43 daemon_status_file_shutdown_step(watcher_steps[step_id].msg);

45 #ifdef ENABLE_SENTRY

@@ -61,6 +67,12 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

67 netdata_log_info("shutdown step: [%d/%d] - {at %s} finished '%s' in %s",

68 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

69 watcher_steps[step_id].msg, step_duration_txt);

70 +

71 +#if defined(FSANITIZE_ADDRESS)

72 + fprintf(stderr, " > shutdown step: [%d/%d] - {at %s} finished '%s' in %s\n",

73 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

74 + watcher_steps[step_id].msg, step_duration_txt);

75 +#endif

76 } else {

77 // Do not call fatal() because it will try to execute the exit

78 // sequence twice.

@@ -68,6 +80,12 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

80 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

81 watcher_steps[step_id].msg, step_duration_txt);

83 +#if defined(FSANITIZE_ADDRESS)

84 + fprintf(stderr, "shutdown step: [%d/%d] - {at %s} timeout '%s' takes too long (%s) - giving up...\n",

85 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

86 + watcher_steps[step_id].msg, step_duration_txt);

87 +#endif

88 +

89 daemon_status_file_shutdown_step("sentry timeout");

90 abort();

91 }

src/daemon/daemon-shutdown.c

+7 -1

@@ -214,7 +214,8 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

214 watcher_step_complete(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS);

215

216 stream_threads_cancel();

217 - service_wait_exit(SERVICE_COLLECTORS | SERVICE_STREAMING, 3 * USEC_PER_SEC);

217 + service_wait_exit(SERVICE_COLLECTORS | SERVICE_STREAMING, 20 * USEC_PER_SEC);

218 + service_signal_exit(SERVICE_STREAMING_CONNECTOR);

219 watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS);

220

221 #ifdef ENABLE_DBENGINE

@@ -352,6 +353,11 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

353 fprintf(stderr, "WARNING: UUIDMAP had %zu UUIDs referenced.\n",

354 uuid_referenced);

355

356 + size_t strings_referenced = string_destroy();

357 + if(strings_referenced)

358 + fprintf(stderr, "WARNING: STRING has %zu strings still allocated.\n",

359 + strings_referenced);

360 +

361 // strings_destroy();

362 // functions_destroy();

363 // dyncfg_destroy();

src/daemon/main.c

+12 -2

@@ -184,13 +184,23 @@ int help(int exitcode) {

184 be set in this procedure to be called in all the relevant code paths.

185 */

186

187 +#if defined(FSANITIZE_ADDRESS)

188 +#define LOG_TO_STDERR(...) fprintf(stderr, __VA_ARGS__)

189 +#else

190 +#define LOG_TO_STDERR(...)

191 +#endif

192 +

193 #define delta_startup_time(msg) \

194 do { \

195 usec_t now_ut = now_monotonic_usec(); \

190 - if(prev_msg) \

196 + if(prev_msg) { \

197 netdata_log_info("NETDATA STARTUP: in %7llu ms, %s - next: %s", (now_ut - last_ut) / USEC_PER_MS, prev_msg, msg); \

192 - else \

198 + LOG_TO_STDERR(" > startup: in %7llu ms, %s - next: %s\n", (now_ut - last_ut) / USEC_PER_MS, prev_msg, msg); \

199 + } \

200 + else { \

201 netdata_log_info("NETDATA STARTUP: next: %s", msg); \

202 + LOG_TO_STDERR(" > startup: next: %s\n", msg); \

203 + } \

204 last_ut = now_ut; \

205 prev_msg = msg; \

206 daemon_status_file_startup_step("startup(" msg ")"); \

src/database/contexts/api_v2_contexts.c

+4 -4

@@ -638,14 +638,14 @@ static bool contexts_conflict_callback(const DICTIONARY_ITEM *item __maybe_unuse

638 ;

639 else if(!(o->flags & RRD_FLAG_COLLECTED) && (n->flags & RRD_FLAG_COLLECTED)) {

640 // keep new

641 - string_freez(o->family);

642 - o->family = string_dup(n->family);

641 + SWAP(o->family, n->family);

642 }

643 else {

644 // merge

645 STRING *old_family = o->family;

646 o->family = string_2way_merge(o->family, n->family);

647 string_freez(old_family);

648 + // n->family will be freed below

649 }

650 }

651

@@ -655,8 +655,7 @@ static bool contexts_conflict_callback(const DICTIONARY_ITEM *item __maybe_unuse

655 ;

656 else if(!(o->flags & RRD_FLAG_COLLECTED) && (n->flags & RRD_FLAG_COLLECTED)) {

657 // keep new

658 - string_freez(o->units);

659 - o->units = string_dup(n->units);

658 + SWAP(o->units, n->units);

659 }

660 else {

661 // keep old

@@ -689,6 +688,7 @@ static bool contexts_conflict_callback(const DICTIONARY_ITEM *item __maybe_unuse

688 o->flags |= n->flags;

689 o->match = MIN(o->match, n->match);

690

691 + string_freez(n->units);

692 string_freez(n->family);

693

694 return true;

src/database/rrdfunctions.c

+20 -20

@@ -28,10 +28,16 @@ static void rrd_functions_insert_callback(const DICTIONARY_ITEM *item __maybe_un

28 // rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");

29 }

31 +static void rrd_functions_cleanup(struct rrd_host_function *rdcf) {

32 + rrd_collector_release(rdcf->collector);

33 + string_freez(rdcf->help);

34 + string_freez(rdcf->tags);

35 +}

36 +

37 static void rrd_functions_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,

38 void *rrdhost __maybe_unused) {

39 struct rrd_host_function *rdcf = func;

34 - rrd_collector_release(rdcf->collector);

40 + rrd_functions_cleanup(rdcf);

41 }

43 static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,

@@ -50,9 +56,8 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

56 dictionary_acquired_item_name(item), rrdhost_hostname(host),

57 rrd_collector_tid(rdcf->collector), rrd_collector_tid(thread_rrd_collector));

53 - struct rrd_collector *old_rdc = rdcf->collector;

59 + new_rdcf->collector = rdcf->collector;

60 rdcf->collector = rrd_collector_acquire_current_thread();

55 - rrd_collector_release(old_rdc);

61 changed = true;

62 }

@@ -72,7 +77,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

77 "FUNCTIONS: function '%s' of host '%s' changed execute callback",

78 dictionary_acquired_item_name(item), rrdhost_hostname(host));

75 - rdcf->execute_cb = new_rdcf->execute_cb;

80 + SWAP(rdcf->execute_cb, new_rdcf->execute_cb);

81 changed = true;

82 }

@@ -81,26 +86,18 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

86 "FUNCTIONS: function '%s' of host '%s' changed help text",

87 dictionary_acquired_item_name(item), rrdhost_hostname(host));

84 - STRING *old = rdcf->help;

85 - rdcf->help = new_rdcf->help;

86 - string_freez(old);

89 + SWAP(rdcf->help, new_rdcf->help);

90 changed = true;

91 }

89 - else

90 - string_freez(new_rdcf->help);

93 if(rdcf->tags != new_rdcf->tags) {

94 nd_log(NDLS_DAEMON, NDLP_DEBUG,

95 "FUNCTIONS: function '%s' of host '%s' changed tags",

96 dictionary_acquired_item_name(item), rrdhost_hostname(host));

97 - STRING *old = rdcf->tags;

98 - rdcf->tags = new_rdcf->tags;

99 - string_freez(old);

98 + SWAP(rdcf->tags, new_rdcf->tags);

99 changed = true;

100 }

102 - else

103 - string_freez(new_rdcf->tags);

101

102 if(rdcf->timeout != new_rdcf->timeout) {

103 nd_log(NDLS_DAEMON, NDLP_DEBUG,

@@ -108,7 +105,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

105 dictionary_acquired_item_name(item), rrdhost_hostname(host),

106 rdcf->timeout, new_rdcf->timeout);

107

111 - rdcf->timeout = new_rdcf->timeout;

108 + SWAP(rdcf->timeout, new_rdcf->timeout);

109 changed = true;

110 }

111

@@ -118,7 +115,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

115 dictionary_acquired_item_name(item), rrdhost_hostname(host),

116 rdcf->version, new_rdcf->version);

117

121 - rdcf->version = new_rdcf->version;

118 + SWAP(rdcf->version, new_rdcf->version);

119 changed = true;

120 }

121

@@ -127,7 +124,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

124 "FUNCTIONS: function '%s' of host '%s' changed priority",

125 dictionary_acquired_item_name(item), rrdhost_hostname(host));

126

130 - rdcf->priority = new_rdcf->priority;

127 + SWAP(rdcf->priority, new_rdcf->priority);

128 changed = true;

129 }

130

@@ -136,7 +133,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

133 "FUNCTIONS: function '%s' of host '%s' changed access level",

134 dictionary_acquired_item_name(item), rrdhost_hostname(host));

135

139 - rdcf->access = new_rdcf->access;

136 + SWAP(rdcf->access, new_rdcf->access);

137 changed = true;

138 }

139

@@ -145,7 +142,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

142 "FUNCTIONS: function '%s' of host '%s' changed sync/async mode",

143 dictionary_acquired_item_name(item), rrdhost_hostname(host));

144

148 - rdcf->sync = new_rdcf->sync;

145 + SWAP(rdcf->sync, new_rdcf->sync);

146 changed = true;

147 }

148

@@ -154,7 +151,7 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

151 "FUNCTIONS: function '%s' of host '%s' changed execute callback data",

152 dictionary_acquired_item_name(item), rrdhost_hostname(host));

153

157 - rdcf->execute_cb_data = new_rdcf->execute_cb_data;

154 + SWAP(rdcf->execute_cb_data, new_rdcf->execute_cb_data);

155 changed = true;

156 }

157

@@ -162,6 +159,8 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

159 // dictionary_acquired_item_name(item), rrdhost_hostname(host),

160 // rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");

161

162 + rrd_functions_cleanup(new_rdcf);

163 +

164 return changed;

165 }

166

@@ -232,6 +231,7 @@ void rrd_function_add(RRDHOST *host, RRDSET *st, const char *name, int timeout,

231 rrd_functions_sanitize(key, name, sizeof(key));

232

233 struct rrd_host_function tmp = {

234 + .collector = NULL,

235 .sync = sync,

236 .timeout = timeout,

237 .version = version,

src/health/health_dyncfg.c

+5 -5

@@ -183,14 +183,14 @@ static bool parse_prototype(json_object *jobj, const char *path, RRD_ALERT_PROTO

183

184 JSONC_PARSE_BOOL_OR_ERROR_AND_RETURN(rule, path, "enabled", ap->match.enabled, error, strict);

185

186 - STRING *type = NULL;

187 - JSONC_PARSE_TXT2STRING_OR_ERROR_AND_RETURN(rule, path, "type", type, error, strict);

188 - if(string_strcmp(type, "template") == 0)

186 + char type[32];

187 + JSONC_PARSE_TXT2CHAR_OR_ERROR_AND_RETURN(rule, path, "type", type, error, strict);

188 + if(strcmp(type, "template") == 0)

189 ap->match.is_template = true;

190 - else if(string_strcmp(type, "instance") == 0)

190 + else if(strcmp(type, "instance") == 0)

191 ap->match.is_template = false;

192 else {

193 - buffer_sprintf(error, "type is '%s', but it can only be 'instance' or 'template'", string2str(type));

193 + buffer_sprintf(error, "type is '%s', but it can only be 'instance' or 'template'", type);

194 return false;

195 }

196

src/health/health_log.c

@@ -246,6 +246,7 @@ inline void health_alarm_log_free_one_nochecks_nounlink(ALARM_ENTRY *ae) {

246 else {

247 string_freez(ae->name);

248 string_freez(ae->chart);

249 + string_freez(ae->chart_name);

250 string_freez(ae->chart_context);

251 string_freez(ae->classification);

252 string_freez(ae->component);

@@ -257,6 +258,7 @@ inline void health_alarm_log_free_one_nochecks_nounlink(ALARM_ENTRY *ae) {

258 string_freez(ae->info);

259 string_freez(ae->old_value_string);

260 string_freez(ae->new_value_string);

261 + string_freez(ae->summary);

262 freez(ae);

263 }

264 }

src/libnetdata/json/json-c-parser-inline.h

+6 -3

@@ -30,9 +30,12 @@

30 if (json_object_object_get_ex(jobj, member, &_j) && json_object_is_type(_j, json_type_string)) { \

31 strncpyz(dst, json_object_get_string(_j), sizeof(dst) - 1); \

32 } \

33 - else if(required) { \

34 - buffer_sprintf(error, "missing or invalid type for '%s.%s' string", path, member); \

35 - return false; \

33 + else { \

34 + dst[0] = '\0'; \

35 + if (required) { \

36 + buffer_sprintf(error, "missing or invalid type for '%s.%s' string", path, member); \

37 + return false; \

38 + } \

39 } \

40 } while(0)

src/libnetdata/string/string.c

+47

@@ -456,6 +456,53 @@ static long unittest_string_entries(void) {

456 return entries;

457 }

458

459 +// returns the number of strings that were freed, but were still referenced

460 +size_t string_destroy(void) {

461 + size_t referenced = 0;

462 +

463 + // Traverse all partitions

464 + for (size_t partition = 0; partition < STRING_PARTITIONS; partition++) {

465 + // Lock the partition to prevent new entries while we're cleaning up

466 + rw_spinlock_write_lock(&string_base[partition].spinlock);

467 +

468 + // Since JudyHS doesn't have simple traversal functions,

469 + // we'll free the entire array at once.

470 + // This is a bit inefficient because we won't be able to

471 + // determine exactly how many strings were referenced.

472 + if (string_base[partition].JudyHSArray) {

473 + // We'll count all entries as "referenced" since we can't check them individually

474 + referenced += string_base[partition].entries;

475 +

476 + // Free the JudyHS array

477 + JudyHSFreeArray(&string_base[partition].JudyHSArray, PJE0);

478 + string_base[partition].JudyHSArray = NULL;

479 + }

480 +

481 + // Reset partition statistics

482 + string_base[partition].inserts = 0;

483 + string_base[partition].deletes = 0;

484 + string_base[partition].entries = 0;

485 + string_base[partition].memory = 0;

486 + string_base[partition].memory_index = 0;

487 +

488 +#ifdef NETDATA_INTERNAL_CHECKS

489 + string_base[partition].atomic.searches = 0;

490 + string_base[partition].atomic.releases = 0;

491 + string_base[partition].atomic.duplications = 0;

492 + string_base[partition].atomic.active_references = 0;

493 + string_base[partition].found_deleted_on_search = 0;

494 + string_base[partition].found_available_on_search = 0;

495 + string_base[partition].found_deleted_on_insert = 0;

496 + string_base[partition].found_available_on_insert = 0;

497 + string_base[partition].spins = 0;

498 +#endif

499 +

500 + rw_spinlock_write_unlock(&string_base[partition].spinlock);

501 + }

502 +

503 + return referenced;

504 +}

505 +

506 #ifdef NETDATA_INTERNAL_CHECKS

507

508 static size_t unittest_string_found_deleted_on_search(void) {

src/libnetdata/string/string.h

@@ -18,6 +18,7 @@ size_t string_strlen(const STRING *string);

18 const char *string2str(const STRING *string) NEVERNULL;

19 bool string_ends_with_string(const STRING *whole, const STRING *end);

20 bool string_starts_with_string(const STRING *whole, const STRING *end);

21 +size_t string_destroy(void);

23 // keep common prefix/suffix and replace everything else with [x]

24 STRING *string_2way_merge(STRING *a, STRING *b);

src/plugins.d/plugins_d.c

+25 -23

@@ -57,28 +57,6 @@ static inline bool plugin_is_running(struct plugind *cd) {

57 return ret;

58 }

60 -static void pluginsd_worker_thread_cleanup(void *pptr) {

61 - struct plugind *cd = CLEANUP_FUNCTION_GET_PTR(pptr);

62 - if(!cd) return;

63 -

64 - worker_unregister();

65 -

66 - spinlock_lock(&cd->unsafe.spinlock);

67 -

68 - cd->unsafe.running = false;

69 - cd->unsafe.thread = 0;

70 -

71 - cd->unsafe.pid = 0;

72 -

73 - POPEN_INSTANCE *pi = cd->unsafe.pi;

74 - cd->unsafe.pi = NULL;

75 -

76 - spinlock_unlock(&cd->unsafe.spinlock);

77 -

78 - if (pi)

79 - spawn_popen_kill(pi, 3 * MSEC_PER_SEC);

80 -}

81 -

60 #define SERIAL_FAILURES_THRESHOLD 10

61 static void pluginsd_worker_thread_handle_success(struct plugind *cd) {

62 if (likely(cd->successful_collections)) {

@@ -143,7 +121,6 @@ static void pluginsd_worker_thread_handle_error(struct plugind *cd, int worker_r

121

122 static void *pluginsd_worker_thread(void *arg) {

123 struct plugind *cd = (struct plugind *) arg;

146 - CLEANUP_FUNCTION_REGISTER(pluginsd_worker_thread_cleanup) cleanup_ptr = cd;

124

125 worker_register("PLUGINSD");

126

@@ -203,6 +180,31 @@ static void *pluginsd_worker_thread(void *arg) {

180 if(unlikely(!plugin_is_enabled(cd)))

181 break;

182 }

183 +

184 + spinlock_lock(&cd->unsafe.spinlock);

185 +

186 + cd->unsafe.running = false;

187 + cd->unsafe.thread = 0;

188 + cd->unsafe.pid = 0;

189 +

190 + POPEN_INSTANCE *pi = cd->unsafe.pi;

191 + cd->unsafe.pi = NULL;

192 +

193 + string_freez(cd->id);

194 + string_freez(cd->cmd);

195 + string_freez(cd->filename);

196 + string_freez(cd->fullfilename);

197 + cd->id = NULL;

198 + cd->cmd = NULL;

199 + cd->filename = NULL;

200 + cd->fullfilename = NULL;

201 +

202 + spinlock_unlock(&cd->unsafe.spinlock);

203 +

204 + if (pi)

205 + spawn_popen_kill(pi, 3 * MSEC_PER_SEC);

206 +

207 + worker_unregister();

208 return NULL;

209 }

210

src/streaming/stream-connector.c

+9 -23

@@ -531,12 +531,15 @@ static void *stream_connector_thread(void *ptr) {

531 worker_register_job_custom_metric(WORKER_SENDER_CONNECTOR_JOB_CANCELLED_NODES, "cancelled nodes", "nodes", WORKER_METRIC_ABSOLUTE);

532

533 unsigned job_id = 0;

534 - while(service_running(SERVICE_STREAMING)) {

535 -

534 + size_t exiting = 0;

535 + while(exiting <= 5) {

536 worker_is_idle();

537 job_id = completion_wait_for_a_job_with_timeout(&sc->completion, job_id, 1000);

538 size_t nodes = 0, connected_nodes = 0, failed_nodes = 0, cancelled_nodes = 0;

539

540 + if(!service_running(SERVICE_STREAMING_CONNECTOR))

541 + exiting++;

542 +

543 spinlock_lock(&sc->queue.spinlock);

544 Word_t idx = 0;

545 for(struct sender_state *s = SENDERS_FIRST(&sc->queue.senders, &idx);

@@ -557,6 +560,7 @@ static void *stream_connector_thread(void *ptr) {

560 spinlock_unlock(&sc->queue.spinlock);

561

562 // do not have the connector lock when calling these

563 + stream_sender_on_disconnect(s);

564 stream_connector_remove(s);

565

566 spinlock_lock(&sc->queue.spinlock);

@@ -564,6 +568,9 @@ static void *stream_connector_thread(void *ptr) {

568 }

569

570 STRCNT_CMD cmd = idx & (STRCNT_CMD_CONNECT| STRCNT_CMD_REMOVE);

571 + if(unlikely(exiting))

572 + cmd = STRCNT_CMD_REMOVE;

573 +

574 switch(cmd) {

575 case STRCNT_CMD_CONNECT:

576 spinlock_unlock(&sc->queue.spinlock);

@@ -619,27 +626,6 @@ static void *stream_connector_thread(void *ptr) {

626 worker_set_metric(WORKER_SENDER_CONNECTOR_JOB_CANCELLED_NODES, (NETDATA_DOUBLE)cancelled_nodes);

627 }

628

622 -#if defined(FSANITIZE_ADDRESS)

623 - // sometimes this thread exits, with localhost still in the queue

624 - sleep(3);

625 -#endif

626 -

627 - spinlock_lock(&sc->queue.spinlock);

628 - Word_t idx = 0;

629 - for(struct sender_state *s = SENDERS_FIRST(&sc->queue.senders, &idx);

630 - s;

631 - s = SENDERS_NEXT(&sc->queue.senders, &idx)) {

632 - SENDERS_DEL(&sc->queue.senders, idx);

633 - spinlock_unlock(&sc->queue.spinlock);

634 -

635 - // do not have the connector lock when calling these

636 - stream_sender_on_disconnect(s);

637 - stream_sender_remove(s, s->exit.reason);

638 -

639 - spinlock_lock(&sc->queue.spinlock);

640 - }

641 - spinlock_unlock(&sc->queue.spinlock);

642 -

629 return NULL;

630 }

631

src/streaming/stream-receiver-connection.c

@@ -99,6 +99,15 @@ void stream_receiver_free(struct receiver_state *rpt) {

99 stream_circular_buffer_destroy(rpt->thread.send_to_child.scb);

100 rpt->thread.send_to_child.scb = NULL;

101

102 + string_freez(rpt->thread.cd.id);

103 + string_freez(rpt->thread.cd.filename);

104 + string_freez(rpt->thread.cd.fullfilename);

105 + string_freez(rpt->thread.cd.cmd);

106 + rpt->thread.cd.id = NULL;

107 + rpt->thread.cd.filename = NULL;

108 + rpt->thread.cd.fullfilename = NULL;

109 + rpt->thread.cd.cmd = NULL;

110 +

111 #ifdef NETDATA_LOG_STREAM_RECEIVER

112 if(rpt->log.fp)

113 fclose(rpt->log.fp);

src/streaming/stream-receiver.c

+3 -3

@@ -449,13 +449,13 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re

449 rpt->thread.cd.id = string_strdupz(buf);

450

451 string_freez(rpt->thread.cd.filename);

452 - rpt->thread.cd.filename = string_strdupz(buf);

452 + rpt->thread.cd.filename = NULL;

453

454 string_freez(rpt->thread.cd.fullfilename);

455 - rpt->thread.cd.fullfilename = string_strdupz(buf);

455 + rpt->thread.cd.fullfilename = NULL;

456

457 string_freez(rpt->thread.cd.cmd);

458 - rpt->thread.cd.cmd = string_strdupz(buf);

458 + rpt->thread.cd.cmd = NULL;

459 }

460

461 rpt->thread.compressed.start = 0;