udp/sendqueue.go

   1 package udpapi
   2
   3 import (
   4         "time"
   5 )
   6
   7 type enqueuedPacket struct {
   8         packet
   9         queue chan packet
  10 }
  11
  12 type sendQueueState struct {
  13         enqueue chan enqueuedPacket
  14 }
  15
  16 var globalQueue sendQueueState
  17
  18 func init() {
  19         globalQueue = sendQueueState{
  20                 enqueue: make(chan enqueuedPacket, 10),
  21         }
  22         go globalQueue.sendQueueDispatch()
  23 }
  24
  25 const (
  26         throttleMinDuration = 2 * time.Second
  27         throttleMaxDuration = 4 * time.Second
  28         throttleIncFactor   = 1.1
  29         throttleDecFactor   = 0.9
  30         throttleDecInterval = 10 * time.Second
  31 )
  32
  33 func sendPacket(p packet, c chan packet) chan bool {
  34         p.sent = make(chan bool, 2)
  35         globalQueue.enqueue <- enqueuedPacket{packet: p, queue: c}
  36         return p.sent
  37 }
  38
  39 func (gq *sendQueueState) sendQueueDispatch() {
  40         pkt := (*enqueuedPacket)(nil)
  41         queue := make([]enqueuedPacket, 0)
  42
  43         nextTimer := time.NewTimer(0)
  44         decTimer := time.NewTimer(0)
  45
  46         currentThrottle := throttleMinDuration
  47
  48         for {
  49                 if pkt == nil && len(queue) > 0 {
  50                         pkt = &queue[0]
  51                         queue = queue[1:]
  52                 }
  53
  54                 nextCh := nextTimer.C
  55                 decCh := decTimer.C
  56
  57                 if pkt == nil {
  58                         nextCh = nil
  59                 }
  60
  61                 select {
  62                 case p := <-gq.enqueue:
  63                         queue = append(queue, p)
  64                 case <-nextCh:
  65                         pkt.queue <- pkt.packet
  66                         <-pkt.packet.sent
  67
  68                         pkt = nil
  69
  70                         currentThrottle = time.Duration(float64(currentThrottle) * throttleIncFactor)
  71                         if currentThrottle > throttleMaxDuration {
  72                                 currentThrottle = throttleMaxDuration
  73                         }
  74                         nextTimer.Reset(currentThrottle)
  75
  76                         decTimer.Reset(throttleDecInterval)
  77                 case <-decCh:
  78                         currentThrottle = time.Duration(float64(currentThrottle) * throttleDecFactor)
  79                         if currentThrottle < throttleMinDuration {
  80                                 currentThrottle = throttleMinDuration
  81                         } else {
  82                                 decTimer.Reset(throttleDecInterval)
  83                         }
  84                 }
  85         }
  86 }