feat: benckmark

benckmark/server/main.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"encoding/binary"
+	"errors"
+	"flag"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+import (
+	"github.com/apache/dubbo-getty"
+	gxsync "github.com/dubbogo/gost/sync"
+)
+
+var (
+	taskPoolMode = flag.Bool("taskPool", false, "task pool mode")
+	taskPoolSize = flag.Int("task_pool_size", 2000, "task poll size")
+	pprofPort    = flag.Int("pprof_port", 65432, "pprof http port")
+)
+
+var (
+	taskPool gxsync.GenericTaskPool
+)
+
+const CronPeriod = time.Second
+
+func main() {
+	flag.Parse()
+
+	go func() {
+		http.ListenAndServe(fmt.Sprintf(":%d", *pprofPort), nil)
+	}()
+
+	options := []getty.ServerOption{getty.WithLocalAddress(":8090")}
+
+	if *taskPoolMode {
+		taskPool = gxsync.NewTaskPoolSimple(*taskPoolSize)
+		options = append(options, getty.WithServerTaskPool(taskPool))
+	}
+
+	server := getty.NewTCPServer(options...)
+
+	go server.RunEventLoop(NewHelloServerSession)
+	log.Printf("getty server start, listening at: 8090")
+
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, os.Interrupt, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
+	<-signals
+	server.Close()
+}
+
+func NewHelloServerSession(session getty.Session) (err error) {
+	var pkgHandler = &PackageHandler{}
+	var EventListener = &MessageHandler{}
+
+	tcpConn, ok := session.Conn().(*net.TCPConn)
+	if !ok {
+		panic(fmt.Sprintf("newSession: %s, session.conn{%#v} is not tcp connection", session.Stat(), session.Conn()))
+	}
+
+	if err = tcpConn.SetNoDelay(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlive(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlivePeriod(10 * time.Second); err != nil {
+		return err
+	}
+	if err = tcpConn.SetReadBuffer(262144); err != nil {
+		return err
+	}
+	if err = tcpConn.SetWriteBuffer(524288); err != nil {
+		return err
+	}
+
+	session.SetName("hello")
+	session.SetMaxMsgLen(128 * 1024) // max message package length is 128k
+	session.SetReadTimeout(time.Second)
+	session.SetWriteTimeout(5 * time.Second)
+	session.SetCronPeriod(int(CronPeriod / 1e6))
+	session.SetWaitTime(time.Second)
+
+	session.SetPkgHandler(pkgHandler)
+	session.SetEventListener(EventListener)
+	return nil
+}
+
+type MessageHandler struct {
+	SessionOnOpen func(session getty.Session)
+}
+
+func (h *MessageHandler) OnOpen(session getty.Session) error {
+	log.Printf("OnOpen session{%s} open", session.Stat())
+	if h.SessionOnOpen != nil {
+		h.SessionOnOpen(session)
+	}
+	return nil
+}
+
+func (h *MessageHandler) OnError(session getty.Session, err error) {
+	log.Printf("OnError session{%s} got error{%v}, will be closed.", session.Stat(), err)
+}
+
+func (h *MessageHandler) OnClose(session getty.Session) {
+	log.Printf("OnClose session{%s} is closing......", session.Stat())
+}
+
+func (h *MessageHandler) OnMessage(session getty.Session, pkg interface{}) {
+	log.Printf("OnMessage....")
+	s, ok := pkg.(string)
+	if !ok {
+		log.Printf("illegal package{%#v}", pkg)
+		return
+	}
+	log.Printf("OnMessage: %s", s)
+}
+
+func (h *MessageHandler) OnCron(session getty.Session) {
+	log.Printf("OnCron....")
+	active := session.GetActive()
+	if CronPeriod < time.Since(active) {
+		log.Printf("OnCorn session{%s} timeout{%s}", session.Stat(), time.Since(active).String())
+		session.Close()
+	}
+}
+
+type PackageHandler struct{}
+
+func (h *PackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
+	dataLen := len(data)
+	if dataLen < 4 {
+		return nil, 0, nil
+	}
+
+	start := 0
+	pos := start + 4
+	pkgLen := int(binary.LittleEndian.Uint32(data[start:pos]))
+	if dataLen < pos+pkgLen {
+		return nil, pos + pkgLen, nil
+	}
+	start = pos
+
+	pos = start + pkgLen
+	s := string(data[start:pos])
+
+	return s, pos, nil
+}
+
+func (h *PackageHandler) Write(ss getty.Session, p interface{}) ([]byte, error) {
+	pkg, ok := p.(string)
+	if !ok {
+		log.Printf("illegal pkg:%+v", p)
+		return nil, errors.New("invalid package")
+	}
+
+	pkgLen := int32(len(pkg))
+	pkgStreams := make([]byte, 0, 4+len(pkg))
+
+	// pkg len
+	start := 0
+	pos := start + 4
+	binary.LittleEndian.PutUint32(pkgStreams[start:pos], uint32(pkgLen))
+	start = pos
+
+	// pkg
+	pos = start + int(pkgLen)
+	copy(pkgStreams[start:pos], pkg[:])
+
+	return pkgStreams[:pos], nil
+}

benckmark/single_client/main.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"encoding/binary"
+	"errors"
+	"flag"
+	"fmt"
+	"log"
+	"net"
+	"time"
+)
+
+import (
+	"github.com/apache/dubbo-getty"
+	"github.com/dubbogo/gost/sync"
+)
+
+var (
+	ip          = flag.String("ip", "127.0.0.1", "server IP")
+	connections = flag.Int("conn", 1, "number of tcp connections")
+
+	taskPoolMode = flag.Bool("taskPool", false, "task pool mode")
+	taskPoolSize = flag.Int("task_pool_size", 2000, "task poll size")
+	pprofPort    = flag.Int("pprof_port", 65431, "pprof http port")
+)
+
+var (
+	taskPool gxsync.GenericTaskPool
+)
+
+const CronPeriod = time.Second
+const WritePkgTimeout = 1e8
+
+func main() {
+	flag.Parse()
+
+	client := getty.NewTCPClient(
+		getty.WithServerAddress(*ip+":8090"),
+		getty.WithConnectionNumber(*connections),
+		getty.WithClientTaskPool(taskPool),
+	)
+
+	client.RunEventLoop(NewHelloClientSession)
+
+	ClientRequest()
+	client.Close()
+}
+
+// NewHelloClientSession use for init client session
+func NewHelloClientSession(session getty.Session) (err error) {
+	var pkgHandler = &PackageHandler{}
+	var EventListener = &MessageHandler{}
+
+	EventListener.SessionOnOpen = func(session getty.Session) {
+		Sessions = append(Sessions, session)
+	}
+
+	tcpConn, ok := session.Conn().(*net.TCPConn)
+	if !ok {
+		panic(fmt.Sprintf("newSession: %s, session.conn{%#v} is not tcp connection", session.Stat(), session.Conn()))
+	}
+
+	if err = tcpConn.SetNoDelay(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlive(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlivePeriod(10 * time.Second); err != nil {
+		return err
+	}
+	if err = tcpConn.SetReadBuffer(262144); err != nil {
+		return err
+	}
+	if err = tcpConn.SetWriteBuffer(524288); err != nil {
+		return err
+	}
+
+	session.SetName("hello")
+	session.SetMaxMsgLen(128 * 1024) // max message package length is 128k
+	session.SetReadTimeout(time.Second)
+	session.SetWriteTimeout(5 * time.Second)
+	session.SetCronPeriod(int(CronPeriod / 1e6))
+	session.SetWaitTime(time.Second)
+
+	session.SetPkgHandler(pkgHandler)
+	session.SetEventListener(EventListener)
+	return nil
+}
+
+type MessageHandler struct {
+	SessionOnOpen func(session getty.Session)
+}
+
+func (h *MessageHandler) OnOpen(session getty.Session) error {
+	log.Printf("OnOpen session{%s} open", session.Stat())
+	if h.SessionOnOpen != nil {
+		h.SessionOnOpen(session)
+	}
+	return nil
+}
+
+func (h *MessageHandler) OnError(session getty.Session, err error) {
+	log.Printf("OnError session{%s} got error{%v}, will be closed.", session.Stat(), err)
+}
+
+func (h *MessageHandler) OnClose(session getty.Session) {
+	log.Printf("OnClose session{%s} is closing......", session.Stat())
+}
+
+func (h *MessageHandler) OnMessage(session getty.Session, pkg interface{}) {
+	log.Printf("OnMessage....")
+	s, ok := pkg.(string)
+	if !ok {
+		log.Printf("illegal package{%#v}", pkg)
+		return
+	}
+	log.Printf("OnMessage: %s", s)
+}
+
+func (h *MessageHandler) OnCron(session getty.Session) {
+	log.Printf("OnCron....")
+	active := session.GetActive()
+	if CronPeriod < time.Since(active) {
+		log.Printf("OnCorn session{%s} timeout{%s}", session.Stat(), time.Since(active).String())
+		session.Close()
+	}
+}
+
+type PackageHandler struct{}
+
+func (h *PackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
+	dataLen := len(data)
+	if dataLen < 4 {
+		return nil, 0, nil
+	}
+
+	start := 0
+	pos := start + 4
+	pkgLen := int(binary.LittleEndian.Uint32(data[start:pos]))
+	if dataLen < pos+pkgLen {
+		return nil, pos + pkgLen, nil
+	}
+	start = pos
+
+	pos = start + pkgLen
+	s := string(data[start:pos])
+
+	return s, pos, nil
+}
+
+func (h *PackageHandler) Write(ss getty.Session, p interface{}) ([]byte, error) {
+	pkg, ok := p.(string)
+	if !ok {
+		log.Printf("illegal pkg:%+v", p)
+		return nil, errors.New("invalid package")
+	}
+
+	pkgLen := int32(len(pkg))
+	pkgStreams := make([]byte, 0, 4+len(pkg))
+
+	// pkg len
+	start := 0
+	pos := start + 4
+	binary.LittleEndian.PutUint32(pkgStreams[start:pos], uint32(pkgLen))
+	start = pos
+
+	// pkg
+	pos = start + int(pkgLen)
+	copy(pkgStreams[start:pos], pkg[:])
+
+	return pkgStreams[:pos], nil
+}
+
+var (
+	Sessions []getty.Session
+)
+
+func ClientRequest() {
+	for _, session := range Sessions {
+		ss := session
+		_, _, err := ss.WritePkg("hello", WritePkgTimeout)
+		if err != nil {
+			log.Printf("session.WritePkg(session{%s}, error{%v}", ss.Stat(), err)
+			ss.Close()
+		}
+	}
+
+}

go.mod

@@ -2,12 +2,14 @@ module github.com/apache/dubbo-getty
 
 go 1.14
 
+replace github.com/dubbogo/gost v1.11.0 => /Users/haohongfan/goproject/dubbo/gost
+
 require (
 	github.com/dubbogo/gost v1.11.0
 	github.com/golang/snappy v0.0.1
 	github.com/gorilla/websocket v1.4.2
 	github.com/pkg/errors v0.9.1
-	github.com/stretchr/testify v1.6.1
+	github.com/stretchr/testify v1.7.0
 	go.uber.org/atomic v1.7.0
-	go.uber.org/zap v1.15.0
+	go.uber.org/zap v1.16.0
 )

session.go

@@ -499,6 +499,7 @@ func (s *session) run() {
 	}
 
 	s.grNum.Add(1)
+	println(1111)
 	if _, err := defaultTimerWheel.AddTimer(sessionTimerLoop, gxtime.TimerLoop, s.period, s); err != nil {
 		panic(fmt.Sprintf("failed to add session %s to defaultTimerWheel", s.Stat()))
 	}