linux环境编程(3): 使用POSIX IPC完成进程间通信

时间:2023-02-06 17:08:09

1. 写在前面

之前的文章总结了使用管道进行进程间通信的方法,除了pipe和fifo,Linux内核还为我们提供了其他更高级的IPC方式,包括共享内存,消息队列,信号量等,本篇文章会通过一个具有完整逻辑功能的示例说明如何使用这些IPC方法。毕竟单纯地查手册,写代码...周而复始,这个过程还是比较枯燥的,而且并没有哪个IPC方法能解决所有的进程间通信问题,每种方法都不是孤立存在的,通过一个小例子把它们串联起来,是一种更好的学习方式。下文中的代码实现可以参考我的代码仓库

2. POSIX IPC概述

进程间通信,主要解决两个问题,即数据传递和同步。POSIX IPC提供了下面三种方法:

  • 消息队列
  • 共享内存
  • 信号量

操作系统中运行的进程,彼此之间是隔离的,要想实现通信,就必须有一个媒介,是通信双方都可以访问到的。从这个角度看,操作系统内核正是每个进程都可以访问到的那个媒介,就像一个"全局变量"。消息队列不过是内核维护的一个队列,保存了用户进程发送来的消息,其他进程可以从队列中取走消息,每个消息可以设置优先级,进程发送和接收消息的行为可以是阻塞或者非阻塞的,这一点类似管道;共享内存就是利用了虚拟地址空间以及物理地址空间,让不同进程的虚拟地址映射到同一个物理页面上,这样就实现了共享,对于映射的地址空间可以设定可读,可写以及可执行的标志;信号量就像一个内核中的整型变量,这个变量的数值记录了某种资源的数量,进程可以对它进行加减操作,合理使用的话就能完成想要的进程之间的同步逻辑。可以看到,这三种IPC方法在内核中都对应了一种数据结构,为了能够让用户进程访问到这些数据结构,POSIX IPC延续了“一切皆文件”的设计思路,我们可以用类似“/somename”这种形式的文件名去创建或者打开这些IPC对象,然后对它们进行各种操作,和文件的访问权限类似,进程操作IPC对象时也会进行权限检查。可能上面对三种POSIX IPC的描述存在不严谨的地方,但对于使用者来说,我们只要在脑子里建立一个合适的,能够描述它们工作方式的模型就可以了,而不是不断重复手册中对每个api的叙述。下面的表格列出了常用的POSIX IPC api:

消息队列 共享内存 信号量
打开 mq_open shm_open sem_open
关闭 mq_close shm_close sem_close
操作 mq_send/mq_receive 内存读写 sem_wait/sem_post
删除 mq_unlink shm_unlink sem_unlink

3. POSIX IPC使用

3.1 项目功能说明

下面将使用三种POSIX IPC实现一个简单的项目,用来记录IPC的使用方法。项目包含一个server进程和若干个client进程,他们各自的功能如下:

  • server进程
    • 首先运行,等待client的连接到来;
    • 收到client的连接,fork出一个新的进程去处理client的请求;
  • client进程
    • 可以同时运行多个;
    • 启动时和server建立连接,连接建立完成之后,接受用于输入,向server发起请求;
    • 可以完成主动断开连接,终止server进程,以及其他操作;

首先启动server进程,然后启动多个client进程向server发送请求,项目实现之后的效果如下:

linux环境编程(3): 使用POSIX IPC完成进程间通信

3.2 项目实现原理

  1. client如何和server建立连接?

    client进程和server进程都可以访问一段共享内存,当server进程启动时,会对这段共享内存进行初始化,初始化完成之后,server对信号量A执行post操作,表明共享内存准备完毕,之后server进程就通过信号量B等待新连接的建立;当有新的client进程想建立连接时,会先通过对信号量A执行wait操作,等待共享内存可用,如果可用,client会把请求参数写到共享内存之中,写入完成后会对信号量B执行post操作,通知server进程有新的连接已经建立。

  2. client建立连接之后如何发送请求?

    client通过两个消息队列实现发送请求和接收响应。在client建立连接时,会在共享内存中写入用于和server通信的两个消息队列的名字,server在处理连接时会打开消息队列,然后和client进行通信。对于每个新建立的连接,server会fork出一个新的进程去处理该连接对应的client发送来的请求。

  3. client如何通过发送请求关闭server?

    client通过向请求消息队列中写入kill_server请求,可以实现关闭server。当server进程fork出的进程从消息队列中读到kill_server请求,该进程会通过管道写入数据,通知server的主进程结束运行。

  4. server和client之间的时序关系:

    通过前面3点的描述可以看出,这个简单的项目几乎用到了全部常用的IPC方法,下面这个时序图更直观地说明了其工作原理:

    linux环境编程(3): 使用POSIX IPC完成进程间通信

    server和client之间的同步操作主要集中在步骤6,7,8,9。当server准备好共享内存之后,通过第6步的信号量A通知client可以建立连接了,之后client向共享内存写入数据,再操作第9步的信号量B通知server连接数据已经写入,最后server会创建子进程去处理client的请求。实际上server的主进程是一个循环,处理请求都是在server的子进程中完成的,以上内容说明了server主进程在循环中完成的工作。

  5. 资源清理

    当我们使用POSIX IPC时,内核会建立相应的数据结构,并且通过文件系统接口展示给用户,但IPC资源不能无限创建,当我们的程序运行结束之后应该清理自己用到的IPC资源。运行程序时创建的POSIX IPC对象可以在/dev/shm以及/dev/mqueue下查看,程序结束之后,server和client会释放掉自己创建的IPC资源。所以,要查看server和client创建的共享内存,信号量以及消息队列,需要在程序运行期间查看上述的两个目录。

3.3 主要代码功能

  • 消息格式:

    server和client之间通过消息队列传递请求和响应数据,消息队列中消息格式定义如下:

    struct msgbuf {
        int type;
        union {
            struct {
                int a;
                int b;
            } request_add;
    
            struct {
                int c;
            } response_add;
    
            struct {
                int disconect;
            } request_disconnect;
    
            struct {
                int kill_server;
            } request_kill_server;
        } data;
    };
    
  • server主进程:

    int main(int argc, char **argv) {
    
        int err = server_init();
        if (err) {
            log_warning("server_init failed\n");
            return -1;
        }
        server_start();
        server_shutdown();
    
        return 0;
    }
    

    其中,在server_init中,server会创建需要使用的共享内存,信号量以及管道。

    int server_init() {
        memset(&ipc_server, 0, sizeof(ipc_server));
    
        // shared memory init
        ipc_server.conn_buf_fd =
            shm_open(CONNECTION_SHM, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
    
        ...
    
        if (ftruncate(ipc_server.conn_buf_fd, CONNECTION_SHM_SIZE) < 0) {
            log_warning("server failed ftruncate\n");
            return -1;
        }
    
        ipc_server.conn_buf = (struct connection *)mmap(
            NULL, CONNECTION_SHM_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED,
            ipc_server.conn_buf_fd, 0);
    
        ...
    
        memset(ipc_server.conn_buf, 0, CONNECTION_SHM_SIZE);
    
        ipc_server.conn_buf_ready =
            sem_open(CONNECTION_BUF_SEM, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR, 0);
    
        ...
    
        ipc_server.conn_new_ready =
            sem_open(CONNECTION_NEW_SEM, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR, 0);
        
        ...
    
        // pipe init
        int pipefd[2];
        if (pipe2(ipc_server.pipefd, O_NONBLOCK)) {
            log_warning("server failed pipe2\n");
            return -1;
        }
    
        log_info("server init done\n");
        return 0;
    }
    

    在server_start中会循环处理来自client的连接。

    void server_start() {
        int err = sem_post(ipc_server.conn_buf_ready);
    
        ...
    
        struct connection conn;
        int stop = 0;
        while (!stop) {
            // handle new connection
            sem_wait(ipc_server.conn_new_ready);
            if (read(ipc_server.pipefd[0], &stop, sizeof(int)) <= 0)
                stop = 0;
    
            if (ipc_server.conn_buf->valid) {
                log_info("new connection established\n");
                memcpy(&conn, ipc_server.conn_buf, sizeof(conn));
                handle_connection(&conn);
                memset(ipc_server.conn_buf, 0, sizeof(struct connection));
                sem_post(ipc_server.conn_buf_ready);
            }
        }
    }
    

    当server主进程退出之后,server_shutdown会清理IPC资源。

  • client进程:

    client启动之后,首先会尝试和server建立连接,建立连接之后会循环处理用户输入,通过消息队列向server的服务进程发送请求。

    int main(int argc, char **argv) {
    
        if (build_connection()) {
            log_info("client failed build_connection\n");
            return -1;
        }
        handle_command();
        cleanup();
    
        log_info("client %d exit\n", getpid());
        return 0;
    }
    

    client建立连接的过程如下:建立连接时client需要等待共享内存可用,并且在写入连接数据之后通知server,这些同步操作都是通过信号量实现的。

    int build_connection() {
        
        ...
    
        connection.mqreq_fd =
            mq_open(connection.mqreq, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR, &attr);
        connection.mqrsp_fd =
            mq_open(connection.mqrsp, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR, &attr);
    
        ...
    
        // open and map shared memory
        int fd = shm_open(CONNECTION_SHM, O_RDWR, 0);
        void *conn_buf = mmap(NULL, CONNECTION_SHM_SIZE, PROT_READ | PROT_WRITE,
                            MAP_SHARED, fd, 0);
    
        // write connection to conn_buf, notify server new conncection is comming
        sem_t *conn_buf_ready = sem_open(CONNECTION_BUF_SEM, O_RDWR);
        sem_t *conn_new_ready = sem_open(CONNECTION_NEW_SEM, O_RDWR);
    
        ...
    
        sem_wait(conn_buf_ready);
        connection.valid = 1;
        memcpy(conn_buf, &connection, sizeof(connection));
        sem_post(conn_new_ready);
    
        ...
    
        return 0;
    }
    

写在最后

通过一个包含server和client的代码示例,说明了POSIX IPC*享内存,消息队列以及信号量的使用方法。具体实现可以参考我的代码仓库