From 8c69d30e99292d0f53a94e21ed1e88a00a6bf4fd Mon Sep 17 00:00:00 2001 From: Bill Gray Date: Nov 14 2023 13:00:43 +0000 Subject: add support for extended NUMA control mode for multi-CPUs intensive tasks --- diff --git a/numad.c b/numad.c index 25363a2..afcff1e 100644 --- a/numad.c +++ b/numad.c @@ -59,9 +59,10 @@ Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA #define VERSION_STRING "20150602" - -#define VAR_RUN_FILE "/var/run/numad.pid" -#define VAR_LOG_FILE "/var/log/numad.log" +#define VAR_RUN_DIR_ADMIN "/var/run" +#define VAR_LOG_DIR_ADMIN "/var/log" +#define VAR_RUN_FILE "numad.pid" +#define VAR_LOG_FILE "numad.log" #define KILOBYTE (1024) #define MEGABYTE (1024 * 1024) @@ -86,6 +87,8 @@ Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA #define DEFAULT_MEMLOCALITY_PERCENT 90 +#define MIN(a, b) ((a) < (b) ? (a) : (b)) + #define CONVERT_DIGITS_TO_NUM(p, n) \ n = *p++ - '0'; \ while (isdigit(*p)) { \ @@ -93,7 +96,9 @@ Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA n += (*p++ - '0'); \ } +int numa_ctrl_mode = 0; +char var_run_file[BUFSIZ], var_log_file[BUFSIZ]; int num_cpus = 0; int num_nodes = 0; int threads_per_core = 0; @@ -119,6 +124,8 @@ int got_sighup = 0; int got_sigterm = 0; int got_sigquit = 0; +int get_daemon_pid(int inited); + void sig_handler(int signum) { switch (signum) { case SIGHUP: got_sighup = 1; break; @@ -159,11 +166,25 @@ void numad_log(int level, const char *fmt, ...) { fflush(log_fs); } +void init_run_file() { + uid_t uid = getuid(); + gid_t gid = getgid(); + + snprintf(var_run_file, BUFSIZ-1, "%s/%s", VAR_RUN_DIR_ADMIN, VAR_RUN_FILE); + snprintf(var_log_file, BUFSIZ-1, "%s/%s", VAR_LOG_DIR_ADMIN, VAR_LOG_FILE); + if (get_daemon_pid(0) == 0 && uid != 0 && gid != 0) { + snprintf(var_run_file, BUFSIZ-1, "%s/%s", "/tmp", VAR_RUN_FILE); + snprintf(var_log_file, BUFSIZ-1, "%s/%s", "/tmp", VAR_LOG_FILE); + } +} + void open_log_file() { - log_fs = fopen(VAR_LOG_FILE, "a"); + log_fs = fopen(var_log_file, "a"); if (log_fs == NULL) { log_fs = stderr; numad_log(LOG_ERR, "Cannot open numad log file (errno: %d) -- using stderr\n", errno); + } else { + fchmod(fileno(log_fs), S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP ); } } @@ -204,7 +225,7 @@ void flush_msg_queue() { } void init_msg_queue() { - key_t msg_key = 0xdeadbeef; + key_t msg_key = 0xaddaadda; int msg_flg = 0660 | IPC_CREAT; msg_qid = msgget(msg_key, msg_flg); if (msg_qid < 0) { @@ -391,6 +412,7 @@ typedef struct node_data { uint64_t CPUs_total; // scaled * ONE_HUNDRED uint64_t CPUs_free; // scaled * ONE_HUNDRED uint64_t magnitude; // hack: MBs * CPUs + uint8_t CPUs_alloc; uint8_t *distance; id_list_p cpu_list_p; } node_data_t, *node_data_p; @@ -420,6 +442,7 @@ typedef struct process_data { uint64_t data_time_stamp; // hundredths of seconds uint64_t bind_time_stamp; uint64_t num_threads; + uint64_t cpu_threads; uint64_t MBs_size; uint64_t MBs_used; uint64_t cpu_util; @@ -705,7 +728,7 @@ pid_list_p remove_pid_from_pid_list(pid_list_p list_ptr, long pid) { void shut_down_numad() { numad_log(LOG_NOTICE, "Shutting down numad\n"); flush_msg_queue(); - unlink(VAR_RUN_FILE); + unlink(var_run_file); close_log_file(); exit(EXIT_SUCCESS); } @@ -722,6 +745,7 @@ void print_usage_and_exit(char *prog_name) { fprintf(stderr, "-C 1 to count inactive file cache as available memory (default 1)\n"); fprintf(stderr, "-C 0 to count inactive file cache memory as unavailable (default 1)\n"); fprintf(stderr, "-d for debug logging (same effect as '-l 7')\n"); + fprintf(stderr, "-e 0|1 enable extended control mode for cpu intensive task (default 0)\n"); fprintf(stderr, "-h to print this usage info\n"); fprintf(stderr, "-H to set THP scan_sleep_ms (default %d)\n", DEFAULT_THP_SCAN_SLEEP_MS); fprintf(stderr, "-i [:] to specify interval seconds\n"); @@ -776,8 +800,8 @@ void check_prereqs(char *prog_name) { } -int get_daemon_pid() { - int fd = open(VAR_RUN_FILE, O_RDONLY, 0); +int get_daemon_pid(int inited) { + int fd = open(var_run_file, O_RDONLY, 0); if (fd < 0) { return 0; } @@ -794,9 +818,9 @@ int get_daemon_pid() { char fname[FNAME_SIZE]; snprintf(fname, FNAME_SIZE, "/proc/%d", pid); if (access(fname, F_OK) < 0) { - if (errno == ENOENT) { + if (inited && errno == ENOENT) { numad_log(LOG_NOTICE, "Removing out-of-date numad run file because %s doesn't exist\n", fname); - unlink(VAR_RUN_FILE); + unlink(var_run_file); } return 0; } @@ -809,7 +833,7 @@ int register_numad_pid() { char buf[BUF_SIZE]; int fd; create_run_file: - fd = open(VAR_RUN_FILE, O_RDWR|O_CREAT|O_EXCL, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH); + fd = open(var_run_file, O_RDWR|O_CREAT|O_EXCL, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH); if (fd >= 0) { pid = getpid(); sprintf(buf, "%d\n", pid); @@ -819,7 +843,7 @@ create_run_file: return pid; } if (errno == EEXIST) { - fd = open(VAR_RUN_FILE, O_RDWR|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH); + fd = open(var_run_file, O_RDWR|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH); if (fd < 0) { goto fail_numad_run_file; } @@ -835,7 +859,7 @@ create_run_file: if (errno == ENOENT) { // Assume run file is out-of-date... numad_log(LOG_NOTICE, "Removing out-of-date numad run file because %s doesn't exist\n", fname); - unlink(VAR_RUN_FILE); + unlink(var_run_file); goto create_run_file; } } @@ -1426,6 +1450,7 @@ int update_nodes() { node[node_ix].CPUs_free = 0; node[node_ix].magnitude = 0; } + node[node_ix].CPUs_alloc = 0; } avg_node_MBs_free = sum_of_node_MBs_free / num_nodes; avg_node_CPUs_free = sum_of_node_CPUs_free / num_nodes; @@ -1766,7 +1791,9 @@ id_list_p pick_numa_nodes(int pid, int cpus, int mbs, int assume_enough_cpus) { } } fclose(fs); - proc_avg_node_CPUs_free = p->CPUs_used; + if (!numa_ctrl_mode) { + proc_avg_node_CPUs_free = p->CPUs_used; + } for (int ix = 0; (ix <= num_nodes); ix++) { p->process_MBs[ix] /= MEGABYTE; if ((log_level >= LOG_DEBUG) && (p->process_MBs[ix] > 0)) { @@ -1776,11 +1803,15 @@ id_list_p pick_numa_nodes(int pid, int cpus, int mbs, int assume_enough_cpus) { numad_log(LOG_DEBUG, "PROCESS_MBs[%d]: %ld\n", ix, p->process_MBs[ix]); } } - if (ID_IS_IN_LIST(ix, p->node_list_p)) { - proc_avg_node_CPUs_free += node[ix].CPUs_free; + if (!numa_ctrl_mode) { + if (ID_IS_IN_LIST(ix, p->node_list_p)) { + proc_avg_node_CPUs_free += node[ix].CPUs_free; + } } } - proc_avg_node_CPUs_free /= NUM_IDS_IN_LIST(p->node_list_p); + if (!numa_ctrl_mode) { + proc_avg_node_CPUs_free /= NUM_IDS_IN_LIST(p->node_list_p); + } if ((process_has_interleaved_memory) && (keep_interleaved_memory)) { // Mark this process as having interleaved memory so we do not // merge the interleaved memory. Time stamp it as done and return. @@ -1810,17 +1841,24 @@ id_list_p pick_numa_nodes(int pid, int cpus, int mbs, int assume_enough_cpus) { // Add back memory used by this process on this node. tmp_node[ix].MBs_free += ((p->process_MBs[ix] * 17) / 16); // Apply light mem bias // Add back CPU used by this process in proportion to the memory used on this node. - tmp_node[ix].CPUs_free += ((p->CPUs_used * p->process_MBs[ix]) / p->MBs_used); + if (!numa_ctrl_mode) { + tmp_node[ix].CPUs_free += ((p->CPUs_used * p->process_MBs[ix]) / p->MBs_used); + } } else { // If the process is currently running on less than all the // nodes, first add back (biased) memory already used by this // process on this node, then assign average process CPU / node // for this process iff the process is present on this node. tmp_node[ix].MBs_free += ((p->process_MBs[ix] * 5) / 4); // Apply heavy mem bias - if (ID_IS_IN_LIST(ix, p->node_list_p)) { - tmp_node[ix].CPUs_free = proc_avg_node_CPUs_free; + if (!numa_ctrl_mode) { + if (ID_IS_IN_LIST(ix, p->node_list_p)) { + tmp_node[ix].CPUs_free = proc_avg_node_CPUs_free; + } } } + if (numa_ctrl_mode) { + tmp_node[ix].CPUs_free = (tmp_node[ix].CPUs_total - tmp_node[ix].CPUs_alloc * ONE_HUNDRED); + } sum_of_node_CPUs_free += tmp_node[ix].CPUs_free; if (tmp_node[ix].CPUs_free > tmp_node[ix].CPUs_total) { tmp_node[ix].CPUs_free = tmp_node[ix].CPUs_total; @@ -1952,12 +1990,15 @@ id_list_p pick_numa_nodes(int pid, int cpus, int mbs, int assume_enough_cpus) { } // Allocate more resources until request is met. best_node_ix = 0; - while ((min_req_nodes > 0) || (mbs > 0) || ((cpus > cpu_flex) && (!assume_enough_cpus))) { + while (best_node_ix < num_nodes && + ((min_req_nodes > 0) || ( (!numa_ctrl_mode) && mbs > 0) || ((cpus > cpu_flex) && assume_enough_cpus))) { if (log_level >= LOG_DEBUG) { numad_log(LOG_DEBUG, "MBs: %d, CPUs: %d\n", mbs, cpus); } - numad_log(LOG_DEBUG, "Assigning resources from node %d\n", index[best_node_ix]); - ADD_ID_TO_LIST(tmp_node[index[best_node_ix]].node_id, target_node_list_p); + if (tmp_node[index[best_node_ix]].CPUs_free > 10) { + numad_log(LOG_DEBUG, "Assigning resources from node %d\n", index[best_node_ix]); + ADD_ID_TO_LIST(tmp_node[index[best_node_ix]].node_id, target_node_list_p); + } min_req_nodes -= 1; if (EQUAL_LISTS(target_node_list_p, all_nodes_list_p)) { // Apparently we must use all resource nodes... @@ -2031,6 +2072,22 @@ id_list_p pick_numa_nodes(int pid, int cpus, int mbs, int assume_enough_cpus) { return target_node_list_p; } +int check_and_update_loads(int cpu_request, id_list_p node_list_p, int update) { + int cpu_free_nodes, cpu_free_alloc, ix = 0; + int real_nodes = 0, orig_nodes = NUM_IDS_IN_LIST(node_list_p); + for ( ; (ix < num_nodes); ix++) { + if (ID_IS_IN_LIST(ix, node_list_p)) { + cpu_free_nodes = (node[ix].CPUs_total / ONE_HUNDRED) - node[ix].CPUs_alloc; + cpu_free_alloc = MIN(cpu_free_nodes * ONE_HUNDRED, cpu_request); + cpu_request -= cpu_free_alloc; + real_nodes += 1; + if (update) + node[ix].CPUs_alloc += cpu_free_alloc / ONE_HUNDRED; + } + } + + return (cpu_request != 0 || orig_nodes != real_nodes); +} int manage_loads() { uint64_t time_stamp = get_time_stamp(); @@ -2064,6 +2121,18 @@ int manage_loads() { } } } + + // Expand resources needed estimate using target_utilization factor. + // Start with the CPUs actually used (capped by number of threads) for + // CPUs required, and the RSS MBs actually used for the MBs + // requirement, + int mem_target_utilization = target_utilization; + int cpu_target_utilization = target_utilization; + // Cap memory utilization at 100 percent (but allow CPUs to oversubscribe) + if (mem_target_utilization > 100) { + mem_target_utilization = 100; + } + // Order candidate considerations using timestamps and magnitude: amount of // CPU used * amount of memory used. Not expecting a long list here. Use // a simplistic sort -- however move all not yet bound to front of list and @@ -2072,26 +2141,28 @@ int manage_loads() { // Within bins, order by bind_time_stamp so oldest bound will be higher // priority to evaluate. Start by moving all unbound to beginning. int num_unbound = 0; - for (int ij = 0; (ij < nprocs); ij++) { - if (pindex[ij]->bind_time_stamp == 0) { - process_data_p tmp = pindex[num_unbound]; - pindex[num_unbound++] = pindex[ij]; - pindex[ij] = tmp; - } - } - // Sort all unbound so biggest magnitude comes first - for (int ij = 0; (ij < num_unbound); ij++) { - int best = ij; - for (int ik = ij + 1; (ik < num_unbound); ik++) { - uint64_t ik_mag = (pindex[ ik]->CPUs_used * pindex[ ik]->MBs_used); - uint64_t best_mag = (pindex[best]->CPUs_used * pindex[best]->MBs_used); - if (ik_mag <= best_mag) continue; - best = ik; - } - if (best != ij) { - process_data_p tmp = pindex[ij]; - pindex[ij] = pindex[best]; - pindex[best] = tmp; + if (!numa_ctrl_mode) { + for (int ij = 0; (ij < nprocs); ij++) { + if (pindex[ij]->bind_time_stamp == 0) { + process_data_p tmp = pindex[num_unbound]; + pindex[num_unbound++] = pindex[ij]; + pindex[ij] = tmp; + } + } + // Sort all unbound so biggest magnitude comes first + for (int ij = 0; (ij < num_unbound); ij++) { + int best = ij; + for (int ik = ij + 1; (ik < num_unbound); ik++) { + uint64_t ik_mag = (pindex[ ik]->CPUs_used * pindex[ ik]->MBs_used); + uint64_t best_mag = (pindex[best]->CPUs_used * pindex[best]->MBs_used); + if (ik_mag <= best_mag) continue; + best = ik; + } + if (best != ij) { + process_data_p tmp = pindex[ij]; + pindex[ij] = pindex[best]; + pindex[best] = tmp; + } } } // Sort the remaining candidates into bins of increasting magnitude, and by @@ -2136,6 +2207,7 @@ int manage_loads() { } // Estimate desired size (+ margin capacity) and // make resource requests for each candidate process + int need_rebind_node = 0; for (int ix = 0; (ix < nprocs); ix++) { process_data_p p = pindex[ix]; // If this process has interleaved memory, recheck it only every 30 minutes... @@ -2147,16 +2219,6 @@ int manage_loads() { } continue; } - // Expand resources needed estimate using target_utilization factor. - // Start with the CPUs actually used (capped by number of threads) for - // CPUs required, and the RSS MBs actually used for the MBs - // requirement, - int mem_target_utilization = target_utilization; - int cpu_target_utilization = target_utilization; - // Cap memory utilization at 100 percent (but allow CPUs to oversubscribe) - if (mem_target_utilization > 100) { - mem_target_utilization = 100; - } // If the process virtual memory size is bigger than one node, and it // is already using more than 80 percent of a node, then request MBs // based on the virtual size rather than on the current amount in use. @@ -2167,6 +2229,28 @@ int manage_loads() { mb_request = (p->MBs_used * 100) / mem_target_utilization; } int cpu_request = (p->CPUs_used * 100) / cpu_target_utilization; + if (numa_ctrl_mode) { + int req_threads = ((cpu_request + 50) * cpu_target_utilization) / (100 * ONE_HUNDRED); + + // To avoid being stuck in a node with fewer CPUs than actually needed, + // if the number of CPUs currently in use is greater than the number of previously converted CPUs. + // reset the number of CPUs required by the number of threads in the process + int cpu_threads = (p->cpu_threads * ONE_HUNDRED * 100) / cpu_target_utilization; + if (req_threads > p->cpu_threads) { + p->cpu_threads = req_threads; + cpu_request = ((p->num_threads * ONE_HUNDRED) * 100) / cpu_target_utilization; + req_threads = cpu_request / ONE_HUNDRED; + } else if (cpu_threads) { + if (p->cpu_threads > p->num_threads) { + p->cpu_threads = p->num_threads; + cpu_threads = (p->cpu_threads * ONE_HUNDRED * 100) / cpu_target_utilization; + } + cpu_request = cpu_threads; + } + numad_log(LOG_DEBUG, "--> Calculating (num_threads=%d, cpu_threads=%d, req_threads=%d) thread of PID %d\n", + p->num_threads, p->cpu_threads, req_threads, p->pid); + } + // But do not give a process more CPUs than it has threads! int thread_limit = p->num_threads; // If process looks like a KVM guest, try to limit thread count to the @@ -2179,7 +2263,7 @@ int manage_loads() { thread_limit = kvm_vcpu_threads; } } - thread_limit *= ONE_HUNDRED; + thread_limit = (thread_limit * ONE_HUNDRED * 100) / cpu_target_utilization; if (cpu_request > thread_limit) { cpu_request = thread_limit; } @@ -2194,18 +2278,30 @@ int manage_loads() { // it if it looks like it should have a better place with // sufficient resources. FIXME: this is currently implemented for // only smallish processes that will fit in a single node. - if ( ( ID_IS_IN_LIST(min_node_CPUs_free_ix, p->node_list_p) || ID_IS_IN_LIST(min_node_MBs_free_ix, p->node_list_p)) - && (cpu_request < node[0].CPUs_total) && (mb_request < node[0].MBs_total) - && (abs(min_node_CPUs_free + p->CPUs_used - avg_node_CPUs_free) - + abs((max_node_CPUs_free - p->CPUs_used) - avg_node_CPUs_free) - < (max_node_CPUs_free - min_node_CPUs_free) - CPU_THRESHOLD) // CPU slop - && (abs(min_node_MBs_free + p->MBs_used - avg_node_MBs_free) - + abs((max_node_MBs_free - p->MBs_used) - avg_node_MBs_free) - < (max_node_MBs_free - min_node_MBs_free)) ) { + if (numa_ctrl_mode) { + if (!need_rebind_node) { + need_rebind_node = (check_and_update_loads(cpu_request, p->node_list_p, 0) > 0); + } else { + need_rebind_node = 0; + } + } + if ( need_rebind_node || + (( ID_IS_IN_LIST(min_node_CPUs_free_ix, p->node_list_p) || ID_IS_IN_LIST(min_node_MBs_free_ix, p->node_list_p)) + && (cpu_request < node[0].CPUs_total) && (mb_request < node[0].MBs_total) + && (abs(min_node_CPUs_free + p->CPUs_used - avg_node_CPUs_free) + + abs((max_node_CPUs_free - p->CPUs_used) - avg_node_CPUs_free) + < (max_node_CPUs_free - min_node_CPUs_free) - CPU_THRESHOLD) // CPU slop + && (abs(min_node_MBs_free + p->MBs_used - avg_node_MBs_free) + + abs((max_node_MBs_free - p->MBs_used) - avg_node_MBs_free) + < (max_node_MBs_free - min_node_MBs_free))) ) { if (log_level >= LOG_DEBUG) { numad_log(LOG_DEBUG, "Bypassing delay for %d because it looks like it can do better.\n", p->pid); } } else { + if (numa_ctrl_mode) { + check_and_update_loads(cpu_request, p->node_list_p, 1); + } + if (log_level >= LOG_DEBUG) { numad_log(LOG_DEBUG, "Skipping evaluation of PID %d because done too recently.\n", p->pid); } @@ -2216,11 +2312,20 @@ int manage_loads() { pthread_mutex_lock(&node_info_mutex); int assume_enough_cpus = (sum_CPUs_used <= sum_CPUs_total); id_list_p node_list_p = pick_numa_nodes(p->pid, cpu_request, mb_request, assume_enough_cpus); - if ((node_list_p != NULL) && (bind_process_and_migrate_memory(p))) { - pthread_mutex_unlock(&node_info_mutex); - // Return minimum interval when actively moving processes - return min_interval; + if (!numa_ctrl_mode) { + if ((node_list_p != NULL) && (bind_process_and_migrate_memory(p))) { + pthread_mutex_unlock(&node_info_mutex); + // Return minimum interval when actively moving processes + return min_interval; + } + } + if (numa_ctrl_mode) { + if (node_list_p != NULL) { + bind_process_and_migrate_memory(p); + } + check_and_update_loads(cpu_request, p->node_list_p, 1); } + pthread_mutex_unlock(&node_info_mutex); } // Return maximum interval when no process movement @@ -2244,6 +2349,14 @@ void *set_dynamic_options(void *arg) { numad_log(LOG_NOTICE, "Counting inactive file cache as unavailable\n"); } break; + case 'e': + numa_ctrl_mode = (msg.body.arg1 != 0); + if (numa_ctrl_mode) { + numad_log(LOG_NOTICE, "Enable extended NUMA control mode for multi-CPUs intensive task.\n"); + } else { + numad_log(LOG_NOTICE, "Disable extended NUMA control mode for multi-CPUs intensive task.\n"); + } + break; case 'H': thp_scan_sleep_ms = msg.body.arg1; set_thp_scan_sleep_ms(thp_scan_sleep_ms); @@ -2380,6 +2493,7 @@ int main(int argc, char *argv[]) { int opt; int C_flag = 0; int d_flag = 0; + int e_flag = 0; int H_flag = 0; int i_flag = 0; int K_flag = 0; @@ -2395,7 +2509,7 @@ int main(int argc, char *argv[]) { int x_flag = 0; int tmp_int = 0; long list_pid = 0; - while ((opt = getopt(argc, argv, "C:dD:hH:i:K:l:m:p:r:R:S:t:u:vVw:x:")) != -1) { + while ((opt = getopt(argc, argv, "C:dD:e:hH:i:K:l:m:p:r:R:S:t:u:vVw:x:")) != -1) { switch (opt) { case 'C': C_flag = 1; @@ -2408,6 +2522,10 @@ int main(int argc, char *argv[]) { case 'D': // obsoleted break; + case 'e': + e_flag = 1; + numa_ctrl_mode = (atoi(optarg) != 0); + break; case 'h': print_usage_and_exit(argv[0]); break; @@ -2507,13 +2625,14 @@ int main(int argc, char *argv[]) { exit(EXIT_FAILURE); } } + init_run_file(); open_log_file(); init_msg_queue(); num_cpus = get_num_cpus(); page_size_in_bytes = sysconf(_SC_PAGESIZE); huge_page_size_in_bytes = get_huge_page_size_in_bytes(); // Figure out if this is the daemon, or a subsequent invocation - int daemon_pid = get_daemon_pid(); + int daemon_pid = get_daemon_pid(1); if (daemon_pid > 0) { // Daemon is already running. So send dynamic options to persistent // thread to handle requests, get the response (if any), and finish. @@ -2521,6 +2640,9 @@ int main(int argc, char *argv[]) { if (C_flag) { send_msg(daemon_pid, 'C', use_inactive_file_cache, 0, ""); } + if (e_flag) { + send_msg(daemon_pid, 'e', numa_ctrl_mode, 0, ""); + } if (H_flag) { send_msg(daemon_pid, 'H', thp_scan_sleep_ms, 0, ""); }