关于 nng/nanomsg 库的一点想法

2019-02-22

NNG/nanomsg 是最近项目上使用到的一个通信库，用来实现进程间过程调用和线程间通信，很是方便。

NNG 是 nanomsg 的继任版本，而 nanomsg 则是流行的 ZMQ 的 C 重写版。

NNG 将通信使用的协议和传输分离，同一个协议可以工作在不同的传输层上，类似与 TCP/IP 的应用层和传输层的分层，同时接口上屏蔽了底层细节，统一用字符串 URL 来描述传输模式。这样当使用场景修改时，可以通过简单修改 URL 来实现适应，极具灵活性。

同时如 NNG 描述所言 “light-weight brokerless messaging”，NNG 中的通信各方是不需要第三方程序介入的，这与 MQTT/Redis 通信需要服务器不同。这样很适合作为通信库来使用而没有其他依赖。

NNG 支持的通信协议主要有以下几种：

  • PAIR 一对一双向通信。
  • PIPELINE(PUSH/PULL) 单向通信，类似与生产者消费者模型的消息队列。
  • PUB/SUB 单向广播。
  • REQ/REP 请求-应答模式，类似与 RPC 模式。
  • BUS 网状连接通信，每个加入节点都可以发送/接受广播消息。
  • SURVEY 用于多节点表决或者服务发现。

NNG 支持的传输模式主要有以下三种常用，其他还有tcp附加tls 1.2加密的tls传输和基于WebSocket的ws传输：

  • inproc 进程内线程间传输
  • ipc 主机内进程间传输
  • tcp 网络内主机间传输

通信协议里除了 PAIR 之外，基本都是一对多的通信模式，这点需要注意，以 PIPELINE 和 PUB/SUB 为例：

  1. PIPELINE 的 PUSH 端是 client，一个 PUSH 可以连接多个 PULL 端，发送数据时会选择其中一个可用的发送；PULL 端是 server，一个 PULL 可以接收多个 PUSH 连接和数据。
  2. PUB/SUB 的 SUB 端是 client，一个 SUB 可以连接多个不同的 PUB 端，接收多个 PUB 端广播的数据；PUB 端是 server，一个 PUB 可以接收多个 SUB 连接并广播数据。

基于以上，多个程序是没办法共用一个 PUB/SUB 通道来广播数据的，这与 ROS 里的 topic 和 LCM 中的 channel 模式不同。如果要实现类似功能，则可以使用 PIPELINE + PUB/SUB 来处理：

  • 独立一个话题发布的程序，拥有一个 PULL 和 PUB。
  • PULL 约定一个 URL，所有需要发布该话题的程序都 PUSH 数据到该 URL 上。
  • PUB 约定一个 URL，所有需要获取该话题的程序都 SUB 到该 URL 上。
  • 程序内部循环将 PULL 读取的数据发送到 PUB 上。

以上则可以模拟出 ROS topic 数据合并 或者 LCM 中 channel 的类似功能。

整体上看，NNG 的 API 很简约，主要是 4 个，open/recv/send/close，open 根据协议不同使用的函数会不同。配置则是 setopt/getopt，与 UNIX API 类似。API 中没有上下文环境（context-less）依赖，只需要一个 nng_socket，这种设计和实现方法值得去学习一下（初步揣测应该是使用指针值作为handle，如果要强制编译器做类型检测，则会套上一层 struct，如 typedef struct { _nng_xxx_socket * p } nng_socket;）。

NNG 协议基本上囊括了常见的通信需求，一些特殊的需求，也可以通过组合协议来实现，比如上面的模拟 ROS topic 或者 LCM channel 的方法。这样一来，如果在程序中使用 NNG，不管是多进程，还是多线程，通过设计，可以进一步增强模块化，同时不乏灵活性。如果环境变化，程序不管是由多进程改成多线程，还是由多线程改成多主机，都很容易实现。

常见模块/进程/线程间通信，可以依据具体需求来使用 PIPELINE（消息队列） 还是 REQ/REP（过程调用），而不是锁+全局变量，每个模块单元只需要做单一相关的具体事务，无需知晓全局状态。

Python tkinter GUI 中调用 shell 脚本

2018-07-25

本文给出一个 Python tkinter GUI 程序中调用 shell 脚本的代码示例，在 GUI 的一个文本框中会不断输出脚本的标准输出和标准错误且不会阻塞 GUI 窗体事件响应，同时也会在脚本执行完毕后根据状态（exit code）弹出提示框，如果在脚本运行中关闭窗口，会弹出提示并中断脚本的执行。

这个样例很适合编写一些提供给非技术人员使用的小工具。

代码如下，略长，使用的是 python 3，因为依赖 select 的原因，仅测试过 linux 下的使用，非 Posix 系统不能保证功能正常。代码也适合调用其他 console 程序，而不仅限于 shell 脚本。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

import tkinter
from tkinter import messagebox, scrolledtext
import subprocess
import select
import os
import signal

WINDOW_SIZE = "600x400"

# Run Button
btn_upd = None
# Output textbox
txt_out = None
# current script job/process
cur_process = None
# mark if Tk is closing
close_flag = False
# Tk root
root = tkinter.Tk()


def disable_allbtn():
    global close_flag
    global btn_upd
    if close_flag:
        return
    btn_upd.configure(state=tkinter.DISABLED)

def enable_allbtn():
    global close_flag
    global btn_upd
    if close_flag:
        return
    btn_upd.configure(state=tkinter.NORMAL)

def cb_runbash_withoutput(script):
    global cur_process
    global txt_out
    global root
    global close_flag
    # clear output
    txt_out.delete(1.0, tkinter.END)
    # create subprocess
    process = subprocess.Popen(script, shell=True, preexec_fn=os.setsid,
        stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    # read subprocess stderr/stdout in non-blocking
    outpoll = select.poll()
    errpoll = select.poll()
    outpoll.register(process.stdout, select.POLLIN)
    errpoll.register(process.stderr, select.POLLIN)
    # store current subprocess
    cur_process = process
    while True:
        output = ""
        errput = ""
        # poll stdout, timeout 2ms
        if outpoll.poll(2):
            output = process.stdout.readline().decode()
            # insert stdout to txt_out
            if output and not close_flag:
                txt_out.insert(tkinter.INSERT, output)
        # poll stderr, timeout 2ms
        if errpoll.poll(2):
            errput = process.stderr.readline().decode()
            # insert stderr to txt_out
            if errput and not close_flag:
                txt_out.insert(tkinter.INSERT, errput)
        # scroll text if have output
        if (output or errput) and not close_flag:
            txt_out.see(tkinter.END)
        # do eventloop when not closing
        if not close_flag:
            root.update()
        # process end
        if output == "" and process.poll() is not None:
            break
    # get process return code
    code = process.poll()
    # reset current subprocess
    cur_process = None
    return code

# callback 函数
def cb_update():
    global close_flag
    disable_allbtn()
    code = cb_runbash_withoutput("./bash_script_sample.sh")
    # exit when closing
    if close_flag:
        return
    if code != 0:
        messagebox.showerror(title="Error", message="run bash script failed")
    else:
        messagebox.showinfo(title="Info", message="run bash script success")
    enable_allbtn()


def on_closing():
    global cur_process
    global root
    global close_flag
    # if cur_process still running
    if cur_process:
        if messagebox.askokcancel("Quit", "There is task running, do you want to quit?"):
            # kill subprocess
            os.killpg(os.getpgid(cur_process.pid), signal.SIGTERM)
            cur_process = None
        else:
            return
    # mark closing
    close_flag = True
    # close window
    root.destroy()

root.title("Hello world")
root.geometry(WINDOW_SIZE)
# register cloing event callback
root.protocol("WM_DELETE_WINDOW", on_closing)

btn_upd = tkinter.Button(root, text="Update", command = cb_update)
btn_upd.pack(fill=tkinter.X, pady=10)

txt_out = scrolledtext.ScrolledText(root)
txt_out.pack(fill=tkinter.BOTH, expand=True, padx=20, pady=20)

tkinter.mainloop()

核心是使用 subprocess 调用 console 程序，然后在 GUI callback 中不断读取程序的 stdout/stderr，同时不阻塞 GUI 事件循环。

在 CMake C\C++ 项目中使用 Git 仓库版本号

2018-07-05

为了便于自动化版本管理，在C\C++项目中通常会考虑使用 Git 标识的版本作为程序版本号，本文介绍如何在 CMake 构建的项目中加入此项功能支持。

核心思路是，利用 CMake 调用 Git 命令获取仓库版本状态，然后将信息输出到一个头文件中，程序需要使用版本信息的，包含头文件并使用对应宏即可。

CMakeLists.txt 例子如下

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

cmake_minimum_required(VERSION 2.8)
project(Foo)

find_package(Git)
# 生成版本描述字符串类似 TAG-X-gHASH
execute_process(COMMAND ${GIT_EXECUTABLE} describe --abbrev=6 --dirty --always --tags
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
    OUTPUT_VARIABLE  GIT_REPO_VERSION
    OUTPUT_STRIP_TRAILING_WHITESPACE
)
# 获取最新 commit 日期，YYYY-MM-DD
execute_process(COMMAND ${GIT_EXECUTABLE} log -1 --format=%cd --date=short
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
    OUTPUT_VARIABLE  GIT_REPO_DATE
    OUTPUT_STRIP_TRAILING_WHITESPACE
)
# 获取最新 commit Hash
execute_process(COMMAND ${GIT_EXECUTABLE} log -1 --format=%H
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
    OUTPUT_VARIABLE  GIT_REPO_HASH
    OUTPUT_STRIP_TRAILING_WHITESPACE
)
# 写入头文件宏定义的版本字符串，日期和 Hash
file(WRITE ${PROJECT_BINARY_DIR}/repo_version.h
    "#define REPO_VERSION \"${GIT_REPO_VERSION}\"\n#define REPO_DATE \"${GIT_REPO_DATE}\"\n#define REPO_HASH \"${GIT_REPO_HASH}\"\n"
)
# 包含引用目录，源文件可以直接用 #include "repo_version.h" 包含
include_directories(${PROJECT_BINARY_DIR})

aux_source_directory(. SRC)
# 添加头文件依赖，保证更改后重新编译
add_executable(foo ${SRC} ${PROJECT_BINARY_DIR}/repo_version.h)

程序示例代码如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


#include <stdio.h>
#include "repo_version.h"


int main(int argc, char** argv)
{
    printf("version: %s\n", REPO_VERSION);
    printf("date: %s\n", REPO_DATE);
    printf("hash: %s\n", REPO_HASH);
    return 0;
}

如果项目小，可以不使用头文件方式，而是直接在 CMake 传入宏定义。宏定义方法不适合中大型项目，因为每次都会导致项目重编译，编译时间又长，影响效率。