09-Unix域
Unix域一种进程间通信的方式，Unix域不仅支持没有继承关系的进程间进行通信，而且支持进程间传递文件描述符。Unix域是Node.js中核心的功能，它是进程间通信的底层基础，child_process和cluster模块都依赖Unix域的能力。从实现和使用上来看，Unix域类似TCP，但是因为它是基于同主机进程的，不像TCP需要面临复杂的网络的问题，所以实现也没有TCP那么复杂。Unix域和传统的socket通信一样，遵循网络编程的那一套流程，由于在同主机内，就不必要使用IP和端口的方式。Node.js中，Unix域采用的是一个文件作为标记。大致原理如下。
1 服务器首先拿到一个socket。
2 服务器bind一个文件，类似bind一个IP和端口一样，对于操作系统来说，就是新建一个文件（不一定是在硬盘中创建，可以设置抽象路径名），然后把文件路径信息存在socket中。
3 调用listen修改socket状态为监听状态。
4 客户端通过同样的文件路径调用connect去连接服务器。这时候用于表示客户端的结构体插入服务器的连接队列，等待处理。
5 服务器调用accept摘取队列的节点，然后新建一个通信socket和客户端进行通信。
Unix域通信本质还是基于内存之间的通信，客户端和服务器都维护一块内存，这块内存分为读缓冲区和写缓冲区。从而实现全双工通信，而Unix域的文件路径，只不过是为了让客户端进程可以找到服务端进程，后续就可以互相往对方维护的内存里写数据，从而实现进程间通信。
9.1 Unix域在Libuv中的使用
接下来我们看一下在Libuv中关于Unix域的实现和使用。
9.1.1 初始化
Unix域使用uv_pipe_t结构体表示，使用之前首先需要初始化uv_pipe_t。下面看一下它的实现逻辑。
| int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle, int ipc) {
uv__stream_init(loop, (uv_stream_t*)handle, UV_NAMED_PIPE);
handle->shutdown_req = NULL;
handle->connect_req = NULL;
handle->pipe_fname = NULL;
handle->ipc = ipc;
return 0;
}
|
uv_pipe_init逻辑很简单，就是初始化uv_pipe_t结构体的一些字段。uv_pipe_t继承于stream，uv__stream_init就是初始化stream（父类）的字段。uv_pipe_t中有一个字段ipc，该字段标记了是否允许在该Unix域通信中传递文件描述符。
9.1.2 绑定Unix域路径
开头说过，Unix域的实现类似TCP的实现。遵循网络socket编程那一套流程。服务端使用bind，listen等函数启动服务。
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 | // name是unix路径名称
int uv_pipe_bind(uv_pipe_t* handle, const char* name) {
struct sockaddr_un saddr;
const char* pipe_fname;
int sockfd;
int err;
pipe_fname = NULL;
pipe_fname = uv__strdup(name);
name = NULL;
// 流式Unix域套接字
sockfd = uv__socket(AF_UNIX, SOCK_STREAM, 0);
memset(&saddr, 0, sizeof saddr);
strncpy(saddr.sun_path, pipe_fname, sizeof(saddr.sun_path) - 1);
saddr.sun_path[sizeof(saddr.sun_path) - 1] = '\0';
saddr.sun_family = AF_UNIX;
// 绑定到路径，TCP是绑定到IP和端口
if (bind(sockfd, (struct sockaddr*)&saddr, sizeof saddr)) {
// ...
}
// 设置绑定成功标记
handle->flags |= UV_HANDLE_BOUND;
// Unix域的路径
handle->pipe_fname = pipe_fname;
// 保存socket对应的fd
handle->io_watcher.fd = sockfd;
return 0;
}
|
uv_pipe_bind函数首先申请一个socket，然后调用操作系统的bind函数把Unix域路径保存到socket中。最后标记已经绑定标记，并且保存Unix域的路径和socket对应的fd到handle中，后续需要使用。我们看到Node.js中Unix域的类型是SOCK_STREAM。Unix域支持两种数据模式。
1 流式（ SOCK_STREAM），类似TCP，数据为字节流，需要应用层处理粘包问题。
2 数据报模式（ SOCK_DGRAM ），类似UDP，不需要处理粘包问题。
通过Unix域虽然可以实现进程间的通信，但是我们拿到的数据可能是"乱的"，这是为什么呢？一般情况下，客户端给服务器发送1个字节，然后服务器处理，如果是基于这种场景，那么数据就不会是乱的。因为每次就是一个需要处理的数据单位。但是如果客户端给服务器发送1个字节，服务器还没来得及处理，客户端又发送了一个字节，那么这时候服务器再处理的时候，就会有问题。因为两个字节混一起了。就好比在一个TCP连接上先后发送两个HTTP请求一样，如果服务器没有办法判断两个请求的数据边界，那么处理就会有问题。所以这时候，我们需要定义一个应用层协议，并且实现封包解包的逻辑，才能真正完成进程间通信。
9.1.3 启动服务
绑定了路径后，就可以调用listen函数使得socket处于监听状态。
1
2
3
4
5
6
7
8
9
10
11
12 | int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {
// uv__stream_fd(handle)得到bind函数中获取的socket
if (listen(uv__stream_fd(handle), backlog))
return UV__ERR(errno);
// 保存回调，有进程调用connect的时候时触发，由uv__server_io函数触发
handle->connection_cb = cb;
// IO观察者的回调
handle->io_watcher.cb = uv__server_io;
// 注册IO观察者到Libuv，等待连接，即读事件到来
uv__io_start(handle->loop, &handle->io_watcher, POLLIN);
return 0;
}
|
uv_pipe_listen执行操作系统的listen函数使得socket成为监听型的套接字。然后把socket对应的文件描述符和回调封装成IO观察者。注册到Libuv中。等到有读事件到来（有连接到来）。就会执行uv__server_io函数，摘下对应的客户端节点。最后执行connection_cb回调。
9.1.4 发起连接
这时候，我们已经成功启动了一个Unix域服务。接下来就是看客户端的逻辑。
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 | void uv_pipe_connect(uv_connect_t* req,
uv_pipe_t* handle,
const char* name,
uv_connect_cb cb) {
struct sockaddr_un saddr;
int new_sock;
int err;
int r;
// 判断是否已经有socket了，没有的话需要申请一个，见下面
new_sock = (uv__stream_fd(handle) == -1);
// 客户端还没有对应的socket fd
if (new_sock) {
handle->io_watcher.fd= uv__socket(AF_UNIX,
SOCK_STREAM,
0);
}
// 需要连接的服务器信息。主要是Unix域路径信息
memset(&saddr, 0, sizeof saddr);
strncpy(saddr.sun_path, name, sizeof(saddr.sun_path) - 1);
saddr.sun_path[sizeof(saddr.sun_path) - 1] = '\0';
saddr.sun_family = AF_UNIX;
// 非阻塞式连接服务器，Unix域路径是name
do {
r = connect(uv__stream_fd(handle),
(struct sockaddr*)&saddr, sizeof saddr);
}
while (r == -1 && errno == EINTR);
// 忽略错误处理逻辑
err = 0;
// 设置socket的可读写属性
if (new_sock) {
err = uv__stream_open((uv_stream_t*)handle,
uv__stream_fd(handle),
UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
}
// 把IO观察者注册到Libuv，等到连接成功或者可以发送请求
if (err == 0)
uv__io_start(handle->loop,
&handle->io_watcher,
POLLIN | POLLOUT);
out:
// 记录错误码，如果有的话
handle->delayed_error = err;
// 保存调用者信息
handle->connect_req = req;
uv__req_init(handle->loop, req, UV_CONNECT);
req->handle = (uv_stream_t*)handle;
req->cb = cb;
QUEUE_INIT(&req->queue);
/*
如果连接出错，在pending阶段会执行uv__stream_io，
从而执行req对应的回调。错误码是delayed_error
*/
if (err)
uv__io_feed(handle->loop, &handle->io_watcher);
}
|
uv_pipe_connect函数首先以非阻塞的方式调用操作系统的connect函数，调用connect后操作系统把客户端对应的socket直接插入服务器socket的待处理socket队列中，等待服务器处理。这时候socket是处于连接中的状态，当服务器调用accept函数处理连接时，会修改连接状态为已连接（这和TCP不一样，TCP是完成三次握手后就会修改为连接状态，而不是accept的时候），并且会触发客户端socket的可写事件。事件驱动模块就会执行相应的回调（uv__stream_io），从而执行C++和JS的回调。
9.1.5 关闭Unix域
我们可以通过uv_close关闭一个Unix域handle。uv_close中会调用uv__pipe_close。
| void uv__pipe_close(uv_pipe_t* handle) {
// 如果是Unix域服务器则需要删除Unix域路径并删除指向的堆内存
if (handle->pipe_fname) {
unlink(handle->pipe_fname);
uv__free((void*)handle->pipe_fname);
handle->pipe_fname = NULL;
}
// 关闭流相关的内容
uv__stream_close((uv_stream_t*)handle);
}
|
关闭Unix域handle时，Libuv会自动删除Unix域路径对应的文件。但是如果进程异常退出时，该文件可能不会被删除，这样会导致下次监听的时候报错listen EADDRINUSE，所以安全起见，我们可以在进程退出或者监听之前判断该文件是否存在，存在的话则删除。另外还有一个问题是，如果两个不相关的进程使用了同一个文件则会导致误删，所以Unix域对应的文件，我们需要小心处理，最好能保证唯一性。
Unix域大致的流程和网络编程一样。分为服务端和客户端两面。Libuv在操作系统提供的API的基础上。和Libuv的异步非阻塞结合。在Libuv中为进程间提供了一种通信方式。下面看一下在Node.js中是如何使用Libuv提供的功能的。
9.2 Unix域在Node.js中的使用
9.2.1 Unix域服务器
在Node.js中，我们可以通过以下代码创建一个Unix域服务器
| const server = net.createServer((client) => {
// 处理client
});
server.listen('/tmp/test.sock', () => {
console.log(`bind uinx domain success`);
});
|
我们从listen函数开始分析这个过程。
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 | Server.prototype.listen = function(...args) {
const normalized = normalizeArgs(args);
let options = normalized[0];
const cb = normalized[1];
// 调用底层的listen函数成功后执行的回调
if (cb !== null) {
this.once('listening', cb);
}
if (options.path && isPipeName(options.path)) {
const pipeName = this._pipeName = options.path;
backlog = options.backlog || backlogFromArgs;
listenIncluster(this, pipeName, -1, -1, backlog, undefined,
options.exclusive);
/*
Unix域使用文件实现的，客户端需要访问该文件的权限才能通信，
这里做权限控制
*/
let mode = 0;
if (options.readableAll === true)
mode |= PipeConstants.UV_READABLE;
if (options.writableAll === true)
mode |= PipeConstants.UV_WRITABLE;
if (mode !== 0) {
// 修改文件的访问属性
const err = this._handle.fchmod(mode);
if (err) {
this._handle.close();
this._handle = null;
throw errnoException(err, 'uv_pipe_chmod');
}
}
return this;
}
}
|
这段代码中最主要的是listenIncluster函数。我们看一下该函数的逻辑。
| function listenIncluster(server, address, port, addressType,
backlog, fd, exclusive, flags) {
exclusive = !!exclusive;
if (cluster === undefined) cluster = require('cluster');
if (cluster.isMaster || exclusive) {
server._listen2(address, port, addressType, backlog, fd, flags);
return;
}
}
|
直接调用_listen2（isMaster只有在cluster.fork创建的进程中才是false，其余情况都是true，包括child_process模块创建的子进程）。我们继续看listen函数。
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 | Server.prototype._listen2 = setupListenHandle;
function setupListenHandle(address,
port,
addressType,
backlog,
fd,
flags) {
this._handle = createServerHandle(address,
port,
addressType,
fd,
flags);
// 有完成连接完成时触发
this._handle.onconnection = onconnection;
const err = this._handle.listen(backlog || 511);
if (err) {
// 触发error事件
}
// 下一个tick触发listen回调
defaultTriggerAsyncIdScope(this[async_id_symbol],
process.nextTick,
emitListeningNT,
this);
}
|
| function createServerHandle(address,
port,
addressType,
fd,
flags) {
let handle = new Pipe(PipeConstants.SERVER);
handle.bind(address, port);
return handle;
}
|
创建了一个Pipe对象，然后调用它的bind和listen函数，我们看new Pipe的逻辑，从pipe_wrap.cc的导出逻辑，我们知道，这时候会新建一个C++对象，然后执行New函数，并且把新建的C++对象等信息作为入参。
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 | void PipeWrap::New(const FunctionCallbackInfo<Value>& args) {
Environment* env = Environment::GetCurrent(args);
// 类型
int type_value = args[0].As<Int32>()->Value();
PipeWrap::SocketType type = static_cast<PipeWrap::SocketType>(type_value);
// 是否是用于IPC
bool ipc;
ProviderType provider;
switch (type) {
case SOCKET:
provider = PROVIDER_PIPEWRAP;
ipc = false;
break;
case SERVER:
provider = PROVIDER_PIPESERVERWRAP;
ipc = false;
break;
case IPC:
provider = PROVIDER_PIPEWRAP;
ipc = true;
break;
default:
UNREACHABLE();
}
new PipeWrap(env, args.This(), provider, ipc);
}
|
New函数处理了参数，然后执行了new PipeWrap创建一个对象。
| PipeWrap::PipeWrap(Environment* env,
Local<Object> object,
ProviderType provider,
bool ipc)
: ConnectionWrap(env, object, provider) {
int r = uv_pipe_init(env->event_loop(), &handle_, ipc);
}
|
9.2.2 Unix域客户端
接着我们看一下Unix域作为客户端使用时的过程。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22 | Socket.prototype.connect = function(...args) {
const path = options.path;
// Unix域路径
var pipe = !!path;
if (!this._handle) {
// 创建一个C++层handle，即pipe_wrap.cc导出的Pipe类
this._handle = pipe ?
new Pipe(PipeConstants.SOCKET) :
new TCP(TCPConstants.SOCKET);
// 挂载onread方法到this中
initSocketHandle(this);
}
if (cb !== null) {
this.once('connect', cb);
}
// 执行internalConnect
defaultTriggerAsyncIdScope(
this[async_id_symbol], internalConnect, this, path
);
return this;
};
|
首先新建一个handle，值是new Pipe。接着执行了internalConnect，internalConnect函数的主要逻辑如下
| const req = new PipeConnectWrap();
// address为Unix域路径
req.address = address;
req.oncomplete = afterConnect;
// 调用C++层connect
err = self._handle.connect(req, address, afterConnect);
|
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 | void PipeWrap::Connect(const FunctionCallbackInfo<Value>& args) {
Environment* env = Environment::GetCurrent(args);
PipeWrap* wrap;
ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
// PipeConnectWrap对象
Local<Object> req_wrap_obj = args[0].As<Object>();
// Unix域路径
node::Utf8Value name(env->isolate(), args[1]);
/*
新建一个ConnectWrap对象，ConnectWrap是对handle进行一次连接请求
的封装，内部维护一个uv_connect_t结构体， req_wrap_obj的一个字段
指向ConnectWrap对象，用于保存对应的请求上下文
*/
ConnectWrap* req_wrap =
new ConnectWrap(env,
req_wrap_obj,
AsyncWrap::PROVIDER_PIPECONNECTWRAP);
// 调用Libuv的connect函数
uv_pipe_connect(req_wrap->req(),
&wrap->handle_,
*name,
AfterConnect);
// req_wrap->req_.data = req_wrap;关联起来
req_wrap->Dispatched();
// uv_pipe_connect() doesn't return errors.
args.GetReturnValue().Set(0);
}
|
uv_pipe_connect函数，第一个参数是uv_connect_t结构体（request），第二个是一个uv_pipe_t结构体（handle），handle是对Unix域客户端的封装，request是请求的封装，它表示基于handle发起一次连接请求。连接成功后会执行AfterConnect。由前面分析我们知道，当连接成功时，首先会执行回调Libuv的uv__stream_io，然后执行C++层的AfterConnect。
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 | // 主动发起连接，成功/失败后的回调
template <typename WrapType,typename UVType> = PipeWrap, uv_pipe_t
void ConnectionWrap<WrapType, UVType>::AfterConnect(uv_connect_t* req
,int status) {
// 在Connect函数里关联起来的
ConnectWrap* req_wrap = static_cast<ConnectWrap*>(req->data);
// 在uv_pipe_connect中完成关联的
WrapType* wrap = static_cast<WrapType*>(req->handle->data);
Environment* env = wrap->env();
HandleScope handle_scope(env->isolate());
Context::Scope context_scope(env->context());
bool readable, writable;
// 是否连接成功
if (status) {
readable = writable = 0;
} else {
readable = uv_is_readable(req->handle) != 0;
writable = uv_is_writable(req->handle) != 0;
}
Local<Value> argv[5] = {
Integer::New(env->isolate(), status),
wrap->object(),
req_wrap->object(),
Boolean::New(env->isolate(), readable),
Boolean::New(env->isolate(), writable)
};
// 执行JS层的oncomplete回调
req_wrap->MakeCallback(env->oncomplete_string(),
arraysize(argv),
argv);
delete req_wrap;
}
|
我们再回到JS层的afterConnect
1
2
3
4
5
6
7
8
9
10
11
12
13
14 | function afterConnect(status, handle, req, readable, writable) {
var self = handle.owner;
handle = self._handle;
if (status === 0) {
self.readable = readable;
self.writable = writable;
self._unrefTimer();
// 触发connect事件
self.emit('connect');
// 可读并且没有处于暂停模式，则注册等待可读事件
if (readable && !self.isPaused())
self.read(0);
}
}
|
至此，作为客户端对服务器的连接就完成了。后续就可以进行通信。