gen-iot/liblpc

Golang high performance, event-driven, asynchronous I/O

liblpc

High performance async network io library

First

if you have any good feature suggestions or bug fixed , any Pull Request or Issues are welcome!

Usage

# install liblpc v2
go get -u github.com/gen-iot/liblpc/v2@latest

// import 
import "github.com/gen-iot/liblpc/v2"

Overview

• World Based On Event Loop 😎
• UnBuffered/Buffered Stream 🚀
• Timers: IO Timeout, Delay... ⏰
• DNS Resolve 🌐
• Lots Of Unix Socket API Toolbox 🔧
• Thread(Goroutine) Safe! 👍

Backend

Platform	Backend	Support
Linux family	Epoll	Done 🎉
mac OS (BSD family)	Kqueue	Done 🎉
Windows	IOCP	Maybe Never... 😢
POSIX Like	Poll	Maybe Never... 🥺 we already have epoll
POSIX Like	Select	Coming Soon 🤡

liblpc using interface Poller and Watcher as abstraction for any backend.

Getting Started

Create Pure EventLoop:

loop,err := liblpc.NewEventLoop()
std.AssertError(err, "new pure event loop")

Create IO EventLoop:

loop, err := liblpc.NewIOEvtLoop(1024 * 4)
std.AssertError(err, "new io event loop")

Loop Lifecycle:

exit a loop

// just call loop.Break in anywhere
loop.Break()

📌Loop.'Close' can't stop a loop but Loop.'Break' can.

📌Loop.'Close' use to cleanup a loop

Cleanup a loop

loop ,err := liblpc.NewEventLoop()
std.AssertError(err, "new event loop")
defer loop.Close()

Run loop synchronously

// block until break loop called
loop.Run()

Run loop asynchronously😂😂😂

go func(){ loop.Run() }()

Create Listener:

// create listen fd first!
listenerFd, err := liblpc.NewListenerFd(
  "127.0.0.1:12345", // serve at
  1024,              // backlog
  true,              // enable reuse addr
  true,              // enable reuse port
)
std.AssertError(err, "new listener fd")
// new listener
listener := liblpc.NewListener(loop, int(listenerFd), onAccept)
listener.Start()

Accept New Conn Stream:

// 📌Note: in accept callback
stream := liblpc.NewConnStream(
  ln.Loop().(*liblpc.IOEvtLoop), // cast Loop to IOEventLoop 
  newFd,                         // incoming fd
  onStreamRead,                  // read callback
  )
stream.SetOnClose(onStreamClose) // register close callback
stream.Start()

Create Client Stream:

cliFd, err := liblpc.NewConnFd(addr)
std.AssertError(err, "new client fd failed")
stream := liblpc.NewConnStream(loop, int(cliFd), nil)
stream.SetOnConnect(func(sw liblpc.StreamWriter, err error) {
  sw.Write([]byte("hello world!"), true)
})
stream.SetOnClose(func(sw liblpc.StreamWriter, err error) {
  log.Println("client close :", err)
  // break loop...
  loop.Break()
})
stream.Start()

📌Stream.'Close' is safe to invoke multi times

📌Anytime you can't find out whether if Stream is 'Closing' or really been 'Closed',Just invoke Stream.'Close'

Example: Simple Read/Write/Close

package main

import (
  "github.com/gen-iot/liblpc"
  "github.com/gen-iot/std"
  "log"
)

func onStreamRead(sw liblpc.StreamWriter, data []byte, len int) {
  // print client data in string format
  log.Println("on read:", string(data[:len]))
  _ = sw.Close()
}

func onStreamClose(sw liblpc.StreamWriter, err error) {
  log.Println("conn closed,err:", err)
  _ = sw.Close() // close remote client
}

func onAccept(ln *liblpc.Listener, newFd int, err error) {
  if err != nil {
    log.Printf("listener got error:%v\n", err)
    return
  }
  stream := liblpc.NewConnStream(
    ln.Loop().(*liblpc.IOEvtLoop), // cast Loop to   IOEventLoop 
    newFd,                         // incoming fd
    onStreamRead,                  // read callback
    )
  stream.SetOnClose(onStreamClose) // register close   callback
  stream.Start()
}

func simpleClient(loop *liblpc.IOEvtLoop, addr string) {
  cliFd, err := liblpc.NewConnFd(addr)
  std.AssertError(err, "new client fd failed")
  stream := liblpc.NewConnStream(loop, int(cliFd), nil)
  stream.SetOnConnect(func(sw liblpc.StreamWriter, err   error) {
    sw.Write([]byte("hello world!"), true)
  })
  stream.SetOnClose(func(sw liblpc.StreamWriter, err error)   {
    log.Println("client close :", err)
    // close itself
    _ = sw.Close()
    // break loop...
    loop.Break()
  })
  stream.Start()
}

func main() {
  loop, err := liblpc.NewIOEvtLoop(1024 * 4)
  std.AssertError(err, "new event loop")
  defer std.CloseIgnoreErr(loop)
  // create listen fd first!
  listenerFd, err := liblpc.NewListenerFd(
    "127.0.0.1:12345", // serve at
    1024,              // backlog
    true,              // enable reuse addr
    true,              // enable reuse port
  )
  std.AssertError(err, "new listener fd")
  // new listener
  listener := liblpc.NewListener(loop, int(listenerFd),   onAccept)
  defer std.CloseIgnoreErr(listener)
  listener.Start()
  // start simple client
  simpleClient(loop, "127.0.0.1:12345")
  //
  loop.Run()
}

License

Released under the MIT License

liblpc's Issues

create eventloop with specified poller

need api

func NewEventLoop2(poller Poller) (EventLoop, error)

support conn timeout

需要超时,在没有收到客户端FIN包,连接不能正常释放.

支持加密传输

至少支持openssl, libressl其中一种backend

EventLoop Not Exit After Ctx.Done

in EventLoop.Run, ctx.Done() detection is after poller.Poll

if Poller currently no event, Run will not quit immediately

Suggest:

func (this *EventLoop) Run(ctx context.Context) {
        // do like cmd.Run :P
	go func() {
		select {
		case <-ctx.Done():
			this.Break()
		}
	}()
	for {
		// poller stuff
	}
}

NewNotifyWatcher eventfd flags typo

previous:

eventFd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_CLOEXEC)

fix:

eventFd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)

WatcherCtl should handle error

如果出现错误,应该把watcher从loop中移除,而不是panic

infinity WRITEABLE on darwin

Caused by: kqueue always report socket always writeable
Fix: add EV_ONESHOT on write flag

loop.Run support context

change:

loop.Run()

to:

// signature
Run(ctx context.Context)

// call
loop.Run(context.Background())

Stream Close有可能导致重复close

fix:

func (this *Stream) Close() error {
	this.Loop().RunInLoop(func() {
		if this.isClose {
			return
		}
		this.isClose = true
		this.onRead(nil, 0, io.EOF)
		if this.DisableRW() {
			this.Update(true)
		}
		_ = this.FdWatcher.Close()
	})
	return nil
}

NewNotifyWatcher handle error

if err !=0 , doesn't need to close fd

osx NewListenerFd ReusePort support

SO_REUSEPORT has different value in osx and linux