@cryptotaxi247 / kubo / commits / 9ff03782e

godep: update h2so5/utp

the godep tool is toying with me...

Henry committed Apr 28, 2015 at 13:40 UTC 9ff03782ee891a7f36841469872d689d9598a615

20 files changed +1962 -1597

Godeps/Godeps.json

+1 -1

@@ -96,7 +96,7 @@

96 },

97 {

98 "ImportPath": "github.com/h2so5/utp",

99 - "Rev": "654d875bb65e96729678180215cf080fe2810371"

99 + "Rev": "6ca83358f5c331028feb9b97c445e9c7354967b0"

100 },

101 {

102 "ImportPath": "github.com/hashicorp/go.net/ipv4",

Godeps/_workspace/src/github.com/h2so5/utp/.travis.yml

+3 -1

@@ -1,7 +1,9 @@

1 language: go

3 script:

4 - - GO_UTP_LOGGING=2 go test -v -bench .

4 + - GO_UTP_LOGGING=2 go test -v

5 + - GOMAXPROCS=4 GO_UTP_LOGGING=2 go test -v

6 - go test -v -race

7 - GO_UTP_LOGGING=2 go run benchmark/main.go -h

8 + - GOMAXPROCS=4 GO_UTP_LOGGING=2 go run benchmark/main.go -h

9 - GO_UTP_LOGGING=2 cd _ucat_test; make test

Godeps/_workspace/src/github.com/h2so5/utp/README.md

+12 -41

@@ -3,55 +3,26 @@ utp

4 μTP (Micro Transport Protocol) implementation

6 -[![Build status](https://ci.appveyor.com/api/projects/status/j1be8y7p6nd2wqqw?svg=true)](https://ci.appveyor.com/project/h2so5/utp)

7 -[![Build Status](https://travis-ci.org/h2so5/utp.svg)](https://travis-ci.org/h2so5/utp)

6 +[![Build status](https://ci.appveyor.com/api/projects/status/j1be8y7p6nd2wqqw?svg=true&branch=master)](https://ci.appveyor.com/project/h2so5/utp)

7 +[![Build Status](https://travis-ci.org/h2so5/utp.svg?branch=master)](https://travis-ci.org/h2so5/utp)

8 [![GoDoc](https://godoc.org/github.com/h2so5/utp?status.svg)](http://godoc.org/github.com/h2so5/utp)

10 http://www.bittorrent.org/beps/bep_0029.html

12 -**warning: This is a buggy alpha version.**

13 -

14 -## Benchmark History

15 -

16 -[![Benchmark status](http://107.170.244.57:80/go-utp-bench.php)]()

17 -

12 ## Installation

14 ```

15 go get github.com/h2so5/utp

16 ```

24 -## Example

25 -

26 -Echo server

27 -

28 -```go

29 -package main

30 -

31 -import (

32 - "time"

33 -

34 - "github.com/h2so5/utp"

35 -)

36 -

37 -func main() {

38 - ln, _ := utp.Listen("utp", ":11000")

39 - defer ln.Close()

40 -

41 - conn, _ := ln.AcceptUTP()

42 - conn.SetKeepAlive(time.Minute)

43 - defer conn.Close()

44 -

45 - for {

46 - var buf [1024]byte

47 - l, err := conn.Read(buf[:])

48 - if err != nil {

49 - break

50 - }

51 - _, err = conn.Write(buf[:l])

52 - if err != nil {

53 - break

54 - }

55 - }

56 -}

18 +## Debug Log

19 +

20 +Use GO_UTP_LOGGING to show debug logs.

21 +

22 +```

23 +GO_UTP_LOGGING=0 go test <- default, no logging

24 +GO_UTP_LOGGING=1 go test

25 +GO_UTP_LOGGING=2 go test

26 +GO_UTP_LOGGING=3 go test

27 +GO_UTP_LOGGING=4 go test <- most verbose

28 ```

Godeps/_workspace/src/github.com/h2so5/utp/addr.go

+22 -14

@@ -2,11 +2,31 @@ package utp

3 import "net"

5 -type UTPAddr struct {

5 +// Addr represents the address of a UTP end point.

6 +type Addr struct {

7 net.Addr

8 }

9 -func (a UTPAddr) Network() string { return "utp" }

10 +// Network returns the address's network name, "utp".

11 +func (a Addr) Network() string { return "utp" }

12 +

13 +// ResolveAddr parses addr as a UTP address of the form "host:port"

14 +// or "[ipv6-host%zone]:port" and resolves a pair of domain name and

15 +// port name on the network net, which must be "utp", "utp4" or

16 +// "utp6". A literal address or host name for IPv6 must be enclosed

17 +// in square brackets, as in "[::1]:80", "[ipv6-host]:http" or

18 +// "[ipv6-host%zone]:80".

19 +func ResolveAddr(n, addr string) (*Addr, error) {

20 + udpnet, err := utp2udp(n)

21 + if err != nil {

22 + return nil, err

23 + }

24 + udp, err := net.ResolveUDPAddr(udpnet, addr)

25 + if err != nil {

26 + return nil, err

27 + }

28 + return &Addr{Addr: udp}, nil

29 +}

31 func utp2udp(n string) (string, error) {

32 switch n {

@@ -20,15 +40,3 @@ func utp2udp(n string) (string, error) {

40 return "", net.UnknownNetworkError(n)

41 }

42 }

23 -

24 -func ResolveUTPAddr(n, addr string) (*UTPAddr, error) {

25 - udpnet, err := utp2udp(n)

26 - if err != nil {

27 - return nil, err

28 - }

29 - udp, err := net.ResolveUDPAddr(udpnet, addr)

30 - if err != nil {

31 - return nil, err

32 - }

33 - return &UTPAddr{Addr: udp}, nil

34 -}

Godeps/_workspace/src/github.com/h2so5/utp/base.go new

+325

@@ -0,0 +1,325 @@

1 +package utp

2 +

3 +import (

4 + "errors"

5 + "net"

6 + "sync"

7 + "sync/atomic"

8 + "syscall"

9 + "time"

10 +)

11 +

12 +var baseConnMap = make(map[string]*baseConn)

13 +var baseConnMutex sync.Mutex

14 +

15 +type packetHandler struct {

16 + send chan<- *packet

17 + closed chan int

18 +}

19 +

20 +type baseConn struct {

21 + addr string

22 + conn net.PacketConn

23 + synPackets *packetRingBuffer

24 + outOfBandPackets *packetRingBuffer

25 +

26 + handlers map[uint16]*packetHandler

27 + handlerMutex sync.RWMutex

28 + ref int32

29 + refMutex sync.RWMutex

30 +

31 + rdeadline time.Time

32 + wdeadline time.Time

33 +

34 + softClosed int32

35 + closed int32

36 +}

37 +

38 +func newBaseConn(n string, addr *Addr) (*baseConn, error) {

39 + udpnet, err := utp2udp(n)

40 + if err != nil {

41 + return nil, err

42 + }

43 + var s string

44 + if addr != nil {

45 + s = addr.String()

46 + } else {

47 + s = ":0"

48 + }

49 + conn, err := net.ListenPacket(udpnet, s)

50 + if err != nil {

51 + return nil, err

52 + }

53 + c := &baseConn{

54 + conn: conn,

55 + synPackets: newPacketRingBuffer(packetBufferSize),

56 + outOfBandPackets: newPacketRingBuffer(packetBufferSize),

57 + handlers: make(map[uint16]*packetHandler),

58 + }

59 + c.Register(-1, nil)

60 + go c.recvLoop()

61 + return c, nil

62 +}

63 +

64 +func getSharedBaseConn(n string, addr *Addr) (*baseConn, error) {

65 + baseConnMutex.Lock()

66 + defer baseConnMutex.Unlock()

67 + var s string

68 + if addr != nil {

69 + s = addr.String()

70 + } else {

71 + s = ":0"

72 + }

73 + if c, ok := baseConnMap[s]; ok {

74 + return c, nil

75 + }

76 + c, err := newBaseConn(n, addr)

77 + if err != nil {

78 + return nil, err

79 + }

80 + c.addr = s

81 + baseConnMap[s] = c

82 + go c.recvLoop()

83 + return c, nil

84 +}

85 +

86 +func (c *baseConn) ok() bool { return c != nil && c.conn != nil }

87 +

88 +func (c *baseConn) LocalAddr() net.Addr {

89 + if !c.ok() {

90 + return nil

91 + }

92 + return &Addr{Addr: c.conn.LocalAddr()}

93 +}

94 +

95 +func (c *baseConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) {

96 + if !c.ok() {

97 + return 0, nil, syscall.EINVAL

98 + }

99 + if !c.isOpen() {

100 + return 0, nil, &net.OpError{

101 + Op: "read",

102 + Net: c.LocalAddr().Network(),

103 + Addr: c.LocalAddr(),

104 + Err: errClosing,

105 + }

106 + }

107 + var d time.Duration

108 + if !c.rdeadline.IsZero() {

109 + d = c.rdeadline.Sub(time.Now())

110 + if d < 0 {

111 + d = 0

112 + }

113 + }

114 + p, err := c.outOfBandPackets.popOne(d)

115 + if err != nil {

116 + return 0, nil, &net.OpError{

117 + Op: "read",

118 + Net: c.LocalAddr().Network(),

119 + Addr: c.LocalAddr(),

120 + Err: err,

121 + }

122 + }

123 + return copy(b, p.payload), p.addr, nil

124 +}

125 +

126 +func (c *baseConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {

127 + if !c.ok() {

128 + return 0, syscall.EINVAL

129 + }

130 + if !c.isOpen() {

131 + return 0, &net.OpError{

132 + Op: "write",

133 + Net: c.LocalAddr().Network(),

134 + Addr: c.LocalAddr(),

135 + Err: errClosing,

136 + }

137 + }

138 + return c.conn.WriteTo(b, addr)

139 +}

140 +

141 +func (c *baseConn) Close() error {

142 + if !c.ok() {

143 + return syscall.EINVAL

144 + }

145 + if c.isOpen() && atomic.CompareAndSwapInt32(&c.softClosed, 0, 1) {

146 + c.Unregister(-1)

147 + } else {

148 + return &net.OpError{

149 + Op: "close",

150 + Net: c.LocalAddr().Network(),

151 + Addr: c.LocalAddr(),

152 + Err: errClosing,

153 + }

154 + }

155 + return nil

156 +}

157 +

158 +func (c *baseConn) SetDeadline(t time.Time) error {

159 + if !c.ok() {

160 + return syscall.EINVAL

161 + }

162 + err := c.SetReadDeadline(t)

163 + if err != nil {

164 + return err

165 + }

166 + return c.SetWriteDeadline(t)

167 +}

168 +

169 +func (c *baseConn) SetReadDeadline(t time.Time) error {

170 + if !c.ok() {

171 + return syscall.EINVAL

172 + }

173 + c.rdeadline = t

174 + return nil

175 +}

176 +

177 +func (c *baseConn) SetWriteDeadline(t time.Time) error {

178 + if !c.ok() {

179 + return syscall.EINVAL

180 + }

181 + c.wdeadline = t

182 + return nil

183 +}

184 +

185 +func (c *baseConn) recvLoop() {

186 + var buf [maxUdpPayload]byte

187 + for {

188 + l, addr, err := c.conn.ReadFrom(buf[:])

189 + if err != nil {

190 + ulog.Printf(3, "baseConn(%v): %v", c.LocalAddr(), err)

191 + return

192 + }

193 + p, err := c.decodePacket(buf[:l])

194 + if err != nil {

195 + ulog.Printf(3, "baseConn(%v): RECV out-of-band packet (len: %d) from %v", c.LocalAddr(), l, addr)

196 + c.outOfBandPackets.push(&packet{payload: append([]byte{}, buf[:l]...), addr: addr})

197 + } else {

198 + p.addr = addr

199 + ulog.Printf(3, "baseConn(%v): RECV: %v from %v", c.LocalAddr(), p, addr)

200 + if p.header.typ == stSyn {

201 + // ignore duplicated syns

202 + if !c.exists(p.header.id + 1) {

203 + c.synPackets.push(p)

204 + }

205 + } else {

206 + c.processPacket(p)

207 + }

208 + }

209 + }

210 +}

211 +

212 +func (c *baseConn) decodePacket(b []byte) (*packet, error) {

213 + var p packet

214 + err := p.UnmarshalBinary(b)

215 + if err != nil {

216 + return nil, err

217 + }

218 + if p.header.ver != version {

219 + return nil, errors.New("unsupported utp version")

220 + }

221 + return &p, nil

222 +}

223 +

224 +func (c *baseConn) exists(id uint16) bool {

225 + c.handlerMutex.RLock()

226 + defer c.handlerMutex.RUnlock()

227 + return c.handlers[id] != nil

228 +}

229 +

230 +func (c *baseConn) processPacket(p *packet) {

231 + c.handlerMutex.RLock()

232 + h, ok := c.handlers[p.header.id]

233 + c.handlerMutex.RUnlock()

234 + if ok {

235 + select {

236 + case <-h.closed:

237 + case h.send <- p:

238 + }

239 + }

240 +}

241 +

242 +func (c *baseConn) Register(id int32, f chan<- *packet) {

243 + if id < 0 {

244 + c.refMutex.Lock()

245 + c.ref++

246 + c.refMutex.Unlock()

247 + } else {

248 + if f == nil {

249 + panic("nil handler not allowed")

250 + }

251 + c.handlerMutex.Lock()

252 + _, ok := c.handlers[uint16(id)]

253 + c.handlerMutex.Unlock()

254 + if !ok {

255 + c.refMutex.Lock()

256 + c.ref++

257 + c.refMutex.Unlock()

258 + c.handlerMutex.Lock()

259 + c.handlers[uint16(id)] = &packetHandler{

260 + send: f,

261 + closed: make(chan int),

262 + }

263 + c.handlerMutex.Unlock()

264 + ulog.Printf(2, "baseConn(%v): register #%d (ref: %d)", c.LocalAddr(), id, c.ref)

265 + }

266 + }

267 +}

268 +

269 +func (c *baseConn) Unregister(id int32) {

270 + if id < 0 {

271 + c.refMutex.Lock()

272 + c.ref--

273 + c.refMutex.Unlock()

274 + } else {

275 + c.handlerMutex.Lock()

276 + f, ok := c.handlers[uint16(id)]

277 + c.handlerMutex.Unlock()

278 + if ok {

279 + c.handlerMutex.Lock()

280 + close(f.closed)

281 + delete(c.handlers, uint16(id))

282 + c.handlerMutex.Unlock()

283 + c.refMutex.Lock()

284 + c.ref--

285 + c.refMutex.Unlock()

286 + }

287 + }

288 + c.refMutex.Lock()

289 + r := c.ref

290 + c.refMutex.Unlock()

291 + if r <= 0 {

292 + baseConnMutex.Lock()

293 + defer baseConnMutex.Unlock()

294 + c.close()

295 + delete(baseConnMap, c.addr)

296 + ulog.Printf(2, "baseConn(%v): unregister #%d (ref: %d)", c.LocalAddr(), id, c.ref)

297 + }

298 +}

299 +

300 +func (c *baseConn) close() {

301 + if atomic.CompareAndSwapInt32(&c.closed, 0, 1) {

302 + c.conn.Close()

303 + }

304 +}

305 +

306 +func (c *baseConn) isOpen() bool {

307 + return atomic.LoadInt32(&c.closed) == 0

308 +}

309 +

310 +func (c *baseConn) Send(p *packet) {

311 + b, err := p.MarshalBinary()

312 + if err != nil {

313 + panic(err)

314 + }

315 + ulog.Printf(3, "baseConn(%v): SEND: %v to %v", c.LocalAddr(), p, p.addr)

316 + _, err = c.conn.WriteTo(b, p.addr)

317 + if err != nil {

318 + ulog.Printf(3, "%v", err)

319 + panic(err)

320 + }

321 +}

322 +

323 +func (c *baseConn) RecvSyn(timeout time.Duration) (*packet, error) {

324 + return c.synPackets.popOne(timeout)

325 +}

Godeps/_workspace/src/github.com/h2so5/utp/base_test.go new

+308

@@ -0,0 +1,308 @@

1 +package utp

2 +

3 +import (

4 + "bytes"

5 + "net"

6 + "sync"

7 + "testing"

8 + "time"

9 +)

10 +

11 +func TestSharedConnRecvPacket(t *testing.T) {

12 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

13 + if err != nil {

14 + t.Fatal(err)

15 + }

16 +

17 + c, err := getSharedBaseConn("utp", addr)

18 + if err != nil {

19 + t.Fatal(err)

20 + }

21 + defer c.Close()

22 +

23 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

24 + if err != nil {

25 + t.Fatal(err)

26 + }

27 +

28 + uc, err := net.DialUDP("udp", nil, uaddr)

29 + if err != nil {

30 + t.Fatal(err)

31 + }

32 + defer uc.Close()

33 +

34 + ch := make(chan *packet)

35 + c.Register(5, ch)

36 +

37 + for i := 0; i < 100; i++ {

38 + p := &packet{header: header{typ: stData, ver: version, id: 5}}

39 + payload, err := p.MarshalBinary()

40 + if err != nil {

41 + t.Fatal(err)

42 + }

43 + go func() {

44 + uc.Write(payload)

45 + }()

46 + <-ch

47 + }

48 +

49 + c.Unregister(5)

50 +}

51 +

52 +func TestSharedConnSendPacket(t *testing.T) {

53 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

54 + if err != nil {

55 + t.Fatal(err)

56 + }

57 +

58 + c, err := getSharedBaseConn("utp", addr)

59 + if err != nil {

60 + t.Fatal(err)

61 + }

62 + defer c.Close()

63 +

64 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

65 + if err != nil {

66 + t.Fatal(err)

67 + }

68 +

69 + uc, err := net.DialUDP("udp", nil, uaddr)

70 + if err != nil {

71 + t.Fatal(err)

72 + }

73 + defer uc.Close()

74 +

75 + for i := 0; i < 100; i++ {

76 + addr, err := net.ResolveUDPAddr("udp", uc.LocalAddr().String())

77 + if err != nil {

78 + t.Fatal(err)

79 + }

80 + p := &packet{header: header{typ: stData, ver: version, id: 5}, addr: addr}

81 + payload, err := p.MarshalBinary()

82 + if err != nil {

83 + t.Fatal(err)

84 + }

85 +

86 + c.Send(p)

87 +

88 + var b [256]byte

89 + l, err := uc.Read(b[:])

90 + if err != nil {

91 + t.Fatal(err)

92 + }

93 +

94 + if !bytes.Equal(b[:l], payload) {

95 + t.Errorf("expected packet of %v; got %v", payload, b[:l])

96 + }

97 + }

98 +}

99 +

100 +func TestSharedConnRecvSyn(t *testing.T) {

101 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

102 + if err != nil {

103 + t.Fatal(err)

104 + }

105 +

106 + c, err := getSharedBaseConn("utp", addr)

107 + if err != nil {

108 + t.Fatal(err)

109 + }

110 + defer c.Close()

111 +

112 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

113 + if err != nil {

114 + t.Fatal(err)

115 + }

116 +

117 + uc, err := net.DialUDP("udp", nil, uaddr)

118 + if err != nil {

119 + t.Fatal(err)

120 + }

121 + defer uc.Close()

122 +

123 + for i := 0; i < 100; i++ {

124 + p := &packet{header: header{typ: stSyn, ver: version}}

125 + payload, err := p.MarshalBinary()

126 + if err != nil {

127 + t.Fatal(err)

128 + }

129 + go func() {

130 + uc.Write(payload)

131 + }()

132 + p, err = c.RecvSyn(time.Duration(0))

133 + if err != nil {

134 + t.Fatal(err)

135 + }

136 + if p == nil {

137 + t.Errorf("packet must not be nil")

138 + }

139 + }

140 +}

141 +

142 +func TestSharedConnRecvOutOfBound(t *testing.T) {

143 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

144 + if err != nil {

145 + t.Fatal(err)

146 + }

147 +

148 + c, err := getSharedBaseConn("utp", addr)

149 + if err != nil {

150 + t.Fatal(err)

151 + }

152 + defer c.Close()

153 +

154 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

155 + if err != nil {

156 + t.Fatal(err)

157 + }

158 +

159 + uc, err := net.DialUDP("udp", nil, uaddr)

160 + if err != nil {

161 + t.Fatal(err)

162 + }

163 + defer uc.Close()

164 +

165 + for i := 0; i < 100; i++ {

166 + payload := []byte("Hello")

167 + go func() {

168 + uc.Write(payload)

169 + }()

170 + var b [256]byte

171 + l, _, err := c.ReadFrom(b[:])

172 + if err != nil {

173 + t.Fatal(err)

174 + }

175 + if !bytes.Equal(payload, b[:l]) {

176 + t.Errorf("expected packet of %v; got %v", payload, b[:l])

177 + }

178 + }

179 +}

180 +

181 +func TestSharedConnSendOutOfBound(t *testing.T) {

182 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

183 + if err != nil {

184 + t.Fatal(err)

185 + }

186 +

187 + c, err := getSharedBaseConn("utp", addr)

188 + if err != nil {

189 + t.Fatal(err)

190 + }

191 + defer c.Close()

192 +

193 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

194 + if err != nil {

195 + t.Fatal(err)

196 + }

197 +

198 + uc, err := net.DialUDP("udp", nil, uaddr)

199 + if err != nil {

200 + t.Fatal(err)

201 + }

202 + defer uc.Close()

203 +

204 + for i := 0; i < 100; i++ {

205 + addr, err := net.ResolveUDPAddr("udp", uc.LocalAddr().String())

206 + if err != nil {

207 + t.Fatal(err)

208 + }

209 + payload := []byte("Hello")

210 + _, err = c.WriteTo(payload, addr)

211 + if err != nil {

212 + t.Fatal(err)

213 + }

214 +

215 + var b [256]byte

216 + l, err := uc.Read(b[:])

217 + if err != nil {

218 + t.Fatal(err)

219 + }

220 +

221 + if !bytes.Equal(payload, b[:l]) {

222 + t.Errorf("expected packet of %v; got %v", payload, b[:l])

223 + }

224 + }

225 +}

226 +

227 +func TestSharedConnReferenceCount(t *testing.T) {

228 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

229 + if err != nil {

230 + t.Fatal(err)

231 + }

232 +

233 + c, err := getSharedBaseConn("utp", addr)

234 + if err != nil {

235 + t.Fatal(err)

236 + }

237 + defer c.Close()

238 +

239 + var w sync.WaitGroup

240 +

241 + c.Register(-1, nil)

242 +

243 + for i := 0; i < 5; i++ {

244 + w.Add(1)

245 + go func(i int) {

246 + defer w.Done()

247 + c.Register(int32(i), make(chan *packet))

248 + }(i)

249 + }

250 +

251 + w.Wait()

252 + for i := 0; i < 5; i++ {

253 + w.Add(1)

254 + go func(i int) {

255 + defer w.Done()

256 + c.Unregister(int32(i))

257 + }(i)

258 + }

259 +

260 + w.Wait()

261 + c.Unregister(-1)

262 + c.Close()

263 +

264 + c = baseConnMap[addr.String()]

265 + if c != nil {

266 + t.Errorf("baseConn should be released", c.ref)

267 + }

268 +}

269 +

270 +func TestSharedConnClose(t *testing.T) {

271 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

272 + if err != nil {

273 + t.Fatal(err)

274 + }

275 +

276 + c, err := getSharedBaseConn("utp", addr)

277 + if err != nil {

278 + t.Fatal(err)

279 + }

280 + defer c.Close()

281 +

282 + for i := 0; i < 5; i++ {

283 + c.Close()

284 + }

285 +

286 + var b [256]byte

287 + _, _, err = c.ReadFrom(b[:])

288 + if err == nil {

289 + t.Fatal("ReadFrom should fail")

290 + }

291 +

292 + uaddr, err := net.ResolveUDPAddr("udp", c.LocalAddr().String())

293 + if err != nil {

294 + t.Fatal(err)

295 + }

296 +

297 + uc, err := net.DialUDP("udp", nil, uaddr)

298 + if err != nil {

299 + t.Fatal(err)

300 + }

301 + defer uc.Close()

302 +

303 + payload := []byte("Hello")

304 + _, err = c.WriteTo(payload, uc.LocalAddr())

305 + if err == nil {

306 + t.Fatal("WriteTo should fail")

307 + }

308 +}

Godeps/_workspace/src/github.com/h2so5/utp/benchmark/main.go

+46 -26

@@ -85,25 +85,27 @@ func main() {

85 }

87 func c2s(l int64, stream bool) float64 {

88 - ln, err := utp.Listen("utp", "127.0.0.1:0")

88 + laddr, err := utp.ResolveAddr("utp", "127.0.0.1:0")

89 if err != nil {

90 log.Fatal(err)

91 }

92 -

93 - raddr, err := utp.ResolveUTPAddr("utp", ln.Addr().String())

92 + ln, err := utp.Listen("utp", laddr)

93 if err != nil {

94 log.Fatal(err)

95 }

98 - c, err := utp.DialUTPTimeout("utp", nil, raddr, 1000*time.Millisecond)

99 - if err != nil {

100 - log.Fatal(err)

101 - }

102 - defer c.Close()

97 + cch := make(chan *utp.Conn)

98 + go func() {

99 + c, err := utp.DialUTPTimeout("utp", nil, ln.Addr().(*utp.Addr), 1000*time.Millisecond)

100 + if err != nil {

101 + log.Fatal(err)

102 + }

103

104 - if err != nil {

105 - log.Fatal(err)

106 - }

104 + if err != nil {

105 + log.Fatal(err)

106 + }

107 + cch <- c

108 + }()

109

110 s, err := ln.Accept()

111 if err != nil {

@@ -112,7 +114,11 @@ func c2s(l int64, stream bool) float64 {

114 defer s.Close()

115 ln.Close()

116

117 + c := <-cch

118 + defer c.Close()

119 +

120 rch := make(chan int)

121 + wch := make(chan int)

122

123 sendHash := md5.New()

124 readHash := md5.New()

@@ -122,12 +128,13 @@ func c2s(l int64, stream bool) float64 {

128 if stream {

129 go func() {

130 defer c.Close()

131 + defer close(wch)

132 io.Copy(io.MultiWriter(c, sendHash, &counter), io.LimitReader(RandReader{}, l))

133 }()

134

135 go func() {

136 + defer close(rch)

137 io.Copy(readHash, s)

130 - close(rch)

138 }()

139

140 go func() {

@@ -148,6 +155,7 @@ func c2s(l int64, stream bool) float64 {

155

156 start := time.Now()

157 <-rch

158 + <-wch

159 bps = float64(l*8) / (float64(time.Now().Sub(start)) / float64(time.Second))

160

161 } else {

@@ -156,16 +164,18 @@ func c2s(l int64, stream bool) float64 {

164

165 go func() {

166 defer c.Close()

167 + defer close(wch)

168 io.Copy(c, &sendBuf)

169 }()

170

171 go func() {

172 + defer close(rch)

173 io.Copy(&readBuf, s)

164 - rch <- 0

174 }()

175

176 start := time.Now()

177 <-rch

178 + <-wch

179 bps = float64(l*8) / (float64(time.Now().Sub(start)) / float64(time.Second))

180

181 io.Copy(sendHash, &sendBuf)

@@ -180,25 +190,27 @@ func c2s(l int64, stream bool) float64 {

190 }

191

192 func s2c(l int64, stream bool) float64 {

183 - ln, err := utp.Listen("utp", "127.0.0.1:0")

193 + laddr, err := utp.ResolveAddr("utp", "127.0.0.1:0")

194 if err != nil {

195 log.Fatal(err)

196 }

187 -

188 - raddr, err := utp.ResolveUTPAddr("utp", ln.Addr().String())

197 + ln, err := utp.Listen("utp", laddr)

198 if err != nil {

199 log.Fatal(err)

200 }

201

193 - c, err := utp.DialUTPTimeout("utp", nil, raddr, 1000*time.Millisecond)

194 - if err != nil {

195 - log.Fatal(err)

196 - }

197 - defer c.Close()

202 + cch := make(chan *utp.Conn)

203 + go func() {

204 + c, err := utp.DialUTPTimeout("utp", nil, ln.Addr().(*utp.Addr), 1000*time.Millisecond)

205 + if err != nil {

206 + log.Fatal(err)

207 + }

208

199 - if err != nil {

200 - log.Fatal(err)

201 - }

209 + if err != nil {

210 + log.Fatal(err)

211 + }

212 + cch <- c

213 + }()

214

215 s, err := ln.Accept()

216 if err != nil {

@@ -207,7 +219,11 @@ func s2c(l int64, stream bool) float64 {

219 defer s.Close()

220 ln.Close()

221

222 + c := <-cch

223 + defer c.Close()

224 +

225 rch := make(chan int)

226 + wch := make(chan int)

227

228 sendHash := md5.New()

229 readHash := md5.New()

@@ -218,12 +234,13 @@ func s2c(l int64, stream bool) float64 {

234 if stream {

235 go func() {

236 defer s.Close()

237 + defer close(wch)

238 io.Copy(io.MultiWriter(s, sendHash, &counter), io.LimitReader(RandReader{}, l))

239 }()

240

241 go func() {

242 + defer close(rch)

243 io.Copy(readHash, c)

226 - close(rch)

244 }()

245

246 go func() {

@@ -244,6 +261,7 @@ func s2c(l int64, stream bool) float64 {

261

262 start := time.Now()

263 <-rch

264 + <-wch

265 bps = float64(l*8) / (float64(time.Now().Sub(start)) / float64(time.Second))

266

267 } else {

@@ -252,16 +270,18 @@ func s2c(l int64, stream bool) float64 {

270

271 go func() {

272 defer s.Close()

273 + defer close(wch)

274 io.Copy(s, &sendBuf)

275 }()

276

277 go func() {

278 + defer close(rch)

279 io.Copy(&readBuf, c)

260 - rch <- 0

280 }()

281

282 start := time.Now()

283 <-rch

284 + <-wch

285 bps = float64(l*8) / (float64(time.Now().Sub(start)) / float64(time.Second))

286

287 io.Copy(sendHash, &sendBuf)

Godeps/_workspace/src/github.com/h2so5/utp/buffer.go

+272 -30

@@ -2,7 +2,9 @@ package utp

3 import (

4 "errors"

5 + "io"

6 "math"

7 + "sync"

8 "time"

9 )

@@ -74,7 +76,7 @@ func (b *packetBuffer) compact() {

76 }

77 }

77 -func (b *packetBuffer) first() *packet {

79 +func (b *packetBuffer) front() *packet {

80 if b.root == nil || b.root.p == nil {

81 return nil

82 }

@@ -157,6 +159,9 @@ func (b *packetBuffer) generateSelectiveACK() []byte {

159 ack = ack[:len(ack)-1]

160 }

161

162 + if len(ack) == 0 {

163 + return nil

164 + }

165 return ack

166 }

167

@@ -185,46 +190,283 @@ func (b *packetBuffer) processSelectiveACK(ack []byte) {

190 }

191 }

192

188 -type timedBuffer struct {

189 - d time.Duration

190 - root *timedBufferNode

193 +type packetRingBuffer struct {

194 + b []*packet

195 + begin int

196 + s int

197 + mutex sync.RWMutex

198 + rch chan int

199 }

200

193 -type timedBufferNode struct {

194 - val float64

195 - next *timedBufferNode

196 - pushed time.Time

201 +func newPacketRingBuffer(s int) *packetRingBuffer {

202 + return &packetRingBuffer{

203 + b: make([]*packet, s),

204 + rch: make(chan int),

205 + }

206 +}

207 +

208 +func (b *packetRingBuffer) size() int {

209 + b.mutex.RLock()

210 + defer b.mutex.RUnlock()

211 + return b.s

212 +}

213 +

214 +func (b *packetRingBuffer) empty() bool {

215 + return b.size() == 0

216 +}

217 +

218 +func (b *packetRingBuffer) push(p *packet) {

219 + b.mutex.Lock()

220 + defer b.mutex.Unlock()

221 + b.b[(b.begin+b.s)%len(b.b)] = p

222 + if b.s < len(b.b) {

223 + b.s++

224 + } else {

225 + b.begin = (b.begin + 1) % len(b.b)

226 + }

227 + select {

228 + case b.rch <- 0:

229 + default:

230 + }

231 +}

232 +

233 +func (b *packetRingBuffer) pop() *packet {

234 + if b.empty() {

235 + return nil

236 + }

237 + b.mutex.Lock()

238 + defer b.mutex.Unlock()

239 + p := b.b[b.begin]

240 + b.begin = (b.begin + 1) % len(b.b)

241 + b.s--

242 + return p

243 +}

244 +

245 +func (b *packetRingBuffer) popOne(timeout time.Duration) (*packet, error) {

246 + t := time.NewTimer(timeout)

247 + defer t.Stop()

248 + if timeout == 0 {

249 + t.Stop()

250 + }

251 + if b.empty() {

252 + select {

253 + case <-b.rch:

254 + case <-t.C:

255 + return nil, errTimeout

256 + }

257 + }

258 + return b.pop(), nil

259 +}

260 +

261 +type byteRingBuffer struct {

262 + b []byte

263 + begin int

264 + s int

265 + mutex sync.RWMutex

266 + rch chan int

267 + closech chan int

268 + closechMutex sync.Mutex

269 +}

270 +

271 +func newByteRingBuffer(s int) *byteRingBuffer {

272 + return &byteRingBuffer{

273 + b: make([]byte, s),

274 + rch: make(chan int),

275 + closech: make(chan int),

276 + }

277 }

278

199 -func (b *timedBuffer) push(val float64) {

200 - var before *timedBufferNode

201 - for n := b.root; n != nil; n = n.next {

202 - if time.Now().Sub(n.pushed) >= b.d {

203 - if before != nil {

204 - before.next = nil

205 - } else {

206 - b.root = nil

279 +func (r *byteRingBuffer) size() int {

280 + r.mutex.RLock()

281 + defer r.mutex.RUnlock()

282 + return r.s

283 +}

284 +

285 +func (r *byteRingBuffer) space() int {

286 + r.mutex.RLock()

287 + defer r.mutex.RUnlock()

288 + return len(r.b) - r.s

289 +}

290 +

291 +func (r *byteRingBuffer) empty() bool {

292 + return r.size() == 0

293 +}

294 +

295 +func (r *byteRingBuffer) Write(b []byte) (int, error) {

296 + r.mutex.Lock()

297 + defer r.mutex.Unlock()

298 +

299 + for len(b) > 0 {

300 + end := (r.begin + r.s) % len(r.b)

301 + n := copy(r.b[end:], b)

302 + b = b[n:]

303 +

304 + s := r.s + n

305 + if s > len(r.b) {

306 + r.begin = (r.begin + s - len(r.b)) % len(r.b)

307 + r.s = len(r.b)

308 + } else {

309 + r.s += n

310 + }

311 + }

312 + select {

313 + case r.rch <- 0:

314 + case <-r.closech:

315 + return 0, io.EOF

316 + default:

317 + }

318 + return len(b), nil

319 +}

320 +

321 +func (r *byteRingBuffer) ReadTimeout(b []byte, timeout time.Duration) (int, error) {

322 + t := time.NewTimer(timeout)

323 + defer t.Stop()

324 + if timeout == 0 {

325 + t.Stop()

326 + }

327 + if r.empty() {

328 + select {

329 + case <-r.rch:

330 + case <-t.C:

331 + return 0, errTimeout

332 + case <-r.closech:

333 + return 0, io.EOF

334 + }

335 + }

336 + l := r.size()

337 + if l > len(b) {

338 + l = len(b)

339 + }

340 + r.mutex.Lock()

341 + defer r.mutex.Unlock()

342 + if r.begin+l > len(r.b) {

343 + n := copy(b, r.b[r.begin:])

344 + n = copy(b[n:], r.b[:])

345 + r.begin = n

346 + } else {

347 + copy(b, r.b[r.begin:r.begin+l])

348 + r.begin = (r.begin + l) % len(r.b)

349 + }

350 + r.s -= l

351 + return l, nil

352 +}

353 +

354 +func (r *byteRingBuffer) Close() error {

355 + r.closechMutex.Lock()

356 + defer r.closechMutex.Unlock()

357 + select {

358 + case <-r.closech:

359 + return errClosing

360 + default:

361 + close(r.closech)

362 + }

363 + return nil

364 +}

365 +

366 +type rateLimitedBuffer struct {

367 + wch chan<- []byte

368 + closech chan int

369 + closechMutex sync.Mutex

370 + size uint32

371 + sizech chan uint32

372 + sizeMutex sync.Mutex

373 +}

374 +

375 +func newRateLimitedBuffer(ch chan<- []byte, size uint32) *rateLimitedBuffer {

376 + return &rateLimitedBuffer{

377 + wch: ch,

378 + closech: make(chan int),

379 + size: size,

380 + sizech: make(chan uint32),

381 + }

382 +}

383 +

384 +func (r *rateLimitedBuffer) WriteTimeout(b []byte, timeout time.Duration) (int, error) {

385 + t := time.NewTimer(timeout)

386 + defer t.Stop()

387 + if timeout == 0 {

388 + t.Stop()

389 + }

390 +

391 + for wrote := uint32(0); wrote < uint32(len(b)); {

392 + r.sizeMutex.Lock()

393 + s := r.size

394 + r.sizeMutex.Unlock()

395 + if s == 0 {

396 + select {

397 + case ns := <-r.sizech:

398 + s = ns

399 + case <-r.closech:

400 + return 0, errClosing

401 }

208 - break

402 }

210 - before = n

403 + if s > uint32(len(b))-wrote {

404 + s = uint32(len(b)) - wrote

405 + }

406 + select {

407 + case r.wch <- append([]byte{}, b[wrote:wrote+s]...):

408 + wrote += s

409 + r.sizeMutex.Lock()

410 + r.size -= uint32(s)

411 + r.sizeMutex.Unlock()

412 + case <-r.closech:

413 + return 0, errClosing

414 + case <-t.C:

415 + return 0, errTimeout

416 + }

417 }

212 - b.root = &timedBufferNode{

213 - val: val,

214 - next: b.root,

215 - pushed: time.Now(),

418 +

419 + return len(b), nil

420 +}

421 +

422 +func (r *rateLimitedBuffer) Reset(size uint32) {

423 + r.sizeMutex.Lock()

424 + defer r.sizeMutex.Unlock()

425 + r.size = size

426 + select {

427 + case r.sizech <- size:

428 + default:

429 }

430 }

431

219 -func (b *timedBuffer) min() float64 {

220 - if b.root == nil {

221 - return 0

432 +func (r *rateLimitedBuffer) Close() error {

433 + r.closechMutex.Lock()

434 + defer r.closechMutex.Unlock()

435 + select {

436 + case <-r.closech:

437 + return errClosing

438 + default:

439 + close(r.closech)

440 }

223 - min := b.root.val

224 - for n := b.root; n != nil; n = n.next {

225 - if min > n.val {

226 - min = n.val

441 + return nil

442 +}

443 +

444 +type baseDelayBuffer struct {

445 + b [6]uint32

446 + last int

447 + min uint32

448 +}

449 +

450 +func (b *baseDelayBuffer) Push(val uint32) {

451 + t := time.Now()

452 + i := t.Second()/20 + (t.Minute()%2)*3

453 + if b.last == i {

454 + if b.b[i] > val {

455 + b.b[i] = val

456 }

457 + } else {

458 + b.b[i] = val

459 + b.last = i

460 }

229 - return min

461 + min := val

462 + for _, v := range b.b {

463 + if v > 0 && min > v {

464 + min = v

465 + }

466 + }

467 + b.min = min

468 +}

469 +

470 +func (b *baseDelayBuffer) Min() uint32 {

471 + return b.min

472 }

Godeps/_workspace/src/github.com/h2so5/utp/buffer_test.go new

+178

@@ -0,0 +1,178 @@

1 +package utp

2 +

3 +import (

4 + "bytes"

5 + "math"

6 + "testing"

7 + "time"

8 +)

9 +

10 +func TestPacketBuffer(t *testing.T) {

11 + size := 12

12 + b := newPacketBuffer(12, 1)

13 +

14 + if b.space() != size {

15 + t.Errorf("expected space == %d; got %d", size, b.space())

16 + }

17 +

18 + for i := 1; i <= size; i++ {

19 + b.push(&packet{header: header{seq: uint16(i)}})

20 + }

21 +

22 + if b.space() != 0 {

23 + t.Errorf("expected space == 0; got %d", b.space())

24 + }

25 +

26 + a := []byte{255, 7}

27 + ack := b.generateSelectiveACK()

28 + if !bytes.Equal(a, ack) {

29 + t.Errorf("expected ack == %v; got %v", a, ack)

30 + }

31 +

32 + err := b.push(&packet{header: header{seq: 15}})

33 + if err == nil {

34 + t.Fatal("push should fail")

35 + }

36 +

37 + all := b.all()

38 + if len(all) != size {

39 + t.Errorf("expected %d packets sequence; got %d", size, len(all))

40 + }

41 +

42 + f := b.fetch(6)

43 + if f == nil {

44 + t.Fatal("fetch should not fail")

45 + }

46 +

47 + b.compact()

48 +

49 + err = b.push(&packet{header: header{seq: 15}})

50 + if err != nil {

51 + t.Fatal(err)

52 + }

53 +

54 + err = b.push(&packet{header: header{seq: 17}})

55 + if err != nil {

56 + t.Fatal(err)

57 + }

58 +

59 + for i := 7; i <= size; i++ {

60 + f := b.fetch(uint16(i))

61 + if f == nil {

62 + t.Fatal("fetch should not fail")

63 + }

64 + }

65 +

66 + a = []byte{128, 2}

67 + ack = b.generateSelectiveACK()

68 + if !bytes.Equal(a, ack) {

69 + t.Errorf("expected ack == %v; got %v", a, ack)

70 + }

71 +

72 + all = b.all()

73 + if len(all) != 2 {

74 + t.Errorf("expected 2 packets sequence; got %d", len(all))

75 + }

76 +

77 + b.compact()

78 + if b.space() != 9 {

79 + t.Errorf("expected space == 9; got %d", b.space())

80 + }

81 +

82 + ack = b.generateSelectiveACK()

83 + b.processSelectiveACK(ack)

84 +

85 + all = b.all()

86 + if len(all) != 1 {

87 + t.Errorf("expected size == 1; got %d", len(all))

88 + }

89 +}

90 +

91 +func TestPacketBufferBoundary(t *testing.T) {

92 + begin := math.MaxUint16 - 3

93 + b := newPacketBuffer(12, begin)

94 + for i := begin; i != 5; i = (i + 1) % (math.MaxUint16 + 1) {

95 + err := b.push(&packet{header: header{seq: uint16(i)}})

96 + if err != nil {

97 + t.Fatal(err)

98 + }

99 + }

100 +}

101 +

102 +func TestPacketRingBuffer(t *testing.T) {

103 + b := newPacketRingBuffer(5)

104 + for i := 0; i < 7; i++ {

105 + b.push(&packet{header: header{seq: uint16(i)}})

106 + }

107 +

108 + if b.size() != 5 {

109 + t.Errorf("expected size == 5; got %d", b.size())

110 + }

111 +

112 + p := b.pop()

113 + if p.header.seq != 2 {

114 + t.Errorf("expected header.seq == 2; got %d", p.header.seq)

115 + }

116 +

117 + if b.size() != 4 {

118 + t.Errorf("expected size == 4; got %d", b.size())

119 + }

120 +

121 + for b.pop() != nil {

122 + }

123 +

124 + if !b.empty() {

125 + t.Errorf("buffer must be empty")

126 + }

127 +

128 + go func() {

129 + for i := 0; i < 5; i++ {

130 + b.push(&packet{header: header{seq: uint16(i)}})

131 + }

132 + }()

133 +

134 + p, err := b.popOne(time.Second)

135 + if err != nil {

136 + t.Fatal(err)

137 + }

138 +

139 + if p.header.seq != 0 {

140 + t.Errorf("expected header.seq == 0; got %d", p.header.seq)

141 + }

142 +}

143 +

144 +func TestByteRingBuffer(t *testing.T) {

145 +

146 + b := newByteRingBuffer(5)

147 + for i := 0; i < 100; i++ {

148 + b.Write([]byte{byte(i)})

149 + }

150 +

151 + var buf [10]byte

152 + l, err := b.ReadTimeout(buf[:], 0)

153 + if err != nil {

154 + t.Fatal(err)

155 + }

156 +

157 + e := []byte{95, 96, 97, 98, 99}

158 + if !bytes.Equal(buf[:l], e) {

159 + t.Errorf("expected payload of %v; got %v", e, buf[:l])

160 + }

161 +

162 + e2 := []byte("abcdefghijklmnopqrstuvwxyz")

163 + go func() {

164 + _, err := b.Write(e2)

165 + if err != nil {

166 + t.Fatal(err)

167 + }

168 + }()

169 +

170 + l, err = b.ReadTimeout(buf[:], 0)

171 + if err != nil {

172 + t.Fatal(err)

173 + }

174 +

175 + if !bytes.Equal(buf[:l], e2[len(e2)-5:]) {

176 + t.Errorf("expected payload of %v; got %v", e2[len(e2)-5:], buf[:l])

177 + }

178 +}

Godeps/_workspace/src/github.com/h2so5/utp/conn.go

+329 -519

@@ -1,48 +1,55 @@

1 package utp

3 import (

4 - "bytes"

5 - "errors"

6 - "io"

4 "math"

8 - "math/rand"

5 "net"

6 + "sync"

7 + "sync/atomic"

8 "syscall"

9 "time"

10 )

14 -type UTPConn struct {

15 - conn net.PacketConn

16 - raddr net.Addr

17 - rid, sid, seq, ack, lastAck uint16

18 - rtt, rttVar, minRtt, rto, dupAck int64

19 - diff, maxWindow uint32

20 - rdeadline, wdeadline time.Time

21 -

22 - state state

23 - lastTimedOut time.Time

24 -

25 - outch chan *outgoingPacket

26 - outchch chan int

27 - sendch chan *outgoingPacket

28 - sendchch chan int

29 - recvch chan *packet

30 - recvchch chan int

31 - readch chan []byte

32 - readchch chan int

33 - winch chan uint32

34 - quitch chan int

35 - activech chan int

36 - connch chan error

37 - finch chan int

38 - closech chan<- uint16

39 - eofid uint16

40 - keepalivech chan time.Duration

12 +// Conn is an implementation of the Conn interface for UTP network

13 +// connections.

14 +type Conn struct {

15 + conn *baseConn

16 + raddr net.Addr

17 + rid, sid, seq, ack, lastAck uint16

18 + rtt, rttVar, minRtt, rto int64

19 + dupAck int

20 + diff, maxWindow uint32

21 +

22 + state int

23 + closed int32

42 - readbuf bytes.Buffer

25 recvbuf *packetBuffer

26 sendbuf *packetBuffer

28 + readbuf *byteRingBuffer

29 + writebuf *rateLimitedBuffer

30 +

31 + baseDelay baseDelayBuffer

32 +

33 + writech chan []byte

34 + ackch chan int

35 + synch chan int

36 +

37 + rdeadline time.Time

38 + wdeadline time.Time

39 + deadlineMutex sync.RWMutex

40 +

41 + recv chan *packet

42 +

43 + closing bool

44 + closingch chan int

45 +

46 + keepalivech chan time.Duration

47 +

48 + connch chan int

49 +

50 + closech chan int

51 + closechMutex sync.Mutex

52 +

53 stat statistics

54 }

@@ -54,351 +61,235 @@ type statistics struct {

61 packetTimedOuts int

62 sentSelectiveACKs int

63 receivedSelectiveACKs int

57 -

58 - rtoSum int

59 - rtoCount int

64 + rtoSum int64

65 + rtoCount int

66 }

62 -func dial(n string, laddr, raddr *UTPAddr, timeout time.Duration) (*UTPConn, error) {

63 - udpnet, err := utp2udp(n)

64 - if err != nil {

65 - return nil, err

66 - }

67 -

68 - // TODO extract

69 - if laddr == nil {

70 - addr, err := net.ResolveUDPAddr(udpnet, ":0")

71 - if err != nil {

72 - return nil, err

73 - }

74 - laddr = &UTPAddr{Addr: addr}

75 - }

76 -

77 - conn, err := net.ListenPacket(udpnet, laddr.Addr.String())

78 - if err != nil {

79 - return nil, err

80 - }

81 -

82 - id := uint16(rand.Intn(math.MaxUint16))

83 -

84 - c := newUTPConn()

85 - c.conn = conn

86 - c.raddr = raddr.Addr

87 - c.rid = id

88 - c.sid = id + 1

89 - c.seq = 1

90 - c.state = state_syn_sent

91 - c.sendbuf = newPacketBuffer(window_size, 1)

92 -

93 - go c.recv()

94 - go c.loop()

95 -

96 - select {

97 - case c.sendch <- &outgoingPacket{st_syn, nil, nil}:

98 - case <-c.sendchch:

99 - return nil, errors.New("use of closed network connection")

100 - }

68 +func newConn() *Conn {

69 + wch := make(chan []byte)

70 + c := &Conn{

71 + minRtt: math.MaxInt64,

72 + maxWindow: mss,

73 + rto: int64(60),

102 - var t <-chan time.Time

103 - if timeout != 0 {

104 - t = time.After(timeout)

105 - }

75 + recv: make(chan *packet),

76 + connch: make(chan int),

107 - select {

108 - case err := <-c.connch:

109 - if err != nil {

110 - c.closed()

111 - return nil, err

112 - }

113 - ulog.Printf(1, "Conn(%v): Connected", c.LocalAddr())

114 - return c, nil

115 - case <-t:

116 - c.quitch <- 0

117 - return nil, &timeoutError{}

118 - }

119 -}

78 + recvbuf: newPacketBuffer(0, 0),

121 -func newUTPConn() *UTPConn {

122 - rto := 60

80 + readbuf: newByteRingBuffer(readBufferSize),

81 + writebuf: newRateLimitedBuffer(wch, mss),

124 - return &UTPConn{

125 - minRtt: math.MaxInt64,

126 - maxWindow: mtu,

127 - rto: int64(rto),

128 -

129 - outch: make(chan *outgoingPacket, 1),

130 - outchch: make(chan int),

131 - sendch: make(chan *outgoingPacket, 1),

132 - sendchch: make(chan int),

133 - recvch: make(chan *packet, 2),

134 - recvchch: make(chan int),

135 - winch: make(chan uint32, 1),

136 - quitch: make(chan int),

137 - activech: make(chan int),

138 - readch: make(chan []byte, 1),

139 - readchch: make(chan int),

140 - connch: make(chan error, 1),

141 - finch: make(chan int, 1),

83 + writech: wch,

84 + ackch: make(chan int),

85 + synch: make(chan int),

87 + closingch: make(chan int),

88 keepalivech: make(chan time.Duration),

144 -

145 - stat: statistics{

146 - rtoSum: rto,

147 - rtoCount: 1,

148 - },

89 + closech: make(chan int),

90 }

91 + return c

92 }

152 -func (c *UTPConn) ok() bool { return c != nil && c.conn != nil }

94 +func (c *Conn) ok() bool { return c != nil && c.conn != nil }

154 -func (c *UTPConn) Close() error {

96 +// Close closes the connection.

97 +func (c *Conn) Close() error {

98 if !c.ok() {

99 return syscall.EINVAL

100 }

158 -

101 + if !c.isOpen() {

102 + return nil

103 + }

104 select {

160 - case <-c.activech:

105 + case <-c.closingch:

106 default:

162 - c.quitch <- 0

163 - ulog.Printf(2, "Conn(%v): Wait for close", c.LocalAddr())

164 - <-c.finch

107 + close(c.closingch)

108 }

166 -

109 + select {

110 + case <-c.connch:

111 + default:

112 + return nil

113 + }

114 + <-c.closech

115 return nil

116 }

117

170 -func (c *UTPConn) LocalAddr() net.Addr {

171 - return &UTPAddr{Addr: c.conn.LocalAddr()}

118 +// LocalAddr returns the local network address.

119 +func (c *Conn) LocalAddr() net.Addr {

120 + if !c.ok() {

121 + return nil

122 + }

123 + return c.conn.LocalAddr()

124 }

125

174 -func (c *UTPConn) RemoteAddr() net.Addr {

175 - return &UTPAddr{Addr: c.raddr}

126 +// RemoteAddr returns the remote network address.

127 +func (c *Conn) RemoteAddr() net.Addr {

128 + if !c.ok() {

129 + return nil

130 + }

131 + return c.raddr

132 }

133

178 -func (c *UTPConn) Read(b []byte) (int, error) {

134 +// Read implements the Conn Read method.

135 +func (c *Conn) Read(b []byte) (int, error) {

136 if !c.ok() {

137 return 0, syscall.EINVAL

138 }

182 -

183 - if c.readbuf.Len() == 0 {

184 - var timeout <-chan time.Time

185 - if !c.rdeadline.IsZero() {

186 - timeout = time.After(c.rdeadline.Sub(time.Now()))

139 + if !c.isOpen() {

140 + return 0, &net.OpError{

141 + Op: "read",

142 + Net: c.LocalAddr().Network(),

143 + Addr: c.LocalAddr(),

144 + Err: errClosing,

145 }

188 -

146 + }

147 + s := c.readbuf.space()

148 + c.deadlineMutex.RLock()

149 + d := timeToDeadline(c.rdeadline)

150 + c.deadlineMutex.RUnlock()

151 + l, err := c.readbuf.ReadTimeout(b, d)

152 + if s < mss && c.readbuf.space() > 0 {

153 select {

190 - case b := <-c.readch:

191 - if b == nil {

192 - return 0, io.EOF

193 - }

194 - _, err := c.readbuf.Write(b)

195 - if err != nil {

196 - return 0, err

197 - }

198 - case <-c.readchch:

199 - loop:

200 - for {

201 - select {

202 - case b := <-c.readch:

203 - _, err := c.readbuf.Write(b)

204 - if err != nil {

205 - return 0, err

206 - }

207 - default:

208 - break loop

209 - }

210 - }

211 - if c.readbuf.Len() == 0 {

212 - return 0, io.EOF

213 - }

214 - case <-timeout:

215 - return 0, &timeoutError{}

154 + case c.ackch <- 0:

155 + default:

156 }

157 }

218 - return c.readbuf.Read(b)

158 + return l, err

159 }

160

221 -func (c *UTPConn) Write(b []byte) (int, error) {

161 +func timeToDeadline(deadline time.Time) (d time.Duration) {

162 + if deadline.IsZero() {

163 + return

164 + }

165 + d = deadline.Sub(time.Now())

166 + if d < 0 {

167 + d = 0

168 + }

169 + return

170 +}

171 +

172 +// Write implements the Conn Write method.

173 +func (c *Conn) Write(b []byte) (int, error) {

174 if !c.ok() {

175 return 0, syscall.EINVAL

176 }

225 -

226 - var wrote uint64

227 - for {

228 - l := uint64(len(b)) - wrote

229 - if l > mss {

230 - l = mss

231 - }

232 - select {

233 - case c.outch <- &outgoingPacket{st_data, nil, b[wrote : wrote+l]}:

234 - case <-c.outchch:

235 - return 0, errors.New("use of closed network connection")

236 - }

237 -

238 - wrote += l

239 - ulog.Printf(4, "Conn(%v): Write %d/%d bytes", c.LocalAddr(), wrote, len(b))

240 - if l < mss {

241 - break

177 + if !c.isOpen() {

178 + return 0, &net.OpError{

179 + Op: "write",

180 + Net: c.LocalAddr().Network(),

181 + Addr: c.LocalAddr(),

182 + Err: errClosing,

183 }

184 }

244 -

245 - return len(b), nil

185 + c.deadlineMutex.RLock()

186 + d := timeToDeadline(c.wdeadline)

187 + c.deadlineMutex.RUnlock()

188 + return c.writebuf.WriteTimeout(b, d)

189 }

190

248 -func (c *UTPConn) SetDeadline(t time.Time) error {

191 +// SetDeadline implements the Conn SetDeadline method.

192 +func (c *Conn) SetDeadline(t time.Time) error {

193 if !c.ok() {

194 return syscall.EINVAL

195 }

252 - if err := c.SetReadDeadline(t); err != nil {

253 - return err

254 - }

255 - if err := c.SetWriteDeadline(t); err != nil {

196 + err := c.SetReadDeadline(t)

197 + if err != nil {

198 return err

199 }

258 - return nil

200 + return c.SetWriteDeadline(t)

201 }

202

261 -func (c *UTPConn) SetReadDeadline(t time.Time) error {

203 +// SetReadDeadline implements the Conn SetReadDeadline method.

204 +func (c *Conn) SetReadDeadline(t time.Time) error {

205 if !c.ok() {

206 return syscall.EINVAL

207 }

208 + c.deadlineMutex.Lock()

209 + defer c.deadlineMutex.Unlock()

210 c.rdeadline = t

211 return nil

212 }

213

269 -func (c *UTPConn) SetWriteDeadline(t time.Time) error {

214 +// SetWriteDeadline implements the Conn SetWriteDeadline method.

215 +func (c *Conn) SetWriteDeadline(t time.Time) error {

216 if !c.ok() {

217 return syscall.EINVAL

218 }

219 + c.deadlineMutex.Lock()

220 + defer c.deadlineMutex.Unlock()

221 c.wdeadline = t

222 return nil

223 }

224

277 -func (c *UTPConn) SetKeepAlive(d time.Duration) error {

225 +// SetKeepAlive sets the keepalive interval associated with the connection.

226 +func (c *Conn) SetKeepAlive(d time.Duration) error {

227 if !c.ok() {

228 return syscall.EINVAL

229 }

281 - select {

282 - case <-c.activech:

283 - default:

284 - c.keepalivech <- d

230 + if !c.isOpen() {

231 + return errClosing

232 }

233 + c.keepalivech <- d

234 return nil

235 }

236

289 -func readPacket(data []byte) (*packet, error) {

290 - p := globalPool.get()

291 - err := p.UnmarshalBinary(data)

292 - if err != nil {

293 - return nil, err

294 - }

295 - if p.header.ver != version {

296 - return nil, errors.New("unsupported header version")

297 - }

298 - return p, nil

299 -}

300 -

301 -func (c *UTPConn) recv() {

302 - for {

303 - var buf [mtu]byte

304 - len, addr, err := c.conn.ReadFrom(buf[:])

305 - if err != nil {

306 - return

307 - }

308 - if addr.String() != c.raddr.String() {

309 - continue

310 - }

311 - p, err := readPacket(buf[:len])

312 - if err == nil {

313 - select {

314 - case c.recvch <- p:

315 - case <-c.recvchch:

316 - return

317 - }

318 - }

319 - }

320 -}

237 +func (c *Conn) loop() {

238 + defer c.conn.Unregister(int32(c.rid))

239

322 -func (c *UTPConn) loop() {

323 - var recvExit, sendExit bool

324 - var lastReceived time.Time

240 + var resendSeq uint16

241 + var resendCont int

242 var keepalive <-chan time.Time

243

327 - go func() {

328 - var window uint32 = window_size * mtu

329 - for {

330 - if window >= mtu {

331 - select {

332 - case b := <-c.outch:

333 - select {

334 - case c.sendch <- b:

335 - window -= mtu

336 - case <-c.sendchch:

337 - return

338 - }

339 - case <-c.outchch:

340 - return

341 - case w := <-c.winch:

342 - window = w

343 - }

344 - } else {

345 - window = <-c.winch

346 - }

347 - }

348 - }()

244 + resend := time.NewTimer(0)

245 + resend.Stop()

246 + defer resend.Stop()

247

248 for {

351 - select {

352 - case <-c.sendchch:

353 - sendExit = true

354 - default:

249 + resend.Stop()

250 + f := c.sendbuf.front()

251 + if f != nil {

252 + resend.Reset(time.Duration(c.rto) * time.Millisecond)

253 }

254 select {

357 - case <-c.recvchch:

358 - recvExit = true

359 - default:

360 - }

361 - select {

362 - case p := <-c.recvch:

363 - ack := c.processPacket(p)

364 - lastReceived = time.Now()

365 - if ack {

366 - out := &outgoingPacket{st_state, nil, nil}

367 - selack := c.sendbuf.generateSelectiveACK()

368 - if len(selack) > 0 {

369 - out.ext = []extension{

370 - extension{

371 - typ: ext_selective_ack,

372 - payload: selack,

373 - },

374 - }

375 - c.stat.sentSelectiveACKs++

255 + case <-c.ackch:

256 + c.sendACK()

257 + case <-c.synch:

258 + c.sendSYN()

259 + case p := <-c.recv:

260 + c.stat.receivedPackets++

261 + c.processPacket(p)

262 + case b := <-c.writech:

263 + c.sendDATA(b)

264 + case <-c.closingch:

265 + c.enterClosing()

266 + case <-resend.C:

267 + if f != nil {

268 + if resendSeq == f.header.seq {

269 + resendCont++

270 + } else {

271 + resendCont = 0

272 + resendSeq = f.header.seq

273 }

377 - c.sendPacket(out)

378 - }

379 -

380 - case b := <-c.sendch:

381 - c.sendPacket(b)

382 -

383 - case <-time.After(time.Duration(c.rto) * time.Millisecond):

384 - if !c.state.active && time.Now().Sub(lastReceived) > reset_timeout {

385 - ulog.Printf(2, "Conn(%v): Connection timed out", c.LocalAddr())

386 - c.sendPacket(&outgoingPacket{st_reset, nil, nil})

387 - c.close()

388 - } else {

389 - t, err := c.sendbuf.frontPushedTime()

390 - if err == nil && c.lastTimedOut != t && time.Now().Sub(t) > time.Duration(c.rto)*time.Millisecond {

391 - c.lastTimedOut = t

392 - c.stat.packetTimedOuts++

274 + c.stat.packetTimedOuts++

275 + if resendCont > maxRetry {

276 + c.sendRST()

277 + c.close()

278 + } else {

279 c.maxWindow /= 2

280 if c.maxWindow < mtu {

281 c.maxWindow = mtu

282 }

283 for _, p := range c.sendbuf.sequence() {

398 - c.resendPacket(p)

284 + c.resend(p)

285 }

286 }

287 }

288 + case <-c.closech:

289 + c.readbuf.Close()

290 + c.state = stateClosed

291 + atomic.StoreInt32(&c.closed, 1)

292 + return

293 case d := <-c.keepalivech:

294 if d <= 0 {

295 keepalive = nil

@@ -407,56 +298,62 @@ func (c *UTPConn) loop() {

298 }

299 case <-keepalive:

300 ulog.Printf(2, "Conn(%v): Send keepalive", c.LocalAddr())

410 - c.sendPacket(&outgoingPacket{st_state, nil, nil})

411 -

412 - case <-c.quitch:

413 - if c.state.exit != nil {

414 - c.state.exit(c)

415 - }

301 + c.sendACK()

302 }

417 - if recvExit && sendExit {

418 - return

303 + if c.closing {

304 + c.tryFIN()

305 + if c.state == stateSynSent || c.state == stateFinSent || (c.recvbuf.empty() && c.sendbuf.empty()) {

306 + c.close()

307 + }

308 }

309 }

310 }

311

423 -func (c *UTPConn) sendPacket(b *outgoingPacket) {

424 - p := c.makePacket(b)

425 - bin, err := p.MarshalBinary()

426 - if err == nil {

427 - ulog.Printf(3, "SEND %v -> %v: %v", c.conn.LocalAddr(), c.raddr, p.String())

428 - c.stat.sentPackets++

429 - _, err = c.conn.WriteTo(bin, c.raddr)

430 - if err != nil {

431 - return

432 - }

433 - if b.typ != st_state {

434 - c.sendbuf.push(p)

435 - } else {

436 - globalPool.put(p)

312 +func (c *Conn) tryFIN() {

313 + if c.state != stateFinSent {

314 + if c.sendFIN() == nil {

315 + c.writebuf.Close()

316 + c.state = stateFinSent

317 }

318 }

319 }

320

441 -func (c *UTPConn) resendPacket(p *packet) {

442 - bin, err := p.MarshalBinary()

443 - if err == nil {

444 - ulog.Printf(3, "RESEND %v -> %v: %v", c.conn.LocalAddr(), c.raddr, p.String())

445 - c.stat.resentPackets++

446 - _, err = c.conn.WriteTo(bin, c.raddr)

447 - if err != nil {

448 - return

449 - }

321 +func (c *Conn) enterClosing() {

322 + if !c.closing {

323 + c.closing = true

324 }

325 }

326

327 +func (c *Conn) close() {

328 + c.closechMutex.Lock()

329 + defer c.closechMutex.Unlock()

330 + select {

331 + case <-c.closech:

332 + default:

333 + close(c.closech)

334 + }

335 + ulog.Printf(1, "Conn(%v): closed", c.LocalAddr())

336 + ulog.Printf(1, "Conn(%v): * SentPackets: %d", c.LocalAddr(), c.stat.sentPackets)

337 + ulog.Printf(1, "Conn(%v): * ResentPackets: %d", c.LocalAddr(), c.stat.resentPackets)

338 + ulog.Printf(1, "Conn(%v): * ReceivedPackets: %d", c.LocalAddr(), c.stat.receivedPackets)

339 + ulog.Printf(1, "Conn(%v): * ReceivedDuplicatedACKs: %d", c.LocalAddr(), c.stat.receivedDuplicatedACKs)

340 + ulog.Printf(1, "Conn(%v): * PacketTimedOuts: %d", c.LocalAddr(), c.stat.packetTimedOuts)

341 + ulog.Printf(1, "Conn(%v): * SentSelectiveACKs: %d", c.LocalAddr(), c.stat.sentSelectiveACKs)

342 + ulog.Printf(1, "Conn(%v): * ReceivedSelectiveACKs: %d", c.LocalAddr(), c.stat.receivedSelectiveACKs)

343 + if c.stat.rtoCount > 0 {

344 + ulog.Printf(1, "Conn(%v): * AverageRTO: %d", c.LocalAddr(), c.stat.rtoSum/int64(c.stat.rtoCount))

345 + }

346 +}

347 +

348 +func (c *Conn) isOpen() bool {

349 + return atomic.LoadInt32(&c.closed) == 0

350 +}

351 +

352 func currentMicrosecond() uint32 {

353 return uint32(time.Now().Nanosecond() / 1000)

354 }

355

457 -func (c *UTPConn) processPacket(p *packet) bool {

458 - var ack bool

459 -

356 +func (c *Conn) processPacket(p *packet) {

357 if p.header.t == 0 {

358 c.diff = 0

359 } else {

@@ -469,15 +366,14 @@ func (c *UTPConn) processPacket(p *packet) bool {

366 }

367 }

368

472 - ulog.Printf(3, "RECV %v -> %v: %v", c.raddr, c.conn.LocalAddr(), p.String())

473 - c.stat.receivedPackets++

369 + c.baseDelay.Push(c.diff)

370

475 - if p.header.typ == st_state {

476 -

477 - f := c.sendbuf.first()

371 + switch p.header.typ {

372 + case stState:

373 + f := c.sendbuf.front()

374 if f != nil && p.header.ack == f.header.seq {

375 for _, e := range p.ext {

480 - if e.typ == ext_selective_ack {

376 + if e.typ == extSelectiveAck {

377 ulog.Printf(3, "Conn(%v): Receive Selective ACK", c.LocalAddr())

378 c.stat.receivedSelectiveACKs++

379 c.sendbuf.processSelectiveACK(e.payload)

@@ -507,12 +403,12 @@ func (c *UTPConn) processPacket(p *packet) bool {

403 } else if c.rto > 1000 {

404 c.rto = 1000

405 }

510 - c.stat.rtoSum += int(c.rto)

406 + c.stat.rtoSum += c.rto

407 c.stat.rtoCount++

408 }

409

514 - if c.diff != 0 {

515 - ourDelay := float64(c.diff)

410 + ourDelay := float64(c.diff - c.baseDelay.Min())

411 + if ourDelay != 0.0 {

412 offTarget := 100000.0 - ourDelay

413 windowFactor := float64(mtu) / float64(c.maxWindow)

414 delayFactor := offTarget / 100000.0

@@ -523,85 +419,142 @@ func (c *UTPConn) processPacket(p *packet) bool {

419 }

420 ulog.Printf(4, "Conn(%v): Update maxWindow: %d", c.LocalAddr(), c.maxWindow)

421 }

526 - globalPool.put(s)

422 }

423 +

424 c.sendbuf.compact()

425 +

426 if c.lastAck == p.header.ack {

427 c.dupAck++

428 if c.dupAck >= 2 {

532 - ulog.Printf(3, "Conn(%v): Receive 3 duplicated acks: %d", c.LocalAddr(), p.header.ack)

429 c.stat.receivedDuplicatedACKs++

534 - p := c.sendbuf.first()

430 + ulog.Printf(3, "Conn(%v): Receive 3 duplicated acks: %d", c.LocalAddr(), p.header.ack)

431 + p := c.sendbuf.front()

432 if p != nil {

433 c.maxWindow /= 2

434 if c.maxWindow < mtu {

435 c.maxWindow = mtu

436 }

437 ulog.Printf(4, "Conn(%v): Update maxWindow: %d", c.LocalAddr(), c.maxWindow)

541 - c.resendPacket(p)

438 + c.resend(p)

439 }

440 c.dupAck = 0

441 }

442 } else {

443 c.dupAck = 0

444 }

445 +

446 c.lastAck = p.header.ack

447 if p.header.ack == c.seq-1 {

448 wnd := p.header.wnd

449 if wnd > c.maxWindow {

450 wnd = c.maxWindow

451 }

554 - ulog.Printf(4, "Conn(%v): Reset window: %d", c.LocalAddr(), wnd)

555 - go func() {

556 - c.winch <- wnd

557 - }()

452 + c.writebuf.Reset(wnd)

453 }

559 - if c.state.state != nil {

560 - c.state.state(c, p)

454 +

455 + if c.state == stateSynSent {

456 + c.recvbuf = newPacketBuffer(windowSize, int(p.header.seq))

457 + c.state = stateConnected

458 + close(c.connch)

459 }

562 - globalPool.put(p)

563 - } else if p.header.typ == st_reset {

564 - globalPool.put(p)

460 +

461 + case stReset:

462 + c.sendRST()

463 c.close()

566 - } else {

567 - if c.recvbuf == nil {

568 - return false

569 - }

570 - ack = true

464 +

465 + default:

466 c.recvbuf.push(p)

467 for _, s := range c.recvbuf.fetchSequence() {

468 c.ack = s.header.seq

574 - switch s.header.typ {

575 - case st_data:

576 - if c.state.data != nil {

577 - c.state.data(c, s)

578 - }

579 - case st_fin:

580 - if c.state.fin != nil {

581 - c.state.fin(c, s)

582 - }

583 - case st_state:

584 - if c.state.state != nil {

585 - c.state.state(c, s)

586 - }

469 + if s.header.typ == stData {

470 + c.readbuf.Write(s.payload)

471 + } else if s.header.typ == stFin {

472 + c.enterClosing()

473 }

588 - globalPool.put(s)

474 + }

475 + c.sendACK()

476 + }

477 +}

478 +

479 +func (c *Conn) sendACK() {

480 + ack := c.makePacket(stState, nil, c.raddr)

481 + selack := c.sendbuf.generateSelectiveACK()

482 + if selack != nil {

483 + c.stat.sentSelectiveACKs++

484 + ack.ext = []extension{

485 + extension{

486 + typ: extSelectiveAck,

487 + payload: selack,

488 + },

489 }

490 }

591 - return ack

491 + c.stat.sentPackets++

492 + c.conn.Send(ack)

493 }

494

594 -func (c *UTPConn) makePacket(b *outgoingPacket) *packet {

595 - wnd := window_size * mtu

495 +func (c *Conn) sendSYN() {

496 + syn := c.makePacket(stSyn, nil, c.raddr)

497 + err := c.sendbuf.push(syn)

498 + if err != nil {

499 + ulog.Printf(2, "Conn(%v): buffer error: %v", c.LocalAddr(), err)

500 + return

501 + }

502 + c.stat.sentPackets++

503 + c.conn.Send(syn)

504 +}

505 +

506 +func (c *Conn) sendFIN() error {

507 + fin := c.makePacket(stFin, nil, c.raddr)

508 + err := c.sendbuf.push(fin)

509 + if err != nil {

510 + ulog.Printf(2, "Conn(%v): buffer error: %v", c.LocalAddr(), err)

511 + return err

512 + }

513 + c.stat.sentPackets++

514 + c.conn.Send(fin)

515 + return nil

516 +}

517 +

518 +func (c *Conn) sendRST() {

519 + rst := c.makePacket(stReset, nil, c.raddr)

520 + c.stat.sentPackets++

521 + c.conn.Send(rst)

522 +}

523 +

524 +func (c *Conn) sendDATA(b []byte) {

525 + for i := 0; i <= len(b)/mss; i++ {

526 + l := len(b) - i*mss

527 + if l > mss {

528 + l = mss

529 + }

530 + data := c.makePacket(stData, b[i*mss:i*mss+l], c.raddr)

531 + c.sendbuf.push(data)

532 + c.stat.sentPackets++

533 + c.conn.Send(data)

534 + }

535 +}

536 +

537 +func (c *Conn) resend(p *packet) {

538 + c.stat.resentPackets++

539 + c.conn.Send(p)

540 + ulog.Printf(3, "Conn(%v): RESEND: %s", c.LocalAddr(), p.String())

541 +}

542 +

543 +func (c *Conn) makePacket(typ int, payload []byte, dst net.Addr) *packet {

544 + wnd := windowSize * mtu

545 if c.recvbuf != nil {

546 wnd = c.recvbuf.space() * mtu

547 }

548 + s := c.readbuf.space()

549 + if wnd > s {

550 + wnd = s

551 + }

552 id := c.sid

600 - if b.typ == st_syn {

553 + if typ == stSyn {

554 id = c.rid

555 }

603 - p := globalPool.get()

604 - p.header.typ = b.typ

556 + p := &packet{}

557 + p.header.typ = typ

558 p.header.ver = version

559 p.header.id = id

560 p.header.t = currentMicrosecond()

@@ -609,153 +562,10 @@ func (c *UTPConn) makePacket(b *outgoingPacket) *packet {

562 p.header.wnd = uint32(wnd)

563 p.header.seq = c.seq

564 p.header.ack = c.ack

612 - if b.typ == st_fin {

613 - c.eofid = c.seq

614 - }

615 - if !(b.typ == st_state && len(b.payload) == 0) {

565 + p.addr = dst

566 + if typ != stState && typ != stFin {

567 c.seq++

568 }

618 - p.payload = p.payload[:len(b.payload)]

619 - copy(p.payload, b.payload)

569 + p.payload = payload

570 return p

571 }

622 -

623 -func (c *UTPConn) close() {

624 - if !c.state.closed {

625 - close(c.outchch)

626 - close(c.readchch)

627 - close(c.sendchch)

628 - close(c.recvchch)

629 - close(c.activech)

630 - close(c.finch)

631 - c.closed()

632 -

633 - // Accepted connection

634 - if c.closech != nil {

635 - c.closech <- c.sid

636 - } else {

637 - c.conn.Close()

638 - }

639 -

640 - ulog.Printf(1, "Conn(%v): Closed", c.LocalAddr())

641 - ulog.Printf(1, "Conn(%v): * SentPackets: %d", c.LocalAddr(), c.stat.sentPackets)

642 - ulog.Printf(1, "Conn(%v): * ResentPackets: %d", c.LocalAddr(), c.stat.resentPackets)

643 - ulog.Printf(1, "Conn(%v): * ReceivedPackets: %d", c.LocalAddr(), c.stat.receivedPackets)

644 - ulog.Printf(1, "Conn(%v): * ReceivedDuplicatedACKs: %d", c.LocalAddr(), c.stat.receivedDuplicatedACKs)

645 - ulog.Printf(1, "Conn(%v): * PacketTimedOuts: %d", c.LocalAddr(), c.stat.packetTimedOuts)

646 - ulog.Printf(1, "Conn(%v): * SentSelectiveACKs: %d", c.LocalAddr(), c.stat.sentSelectiveACKs)

647 - ulog.Printf(1, "Conn(%v): * ReceivedSelectiveACKs: %d", c.LocalAddr(), c.stat.receivedSelectiveACKs)

648 - ulog.Printf(1, "Conn(%v): * AverageRTO: %d", c.LocalAddr(), c.stat.rtoSum/c.stat.rtoCount)

649 - }

650 -}

651 -

652 -func (c *UTPConn) closed() {

653 - ulog.Printf(2, "Conn(%v): Change state: CLOSED", c.LocalAddr())

654 - c.state = state_closed

655 -}

656 -

657 -func (c *UTPConn) closing() {

658 - ulog.Printf(2, "Conn(%v): Change state: CLOSING", c.LocalAddr())

659 - c.state = state_closing

660 -}

661 -

662 -func (c *UTPConn) syn_sent() {

663 - ulog.Printf(2, "Conn(%v): Change state: SYN_SENT", c.LocalAddr())

664 - c.state = state_syn_sent

665 -}

666 -

667 -func (c *UTPConn) connected() {

668 - ulog.Printf(2, "Conn(%v): Change state: CONNECTED", c.LocalAddr())

669 - c.state = state_connected

670 -}

671 -

672 -func (c *UTPConn) fin_sent() {

673 - ulog.Printf(2, "Conn(%v): Change state: FIN_SENT", c.LocalAddr())

674 - c.state = state_fin_sent

675 -}

676 -

677 -type state struct {

678 - data func(c *UTPConn, p *packet)

679 - fin func(c *UTPConn, p *packet)

680 - state func(c *UTPConn, p *packet)

681 - exit func(c *UTPConn)

682 - active bool

683 - closed bool

684 -}

685 -

686 -var state_closed state = state{

687 - closed: true,

688 -}

689 -

690 -var state_closing state = state{

691 - data: func(c *UTPConn, p *packet) {

692 - select {

693 - case c.readch <- append([]byte(nil), p.payload...):

694 - case <-c.readchch:

695 - }

696 - if c.recvbuf.empty() && c.sendbuf.empty() {

697 - c.close()

698 - }

699 - },

700 - state: func(c *UTPConn, p *packet) {

701 - if c.recvbuf.empty() && c.sendbuf.empty() {

702 - c.close()

703 - }

704 - },

705 -}

706 -

707 -var state_syn_sent state = state{

708 - state: func(c *UTPConn, p *packet) {

709 - c.recvbuf = newPacketBuffer(window_size, int(p.header.seq))

710 - c.connected()

711 - c.connch <- nil

712 - },

713 - exit: func(c *UTPConn) {

714 - go func() {

715 - select {

716 - case c.outch <- &outgoingPacket{st_fin, nil, nil}:

717 - case <-c.outchch:

718 - }

719 - }()

720 - c.fin_sent()

721 - },

722 - active: true,

723 -}

724 -

725 -var state_connected state = state{

726 - data: func(c *UTPConn, p *packet) {

727 - select {

728 - case c.readch <- append([]byte(nil), p.payload...):

729 - case <-c.readchch:

730 - }

731 - },

732 - fin: func(c *UTPConn, p *packet) {

733 - if c.recvbuf.empty() && c.sendbuf.empty() {

734 - c.close()

735 - } else {

736 - c.closing()

737 - }

738 - },

739 - exit: func(c *UTPConn) {

740 - go func() {

741 - select {

742 - case c.outch <- &outgoingPacket{st_fin, nil, nil}:

743 - case <-c.outchch:

744 - }

745 - }()

746 - c.fin_sent()

747 - },

748 - active: true,

749 -}

750 -

751 -var state_fin_sent state = state{

752 - state: func(c *UTPConn, p *packet) {

753 - if p.header.ack == c.eofid {

754 - if c.recvbuf.empty() && c.sendbuf.empty() {

755 - c.close()

756 - } else {

757 - c.closing()

758 - }

759 - }

760 - },

761 -}

Godeps/_workspace/src/github.com/h2so5/utp/conn_test.go new

+87

@@ -0,0 +1,87 @@

1 +package utp

2 +

3 +import (

4 + "bytes"

5 + "testing"

6 +)

7 +

8 +func TestReadWrite(t *testing.T) {

9 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

10 + if err != nil {

11 + t.Fatal(err)

12 + }

13 +

14 + l, err := Listen("utp", addr)

15 + if err != nil {

16 + t.Fatal(err)

17 + }

18 + defer l.Close()

19 +

20 + payload := []byte("abcdefgh")

21 +

22 + ch := make(chan int)

23 + go func() {

24 + c, err := l.Accept()

25 + if err != nil {

26 + t.Fatal(err)

27 + }

28 + defer c.Close()

29 +

30 + var buf [256]byte

31 + length, err := c.Read(buf[:])

32 + if err != nil {

33 + t.Fatal(err)

34 + }

35 + if !bytes.Equal(payload, buf[:length]) {

36 + t.Errorf("expected payload of %v; got %v", payload, buf[:length])

37 + }

38 +

39 + ch <- 0

40 + }()

41 +

42 + c, err := DialUTP("utp", nil, l.Addr().(*Addr))

43 + if err != nil {

44 + t.Fatal(err)

45 + }

46 + defer c.Close()

47 +

48 + _, err = c.Write(payload)

49 + if err != nil {

50 + t.Fatal(err)

51 + }

52 +

53 + <-ch

54 +}

55 +

56 +func TestClose(t *testing.T) {

57 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

58 + if err != nil {

59 + t.Fatal(err)

60 + }

61 +

62 + l, err := Listen("utp", addr)

63 + if err != nil {

64 + t.Fatal(err)

65 + }

66 + defer l.Close()

67 +

68 + go func() {

69 + c, err := l.Accept()

70 + if err != nil {

71 + t.Fatal(err)

72 + }

73 + c.Close()

74 + }()

75 +

76 + c, err := DialUTP("utp", nil, l.Addr().(*Addr))

77 + if err != nil {

78 + t.Fatal(err)

79 + }

80 + defer c.Close()

81 +

82 + var b [128]byte

83 + _, err = c.Read(b[:])

84 + if err == nil {

85 + t.Fatal("Read should fail")

86 + }

87 +}

Godeps/_workspace/src/github.com/h2so5/utp/dial.go

+48 -14

@@ -2,24 +2,58 @@ package utp

3 import (

4 "errors"

5 + "math"

6 + "math/rand"

7 "net"

8 "time"

9 )

9 -func Dial(n, addr string) (*UTPConn, error) {

10 - raddr, err := ResolveUTPAddr(n, addr)

11 +// DialUTP connects to the remote address raddr on the network net,

12 +// which must be "utp", "utp4", or "utp6". If laddr is not nil, it is

13 +// used as the local address for the connection.

14 +func DialUTP(n string, laddr, raddr *Addr) (*Conn, error) {

15 + return DialUTPTimeout(n, laddr, raddr, 0)

16 +}

17 +

18 +// DialUTPTimeout acts like Dial but takes a timeout.

19 +// The timeout includes name resolution, if required.

20 +func DialUTPTimeout(n string, laddr, raddr *Addr, timeout time.Duration) (*Conn, error) {

21 + conn, err := getSharedBaseConn(n, laddr)

22 if err != nil {

23 return nil, err

24 }

14 - return DialUTP(n, nil, raddr)

15 -}

17 -func DialUTP(n string, laddr, raddr *UTPAddr) (*UTPConn, error) {

18 - return dial(n, laddr, raddr, 0)

19 -}

26 + id := uint16(rand.Intn(math.MaxUint16))

27 + c := newConn()

28 + c.conn = conn

29 + c.raddr = raddr.Addr

30 + c.rid = id

31 + c.sid = id + 1

32 + c.seq = 1

33 + c.state = stateSynSent

34 + c.sendbuf = newPacketBuffer(windowSize*2, 1)

35 + c.conn.Register(int32(c.rid), c.recv)

36 + go c.loop()

37 + c.synch <- 0

21 -func DialUTPTimeout(n string, laddr, raddr *UTPAddr, timeout time.Duration) (*UTPConn, error) {

22 - return dial(n, laddr, raddr, timeout)

39 + t := time.NewTimer(timeout)

40 + defer t.Stop()

41 + if timeout == 0 {

42 + t.Stop()

43 + }

44 +

45 + select {

46 + case <-c.connch:

47 + case <-t.C:

48 + c.Close()

49 + return nil, &net.OpError{

50 + Op: "dial",

51 + Net: c.LocalAddr().Network(),

52 + Addr: c.LocalAddr(),

53 + Err: errTimeout,

54 + }

55 + }

56 + return c, nil

57 }

59 // A Dialer contains options for connecting to an address.

@@ -49,18 +83,18 @@ type Dialer struct {

83 // Dial connects to the address on the named network.

84 //

85 // See func Dial for a description of the network and address parameters.

52 -func (d *Dialer) Dial(n, addr string) (*UTPConn, error) {

53 - raddr, err := ResolveUTPAddr(n, addr)

86 +func (d *Dialer) Dial(n, addr string) (*Conn, error) {

87 + raddr, err := ResolveAddr(n, addr)

88 if err != nil {

89 return nil, err

90 }

58 - var laddr *UTPAddr

92 + var laddr *Addr

93 if d.LocalAddr != nil {

94 var ok bool

61 - laddr, ok = d.LocalAddr.(*UTPAddr)

95 + laddr, ok = d.LocalAddr.(*Addr)

96 if !ok {

63 - return nil, errors.New("Dialer.LocalAddr is not a UTPAddr")

97 + return nil, errors.New("Dialer.LocalAddr is not a Addr")

98 }

99 }

100

Godeps/_workspace/src/github.com/h2so5/utp/dial_test.go new

+52

@@ -0,0 +1,52 @@

1 +package utp

2 +

3 +import (

4 + "testing"

5 + "time"

6 +)

7 +

8 +func TestDial(t *testing.T) {

9 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

10 + if err != nil {

11 + t.Fatal(err)

12 + }

13 +

14 + l, err := Listen("utp", addr)

15 + if err != nil {

16 + t.Fatal(err)

17 + }

18 + defer l.Close()

19 +

20 + ch := make(chan struct{})

21 + go func() {

22 + l.Accept()

23 + close(ch)

24 + }()

25 +

26 + c, err := DialUTP("utp", nil, l.Addr().(*Addr))

27 + if err != nil {

28 + t.Fatal(err)

29 + }

30 + defer c.Close()

31 +

32 + <-ch

33 +}

34 +

35 +func TestDialFastTimeout(t *testing.T) {

36 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

37 + if err != nil {

38 + t.Fatal(err)

39 + }

40 +

41 + l, err := Listen("utp", addr)

42 + if err != nil {

43 + t.Fatal(err)

44 + }

45 + defer l.Close()

46 + _, err = (&Dialer{

47 + Timeout: time.Nanosecond,

48 + }).Dial("utp", l.Addr().String())

49 + if err == nil {

50 + t.Fatal("expected an error")

51 + }

52 +}

Godeps/_workspace/src/github.com/h2so5/utp/listener.go

+91 -274

@@ -1,329 +1,146 @@

1 package utp

3 import (

4 - "errors"

4 "math"

5 "math/rand"

6 "net"

7 + "sync"

8 + "sync/atomic"

9 "syscall"

10 "time"

11 )

12 -type UTPListener struct {

13 +// Listener is a UTP network listener. Clients should typically

14 +// use variables of type Listener instead of assuming UTP.

15 +type Listener struct {

16 // RawConn represents an out-of-band connection.

17 // This allows a single socket to handle multiple protocols.

18 RawConn net.PacketConn

17 - conn net.PacketConn

18 - conns map[uint16]*UTPConn

19 - accept chan (*UTPConn)

20 - err chan (error)

21 - lasterr error

22 - deadline time.Time

23 - closech chan int

24 - connch chan uint16

25 - closed bool

20 + conn *baseConn

21 + deadline time.Time

22 + deadlineMutex sync.RWMutex

23 + closed int32

24 }

28 -func Listen(n, laddr string) (*UTPListener, error) {

29 - addr, err := ResolveUTPAddr(n, laddr)

30 - if err != nil {

31 - return nil, err

32 - }

33 - return ListenUTP(n, addr)

34 -}

26 +func (l *Listener) ok() bool { return l != nil && l.conn != nil }

36 -func ListenUTP(n string, laddr *UTPAddr) (*UTPListener, error) {

37 - udpnet, err := utp2udp(n)

38 - if err != nil {

39 - return nil, err

40 - }

41 - conn, err := listenPacket(udpnet, laddr.Addr.String())

28 +// Listen announces on the UTP address laddr and returns a UTP

29 +// listener. Net must be "utp", "utp4", or "utp6". If laddr has a

30 +// port of 0, ListenUTP will choose an available port. The caller can

31 +// use the Addr method of Listener to retrieve the chosen address.

32 +func Listen(n string, laddr *Addr) (*Listener, error) {

33 + conn, err := newBaseConn(n, laddr)

34 if err != nil {

35 return nil, err

36 }

45 -

46 - l := UTPListener{

47 - RawConn: newRawConn(conn),

37 + l := &Listener{

38 + RawConn: conn,

39 conn: conn,

49 - conns: make(map[uint16]*UTPConn),

50 - accept: make(chan (*UTPConn), 10),

51 - err: make(chan (error), 1),

52 - closech: make(chan int),

53 - connch: make(chan uint16),

54 - lasterr: nil,

40 }

56 -

57 - l.listen()

58 - return &l, nil

41 + conn.Register(-1, nil)

42 + return l, nil

43 }

61 -type incoming struct {

62 - p *packet

63 - addr net.Addr

64 -}

65 -

66 -func (l *UTPListener) listen() {

67 - inch := make(chan incoming)

68 - raw := l.RawConn.(*rawConn)

69 -

70 - // reads udp packets

71 - go func() {

72 - for {

73 - var buf [mtu]byte

74 - len, addr, err := l.conn.ReadFrom(buf[:])

75 - if err != nil {

76 - l.err <- err

77 - return

78 - }

79 - p, err := readPacket(buf[:len])

80 - if err == nil {

81 - inch <- incoming{p, addr}

82 - } else {

83 - select {

84 - case <-raw.closed:

85 - default:

86 - i := rawIncoming{b: buf[:len], addr: addr}

87 - select {

88 - case raw.in <- i:

89 - default:

90 - // discard the oldest packet

91 - <-raw.in

92 - raw.in <- i

93 - }

94 - }

95 - }

96 - }

97 - }()

98 -

99 - go func() {

100 - for {

101 - select {

102 - case i := <-inch:

103 - l.processPacket(i.p, i.addr)

104 - case <-l.closech:

105 - ulog.Printf(2, "Listener(%v): Stop listening", l.conn.LocalAddr())

106 - close(l.accept)

107 - l.closed = true

108 - case id := <-l.connch:

109 - if _, ok := l.conns[id]; !ok {

110 - delete(l.conns, id+1)

111 - ulog.Printf(2, "Listener(%v): Connection closed #%d (alive: %d)", l.conn.LocalAddr(), id, len(l.conns))

112 - if l.closed && len(l.conns) == 0 {

113 - ulog.Printf(2, "Listener(%v): All accepted connections are closed", l.conn.LocalAddr())

114 - l.conn.Close()

115 - ulog.Printf(1, "Listener(%v): Closed", l.conn.LocalAddr())

116 - return

117 - }

118 - }

119 - }

120 - }

121 - }()

122 -

123 - ulog.Printf(1, "Listener(%v): Start listening", l.conn.LocalAddr())

124 -}

125 -

126 -func listenPacket(n, addr string) (net.PacketConn, error) {

127 - if n == "mem" {

128 - return nil, errors.New("TODO implement in-memory packet connection")

129 - }

130 - return net.ListenPacket(n, addr)

131 -}

132 -

133 -func (l *UTPListener) processPacket(p *packet, addr net.Addr) {

134 - switch p.header.typ {

135 - case st_data, st_fin, st_state, st_reset:

136 - if c, ok := l.conns[p.header.id]; ok {

137 - select {

138 - case c.recvch <- p:

139 - case <-c.recvchch:

140 - }

141 - }

142 - case st_syn:

143 - if l.closed {

144 - return

145 - }

146 - sid := p.header.id + 1

147 - if _, ok := l.conns[p.header.id]; !ok {

148 - seq := rand.Intn(math.MaxUint16)

149 -

150 - c := newUTPConn()

151 - c.conn = l.conn

152 - c.raddr = addr

153 - c.rid = p.header.id + 1

154 - c.sid = p.header.id

155 - c.seq = uint16(seq)

156 - c.ack = p.header.seq

157 - c.diff = currentMicrosecond() - p.header.t

158 - c.state = state_connected

159 - c.closech = l.connch

160 - c.recvbuf = newPacketBuffer(window_size, int(p.header.seq))

161 - c.sendbuf = newPacketBuffer(window_size, seq)

162 -

163 - go c.loop()

164 - select {

165 - case c.recvch <- p:

166 - case <-c.recvchch:

167 - }

168 -

169 - l.conns[sid] = c

170 - ulog.Printf(2, "Listener(%v): New incoming connection #%d from %v (alive: %d)", l.conn.LocalAddr(), sid, addr, len(l.conns))

171 -

172 - l.accept <- c

173 - }

174 - }

175 -}

176 -

177 -func (l *UTPListener) Accept() (net.Conn, error) {

45 +// Accept implements the Accept method in the Listener interface; it

46 +// waits for the next call and returns a generic Conn.

47 +func (l *Listener) Accept() (net.Conn, error) {

48 return l.AcceptUTP()

49 }

181 -func (l *UTPListener) AcceptUTP() (*UTPConn, error) {

182 - if l == nil || l.conn == nil {

51 +// AcceptUTP accepts the next incoming call and returns the new

52 +// connection.

53 +func (l *Listener) AcceptUTP() (*Conn, error) {

54 + if !l.ok() {

55 return nil, syscall.EINVAL

56 }

185 - if l.lasterr != nil {

186 - return nil, l.lasterr

187 - }

188 - var timeout <-chan time.Time

189 - if !l.deadline.IsZero() {

190 - timeout = time.After(l.deadline.Sub(time.Now()))

191 - }

192 - select {

193 - case conn := <-l.accept:

194 - if conn == nil {

195 - return nil, errors.New("use of closed network connection")

57 + if !l.isOpen() {

58 + return nil, &net.OpError{

59 + Op: "accept",

60 + Net: l.conn.LocalAddr().Network(),

61 + Addr: l.conn.LocalAddr(),

62 + Err: errClosing,

63 }

197 - return conn, nil

198 - case err := <-l.err:

199 - l.lasterr = err

200 - return nil, err

201 - case <-timeout:

202 - return nil, &timeoutError{}

203 - }

204 -}

205 -

206 -func (l *UTPListener) Addr() net.Addr {

207 - return &UTPAddr{Addr: l.conn.LocalAddr()}

208 -}

209 -

210 -func (l *UTPListener) Close() error {

211 - if l == nil || l.conn == nil {

212 - return syscall.EINVAL

64 }

214 - l.closech <- 0

215 - l.RawConn.Close()

216 - return nil

217 -}

218 -

219 -func (l *UTPListener) SetDeadline(t time.Time) error {

220 - if l == nil || l.conn == nil {

221 - return syscall.EINVAL

65 + l.deadlineMutex.RLock()

66 + d := timeToDeadline(l.deadline)

67 + l.deadlineMutex.RUnlock()

68 + p, err := l.conn.RecvSyn(d)

69 + if err != nil {

70 + return nil, &net.OpError{

71 + Op: "accept",

72 + Net: l.conn.LocalAddr().Network(),

73 + Addr: l.conn.LocalAddr(),

74 + Err: errClosing,

75 + }

76 }

223 - l.deadline = t

224 - return nil

225 -}

227 -type rawIncoming struct {

228 - b []byte

229 - addr net.Addr

230 -}

78 + seq := rand.Intn(math.MaxUint16)

79 + rid := p.header.id + 1

232 -type rawConn struct {

233 - conn net.PacketConn

234 - rdeadline, wdeadline time.Time

235 - in chan rawIncoming

236 - closed chan int

237 -}

81 + c := newConn()

82 + c.state = stateConnected

83 + c.conn = l.conn

84 + c.raddr = p.addr

85 + c.rid = p.header.id + 1

86 + c.sid = p.header.id

87 + c.seq = uint16(seq)

88 + c.ack = p.header.seq

89 + c.recvbuf = newPacketBuffer(windowSize, int(p.header.seq))

90 + c.sendbuf = newPacketBuffer(windowSize*2, seq)

91 + l.conn.Register(int32(rid), c.recv)

92 + go c.loop()

93 + c.recv <- p

239 -func newRawConn(conn net.PacketConn) *rawConn {

240 - return &rawConn{

241 - conn: conn,

242 - in: make(chan rawIncoming, 100),

243 - closed: make(chan int),

244 - }

95 + ulog.Printf(2, "baseConn(%v): accept #%d from %v", c.LocalAddr(), c.rid, c.raddr)

96 + return c, nil

97 }

247 -func (c *rawConn) ok() bool { return c != nil && c.conn != nil }

248 -

249 -func (c *rawConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) {

250 - if !c.ok() {

251 - return 0, nil, syscall.EINVAL

252 - }

253 - select {

254 - case <-c.closed:

255 - return 0, nil, errors.New("use of closed network connection")

256 - default:

257 - }

258 - var timeout <-chan time.Time

259 - if !c.rdeadline.IsZero() {

260 - timeout = time.After(c.rdeadline.Sub(time.Now()))

261 - }

262 - select {

263 - case r := <-c.in:

264 - return copy(b, r.b), r.addr, nil

265 - case <-timeout:

266 - return 0, nil, &timeoutError{}

267 - }

268 -}

269 -

270 -func (c *rawConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {

271 - if !c.ok() {

272 - return 0, syscall.EINVAL

273 - }

274 - select {

275 - case <-c.closed:

276 - return 0, errors.New("use of closed network connection")

277 - default:

99 +// Addr returns the listener's network address, a *Addr.

100 +func (l *Listener) Addr() net.Addr {

101 + if !l.ok() {

102 + return nil

103 }

279 - return c.conn.WriteTo(b, addr)

104 + return l.conn.LocalAddr()

105 }

106

282 -func (c *rawConn) Close() error {

283 - if !c.ok() {

107 +// Close stops listening on the UTP address.

108 +// Already Accepted connections are not closed.

109 +func (l *Listener) Close() error {

110 + if !l.ok() {

111 return syscall.EINVAL

112 }

286 - select {

287 - case <-c.closed:

288 - return errors.New("use of closed network connection")

289 - default:

290 - close(c.closed)

113 + if !l.close() {

114 + return &net.OpError{

115 + Op: "close",

116 + Net: l.conn.LocalAddr().Network(),

117 + Addr: l.conn.LocalAddr(),

118 + Err: errClosing,

119 + }

120 }

121 return nil

122 }

123

295 -func (c *rawConn) LocalAddr() net.Addr {

296 - if !c.ok() {

297 - return nil

298 - }

299 - return c.conn.LocalAddr()

300 -}

301 -

302 -func (c *rawConn) SetDeadline(t time.Time) error {

303 - if !c.ok() {

124 +// SetDeadline sets the deadline associated with the listener.

125 +// A zero time value disables the deadline.

126 +func (l *Listener) SetDeadline(t time.Time) error {

127 + if !l.ok() {

128 return syscall.EINVAL

129 }

306 - if err := c.SetReadDeadline(t); err != nil {

307 - return err

308 - }

309 - if err := c.SetWriteDeadline(t); err != nil {

310 - return err

311 - }

130 + l.deadlineMutex.Lock()

131 + defer l.deadlineMutex.Unlock()

132 + l.deadline = t

133 return nil

134 }

135

315 -func (c *rawConn) SetReadDeadline(t time.Time) error {

316 - if !c.ok() {

317 - return syscall.EINVAL

136 +func (l *Listener) close() bool {

137 + if atomic.CompareAndSwapInt32(&l.closed, 0, 1) {

138 + l.conn.Unregister(-1)

139 + return true

140 }

319 - c.rdeadline = t

320 - return nil

141 + return false

142 }

143

323 -func (c *rawConn) SetWriteDeadline(t time.Time) error {

324 - if !c.ok() {

325 - return syscall.EINVAL

326 - }

327 - c.wdeadline = t

328 - return nil

144 +func (l *Listener) isOpen() bool {

145 + return atomic.LoadInt32(&l.closed) == 0

146 }

Godeps/_workspace/src/github.com/h2so5/utp/listener_test.go new

+68

@@ -0,0 +1,68 @@

1 +package utp

2 +

3 +import (

4 + "net"

5 + "testing"

6 +)

7 +

8 +func TestListenerAccept(t *testing.T) {

9 + addr, err := ResolveAddr("utp", "127.0.0.1:0")

10 + if err != nil {

11 + t.Fatal(err)

12 + }

13 +

14 + l, err := Listen("utp", addr)

15 + if err != nil {

16 + t.Fatal(err)

17 + }

18 + defer l.Close()

19 +

20 + uaddr, err := net.ResolveUDPAddr("udp", l.Addr().String())

21 + if err != nil {

22 + t.Fatal(err)

23 + }

24 +

25 + uc, err := net.DialUDP("udp", nil, uaddr)

26 + if err != nil {

27 + t.Fatal(err)

28 + }

29 + defer uc.Close()

30 +

31 + for i := 0; i < 1; i++ {

32 + p := &packet{header: header{typ: stSyn, ver: version, id: uint16(i)}}

33 + payload, err := p.MarshalBinary()

34 + if err != nil {

35 + t.Fatal(err)

36 + }

37 + go func() {

38 + uc.Write(payload)

39 + }()

40 +

41 + a, err := l.Accept()

42 + if err != nil {

43 + t.Fatal(err)

44 + }

45 + a.Close()

46 + }

47 +}

48 +

49 +func TestListenerClose(t *testing.T) {

50 + addr, err := ResolveAddr("utp", ":0")

51 + if err != nil {

52 + t.Fatal(err)

53 + }

54 +

55 + l, err := Listen("utp", addr)

56 + if err != nil {

57 + t.Fatal(err)

58 + }

59 +

60 + for i := 0; i < 5; i++ {

61 + l.Close()

62 + }

63 +

64 + _, err = l.Accept()

65 + if err == nil {

66 + t.Fatal("Accept should fail")

67 + }

68 +}

Godeps/_workspace/src/github.com/h2so5/utp/packet.go

+16 -58

@@ -5,7 +5,8 @@ import (

5 "encoding/binary"

6 "fmt"

7 "io"

8 - "sync"

8 + "io/ioutil"

9 + "net"

10 )

12 type header struct {

@@ -24,16 +25,11 @@ type packet struct {

25 header header

26 ext []extension

27 payload []byte

27 -}

28 -

29 -type outgoingPacket struct {

30 - typ int

31 - ext []extension

32 - payload []byte

28 + addr net.Addr

29 }

31 func (p *packet) MarshalBinary() ([]byte, error) {

36 - firstExt := ext_none

32 + firstExt := extNone

33 if len(p.ext) > 0 {

34 firstExt = p.ext[0].typ

35 }

@@ -68,7 +64,7 @@ func (p *packet) MarshalBinary() ([]byte, error) {

65 if len(p.ext) > 0 {

66 for i, e := range p.ext {

71 - next := ext_none

67 + next := extNone

68 if i < len(p.ext)-1 {

69 next = p.ext[i+1].typ

70 }

@@ -123,7 +119,7 @@ func (p *packet) UnmarshalBinary(data []byte) error {

119 }

120 }

121

126 - for e != ext_none {

122 + for e != extNone {

123 currentExt := int(e)

124 var l uint8

125 var ext = []interface{}{

@@ -173,68 +169,30 @@ func (p *packet) UnmarshalBinary(data []byte) error {

169 p.header.typ = int((tv >> 4) & 0xF)

170 p.header.ver = int(tv & 0xF)

171

176 - l := buf.Len()

177 - if l > 0 {

178 - p.payload = p.payload[:l]

179 - _, err := buf.Read(p.payload[:])

180 - if err != nil {

181 - return err

182 - }

172 + data, err := ioutil.ReadAll(buf)

173 + if err != nil {

174 + return err

175 }

176 + p.payload = data

177

178 return nil

179 }

180

181 func (p packet) String() string {

189 - var s string = fmt.Sprintf("[%d ", p.header.id)

182 + s := fmt.Sprintf("[%d ", p.header.id)

183 switch p.header.typ {

191 - case st_data:

184 + case stData:

185 s += "ST_DATA"

193 - case st_fin:

186 + case stFin:

187 s += "ST_FIN"

195 - case st_state:

188 + case stState:

189 s += "ST_STATE"

197 - case st_reset:

190 + case stReset:

191 s += "ST_RESET"

199 - case st_syn:

192 + case stSyn:

193 s += "ST_SYN"

194 }

195 s += fmt.Sprintf(" seq:%d ack:%d len:%d", p.header.seq, p.header.ack, len(p.payload))

196 s += "]"

197 return s

198 }

206 -

207 -var globalPool packetPool

208 -

209 -type packetPool struct {

210 - root *packetPoolNode

211 - mutex sync.Mutex

212 -}

213 -

214 -type packetPoolNode struct {

215 - p *packet

216 - next *packetPoolNode

217 -}

218 -

219 -func (o *packetPool) get() *packet {

220 - o.mutex.Lock()

221 - defer o.mutex.Unlock()

222 - r := o.root

223 - if r != nil {

224 - o.root = o.root.next

225 - return r.p

226 - } else {

227 - return &packet{

228 - payload: make([]byte, 0, mss),

229 - }

230 - }

231 -}

232 -

233 -func (o *packetPool) put(p *packet) {

234 - o.mutex.Lock()

235 - defer o.mutex.Unlock()

236 - o.root = &packetPoolNode{

237 - p: p,

238 - next: o.root,

239 - }

240 -}

Godeps/_workspace/src/github.com/h2so5/utp/packet_test.go new

+64

@@ -0,0 +1,64 @@

1 +package utp

2 +

3 +import (

4 + "io"

5 + "reflect"

6 + "testing"

7 +)

8 +

9 +func TestPacketBinary(t *testing.T) {

10 + h := header{

11 + typ: stFin,

12 + ver: version,

13 + id: 100,

14 + t: 50000,

15 + diff: 10000,

16 + wnd: 65535,

17 + seq: 100,

18 + ack: 200,

19 + }

20 +

21 + e := []extension{

22 + extension{

23 + typ: extSelectiveAck,

24 + payload: []byte{0, 1, 0, 1},

25 + },

26 + extension{

27 + typ: extSelectiveAck,

28 + payload: []byte{100, 0, 200, 0},

29 + },

30 + }

31 +

32 + p := packet{

33 + header: h,

34 + ext: e,

35 + payload: []byte("abcdefg"),

36 + }

37 +

38 + b, err := p.MarshalBinary()

39 + if err != nil {

40 + t.Fatal(err)

41 + }

42 +

43 + p2 := packet{payload: make([]byte, 0, mss)}

44 + err = p2.UnmarshalBinary(b)

45 + if err != nil {

46 + t.Fatal(err)

47 + }

48 +

49 + if !reflect.DeepEqual(p, p2) {

50 + t.Errorf("expected packet of %v; got %v", p, p2)

51 + }

52 +}

53 +

54 +func TestUnmarshalShortPacket(t *testing.T) {

55 + b := make([]byte, 18)

56 + p := packet{}

57 + err := p.UnmarshalBinary(b)

58 +

59 + if err == nil {

60 + t.Fatal("UnmarshalBinary should fail")

61 + } else if err != io.EOF {

62 + t.Fatal(err)

63 + }

64 +}

Godeps/_workspace/src/github.com/h2so5/utp/ucat/ucat.go

+6 -2

@@ -111,7 +111,11 @@ func log(format string, vals ...interface{}) {

111 // Listen listens and accepts one incoming uTP connection on a given port,

112 // and pipes all incoming data to os.Stdout.

113 func Listen(localAddr string) error {

114 - l, err := utp.Listen("utp", localAddr)

114 + laddr, err := utp.ResolveAddr("utp", localAddr)

115 + if err != nil {

116 + return fmt.Errorf("failed to resolve address %s", localAddr)

117 + }

118 + l, err := utp.Listen("utp", laddr)

119 if err != nil {

120 return err

121 }

@@ -138,7 +142,7 @@ func Dial(localAddr, remoteAddr string) error {

142 var laddr net.Addr

143 var err error

144 if localAddr != "" {

141 - laddr, err = utp.ResolveUTPAddr("utp", localAddr)

145 + laddr, err = utp.ResolveAddr("utp", localAddr)

146 if err != nil {

147 return fmt.Errorf("failed to resolve address %s", localAddr)

148 }

Godeps/_workspace/src/github.com/h2so5/utp/utp.go

+34 -16

@@ -1,25 +1,38 @@

1 package utp

3 -import "time"

3 +import (

4 + "errors"

5 + "time"

6 +)

8 const (

9 version = 1

8 - st_data = 0

9 - st_fin = 1

10 - st_state = 2

11 - st_reset = 3

12 - st_syn = 4

13 -

14 - ext_none = 0

15 - ext_selective_ack = 1

16 -

17 - header_size = 20

18 - mtu = 3200

19 - mss = mtu - header_size

20 - window_size = 100

21 -

22 - reset_timeout = time.Second

11 + stData = 0

12 + stFin = 1

13 + stState = 2

14 + stReset = 3

15 + stSyn = 4

16 +

17 + stateClosed = iota

18 + stateClosing

19 + stateSynSent

20 + stateConnected

21 + stateFinSent

22 +

23 + extNone = 0

24 + extSelectiveAck = 1

25 +

26 + headerSize = 20

27 + mtu = 3200

28 + mss = mtu - headerSize

29 + windowSize = 100

30 + packetBufferSize = 256

31 + readBufferSize = 1048576

32 + maxRetry = 3

33 +

34 + maxUdpPayload = 65507

35 + resetTimeout = time.Second

36 )

38 type timeoutError struct{}

@@ -27,3 +40,8 @@ type timeoutError struct{}

40 func (e *timeoutError) Error() string { return "i/o timeout" }

41 func (e *timeoutError) Timeout() bool { return true }

42 func (e *timeoutError) Temporary() bool { return true }

43 +

44 +var (

45 + errTimeout error = &timeoutError{}

46 + errClosing = errors.New("use of closed network connection")

47 +)

Godeps/_workspace/src/github.com/h2so5/utp/utp_test.go deleted

-601

@@ -1,601 +0,0 @@

1 -package utp

2 -

3 -import (

4 - "bytes"

5 - "io"

6 - "io/ioutil"

7 - "math"

8 - "math/rand"

9 - "net"

10 - "reflect"

11 - "testing"

12 - "time"

13 -)

14 -

15 -func init() {

16 - rand.Seed(time.Now().Unix())

17 -}

18 -

19 -func TestReadWrite(t *testing.T) {

20 - ln, err := Listen("utp", "127.0.0.1:0")

21 - if err != nil {

22 - t.Fatal(err)

23 - }

24 -

25 - raddr, err := ResolveUTPAddr("utp", ln.Addr().String())

26 - if err != nil {

27 - t.Fatal(err)

28 - }

29 -

30 - c, err := DialUTPTimeout("utp", nil, raddr, 1000*time.Millisecond)

31 - if err != nil {

32 - t.Fatal(err)

33 - }

34 - defer c.Close()

35 -

36 - err = ln.SetDeadline(time.Now().Add(1000 * time.Millisecond))

37 - if err != nil {

38 - t.Fatal(err)

39 - }

40 -

41 - s, err := ln.Accept()

42 - if err != nil {

43 - t.Fatal(err)

44 - }

45 - ln.Close()

46 -

47 - payload := []byte("Hello!")

48 - _, err = c.Write(payload)

49 - if err != nil {

50 - t.Fatal(err)

51 - }

52 -

53 - err = s.SetDeadline(time.Now().Add(1000 * time.Millisecond))

54 - if err != nil {

55 - t.Fatal(err)

56 - }

57 -

58 - var buf [256]byte

59 - l, err := s.Read(buf[:])

60 - if err != nil {

61 - t.Fatal(err)

62 - }

63 -

64 - if !bytes.Equal(payload, buf[:l]) {

65 - t.Errorf("expected payload of %v; got %v", payload, buf[:l])

66 - }

67 -

68 - payload2 := []byte("World!")

69 - _, err = s.Write(payload2)

70 - if err != nil {

71 - t.Fatal(err)

72 - }

73 -

74 - err = c.SetDeadline(time.Now().Add(1000 * time.Millisecond))

75 - if err != nil {

76 - t.Fatal(err)

77 - }

78 -

79 - l, err = c.Read(buf[:])

80 - if err != nil {

81 - t.Fatal(err)

82 - }

83 -

84 - if !bytes.Equal(payload2, buf[:l]) {

85 - t.Errorf("expected payload of %v; got %v", payload2, buf[:l])

86 - }

87 -}

88 -

89 -func TestRawReadWrite(t *testing.T) {

90 - ln, err := Listen("utp", "127.0.0.1:0")

91 - if err != nil {

92 - t.Fatal(err)

93 - }

94 - defer ln.Close()

95 -

96 - raddr, err := net.ResolveUDPAddr("udp", ln.Addr().String())

97 - if err != nil {

98 - t.Fatal(err)

99 - }

100 -

101 - c, err := net.DialUDP("udp", nil, raddr)

102 - if err != nil {

103 - t.Fatal(err)

104 - }

105 - defer c.Close()

106 -

107 - payload := []byte("Hello!")

108 - _, err = c.Write(payload)

109 - if err != nil {

110 - t.Fatal(err)

111 - }

112 -

113 - var buf [256]byte

114 - n, addr, err := ln.RawConn.ReadFrom(buf[:])

115 - if !bytes.Equal(payload, buf[:n]) {

116 - t.Errorf("expected payload of %v; got %v", payload, buf[:n])

117 - }

118 - if addr.String() != c.LocalAddr().String() {

119 - t.Errorf("expected addr of %v; got %v", c.LocalAddr(), addr.String())

120 - }

121 -}

122 -

123 -func TestLongReadWriteC2S(t *testing.T) {

124 - ln, err := Listen("utp", "127.0.0.1:0")

125 - if err != nil {

126 - t.Fatal(err)

127 - }

128 -

129 - raddr, err := ResolveUTPAddr("utp", ln.Addr().String())

130 - if err != nil {

131 - t.Fatal(err)

132 - }

133 -

134 - c, err := DialUTPTimeout("utp", nil, raddr, 1000*time.Millisecond)

135 - if err != nil {

136 - t.Fatal(err)

137 - }

138 - defer c.Close()

139 -

140 - err = ln.SetDeadline(time.Now().Add(1000 * time.Millisecond))

141 - if err != nil {

142 - t.Fatal(err)

143 - }

144 -

145 - s, err := ln.Accept()

146 - if err != nil {

147 - t.Fatal(err)

148 - }

149 - defer s.Close()

150 - ln.Close()

151 -

152 - var payload [10485760]byte

153 - for i := range payload {

154 - payload[i] = byte(rand.Int())

155 - }

156 -

157 - rch := make(chan []byte)

158 - ech := make(chan error, 2)

159 -

160 - go func() {

161 - defer c.Close()

162 - _, err := c.Write(payload[:])

163 - if err != nil {

164 - ech <- err

165 - }

166 - }()

167 -

168 - go func() {

169 - b, err := ioutil.ReadAll(s)

170 - if err != nil {

171 - ech <- err

172 - rch <- nil

173 - } else {

174 - ech <- nil

175 - rch <- b

176 - }

177 - }()

178 -

179 - err = <-ech

180 - if err != nil {

181 - t.Fatal(err)

182 - }

183 -

184 - r := <-rch

185 - if r == nil {

186 - return

187 - }

188 -

189 - if !bytes.Equal(r, payload[:]) {

190 - t.Errorf("expected payload of %d; got %d", len(payload[:]), len(r))

191 - }

192 -}

193 -

194 -func TestLongReadWriteS2C(t *testing.T) {

195 - ln, err := Listen("utp", "127.0.0.1:0")

196 - if err != nil {

197 - t.Fatal(err)

198 - }

199 -

200 - raddr, err := ResolveUTPAddr("utp", ln.Addr().String())

201 - if err != nil {

202 - t.Fatal(err)

203 - }

204 -

205 - c, err := DialUTPTimeout("utp", nil, raddr, 1000*time.Millisecond)

206 - if err != nil {

207 - t.Fatal(err)

208 - }

209 - defer c.Close()

210 -

211 - err = ln.SetDeadline(time.Now().Add(1000 * time.Millisecond))

212 - if err != nil {

213 - t.Fatal(err)

214 - }

215 -

216 - s, err := ln.Accept()

217 - if err != nil {

218 - t.Fatal(err)

219 - }

220 - defer s.Close()

221 - ln.Close()

222 -

223 - var payload [10485760]byte

224 - for i := range payload {

225 - payload[i] = byte(rand.Int())

226 - }

227 -

228 - rch := make(chan []byte)

229 - ech := make(chan error, 2)

230 -

231 - go func() {

232 - defer s.Close()

233 - _, err := s.Write(payload[:])

234 - if err != nil {

235 - ech <- err

236 - }

237 - }()

238 -

239 - go func() {

240 - b, err := ioutil.ReadAll(c)

241 - if err != nil {

242 - ech <- err

243 - rch <- nil

244 - } else {

245 - ech <- nil

246 - rch <- b

247 - }

248 - }()

249 -

250 - err = <-ech

251 - if err != nil {

252 - t.Fatal(err)

253 - }

254 -

255 - r := <-rch

256 - if r == nil {

257 - return

258 - }

259 -

260 - if !bytes.Equal(r, payload[:]) {

261 - t.Errorf("expected payload of %d; got %d", len(payload[:]), len(r))

262 - }

263 -}

264 -

265 -func TestAccept(t *testing.T) {

266 - ln, err := Listen("utp", "127.0.0.1:0")

267 - if err != nil {

268 - t.Fatal(err)

269 - }

270 - defer ln.Close()

271 -

272 - c, err := DialUTPTimeout("utp", nil, ln.Addr().(*UTPAddr), 200*time.Millisecond)

273 - if err != nil {

274 - t.Fatal(err)

275 - }

276 - defer c.Close()

277 -

278 - err = ln.SetDeadline(time.Now().Add(100 * time.Millisecond))

279 - _, err = ln.Accept()

280 - if err != nil {

281 - t.Fatal(err)

282 - }

283 -}

284 -

285 -func TestAcceptDeadline(t *testing.T) {

286 - ln, err := Listen("utp", "127.0.0.1:0")

287 - if err != nil {

288 - t.Fatal(err)

289 - }

290 - defer ln.Close()

291 - err = ln.SetDeadline(time.Now().Add(time.Millisecond))

292 - _, err = ln.Accept()

293 - if err == nil {

294 - t.Fatal("Accept should failed")

295 - }

296 -}

297 -

298 -func TestAcceptClosedListener(t *testing.T) {

299 - ln, err := Listen("utp", "127.0.0.1:0")

300 - if err != nil {

301 - t.Fatal(err)

302 - }

303 - err = ln.Close()

304 - if err != nil {

305 - t.Fatal(err)

306 - }

307 - _, err = ln.Accept()

308 - if err == nil {

309 - t.Fatal("Accept should failed")

310 - }

311 - _, err = ln.Accept()

312 - if err == nil {

313 - t.Fatal("Accept should failed")

314 - }

315 -}

316 -

317 -func TestDialer(t *testing.T) {

318 - ln, err := Listen("utp", "127.0.0.1:0")

319 - if err != nil {

320 - t.Fatal(err)

321 - }

322 - defer ln.Close()

323 -

324 - d := Dialer{}

325 - c, err := d.Dial("utp", ln.Addr().String())

326 - if err != nil {

327 - t.Fatal(err)

328 - }

329 - defer c.Close()

330 -}

331 -

332 -func TestDialerAddrs(t *testing.T) {

333 - ln, err := Listen("utp", "127.0.0.1:0")

334 - if err != nil {

335 - t.Fatal(err)

336 - }

337 - defer ln.Close()

338 -

339 - laddr, err := ResolveUTPAddr("utp", "127.0.0.1:45678")

340 - if err != nil {

341 - t.Fatal(err)

342 - }

343 -

344 - d := Dialer{LocalAddr: laddr}

345 - c1, err := d.Dial("utp", ln.Addr().String())

346 - if err != nil {

347 - t.Fatal(err)

348 - }

349 - defer c1.Close()

350 -

351 - c2, err := ln.Accept()

352 - if err != nil {

353 - t.Fatal(err)

354 - }

355 - defer c2.Close()

356 -

357 - eq := func(a, b net.Addr) bool {

358 - return a.String() == b.String()

359 - }

360 -

361 - if !eq(d.LocalAddr, c2.RemoteAddr()) {

362 - t.Fatal("dialer.LocalAddr not equal to c2.RemoteAddr ")

363 - }

364 - if !eq(c1.LocalAddr(), c2.RemoteAddr()) {

365 - t.Fatal("c1.LocalAddr not equal to c2.RemoteAddr ")

366 - }

367 - if !eq(c2.LocalAddr(), c1.RemoteAddr()) {

368 - t.Fatal("c2.LocalAddr not equal to c1.RemoteAddr ")

369 - }

370 -}

371 -

372 -func TestDialerTimeout(t *testing.T) {

373 - timeout := time.Millisecond * 200

374 - d := Dialer{Timeout: timeout}

375 - done := make(chan struct{})

376 -

377 - go func() {

378 - _, err := d.Dial("utp", "127.0.0.1:34567")

379 - if err == nil {

380 - t.Fatal("should not connect")

381 - }

382 - done <- struct{}{}

383 - }()

384 -

385 - select {

386 - case <-time.After(timeout * 2):

387 - t.Fatal("should have ended already")

388 - case <-done:

389 - }

390 -}

391 -

392 -func TestPacketBinary(t *testing.T) {

393 - h := header{

394 - typ: st_fin,

395 - ver: version,

396 - id: 100,

397 - t: 50000,

398 - diff: 10000,

399 - wnd: 65535,

400 - seq: 100,

401 - ack: 200,

402 - }

403 -

404 - e := []extension{

405 - extension{

406 - typ: ext_selective_ack,

407 - payload: []byte{0, 1, 0, 1},

408 - },

409 - extension{

410 - typ: ext_selective_ack,

411 - payload: []byte{100, 0, 200, 0},

412 - },

413 - }

414 -

415 - p := packet{

416 - header: h,

417 - ext: e,

418 - payload: []byte("abcdefg"),

419 - }

420 -

421 - b, err := p.MarshalBinary()

422 - if err != nil {

423 - t.Fatal(err)

424 - }

425 -

426 - p2 := packet{payload: make([]byte, 0, mss)}

427 - err = p2.UnmarshalBinary(b)

428 - if err != nil {

429 - t.Fatal(err)

430 - }

431 -

432 - if !reflect.DeepEqual(p, p2) {

433 - t.Errorf("expected packet of %v; got %v", p, p2)

434 - }

435 -}

436 -

437 -func TestUnmarshalShortPacket(t *testing.T) {

438 - b := make([]byte, 18)

439 - p := packet{}

440 - err := p.UnmarshalBinary(b)

441 -

442 - if err == nil {

443 - t.Fatal("UnmarshalBinary should fail")

444 - } else if err != io.EOF {

445 - t.Fatal(err)

446 - }

447 -}

448 -

449 -func TestWriteOnClosedChannel(t *testing.T) {

450 - ln, err := Listen("utp", "127.0.0.1:0")

451 - if err != nil {

452 - t.Fatal(err)

453 - }

454 - defer ln.Close()

455 -

456 - c, err := DialUTPTimeout("utp", nil, ln.Addr().(*UTPAddr), 200*time.Millisecond)

457 - if err != nil {

458 - t.Fatal(err)

459 - }

460 -

461 - go func() {

462 - for {

463 - _, err := c.Write([]byte{100})

464 - if err != nil {

465 - return

466 - }

467 - }

468 - }()

469 -

470 - c.Close()

471 -}

472 -

473 -func TestReadOnClosedChannel(t *testing.T) {

474 - ln, err := Listen("utp", "127.0.0.1:0")

475 - if err != nil {

476 - t.Fatal(err)

477 - }

478 - defer ln.Close()

479 -

480 - c, err := DialUTPTimeout("utp", nil, ln.Addr().(*UTPAddr), 200*time.Millisecond)

481 - if err != nil {

482 - t.Fatal(err)

483 - }

484 -

485 - go func() {

486 - for {

487 - var buf [16]byte

488 - _, err := c.Read(buf[:])

489 - if err != nil {

490 - return

491 - }

492 - }

493 - }()

494 -

495 - c.Close()

496 -}

497 -

498 -func TestPacketBuffer(t *testing.T) {

499 - size := 12

500 - b := newPacketBuffer(12, 1)

501 -

502 - if b.space() != size {

503 - t.Errorf("expected space == %d; got %d", size, b.space())

504 - }

505 -

506 - for i := 1; i <= size; i++ {

507 - b.push(&packet{header: header{seq: uint16(i)}})

508 - }

509 -

510 - if b.space() != 0 {

511 - t.Errorf("expected space == 0; got %d", b.space())

512 - }

513 -

514 - a := []byte{255, 7}

515 - ack := b.generateSelectiveACK()

516 - if !bytes.Equal(a, ack) {

517 - t.Errorf("expected ack == %v; got %v", a, ack)

518 - }

519 -

520 - err := b.push(&packet{header: header{seq: 15}})

521 - if err == nil {

522 - t.Fatal("push should fail")

523 - }

524 -

525 - all := b.all()

526 - if len(all) != size {

527 - t.Errorf("expected %d packets sequence; got %d", size, len(all))

528 - }

529 -

530 - f := b.fetch(6)

531 - if f == nil {

532 - t.Fatal("fetch should not fail")

533 - }

534 -

535 - b.compact()

536 -

537 - err = b.push(&packet{header: header{seq: 15}})

538 - if err != nil {

539 - t.Fatal(err)

540 - }

541 -

542 - err = b.push(&packet{header: header{seq: 17}})

543 - if err != nil {

544 - t.Fatal(err)

545 - }

546 -

547 - for i := 7; i <= size; i++ {

548 - f := b.fetch(uint16(i))

549 - if f == nil {

550 - t.Fatal("fetch should not fail")

551 - }

552 - }

553 -

554 - a = []byte{128, 2}

555 - ack = b.generateSelectiveACK()

556 - if !bytes.Equal(a, ack) {

557 - t.Errorf("expected ack == %v; got %v", a, ack)

558 - }

559 -

560 - all = b.all()

561 - if len(all) != 2 {

562 - t.Errorf("expected 2 packets sequence; got %d", len(all))

563 - }

564 -

565 - b.compact()

566 - if b.space() != 9 {

567 - t.Errorf("expected space == 9; got %d", b.space())

568 - }

569 -

570 - ack = b.generateSelectiveACK()

571 - b.processSelectiveACK(ack)

572 -

573 - all = b.all()

574 - if len(all) != 1 {

575 - t.Errorf("expected size == 1; got %d", len(all))

576 - }

577 -}

578 -

579 -func TestPacketBufferBoundary(t *testing.T) {

580 - begin := math.MaxUint16 - 3

581 - b := newPacketBuffer(12, begin)

582 - for i := begin; i != 5; i = (i + 1) % (math.MaxUint16 + 1) {

583 - err := b.push(&packet{header: header{seq: uint16(i)}})

584 - if err != nil {

585 - t.Fatal(err)

586 - }

587 - }

588 -}

589 -

590 -func TestTimedBufferNode(t *testing.T) {

591 - b := timedBuffer{d: time.Millisecond * 100}

592 - b.push(100)

593 - b.push(200)

594 - time.Sleep(time.Millisecond * 200)

595 - b.push(300)

596 - b.push(400)

597 - m := b.min()

598 - if m != 300 {

599 - t.Errorf("expected min == 300; got %d", m)

600 - }

601 -}