golang-实现自己的事件驱动

golang实现自己的事件驱动

众所周知，go中的异步操作都已经封装在了运行时的过程中，有关socket的网络的异步操作都封装到了go的netpoll中，从而简化了编程形式。本文也就根据evio库总结而来。

golang跨平台库

如何编写golang的跨平台库，现在主流的方式如下：

evserver
│   go.mod
│   main.go    
│
└───show
│   │   show.go
│   │   show_darwin.go
│   │   show_linux.go

通过创建一个package，然后通过命名文件后缀为_{平台后缀}.go的形式，在不同平台运行的时候会自动编译对应的代码。

main.go代码

package main

import (
	"evserver/show"
)


func main() {
	show := show.GetDefault()
	show.ShowHello()
}

show.go

package show


type Show interface {
	ShowHello()
}

show_linux.go

package show


import "fmt"

type osDefaulter struct {
}

func GetDefault() osDefaulter{
	return osDefaulter{}
}


func(s osDefaulter) ShowHello(){
	fmt.Println("linux show")
}

show_darwin.go

package show


import "fmt"

type osDefaulter struct {
}

func GetDefault() osDefaulter{
	return osDefaulter{}
}


func(s osDefaulter) ShowHello(){
	fmt.Println("darwin show")
}

常见的设计的方式就是通过在package中定义一个接口，然后通过不同平台的后缀文件去调用不同的方式去实现，从而完成package统一的对外提供服务的方式，当然跨平台的库也可以有另外一种方式，即如下：

func ShowHello() string {
    if runtime.GOOS == "windows" {
        return "windows hello"
    } else {
        return "other paltform hello"
    }
}

但是这种方式针对简单的跨平台性能还可以，针对复杂的跨平台的功能就对代码侵入比较严重。

https://techblog.steelseries.com/2014/04/08/multi-platform-development-go.html

https://blog.gopheracademy.com/advent-2013/day-23-multi-platform-applications/

golang事件驱动-Kqueue

总体的代码目录如下：

evserver
│   go.mod
│   main.go    
│
└───poll
│   │   poller.go
│   │   poller_darwin.go
│   │   poller_linux.go

main.go

package main

import (
	"evserver/poll"
	"fmt"
)

func main() {
	IP := "127.0.0.1"
	Port := 6667
	s := &poll.Server{
		Ip:IP,
		Port: Port,
	}
	s.Init()
	s.Data = func(c *poll.Conn, in []byte) (out []byte) {
		out = in
		out = append(out, []byte("back")...)
		return
	}
	fmt.Printf(" running in %s:%d\n", IP, Port)
	poll.LoopRun(s)
}

poller.go

package poll

import (
	"log"
	"net"
	"syscall"
)

const READ_FLAG  = 1
const WRITE_FLAG  = 2


type Conn struct {
	fd         int              // file descriptor
	lnidx      int              // listener index in the server lns list
	out        []byte           // write buffer
	sa         syscall.Sockaddr // remote socket address
	reuse      bool             // should reuse input buffer
	opened     bool             // Connection opened event fired
	ctx        interface{}      // user-defined context
	loop       *loop            // Connected loop
}

func (c *Conn) Context() interface{}       { return c.ctx }
func (c *Conn) SetContext(ctx interface{}) { c.ctx = ctx }
func (c *Conn) Wake() {
	//if c.loop != nil {
	//	c.loop.poll.Trigger(c)
	//}
}


type loop struct {
	idx     int            // loop index in the server loops list
	poll    *Poll // epoll or kqueue
	packet  []byte         // read packet buffer
	fdConns map[int]*Conn  // loop Connections fd -> Conn
}


type Server struct {
	Ip string
	Port int
	fd int
	Data func(c *Conn, in []byte)(out []byte)
}

func (s *Server)Init()  {
	fd ,err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, syscall.IPPROTO_TCP)
	if err != nil {
		panic(err)
	}

	var serverAddr [4]byte

	ip := s.Ip
	IP := net.ParseIP(ip)
	if IP == nil {
		log.Fatal("Unable to process IP: ", ip)
	}

	copy(serverAddr[:], IP[12:16])

	if err = syscall.SetsockoptInt(fd, syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1); err != nil {
		syscall.Close(fd)
		panic(err)
	}

	err = syscall.Bind(fd, &syscall.SockaddrInet4{Port: s.Port, Addr:serverAddr})
	if err != nil {
		panic(err)
	}

	err = syscall.Listen(fd, 1024)
	if err != nil {
		panic(err)
	}
	s.fd = fd
}

func LoopRun(s *Server){
	l := &loop{
		poll:OpenPoll(),
		packet:  make([]byte, 0xFFFF),
		fdConns: make(map[int]*Conn),
	}

	l.poll.AddRead(s.fd)

	err := l.poll.Wait(func(fd int, note interface{}) error {
		if fd == s.fd {
			err := LoopAccept(s, l, fd)
			if err != nil {
				panic(err)
			}
		} else {
			c := l.fdConns[fd]
			flag := note.(int)
			if flag == READ_FLAG {
				LoopRead(s, l, c)
			} else {
				LoopWrite(s, l, c)
			}
		}

		return nil
	})
	if err != nil {
		panic(err)
	}
}


func LoopAccept(s *Server, l *loop, fd int) error{
	nfd, sa, err := syscall.Accept(fd)
	if err != nil {
		if err != nil {
			if err == syscall.EAGAIN {
				return nil
			}
			return err
		}
	}
	if err := syscall.SetNonblock(nfd, true); err != nil {
		return err
	}

	c := &Conn{fd: nfd, sa: sa, loop: l}
	c.out = []byte{}
	l.fdConns[c.fd] = c
	l.poll.AddReadWrite(c.fd)
	return nil
}


func LoopRead(s *Server, l *loop, c *Conn)error {
	var in []byte
	n, err := syscall.Read(c.fd, l.packet)
	if n == 0 || err != nil {
		if err == syscall.EAGAIN {
			return nil
		}
		return LoopClose(s, l, c)
	}
	in = l.packet[:n]
	if !c.reuse {
		in = append([]byte{}, in...)
	}
	if s.Data != nil {
		out := s.Data(c, in)
		if len(out) > 0 {
			c.out = append(c.out[:0], out...)
		}
	}
	if len(c.out) != 0 {
		l.poll.ModReadWrite(c.fd)
	}
	return nil
}


func LoopWrite(s *Server, l *loop, c *Conn)error {
	if c == nil {
		return nil
	}
	if c.out == nil || len(c.out) == 0 {
		return nil
	}
	n, err := syscall.Write(c.fd, c.out)
	if err != nil {
		if err == syscall.EAGAIN {
			return nil
		}
		return LoopClose(s, l, c)
	}
	if n == len(c.out) {
		// release the connection output page if it goes over page size,
		// otherwise keep reusing existing page.
		if cap(c.out) > 4096 {
			c.out = nil
		} else {
			c.out = c.out[:0]
		}
	} else {
		c.out = c.out[n:]
	}
	if len(c.out) == 0{
		l.poll.ModRead(c.fd)
	}
	return nil
}


func LoopClose(s *Server, l *loop, c *Conn)error{
	delete(l.fdConns, c.fd)
	err := syscall.Close(c.fd)
	return err
}

poller_darwin.go

package poll

import (
	"syscall"
)

// Poll ...
type Poll struct {
	fd      int
	changes []syscall.Kevent_t
}

// OpenPoll ...
func OpenPoll() *Poll {
	l := new(Poll)
	p, err := syscall.Kqueue()
	if err != nil {
		panic(err)
	}
	l.fd = p
	_, err = syscall.Kevent(l.fd, []syscall.Kevent_t{{
		Ident:  0,
		Filter: syscall.EVFILT_USER,
		Flags:  syscall.EV_ADD | syscall.EV_CLEAR,
	}}, nil, nil)
	if err != nil {
		panic(err)
	}

	return l
}

// Close ...
func (p *Poll) Close() error {
	return syscall.Close(p.fd)
}

// Wait ...
func (p *Poll) Wait(iter func(fd int, note interface{}) error) error {
	events := make([]syscall.Kevent_t, 128)
	for {
		n, err := syscall.Kevent(p.fd, p.changes, events, nil)
		if err != nil && err != syscall.EINTR {
			return err
		}
		p.changes = p.changes[:0]
		for i := 0; i < n; i++ {
			if fd := int(events[i].Ident); fd != 0 {
				var flag int
				if events[i].Filter == syscall.EVFILT_READ {
					flag = READ_FLAG
				} else {
					flag = WRITE_FLAG
				}
				if err := iter(fd, flag); err != nil {
					return err
				}
			}
		}
	}
}

// AddRead ...
func (p *Poll) AddRead(fd int) {
	p.changes = append(p.changes,
		syscall.Kevent_t{
			Ident: uint64(fd), Flags: syscall.EV_ADD, Filter: syscall.EVFILT_READ,
		},
	)
}

// AddReadWrite ...
func (p *Poll) AddReadWrite(fd int) {
	p.changes = append(p.changes,
		syscall.Kevent_t{
			Ident: uint64(fd), Flags: syscall.EV_ADD, Filter: syscall.EVFILT_READ,
		},
		syscall.Kevent_t{
			Ident: uint64(fd), Flags: syscall.EV_ADD, Filter: syscall.EVFILT_WRITE,
		},
	)
}

// ModRead ...
func (p *Poll) ModRead(fd int) {
	p.changes = append(p.changes, syscall.Kevent_t{
		Ident: uint64(fd), Flags: syscall.EV_DELETE, Filter: syscall.EVFILT_WRITE,
	})
}

// ModReadWrite ...
func (p *Poll) ModReadWrite(fd int) {
	p.changes = append(p.changes, syscall.Kevent_t{
		Ident: uint64(fd), Flags: syscall.EV_ADD, Filter: syscall.EVFILT_WRITE,
	})
}

// ModDetach ...
func (p *Poll) ModDetach(fd int) {
	p.changes = append(p.changes,
		syscall.Kevent_t{
			Ident: uint64(fd), Flags: syscall.EV_DELETE, Filter: syscall.EVFILT_READ,
		},
		syscall.Kevent_t{
			Ident: uint64(fd), Flags: syscall.EV_DELETE, Filter: syscall.EVFILT_WRITE,
		},
	)
}

运行测试

import socket


sock = socket.socket()
sock.connect(("127.0.0.1", 6667))

print(sock.send(b"data123121231232"))
print(sock.recv(4096))

此时运行程序go run main.go和测试客户端程序。

此时就可以看出简单的事件驱动的过程。

压测

将main.go进行修改，修改为http的处理方式。

package main

import (
	"evserver/poll"
	"fmt"
	"strconv"
	"strings"
	"time"
)

func main() {
	IP := "127.0.0.1"
	Port := 6667
	s := &poll.Server{
		Ip:IP,
		Port: Port,
	}
	s.Init()
	s.Data = func(c *poll.Conn, in []byte) (out []byte) {
		if in == nil {
			return
		}
		data := in
		//if noparse && bytes.Contains(data, []byte("\r\n\r\n")) {
		//	// for testing minimal single packet request -> response.
		//	out = appendresp(nil, "200 OK", "", res)
		//	return
		//}
		// process the pipeline
		var req request
		for {
			leftover, err := parsereq(data, &req)
			if err != nil {
				// bad thing happened
				out = appendresp(out, "500 Error", "", err.Error()+"\n")
				break
			} else if len(leftover) == len(data) {
				// request not ready, yet
				break
			}
			// handle the request
			out = appendhandle(out, &req)
			data = leftover
		}
		return
	}

	res = "Hello World!\r\n"

	fmt.Printf(" running in %s:%d\n", IP, Port)
	poll.LoopRun(s)
}


var res string

type request struct {
	proto, method string
	path, query   string
	head, body    string
	remoteAddr    string
}


// appendhandle handles the incoming request and appends the response to
// the provided bytes, which is then returned to the caller.
func appendhandle(b []byte, req *request) []byte {
	return appendresp(b, "200 OK", "", res)
}

// appendresp will append a valid http response to the provide bytes.
// The status param should be the code plus text such as "200 OK".
// The head parameter should be a series of lines ending with "\r\n" or empty.
func appendresp(b []byte, status, head, body string) []byte {
	b = append(b, "HTTP/1.1"...)
	b = append(b, ' ')
	b = append(b, status...)
	b = append(b, '\r', '\n')
	b = append(b, "Server: error\r\n"...)
	b = append(b, "Date: "...)
	b = time.Now().AppendFormat(b, "Mon, 02 Jan 2006 15:04:05 GMT")
	b = append(b, '\r', '\n')
	if len(body) > 0 {
		b = append(b, "Content-Length: "...)
		b = strconv.AppendInt(b, int64(len(body)), 10)
		b = append(b, '\r', '\n')
	}
	b = append(b, head...)
	b = append(b, '\r', '\n')
	if len(body) > 0 {
		b = append(b, body...)
	}
	return b
}

// parsereq is a very simple http request parser. This operation
// waits for the entire payload to be buffered before returning a
// valid request.
func parsereq(data []byte, req *request) (leftover []byte, err error) {
	sdata := string(data)
	var i, s int
	var top string
	var clen int
	var q = -1
	// method, path, proto line
	for ; i < len(sdata); i++ {
		if sdata[i] == ' ' {
			req.method = sdata[s:i]
			for i, s = i+1, i+1; i < len(sdata); i++ {
				if sdata[i] == '?' && q == -1 {
					q = i - s
				} else if sdata[i] == ' ' {
					if q != -1 {
						req.path = sdata[s:q]
						req.query = req.path[q+1 : i]
					} else {
						req.path = sdata[s:i]
					}
					for i, s = i+1, i+1; i < len(sdata); i++ {
						if sdata[i] == '\n' && sdata[i-1] == '\r' {
							req.proto = sdata[s:i]
							i, s = i+1, i+1
							break
						}
					}
					break
				}
			}
			break
		}
	}
	if req.proto == "" {
		return data, fmt.Errorf("malformed request")
	}
	top = sdata[:s]
	for ; i < len(sdata); i++ {
		if i > 1 && sdata[i] == '\n' && sdata[i-1] == '\r' {
			line := sdata[s : i-1]
			s = i + 1
			if line == "" {
				req.head = sdata[len(top)+2 : i+1]
				i++
				if clen > 0 {
					if len(sdata[i:]) < clen {
						break
					}
					req.body = sdata[i : i+clen]
					i += clen
				}
				return data[i:], nil
			}
			if strings.HasPrefix(line, "Content-Length:") {
				n, err := strconv.ParseInt(strings.TrimSpace(line[len("Content-Length:"):]), 10, 64)
				if err == nil {
					clen = int(n)
				}
			}
		}
	}
	// not enough data
	return data, nil
}

此时运行go run main.go，并将前文编写的原生的http和evio的http进行对比。同样进行wrk进行测试。

原生http

 wrk -t8 -c200 -d60s --latency  http://127.0.0.1:8000
Running 1m test @ http://127.0.0.1:8000
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     4.60ms    6.67ms 229.63ms   93.36%
    Req/Sec     7.41k     1.73k   14.17k    72.51%
  Latency Distribution
     50%    2.86ms
     75%    4.06ms
     90%    7.82ms
     99%   35.12ms
  3537128 requests in 1.00m, 431.78MB read
  Socket errors: connect 0, read 27, write 0, timeout 0
Requests/sec:  58849.98
Transfer/sec:      7.18MB

evio的http(开启3个loop)

wrk -t8 -c200 -d60s --latency  http://127.0.0.1:7979
Running 1m test @ http://127.0.0.1:7979
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     2.19ms  730.09us  36.98ms   93.56%
    Req/Sec    11.39k     1.37k   17.97k    68.33%
  Latency Distribution
     50%    2.02ms
     75%    2.37ms
     90%    2.68ms
     99%    4.24ms
  5444990 requests in 1.00m, 540.05MB read
Requests/sec:  90669.19
Transfer/sec:      8.99MB

自己实现的http

wrk -t8 -c200 -d60s --latency  http://127.0.0.1:6667
Running 1m test @ http://127.0.0.1:6667
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     2.32ms    1.38ms  47.39ms   95.27%
    Req/Sec    11.06k     1.98k   31.58k    74.97%
  Latency Distribution
     50%    2.04ms
     75%    2.41ms
     90%    2.90ms
     99%    6.85ms
  5286393 requests in 1.00m, 529.36MB read
Requests/sec:  87953.74
Transfer/sec:      8.81MB

	原生http	evio处理http	手工实现http
Qps	58849.98	90669.19	87953.74

总结

本文主要就是梳理了一下go的跨平台的主流方式，并简单实现了有关mac的kqueue的过程，后续大家有兴趣可自行开发事件驱动的框架，当前比较火热的是evio，gev和gnet。由于本人才疏学浅，如有错误请批评指正。