yggdrasil-network/yggdrasil-go
1
0
Fork 0
mirror of https://github.com/yggdrasil-network/yggdrasil-go.git synced 2025-04-28 22:25:07 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #475 from Arceliar/misc

Browse source 
Misc
This commit is contained in:
Neil Alexander 2019-07-29 20:24:49 +01:00 • committed by GitHub
commit 853054eb62
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 83 additions and 136 deletions
Download patch file Download diff file Expand all files Collapse all files

60
src/yggdrasil/conn.go
View file

@ -152,11 +152,9 @@ func (c *Conn) Read(b []byte) (int, error) {
if !ok {
return 0, ConnError{errors.New("session closed"), false, false, true, 0}
}
defer util.PutBytes(p.Payload)
var err error
done := make(chan struct{})
workerFunc := func() {
defer close(done)
sessionFunc := func() {
defer util.PutBytes(p.Payload)
// If the nonce is bad then drop the packet and return an error
if !sinfo.nonceIsOK(&p.Nonce) {
err = ConnError{errors.New("packet dropped due to invalid nonce"), false, true, false, 0}
@ -176,33 +174,7 @@ func (c *Conn) Read(b []byte) (int, error) {
sinfo.time = time.Now()
sinfo.bytesRecvd += uint64(len(bs))
}
// Hand over to the session worker
defer func() {
if recover() != nil {
err = ConnError{errors.New("read failed, session already closed"), false, false, true, 0}
close(done)
}
}() // In case we're racing with a close
// Send to worker
select {
case sinfo.worker <- workerFunc:
case <-cancel.Finished():
if cancel.Error() == util.CancellationTimeoutError {
return 0, ConnError{errors.New("read timeout"), true, false, false, 0}
} else {
return 0, ConnError{errors.New("session closed"), false, false, true, 0}
}
}
// Wait for the worker to finish
select {
case <-done: // Wait for the worker to finish, failing this can cause memory errors (util.[Get||Put]Bytes stuff)
case <-cancel.Finished():
if cancel.Error() == util.CancellationTimeoutError {
return 0, ConnError{errors.New("read timeout"), true, false, false, 0}
} else {
return 0, ConnError{errors.New("session closed"), false, false, true, 0}
}
}
sinfo.doFunc(sessionFunc)
// Something went wrong in the session worker so abort
if err != nil {
if ce, ok := err.(*ConnError); ok && ce.Temporary() {
@ -223,10 +195,8 @@ func (c *Conn) Read(b []byte) (int, error) {
func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
sinfo := c.session
var packet []byte
done := make(chan struct{})
written := len(b)
workerFunc := func() {
defer close(done)
sessionFunc := func() {
// Does the packet exceed the permitted size for the session?
if uint16(len(b)) > sinfo.getMTU() {
written, err = 0, ConnError{errors.New("packet too big"), true, false, false, int(sinfo.getMTU())}
@ -273,27 +243,7 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
default: // Don't do anything, to keep traffic throttled
}
}
// Set up a timer so this doesn't block forever
cancel := c.getDeadlineCancellation(&c.writeDeadline)
defer cancel.Cancel(nil)
// Hand over to the session worker
defer func() {
if recover() != nil {
err = ConnError{errors.New("write failed, session already closed"), false, false, true, 0}
close(done)
}
}() // In case we're racing with a close
select { // Send to worker
case sinfo.worker <- workerFunc:
case <-cancel.Finished():
if cancel.Error() == util.CancellationTimeoutError {
return 0, ConnError{errors.New("write timeout"), true, false, false, 0}
} else {
return 0, ConnError{errors.New("session closed"), false, false, true, 0}
}
}
// Wait for the worker to finish, otherwise there are memory errors ([Get||Put]Bytes stuff)
<-done
sinfo.doFunc(sessionFunc)
// Give the packet to the router
if written > 0 {
sinfo.core.router.out(packet)