Golang implementation of the Reliable UDP transport, revised from an original documented mostly in chinese.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

rudp.go 9.2KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459
  1. package rudp
  2. import (
  3. "bytes"
  4. "errors"
  5. "time"
  6. "go.uber.org/atomic"
  7. )
  8. const (
  9. TYPE_PING = iota
  10. TYPE_EOF
  11. TYPE_CORRUPT
  12. TYPE_REQUEST
  13. TYPE_MISSING
  14. TYPE_NORMAL
  15. )
  16. const (
  17. MAX_MSG_HEAD = 4
  18. GENERAL_PACKAGE = 576 - 60 - 8
  19. MAX_PACKAGE = 0x7fff - TYPE_NORMAL
  20. )
  21. type Err struct {
  22. atomic.Int32
  23. }
  24. const (
  25. ERROR_NIL int32 = iota
  26. ERROR_EOF
  27. ERROR_REMOTE_EOF
  28. ERROR_CORRUPT
  29. ERROR_MSG_SIZE
  30. )
  31. func (e Err) Error() error {
  32. switch e.Int32.Load() {
  33. case ERROR_EOF:
  34. return errors.New("EOF")
  35. case ERROR_REMOTE_EOF:
  36. return errors.New("remote EOF")
  37. case ERROR_CORRUPT:
  38. return errors.New("corrupt")
  39. case ERROR_MSG_SIZE:
  40. return errors.New("recive msg size error")
  41. default:
  42. return nil
  43. }
  44. }
  45. type Package struct {
  46. Next *Package
  47. Bts []byte
  48. }
  49. type packageBuffer struct {
  50. tmp bytes.Buffer
  51. num int
  52. head *Package
  53. tail *Package
  54. }
  55. func (pb *packageBuffer) packRequest(min, max int, tag int) {
  56. if pb.tmp.Len()+5 > GENERAL_PACKAGE {
  57. pb.newPackage()
  58. }
  59. pb.tmp.WriteByte(byte(tag))
  60. pb.tmp.WriteByte(byte((min & 0xff00) >> 8))
  61. pb.tmp.WriteByte(byte(min & 0xff))
  62. pb.tmp.WriteByte(byte((max & 0xff00) >> 8))
  63. pb.tmp.WriteByte(byte(max & 0xff))
  64. }
  65. func (pb *packageBuffer) fillHeader(head, id int) {
  66. if head < 128 {
  67. pb.tmp.WriteByte(byte(head))
  68. } else {
  69. pb.tmp.WriteByte(byte(((head & 0x7f00) >> 8) | 0x80))
  70. pb.tmp.WriteByte(byte(head & 0xff))
  71. }
  72. pb.tmp.WriteByte(byte((id & 0xff00) >> 8))
  73. pb.tmp.WriteByte(byte(id & 0xff))
  74. }
  75. func (pb *packageBuffer) packMessage(m *message) {
  76. if m.buf.Len()+4+pb.tmp.Len() >= GENERAL_PACKAGE {
  77. pb.newPackage()
  78. }
  79. pb.fillHeader(m.buf.Len()+TYPE_NORMAL, m.id)
  80. pb.tmp.Write(m.buf.Bytes())
  81. }
  82. func (pb *packageBuffer) newPackage() {
  83. if pb.tmp.Len() <= 0 {
  84. return
  85. }
  86. p := &Package{Bts: make([]byte, pb.tmp.Len())}
  87. copy(p.Bts, pb.tmp.Bytes())
  88. pb.tmp.Reset()
  89. pb.num++
  90. if pb.tail == nil {
  91. pb.head = p
  92. pb.tail = p
  93. } else {
  94. pb.tail.Next = p
  95. pb.tail = p
  96. }
  97. }
  98. func New() *Rudp {
  99. return &Rudp{reqSendAgain: make(chan [2]int, 1<<10),
  100. addSendAgain: make(chan [2]int, 1<<10), recvSkip: make(map[int]int)}
  101. }
  102. type Rudp struct {
  103. recvQueue messageQueue
  104. recvSkip map[int]int
  105. reqSendAgain chan [2]int
  106. recvIDMin int
  107. recvIDMax int
  108. sendQueue messageQueue
  109. sendHistory messageQueue
  110. addSendAgain chan [2]int
  111. sendID int
  112. corrupt Err
  113. currentTick int
  114. lastRecvTick int
  115. lastExpiredTick int
  116. lastSendDelayTick int
  117. }
  118. func (r *Rudp) Recv(bts []byte) (int, error) {
  119. if err := r.corrupt.Load(); err != ERROR_NIL {
  120. return 0, r.corrupt.Error()
  121. }
  122. m := r.recvQueue.pop(r.recvIDMin)
  123. if m == nil {
  124. return 0, nil
  125. }
  126. r.recvIDMin++
  127. // TODO: can this copy be avoided? freelists?
  128. copy(bts, m.buf.Bytes())
  129. return m.buf.Len(), nil
  130. }
  131. func (r *Rudp) Send(bts []byte) (n int, err error) {
  132. if err := r.corrupt.Load(); err != ERROR_NIL {
  133. return 0, r.corrupt.Error()
  134. }
  135. if len(bts) > MAX_PACKAGE {
  136. return 0, nil
  137. }
  138. m := &message{}
  139. m.buf.Write(bts)
  140. m.id = r.sendID
  141. r.sendID++
  142. m.tick = r.currentTick
  143. r.sendQueue.push(m)
  144. return len(bts), nil
  145. }
  146. func (r *Rudp) Update(tick int) *Package {
  147. if r.corrupt.Load() != ERROR_NIL {
  148. return nil
  149. }
  150. r.currentTick += tick
  151. if r.currentTick >= r.lastExpiredTick+expiredTick {
  152. r.lastExpiredTick = r.currentTick
  153. r.clearSendExpired()
  154. }
  155. if r.currentTick >= r.lastRecvTick+corruptTick {
  156. r.corrupt.Store(ERROR_CORRUPT)
  157. }
  158. if r.currentTick >= r.lastSendDelayTick+sendDelayTick {
  159. r.lastSendDelayTick = r.currentTick
  160. return r.outPut()
  161. }
  162. return nil
  163. }
  164. type message struct {
  165. next *message
  166. buf bytes.Buffer
  167. id int
  168. tick int
  169. }
  170. type messageQueue struct {
  171. head *message
  172. tail *message
  173. num int
  174. }
  175. func (mq *messageQueue) pop(id int) *message {
  176. if mq.head == nil {
  177. return nil
  178. }
  179. m := mq.head
  180. if id >= 0 && m.id != id {
  181. return nil
  182. }
  183. mq.head = m.next
  184. m.next = nil
  185. if mq.head == nil {
  186. mq.tail = nil
  187. }
  188. mq.num--
  189. return m
  190. }
  191. func (mq *messageQueue) push(m *message) {
  192. if mq.tail == nil {
  193. mq.head = m
  194. mq.tail = m
  195. } else {
  196. mq.tail.next = m
  197. mq.tail = m
  198. }
  199. mq.num++
  200. }
  201. func (r *Rudp) getID(max int, bt1, bt2 byte) int {
  202. n1, n2 := int(bt1), int(bt2)
  203. id := n1*256 + n2
  204. id |= max & ^0xffff
  205. if id < max-0x8000 {
  206. id += 0x10000
  207. dbg("id < max-0x8000 ,net %v,id %v,min %v,max %v,cur %v",
  208. n1*256+n2, id, r.recvIDMin, max, id+0x10000)
  209. } else if id > max+0x8000 {
  210. id -= 0x10000
  211. dbg("id > max-0x8000 ,net %v,id %v,min %v,max %v,cur %v",
  212. n1*256+n2, id, r.recvIDMin, max, id+0x10000)
  213. }
  214. return id
  215. }
  216. func (r *Rudp) outPut() *Package {
  217. var tmp packageBuffer
  218. r.reqMissing(&tmp)
  219. r.replyRequest(&tmp)
  220. r.sendMessage(&tmp)
  221. if tmp.head == nil && tmp.tmp.Len() == 0 {
  222. tmp.tmp.WriteByte(byte(TYPE_PING))
  223. }
  224. tmp.newPackage()
  225. return tmp.head
  226. }
  227. func (r *Rudp) Input(bts []byte) {
  228. sz := len(bts)
  229. if sz > 0 {
  230. r.lastRecvTick = r.currentTick
  231. }
  232. for sz > 0 {
  233. length := int(bts[0])
  234. if length > 127 {
  235. if sz <= 1 {
  236. r.corrupt.Store(ERROR_MSG_SIZE)
  237. return
  238. }
  239. length = (length*256 + int(bts[1])) & 0x7fff
  240. bts = bts[2:]
  241. sz -= 2
  242. } else {
  243. bts = bts[1:]
  244. sz -= 1
  245. }
  246. switch length {
  247. case TYPE_PING:
  248. r.checkMissing(false)
  249. case TYPE_EOF:
  250. r.corrupt.Store(ERROR_EOF)
  251. case TYPE_CORRUPT:
  252. r.corrupt.Store(ERROR_REMOTE_EOF)
  253. return
  254. case TYPE_REQUEST, TYPE_MISSING:
  255. if sz < 4 {
  256. r.corrupt.Store(ERROR_MSG_SIZE)
  257. return
  258. }
  259. exe := r.addRequest
  260. max := r.sendID
  261. if length == TYPE_MISSING {
  262. exe = r.addMissing
  263. max = r.recvIDMax
  264. }
  265. // this eliminates multiple BCs in the exe function invocation
  266. _ = bts[3]
  267. exe(r.getID(max, bts[0], bts[1]), r.getID(max, bts[2], bts[3]))
  268. bts = bts[4:]
  269. sz -= 4
  270. default:
  271. length -= TYPE_NORMAL
  272. if sz < length+2 {
  273. r.corrupt.Store(ERROR_MSG_SIZE)
  274. return
  275. }
  276. r.insertMessage(r.getID(r.recvIDMax, bts[0], bts[1]), bts[2:length+2])
  277. bts = bts[length+2:]
  278. sz -= length + 2
  279. }
  280. }
  281. r.checkMissing(false)
  282. }
  283. func (r *Rudp) checkMissing(direct bool) {
  284. head := r.recvQueue.head
  285. if head != nil && head.id > r.recvIDMin {
  286. nano := int(time.Now().UnixNano())
  287. last := r.recvSkip[r.recvIDMin]
  288. if !direct && last == 0 {
  289. r.recvSkip[r.recvIDMin] = nano
  290. dbg("miss start %v-%v,max %v", r.recvIDMin, head.id-1, r.recvIDMax)
  291. } else if direct || last+missingTime < nano {
  292. delete(r.recvSkip, r.recvIDMin)
  293. r.reqSendAgain <- [2]int{r.recvIDMin, head.id - 1}
  294. dbg("req miss %v-%v,direct %v,wait num %v",
  295. r.recvIDMin, head.id-1, direct, r.recvQueue.num)
  296. }
  297. }
  298. }
  299. func (r *Rudp) insertMessage(id int, bts []byte) {
  300. if id < r.recvIDMin {
  301. dbg("already recv %v,len %v", id, len(bts))
  302. return
  303. }
  304. delete(r.recvSkip, id)
  305. if id > r.recvIDMax || r.recvQueue.head == nil {
  306. m := &message{}
  307. m.buf.Write(bts)
  308. m.id = id
  309. r.recvQueue.push(m)
  310. r.recvIDMax = id
  311. } else {
  312. m := r.recvQueue.head
  313. last := &r.recvQueue.head
  314. for m != nil {
  315. if m.id == id {
  316. dbg("repeat recv id %v,len %v", id, len(bts))
  317. } else if m.id > id {
  318. tmp := &message{}
  319. tmp.buf.Write(bts)
  320. tmp.id = id
  321. tmp.next = m
  322. *last = tmp
  323. r.recvQueue.num++
  324. return
  325. }
  326. last = &m.next
  327. m = m.next
  328. }
  329. }
  330. }
  331. func (r *Rudp) sendMessage(tmp *packageBuffer) {
  332. m := r.sendQueue.head
  333. for m != nil {
  334. tmp.packMessage(m)
  335. m = m.next
  336. }
  337. if r.sendQueue.head != nil {
  338. if r.sendHistory.tail == nil {
  339. r.sendHistory = r.sendQueue
  340. } else {
  341. r.sendHistory.tail.next = r.sendQueue.head
  342. r.sendHistory.tail = r.sendQueue.tail
  343. }
  344. r.sendQueue.head = nil
  345. r.sendQueue.tail = nil
  346. }
  347. }
  348. func (r *Rudp) clearSendExpired() {
  349. m := r.sendHistory.head
  350. for m != nil {
  351. if m.tick >= r.lastExpiredTick {
  352. break
  353. }
  354. m = m.next
  355. }
  356. r.sendHistory.head = m
  357. if m == nil {
  358. r.sendHistory.tail = nil
  359. }
  360. }
  361. func (r *Rudp) addRequest(min, max int) {
  362. dbg("add request %v-%v,max send id %v", min, max, r.sendID)
  363. r.addSendAgain <- [2]int{min, max}
  364. }
  365. func (r *Rudp) addMissing(min, max int) {
  366. if max < r.recvIDMin {
  367. dbg("add missing %v-%v fail,already recv,min %v", min, max, r.recvIDMin)
  368. return
  369. }
  370. if min > r.recvIDMin {
  371. dbg("add missing %v-%v fail, more than min %v", min, max, r.recvIDMin)
  372. return
  373. }
  374. head := 0
  375. if r.recvQueue.head != nil {
  376. head = r.recvQueue.head.id
  377. }
  378. dbg("add missing %v-%v,min %v,head %v", min, max, r.recvIDMin, head)
  379. r.recvIDMin = max + 1
  380. r.checkMissing(true)
  381. }
  382. func (r *Rudp) replyRequest(tmp *packageBuffer) {
  383. for {
  384. select {
  385. case again := <-r.addSendAgain:
  386. history := r.sendHistory.head
  387. min, max := again[0], again[1]
  388. if history == nil || max < history.id {
  389. dbg("send again miss %v-%v,send max %v", min, max, r.sendID)
  390. tmp.packRequest(min, max, TYPE_MISSING)
  391. } else {
  392. var start, end, num int
  393. for {
  394. if history == nil || max < history.id {
  395. //expired
  396. break
  397. } else if min <= history.id {
  398. tmp.packMessage(history)
  399. if start == 0 {
  400. start = history.id
  401. }
  402. end = history.id
  403. num++
  404. }
  405. history = history.next
  406. }
  407. if min < start {
  408. tmp.packRequest(min, start-1, TYPE_MISSING)
  409. dbg("send again miss %v-%v,send max %v", min, start-1, r.sendID)
  410. }
  411. dbg("send again %v-%v of %v-%v,all %v,max send id %v", start, end, min, max, num, r.sendID)
  412. }
  413. default:
  414. return
  415. }
  416. }
  417. }
  418. func (r *Rudp) reqMissing(tmp *packageBuffer) {
  419. for {
  420. select {
  421. case req := <-r.reqSendAgain:
  422. tmp.packRequest(req[0], req[1], TYPE_REQUEST)
  423. default:
  424. return
  425. }
  426. }
  427. }