examples: add vanilla_http_server - a fast, multi-threaded, non-blocking, port and host reuse, thread-safe, epoll server (#23094)

cmd/tools/modules/testing/common.v

@@ -242,6 +242,7 @@ pub fn new_test_session(_vargs string, will_compile bool) TestSession {
 			skip_files << 'examples/database/psql/customer.v'
 		}
 		$if windows {
+			skip_files << 'examples/vanilla_http_server' // requires epoll
 			skip_files << 'examples/1brc/solution/main.v' // requires mmap
 			skip_files << 'examples/database/mysql.v'
 			skip_files << 'examples/database/orm.v'

cmd/tools/vbuild-examples.v

@@ -10,6 +10,7 @@ const efolders = [
 	'examples/viewer',
 	'examples/vweb_orm_jwt',
 	'examples/vweb_fullstack',
+	'examples/vanilla_http_server',
 ]
 
 pub fn normalised_vroot_path(path string) string {

examples/vanilla_http_server/.editorconfig

@@ -0,0 +1,8 @@
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.v]
+indent_style = tab

examples/vanilla_http_server/.gitattributes

@@ -0,0 +1,8 @@
+* text=auto eol=lf
+*.bat eol=crlf
+
+*.v linguist-language=V
+*.vv linguist-language=V
+*.vsh linguist-language=V
+v.mod linguist-language=V
+.vdocignore linguist-language=ignore

examples/vanilla_http_server/.gitignore

@@ -0,0 +1,23 @@
+# Binaries for programs and plugins
+main
+*.exe
+*.exe~
+*.so
+*.dylib
+*.dll
+
+# Ignore binary output folders
+bin/
+
+# Ignore common editor/system specific metadata
+.DS_Store
+.idea/
+.vscode/
+*.iml
+
+# ENV
+.env
+
+# vweb and database
+*.db
+*.js

examples/vanilla_http_server/README.md

@@ -0,0 +1,101 @@
+# Vanilla
+
+Vanilla is a raw V server.
+
+## Description
+
+This project is a simple server written in the V programming language.
+It aims to provide a minimalistic and efficient server implementation.
+
+## Features
+
+- Lightweight and fast
+- Minimal dependencies
+- Easy to understand and extend
+
+## Installation
+
+To install Vanilla, you need to have the V compiler installed.
+You can download it from the [official V website](https://vlang.io).
+
+## Usage
+
+To run the server, use the following command:
+
+```sh
+v -prod crun .
+```
+
+This will start the server, and you can access it at `http://localhost:3000`.
+
+## Code Overview
+
+### Main Server
+
+The main server logic is implemented in [src/main.v](v/vanilla/src/main.v).
+The server is initialized and started in the `main` function:
+
+```v ignore
+module main
+
+const port = 3000
+
+fn main() {
+	mut server := Server{
+		router: setup_router()
+	}
+
+	server.socket_fd = create_server_socket(port)
+	if server.socket_fd < 0 {
+		return
+	}
+	server.epoll_fd = C.epoll_create1(0)
+	if server.epoll_fd < 0 {
+		C.perror('epoll_create1 failed'.str)
+		C.close(server.socket_fd)
+		return
+	}
+
+	server.lock_flag.lock()
+	if add_fd_to_epoll(server.epoll_fd, server.socket_fd, u32(C.EPOLLIN)) == -1 {
+		C.close(server.socket_fd)
+		C.close(server.epoll_fd)
+
+		server.lock_flag.unlock()
+		return
+	}
+
+	server.lock_flag.unlock()
+
+	server.lock_flag.init()
+	for i := 0; i < 16; i++ {
+		server.threads[i] = spawn process_events(&server)
+	}
+	println('listening on http://localhost:${port}/')
+	event_loop(&server)
+}
+```
+
+## Test
+
+### CURL
+
+```sh
+curl -X GET --verbose http://localhost:3000/ &&
+curl -X POST --verbose http://localhost:3000/user &&
+curl -X GET --verbose http://localhost:3000/user/1
+
+```
+
+### WRK
+
+```sh
+wrk --connection 512 --threads 16 --duration 10s http://localhost:3000
+```
+
+### Valgrind
+```sh
+# Race condition check
+v -prod -gc none . 
+valgrind --tool=helgrind ./vanilla_http_server
+```

examples/vanilla_http_server/src/controllers.v

@@ -0,0 +1,42 @@
+module main
+
+import strings
+
+const http_ok_response = 'HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n'.bytes()
+
+const http_created_response = 'HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n'.bytes()
+
+fn home_controller(params []string) ![]u8 {
+	return http_ok_response
+}
+
+fn get_users_controller(params []string) ![]u8 {
+	return http_ok_response
+}
+
+@[direct_array_access; manualfree]
+fn get_user_controller(params []string) ![]u8 {
+	if params.len == 0 {
+		return tiny_bad_request_response
+	}
+	id := params[0]
+	response_body := id
+
+	mut sb := strings.new_builder(200)
+	sb.write_string('HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: ')
+	sb.write_string(response_body.len.str())
+	sb.write_string('\r\nConnection: keep-alive\r\n\r\n')
+	sb.write_string(response_body)
+
+	defer {
+		unsafe {
+			response_body.free()
+			params.free()
+		}
+	}
+	return sb
+}
+
+fn create_user_controller(params []string) ![]u8 {
+	return http_created_response
+}

examples/vanilla_http_server/src/main.c.v

@@ -0,0 +1,32 @@
+module main
+
+// handle_request finds and executes the handler for a given route.
+// It takes an HttpRequest object as an argument and returns the response as a byte array.
+fn handle_request(req HttpRequest) ![]u8 {
+	method := unsafe { tos(&req.buffer[req.method.start], req.method.len) }
+	path := unsafe { tos(&req.buffer[req.path.start], req.path.len) }
+
+	if method == 'GET' {
+		if path == '/' {
+			return home_controller([])
+		} else if path.starts_with('/user/') {
+			id := path[6..]
+			return get_user_controller([id])
+		}
+	} else if method == 'POST' {
+		if path == '/user' {
+			return create_user_controller([])
+		}
+	}
+
+	return tiny_bad_request_response
+}
+
+fn main() {
+	mut server := Server{
+		request_handler: handle_request
+		port:            3001
+	}
+
+	server.run()
+}

examples/vanilla_http_server/src/request_parser.v

@@ -0,0 +1,71 @@
+module main
+
+struct Slice {
+	start int
+	len   int
+}
+
+struct HttpRequest {
+mut:
+	buffer  []u8
+	method  Slice
+	path    Slice
+	version Slice
+}
+
+@[direct_array_access]
+fn parse_request_line(mut req HttpRequest) ! {
+	mut i := 0
+	// Parse HTTP method
+	for i < req.buffer.len && req.buffer[i] != ` ` {
+		i++
+	}
+	req.method = Slice{
+		start: 0
+		len:   i
+	}
+	i++
+
+	// Parse path
+	mut path_start := i
+	for i < req.buffer.len && req.buffer[i] != ` ` {
+		i++
+	}
+	req.path = Slice{
+		start: path_start
+		len:   i - path_start
+	}
+	i++
+
+	// Parse HTTP version
+	mut version_start := i
+	for i < req.buffer.len && req.buffer[i] != `\r` {
+		i++
+	}
+	req.version = Slice{
+		start: version_start
+		len:   i - version_start
+	}
+
+	// Move to the end of the request line
+	if i + 1 < req.buffer.len && req.buffer[i] == `\r` && req.buffer[i + 1] == `\n` {
+		i += 2
+	} else {
+		return error('Invalid HTTP request line')
+	}
+}
+
+fn decode_http_request(buffer []u8) !HttpRequest {
+	mut req := HttpRequest{
+		buffer: buffer
+	}
+
+	parse_request_line(mut req)!
+
+	return req
+}
+
+// Helper function to convert Slice to string for debugging
+fn slice_to_string(buffer []u8, s Slice) string {
+	return buffer[s.start..s.start + s.len].bytestr()
+}

examples/vanilla_http_server/src/server.c.v

@@ -0,0 +1,282 @@
+// This module implements a basic HTTP server using epoll for handling multiple client connections efficiently.
+// The server is designed to be non-blocking and uses multiple threads to handle incoming requests concurrently.
+//
+// Performance Considerations:
+// - Non-blocking I/O: The server uses non-blocking sockets to ensure that it can handle multiple connections without being blocked by any single connection.
+// - Epoll: The use of epoll allows the server to efficiently monitor multiple file descriptors to see if I/O is possible on any of them.
+// - Threading: The server spawns multiple threads to handle client requests, which can improve performance on multi-core systems.
+// - Memory Management: Care is taken to allocate and free memory appropriately to avoid memory leaks and ensure efficient memory usage.
+// - Error Handling: The server includes error handling to manage and log errors without crashing, ensuring robustness and reliability.
+// - SO_REUSEPORT: The server sets the SO_REUSEPORT socket option to allow multiple sockets on the same host and port, which can improve performance in certain scenarios.
+// - Connection Handling: The server efficiently handles client connections, including accepting new connections, reading requests, and sending responses.
+// - Mutex Locking: The server uses mutex locks to manage access to shared resources, ensuring thread safety while minimizing contention.
+module main
+
+const tiny_bad_request_response = 'HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n'.bytes()
+
+#include <fcntl.h>
+#include <errno.h>
+
+$if !windows {
+	#include <sys/epoll.h>
+	#include <netinet/in.h>
+}
+
+fn C.socket(socket_family int, socket_type int, protocol int) int
+
+fn C.bind(sockfd int, addr &C.sockaddr_in, addrlen u32) int
+
+fn C.send(__fd int, __buf voidptr, __n usize, __flags int) int
+
+fn C.recv(__fd int, __buf voidptr, __n usize, __flags int) int
+
+fn C.setsockopt(__fd int, __level int, __optname int, __optval voidptr, __optlen u32) int
+
+fn C.listen(__fd int, __n int) int
+
+fn C.perror(s &u8)
+
+fn C.close(fd int) int
+
+fn C.accept(sockfd int, address &C.sockaddr_in, addrlen &u32) int
+
+fn C.htons(__hostshort u16) u16
+
+fn C.epoll_create1(__flags int) int
+
+fn C.epoll_ctl(__epfd int, __op int, __fd int, __event &C.epoll_event) int
+
+fn C.epoll_wait(__epfd int, __events &C.epoll_event, __maxevents int, __timeout int) int
+
+fn C.fcntl(fd int, cmd int, arg int) int
+
+struct C.in_addr {
+	s_addr u32
+}
+
+struct C.sockaddr_in {
+	sin_family u16
+	sin_port   u16
+	sin_addr   C.in_addr
+	sin_zero   [8]u8
+}
+
+union C.epoll_data {
+	ptr voidptr
+	fd  int
+	u32 u32
+	u64 u64
+}
+
+struct C.epoll_event {
+	events u32
+	data   C.epoll_data
+}
+
+struct Server {
+pub:
+	port int = 3000
+mut:
+	socket_fd       int
+	epoll_fds       [max_thread_pool_size]int
+	threads         [max_thread_pool_size]thread
+	request_handler fn (HttpRequest) ![]u8 @[required]
+}
+
+const max_connection_size = 1024
+
+const max_thread_pool_size = 8
+
+fn set_blocking(fd int, blocking bool) {
+	flags := C.fcntl(fd, C.F_GETFL, 0)
+	if flags == -1 {
+		// TODO: better error handling
+		eprintln(@LOCATION)
+		return
+	}
+	if blocking {
+		// This removes the O_NONBLOCK flag from flags and set it.
+		C.fcntl(fd, C.F_SETFL, flags & ~C.O_NONBLOCK)
+	} else {
+		// This adds the O_NONBLOCK flag from flags and set it.
+		C.fcntl(fd, C.F_SETFL, flags | C.O_NONBLOCK)
+	}
+}
+
+fn close_socket(fd int) {
+	C.close(fd)
+}
+
+fn create_server_socket(port int) int {
+	// Create a socket with non-blocking mode
+	server_fd := C.socket(C.AF_INET, C.SOCK_STREAM, 0)
+	if server_fd < 0 {
+		eprintln(@LOCATION)
+		C.perror('Socket creation failed'.str)
+		return -1
+	}
+
+	// set_blocking(server_fd, false)
+
+	// Enable SO_REUSEPORT
+	opt := 1
+	if C.setsockopt(server_fd, C.SOL_SOCKET, C.SO_REUSEPORT, &opt, sizeof(opt)) < 0 {
+		eprintln(@LOCATION)
+		C.perror('setsockopt SO_REUSEPORT failed'.str)
+		close_socket(server_fd)
+		return -1
+	}
+
+	server_addr := C.sockaddr_in{
+		sin_family: u16(C.AF_INET)
+		sin_port:   C.htons(port)
+		sin_addr:   C.in_addr{u32(C.INADDR_ANY)}
+		sin_zero:   [8]u8{}
+	}
+
+	if C.bind(server_fd, voidptr(&server_addr), sizeof(server_addr)) < 0 {
+		eprintln(@LOCATION)
+		C.perror('Bind failed'.str)
+		close_socket(server_fd)
+		return -1
+	}
+	if C.listen(server_fd, max_connection_size) < 0 {
+		eprintln(@LOCATION)
+		C.perror('Listen failed'.str)
+		close_socket(server_fd)
+		return -1
+	}
+	return server_fd
+}
+
+// Function to add a file descriptor to the epoll instance
+fn add_fd_to_epoll(epoll_fd int, fd int, events u32) int {
+	mut ev := C.epoll_event{
+		events: events
+	}
+	ev.data.fd = fd
+	if C.epoll_ctl(epoll_fd, C.EPOLL_CTL_ADD, fd, &ev) == -1 {
+		eprintln(@LOCATION)
+		C.perror('epoll_ctl'.str)
+		return -1
+	}
+	return 0
+}
+
+// Function to remove a file descriptor from the epoll instance
+fn remove_fd_from_epoll(epoll_fd int, fd int) {
+	C.epoll_ctl(epoll_fd, C.EPOLL_CTL_DEL, fd, C.NULL)
+}
+
+fn handle_accept_loop(mut server Server) {
+	for {
+		client_fd := C.accept(server.socket_fd, C.NULL, C.NULL)
+		if client_fd < 0 {
+			// Check for EAGAIN or EWOULDBLOCK, usually represented by errno 11.
+			if C.errno == C.EAGAIN || C.errno == C.EWOULDBLOCK {
+				break // No more incoming connections; exit loop.
+			}
+
+			eprintln(@LOCATION)
+			C.perror('Accept failed'.str)
+			return
+		}
+
+		unsafe {
+			// Distribute client connections among epoll_fds
+			epoll_fd := server.epoll_fds[client_fd % max_thread_pool_size]
+			if add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 {
+				close_socket(client_fd)
+			}
+		}
+	}
+}
+
+fn handle_client_closure(server &Server, client_fd int) {
+	unsafe {
+		remove_fd_from_epoll(client_fd, client_fd)
+	}
+}
+
+fn process_events(mut server Server, epoll_fd int) {
+	for {
+		events := [max_connection_size]C.epoll_event{}
+		num_events := C.epoll_wait(epoll_fd, &events[0], max_connection_size, -1)
+		for i := 0; i < num_events; i++ {
+			if events[i].events & u32((C.EPOLLHUP | C.EPOLLERR)) != 0 {
+				handle_client_closure(server, unsafe { events[i].data.fd })
+				continue
+			}
+			if events[i].events & u32(C.EPOLLIN) != 0 {
+				request_buffer := [140]u8{}
+				bytes_read := C.recv(unsafe { events[i].data.fd }, &request_buffer[0],
+					140 - 1, 0)
+				if bytes_read > 0 {
+					mut readed_request_buffer := []u8{cap: bytes_read}
+
+					unsafe {
+						readed_request_buffer.push_many(&request_buffer[0], bytes_read)
+					}
+
+					decoded_http_request := decode_http_request(readed_request_buffer) or {
+						eprintln('Error decoding request ${err}')
+						C.send(unsafe { events[i].data.fd }, tiny_bad_request_response.data,
+							tiny_bad_request_response.len, 0)
+						handle_client_closure(server, unsafe { events[i].data.fd })
+						continue
+					}
+
+					// This lock is a workaround for avoiding race condition in router.params
+					// This slows down the server, but it's a temporary solution
+
+					response_buffer := server.request_handler(decoded_http_request) or {
+						eprintln('Error handling request ${err}')
+						C.send(unsafe { events[i].data.fd }, tiny_bad_request_response.data,
+							tiny_bad_request_response.len, 0)
+						handle_client_closure(server, unsafe { events[i].data.fd })
+
+						continue
+					}
+
+					C.send(unsafe { events[i].data.fd }, response_buffer.data, response_buffer.len,
+						0)
+					handle_client_closure(server, unsafe { events[i].data.fd })
+				} else if bytes_read == 0
+					|| (bytes_read < 0 && C.errno != C.EAGAIN && C.errno != C.EWOULDBLOCK) {
+					handle_client_closure(server, unsafe { events[i].data.fd })
+				}
+			}
+		}
+	}
+}
+
+fn (mut server Server) run() {
+	$if windows {
+		eprintln('Windows is not supported yet')
+		return
+	}
+	server.socket_fd = create_server_socket(server.port)
+	if server.socket_fd < 0 {
+		return
+	}
+
+	for i := 0; i < max_thread_pool_size; i++ {
+		server.epoll_fds[i] = C.epoll_create1(0)
+		if server.epoll_fds[i] < 0 {
+			C.perror('epoll_create1 failed'.str)
+			close_socket(server.socket_fd)
+			return
+		}
+
+		if add_fd_to_epoll(server.epoll_fds[i], server.socket_fd, u32(C.EPOLLIN)) == -1 {
+			close_socket(server.socket_fd)
+			close_socket(server.epoll_fds[i])
+
+			return
+		}
+		server.threads[i] = spawn process_events(mut server, server.epoll_fds[i])
+	}
+
+	println('listening on http://localhost:${server.port}/')
+	handle_accept_loop(mut server)
+}

examples/vanilla_http_server/v.mod

@@ -0,0 +1,7 @@
+Module {
+	name: 'vanilla'
+	description: 'A raw V server'
+	version: '0.0.1'
+	license: 'MIT'
+	dependencies: []
+}