Cloak/internal/multiplex/switchboard.go

153 lines
3.5 KiB
Go
Raw Normal View History

2018-10-05 22:44:20 +00:00
package multiplex
import (
2018-10-28 22:51:00 +00:00
"errors"
2019-08-02 14:45:33 +00:00
log "github.com/sirupsen/logrus"
2018-10-05 22:44:20 +00:00
"net"
2018-10-28 21:22:38 +00:00
"sync"
"sync/atomic"
2018-10-05 22:44:20 +00:00
)
2018-10-20 20:41:01 +00:00
// switchboard is responsible for keeping the reference of TLS connections between client and server
2018-10-05 22:44:20 +00:00
type switchboard struct {
session *Session
2018-11-07 21:16:13 +00:00
*Valve
2018-10-28 22:51:00 +00:00
2018-11-07 21:16:13 +00:00
// optimum is the connEnclave with the smallest sendQueue
optimum atomic.Value // *connEnclave
2018-10-28 21:22:38 +00:00
cesM sync.RWMutex
ces []*connEnclave
2018-11-07 21:16:13 +00:00
}
func (sb *switchboard) getOptimum() *connEnclave {
if i := sb.optimum.Load(); i == nil {
return nil
} else {
return i.(*connEnclave)
}
}
2018-10-05 22:44:20 +00:00
// Some data comes from a Stream to be sent through one of the many
// remoteConn, but which remoteConn should we use to send the data?
//
// In this case, we pick the remoteConn that has about the smallest sendQueue.
type connEnclave struct {
remoteConn net.Conn
2018-10-28 21:22:38 +00:00
sendQueue uint32
2018-10-05 22:44:20 +00:00
}
2018-11-07 21:16:13 +00:00
func makeSwitchboard(sesh *Session, valve *Valve) *switchboard {
// rates are uint64 because in the usermanager we want the bandwidth to be atomically
// operated (so that the bandwidth can change on the fly).
2018-10-05 22:44:20 +00:00
sb := &switchboard{
2018-10-28 21:22:38 +00:00
session: sesh,
2018-11-07 21:16:13 +00:00
Valve: valve,
2018-10-28 21:22:38 +00:00
ces: []*connEnclave{},
2018-10-05 22:44:20 +00:00
}
return sb
}
2018-10-28 22:51:00 +00:00
var errNilOptimum error = errors.New("The optimal connection is nil")
2018-10-28 21:22:38 +00:00
func (sb *switchboard) send(data []byte) (int, error) {
2018-11-07 21:16:13 +00:00
ce := sb.getOptimum()
2018-10-28 22:51:00 +00:00
if ce == nil {
return 0, errNilOptimum
}
2018-10-28 21:22:38 +00:00
atomic.AddUint32(&ce.sendQueue, uint32(len(data)))
go sb.updateOptimum()
2018-10-20 16:03:39 +00:00
n, err := ce.remoteConn.Write(data)
2018-10-14 19:32:54 +00:00
if err != nil {
2018-11-07 21:16:13 +00:00
return n, err
2018-10-14 19:32:54 +00:00
}
2018-12-26 00:46:39 +00:00
sb.txWait(n)
sb.Valve.AddTx(int64(n))
2018-10-28 21:22:38 +00:00
atomic.AddUint32(&ce.sendQueue, ^uint32(n-1))
go sb.updateOptimum()
return n, nil
}
2018-10-20 16:03:39 +00:00
2018-10-28 21:22:38 +00:00
func (sb *switchboard) updateOptimum() {
2018-11-07 21:16:13 +00:00
currentOpti := sb.getOptimum()
2018-10-28 21:22:38 +00:00
currentOptiQ := atomic.LoadUint32(&currentOpti.sendQueue)
sb.cesM.RLock()
for _, ce := range sb.ces {
ceQ := atomic.LoadUint32(&ce.sendQueue)
if ceQ < currentOptiQ {
currentOpti = ce
currentOptiQ = ceQ
}
2018-10-20 16:03:39 +00:00
}
2018-10-28 21:22:38 +00:00
sb.cesM.RUnlock()
sb.optimum.Store(currentOpti)
2018-10-28 21:22:38 +00:00
}
2018-10-20 16:03:39 +00:00
2018-10-28 21:22:38 +00:00
func (sb *switchboard) addConn(conn net.Conn) {
var sendQueue uint32
2018-10-28 21:22:38 +00:00
newCe := &connEnclave{
remoteConn: conn,
sendQueue: sendQueue,
2018-10-28 21:22:38 +00:00
}
sb.cesM.Lock()
sb.ces = append(sb.ces, newCe)
sb.cesM.Unlock()
sb.optimum.Store(newCe)
2018-10-28 21:22:38 +00:00
go sb.deplex(newCe)
2018-10-05 22:44:20 +00:00
}
2018-10-28 21:22:38 +00:00
func (sb *switchboard) removeConn(closing *connEnclave) {
sb.cesM.Lock()
for i, ce := range sb.ces {
if closing == ce {
sb.ces = append(sb.ces[:i], sb.ces[i+1:]...)
break
2018-10-05 22:44:20 +00:00
}
}
2018-10-28 21:22:38 +00:00
if len(sb.ces) == 0 {
sb.session.SetTerminalMsg("no underlying connection left")
2018-10-28 21:22:38 +00:00
sb.session.Close()
}
2018-11-07 21:16:13 +00:00
sb.cesM.Unlock()
2018-10-28 21:22:38 +00:00
}
2018-12-30 01:15:46 +00:00
// actively triggered by session.Close()
2018-12-31 11:30:39 +00:00
func (sb *switchboard) closeAll() {
sb.cesM.RLock()
2018-10-28 21:22:38 +00:00
for _, ce := range sb.ces {
ce.remoteConn.Close()
}
sb.cesM.RUnlock()
2018-10-05 22:44:20 +00:00
}
2019-08-02 15:37:48 +00:00
// deplex function costantly reads from a TCP connection, call Deobfs and distribute it
2018-12-31 11:30:39 +00:00
// to the corresponding stream
2018-10-09 20:53:55 +00:00
func (sb *switchboard) deplex(ce *connEnclave) {
2018-10-16 20:13:19 +00:00
buf := make([]byte, 20480)
2018-10-09 20:53:55 +00:00
for {
2019-08-02 15:37:48 +00:00
n, err := sb.session.unitRead(ce.remoteConn, buf)
2018-11-07 21:16:13 +00:00
sb.rxWait(n)
sb.Valve.AddRx(int64(n))
2018-10-09 20:53:55 +00:00
if err != nil {
2019-08-02 14:45:33 +00:00
log.Tracef("a connection for session %v has closed: %v", sb.session.id, err)
2018-10-09 20:53:55 +00:00
go ce.remoteConn.Close()
2018-10-28 21:22:38 +00:00
sb.removeConn(ce)
2018-10-09 20:53:55 +00:00
return
}
2019-08-02 15:37:48 +00:00
frame, err := sb.session.Deobfs(buf[:n])
2018-12-09 23:45:06 +00:00
if err != nil {
2019-08-02 14:45:33 +00:00
log.Debugf("Failed to decrypt a frame for session %v: %v", sb.session.id, err)
2018-12-09 23:45:06 +00:00
continue
}
2018-11-07 21:16:13 +00:00
2018-12-31 11:30:39 +00:00
stream := sb.session.getStream(frame.StreamID, frame.Closing == 1)
2018-11-24 00:55:26 +00:00
// if the frame is telling us to close a closed stream
// (this happens when ss-server and ss-local closes the stream
// simutaneously), we don't do anything
if stream != nil {
stream.writeNewFrame(frame)
2018-10-20 10:35:50 +00:00
}
2018-10-05 22:44:20 +00:00
}
}