2
\$\begingroup\$

I am writing a program in an environment that makes use of cgroups to identify and group processes together. I want to parse the CPU utilization of each cgroup by sampling /sys/fs/cgroup/cpuacct/.../cpuacct.stat over a 1 second interval a few times and then averaging them. It also calculates the overall CPU utilization of the system by sampling /proc/stat in the same manner.

I want to do this in a multithreaded way because if I did it serially, the samples would happen serially, and the program would take a while to run.

I also use PCRE to distinguish between two different process types.

#include <sys/types.h>
#include <sys/wait.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <semaphore.h>
#include <pthread.h>
#include <sys/uio.h>
#include <pcre.h>

#define ITERATIONS 3

#if !defined(CLK_TIME_PER_SECOND) && defined(_SC_CLK_TCK)
#  define CLK_TIME_PER_SECOND ((int) sysconf (_SC_CLK_TCK))
#endif
#if !defined(CLK_TIME_PER_SECOND) && defined(CLK_TCK)
#  define CLK_TIME_PER_SECOND ((int) CLK_TCK)
#endif
#if !defined(CLK_TIME_PER_SECOND) && defined(CLOCKS_PER_SEC)
#  define CLK_TIME_PER_SECOND ((int) CLOCKS_PER_SEC)
#endif
#if !defined(CLK_TIME_PER_SECOND)
#  define CLK_TIME_PER_SECOND 100
#endif

struct job_struct {
    char procid[50]; //store job information as procid,procType tuple
    char procType[50];
};

typedef struct {
    char *procid; //for passing to the threads
    char *procType;
} sample_struct;

int sumFlag, clk; //TODO: localize?
long double sum1, sum2, systemUtilization, otherCPU;
char *aStrRegex;
const char *pcreErrorStr;
int pcreErrorOffset, pcreExecRet;
int subStrVec[30];
pcre *reCompiled;
pcre_extra *pcreExtra;

pthread_mutex_t total_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t ok_to_add = PTHREAD_COND_INITIALIZER;

int sampleCGroup(char procid[], char procType[]);

void *addSampletoTotal();

void *overallUtilization();

long double getSystemUtilization(void);

int getOSVersion(void);

int getCPUCount(void);

int main(void) {
    int cpus, i, jobCount, osVersion;
    FILE *fp;
    char buff[1024];

    cpus = getCPUCount();
    clk = CLK_TIME_PER_SECOND;

    osVersion = getOSVersion();
    if (osVersion <= 10) {
        systemUtilization = getSystemUtilization();
        printf("sys=%Lf\n", systemUtilization); //cgroups are not supported on certain OS versions, so print the overall utilization and exit, otherwise, try to run
        pthread_mutex_destroy(&total_lock); //clean up mutex lock
        pthread_cond_destroy(&ok_to_add);
        return (0);
    }

    struct job_struct jobs[cpus]; //initially size the array to be the number of CPUs

    sumFlag = 1; //condition variable for mutex

    pthread_t systr; //thread id for overall system utilization, note this should only execute on certain versions, the conditional above should have returned if not
    pthread_create(&systr, NULL, overallUtilization, NULL); //create the thread

    aStrRegex = "proc_type1"; //type1
    reCompiled = pcre_compile(aStrRegex, 0, &pcreErrorStr, &pcreErrorOffset, NULL); //compile the regex
    if (reCompiled == NULL) {
        printf("ERROR: Could not compile '%s': %s\n", aStrRegex, pcreErrorStr);
        pthread_join(systr, NULL); //join the system utilization thread
        if (systemUtilization > (long double) cpus) { //sanity check for HT
            systemUtilization = (long double) cpus;
        }
        printf("sys=%Lf\n", systemUtilization); //print info
        printf("cgrType2=%f\n", 0.0);
        printf("cgrType1=%f\n", 0.0);
        printf("cgrOther=%Lf\n", systemUtilization);
        pthread_mutex_destroy(&total_lock); //clean up mutex lock
        pthread_cond_destroy(&ok_to_add);
        return (-1);
    }
    // Optimize the regex
    pcreExtra = pcre_study(reCompiled, 0, &pcreErrorStr);
    /* pcre_study() returns NULL for both errors and when it can not optimize the regex.  The last argument is how one checks for
     errors (it is NULL if everything works, and points to an error string otherwise. */
    if (pcreErrorStr != NULL) {
        printf("ERROR: Could not study '%s': %s\n", aStrRegex, pcreErrorStr);
        pthread_join(systr, NULL); //join the system utilization thread
        if (systemUtilization > (long double) cpus) { //sanity check for HT
            systemUtilization = (long double) cpus;
        }
        printf("sys=%Lf\n", systemUtilization); //print info
        printf("cgrType2=%f\n", 0.0);
        printf("cgrType1=%f\n", 0.0);
        printf("cgrOther=%Lf\n", systemUtilization);
        pthread_mutex_destroy(&total_lock); //clean up mutex lock
        pthread_cond_destroy(&ok_to_add);
        return (-1);
    } /* end if */

    fp = popen("/var/bin/getProcs", "r"); //open getProcs to list the procs, this returns the cgroup ID and the type
    if (fp == NULL) {
        perror("/var/bin/getProcs");
        pthread_join(systr, NULL); //join the system utilization thread
        if (systemUtilization > (long double) cpus) { //sanity check for HT
            systemUtilization = (long double) cpus;
        }
        printf("sys=%Lf\n", systemUtilization); //print info
        printf("cgrType2=%f\n", 0.0);
        printf("cgrType1=%f\n", 0.0);
        printf("cgrOther=%Lf\n", systemUtilization);
        pthread_mutex_destroy(&total_lock); //clean up mutex lock
        pthread_cond_destroy(&ok_to_add);
        return (-1);
    }

    jobCount = 0;
    while (fgets(buff, sizeof(buff), fp) != NULL) {
        sscanf(buff, "%[^,],%[^,]", jobs[jobCount].procid, jobs[jobCount].procType);
        jobCount++;
    }

    if (jobCount == 0) {
        pthread_join(systr, NULL); //join the system utilization thread
        if (systemUtilization > (long double) cpus) {
            systemUtilization = (long double) cpus;
        }
        printf("sys=%Lf\n", systemUtilization); //print info
        printf("cgrType2=%f\n", 0.0);
        printf("cgrType1=%f\n", 0.0);
        printf("cgrOther=%Lf\n", systemUtilization);
        pclose(fp); //close the command
        pthread_mutex_destroy(&total_lock); //destroy mutex lock
        pthread_cond_destroy(&ok_to_add);
        return (0);
    }

    pthread_t tid[jobCount]; // the thread identifiers, one for each job

    for (i = 0; i < jobCount; i++) {
        sample_struct *args = malloc(sizeof *args); //allocate memory for passing the argument struct to the threads
        args->procid = jobs[i].procid; //assign info in job array to the arg struct
        args->procType = jobs[i].procType;
        pthread_create(&tid[i], NULL, addSampletoTotal, args); //create the thread
    }

    pthread_join(systr, NULL); //join the system utilization thread
    for (i = 0; i < jobCount; i++) {
        pthread_join(tid[i], NULL); //finish threads
    }

    pthread_mutex_destroy(&total_lock); //destroy mutex lock
    pthread_cond_destroy(&ok_to_add);

    if (systemUtilization > (long double) cpus) { //sanity check for HT
        systemUtilization = (long double) cpus;
    }
    if (sum2 > (long double) cpus) {
        sum2 = (long double) cpus;
    }
    if (sum1 > (long double) cpus) {
        sum1 = (long double) cpus;
    }

    otherCPU = systemUtilization - sum2 - sum1;

    if (otherCPU < 0.0) {
        otherCPU = 0.0;
    }

    printf("sys=%Lf\n", systemUtilization); //print info
    printf("cgrType2=%Lf\n", sum2);
    printf("cgrType1=%Lf\n", sum1);
    printf("cgrOther=%Lf\n", otherCPU);
    pclose(fp); //close the command
    return (0);
}

int sampleCGroup(char procid[], char procType[]) {
    //open the cgroup cpustat file for the job based on the procid arg
    char *pidPath = malloc(
            strlen("/sys/fs/cgroup/cpuacct/cpuacct.stat/") + strlen(procid) + 1);//+1 for the zero-terminator
    if (pidPath == NULL) {
        perror("Could not allocate memory for str\n");
        return (-1);
    } else {
        strcpy(pidPath, "/sys/fs/cgroup/cpuacct/"); //create the path to the cgroup info
        strcat(pidPath, procid);
        strcat(pidPath, "/cpuacct.stat");
    }

    long double a[2], b[2];
    long double loadavg = 0.0;
    int i;
    FILE *fp;

    for (i = 0; i < ITERATIONS; i++) {
        fp = fopen(pidPath, "r"); //first sample
        if (fp == NULL) {
            perror(pidPath);
            return (-1);
        } else {
            if (fscanf(fp, "%*4s %Lf %*6s %Lf", &a[0], &a[1]) != 2) {
                perror("Could not parse first cgroup CPU sample!\n");
                return (-1);
            }
            fclose(fp);
            sleep(1); //let 1s go by
        }

        fp = fopen(pidPath, "r"); //second sample
        if (fp == NULL) {
            perror(pidPath);
            return (-1);
        } else {
            if (fscanf(fp, "%*4s %Lf %*6s %Lf", &b[0], &b[1]) != 2) {
                perror("Could not parse second cgroup CPU sample!\n");
                return (-1);
            }
            fclose(fp);
        }

        loadavg += ((b[0] + b[1]) - (a[0] + a[1])); //delta
    }

    loadavg = (loadavg / (clk * ITERATIONS)); //calculate the CPU use by delta divided by clock ticks per one second times iterations
    free(pidPath); //free the memory allocated to the path variable

    //CRITICAL SECTION BEGIN
    pthread_mutex_lock(&total_lock); //lock mutex
    while (sumFlag == 0) {
        pthread_cond_wait(&ok_to_add, &total_lock); //wait on ok to add signal
    }

    pcreExecRet = pcre_exec(reCompiled,
                            pcreExtra,
                            procType,
                            strlen(procType),          // length of string
                            0,                      // Start looking at this point
                            0,                      // OPTIONS
                            subStrVec,
                            30);                    // Length of subStrVec

    if (pcreExecRet > 0) {
        sum1 += loadavg;
    } else if (pcreExecRet == PCRE_ERROR_NOMATCH) {
        sum2 += loadavg;
    } else {
        sum2 += loadavg; //assume type2 if this fails
        perror("Could not determine type!\n");
    }

    sumFlag = 1;

    pthread_cond_signal(&ok_to_add); //signal that it is ok to add
    pthread_mutex_unlock(&total_lock); //unlock mutex
    //CRITICAL SECTION END

    return 0; //success
}

void *addSampletoTotal(void *args) {
    sample_struct *actual_args = args; //get the jobs argument struct
    sampleCGroup(actual_args->procid, actual_args->procType); //sample the cgroup for the job passed to the thread
    free(actual_args); //free the memory allocated to args
    pthread_exit(0); //exit thread
}

void *overallUtilization() {
    systemUtilization = getSystemUtilization(); //sample the cgroup for the job passed to the thread
    pthread_exit(0); //exit thread
}

long double getSystemUtilization(void) {
    long double a[3], b[3];
    long double loadavg = 0.0;
    int i;
    FILE *fp;

    /* /proc/stat columns:
     * 1st column : user = normal processes executing in user mode
     * 2nd column : nice = niced processes executing in user mode
     * 3rd column : system = processes executing in kernel mode
     * 4th column : idle = twiddling thumbs
     * 5th column : iowait = waiting for I/O to complete
     * 6th column : irq = servicing interrupts
     * 7th column : softirq = servicing softirqs
     * */

    for (i = 0; i < ITERATIONS; i++) {
        fp = fopen("/proc/stat", "r"); //first sample
        if (fp == NULL) {
            perror("/proc/stat");
            return (-1);
        } else {
            if (fscanf(fp, "%*s %Lf %Lf %Lf", &a[0], &a[1], &a[2]) != 3 ) {
                perror("Could not parse first system CPU sample!\n");
                return (-1);
            }
            fclose(fp);
            sleep(1); //let 1s go by
        }

        fp = fopen("/proc/stat", "r"); //second sample
        if (fp == NULL) {
            perror("/proc/stat");
            return (-1);
        } else {
            if (fscanf(fp, "%*s %Lf %Lf %Lf", &b[0], &b[1], &b[2]) != 3) {
                perror("Could not parse second system CPU sample!\n");
                return (-1);
            }
            fclose(fp);
        }

        loadavg += ((b[0] + b[1] + b[2]) - (a[0] + a[1] + a[2]));  //delta
    }

    loadavg = (loadavg / (clk * ITERATIONS)); //calculate the CPU use by delta divided by clock ticks per one second times iterations
    return loadavg;
}

int getOSVersion(void) {
    FILE *os;
    int version, checkVersion, patchLevel;
    os = fopen("/etc/SuSE-release", "r"); //first sample
    if (os == NULL) {
        perror("/etc/SuSE-release");
        return (-1);
    } else {
        if (fscanf(os, "%*s %*s %*s %*s %d %*s VERSION = %d PATCHLEVEL = %d", &version, &checkVersion, &patchLevel) != 3) {
            perror("Unable to read SuSE release");
            return -1;
        }
        fclose(os);
    }
    if (version == checkVersion) {
        return version;
    } else {
        return -1;
    }
}

int getCPUCount(void) {
#if defined _SC_NPROCESSORS_ONLN
    {
            long int nprocs = sysconf (_SC_NPROCESSORS_ONLN);
            if (nprocs > 0)
            return nprocs;
        }
#elif HAVE_SCHED_GETAFFINITY_LIKE_GLIBC /* glibc >= 2.3.4 */
    {
        cpu_set_t set;

        if (sched_getaffinity (0, sizeof (set), &set) == 0)
          {
            unsigned long count;

# ifdef CPU_COUNT
            /* glibc >= 2.6 has the CPU_COUNT macro.  */
            count = CPU_COUNT (&set);
# else
            size_t i;

            count = 0;
            for (i = 0; i < CPU_SETSIZE; i++)
              if (CPU_ISSET (i, &set))
                count++;
# endif
            if (count > 0)
              return count;
          }
      }
#endif

    return 1;
}

In particular I want to be sure this is safe multithreading.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.