Commit 509f1bdd authored by AlexStocks's avatar AlexStocks

write udp pkg asap

parent c3fb7ea5
......@@ -150,6 +150,7 @@ func (this *EchoClient) getClientEchoSession(session getty.Session) (clientEchoS
func (this *EchoClient) heartbeat(session getty.Session) {
var (
err error
pkg EchoPackage
ctx getty.UDPContext
)
......@@ -166,7 +167,7 @@ func (this *EchoClient) heartbeat(session getty.Session) {
ctx.PeerAddr = &(this.serverAddr)
//if err := session.WritePkg(ctx, WritePkgTimeout); err != nil {
if err := session.WritePkg(ctx, WritePkgASAP); err != nil {
if err = session.WritePkg(ctx, WritePkgASAP); err != nil {
log.Warn("session.WritePkg(session{%s}, context{%#v}) = error{%v}", session.Stat(), ctx, err)
session.Close()
......
......@@ -16,6 +16,7 @@ import (
import (
"github.com/AlexStocks/getty"
"github.com/AlexStocks/goext/log"
log "github.com/AlexStocks/log4go"
)
......@@ -69,6 +70,7 @@ func (this *EchoMessageHandler) OnMessage(session getty.Session, udpCtx interfac
}
log.Debug("get echo package{%s}", p)
gxlog.CError("session:%s, get echo package{%s}", session.Stat(), p)
this.client.updateSession(session)
}
......
......@@ -17,7 +17,7 @@ import (
_ "net/http/pprof"
"os"
"os/signal"
// "strings"
"sync/atomic"
"syscall"
"time"
......@@ -81,6 +81,7 @@ func newSession(session getty.Session) error {
udpConn *net.UDPConn
gettyClient getty.Client
client *EchoClient
sessionName string
)
if gettyClient, ok = session.EndPoint().(getty.Client); !ok {
......@@ -90,9 +91,11 @@ func newSession(session getty.Session) error {
switch gettyClient {
case connectedClient.gettyClient:
client = &connectedClient
sessionName = "connected-" + conf.GettySessionParam.SessionName
case unconnectedClient.gettyClient:
client = &unconnectedClient
sessionName = "unconnected-" + conf.GettySessionParam.SessionName
default:
panic(fmt.Sprintf("illegal session{%#v} endpoint", session))
......@@ -109,7 +112,7 @@ func newSession(session getty.Session) error {
udpConn.SetReadBuffer(conf.GettySessionParam.UdpRBufSize)
udpConn.SetWriteBuffer(conf.GettySessionParam.UdpWBufSize)
session.SetName(conf.GettySessionParam.SessionName)
session.SetName(sessionName)
session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
session.SetPkgHandler(NewEchoPackageHandler())
session.SetEventListener(newEchoMessageHandler(client))
......@@ -175,6 +178,7 @@ func initSignal() {
func echo(client *EchoClient) {
var (
err error
pkg EchoPackage
ctx getty.UDPContext
)
......@@ -191,7 +195,8 @@ func echo(client *EchoClient) {
ctx.PeerAddr = &(client.serverAddr)
if session := client.selectSession(); session != nil {
err := session.WritePkg(ctx, WritePkgTimeout)
// err := session.WritePkg(ctx, WritePkgTimeout)
err = session.WritePkg(ctx, WritePkgASAP)
if err != nil {
log.Warn("session.WritePkg(session{%s}, UDPContext{%#v}) = error{%v}", session.Stat(), ctx, err)
session.Close()
......@@ -207,7 +212,7 @@ func testEchoClient(client *EchoClient) {
)
for {
if unconnectedClient.isAvailable() {
if client.isAvailable() {
break
}
time.Sleep(3e9)
......@@ -215,11 +220,10 @@ func testEchoClient(client *EchoClient) {
counter.Start()
for i := 0; i < conf.EchoTimes; i++ {
echo(&unconnectedClient)
echo(client)
}
cost = counter.Count()
log.Info("after loop %d times, client:%s echo cost %d ms",
conf.EchoTimes, client.gettyClient.EndPointType(), cost/1e6)
log.Info("after loop %d times, echo cost %d ms", conf.EchoTimes, cost/1e6)
}
func test() {
......
......@@ -66,7 +66,7 @@ func (this *EchoPackageHandler) Write(ss getty.Session, udpCtx interface{}) erro
startTime = time.Now()
if echoPkg, ok = ctx.Pkg.(*EchoPackage); !ok {
log.Error("illegal pkg:%+v\n", ctx.Pkg)
log.Error("illegal pkg:%+v, its type:%T\n", ctx.Pkg, ctx.Pkg)
return errors.New("invalid echo package!")
}
......@@ -76,8 +76,7 @@ func (this *EchoPackageHandler) Write(ss getty.Session, udpCtx interface{}) erro
return err
}
// _, err = ss.Write(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr})
err = ss.WritePkg(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr}, WritePkgASAP)
_, err = ss.Write(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr})
log.Info("WriteEchoPkgTimeMs = %s", time.Since(startTime).String())
return err
......
......@@ -26,7 +26,7 @@ SessionTimeout = "20s"
# client echo request string
EchoString = "Hello, getty!"
# 发送echo请求次数
EchoTimes = 10000
EchoTimes = 100000
# app fail fast
FailFastTimeout = "3s"
......
......@@ -11,12 +11,12 @@ package main
import (
"errors"
"fmt"
"sync"
"time"
)
import (
"fmt"
"github.com/AlexStocks/getty"
log "github.com/AlexStocks/log4go"
)
......@@ -42,8 +42,9 @@ type HeartbeatHandler struct{}
func (this *HeartbeatHandler) Handle(session getty.Session, ctx getty.UDPContext) error {
var (
pkg *EchoPackage
ok bool
pkg *EchoPackage
rspPkg EchoPackage
)
log.Debug("get echo heartbeat udp context{%#v}", ctx)
......@@ -51,7 +52,6 @@ func (this *HeartbeatHandler) Handle(session getty.Session, ctx getty.UDPContext
return fmt.Errorf("illegal @ctx.Pkg:%#v", ctx.Pkg)
}
var rspPkg EchoPackage
rspPkg.H = pkg.H
rspPkg.B = echoHeartbeatResponseString
rspPkg.H.Len = uint16(len(rspPkg.B) + 1)
......
......@@ -67,7 +67,7 @@ func (this *EchoPackageHandler) Write(ss getty.Session, udpCtx interface{}) erro
startTime = time.Now()
if echoPkg, ok = ctx.Pkg.(*EchoPackage); !ok {
log.Error("illegal pkg:%+v\n", ctx.Pkg)
log.Error("illegal pkg:%+v, addr:%s\n", ctx.Pkg, ctx.PeerAddr)
return errors.New("invalid echo package!")
}
......@@ -77,8 +77,7 @@ func (this *EchoPackageHandler) Write(ss getty.Session, udpCtx interface{}) erro
return err
}
// _, err = ss.Write(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr})
err = ss.WritePkg(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr}, WritePkgASAP)
_, err = ss.Write(getty.UDPContext{Pkg: buf.Bytes(), PeerAddr: ctx.PeerAddr})
log.Info("WriteEchoPkgTimeMs = %s", time.Since(startTime).String())
return err
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment