@cryptotaxi247 / netdata-1 / commits / 6ffa84229

use nd threads in exporting (#20212)

* use nd threads in exporting * fix thread names * fix thread names again * fix use of mutex and conditional variables

Costa Tsaousis committed Apr 30, 2025 at 16:42 UTC 6ffa84229877d54fb2e32cd07b4331be0e3e539e

11 files changed +42 -34

src/exporting/aws_kinesis/aws_kinesis.c

+2 -5

@@ -94,16 +94,12 @@ int init_aws_kinesis_instance(struct instance *instance)

94 *

95 * @param instance_p an instance data structure.

96 */

97 -void aws_kinesis_connector_worker(void *instance_p)

97 +void *aws_kinesis_connector_worker(void *instance_p)

98 {

99 struct instance *instance = (struct instance *)instance_p;

100 struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;

101 struct aws_kinesis_specific_data *connector_specific_data = instance->connector_specific_data;

102

103 - char threadname[ND_THREAD_TAG_MAX + 1];

104 - snprintfz(threadname, ND_THREAD_TAG_MAX, "EXPKNSS[%zu]", instance->index);

105 - uv_thread_set_name_np(threadname);

106 -

103 while (!instance->engine->exit) {

104 unsigned long long partition_key_seq = 0;

105 struct stats *stats = &instance->stats;

@@ -220,4 +216,5 @@ void aws_kinesis_connector_worker(void *instance_p)

216 }

217

218 aws_kinesis_cleanup(instance);

219 + return NULL;

220 }

src/exporting/aws_kinesis/aws_kinesis.h

+1 -1

@@ -11,6 +11,6 @@

11 #define KINESIS_RECORD_MAX 1024 * 1024

13 int init_aws_kinesis_instance(struct instance *instance);

14 -void aws_kinesis_connector_worker(void *instance_p);

14 +void *aws_kinesis_connector_worker(void *instance_p);

16 #endif //NETDATA_EXPORTING_KINESIS_H

src/exporting/exporting_engine.c

+5 -3

@@ -138,9 +138,11 @@ static void exporting_main_cleanup(void *pptr)

138

139 if (!instance->exited) {

140 netdata_log_info("EXPORTING: signaling worker '%s' to stop...", instance->config.name);

141 - uv_mutex_unlock(&instance->mutex);

141 + // Lock the mutex before signaling the condition variable

142 + uv_mutex_lock(&instance->mutex);

143 instance->data_is_ready = 1;

144 uv_cond_signal(&instance->cond_var);

145 + uv_mutex_unlock(&instance->mutex);

146 }

147 else

148 netdata_log_info("EXPORTING: found worker '%s' already stopped", instance->config.name);

@@ -157,8 +159,8 @@ static void exporting_main_cleanup(void *pptr)

159 exited++;

160

161 if(instance->thread) {

160 - uv_thread_join(&instance->thread);

161 - instance->thread = 0;

162 + nd_thread_join(instance->thread);

163 + instance->thread = NULL;

164 }

165 }

166 else if(iterations % 100 == 0)

src/exporting/exporting_engine.h

+4 -3

@@ -74,6 +74,7 @@ struct instance_config {

74 const char *prefix;

75 const char *label_prefix;

76 const char *hostname;

77 + const char *thread_tag;

79 int update_every;

80 int buffer_on_failures;

@@ -195,7 +196,7 @@ struct stats {

196 struct instance {

197 struct instance_config config;

198 void *buffer;

198 - void (*worker)(void *instance_p);

199 + void *(*worker)(void *instance_p);

200 struct stats stats;

201

202 int scheduled;

@@ -208,7 +209,7 @@ struct instance {

209 time_t after;

210 time_t before;

211

211 - uv_thread_t thread;

212 + ND_THREAD *thread;

213 uv_mutex_t mutex;

214 uv_cond_t cond_var;

215 int data_is_ready;

@@ -290,7 +291,7 @@ int exporting_discard_response(BUFFER *buffer, struct instance *instance);

291 void simple_connector_receive_response(int *sock, struct instance *instance);

292 void simple_connector_send_buffer(

293 int *sock, int *failures, struct instance *instance, BUFFER *header, BUFFER *buffer, size_t buffered_metrics);

293 -void simple_connector_worker(void *instance_p);

294 +void *simple_connector_worker(void *instance_p);

295

296 void create_main_rusage_chart(RRDSET **st_rusage, RRDDIM **rd_user, RRDDIM **rd_system);

297 void send_main_rusage(RRDSET *st_rusage, RRDDIM *rd_user, RRDDIM *rd_system);

src/exporting/init_connectors.c

+19 -5

@@ -34,67 +34,81 @@ int init_connectors(struct engine *engine)

34 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {

35 instance->index = engine->instance_num++;

36 instance->after = engine->now;

37 -

37 +

38 + // Set thread tag and initialize connector based on type

39 switch (instance->config.type) {

40 case EXPORTING_CONNECTOR_TYPE_GRAPHITE:

41 + instance->config.thread_tag = "EXPGRPH";

42 if (init_graphite_instance(instance) != 0)

43 return 1;

44 break;

45 case EXPORTING_CONNECTOR_TYPE_GRAPHITE_HTTP:

46 + instance->config.thread_tag = "EXPGRPH";

47 if (init_graphite_instance(instance) != 0)

48 return 1;

49 break;

50 case EXPORTING_CONNECTOR_TYPE_JSON:

51 + instance->config.thread_tag = "EXPJSON";

52 if (init_json_instance(instance) != 0)

53 return 1;

54 break;

55 case EXPORTING_CONNECTOR_TYPE_JSON_HTTP:

56 + instance->config.thread_tag = "EXPJSON";

57 if (init_json_http_instance(instance) != 0)

58 return 1;

59 break;

60 case EXPORTING_CONNECTOR_TYPE_OPENTSDB:

61 + instance->config.thread_tag = "EXPTSDB";

62 if (init_opentsdb_telnet_instance(instance) != 0)

63 return 1;

64 break;

65 case EXPORTING_CONNECTOR_TYPE_OPENTSDB_HTTP:

66 + instance->config.thread_tag = "EXPTSDB";

67 if (init_opentsdb_http_instance(instance) != 0)

68 return 1;

69 break;

70 case EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE:

71 + instance->config.thread_tag = "EXPPRW";

72 #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE

73 if (init_prometheus_remote_write_instance(instance) != 0)

74 return 1;

75 #endif

76 break;

77 case EXPORTING_CONNECTOR_TYPE_KINESIS:

78 + instance->config.thread_tag = "EXPKINS";

79 #if HAVE_KINESIS

80 if (init_aws_kinesis_instance(instance) != 0)

81 return 1;

82 #endif

83 break;

84 case EXPORTING_CONNECTOR_TYPE_PUBSUB:

85 + instance->config.thread_tag = "EXPPUBS";

86 #if ENABLE_EXPORTING_PUBSUB

87 if (init_pubsub_instance(instance) != 0)

88 return 1;

89 #endif

90 break;

91 case EXPORTING_CONNECTOR_TYPE_MONGODB:

92 + instance->config.thread_tag = "EXPMNG";

93 #ifdef HAVE_MONGOC

94 if (init_mongodb_instance(instance) != 0)

95 return 1;

96 #endif

97 break;

98 default:

99 + instance->config.thread_tag = "EXPCON";

100 netdata_log_error("EXPORTING: unknown exporting connector type");

101 return 1;

102 }

103

104 // dispatch the instance worker thread

93 - int error = uv_thread_create(&instance->thread, instance->worker, instance);

94 - if (error) {

95 - netdata_log_error("EXPORTING: cannot create thread worker. uv_thread_create(): %s", uv_strerror(error));

105 + char threadname[ND_THREAD_TAG_MAX + 1];

106 + snprintfz(threadname, ND_THREAD_TAG_MAX, "%s[%zu]", instance->config.thread_tag, instance->index);

107 +

108 + instance->thread = nd_thread_create(threadname, NETDATA_THREAD_OPTION_JOINABLE, instance->worker, instance);

109 + if (!instance->thread) {

110 + netdata_log_error("EXPORTING: cannot create thread worker for instance %s", instance->config.name);

111 instance->exited = 1;

97 - instance->thread = 0;

112 return 1;

113 }

114

src/exporting/mongodb/mongodb.c

+2 -5

@@ -276,7 +276,7 @@ void mongodb_cleanup(struct instance *instance)

276 *

277 * @param instance_p an instance data structure.

278 */

279 -void mongodb_connector_worker(void *instance_p)

279 +void *mongodb_connector_worker(void *instance_p)

280 {

281 struct instance *instance = (struct instance *)instance_p;

282 #ifdef NETDATA_INTERNAL_CHECKS

@@ -285,10 +285,6 @@ void mongodb_connector_worker(void *instance_p)

285 struct mongodb_specific_data *connector_specific_data =

286 (struct mongodb_specific_data *)instance->connector_specific_data;

287

288 - char threadname[ND_THREAD_TAG_MAX + 1];

289 - snprintfz(threadname, ND_THREAD_TAG_MAX, "EXPMNG[%zu]", instance->index);

290 - uv_thread_set_name_np(threadname);

291 -

288 while (!instance->engine->exit) {

289 struct stats *stats = &instance->stats;

290

@@ -393,4 +389,5 @@ void mongodb_connector_worker(void *instance_p)

389 }

390

391 mongodb_cleanup(instance);

392 + return NULL;

393 }

src/exporting/mongodb/mongodb.h

+1 -1

@@ -30,6 +30,6 @@ void mongodb_cleanup(struct instance *instance);

31 int init_mongodb_instance(struct instance *instance);

32 int format_batch_mongodb(struct instance *instance);

33 -void mongodb_connector_worker(void *instance_p);

33 +void *mongodb_connector_worker(void *instance_p);

35 #endif //NETDATA_EXPORTING_MONGODB_H

src/exporting/process_data.c

+1 -1

@@ -316,9 +316,9 @@ void end_batch_formatting(struct engine *engine)

316 disable_instance(instance);

317 continue;

318 }

319 - uv_mutex_unlock(&instance->mutex);

319 instance->data_is_ready = 1;

320 uv_cond_signal(&instance->cond_var);

321 + uv_mutex_unlock(&instance->mutex);

322

323 instance->scheduled = 0;

324 instance->after = instance->before;

src/exporting/pubsub/pubsub.c

+2 -5

@@ -93,16 +93,12 @@ void clean_pubsub_instance(struct instance *instance)

93 *

94 * @param instance_p an instance data structure.

95 */

96 -void pubsub_connector_worker(void *instance_p)

96 +void *pubsub_connector_worker(void *instance_p)

97 {

98 struct instance *instance = (struct instance *)instance_p;

99 struct pubsub_specific_config *connector_specific_config = instance->config.connector_specific_config;

100 struct pubsub_specific_data *connector_specific_data = instance->connector_specific_data;

101

102 - char threadname[ND_THREAD_TAG_MAX + 1];

103 - snprintfz(threadname, ND_THREAD_TAG_MAX, "EXPPBSB[%zu]", instance->index);

104 - uv_thread_set_name_np(threadname);

105 -

102 while (!instance->engine->exit) {

103 struct stats *stats = &instance->stats;

104 char error_message[ERROR_LINE_MAX + 1] = "";

@@ -196,4 +192,5 @@ void pubsub_connector_worker(void *instance_p)

192 }

193

194 clean_pubsub_instance(instance);

195 + return NULL;

196 }

src/exporting/pubsub/pubsub.h

+1 -1

@@ -9,6 +9,6 @@

10 int init_pubsub_instance(struct instance *instance);

11 void clean_pubsub_instance(struct instance *instance);

12 -void pubsub_connector_worker(void *instance_p);

12 +void *pubsub_connector_worker(void *instance_p);

14 #endif //NETDATA_EXPORTING_PUBSUB_H

src/exporting/send_data.c

+4 -4

@@ -205,14 +205,12 @@ void simple_connector_send_buffer(

205 *

206 * @param instance_p an instance data structure.

207 */

208 -void simple_connector_worker(void *instance_p)

208 +void *simple_connector_worker(void *instance_p)

209 {

210 struct instance *instance = (struct instance*)instance_p;

211 struct simple_connector_data *connector_specific_data = instance->connector_specific_data;

212

213 - char threadname[ND_THREAD_TAG_MAX + 1];

214 - snprintfz(threadname, ND_THREAD_TAG_MAX, "EXPSMPL[%zu]", instance->index);

215 - uv_thread_set_name_np(threadname);

213 + // Thread name is set during creation

214

215 uint32_t options = (uint32_t)instance->config.options;

216

@@ -392,4 +390,6 @@ void simple_connector_worker(void *instance_p)

390 #endif

391

392 simple_connector_cleanup(instance);

393 +

394 + return NULL;

395 }