send_queue.go 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  1. package quic
  2. type sender interface {
  3. Send(p *packetBuffer)
  4. Run() error
  5. WouldBlock() bool
  6. Available() <-chan struct{}
  7. Close()
  8. }
  9. type sendQueue struct {
  10. queue chan *packetBuffer
  11. closeCalled chan struct{} // runStopped when Close() is called
  12. runStopped chan struct{} // runStopped when the run loop returns
  13. available chan struct{}
  14. conn sendConn
  15. }
  16. var _ sender = &sendQueue{}
  17. const sendQueueCapacity = 8
  18. func newSendQueue(conn sendConn) sender {
  19. return &sendQueue{
  20. conn: conn,
  21. runStopped: make(chan struct{}),
  22. closeCalled: make(chan struct{}),
  23. available: make(chan struct{}, 1),
  24. queue: make(chan *packetBuffer, sendQueueCapacity),
  25. }
  26. }
  27. // Send sends out a packet. It's guaranteed to not block.
  28. // Callers need to make sure that there's actually space in the send queue by calling WouldBlock.
  29. // Otherwise Send will panic.
  30. func (h *sendQueue) Send(p *packetBuffer) {
  31. select {
  32. case h.queue <- p:
  33. case <-h.runStopped:
  34. default:
  35. panic("sendQueue.Send would have blocked")
  36. }
  37. }
  38. func (h *sendQueue) WouldBlock() bool {
  39. return len(h.queue) == sendQueueCapacity
  40. }
  41. func (h *sendQueue) Available() <-chan struct{} {
  42. return h.available
  43. }
  44. func (h *sendQueue) Run() error {
  45. defer close(h.runStopped)
  46. var shouldClose bool
  47. for {
  48. if shouldClose && len(h.queue) == 0 {
  49. return nil
  50. }
  51. select {
  52. case <-h.closeCalled:
  53. h.closeCalled = nil // prevent this case from being selected again
  54. // make sure that all queued packets are actually sent out
  55. shouldClose = true
  56. case p := <-h.queue:
  57. if err := h.conn.Write(p.Data); err != nil {
  58. return err
  59. }
  60. p.Release()
  61. select {
  62. case h.available <- struct{}{}:
  63. default:
  64. }
  65. }
  66. }
  67. }
  68. func (h *sendQueue) Close() {
  69. close(h.closeCalled)
  70. // wait until the run loop returned
  71. <-h.runStopped
  72. }