X-Git-Url: https://arthur.barton.de/gitweb/?a=blobdiff_plain;f=src%2Fplugins_d.c;h=7fe46e46f18feac90a817975cd807334885f401e;hb=655715342478c5da5d43c5a8e619bd214585db4d;hp=0f3b8bc6450cd7fbfb296a4a7f991bd85e7c5852;hpb=6911559e3af6b73b7be8fb1171b4645121ebe1ab;p=netdata.git diff --git a/src/plugins_d.c b/src/plugins_d.c index 0f3b8bc6..7fe46e46 100644 --- a/src/plugins_d.c +++ b/src/plugins_d.c @@ -87,10 +87,12 @@ static int pluginsd_split_words(char *str, char **words, int max_words) { void *pluginsd_worker_thread(void *arg) { struct plugind *cd = (struct plugind *)arg; + cd->obsolete = 0; + char line[PLUGINSD_LINE_MAX + 1]; #ifdef DETACH_PLUGINS_FROM_NETDATA - unsigned long long usec = 0, susec = 0; + usec_t usec = 0, susec = 0; struct timeval last = {0, 0} , now = {0, 0}; #endif @@ -109,7 +111,7 @@ void *pluginsd_worker_thread(void *arg) size_t count = 0; - while(likely(1)) { + for(;;) { if(unlikely(netdata_exit)) break; FILE *fp = mypopen(cd->cmd, &cd->pid); @@ -121,7 +123,6 @@ void *pluginsd_worker_thread(void *arg) info("PLUGINSD: '%s' running on pid %d", cd->fullfilename, cd->pid); RRDSET *st = NULL; - char *s; uint32_t hash; while(likely(fgets(line, PLUGINSD_LINE_MAX, fp) != NULL)) { @@ -132,7 +133,7 @@ void *pluginsd_worker_thread(void *arg) // debug(D_PLUGINSD, "PLUGINSD: %s: %s", cd->filename, line); int w = pluginsd_split_words(line, words, MAX_WORDS); - s = words[0]; + char *s = words[0]; if(unlikely(!s || !*s || !w)) { // debug(D_PLUGINSD, "PLUGINSD: empty line"); continue; @@ -142,7 +143,7 @@ void *pluginsd_worker_thread(void *arg) hash = simple_hash(s); - if(likely(hash == SET_HASH && !strcmp(s, "SET"))) { + if(likely(hash == SET_HASH && !strsame(s, "SET"))) { char *dimension = words[1]; char *value = words[2]; @@ -166,7 +167,7 @@ void *pluginsd_worker_thread(void *arg) if(value) rrddim_set(st, dimension, strtoll(value, NULL, 0)); } - else if(likely(hash == BEGIN_HASH && !strcmp(s, "BEGIN"))) { + else if(likely(hash == BEGIN_HASH && !strsame(s, "BEGIN"))) { char *id = words[1]; char *microseconds_txt = words[2]; @@ -186,13 +187,13 @@ void *pluginsd_worker_thread(void *arg) } if(likely(st->counter_done)) { - unsigned long long microseconds = 0; - if(microseconds_txt && *microseconds_txt) microseconds = strtoull(microseconds_txt, NULL, 10); + usec_t microseconds = 0; + if(microseconds_txt && *microseconds_txt) microseconds = str2ull(microseconds_txt); if(microseconds) rrdset_next_usec(st, microseconds); - else rrdset_next_plugins(st); + else rrdset_next(st); } } - else if(likely(hash == END_HASH && !strcmp(s, "END"))) { + else if(likely(hash == END_HASH && !strsame(s, "END"))) { if(unlikely(!st)) { error("PLUGINSD: '%s' is requesting an END, without a BEGIN. Disabling it.", cd->fullfilename); cd->enabled = 0; @@ -207,15 +208,15 @@ void *pluginsd_worker_thread(void *arg) count++; } - else if(likely(hash == FLUSH_HASH && !strcmp(s, "FLUSH"))) { + else if(likely(hash == FLUSH_HASH && !strsame(s, "FLUSH"))) { debug(D_PLUGINSD, "PLUGINSD: '%s' is requesting a FLUSH", cd->fullfilename); st = NULL; } - else if(likely(hash == CHART_HASH && !strcmp(s, "CHART"))) { + else if(likely(hash == CHART_HASH && !strsame(s, "CHART"))) { int noname = 0; st = NULL; - if((words[1]) != NULL && (words[2]) != NULL && strcmp(words[1], words[2]) == 0) + if((words[1]) != NULL && (words[2]) != NULL && strsame(words[1], words[2]) == 0) noname = 1; char *type = words[1]; @@ -241,10 +242,10 @@ void *pluginsd_worker_thread(void *arg) } int priority = 1000; - if(likely(priority_s)) priority = atoi(priority_s); + if(likely(priority_s)) priority = str2i(priority_s); int update_every = cd->update_every; - if(likely(update_every_s)) update_every = atoi(update_every_s); + if(likely(update_every_s)) update_every = str2i(update_every_s); if(unlikely(!update_every)) update_every = cd->update_every; int chart_type = RRDSET_TYPE_LINE; @@ -271,7 +272,7 @@ void *pluginsd_worker_thread(void *arg) } else debug(D_PLUGINSD, "PLUGINSD: Chart '%s' already exists. Not adding it again.", st->id); } - else if(likely(hash == DIMENSION_HASH && !strcmp(s, "DIMENSION"))) { + else if(likely(hash == DIMENSION_HASH && !strsame(s, "DIMENSION"))) { char *id = words[1]; char *name = words[2]; char *algorithm = words[3]; @@ -325,40 +326,40 @@ void *pluginsd_worker_thread(void *arg) } else if(unlikely(st->debug)) debug(D_PLUGINSD, "PLUGINSD: dimension %s/%s already exists. Not adding it again.", st->id, id); } - else if(unlikely(hash == DISABLE_HASH && !strcmp(s, "DISABLE"))) { - error("PLUGINSD: '%s' called DISABLE. Disabling it.", cd->fullfilename); + else if(unlikely(hash == DISABLE_HASH && !strsame(s, "DISABLE"))) { + info("PLUGINSD: '%s' called DISABLE. Disabling it.", cd->fullfilename); cd->enabled = 0; killpid(cd->pid, SIGTERM); break; } #ifdef DETACH_PLUGINS_FROM_NETDATA - else if(likely(hash == MYPID_HASH && !strcmp(s, "MYPID"))) { + else if(likely(hash == MYPID_HASH && !strsame(s, "MYPID"))) { char *pid_s = words[1]; pid_t pid = strtod(pid_s, NULL, 0); if(likely(pid)) cd->pid = pid; debug(D_PLUGINSD, "PLUGINSD: %s is on pid %d", cd->id, cd->pid); } - else if(likely(hash == STOPPING_WAKE_ME_UP_PLEASE_HASH && !strcmp(s, "STOPPING_WAKE_ME_UP_PLEASE"))) { + else if(likely(hash == STOPPING_WAKE_ME_UP_PLEASE_HASH && !strsame(s, "STOPPING_WAKE_ME_UP_PLEASE"))) { error("PLUGINSD: '%s' (pid %d) called STOPPING_WAKE_ME_UP_PLEASE.", cd->fullfilename, cd->pid); - gettimeofday(&now, NULL); + now_realtime_timeval(&now); if(unlikely(!usec && !susec)) { // our first run - susec = cd->rrd_update_every * 1000000ULL; + susec = cd->rrd_update_every * USEC_PER_SEC; } else { // second+ run - usec = usecdiff(&now, &last) - susec; + usec = dt_usec(&now, &last) - susec; error("PLUGINSD: %s last loop took %llu usec (worked for %llu, sleeped for %llu).\n", cd->fullfilename, usec + susec, usec, susec); - if(unlikely(usec < (rrd_update_every * 1000000ULL / 2ULL))) susec = (rrd_update_every * 1000000ULL) - usec; - else susec = rrd_update_every * 1000000ULL / 2ULL; + if(unlikely(usec < (rrd_update_every * USEC_PER_SEC / 2ULL))) susec = (rrd_update_every * USEC_PER_SEC) - usec; + else susec = rrd_update_every * USEC_PER_SEC / 2ULL; } error("PLUGINSD: %s sleeping for %llu. Will kill with SIGCONT pid %d to wake it up.\n", cd->fullfilename, susec, cd->pid); usleep(susec); killpid(cd->pid, SIGCONT); - bcopy(&now, &last, sizeof(struct timeval)); + memmove(&last, &now, sizeof(struct timeval)); break; } #endif @@ -380,16 +381,9 @@ void *pluginsd_worker_thread(void *arg) // get the return code int code = mypclose(fp, cd->pid); - - if(netdata_exit) { - cd->pid = 0; - cd->enabled = 0; - cd->obsolete = 1; - pthread_exit(NULL); - return NULL; - } - - if(code != 0) { + + if(unlikely(netdata_exit)) break; + else if(code != 0) { // the plugin reports failure if(likely(!cd->successful_collections)) { @@ -401,7 +395,7 @@ void *pluginsd_worker_thread(void *arg) // we have collected something if(likely(cd->serial_failures <= 10)) { - error("PLUGINSD: '%s' exited with error code %d, but has given useful output in the past (%zu times). Waiting a bit before starting it again.", cd->fullfilename, code, cd->successful_collections); + error("PLUGINSD: '%s' exited with error code %d, but has given useful output in the past (%zu times). %s", cd->fullfilename, code, cd->successful_collections, cd->enabled?"Waiting a bit before starting it again.":"Will not start it again - it is disabled."); sleep((unsigned int) (cd->update_every * 10)); } else { @@ -417,7 +411,7 @@ void *pluginsd_worker_thread(void *arg) // we have collected nothing so far if(likely(cd->serial_failures <= 10)) { - error("PLUGINSD: '%s' (pid %d) does not generate useful output but it reports success (exits with 0). Waiting a bit before starting it again.", cd->fullfilename, cd->pid); + error("PLUGINSD: '%s' (pid %d) does not generate useful output but it reports success (exits with 0). %s.", cd->fullfilename, cd->pid, cd->enabled?"Waiting a bit before starting it again.":"Will not start it again - it is disabled."); sleep((unsigned int) (cd->update_every * 10)); } else { @@ -430,17 +424,19 @@ void *pluginsd_worker_thread(void *arg) } cd->pid = 0; - if(unlikely(!cd->enabled)) - break; + if(unlikely(!cd->enabled)) break; } + info("PLUGINSD: '%s' thread exiting", cd->fullfilename); + cd->obsolete = 1; + cd->thread = (pthread_t)NULL; pthread_exit(NULL); return NULL; } void *pluginsd_main(void *ptr) { - (void)ptr; + struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr; info("PLUGINS.D thread created with task id %d", gettid()); @@ -462,14 +458,13 @@ void *pluginsd_main(void *ptr) { if(scan_frequency < 1) scan_frequency = 1; - while(likely(1)) { + for(;;) { if(unlikely(netdata_exit)) break; dir = opendir(dir_name); if(unlikely(!dir)) { error("Cannot open directory '%s'.", dir_name); - pthread_exit(NULL); - return NULL; + goto cleanup; } while(likely((file = readdir(dir)))) { @@ -477,11 +472,11 @@ void *pluginsd_main(void *ptr) { debug(D_PLUGINSD, "PLUGINSD: Examining file '%s'", file->d_name); - if(unlikely(strcmp(file->d_name, ".") == 0 || strcmp(file->d_name, "..") == 0)) continue; + if(unlikely(strsame(file->d_name, ".") == 0 || strsame(file->d_name, "..") == 0)) continue; int len = (int) strlen(file->d_name); if(unlikely(len <= (int)PLUGINSD_FILE_SUFFIX_LEN)) continue; - if(unlikely(strcmp(PLUGINSD_FILE_SUFFIX, &file->d_name[len - (int)PLUGINSD_FILE_SUFFIX_LEN]) != 0)) { + if(unlikely(strsame(PLUGINSD_FILE_SUFFIX, &file->d_name[len - (int)PLUGINSD_FILE_SUFFIX_LEN]) != 0)) { debug(D_PLUGINSD, "PLUGINSD: File '%s' does not end in '%s'.", file->d_name, PLUGINSD_FILE_SUFFIX); continue; } @@ -496,9 +491,9 @@ void *pluginsd_main(void *ptr) { } // check if it runs already - for(cd = pluginsd_root ; likely(cd) ; cd = cd->next) { - if(unlikely(strcmp(cd->filename, file->d_name) == 0)) break; - } + for(cd = pluginsd_root ; cd ; cd = cd->next) + if(unlikely(strsame(cd->filename, file->d_name) == 0)) break; + if(likely(cd && !cd->obsolete)) { debug(D_PLUGINSD, "PLUGINSD: plugin '%s' is already running", cd->filename); continue; @@ -516,7 +511,7 @@ void *pluginsd_main(void *ptr) { cd->enabled = enabled; cd->update_every = (int) config_get_number(cd->id, "update every", rrd_update_every); - cd->started_t = time(NULL); + cd->started_t = now_realtime_sec(); char *def = ""; snprintfz(cd->cmd, PLUGINSD_CMD_MAX, "exec %s %d %s", cd->fullfilename, cd->update_every, config_get(cd->id, "command options", def)); @@ -524,24 +519,29 @@ void *pluginsd_main(void *ptr) { // link it if(likely(pluginsd_root)) cd->next = pluginsd_root; pluginsd_root = cd; - } - cd->obsolete = 0; - if(unlikely(!cd->enabled)) continue; - - // spawn a new thread for it - if(unlikely(pthread_create(&cd->thread, NULL, pluginsd_worker_thread, cd) != 0)) { - error("PLUGINSD: failed to create new thread for plugin '%s'.", cd->filename); + // it is not currently running cd->obsolete = 1; + + if(cd->enabled) { + // spawn a new thread for it + if(unlikely(pthread_create(&cd->thread, NULL, pluginsd_worker_thread, cd) != 0)) + error("PLUGINSD: failed to create new thread for plugin '%s'.", cd->filename); + + else if(unlikely(pthread_detach(cd->thread) != 0)) + error("PLUGINSD: Cannot request detach of newly created thread for plugin '%s'.", cd->filename); + } } - else if(unlikely(pthread_detach(cd->thread) != 0)) - error("PLUGINSD: Cannot request detach of newly created thread for plugin '%s'.", cd->filename); } closedir(dir); sleep((unsigned int) scan_frequency); } +cleanup: + info("PLUGINS.D thread exiting"); + + static_thread->enabled = 0; pthread_exit(NULL); return NULL; }