@cryptotaxi247 / netdata-1 / commits / c97645d6f

optimize workers statistics performance (#14077)

* find the issue with workers performance and optimize it * have 2 independent locks in workers * break down workers statistics work to measure the time of each module * optimize reading cpu for workers; cleanup procfile from unecessary allocations and operations * cleanup workers * fix the workers statistics dimension name

Costa Tsaousis committed Dec 1, 2022 at 17:17 UTC c97645d6f94c178f8c5a6810d60401c99dce4c2f

6 files changed +140 -97

collectors/proc.plugin/ipc.c

+4 -4

@@ -195,7 +195,7 @@ int ipc_msq_get_info(char *msg_filename, struct message_queue **message_queue_ro

195 size_t words = 0;

196

197 if(unlikely(lines < 2)) {

198 - error("Cannot read %s. Expected 2 or more lines, read %zu.", ff->filename, lines);

198 + error("Cannot read %s. Expected 2 or more lines, read %zu.", procfile_filename(ff), lines);

199 return 1;

200 }

201

@@ -205,7 +205,7 @@ int ipc_msq_get_info(char *msg_filename, struct message_queue **message_queue_ro

205 words = procfile_linewords(ff, l);

206 if(unlikely(words < 2)) continue;

207 if(unlikely(words < 14)) {

208 - error("Cannot read %s line. Expected 14 params, read %zu.", ff->filename, words);

208 + error("Cannot read %s line. Expected 14 params, read %zu.", procfile_filename(ff), words);

209 continue;

210 }

211

@@ -250,7 +250,7 @@ int ipc_shm_get_info(char *shm_filename, struct shm_stats *shm) {

250 size_t words = 0;

251

252 if(unlikely(lines < 2)) {

253 - error("Cannot read %s. Expected 2 or more lines, read %zu.", ff->filename, lines);

253 + error("Cannot read %s. Expected 2 or more lines, read %zu.", procfile_filename(ff), lines);

254 return 1;

255 }

256

@@ -263,7 +263,7 @@ int ipc_shm_get_info(char *shm_filename, struct shm_stats *shm) {

263 words = procfile_linewords(ff, l);

264 if(unlikely(words < 2)) continue;

265 if(unlikely(words < 16)) {

266 - error("Cannot read %s line. Expected 16 params, read %zu.", ff->filename, words);

266 + error("Cannot read %s line. Expected 16 params, read %zu.", procfile_filename(ff), words);

267 continue;

268 }

269

daemon/global_statistics.c

+32 -17

@@ -4,14 +4,14 @@

5 #define GLOBAL_STATS_RESET_WEB_USEC_MAX 0x01

7 -#define WORKER_JOB_GLOBAL 0

8 -#define WORKER_JOB_REGISTRY 1

9 -#define WORKER_JOB_WORKERS 2

10 -#define WORKER_JOB_DBENGINE 3

11 -#define WORKER_JOB_HEARTBEAT 4

12 -#define WORKER_JOB_STRINGS 5

13 -#define WORKER_JOB_DICTIONARIES 6

14 -#define WORKER_JOB_MALLOC_TRACE 7

7 +#define WORKER_JOB_GLOBAL 0

8 +#define WORKER_JOB_REGISTRY 1

9 +#define WORKER_JOB_DBENGINE 2

10 +#define WORKER_JOB_HEARTBEAT 3

11 +#define WORKER_JOB_STRINGS 4

12 +#define WORKER_JOB_DICTIONARIES 5

13 +#define WORKER_JOB_MALLOC_TRACE 6

14 +#define WORKER_JOB_WORKERS 7

16 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 8

17 #error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 8

@@ -2447,21 +2447,37 @@ static void workers_utilization_reset_statistics(struct worker_utilization *wu)

2447 }

2448 }

2449

2450 +#define TASK_STAT_PREFIX "/proc/self/task/"

2451 +#define TASK_STAT_SUFFIX "/stat"

2452 +

2453 static int read_thread_cpu_time_from_proc_stat(pid_t pid __maybe_unused, kernel_uint_t *utime __maybe_unused, kernel_uint_t *stime __maybe_unused) {

2454 #ifdef __linux__

2452 - char filename[200 + 1];

2453 - snprintfz(filename, 200, "/proc/self/task/%d/stat", pid);

2455 + static char filename[sizeof(TASK_STAT_PREFIX) + sizeof(TASK_STAT_SUFFIX) + 20] = TASK_STAT_PREFIX;

2456 + static size_t start_pos = sizeof(TASK_STAT_PREFIX) - 1;

2457 + static procfile *ff = NULL;

2458 +

2459 + // construct the filename

2460 + size_t end_pos = snprintfz(&filename[start_pos], 20, "%d", pid);

2461 + strcpy(&filename[start_pos + end_pos], TASK_STAT_SUFFIX);

2462

2455 - procfile *ff = procfile_open(filename, " ", PROCFILE_FLAG_NO_ERROR_ON_FILE_IO);

2456 - if(!ff) return -1;

2463 + // (re)open the procfile to the new filename

2464 + bool set_quotes = (ff == NULL) ? true : false;

2465 + ff = procfile_reopen(ff, filename, NULL, PROCFILE_FLAG_DEFAULT);

2466 + if(unlikely(!ff)) return -1;

2467

2468 + if(set_quotes)

2469 + procfile_set_open_close(ff, "(", ")");

2470 +

2471 + // read the entire file and split it to lines and words

2472 ff = procfile_readall(ff);

2459 - if(!ff) return -1;

2473 + if(unlikely(!ff)) return -1;

2474

2475 + // parse the numbers we are interested

2476 *utime = str2kernel_uint_t(procfile_lineword(ff, 0, 13));

2477 *stime = str2kernel_uint_t(procfile_lineword(ff, 0, 14));

2478

2464 - procfile_close(ff);

2479 + // leave the file open for the next iteration

2480 +

2481 return 0;

2482 #else

2483 // TODO: add here cpu time detection per thread, for FreeBSD and MacOS

@@ -2621,8 +2637,7 @@ static void worker_utilization_charts(void) {

2637 static size_t iterations = 0;

2638 iterations++;

2639

2624 - int i;

2625 - for(i = 0; all_workers_utilization[i].name ;i++) {

2640 + for(int i = 0; all_workers_utilization[i].name ;i++) {

2641 workers_utilization_reset_statistics(&all_workers_utilization[i]);

2642 workers_foreach(all_workers_utilization[i].name, worker_utilization_charts_callback, &all_workers_utilization[i]);

2643

@@ -2672,11 +2687,11 @@ static void global_statistics_register_workers(void) {

2687 worker_register("STATS");

2688 worker_register_job_name(WORKER_JOB_GLOBAL, "global");

2689 worker_register_job_name(WORKER_JOB_REGISTRY, "registry");

2675 - worker_register_job_name(WORKER_JOB_WORKERS, "workers");

2690 worker_register_job_name(WORKER_JOB_DBENGINE, "dbengine");

2691 worker_register_job_name(WORKER_JOB_STRINGS, "strings");

2692 worker_register_job_name(WORKER_JOB_DICTIONARIES, "dictionaries");

2693 worker_register_job_name(WORKER_JOB_MALLOC_TRACE, "malloc_trace");

2694 + worker_register_job_name(WORKER_JOB_WORKERS, "workers");

2695 }

2696

2697 static void global_statistics_cleanup(void *ptr)

libnetdata/procfile/procfile.c

+32 -35

@@ -22,16 +22,21 @@ size_t procfile_max_allocation = PROCFILE_INCREMENT_BUFFER;

22 // ----------------------------------------------------------------------------

24 char *procfile_filename(procfile *ff) {

25 - if(ff->filename[0]) return ff->filename;

25 + if(ff->filename)

26 + return ff->filename;

28 + char filename[FILENAME_MAX + 1];

29 char buffer[FILENAME_MAX + 1];

30 snprintfz(buffer, FILENAME_MAX, "/proc/self/fd/%d", ff->fd);

30 - ssize_t l = readlink(buffer, ff->filename, FILENAME_MAX);

32 + ssize_t l = readlink(buffer, filename, FILENAME_MAX);

33 if(unlikely(l == -1))

32 - snprintfz(ff->filename, FILENAME_MAX, "unknown filename for fd %d", ff->fd);

34 + snprintfz(filename, FILENAME_MAX, "unknown filename for fd %d", ff->fd);

35 else

34 - ff->filename[l] = '\0';

36 + filename[l] = '\0';

37 +

38 +

39 + ff->filename = strdupz(filename);

41 // on non-linux systems, something like this will be needed

42 // fcntl(ff->fd, F_GETPATH, ff->filename)

@@ -141,8 +146,9 @@ void procfile_close(procfile *ff) {

146

147 debug(D_PROCFILE, PF_PREFIX ": Closing file '%s'", procfile_filename(ff));

148

144 - if(likely(ff->lines)) procfile_lines_free(ff->lines);

145 - if(likely(ff->words)) procfile_words_free(ff->words);

149 + freez(ff->filename);

150 + procfile_lines_free(ff->lines);

151 + procfile_words_free(ff->words);

152

153 if(likely(ff->fd != -1)) close(ff->fd);

154 freez(ff);

@@ -319,40 +325,31 @@ procfile *procfile_readall(procfile *ff) {

325 return ff;

326 }

327

322 -NOINLINE

323 -static void procfile_set_separators(procfile *ff, const char *separators) {

324 - static PF_CHAR_TYPE def[256];

325 - static char initialized = 0;

326 -

327 - if(unlikely(!initialized)) {

328 - // this is thread safe

329 - // if initialized is zero, multiple threads may be executing

330 - // this code at the same time, setting in def[] the exact same values

331 - int i = 256;

332 - while(i--) {

333 - if(unlikely(i == '\n' || i == '\r'))

334 - def[i] = PF_CHAR_IS_NEWLINE;

328 +static PF_CHAR_TYPE procfile_default_separators[256];

329 +__attribute__((constructor)) void procfile_initialize_default_separators(void) {

330 + int i = 256;

331 + while(i--) {

332 + if(unlikely(i == '\n' || i == '\r'))

333 + procfile_default_separators[i] = PF_CHAR_IS_NEWLINE;

334

336 - else if(unlikely(isspace(i) || !isprint(i)))

337 - def[i] = PF_CHAR_IS_SEPARATOR;

335 + else if(unlikely(isspace(i) || !isprint(i)))

336 + procfile_default_separators[i] = PF_CHAR_IS_SEPARATOR;

337

339 - else

340 - def[i] = PF_CHAR_IS_WORD;

341 - }

342 -

343 - initialized = 1;

338 + else

339 + procfile_default_separators[i] = PF_CHAR_IS_WORD;

340 }

341 +}

342

346 - // copy the default

347 - PF_CHAR_TYPE *ffs = ff->separators, *ffd = def, *ffe = &def[256];

348 - while(ffd != ffe)

349 - *ffs++ = *ffd++;

350 -

343 +NOINLINE

344 +static void procfile_set_separators(procfile *ff, const char *separators) {

345 // set the separators

346 if(unlikely(!separators))

347 separators = " \t=|";

348

355 - ffs = ff->separators;

349 + // copy the default

350 + memcpy(ff->separators, procfile_default_separators, 256 * sizeof(PF_CHAR_TYPE));

351 +

352 + PF_CHAR_TYPE *ffs = ff->separators;

353 const char *s = separators;

354 while(*s)

355 ffs[(int)*s++] = PF_CHAR_IS_SEPARATOR;

@@ -416,8 +413,7 @@ procfile *procfile_open(const char *filename, const char *separators, uint32_t f

413 procfile *ff = mallocz(sizeof(procfile) + size);

414

415 //strncpyz(ff->filename, filename, FILENAME_MAX);

419 - ff->filename[0] = '\0';

420 -

416 + ff->filename = NULL;

417 ff->fd = fd;

418 ff->size = size;

419 ff->len = 0;

@@ -449,7 +445,8 @@ procfile *procfile_reopen(procfile *ff, const char *filename, const char *separa

445 // info("PROCFILE: opened '%s' on fd %d", filename, ff->fd);

446

447 //strncpyz(ff->filename, filename, FILENAME_MAX);

452 - ff->filename[0] = '\0';

448 + freez(ff->filename);

449 + ff->filename = NULL;

450 ff->flags = flags;

451

452 // do not do the separators again if NULL is given

libnetdata/procfile/procfile.h

+7 -8

@@ -37,7 +37,7 @@ typedef struct {

37 #define PROCFILE_FLAG_DEFAULT 0x00000000

38 #define PROCFILE_FLAG_NO_ERROR_ON_FILE_IO 0x00000001

40 -typedef enum procfile_separator {

40 +typedef enum __attribute__ ((__packed__)) procfile_separator {

41 PF_CHAR_IS_SEPARATOR,

42 PF_CHAR_IS_NEWLINE,

43 PF_CHAR_IS_WORD,

@@ -46,17 +46,16 @@ typedef enum procfile_separator {

46 PF_CHAR_IS_CLOSE

47 } PF_CHAR_TYPE;

49 -typedef struct {

50 - char filename[FILENAME_MAX + 1]; // not populated until profile_filename() is called

51 -

49 +typedef struct procfile {

50 + char *filename; // not populated until procfile_filename() is called

51 uint32_t flags;

53 - int fd; // the file descriptor

54 - size_t len; // the bytes we have placed into data

55 - size_t size; // the bytes we have allocated for data

52 + int fd; // the file descriptor

53 + size_t len; // the bytes we have placed into data

54 + size_t size; // the bytes we have allocated for data

55 pflines *lines;

56 pfwords *words;

57 PF_CHAR_TYPE separators[256];

59 - char data[]; // allocated buffer to keep file contents

58 + char data[]; // allocated buffer to keep file contents

59 } procfile;

61 // close the proc file and free all related memory

libnetdata/worker_utilization/worker_utilization.c

+63 -31

@@ -44,35 +44,55 @@ struct worker {

44 struct worker *prev;

45 };

47 -static netdata_mutex_t workers_base_lock = NETDATA_MUTEX_INITIALIZER;

48 -static __thread struct worker *worker = NULL;

49 -static Pvoid_t workers_per_workname_JudyHS_array = NULL;

47 +struct workers_workname { // this is what we add to JudyHS

48 + SPINLOCK spinlock;

49 + struct worker *base;

50 +};

51 +

52 +static struct workers_globals {

53 + SPINLOCK spinlock;

54 + Pvoid_t worknames_JudyHS;

55 +

56 +} workers_globals = { // workers globals, the base of all worknames

57 + .spinlock = NETDATA_SPINLOCK_INITIALIZER, // a lock for the worknames index

58 + .worknames_JudyHS = NULL, // the worknames index

59 +};

51 -void worker_register(const char *workname) {

61 +static __thread struct worker *worker = NULL; // the current thread worker

62 +

63 +void worker_register(const char *name) {

64 if(unlikely(worker)) return;

66 worker = callocz(1, sizeof(struct worker));

67 worker->pid = gettid();

68 worker->tag = strdupz(netdata_thread_tag());

57 - worker->workname = strdupz(workname);

69 + worker->workname = strdupz(name);

71 usec_t now = now_monotonic_usec();

72 worker->statistics_last_checkpoint = now;

73 worker->last_action_timestamp = now;

74 worker->last_action = WORKER_IDLE;

64 - size_t workname_size = strlen(workname) + 1;

65 - netdata_mutex_lock(&workers_base_lock);

76 + size_t name_size = strlen(name) + 1;

77 + netdata_spinlock_lock(&workers_globals.spinlock);

67 - Pvoid_t *PValue = JudyHSGet(workers_per_workname_JudyHS_array, (void *)workname, workname_size);

79 + Pvoid_t *PValue = JudyHSGet(workers_globals.worknames_JudyHS, (void *)name, name_size);

80 if(!PValue)

69 - PValue = JudyHSIns(&workers_per_workname_JudyHS_array, (void *)workname, workname_size, PJE0);

81 + PValue = JudyHSIns(&workers_globals.worknames_JudyHS, (void *)name, name_size, PJE0);

82 +

83 + struct workers_workname *workname = *PValue;

84 + if(!workname) {

85 + workname = mallocz(sizeof(struct workers_workname));

86 + workname->spinlock = NETDATA_SPINLOCK_INITIALIZER;

87 + workname->base = NULL;

88 + *PValue = workname;

89 + }

71 - struct worker *base = *PValue;

72 - DOUBLE_LINKED_LIST_APPEND_UNSAFE(base, worker, prev, next);

73 - *PValue = base;

91 + netdata_spinlock_lock(&workname->spinlock);

92 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(workname->base, worker, prev, next);

93 + netdata_spinlock_unlock(&workname->spinlock);

75 - netdata_mutex_unlock(&workers_base_lock);

95 + netdata_spinlock_unlock(&workers_globals.spinlock);

96 }

98 void worker_register_job_custom_metric(size_t job_id, const char *name, const char *units, WORKER_METRIC_TYPE type) {

@@ -105,17 +125,20 @@ void worker_unregister(void) {

125 if(unlikely(!worker)) return;

126

127 size_t workname_size = strlen(worker->workname) + 1;

108 - netdata_mutex_lock(&workers_base_lock);

109 - Pvoid_t *PValue = JudyHSGet(workers_per_workname_JudyHS_array, (void *)worker->workname, workname_size);

128 + netdata_spinlock_lock(&workers_globals.spinlock);

129 + Pvoid_t *PValue = JudyHSGet(workers_globals.worknames_JudyHS, (void *)worker->workname, workname_size);

130 if(PValue) {

111 - struct worker *base = *PValue;

112 - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(base, worker, prev, next);

113 - *PValue = base;

114 -

115 - if(!base)

116 - JudyHSDel(&workers_per_workname_JudyHS_array, (void *)worker->workname, workname_size, PJE0);

131 + struct workers_workname *workname = *PValue;

132 + netdata_spinlock_lock(&workname->spinlock);

133 + DOUBLE_LINKED_LIST_REMOVE_UNSAFE(workname->base, worker, prev, next);

134 + netdata_spinlock_unlock(&workname->spinlock);

135 +

136 + if(!workname->base) {

137 + JudyHSDel(&workers_globals.worknames_JudyHS, (void *) worker->workname, workname_size, PJE0);

138 + freez(workname);

139 + }

140 }

118 - netdata_mutex_unlock(&workers_base_lock);

141 + netdata_spinlock_unlock(&workers_globals.spinlock);

142

143 for(int i = 0; i < WORKER_UTILIZATION_MAX_JOB_TYPES ;i++) {

144 string_freez(worker->per_job_type[i].name);

@@ -187,7 +210,7 @@ void worker_set_metric(size_t job_id, NETDATA_DOUBLE value) {

210

211 // statistics interface

212

190 -void workers_foreach(const char *workname, void (*callback)(

213 +void workers_foreach(const char *name, void (*callback)(

214 void *data

215 , pid_t pid

216 , const char *thread_tag

@@ -203,18 +226,27 @@ void workers_foreach(const char *workname, void (*callback)(

226 , NETDATA_DOUBLE *job_custom_values

227 )

228 , void *data) {

206 - netdata_mutex_lock(&workers_base_lock);

229 + netdata_spinlock_lock(&workers_globals.spinlock);

230 usec_t busy_time, delta;

231 size_t i, jobs_started, jobs_running;

232

210 - size_t workname_size = strlen(workname) + 1;

211 - struct worker *base = NULL;

212 - Pvoid_t *PValue = JudyHSGet(workers_per_workname_JudyHS_array, (void *)workname, workname_size);

213 - if(PValue)

214 - base = *PValue;

233 + size_t workname_size = strlen(name) + 1;

234 + struct workers_workname *workname;

235 + Pvoid_t *PValue = JudyHSGet(workers_globals.worknames_JudyHS, (void *)name, workname_size);

236 + if(PValue) {

237 + workname = *PValue;

238 + netdata_spinlock_lock(&workname->spinlock);

239 + }

240 + else

241 + workname = NULL;

242 +

243 + netdata_spinlock_unlock(&workers_globals.spinlock);

244 +

245 + if(!workname)

246 + return;

247

248 struct worker *p;

217 - DOUBLE_LINKED_LIST_FOREACH_FORWARD(base, p, prev, next) {

249 + DOUBLE_LINKED_LIST_FOREACH_FORWARD(workname->base, p, prev, next) {

250 usec_t now = now_monotonic_usec();

251

252 // find per job type statistics

@@ -326,5 +358,5 @@ void workers_foreach(const char *workname, void (*callback)(

358 );

359 }

360

329 - netdata_mutex_unlock(&workers_base_lock);

361 + netdata_spinlock_unlock(&workname->spinlock);

362 }

libnetdata/worker_utilization/worker_utilization.h

+2 -2

@@ -15,7 +15,7 @@ typedef enum {

15 WORKER_METRIC_INCREMENTAL_TOTAL = 4,

16 } WORKER_METRIC_TYPE;

18 -void worker_register(const char *workname);

18 +void worker_register(const char *name);

19 void worker_register_job_name(size_t job_id, const char *name);

20 void worker_register_job_custom_metric(size_t job_id, const char *name, const char *units, WORKER_METRIC_TYPE type);

21 void worker_unregister(void);

@@ -26,7 +26,7 @@ void worker_set_metric(size_t job_id, NETDATA_DOUBLE value);

27 // statistics interface

29 -void workers_foreach(const char *workname, void (*callback)(

29 +void workers_foreach(const char *name, void (*callback)(

30 void *data

31 , pid_t pid

32 , const char *thread_tag