Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Send router packets asynchronously #223

Merged
merged 3 commits into from
Mar 12, 2020
Merged

Send router packets asynchronously #223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
05bcab9
Send router packets asynchronously
nkryuchkov Mar 10, 2020
c681996
Make changes suggested by @evanlinjin
nkryuchkov Mar 11, 2020
b975415
Fix possible double logging of stopping
nkryuchkov Mar 12, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ require (
github.com/go-chi/chi v4.0.2+incompatible
github.com/google/uuid v1.1.1
github.com/gorilla/securecookie v1.1.1
github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b // indirect
github.com/mholt/archiver/v3 v3.3.0
github.com/pkg/profile v1.3.0
github.com/prometheus/client_golang v1.3.0
Expand Down
14 changes: 14 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ github.com/SkycoinProject/dmsg v0.0.0-20200306152741-acee74fa4514/go.mod h1:Dzyk
github.com/SkycoinProject/skycoin v0.26.0/go.mod h1:xqPLOKh5B6GBZlGA7B5IJfQmCy7mwimD9NlqxR3gMXo=
github.com/SkycoinProject/skycoin v0.27.0 h1:N3IHxj8ossHOcsxLYOYugT+OaELLncYHJHxbbYLPPmY=
github.com/SkycoinProject/skycoin v0.27.0/go.mod h1:xqPLOKh5B6GBZlGA7B5IJfQmCy7mwimD9NlqxR3gMXo=
github.com/SkycoinProject/skywire-mainnet v0.0.0-20200309204032-14af5342da86/go.mod h1:xuOpE5ZZU2kR39u0tJWtOpak/sJpnEFj1HpTxtyPU/A=
github.com/SkycoinProject/yamux v0.0.0-20191213015001-a36efeefbf6a h1:6nHCJqh7trsuRcpMC5JmtDukUndn2VC9sY64K6xQ7hQ=
github.com/SkycoinProject/yamux v0.0.0-20191213015001-a36efeefbf6a/go.mod h1:IaE1dxncLQs4RJcQTZPikJfAZY4szH87u2h0lT0SDuM=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
Expand Down Expand Up @@ -117,18 +118,25 @@ github.com/klauspost/pgzip v1.2.1 h1:oIPZROsWuPHpOdMVWLuJZXwgjhrW8r1yEX8UqMyeNHM
github.com/klauspost/pgzip v1.2.1/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA=
github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
github.com/mattn/go-isatty v0.0.8 h1:HLtExJ+uU2HOZ+wI0Tt5DtUDrx8yhUqDcp7fYERX4CE=
github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b h1:j7+1HpAFS1zy5+Q4qx1fWh90gTKwiN4QCGoY9TWyyO4=
Expand Down Expand Up @@ -229,6 +237,8 @@ golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnf
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 h1:ObdrDkeb4kJdCP557AjRjq69pTHfNouLtWZG7j9rPN8=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073 h1:xMPOj6Pz6UipU1wXLkrtqpHbR0AVFnyPEQq/wRWz9lM=
golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
Expand Down Expand Up @@ -259,6 +269,10 @@ golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191220142924-d4481acd189f h1:68K/z8GLUxV76xGSqwTWw2gyk/jwn79LUL43rES2g8o=
golang.org/x/sys v0.0.0-20191220142924-d4481acd189f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 h1:uYVVQ9WP/Ds2ROhcaGPeIdVq0RIXVLwsHlnvJ+cT1So=
golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
Expand Down
41 changes: 23 additions & 18 deletions pkg/router/route_group.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -367,35 +367,37 @@ func (rg *RouteGroup) keepAliveLoop(interval time.Duration) {
continue
}

if err := rg.sendKeepAlive(); err != nil {
rg.logger.Warnf("Failed to send keepalive: %v", err)
}
rg.sendKeepAlive()
}
}
}

func (rg *RouteGroup) sendKeepAlive() error {
func (rg *RouteGroup) sendKeepAlive() {
rg.mu.Lock()
defer rg.mu.Unlock()

if len(rg.tps) == 0 || len(rg.fwd) == 0 {
// if no transports, no rules, then no keepalive
return nil
return
}

tp := rg.tps[0]
rule := rg.fwd[0]
for i := 0; i < len(rg.tps); i++ {
tp := rg.tps[i]
rule := rg.fwd[i]

if tp == nil {
return ErrBadTransport
}
if tp == nil {
continue
}

packet := routing.MakeKeepAlivePacket(rule.NextRouteID())
if err := rg.writePacket(context.Background(), tp, packet, rule.KeyRouteID()); err != nil {
return err
}
packet := routing.MakeKeepAlivePacket(rule.NextRouteID())
errCh := rg.writePacketAsync(context.Background(), tp, packet, rule.KeyRouteID())

return nil
go func() {
if err := <-errCh; err != nil {
rg.logger.WithError(err).Warnf("Failed to send keepalive")
}
}()
}
}

// Close closes a RouteGroup with the specified close `code`:
Expand Down Expand Up @@ -490,9 +492,12 @@ func (rg *RouteGroup) handleClosePacket(code routing.CloseCode) error {
func (rg *RouteGroup) broadcastClosePackets(code routing.CloseCode) {
for i := 0; i < len(rg.tps); i++ {
packet := routing.MakeClosePacket(rg.fwd[i].NextRouteID(), code)
if err := rg.writePacket(context.Background(), rg.tps[i], packet, rg.fwd[i].KeyRouteID()); err != nil {
rg.logger.WithError(err).Errorf("Failed to send close packet to %s", rg.tps[i].Remote())
}
errCh := rg.writePacketAsync(context.Background(), rg.tps[i], packet, rg.fwd[i].KeyRouteID())
go func(tp *transport.ManagedTransport) {
if err := <-errCh; err != nil {
rg.logger.WithError(err).Errorf("Failed to send close packet to %s", tp.Remote())
}
}(rg.tps[i])
}
}

Expand Down
40 changes: 27 additions & 13 deletions pkg/router/router.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,24 +294,38 @@ func (r *router) Serve(ctx context.Context) error {
}

func (r *router) serveTransportManager(ctx context.Context) {
var once sync.Once
done := make(chan struct{})

for {
packet, err := r.tm.ReadPacket()
if err != nil {
if err == transport.ErrNotServing {
r.logger.WithError(err).Info("Stopped reading packets")
return
}
r.logger.WithError(err).Error("Stopped reading packets due to unexpected error.")
select {
case <-done:
return
}

if err := r.handleTransportPacket(ctx, packet); err != nil {
if err == transport.ErrNotServing {
r.logger.WithError(err).Warnf("Stopped serving Transport.")
default:
packet, err := r.tm.ReadPacket()
if err != nil {
if err == transport.ErrNotServing {
r.logger.WithError(err).Info("Stopped reading packets")
return
}
r.logger.WithError(err).Error("Stopped reading packets due to unexpected error.")
return
}

r.logger.Warnf("Failed to handle transport frame: %v", err)
go func(packet routing.Packet) {
if err := r.handleTransportPacket(ctx, packet); err != nil {
if err == transport.ErrNotServing {
once.Do(func() {
r.logger.WithError(err).Warnf("Stopped serving Transport.")
close(done)
})

return
}

r.logger.Warnf("Failed to handle transport frame: %v", err)
}
}(packet)
}
}
}
Expand Down