test CFS fairness and latency

gistfile1.txt

#include <fcntl.h>
#include <errno.h>
#include <getopt.h>
#include <poll.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>
#include <string>
#include <sys/types.h>
#include <sys/syscall.h>

#if !defined(__i386__) && !defined(__x86_64__)
#error    "Arch is not supported!"
#endif

template <class T>
inline T atomic_load(volatile T* ptr) {
return *ptr;
}

template <class T>
inline void atomic_store(volatile T* ptr, T v) {
*ptr = v;
}

long g_max_delay_time;
int g_num_of_eat_cpu_threads = 32;
long g_delay_ms = 1000;    // 1000ms

using std::string;

static inline long current_time() {
struct timeval now;
gettimeofday(&now, NULL);
return now.tv_sec * 1000000L + now.tv_usec;
}

class Pipe {
int _fds[2];

// disable evil copy constructors
Pipe(const Pipe&);
Pipe& operator=(const Pipe&);

int set_nonblock(int fd) {
    long l = fcntl(fd, F_GETFL);
    if (l & O_NONBLOCK) {
        return 0;
    }
    return fcntl(fd, F_SETFL, l | O_NONBLOCK);
}

public:
Pipe() {
}

bool init() {
    int ret = pipe(_fds);
    if (ret != 0) {
        printf("create pipe failed, err: %s\n", strerror(errno));
        return false;
    } else {
        set_nonblock(_fds[1]);
        set_nonblock(_fds[0]);
        return true;
    }
}

bool write_bytes(const string& buffer) {
    size_t send_bytes = 0;
    while (send_bytes < buffer.size()) {
        ssize_t s = write(_fds[1], buffer.data() + send_bytes, buffer.size() - send_bytes);
        if (s < 0) {
            fprintf(stderr, "write to pipe failed %s\n", strerror(errno));
            return false;
        }
        send_bytes += s;
    }
    return true;
}

bool read_bytes(string* buffer, int timeout_ms) {
    struct pollfd fds[1];
    fds[0].fd = _fds[0];
    fds[0].events = POLLIN;
    long start = current_time();
    poll(fds, 1, timeout_ms);
    long ecapsed = current_time() - start - timeout_ms * 1000L;
    if (ecapsed > 10000L) { // 10ms
        printf("Exceeded deadline %d ms\n", ecapsed / 1000);
    }
    if (fds[0].revents & POLLIN) {
        // flush the pipe
        char b[128];
        while (true) {
            ssize_t s = read(_fds[0], b, sizeof(b));
            if (s > 0) {
                buffer->append(b, s);
            }
            if (s < 0 || s != sizeof(b)) {
                break;
            }
        }
    }
    return !buffer->empty();
}

};

void* writer_routine(void* arg) {
Pipe* pipe = (Pipe*) arg;
while (true) {
    const int kInts = 512;
    char b[kInts * sizeof(int)];
    for (int i = 0; i < kInts; i += sizeof(int)) {
        *(int*)&b[i] = rand();
    }
    pipe->write_bytes(string(b, sizeof(b)));
    sleep(1);
}
return NULL;
}

void* reader_routine(void* arg) {
Pipe* pipe = (Pipe*) arg;
while (true) {
    //long start = current_time();
    string buffer;
    pipe->read_bytes(&buffer, 1000);
    //printf("out: len %lu, time: %lu\n", buffer.size(), current_time() - start);
}
return NULL;
}

void* eat_cpu(void*) {
long seed = 0x5eed;
long last_time = current_time();
while (true) {
    // take 10ms
    for (int i = 0; i < (1 << 16) / 5; i++) {
        seed += seed * 31 ^ 0x123456789L;
        seed *= 11;
        seed /= 17;
        seed -= 0xabcdef ^ 0x987654321L;
    }
    long now = current_time();
    if (now - last_time > atomic_load(&g_max_delay_time)) {
        printf("%d delay %d ms\n", syscall(__NR_gettid), (now - last_time) / 1000);
        atomic_store(&g_max_delay_time, now - last_time);
    }
    last_time = current_time();
}
return NULL;
}

// BUILD: g++ -lpthread -o pipe_test pipe_test.cpp
int main(int argc, char** argv) {
    const char* const short_opts = "n:d:h";
    const option long_opts[] = {
        {"num_of_threads", 1, 0, 'n'},
        {"delay_ms", 1, 0, 'd'},
        {"help", 1, 0, 'h'},
        {0, 0, 0, 0}
    };

    while (true) {
        int c = getopt_long(argc, argv, short_opts, long_opts, NULL);
        if (c == -1) {
            break;
        }
        switch (c) {
            case 'n':
                g_num_of_eat_cpu_threads = atoi(optarg);
                break;
            case 'd':
                g_delay_ms = atoi(optarg);
                break;
            case 'h':
                printf("%s --num_of_threads <> --delay_ms <>\n", argv[0]);
                return -1;
            default:
                break;
        }
    }

    g_max_delay_time = g_delay_ms * 1000L;
    printf("eat cpu with %ld threads, delay time %ld ms\n", g_num_of_eat_cpu_threads, g_delay_ms);

    Pipe pipe;
    pipe.init();

    pthread_t writer_tid;
    pthread_create(&writer_tid, NULL, writer_routine, &pipe);

    pthread_t reader_tid;
    pthread_create(&reader_tid, NULL, reader_routine, &pipe);

    for (int i = 0; i < g_num_of_eat_cpu_threads - 1; i++) {
        pthread_t tid;
        pthread_create(&tid, NULL, eat_cpu, NULL);
    }
    eat_cpu(NULL);
    return 0;
}