2018年3月17日 星期六

An Sample Code for threads control - A Wrap for service/thread start/stop/periodical run


multi-thread programming設計很常見, 寫法上與功能上也都很相似, 最基本的幾個操作就是, create/start/stop/periodical run/destroy等等. 最好還能monitor這些thread狀況, 比如被執行幾次, 執行時間多久, 是否有dead lock等等, 於是就寫了這個pattern, 提供這類功能.

service-reg.h

#ifndef SERVICE_REG_H
#define SERVICE_REG_H
#include <sys/time.h>

typedef void *(*srv_fp) (void *);

char * srv_dump(char *buf, int sz);

int _srv_reg(char *srv_name, srv_fp fp, char *fp_name, void *srv_data);
#define srv_reg(srv_name, fp, srv_data) _srv_reg(srv_name, fp, #fp, srv_data)
int srv_start(char *srv_name); // run forever without any delay
int srv_stop(char *srv_name);
int srv_unreg(char *srv_name);
int srv_start_periodical(char *srv_name, unsigned long sec, unsigned long nsec); // run forever with delay

#define SRV_NO_MEM -1

#endif


service-reg.c

#include <pthread.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <ctype.h>
#include <time.h>
#include <sys/time.h>
#include <sys/timerfd.h>
#include <stdint.h>
#include <unistd.h>
#include "service-reg.h"
#include "lookup_table.h"

#define SEC2NSEC 1000000000.0

struct srv_info {
    void *srv_data;
    char *srv_name;
    srv_fp fp;
    int in_fp;
    char *fp_name;
    unsigned run_cnt;
    pthread_t thread_id; /* ID returned by pthread_create() */
    pthread_cond_t cond;
    pthread_mutex_t mutex;
    struct srv_info *next;
    volatile int running; /* 1 for running, 0 for stop */
    struct timespec active_time;
    struct timespec last_enter;
    struct timespec last_exit;
    int tm_fd; /* used to delay */
};

struct srv_tab {
    pthread_mutex_t mutex;
    struct srv_info *list;
    unsigned int num_srv;
} srv_tab = {
    .mutex = PTHREAD_MUTEX_INITIALIZER,
    .list = NULL,
    .num_srv = 0,
};

static void _srv_update_active_time(struct srv_info *s)
{
    clock_gettime(CLOCK_MONOTONIC, &(s->last_exit));

    s->active_time.tv_sec += s->last_exit.tv_sec - s->last_enter.tv_sec;
    if (s->last_exit.tv_nsec > s->last_enter.tv_nsec) {
        s->active_time.tv_nsec += s->last_exit.tv_nsec - s->last_enter.tv_nsec;
    } else {
        s->active_time.tv_sec--;
        s->active_time.tv_nsec += SEC2NSEC + s->last_exit.tv_nsec - s->last_enter.tv_nsec;
    }

    if (s->active_time.tv_nsec >= SEC2NSEC) {
        s->active_time.tv_nsec -= SEC2NSEC;
        s->active_time.tv_sec++;
    }
}

static void *_srv_wrap(void *v)
{
    struct srv_info *s = (struct srv_info *) v;
    while (1) {
recheck:
        if (s->running == 0) {
stoprun:
            pthread_cond_wait(&(s->cond), &(s->mutex));
            goto recheck;
        }

        /* if timer is enabled, than wait */
again:
        if (s->tm_fd != -1) {
            uint64_t exp;
            ssize_t sz;
            sz = read(s->tm_fd, &exp, sizeof(exp));
            if (sz != sizeof(exp)) {
                printf("timerfd_settime failed: s->tm_fd:%d, errno:%d/%s\n",
                            s->tm_fd, errno, strerror(errno));
                if (errno == EINTR) {
                    goto again;
                }
            }

            if (s->running == 0) {
                goto stoprun;
            }
        }

        clock_gettime(CLOCK_MONOTONIC, &(s->last_enter));
        s->in_fp = 1;
        s->fp(s->srv_data);
        s->in_fp = 0;
        _srv_update_active_time(s);
        s->run_cnt++;
    }
    return NULL;
}

int srv_init(void)
{
    return 0;
}

char * srv_dump(char *buf, int sz)
{
    struct srv_info *s;
    char *p = buf;
    int i = 0;
    struct timespec cur_clk;
    struct itimerspec cur_timer;

    clock_gettime(CLOCK_MONOTONIC, &cur_clk);

    if (buf == NULL) {
        return NULL;
    }

    memset(buf, 0, sz);
    pthread_mutex_lock(&(srv_tab.mutex));
    s = srv_tab.list;

    p += snprintf(p, sz - (buf - p), "cur-clk: %f\n", cur_clk.tv_sec + cur_clk.tv_nsec/SEC2NSEC);
    p += snprintf(p, sz - (buf - p), "id   srv_name    running/in_fp\t\tfp/fname/dp\n");
    p += snprintf(p, sz - (buf - p), "\t\t\trun-cnt/run-time\tenter-clk/leave-clk\n");
    p += snprintf(p, sz - (buf - p), "\t\t\ttm_fd/next_expir/perodic\n");
    while (s != NULL) {
        p += snprintf(p, sz - (buf - p), "%-8d%-16s%2d/%-2d\t\t%16p/%s/%-16p\t\t\t\n",
                        i++, s->srv_name, s->running, s->in_fp, s->fp, s->fp_name, s->srv_data);
        p += snprintf(p, sz - (buf - p), "\t\t   %10d/%-12f\t%12f/%-12f\n",
                        s->run_cnt,
                        s->active_time.tv_sec + s->active_time.tv_nsec/SEC2NSEC,
                        s->last_enter.tv_sec + s->last_enter.tv_nsec/SEC2NSEC,
                        s->last_exit.tv_sec + s->last_exit.tv_nsec/SEC2NSEC);
        if (s->tm_fd != -1) {
            if (timerfd_gettime(s->tm_fd, &cur_timer) < 0) {
                printf("timerfd_get failed: errno:%d/%s\n", errno, strerror(errno));
                p += snprintf(p, sz - (buf - p), "\t\t   %6d\n", s->tm_fd);
            } else {
                p += snprintf(p, sz - (buf - p), "\t\t   %6d/%f/%-16f\n",
                                s->tm_fd, cur_timer.it_value.tv_sec + cur_timer.it_value.tv_nsec/SEC2NSEC,
                                cur_timer.it_interval.tv_sec + cur_timer.it_interval.tv_nsec/SEC2NSEC);
            }
        }
        s = s->next;
    }
    pthread_mutex_unlock(&(srv_tab.mutex));
    return buf;
}

int _srv_reg(char *srv_name, srv_fp fp, char *fp_name, void *srv_data)
{
    struct srv_info *s = (struct srv_info *) malloc(sizeof(struct srv_info));
    if (s == NULL) {
        return SRV_NO_MEM;
    }
    memset(s, 0, sizeof(struct srv_info));
    s->fp = fp;
    s->srv_data = srv_data;
    s->srv_name = strdup(srv_name);
    s->fp_name = strdup(fp_name);
    pthread_cond_init(&s->cond, NULL);
    pthread_mutex_init(&s->mutex, NULL);
    s->tm_fd = -1;

    if (!s->srv_name) {
        free(s);
        return SRV_NO_MEM;
    }

    /* insert into tab */
    pthread_mutex_lock(&(srv_tab.mutex));
    s->next = srv_tab.list;
    srv_tab.list = s;
    srv_tab.num_srv++;
    pthread_mutex_unlock(&(srv_tab.mutex));

    /* create the thread */
    pthread_create(&s->thread_id, NULL , _srv_wrap, s);
    return 0;
}

static int _srv_update_timer(struct srv_info *s, unsigned long sec, unsigned long nsec)
{
    struct timespec now_tm;
    struct itimerspec new_tm;

    if (s->tm_fd < 0) {
        s->tm_fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (s->tm_fd < 0) {
            printf("timerfd_create failed: errno:%d/%s\n", errno, strerror(errno));
            return -1;
        }
    }

    if (clock_gettime(CLOCK_MONOTONIC, &now_tm) == -1) {
        printf("timerfd_create failed: errno:%d/%s\n", errno, strerror(errno));
        return -1;
    }

    new_tm.it_value.tv_sec = now_tm.tv_sec + sec;
    new_tm.it_value.tv_nsec = now_tm.tv_nsec + nsec;

    new_tm.it_interval.tv_sec = sec;
    new_tm.it_interval.tv_nsec = nsec;

    if (timerfd_settime(s->tm_fd, TFD_TIMER_ABSTIME, &new_tm, NULL) < 0) {
        printf("timerfd_settime failed: errno:%d/%s\n", errno, strerror(errno));
        return -1;
    }

    return 0;
}

inline static int _srv_name_equal(struct srv_info *s, char *srv_name)
{
    return strlen(s->srv_name) == strlen(srv_name) && !strcmp(s->srv_name, srv_name);
}

static int _srv_set_running(char *srv_name, int running, unsigned long sec, unsigned long nsec)
{
    struct srv_info *s;
    int ret = 0;

    pthread_mutex_lock(&(srv_tab.mutex));
    s = srv_tab.list;
    while (s != NULL) {
        if (_srv_name_equal(s, srv_name)) {
            s->running = running;
            if (sec || nsec) {
                ret = _srv_update_timer(s, sec, nsec);
            } else {
                /* reset timer to zero */
                _srv_update_timer(s, 0, 0);
                /* clear timer */
                close(s->tm_fd);
                s->tm_fd = -1;
            }
            pthread_cond_signal(&s->cond);
            pthread_mutex_unlock(&(srv_tab.mutex));
            return ret;
        }
        s = s->next;
    }
    pthread_mutex_unlock(&(srv_tab.mutex));

    return -1;
}

int srv_start(char *srv_name)
{
    return _srv_set_running(srv_name, 1 /* running */, 0 /* delay.sec */, 0 /* delay.nsec */);
}

int srv_stop(char *srv_name)
{
    return _srv_set_running(srv_name, 0 /* running */, 0 /* delay.sec */, 0 /* delay.nsec */);
}

static int _srv_free(struct srv_info *s)
{
    int ret;
    s->running = 0;

    /* cancel the thread */
    ret = pthread_cancel(s->thread_id);
    if (ret != 0) {
        printf("pthread_cancel failed: %d/%d\n", ret, errno);
        /* FIXME */
    }

    ret = pthread_join(s->thread_id, NULL);
    if (ret != 0) {
        printf("pthread_join failed: %d/%d\n", ret, errno);
        /* FIXME */
    }

    /* free resouce */
    pthread_cond_destroy(&(s->cond));
    pthread_mutex_destroy(&(s->mutex));
    free(s->srv_name);
    free(s->fp_name);
    if (s->tm_fd != -1) {
        close(s->tm_fd);
    }
    free(s);
    return ret;
}

int srv_unreg(char *srv_name)
{
    struct srv_info **s, *sn = NULL;
    int ret;
    s = &(srv_tab.list);
    pthread_mutex_lock(&(srv_tab.mutex));
    while ((*s) != NULL) {
        if (_srv_name_equal(*s, srv_name)) {
            sn = *s;
            *s = (*s)->next;
            srv_tab.num_srv--;
            break;
        }
        s = &((*s)->next);
    }
    pthread_mutex_unlock(&(srv_tab.mutex));
    if (sn) {
        ret = _srv_free(sn);
    }

    return 0;
}

int srv_start_periodical(char *srv_name, unsigned long sec, unsigned long nsec)
{
    return _srv_set_running(srv_name, 1 /* running */, sec /* delay.sec */,  nsec /* delay.nsec */);
}




main.c (test program)

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>
#include <stdarg.h>
#include "service-reg.h"

int pr(const char *fmt, ...)
{
    va_list ap;
    int ret;
    struct timespec cur_clk;
    clock_gettime(CLOCK_MONOTONIC, &cur_clk);

    printf("***CLK: %f\n", cur_clk.tv_sec + cur_clk.tv_nsec/1000000000.0);
    va_start(ap, fmt);
    ret = vprintf(fmt, ap);
    va_end(ap);
    return ret;
}

void *sms_service(void *v)
{
    int *i = (int *) v;
    pr("   ->hello, %p/%d, srv_start_periodical %d\n", v, (*i)++, 2);
    pr("   ->%d\n", srv_start_periodical("brook", 2, 0));
}

int main(int argc, char *argv[])
{
    int i = 0, data = 0, ret;
    char name[] = "brook", buf[1024];

    pr("srv_reg %s\n", name);
    srv_reg(name, sms_service, (void*)&data);

    pr("srv_start %s\n", name);
    srv_start(name);

    pr("sleep 3\n");
    sleep(3);
    sleep(3);

    pr("srv_stop %s\n", name);
    srv_stop(name);

    pr("srv_start_periodical %s with %dsec\n", name, 1);
    srv_start_periodical(name, 1, 0);

    pr("srv_dump %s\n", name);
    pr("%s\n", srv_dump(buf, sizeof(buf)));

    pr("sleep 2\n");
    sleep(2);

    pr("srv_unreg %s\n", name);
    srv_unreg(name);

    pr("over\n");
    return 0;
}



執行結果

brook@vista:~$ ./service-reg
***CLK: 7914669.498393
srv_reg brook
***CLK: 7914669.498469
srv_start brook
***CLK: 7914669.498481
sleep 3
***CLK: 7914669.498531
        ->hello, 0x7ffc24e80df8/0, srv_start_periodical 2
***CLK: 7914669.498541
        ->0
***CLK: 7914671.498595
        ->hello, 0x7ffc24e80df8/1, srv_start_periodical 2
***CLK: 7914671.498647
        ->0
***CLK: 7914672.498576
srv_stop brook
***CLK: 7914672.498599
srv_start_periodical brook with 1sec
***CLK: 7914672.498606
srv_dump brook
***CLK: 7914672.498621
        ->hello, 0x7ffc24e80df8/2, srv_start_periodical 2
***CLK: 7914672.498662
        ->0
***CLK: 7914672.498625
cur-clk: 7914672.498611
id      srv_name        running/in_fp           fp/fname/dp
                        run-cnt/run-time        enter-clk/leave-clk
                        tm_fd/next_expir/perodic
0       brook            1/0                    0x40106f/sms_service/0x7ffc24e80df8
                            2/0.000082          7914671.498594/7914671.498656
                        3/0.999982/1.000000

***CLK: 7914672.498680
sleep 2
***CLK: 7914674.498698
        ->hello, 0x7ffc24e80df8/3, srv_start_periodical 2
***CLK: 7914674.498730
        ->0
***CLK: 7914674.498742
srv_unreg brook
***CLK: 7914674.498997
over




6 則留言:

  1. CLOCK_REALTIME is affected by NTP, and can move forwards and backwards. CLOCK_MONOTONIC is not, and advances at one tick per tick.

    回覆刪除
    回覆
    1. https://lwn.net/Articles/420142/

      CLOCK_MONOTONIC not incrementing during suspend.

      Introduce CLOCK_BOOTTIME allow applications that wanted to be aware of time passing during
      suspend without having to deal with the inconsistencies of
      CLOCK_REALTIME caused by calls to settimeofday.

      刪除
  2. https://lwn.net/Articles/342018/
    CLOCK_REALTIME_COARSE and CLOCK_MONOTONIC_COARSE
    which returns the time at the last tick. This is very fast as we don't
    have to access any hardware (which can be very painful if you're using
    something like the acpi_pm clocksource), and we can even use the vdso
    clock_gettime() method to avoid the syscall. The only trade off is you
    only get low-res tick grained time resolution.

    回覆刪除
  3. http://adrianhuang.blogspot.tw/2007/10/linux-kernel-hz-tick-and-jiffies.html
    Linux核心每隔固定週期會發出timer interrupt (IRQ 0),HZ是用來定義每一秒有幾次timer interrupts。舉例來說,HZ為1000,代表每秒有1000次timer interrupts。
    Tick
    Tick是HZ的倒數,意即timer interrupt每發生一次中斷的時間。如HZ為250時,tick為4毫秒 (millisecond)。

    回覆刪除
  4. brook@vista:~$ cat clock_getres.c
    #include
    #include

    int main()
    {
    struct timespec tt;
    clock_getres(CLOCK_REALTIME, &tt);
    printf("CLOCK_REALTIME resolution: %ld ns\n", tt.tv_nsec);
    clock_getres(CLOCK_REALTIME_COARSE, &tt);
    printf("CLOCK_REALTIME_COARSE resolution: %ld ns\n", tt.tv_nsec);
    return 0;
    }
    brook@vista:~$ ./clock_getres
    CLOCK_REALTIME resolution: 1 ns
    CLOCK_REALTIME_COARSE resolution: 4000000 ns
    brook@vista:~$ grep CONFIG_HZ /boot/config-`uname -r`
    # CONFIG_HZ_PERIODIC is not set
    # CONFIG_HZ_100 is not set
    CONFIG_HZ_250=y
    # CONFIG_HZ_300 is not set
    # CONFIG_HZ_1000 is not set
    CONFIG_HZ=250

    回覆刪除
  5. An bug on tv_nsec > 999999999. please update it by yourself.

    回覆刪除

熱門文章