v/vlib/picoev/picoev.v

// Copyright (c) 2019-2020 Alexander Medvednikov. All rights reserved.
// Use of this source code is governed by an MIT license
// that can be found in the LICENSE file.
module picoev

import picohttpparser

#include <errno.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <signal.h>

#flag -I @VROOT/thirdparty/picoev
#flag -L @VROOT/thirdparty/picoev
#flag @VROOT/thirdparty/picoev/picoev.o

#include "src/picoev.h"

const (
	max_fds = 1024
	timeout_secs = 8
	max_timeout = 10
	max_read = 4096
	max_write = 8192
)

struct C.in_addr {
mut:
	s_addr int
}

struct C.sockaddr_in {
mut:
	sin_family int
	sin_port   int
	sin_addr   C.in_addr
}

struct C.sockaddr_storage {}

fn C.atoi() int
fn C.strncasecmp() int
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.recv() int
//fn C.read() int
fn C.shutdown() int
//fn C.close() int
fn C.ntohs() int
fn C.getsockname() int

fn C.fcntl() int
//fn C.write() int

struct C.picoev_loop {}

struct Picoev {
	loop &C.picoev_loop
	cb   fn(req picohttpparser.Request, mut res picohttpparser.Response)
mut:
	date byteptr
	buf  byteptr
	idx  [1024]int
	out  byteptr
	oidx [1024]int
}

fn C.picoev_del(&C.picoev_loop, int) int
fn C.picoev_set_timeout(&C.picoev_loop, int, int)
fn C.picoev_add(&C.picoev_loop, int, int, int, &C.picoev_handler, voidptr) int
fn C.picoev_init(int) int
fn C.picoev_create_loop(int) &C.picoev_loop
fn C.picoev_loop_once(&C.picoev_loop, int) int
fn C.picoev_destroy_loop(&C.picoev_loop) int
fn C.picoev_deinit() int
fn C.phr_parse_request() int
fn C.phr_parse_request_path_pipeline() int
fn C.phr_parse_request_path() int

[inline]
fn setup_sock(fd int) {
	on := 1
	if C.setsockopt(fd, C.IPPROTO_TCP, C.TCP_NODELAY, &on, sizeof(int)) < 0 {
		println('setup_sock.setup_sock failed')
	}
	if C.fcntl(fd, C.F_SETFL, C.O_NONBLOCK) != 0 {
		println('fcntl failed')
	}
}

[inline]
fn close_conn(loop &C.picoev_loop, fd int) {
	C.picoev_del(loop, fd)
	C.close(fd)
}

[inline]
fn myread(fd int, b byteptr, max_len, idx int) int {
	unsafe {
		return C.read(fd, b + idx, max_len - idx)
	}
}

[inline]
fn mysubstr(s byteptr, from, len int) string {
	unsafe {
		return tos(s + from, len)
	}
}

fn rw_callback(loop &C.picoev_loop, fd, events int, cb_arg voidptr) {
	mut p := &Picoev(cb_arg)
	if (events & C.PICOEV_TIMEOUT) != 0 {
		close_conn(loop, fd)
		p.idx[fd] = 0
		return
	}
	else if (events & C.PICOEV_READ) != 0 {
		C.picoev_set_timeout(loop, fd, timeout_secs)
		mut buf := p.buf
		unsafe {
			buf += fd * max_read
		}
		idx := p.idx[fd]
		mut r := myread(fd, buf, max_read, idx)
		if r == 0 {
			close_conn(loop, fd)
			p.idx[fd] = 0
			return
		} else if r == -1 {
			if false { //errno == C.EAGAIN || errno == C.EWOULDBLOCK {
				// TODO
			} else {
				close_conn(loop, fd)
				p.idx[fd] = 0
				return
			}
		} else {
			r += idx
			mut s := tos(buf, r)
			mut out := p.out
			unsafe {
				out += fd * max_write
			}
			mut res := picohttpparser.Response{
				fd: fd
				date: p.date
				buf_start: out
				buf: out
			}
			unsafe {
				res.buf += p.oidx[fd]
			}
			mut req := picohttpparser.Request{}
			for {
				pret := req.parse_request(s, 100)
				if pret <= 0 && s.len > 0 {
					unsafe { C.memmove(buf, s.str, s.len) }
					p.idx[fd] = s.len
					p.oidx[fd] = int(res.buf) - int(res.buf_start)
					break
				}
				c0 := unsafe { req.method.str[0] }
				if c0 ==`p` || c0 == `P` || c0 == `d` || c0 == `D`  {
					mut j := 0
					for {
						if j == req.num_headers {
							break
						}
						if req.headers[j].name_len == 14 && C.strncasecmp(req.headers[j].name, "content-length", 14) == 0 {
							//cont_length := C.atoi(tos(req.headers[j].value, req.headers[j].value_len).str)
							//println('$cont_length')
							//TODO need to maintain state of incomplete request to collect body later
						}
						j = j+1
					}
				}
				p.cb(req, mut &res)
				if pret >= s.len {
					p.idx[fd] = 0
					p.oidx[fd] = 0
					if res.end() < 0 {
						close_conn(loop, fd)
						return
					}
					break
				}
				s = mysubstr(buf, pret, s.len - pret)
			}
		}
	}
}

fn accept_callback(loop &C.picoev_loop, fd, events int, cb_arg voidptr) {
	newfd := C.accept(fd, 0, 0)
	if newfd != -1 {
		setup_sock(newfd)
		C.picoev_add(loop, newfd, C.PICOEV_READ, timeout_secs, rw_callback, cb_arg)
	}
}

pub fn new(port int, cb voidptr) &Picoev {
	fd := C.socket(C.AF_INET, C.SOCK_STREAM, 0)
	assert fd != -1

	flag := 1
	assert C.setsockopt(fd, C.SOL_SOCKET, C.SO_REUSEADDR, &flag, sizeof(int)) == 0
	assert C.setsockopt(fd, C.SOL_SOCKET, C.SO_REUSEPORT, &flag, sizeof(int)) == 0
	$if linux {
		assert C.setsockopt(fd, C.IPPROTO_TCP, C.TCP_QUICKACK, &flag, sizeof(int)) == 0
		timeout := 10
		assert C.setsockopt(fd, C.IPPROTO_TCP, C.TCP_DEFER_ACCEPT, &timeout, sizeof(int)) == 0
		queue_len := 4096
		assert C.setsockopt(fd, C.IPPROTO_TCP, C.TCP_FASTOPEN, &queue_len, sizeof(int)) == 0
	}

	mut addr := C.sockaddr_in{}
	addr.sin_family = C.AF_INET
	addr.sin_port = C.htons(port)
	addr.sin_addr.s_addr = C.htonl(C.INADDR_ANY)
	size := 16 // sizeof(C.sockaddr_in)
	bind_res := C.bind(fd, &addr, size)
	assert bind_res == 0

	listen_res := C.listen(fd, C.SOMAXCONN)
	assert listen_res == 0

	setup_sock(fd)

	C.picoev_init(max_fds)
	loop := C.picoev_create_loop(max_timeout)
	mut pv := &Picoev{
		loop: loop
		cb: cb
		date: C.get_date()
		buf: malloc(max_fds * max_read + 1)
		out: malloc(max_fds * max_write + 1)
	}
	C.picoev_add(loop, fd, C.PICOEV_READ, 0, accept_callback, pv)

	go update_date(mut pv)

	return pv
}

pub fn (p Picoev) serve() {
	for {
		C.picoev_loop_once(p.loop, 1)
	}
}

fn update_date(mut p Picoev) {
	for {
		p.date = C.get_date()
		C.usleep(1000000)
	}
}