net: move unix socket code to net.unix and skip net/unix/unix_test.v … (#8681)

vlib/net/unix/aasocket.c.v

@@ -0,0 +1,97 @@
+module unix
+
+// Select represents a select operation
+enum Select {
+	read
+	write
+	except
+}
+
+// SocketType are the available sockets
+enum SocketType {
+	dgram = C.SOCK_DGRAM
+	stream = C.SOCK_STREAM
+	seqpacket = C.SOCK_SEQPACKET
+}
+
+struct C.sockaddr {
+	sa_family u16
+}
+
+struct C.sockaddr_un {
+mut:
+	sun_family int
+	sun_path   charptr
+}
+
+struct C.addrinfo {
+mut:
+	ai_family    int
+	ai_socktype  int
+	ai_flags     int
+	ai_protocol  int
+	ai_addrlen   int
+	ai_addr      voidptr
+	ai_canonname voidptr
+	ai_next      voidptr
+}
+
+struct C.sockaddr_storage {
+}
+
+fn C.socket() int
+
+fn C.setsockopt() int
+
+fn C.htonl() int
+
+fn C.htons() int
+
+fn C.bind() int
+
+fn C.listen() int
+
+fn C.accept() int
+
+fn C.getaddrinfo() int
+
+fn C.connect() int
+
+fn C.send() int
+
+fn C.sendto() int
+
+fn C.recv() int
+
+fn C.recvfrom() int
+
+fn C.shutdown() int
+
+fn C.ntohs() int
+
+fn C.getpeername() int
+
+fn C.inet_ntop(af int, src voidptr, dst charptr, dst_size int) charptr
+
+fn C.WSAAddressToStringA() int
+
+fn C.getsockname() int
+
+// defined in builtin
+// fn C.read() int
+// fn C.close() int
+
+fn C.ioctlsocket() int
+
+fn C.fcntl() int
+
+fn C.@select() int
+
+fn C.FD_ZERO()
+
+fn C.FD_SET()
+
+fn C.FD_ISSET() bool
+
+[typedef]
+struct C.fd_set {}

vlib/net/unix/common.v

@@ -0,0 +1,127 @@
+module unix
+
+import time
+import net
+
+fn C.SUN_LEN(C.sockaddr_un) int
+
+fn C.strncpy(charptr, charptr, int)
+
+// Shutdown shutsdown a socket and closes it
+fn shutdown(handle int) ? {
+	$if windows {
+		C.shutdown(handle, C.SD_BOTH)
+		net.socket_error(C.closesocket(handle)) ?
+	} $else {
+		C.shutdown(handle, C.SHUT_RDWR)
+		net.socket_error(C.close(handle)) ?
+	}
+
+	return none
+}
+
+// Select waits for an io operation (specified by parameter `test`) to be available
+fn @select(handle int, test Select, timeout time.Duration) ?bool {
+	set := C.fd_set{}
+
+	C.FD_ZERO(&set)
+	C.FD_SET(handle, &set)
+
+	seconds := timeout.milliseconds() / 1000
+	microseconds := timeout - (seconds * time.second)
+
+	mut tt := C.timeval{
+		tv_sec: u64(seconds)
+		tv_usec: u64(microseconds)
+	}
+
+	mut timeval_timeout := &tt
+
+	// infinite timeout is signaled by passing null as the timeout to
+	// select
+	if timeout == unix.infinite_timeout {
+		timeval_timeout = &C.timeval(0)
+	}
+
+	match test {
+		.read {
+			net.socket_error(C.@select(handle + 1, &set, C.NULL, C.NULL, timeval_timeout)) ?
+		}
+		.write {
+			net.socket_error(C.@select(handle + 1, C.NULL, &set, C.NULL, timeval_timeout)) ?
+		}
+		.except {
+			net.socket_error(C.@select(handle + 1, C.NULL, C.NULL, &set, timeval_timeout)) ?
+		}
+	}
+
+	return C.FD_ISSET(handle, &set)
+}
+
+// wait_for_common wraps the common wait code
+fn wait_for_common(handle int, deadline time.Time, timeout time.Duration, test Select) ? {
+	if deadline.unix == 0 {
+		// only accept infinite_timeout as a valid
+		// negative timeout - it is handled in @select however
+		if timeout < 0 && timeout != unix.infinite_timeout {
+			return net.err_timed_out
+		}
+		ready := @select(handle, test, timeout) ?
+		if ready {
+			return none
+		}
+		return net.err_timed_out
+	}
+	// Convert the deadline into a timeout
+	// and use that
+	d_timeout := deadline.unix - time.now().unix
+	if d_timeout < 0 {
+		// deadline is in the past so this has already
+		// timed out
+		return net.err_timed_out
+	}
+
+	ready := @select(handle, test, d_timeout) ?
+	if ready {
+		return none
+	}
+	return net.err_timed_out
+}
+
+// wait_for_write waits for a write io operation to be available
+fn wait_for_write(handle int, deadline time.Time, timeout time.Duration) ? {
+	return wait_for_common(handle, deadline, timeout, .write)
+}
+
+// wait_for_read waits for a read io operation to be available
+fn wait_for_read(handle int, deadline time.Time, timeout time.Duration) ? {
+	return wait_for_common(handle, deadline, timeout, .read)
+}
+
+// no_deadline should be given to functions when no deadline is wanted (i.e. all functions
+// return instantly)
+const (
+	no_deadline = time.Time{
+		unix: 0
+	}
+)
+
+// no_timeout should be given to functions when no timeout is wanted (i.e. all functions
+// return instantly)
+const (
+	no_timeout = time.Duration(0)
+)
+
+// infinite_timeout should be given to functions when an infinite_timeout is wanted (i.e. functions
+// only ever return with data)
+const (
+	infinite_timeout = time.Duration(-1)
+)
+
+[inline]
+fn wrap_read_result(result int) ?int {
+	if result != 0 {
+		return result
+	}
+	return none
+}

vlib/net/unix/stream_nix.v

@@ -0,0 +1,288 @@
+module unix
+
+import time
+import os
+import net
+
+const (
+	unix_default_read_timeout  = 30 * time.second
+	unix_default_write_timeout = 30 * time.second
+	connect_timeout            = 5 * time.second
+	msg_nosignal               = 0x4000
+)
+
+struct StreamSocket {
+pub:
+	handle int
+mut:
+	path string
+}
+
+struct StreamConn {
+pub mut:
+	sock StreamSocket
+mut:
+	write_deadline time.Time
+	read_deadline  time.Time
+	read_timeout   time.Duration
+	write_timeout  time.Duration
+}
+
+struct StreamListener {
+pub mut:
+	sock StreamSocket
+mut:
+	accept_timeout  time.Duration
+	accept_deadline time.Time
+}
+
+fn error_code() int {
+	return C.errno
+}
+
+fn new_stream_socket() ?StreamSocket {
+	sockfd := net.socket_error(C.socket(C.AF_UNIX, SocketType.stream, 0)) ?
+	mut s := StreamSocket{
+		handle: sockfd
+	}
+	return s
+}
+
+fn (mut s StreamSocket) close() ? {
+	os.rm(s.path) ?
+	return shutdown(s.handle)
+}
+
+fn (mut s StreamSocket) @select(test Select, timeout time.Duration) ?bool {
+	return @select(s.handle, test, timeout)
+}
+
+fn (mut s StreamSocket) connect(a string) ? {
+	if a.len >= 108 {
+		return error('Socket path too long! Max length: 107 chars.')
+	}
+	mut addr := C.sockaddr_un{}
+	unsafe { C.memset(&addr, 0, sizeof(C.sockaddr_un)) }
+	addr.sun_family = C.AF_UNIX
+	C.strncpy(addr.sun_path, a.str, 108)
+	size := C.SUN_LEN(&addr)
+	sockaddr := unsafe { &C.sockaddr(&addr) }
+	res := C.connect(s.handle, sockaddr, size)
+	// if res != 1 {
+	// return none
+	//}
+	if res == 0 {
+		return none
+	}
+	_ := error_code()
+	write_result := s.@select(.write, unix.connect_timeout) ?
+	if write_result {
+		// succeeded
+		return none
+	}
+	except_result := s.@select(.except, unix.connect_timeout) ?
+	if except_result {
+		return net.err_connect_failed
+	}
+	// otherwise we timed out
+	return net.err_connect_timed_out
+}
+
+pub fn listen_stream(sock string) ?&StreamListener {
+	if sock.len >= 108 {
+		return error('Socket path too long! Max length: 107 chars.')
+	}
+	mut s := new_stream_socket() ?
+	s.path = sock
+	mut addr := C.sockaddr_un{}
+	unsafe { C.memset(&addr, 0, sizeof(C.sockaddr_un)) }
+	addr.sun_family = C.AF_UNIX
+	C.strncpy(addr.sun_path, sock.str, 108)
+	size := C.SUN_LEN(&addr)
+	sockaddr := unsafe { &C.sockaddr(&addr) }
+	net.socket_error(C.bind(s.handle, sockaddr, size)) ?
+	net.socket_error(C.listen(s.handle, 128)) ?
+	return &StreamListener{
+		sock: s
+	}
+}
+
+pub fn connect_stream(path string) ?&StreamConn {
+	mut s := new_stream_socket() ?
+	s.connect(path) ?
+	return &StreamConn{
+		sock: s
+		read_timeout: unix.unix_default_read_timeout
+		write_timeout: unix.unix_default_write_timeout
+	}
+}
+
+pub fn (mut l StreamListener) accept() ?&StreamConn {
+	mut new_handle := C.accept(l.sock.handle, 0, 0)
+	if new_handle <= 0 {
+		l.wait_for_accept() ?
+		new_handle = C.accept(l.sock.handle, 0, 0)
+		if new_handle == -1 || new_handle == 0 {
+			return none
+		}
+	}
+	new_sock := StreamSocket{
+		handle: new_handle
+	}
+	return &StreamConn{
+		sock: new_sock
+		read_timeout: unix.unix_default_read_timeout
+		write_timeout: unix.unix_default_write_timeout
+	}
+}
+
+pub fn (c &StreamListener) accept_deadline() ?time.Time {
+	if c.accept_deadline.unix != 0 {
+		return c.accept_deadline
+	}
+	return none
+}
+
+pub fn (mut c StreamListener) set_accept_deadline(deadline time.Time) {
+	c.accept_deadline = deadline
+}
+
+pub fn (c &StreamListener) accept_timeout() time.Duration {
+	return c.accept_timeout
+}
+
+pub fn (mut c StreamListener) set_accept_timeout(t time.Duration) {
+	c.accept_timeout = t
+}
+
+pub fn (mut c StreamListener) wait_for_accept() ? {
+	return wait_for_read(c.sock.handle, c.accept_deadline, c.accept_timeout)
+}
+
+pub fn (mut c StreamListener) close() ? {
+	c.sock.close() ?
+	return none
+}
+
+pub fn (mut c StreamConn) close() ? {
+	c.sock.close() ?
+	return none
+}
+
+// write_ptr blocks and attempts to write all data
+pub fn (mut c StreamConn) write_ptr(b byteptr, len int) ? {
+	$if trace_unix ? {
+		eprintln(
+			'>>> StreamConn.write_ptr | c.sock.handle: $c.sock.handle | b: ${ptr_str(b)} len: $len |\n' +
+			unsafe { b.vstring_with_len(len) })
+	}
+	unsafe {
+		mut ptr_base := byteptr(b)
+		mut total_sent := 0
+		for total_sent < len {
+			ptr := ptr_base + total_sent
+			remaining := len - total_sent
+			mut sent := C.send(c.sock.handle, ptr, remaining, unix.msg_nosignal)
+			if sent < 0 {
+				code := error_code()
+				if code == int(net.error_ewouldblock) {
+					c.wait_for_write() ?
+					continue
+				} else {
+					net.wrap_error(code) ?
+				}
+			}
+			total_sent += sent
+		}
+	}
+	return none
+}
+
+// write blocks and attempts to write all data
+pub fn (mut c StreamConn) write(bytes []byte) ? {
+	return c.write_ptr(bytes.data, bytes.len)
+}
+
+// write_str blocks and attempts to write all data
+pub fn (mut c StreamConn) write_str(s string) ? {
+	return c.write_ptr(s.str, s.len)
+}
+
+pub fn (mut c StreamConn) read_ptr(buf_ptr byteptr, len int) ?int {
+	mut res := wrap_read_result(C.recv(c.sock.handle, buf_ptr, len, 0)) ?
+	$if trace_unix ? {
+		eprintln('<<< StreamConn.read_ptr  | c.sock.handle: $c.sock.handle | buf_ptr: ${ptr_str(buf_ptr)} len: $len | res: $res')
+	}
+	if res > 0 {
+		return res
+	}
+	code := error_code()
+	if code == int(net.error_ewouldblock) {
+		c.wait_for_read() ?
+		res = wrap_read_result(C.recv(c.sock.handle, buf_ptr, len, 0)) ?
+		$if trace_unix ? {
+			eprintln('<<< StreamConn.read_ptr  | c.sock.handle: $c.sock.handle | buf_ptr: ${ptr_str(buf_ptr)} len: $len | res: $res')
+		}
+		return net.socket_error(res)
+	} else {
+		net.wrap_error(code) ?
+	}
+	return none
+}
+
+pub fn (mut c StreamConn) read(mut buf []byte) ?int {
+	return c.read_ptr(buf.data, buf.len)
+}
+
+pub fn (mut c StreamConn) read_deadline() ?time.Time {
+	if c.read_deadline.unix == 0 {
+		return c.read_deadline
+	}
+	return none
+}
+
+pub fn (mut c StreamConn) set_read_deadline(deadline time.Time) {
+	c.read_deadline = deadline
+}
+
+pub fn (mut c StreamConn) write_deadline() ?time.Time {
+	if c.write_deadline.unix == 0 {
+		return c.write_deadline
+	}
+	return none
+}
+
+pub fn (mut c StreamConn) set_write_deadline(deadline time.Time) {
+	c.write_deadline = deadline
+}
+
+pub fn (c &StreamConn) read_timeout() time.Duration {
+	return c.read_timeout
+}
+
+pub fn (mut c StreamConn) set_read_timeout(t time.Duration) {
+	c.read_timeout = t
+}
+
+pub fn (c &StreamConn) write_timeout() time.Duration {
+	return c.write_timeout
+}
+
+pub fn (mut c StreamConn) set_write_timeout(t time.Duration) {
+	c.write_timeout = t
+}
+
+[inline]
+pub fn (mut c StreamConn) wait_for_read() ? {
+	return wait_for_read(c.sock.handle, c.read_deadline, c.read_timeout)
+}
+
+[inline]
+pub fn (mut c StreamConn) wait_for_write() ? {
+	return wait_for_write(c.sock.handle, c.write_deadline, c.write_timeout)
+}
+
+pub fn (c StreamConn) str() string {
+	s := c.sock.str().replace('\n', ' ').replace('  ', ' ')
+	return 'StreamConn{ write_deadline: $c.write_deadline, read_deadline: $c.read_deadline, read_timeout: $c.read_timeout, write_timeout: $c.write_timeout, sock: $s }'
+}

vlib/net/unix/unix_test.v

@@ -0,0 +1,51 @@
+import net.unix
+
+const (
+	test_port = 'test'
+)
+
+fn handle_conn(mut c unix.StreamConn) {
+	for {
+		mut buf := []byte{len: 100, init: 0}
+		read := c.read(mut buf) or {
+			println('Server: connection dropped')
+			return
+		}
+		c.write(buf[..read]) or {
+			println('Server: connection dropped')
+			return
+		}
+	}
+}
+
+fn echo_server(mut l unix.StreamListener) ? {
+	for {
+		mut new_conn := l.accept() or { continue }
+		go handle_conn(mut new_conn)
+	}
+	return none
+}
+
+fn echo() ? {
+	mut c := unix.connect_stream('test') ?
+	defer {
+		c.close() or { }
+	}
+	data := 'Hello from vlib/net!'
+	c.write_str(data) ?
+	mut buf := []byte{len: 4096}
+	read := c.read(mut buf) ?
+	assert read == data.len
+	for i := 0; i < read; i++ {
+		assert buf[i] == data[i]
+	}
+	println('Got "$buf.bytestr()"')
+	return none
+}
+
+fn test_tcp() {
+	mut l := unix.listen_stream(test_port) or { panic(err) }
+	go echo_server(mut l)
+	echo() or { panic(err) }
+	l.close() or { }
+}