From 8599c87f0ca5a0536aaacb112983fff5c98287d9 Mon Sep 17 00:00:00 2001 From: "dmy@lab" Date: Mon, 11 Jan 2016 21:29:48 +0800 Subject: [PATCH] =?UTF-8?q?1.=E5=8F=AA=E7=95=99=E4=B8=8BBlockingQueueCache?= =?UTF-8?q?=202.=E4=B8=BA=E4=BA=86=E9=81=BF=E5=85=8D=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=E5=BC=95=E7=94=A8=EF=BC=8C=E9=87=8D=E6=96=B0=E6=95=B4=E7=90=86?= =?UTF-8?q?package=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cache/queue.go | 32 ++-------------- cache/queue_test.go | 37 ++++++++++--------- .../constant.go => constant/procotolConst.go | 2 +- {protocol => model}/tunnel.go | 15 ++++---- protocol/tcp.go | 10 +++-- stat/tunnelstatus.go | 20 ++++++++-- tscipher/cipher.go | 20 +--------- 7 files changed, 57 insertions(+), 79 deletions(-) rename protocol/constant.go => constant/procotolConst.go (81%) rename {protocol => model}/tunnel.go (90%) diff --git a/cache/queue.go b/cache/queue.go index 97eea15..d366d25 100644 --- a/cache/queue.go +++ b/cache/queue.go @@ -9,33 +9,7 @@ type QueueCache struct { //FIFO queue chan *Cache } -type BlockingQueueCache QueueCache - -type UnblockingQueueCache QueueCache - -func NewUnblockingQueueCache(quN int) *UnblockingQueueCache { - t := &UnblockingQueueCache{ - queue: make(chan *Cache, quN), - } - return t -} - -func (this *UnblockingQueueCache) Put(bits []byte, l int) { - this.queue <- &Cache{ - cache: bits, - l: l, - } -} - -func (this *UnblockingQueueCache) Get() ([]byte, int) { - var t *Cache - // select { - t = <-this.queue - // default: - // return make([]byte, 1024*4), 1024 * 4 - // } - return t.cache, t.l -} +type BlockingQueueCache QueueCache func NewBlockingQueueCache(quN int) *BlockingQueueCache { t := &BlockingQueueCache{ @@ -52,6 +26,6 @@ func (this *BlockingQueueCache) Put(bits []byte, l int) { } func (this *BlockingQueueCache) Get() ([]byte, int) { - t := <-this.queue + t := <-this.queue return t.cache, t.l -} \ No newline at end of file +} diff --git a/cache/queue_test.go b/cache/queue_test.go index 454edc3..edea5b7 100644 --- a/cache/queue_test.go +++ b/cache/queue_test.go @@ -1,25 +1,28 @@ package cache -import( - "testing" +import ( "fmt" + "testing" ) - -func GetAndPut(name string,get *BlockingQueueCache,put *BlockingQueueCache){ - for{ - r,_:=get.Get() - fmt.Printf("%s Get\n",name) - put.Put(r,1) - fmt.Printf("%s Put\n",name) +func GetAndPut(name string, get *BlockingQueueCache, put *BlockingQueueCache) { + r, _ := get.Get() + fmt.Printf("%s Get\n", name) + put.Put(r, 1) + fmt.Printf("%s Put\n", name) } -} +func TestAsynchronousCache(t *testing.T) { + a := NewBlockingQueueCache(1) + b := NewBlockingQueueCache(1) + a.Put(make([]byte, 1), 1) + go func() { + for { + GetAndPut("A", a, b) + } + }() + for i := 0; i < 1000; i++ { + GetAndPut("B", b, a) + } -func TestAsynchronousCache(t *testing.T){ - a:=NewBlockingQueueCache(1) - b:=NewBlockingQueueCache(1) - a.Put(make([]byte,1),1) - go GetAndPut("A",a,b) - GetAndPut("B",b,a) -} \ No newline at end of file +} diff --git a/protocol/constant.go b/constant/procotolConst.go similarity index 81% rename from protocol/constant.go rename to constant/procotolConst.go index a61cc4f..f45a365 100644 --- a/protocol/constant.go +++ b/constant/procotolConst.go @@ -1,4 +1,4 @@ -package protocol +package constant type Direction int diff --git a/protocol/tunnel.go b/model/tunnel.go similarity index 90% rename from protocol/tunnel.go rename to model/tunnel.go index 6ca234e..e25dd23 100644 --- a/protocol/tunnel.go +++ b/model/tunnel.go @@ -1,9 +1,10 @@ -package protocol +package model import ( // "fmt" "github.com/TransX/cache" + "github.com/TransX/constant" "github.com/TransX/log" "github.com/TransX/tscipher" "net" @@ -14,12 +15,12 @@ type Tunnel struct { id string src net.Conn dest net.Conn - cipherDirection Direction + cipherDirection constant.Direction regChan chan interface{} unregChan chan interface{} } -func NewTunnel(id string, src, dest net.Conn, cipherDirection Direction) *Tunnel { +func NewTunnel(id string, src, dest net.Conn, cipherDirection constant.Direction) *Tunnel { return &Tunnel{ id: id, src: src, @@ -28,7 +29,7 @@ func NewTunnel(id string, src, dest net.Conn, cipherDirection Direction) *Tunnel } } -func (this *Tunnel) GetID(id string) string { +func (this *Tunnel) GetID() string { return this.id } @@ -54,7 +55,7 @@ func (this *Tunnel) Run() { //单向的，从src发送到dest id := this.id // cache := make([]byte, 1024*4) //4kB //构建Carrier - queCache := cache.NewUnblockingQueueCache(1) + queCache := cache.NewBlockingQueueCache(1) for i := 0; i < 1; i++ { queCache.Put(make([]byte, 1024*4), 0) } @@ -88,7 +89,7 @@ func (this *Tunnel) receive(revCarrier *tscipher.Carrier) { var err error for { rTimer := time.Now() //receive timer - if cipherDirection != RECEIVE { + if cipherDirection != constant.RECEIVE { revCarrier.Cipher = nil n, err = tscipher.RowReceiveData(revCarrier) } else { @@ -119,7 +120,7 @@ func (this *Tunnel) send(sendCarrier *tscipher.Carrier) { } } }() - if cipherDirection != SEND { + if cipherDirection != constant.SEND { sendCarrier.Cipher = nil } for { diff --git a/protocol/tcp.go b/protocol/tcp.go index 8fc3100..109f392 100644 --- a/protocol/tcp.go +++ b/protocol/tcp.go @@ -3,7 +3,9 @@ package protocol import ( "errors" "fmt" + "github.com/TransX/constant" "github.com/TransX/log" + "github.com/TransX/model" "github.com/TransX/stat" "github.com/TransX/utils" "net" @@ -77,11 +79,11 @@ func (this *TransTCP) Start(listenPort, destIP, destPort string, clientOrServer //tunnel model : [ -->>server ---- client -->> ](this is a tunnel) if clientOrServer == "client" { sendID := utils.TunnelID() - ntSend := NewTunnel(sendID, listenerConn, destConn, SEND) + ntSend := model.NewTunnel(sendID, listenerConn, destConn, constant.SEND) ntSend.SetRegChan(tunMng.GetRegChan()) ntSend.SetUnRegChan(tunMng.GetUnregChan()) receiveID := utils.TunnelID() - ntRev := NewTunnel(receiveID, destConn, listenerConn, RECEIVE) + ntRev := model.NewTunnel(receiveID, destConn, listenerConn, constant.RECEIVE) ntRev.SetRegChan(tunMng.GetRegChan()) ntRev.SetUnRegChan(tunMng.GetUnregChan()) printModelDetail(sendID, receiveID) @@ -90,11 +92,11 @@ func (this *TransTCP) Start(listenPort, destIP, destPort string, clientOrServer } if clientOrServer == "server" { receiveID := utils.TunnelID() - ntRev := NewTunnel(receiveID, listenerConn, destConn, RECEIVE) + ntRev := model.NewTunnel(receiveID, listenerConn, destConn, constant.RECEIVE) ntRev.SetRegChan(tunMng.GetRegChan()) ntRev.SetUnRegChan(tunMng.GetUnregChan()) sendID := utils.TunnelID() - ntSend := NewTunnel(sendID, destConn, listenerConn, SEND) + ntSend := model.NewTunnel(sendID, destConn, listenerConn, constant.SEND) ntSend.SetRegChan(tunMng.GetRegChan()) ntSend.SetUnRegChan(tunMng.GetUnregChan()) printModelDetail(sendID, receiveID) diff --git a/stat/tunnelstatus.go b/stat/tunnelstatus.go index 24f03c9..2eb12a5 100644 --- a/stat/tunnelstatus.go +++ b/stat/tunnelstatus.go @@ -4,6 +4,7 @@ import ( "container/list" "fmt" "github.com/TransX/log" + "github.com/TransX/model" "sync" ) @@ -40,12 +41,25 @@ func (this *TunnelStatusManager) unregister(t interface{}) { this.mux.Lock() defer this.mux.Unlock() l := this.tunnelList + log.Debug("%d tunnels before remove.", this.tunnelList.Len()) + n := 0 + nl := new(list.List) for e := l.Front(); e != nil; e = e.Next() { - if e == t { - l.Remove(e) - break + if e.Value != t { + // log.Debug("Found") + // l.Remove(e) + // break + n++ + nl.PushBack(e.Value) } + } + if n == this.tunnelList.Len() { + m := t.(*model.Tunnel) + log.Debug("check tunntel %s", m.GetID()) + } + this.tunnelList = *nl + log.Debug("%d tunnels after remove.", this.tunnelList.Len()) } diff --git a/tscipher/cipher.go b/tscipher/cipher.go index d0af920..2e7a1d9 100644 --- a/tscipher/cipher.go +++ b/tscipher/cipher.go @@ -39,13 +39,13 @@ type Cipher interface { type Carrier struct { Conn net.Conn Cipher Cipher - Cache *cache.UnblockingQueueCache + Cache *cache.BlockingQueueCache Msg *cache.BlockingQueueCache AttachedTunnelID string receiveBuff []byte } -func NewCarrier(conn net.Conn, cipher Cipher, queCache *cache.UnblockingQueueCache, msg *cache.BlockingQueueCache, id string) *Carrier { +func NewCarrier(conn net.Conn, cipher Cipher, queCache *cache.BlockingQueueCache, msg *cache.BlockingQueueCache, id string) *Carrier { t := new(Carrier) t.Conn = conn t.Cipher = cipher @@ -137,8 +137,6 @@ func UnwrapPackage(pacakge []byte) (data []byte, rest []byte, err error) { func SendData(carrier *Carrier) (n int, err error) { msg, nByte := carrier.Msg.Get() - id := carrier.AttachedTunnelID - log.Info("id %s Get Msg", id) if len(msg) < nByte { log.Panic("Cache of send is too small") } @@ -147,7 +145,6 @@ func SendData(carrier *Carrier) (n int, err error) { carrier.Cache.Put(make([]byte, 1024*4), 1024*4) return } - log.Info("id %s AAAAAAAaaa", id) encrypedByte, err := carrier.Cipher.Encrypt(msg[:nByte]) if err != nil { n = 0 @@ -157,35 +154,24 @@ func SendData(carrier *Carrier) (n int, err error) { wraped := WrapPackage(encrypedByte[:nByte]) n, err = carrier.Conn.Write(wraped) carrier.Cache.Put(make([]byte, 1024*4), 1024*4) - log.Info("id %s give back cache", id) return } func RowReceiveData(carrier *Carrier) (n int, err error) { cache, _ := carrier.Cache.Get() - log.Info("id %s get Cache", carrier.AttachedTunnelID) n, err = carrier.Conn.Read(cache) if err != nil { n = 0 } carrier.Msg.Put(cache, n) - id := carrier.AttachedTunnelID - log.Info("id %s put Msg", id) return } func ReceiveData(carrier *Carrier) (n int, err error) { - // defer func() { - // if r := recover(); r != nil { - // log.Error("ReceiveData err %s", r) - // } - // }() - // log.Debug("id %s wrapedPackage := carrier.GetReceiveBuff()", carrier.AttachedTunnelID) wrapedPackage := carrier.GetReceiveBuff() //make([]byte, 0, cap(carrier.Cache)) var packageData []byte var _rest []byte cache, _ := carrier.Cache.Get() - log.Info("id %s get Cache", carrier.AttachedTunnelID) for { //首先检查这个是不是完整的包，是就返回好了，免得被阻塞 data, rest, err := UnwrapPackage(wrapedPackage) @@ -226,7 +212,5 @@ func ReceiveData(carrier *Carrier) (n int, err error) { } n = len(decrypted) carrier.Msg.Put(decrypted, n) - id := carrier.AttachedTunnelID - log.Info("id %s put Msg", id) return }