C语言实用第三方库Melon开箱即用之多线程模型

2024-01-08 20:23:03

在这里插入图片描述
在之前的文章中(开发利器——C 语言必备实用第三方库),笔者介绍了一款Linux/UNIX下C语言库Melon的基本功能,并给出了一个简单的多进程开箱即用的例子。

本文将给大家介绍Melon中多线程的使用方法。

在Melon中有三种多线程模式:

  • 多线程框架
  • 线程池
  • I/O线程

我们将逐一给出实例。

Melon的Github仓库:https://github.com/Water-Melon/Melon。

多线程框架

注意:Windows下目前不支持本功能。

多线程框架与前面介绍的线程池不同,是一种模块化线程。模块化线程是指,每一个线程都是一个独立的代码模块,都有各自对应的入口函数(类似于每一个 C 语言程序有一个 main 函数一样)。

自2.3.0版本起,多线程模块不再是以在Melon目录下的threads目录中编写代码文件的方式进行集成,而是采用注册函数,由使用者在程序中进行注册加载。这样可以解除线程模块代码与Melon库在目录结构上的耦合。

开发流程

  1. 首先,编写一个源文件:

    #include <stdio.h>
    #include <assert.h>
    #include <string.h>
    #include <errno.h>
    #include "mln_framework.h"
    #include "mln_log.h"
    #include "mln_thread.h"
    
    static int haha(int argc, char **argv)
    {
        int fd = atoi(argv[argc-1]);
        mln_thread_msg_t msg;
        int nfds;
        fd_set rdset;
        for (;;) {
            FD_ZERO(&rdset);
            FD_SET(fd, &rdset);
            nfds = select(fd+1, &rdset, NULL, NULL, NULL);
            if (nfds < 0) {
                if (errno == EINTR) continue;
                mln_log(error, "select error. %s\n", strerror(errno));
                return -1;
            }
            memset(&msg, 0, sizeof(msg));
    #if defined(WIN32)
            int n = recv(fd, (char *)&msg, sizeof(msg), 0);
    #else
            int n = recv(fd, &msg, sizeof(msg), 0);
    #endif
            if (n != sizeof(msg)) {
                mln_log(debug, "recv error. n=%d. %s\n", n, strerror(errno));
                return -1;
            }
            mln_log(debug, "!!!src:%S auto:%l char:%c\n", msg.src, msg.sauto, msg.c);
            mln_thread_clear_msg(&msg);
        }
        return 0;
    }
    
    static void hello_cleanup(void *data)
    {
        mln_log(debug, "@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n");
    }
    
    static int hello(int argc, char **argv)
    {
        mln_thread_cleanup_set(hello_cleanup, NULL);
        int i;
        for (i = 0; i < 1; ++i)  {
            int fd = atoi(argv[argc-1]);
            mln_thread_msg_t msg;
            memset(&msg, 0, sizeof(msg));
            msg.dest = mln_string_new("haha");
            assert(msg.dest);
            msg.sauto = 9736;
            msg.c = 'N';
            msg.type = ITC_REQUEST;
            msg.need_clear = 1;
    #if defined(WIN32)
            int n = send(fd, (char *)&msg, sizeof(msg), 0);
    #else
            int n = send(fd, &msg, sizeof(msg), 0);
    #endif
            if (n != sizeof(msg)) {
                mln_log(debug, "send error. n=%d. %s\n", n, strerror(errno));
                mln_string_free(msg.dest);
                return -1;
            }
        }
        usleep(100000);
        return 0;
    }
    
    int main(int argc, char *argv[])
    {
        mln_thread_module_t modules[] = {
          {"haha", haha},
          {"hello", hello},
        };
    
        struct mln_framework_attr cattr;
    
        cattr.argc = argc;
        cattr.argv = argv;
        cattr.global_init = NULL;
        cattr.main_thread = NULL;
        cattr.worker_process = NULL;
        cattr.master_process = NULL;
    
        mln_thread_module_set(modules, 2);
    
        if (mln_framework_init(&cattr) < 0) {
           fprintf(stderr, "Melon init failed.\n");
           return -1;
        }
    
        return 0;
    }
    

    这段代码中有几个注意事项:

    • 在main中,我们使用mln_thread_module_set函数将每个线程模块的入口与线程模块名映射关系加载到Melon库中。
    • mln_framework_init会根据下面步骤中对配置文件的设置来自动初始化子线程,并进入线程开始运行。
    • 可以看到,两个线程模块之间是可以通信的,通信的方式是使用最后一个参数(argv的最后一个),来向主线程发送消息,消息中包含了目的线程的名字。这里注意,argv是在后面步骤中在配置文件中给出的,但是最后一个参数是程序自动在配置项后增加的一个参数,为主子线程间的通信套接字文件描述符。
    • 在 hello 这个模块中,调用了mln_thread_cleanup_set函数,这个函数的作用是:在从当前线程模块的入口函数返回至上层函数后,将会被调用,用于清理自定义资源。每一个线程模块的清理函数只能被设置一个,多次设置会被覆盖,清理函数是线程独立的,因此不会出现覆盖其他线程处理函数的情况(当然,你也可以故意这样来构造,比如传一个处理函数指针给别的模块,然后那个模块再进行设置)。
  2. 对该文件进行编译:

    $ cc -o test test.c -I /path/to/melon/include -L /path/to/melon/lib -lmelon -lpthread
    
  3. 修改配置文件

    log_level "none";
    //user "root";
    daemon off;
    core_file_size "unlimited";
    //max_nofile 1024;
    worker_proc 1;
    framework "multithread";
    log_path "/home/niklaus/melon/logs/melon.log";
    /*
     * Configurations in the 'proc_exec' are the
     * processes which are customized by user.
     *
     * Here is an example to show you how to
     * spawn a program.
     *     keepalive "/tmp/a.out" ["arg1" "arg2" ...]
     * The command in this example is 'keepalive' that
     * indicate master process to supervise this
     * process. If process is killed, master process
     * would restart this program.
     * If you don't want master to restart it, you can
     *     default "/tmp/a.out" ["arg1" "arg2" ...]
     *
     * But you should know that there is another
     * arugment after the last argument you write here.
     * That is the file descriptor which is used to
     * communicate with master process.
     */
    proc_exec {
       // keepalive "/tmp/a";
    }
    thread_exec {
        restart "hello" "hello" "world";
        default "haha";
    }
    

    这里主要关注framework以及thread_exec的配置项。thread_exec配置块专门用于模块化线程之用,其内部每一个配置项均为线程模块。

    以 hello 为例:

    restart "hello" "hello" "world";
    

    restart或者default是指令,restart表示线程退出主函数后,再次启动线程。而default则表示一旦退出便不再启动。其后的hello字符串就是模块的名称,其余则为模块参数,即入口函数的argcargv的部分。而与主线程通信的套接字则不必写在此处,而是线程启动后进入入口函数前自动添加的。

  4. 运行程序

    $ ./test
    Start up worker process No.1
    Start thread 'hello'
    Start thread 'haha'
    04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
    04/14/2022 14:50:16 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
    04/14/2022 14:50:16 UTC REPORT: PID:552165 Thread 'hello' return 0.
    04/14/2022 14:50:16 UTC REPORT: PID:552165 Child thread 'hello' exit.
    04/14/2022 14:50:16 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
    04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
    04/14/2022 14:50:16 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
    04/14/2022 14:50:16 UTC REPORT: PID:552165 Thread 'hello' return 0.
    04/14/2022 14:50:16 UTC REPORT: PID:552165 Child thread 'hello' exit.
    04/14/2022 14:50:16 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
    04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
    04/14/2022 14:50:17 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
    04/14/2022 14:50:17 UTC REPORT: PID:552165 Thread 'hello' return 0.
    04/14/2022 14:50:17 UTC REPORT: PID:552165 Child thread 'hello' exit.
    04/14/2022 14:50:17 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
    ...
    

    可以看到,事实上 Melon 中会启动工作进程来拉起其子线程,而工作进程数量由worker_proc配置项控制,如果多于一个,则每个工作进程都会拉起一组haha和hello线程。此外,我们也看到,hello线程退出后,清理函数被调用。

高级使用

除了使用mln_thread_module_set注册线程模块外,还可以使用API动态添加和删除线程。这一功能可以使得线程可以根据外部需求进行动态部署和下线。

我们来看一个简单的例子:

#include <stdio.h>
#include <errno.h>
#include "mln_framework.h"
#include "mln_log.h"
#include "mln_thread.h"
#include <unistd.h>

int sw = 0;
char name[] = "hello";
static void main_thread(mln_event_t *ev);

static int hello(int argc, char *argv[])
{
    while (1) {
        printf("%d: Hello\n", getpid());
        usleep(10);
    }
    return 0;
}

static void timer_handler(mln_event_t *ev, void *data)
{
    if (sw) {
        mln_string_t alias = mln_string("hello");
        mln_thread_kill(&alias);
        sw = !sw;
        mln_event_timer_set(ev, 1000, NULL, timer_handler);
    } else {
        main_thread(ev);
    }
}

static void main_thread(mln_event_t *ev)
{
    char **argv = (char **)calloc(3, sizeof(char *));
    if (argv != NULL) {
        argv[0] = name;
        argv[1] = NULL;
        argv[2] = NULL;
        mln_thread_create(ev, "hello", THREAD_DEFAULT, hello, 1, argv);
        sw = !sw;
        mln_event_timer_set(ev, 1000, NULL, timer_handler);
    }
}

int main(int argc, char *argv[])
{
    struct mln_framework_attr cattr;

    cattr.argc = argc;
    cattr.argv = argv;
    cattr.global_init = NULL;
    cattr.main_thread = main_thread;
    cattr.worker_process = NULL;
    cattr.master_process = NULL;

    if (mln_framework_init(&cattr) < 0) {
       fprintf(stderr, "Melon init failed.\n");
       return -1;
    }

    return 0;
}

这段代码中,我们使用main_thread这个回调来让worker进程的主线程增加一些初始化处理。

main_thread中分配了一个指针数组,用来作为线程入口参数。并且利用mln_thread_create函数创建了一个名为hello的线程,线程的入口函数是hello,这个线程的类型是THREAD_DEFAULT(退出后不会重启)。然后设置了一个1秒的定时器。

每秒钟进入一次定时处理函数,函数中通过全局变量sw来杀掉和创建hello线程。

hello线程则是死循环输出hello字符串。

这里有一个点要注意:如果要使用mln_thread_kill杀掉子线程,则子线程内不能使用mln_log来打印日志,因为可能会导致日志函数锁无法释放而致使主线程死锁。

线程池

在Melon中支持两种多线程模式,线程池是其中一种,另一种请参见后续的多线程框架文章。

注意:在每个进程中仅允许存在一个线程池。

头文件

#include "mln_thread_pool.h"

模块名

thread_pool

函数

mln_thread_pool_run
int mln_thread_pool_run(struct mln_thread_pool_attr *tpattr);

struct mln_thread_pool_attr {
    void                              *main_data;
    mln_thread_process                 child_process_handler;
    mln_thread_process                 main_process_handler;
    mln_thread_data_free                free_handler;
    mln_u64_t                          cond_timeout; /*ms*/
    mln_u32_t                          max;
    mln_u32_t                          concurrency;
};
typedef int  (*mln_thread_process)(void *);
typedef void (*mln_thread_data_free)(void *);

描述:创建并运行内存池。

线程池由主线程进行管理和做一部分处理后下发任务,子线程组则接受任务进行处理。

初始状态下,是不存在子线程的,当有任务需要下发时会自动创建子线程。当任务处理完后,子线程会延迟释放,避免频繁分配释放资源。

其中参数结构体的每个成员含义如下:

  • main_data 为主线程的用户自定义数据。
  • child_process_handler 每个子线程的处理函数,该函数有一个参数为主线程下发任务时给出的数据结构指针,返回值为0表示处理正常,非0表示处理异常,异常时会有日志输出。
  • main_process_handler 主线程的处理函数,该函数有一个参数为main_data,返回值为0表示处理正常,非0表示处理异常,异常时会有日志输出。一般情况下,主线程处理函数不应随意自行返回,一旦返回代表线程池处理结束,线程池会被销毁
  • free_handler 为资源释放函数。其资源为主线程下发给子线程的数据结构指针所指向的内容。
  • cond_timeout为闲置子线程回收定时器,单位为毫秒。当子线程无任务处理,且等待时间超过该定时器时长后,会自行退出。
  • max线程池允许的最大子线程数量。
  • concurrency用于pthread_setconcurrency设置并行级别参考值,但部分系统并为实现该功能,因此不应该过多依赖该值。在Linux下,该值设为零表示交由本系统实现自行确定并行度。

返回值:本函数返回值与主线程处理函数的返回值保持一致

mln_thread_pool_resource_add
int mln_thread_pool_resource_add(void *data);

描述:将资源data放入到资源池中。本函数仅应由主线程调用,用于主线程向子线程下发任务所用。

返回值:成功则返回0,否则返回非0

mln_thread_quit
void mln_thread_quit(void);

描述:本函数用于告知线程池,关闭并销毁线程池。

返回值:无

mln_thread_resource_info
void mln_thread_resource_info(struct mln_thread_pool_info *info);

struct mln_thread_pool_info {
    mln_u32_t                          max_num;
    mln_u32_t                          idle_num;
    mln_u32_t                          cur_num;
    mln_size_t                         res_num;
};

描述:获取当前线程池信息。信息会写入参数结构体中,结构体每个参数含义如下:

  • max_num:线程池最大子线程数量
  • idle_num:当前闲置子线程数量
  • cur_num:当前子线程数量(包含闲置和工作中的子线程)
  • res_num:当前尚未被处理的资源数量

返回值:无

示例

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "mln_thread_pool.h"

static int main_process_handler(void *data);
static int child_process_handler(void *data);
static void free_handler(void *data);

int main(int argc, char *argv[])
{
    struct mln_thread_pool_attr tpattr;

    tpattr.main_data = NULL;
    tpattr.child_process_handler = child_process_handler;
    tpattr.main_process_handler = main_process_handler;
    tpattr.free_handler = free_handler;
    tpattr.cond_timeout = 10;
    tpattr.max = 10;
    tpattr.concurrency = 10;
    return mln_thread_pool_run(&tpattr);
}

static int child_process_handler(void *data)
{
    printf("%s\n", (char *)data);
    return 0;
}

static int main_process_handler(void *data)
{
    int n;
    char *text;

    while (1) {
        if ((text = (char *)malloc(16)) == NULL) {
            return -1;
        }
        n = snprintf(text, 15, "hello world");
        text[n] = 0;
        mln_thread_pool_resource_add(text);
        usleep(1000);
    }
}

static void free_handler(void *data)
{
    free(data);
}

I/O线程

I/O线程算是一种另类线程池结构。但是这个组件主要用于图形界面类的应用。通常情况下,图形界面应用都会存在一个用户线程和一个I/O线程,这样当I/O处理时就不会无法响应用户的操作(如:点击)了。

视频介绍

头文件

#include "mln_iothread.h"

模块名

iothread

函数/宏

mln_iothread_init
int mln_iothread_init(mln_iothread_t *t, struct mln_iothread_attr *attr);

struct mln_iothread_attr {
    mln_u32_t                   nthread; //几个I/O线程
    mln_iothread_entry_t        entry; //I/O线程入口函数
    void                       *args; //I/O线程入口参数
    mln_iothread_msg_process_t  handler; //消息处理函数
};

typedef void *(*mln_iothread_entry_t)(void *); //线程入口
typedef void (*mln_iothread_msg_process_t)(mln_iothread_t *t, mln_iothread_ep_type_t from, mln_iothread_msg_t *msg);//消息处理函数

描述:依据attrt进行初始化。

返回值:成功返回0,否则返回-1

mln_iothread_destroy
void mln_iothread_destroy(mln_iothread_t *t);

描述:销毁一个iothread实例。

返回值:无

mln_iothread_send
extern int mln_iothread_send(mln_iothread_t *t, mln_u32_t type, void *data, mln_iothread_ep_type_t to, int feedback);

描述:发送一个消息类型为type,消息数据为data的消息给to的一端,并根据feedback来确定是否阻塞等待反馈。

返回值:

  • 0 - 成功
  • -1 - 失败
  • 1 - 发送缓冲区满
mln_iothread_recv
int mln_iothread_recv(mln_iothread_t *t, mln_iothread_ep_type_t from);

描述:从from的一端接收消息。接收后会调用初始化时设置好的消息处理函数,对消息进行处理。

返回值:已接收并处理的消息个数

mln_iothread_sockfd_get
 mln_iothread_iofd_get(p,t)

描述:从p所指代的mln_iothread_t结构中,根据t的值,获取I/O线程或用户线程的通信套接字。一般是为了将其加入到事件中。

返回值:套接字描述符

注意:套接字仅是用来通知对方线程(或线程组),另一端线程(或线程组)有消息发送过来,用户可以使用epoll、kqueue、select等事件机制进行监听。

mln_iothread_msg_hold
mln_iothread_msg_hold(m)

描述:将消息m持有,此时消息处理函数返回,该消息也不会被释放。主要用于流程较长的场景。该函数仅作用于feedback类型的消息,无需反馈的消息理论上也不需要持有,因为不在乎消息被处理的结果。

返回值:无

mln_iothread_msg_release
mln_iothread_msg_release(m)

描述:释放持有的消息。该消息应该是feedback类型消息,非该类型消息则可能导致执行流程异常。

返回值:无

mln_iothread_msg_type
mln_iothread_msg_type(m)

描述:获取消息的消息类型。

返回值:无符号整型

mln_iothread_msg_data
mln_iothread_msg_data(m)

描述:获取消息的用户自定义数据。

返回值:用户自定义数据结构指针

示例

#include "mln_iothread.h"
#include <string.h>
#include <stdio.h>
#include <errno.h>

static void msg_handler(mln_iothread_t *t, mln_iothread_ep_type_t from, mln_iothread_msg_t *msg)
{
    mln_u32_t type = mln_iothread_msg_type(msg);
    printf("msg type: %u\n", type);
}

static void *entry(void *args)
{
    int n;
    mln_iothread_t *t = (mln_iothread_t *)args;
    while (1) {
        n = mln_iothread_recv(t, user_thread);
        printf("recv %d message(s)\n", n);
    }

    return NULL;
}

int main(void)
{
    int i, rc;
    mln_iothread_t t;
    struct mln_iothread_attr tattr;

    tattr.nthread = 1;
    tattr.entry = (mln_iothread_entry_t)entry;
    tattr.args = &t;
    tattr.handler = (mln_iothread_msg_process_t)msg_handler;
    if (mln_iothread_init(&t, &tattr) < 0) {
        fprintf(stderr, "iothread init failed\n");
        return -1;
    }
    for (i = 0; i < 1000000; ++i) {
        if ((rc = mln_iothread_send(&t, i, NULL, io_thread, 1)) < 0) {
            fprintf(stderr, "send failed\n");
            return -1;
        } else if (rc > 0)
            continue;
    }
    sleep(1);
    mln_iothread_destroy(&t);
    sleep(3);
    printf("DONE\n");

    return 0;
}

文章来源:https://blog.csdn.net/weixin_40960130/article/details/135402201
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。