babyos2（29） socket(AF_LOCAL), IPC

这几天为babyos2实现了通过socket进行进程间通信，即AF_LOCAL相关功能，主要参考linux 1.2。

socket常翻译为套接字，也有些地方翻译成插口，挺形象。把socket想象成类似网线插口的东西，则通过一根网线（连接），把两个socket连接起来，就可以进行通信。

AF_LOCAL域的socket比较简单，因为在同一个系统中，就可以通过addr(path)查找要连接的socket，而读写只需要在内核中读写相关的缓冲区即可。

babyos2 的实现中，每个socket会对应一个文件，但跟linux 1.2不同，不会对应inode。文件设置了一个SOCKET的类型，当通过一个fd去读写时，会找到这个文件，然后若发现文件类型是SOCKET，会通过文件中记录的socket指针，调用它的读写方法。

1.系统调用

babyos2新增加了一个SYS_SOCKET的系统调用，但这个系统调用会根据参数决定具体行为，目前支持：socket, bind, listen, accept, connect。

int32 syscall_t::sys_socket(trap_frame_t* frame)
{
return sys_socket_t::do_sys_socket(frame);
}

int32 sys_socket_t::do_sys_socket(trap_frame_t* frame)
{
uint32 id = frame->ebx;
if (id >= sys_socket_t::MAX_SYS_SOCKET) {
console()->kprintf(RED, "unknown sys_socket call %x, current: %p\n", id, current->m_pid);
return -1;
}
else {
return s_sys_socket_table[id](frame);
}
}

sys_socket_t是类似于syscall_t的一个类，实现socket相关的系统调用。

/* socket syscalls */
int32 sys_socket_t::sys_socket(trap_frame_t* frame)
{
uint32 family = frame->ecx, type = frame->edx, protocol = frame->esi;
return socket(family, type, protocol);
}

int32 sys_socket_t::sys_bind(trap_frame_t* frame)
{
int fd = frame->ecx;
sock_addr_t* myaddr = (sock_addr_t *) frame->edx;
return bind(fd, myaddr);
}

int32 sys_socket_t::sys_listen(trap_frame_t* frame)
{
int fd = frame->ecx;
uint32 backlog = frame->edx;
return listen(fd, backlog);
}

int32 sys_socket_t::sys_accept(trap_frame_t* frame)
{
int fd = frame->ecx;
sock_addr_t* peer_addr = (sock_addr_t *) frame->edx;
return accept(fd, peer_addr);
}

int32 sys_socket_t::sys_connect(trap_frame_t* frame)
{
int fd = frame->ecx;
sock_addr_t* user_addr = (sock_addr_t *) frame->edx;
return connect(fd, user_addr);
}

这几个函数比较简单，只是解析参数，具体功能由调用的函数实现。

2.socket

int32 sys_socket_t::socket(uint32 family, uint32 type, uint32 protocol)
{
//console()->kprintf(WHITE, "socket, pid: %u\n", current->m_pid);
if (family >= socket_t::AF_MAX) {
return -EINVAL;
}

/* alloc a socket */
socket_t* socket = alloc_socket(family);
if (socket == NULL) {
return -ENOSR;
}
socket->create(family, type, protocol);

/* alloc a file */
file_t* file = os()->get_fs()->alloc_file();
if (file == NULL) {
release_socket(socket);
return -ENOSR;
}
file->init(file_t::TYPE_SOCKET, socket);

/* alloc a fd and bind to file */
int fd = current->alloc_fd(file);
if (fd < 0) {
os()->get_fs()->close_file(file);
release_socket(socket);
return -ENOSR;
}

return fd;
}

socket函数，首先会检查是不是AF_LOCAL，目前只支持这一种域。

然后分配一个socket：

socket_t* sys_socket_t::alloc_socket(uint32 family)
{
/* now only support AF_LOCAL */
if (family == socket_t::AF_LOCAL) {
return socket_local_t::alloc_local_socket();
}

return NULL;
}

socket_t* socket_local_t::alloc_local_socket()
{
locker_t locker(s_lock);

socket_local_t* socket = s_local_sockets;
for (int i = 0; i < MAX_LOCAL_SOCKET; i++, socket++) {
if (socket->m_ref == 0) {
socket->m_ref = 1;
return socket;
}
}

return NULL;
}

最终是通过socket_local_t的一个静态函数分配，该类是socket_t的子类，用于统一socket相关函数的接口。

然后做一些初始化工作，之后分配一个文件，并申请一个文件描述符绑定在一起，这个socket就算创建完成了。

3.bind

int32 sys_socket_t::bind(int fd, sock_addr_t* myaddr)
{
//console()->kprintf(WHITE, "bind, pid: %u, fd: %u\n", current->m_pid, fd);
socket_t* socket = look_up_socket(fd);
if (socket == NULL) {
return -EBADF;
}

return socket->bind(myaddr);
}

int32 socket_local_t::bind(sock_addr_t* myaddr)
{
sock_addr_local_t* addr = (sock_addr_local_t *) myaddr;
return socket_local_t::bind_local_socket(this, addr);
}

int32 socket_local_t::bind_local_socket(socket_local_t* socket, sock_addr_local_t* addr)
{
locker_t locker(s_lock);

socket_local_t* s = s_local_sockets;
for (int i = 0; i < MAX_LOCAL_SOCKET; i++, s++) {
if (s->m_addr == *addr && s->m_ref != 0) {
//console()->kprintf(RED, "socket %p, path: %s, ref: %u\n", s, s->m_addr.m_path, s->m_ref);
return -1;
}
}

memcpy(&socket->m_addr, addr, sizeof(sock_addr_local_t));
return 0;
}

bind会先去系统初始化时创建好的一个socket列表中查找看是否已经创建了同一个path的socket，若已创建，则bind返回失败，否则绑定该address。

4.listen

int32 sys_socket_t::listen(int fd, uint32 backlog)
{
//console()->kprintf(WHITE, "listen, pid: %u, fd: %u\n", current->m_pid, fd);
socket_t* socket = look_up_socket(fd);
if (socket == NULL) {
return -EBADF;
}

if (socket->m_state != socket_t::SS_UNCONNECTED) {
return -EINVAL;
}

socket->listen(backlog);
socket->m_flags |= socket_t::SF_ACCEPTCON;

return 0;
}

listen会设置ACCEPTCON的标记，表示这是一个server端的socket，准备接受客户端的连接。

5.accept

int32 sys_socket_t::accept(int fd, sock_addr_t* client_addr)
{
//console()->kprintf(WHITE, "accept, pid: %u, fd: %u\n", current->m_pid, fd);
socket_t* socket = look_up_socket(fd);

/* not find a socket */
if (socket == NULL) {
return -EBADF;
}

/* not connected */
if (socket->m_state != socket_t::SS_UNCONNECTED) {
return -EINVAL;
}

/* not listening */
if (!(socket->m_flags & socket_t::SF_ACCEPTCON)) {
return -EINVAL;
}

/* alloc a new socket to accept the connect */
socket_t* new_socket = alloc_socket(socket->m_family);
if (new_socket == NULL) {
return -ENOSR;
}
new_socket->dup(socket);

/* alloc a file */
file_t* file = os()->get_fs()->alloc_file();
if (file == NULL) {
release_socket(new_socket);
return -ENOSR;
}
file->init(file_t::TYPE_SOCKET, new_socket);

/* alloc a fd and bind to file */
int new_fd = current->alloc_fd(file);
if (new_fd < 0) {
os()->get_fs()->close_file(file);
release_socket(new_socket);
return -ENOSR;
}

//console()->kprintf(PINK, "server socket: %p, create new socket: %p to wait for connect.\n", socket, new_socket);
uint32 ret = new_socket->accept(socket);
if (ret < 0) {
return ret;
}

if (client_addr != NULL) {
if (new_socket->get_name(client_addr) < 0) {
return -ECONNABORTED;
}
}

return new_fd;
}

accept会检查自己的状态，及是否有ACCEPTCON的标记，然后创建一个新的socket用于跟客户端建立连接，而自己则继续等待其他连接。

创建完成后，会拷贝自己的内容，并走类似创建socket的流程，然后：

int32 socket_local_t::accept(socket_t* server_socket)
{
/* wait for connect */
while (server_socket->m_connecting_list.empty()) {
server_socket->m_flags |= socket_t::SF_WAITDATA;
server_socket->m_wait_connect_sem.down();
//console()->kprintf(YELLOW, "wait connect waked up\n");
server_socket->m_flags &= ~socket_t::SF_WAITDATA;
}

/* get a connect */
socket_t* client_socket = NULL;

spinlock_t* lock = server_socket->m_connecting_list.get_lock();
lock->lock_irqsave();
client_socket = *(server_socket->m_connecting_list.begin());
server_socket->m_connecting_list.pop_front();
lock->unlock_irqrestore();

client_socket->m_connected_socket = this;
client_socket->m_state = socket_t::SS_CONNECTED;

this->m_connected_socket = client_socket;
this->m_state = socket_t::SS_CONNECTED;
this->m_ref++;
//console()->kprintf(PINK, "inc ref, socket %p ref: %u\n", this, m_ref);

/* wake up client that accepted */
client_socket->m_wait_accept_sem.up();

return 0;
}

然后等待在一个semaphore上，直到server socket的等待连接的队列不空。

等到后，则设置状态，建立连接关系，然后通知client端连接已建立，唤醒等待在另一个semaphore上的client进程。

6.connect

connect是客户端调用，用于跟server建立连接。

int32 socket_local_t::connect(sock_addr_t* server_addr)
{
/* check state */
if (m_state == SS_CONNECTING) {
return -EINPROGRESS;
}
if (m_state == SS_CONNECTED) {
return -EISCONN;
}

/* check server addr */
if (server_addr->m_family != AF_LOCAL) {
return -EINVAL;
}

/* get server socket */
socket_t* server_socket = socket_local_t::look_up_local_socket((sock_addr_local_t *) server_addr);
if (server_socket == NULL || server_socket->m_state != socket_t::SS_UNCONNECTED) {
return -EINVAL;
}

/* check server */
if (!(server_socket->m_flags & SF_ACCEPTCON)) {
return -EINVAL;
}

/* add this to server socket's connecting list */
spinlock_t* lock = server_socket->m_connecting_list.get_lock();
lock->lock_irqsave();
server_socket->m_connecting_list.push_back(this);
lock->unlock_irqrestore();

/* notify server there is a new connect */
server_socket->m_wait_connect_sem.up();

/* wait for server finish accept */
m_wait_accept_sem.down();

/* check if connect correctly */
if (this->m_state != SS_CONNECTED) {
return m_connected_socket ? -EINTR : -EACCES;
}

m_ref++;
//console()->kprintf(PINK, "inc ref, socket %p ref: %u\n", this, m_ref);
return 0;
}

该函数检查状态后，会去socket列表中搜索要连接的server端socket是否存在，若存在，且正在等待连接，则向server的等待连接的队列中添加一项，并通知server有新连接到来，自己则睡眠等待连接建立。当server被唤醒被建立连接后会唤醒client进程，至此client跟server相互之间的连接建立。

注：babyos2建立连接很简单，就是设置一个指针，及设置下状态。

7.read/write

读写babyos实现的比较简单，每个socket会有一块类似前面实现的pipe的缓冲区，当要写时，会把数据写到对端的缓冲区，而读则是从自己的缓冲区读。

int32 socket_local_t::read(void* buf, uint32 size)
{
char* p = (char *) buf;
char ch;
for (uint32 i = 0; i < size; i++) {
if (m_sock_buf.get_char(ch) < 0) {
return -1;
}
*p++ = ch;
}

return size;
}

int32 socket_local_t::write(void* buf, uint32 size)
{
sock_buffer_t* connect_sock_buf = &((socket_local_t *) (m_connected_socket))->m_sock_buf;
char* p = (char *) buf;
for (uint32 i = 0; i < size; i++) {
if (connect_sock_buf->put_char(*p++) < 0) {
return -1;
}
}

return size;
}

/*
* guzhoudiaoke@126.com
* 2018-01-20
*/

#include "socket.h"
#include "babyos.h"

void sock_buffer_t::init(socket_t* socket)
{
m_socket = socket;
m_read_index = 0;
m_write_index = 0;
m_lock.init();
m_wait_space.init(SOCK_BUF_SIZE);
m_wait_item.init(0);
}

int sock_buffer_t::get_char(char& ch)
{
int ret = -1;
m_wait_item.down();
m_lock.lock();
if (m_socket->m_state == socket_t::SS_CONNECTED) {
ch = m_buffer[m_read_index];
m_read_index = (m_read_index + 1) % SOCK_BUF_SIZE;
ret = 0;
}
m_lock.unlock();
m_wait_space.up();

return ret;
}

int sock_buffer_t::put_char(char ch)
{
int ret = -1;
m_wait_space.down();
m_lock.lock();
if (m_socket->m_state == socket_t::SS_CONNECTED) {
m_buffer[m_write_index] = ch;
m_write_index = (m_write_index + 1) % SOCK_BUF_SIZE;
ret = 0;
}
m_lock.unlock();
m_wait_item.up();

return ret;
}

8.test

static void do_server(int sockfd)
{
int data = 0;
if (userlib_t::read(sockfd, &data, sizeof(int)) < 0) {
userlib_t::printf("server read error.\n");
return;
}
userlib_t::printf("server read  %d from client.\n", data);
data++;

if (userlib_t::write(sockfd, &data, sizeof(int)) < 0) {
userlib_t::printf("server write error.\n");
return;
}
userlib_t::printf("server write %d to   client.\n", data);
}

static void socket_server()
{
int listen_fd = userlib_t::socket(socket_t::AF_LOCAL, 0, 0);
if (listen_fd < 0) {
userlib_t::printf("err, server create socket failed, error %u\n", listen_fd);
return;
}
userlib_t::printf("server create socket succ: %u\n", listen_fd);

sock_addr_local_t addr;
addr.m_family = socket_t::AF_LOCAL;
userlib_t::strcpy(addr.m_path, "/test_socket");

int ret = 0;
if ((ret = userlib_t::bind(listen_fd, &addr)) < 0) {
userlib_t::printf("err, server bind to %u failed, error %u\n", listen_fd, ret);
return;
}
userlib_t::printf("server bind succ\n");

if ((ret = userlib_t::listen(listen_fd, 1)) < 0) {
userlib_t::printf("err, server listen failed, error %u\n", ret);
return;
}
userlib_t::printf("server listen succ\n");

int conn_fd = -1;
sock_addr_local_t client_addr;
for (int i = 0; i < 2; i++) {
if ((conn_fd = userlib_t::accept(listen_fd, &client_addr)) < 0) {
userlib_t::printf("accept failed.\n");
continue;
}

userlib_t::printf("server accept success: %u\n", conn_fd);
if (userlib_t::fork() == 0) {
userlib_t::close(listen_fd);
do_server(conn_fd);
userlib_t::sleep(1);
userlib_t::exit(0);
}
else {
userlib_t::close(conn_fd);
}
}
}

static void do_client(int sockfd, int data)
{
userlib_t::write(sockfd, &data, sizeof(int));
userlib_t::printf("client write %d to   server.\n", data);

userlib_t::read(sockfd, &data, sizeof(int));
userlib_t::printf("client read  %d from server.\n", data);
}

static void socket_client(int data)
{
int sock_fd = userlib_t::socket(socket_t::AF_LOCAL, 0, 0);
if (sock_fd < 0) {
userlib_t::printf("client create socket failed, error %u\n", sock_fd);
return;
}
userlib_t::printf("client create socket success, fd: %u\n", sock_fd);

sock_addr_local_t addr;
addr.m_family = socket_t::AF_LOCAL;
userlib_t::strcpy(addr.m_path, "/test_socket");

int ret = 0;
if ((ret = userlib_t::connect(sock_fd, &addr)) < 0) {
userlib_t::printf("client connect to fd: %u failed, error %u\n", sock_fd, ret);
return;
}

userlib_t::printf("client connect success\n");
do_client(sock_fd, data);
}

static void test_socket()
{
int32 pid1 = -1;
int32 pid2 = -1;
int32 pid3 = -1;

pid1 = userlib_t::fork();
if (pid1 == 0) {
/* server */
socket_server();
userlib_t::exit(0);
}

userlib_t::sleep(1);
pid2 = userlib_t::fork();
if (pid2 == 0) {
/* client */
socket_client(1234);
userlib_t::sleep(1);
userlib_t::exit(0);
}

userlib_t::sleep(1);
pid3 = userlib_t::fork();
if (pid3 == 0) {
/* client */
socket_client(5678);
userlib_t::sleep(1);
userlib_t::exit(0);
}

/* shell */
userlib_t::wait(pid1);
userlib_t::wait(pid2);
userlib_t::wait(pid3);
}

然后写了一个简单的socket通信程序，会创建一个新进程server，server会等待两个连接，之后退出，另外创建两个client进程，会跟server建立连接。

连接建立后，两个client分别发送一个数字到server，server读到后将数字加1然后发会客户端。客户端接到后退出。