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

在之前的文章中（开发利器——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字符串就是模块的名称，其余则为模块参数，即入口函数的argc和argv的部分。而与主线程通信的套接字则不必写在此处，而是线程启动后进入入口函数前自动添加的。

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);//消息处理函数

描述：依据attr对t进行初始化。

返回值：成功返回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;
}