fix(udp_task_pool): panic: close of closed channel (#570)

control/udp_task_pool.go

@@ -10,6 +10,7 @@ const UdpTaskQueueLength = 128
 type UdpTask = func()
 
 type UdpTaskQueue struct {
+	key       string
 	ch        chan UdpTask
 	timer     *time.Timer
 	agingTime time.Duration
@@ -47,7 +48,9 @@ func (p *UdpTaskPool) convoy(q *UdpTaskQueue) {
 		clearloop:
 			for {
 				select {
-				case <-q.ch:
+				case t := <-q.ch:
+					// Emit it back due to closed q.
+					p.EmitTask(q.key, t)
 				default:
 					break clearloop
 				}
@@ -66,6 +69,7 @@ func (p *UdpTaskPool) EmitTask(key string, task UdpTask) {
 	if !ok {
 		ch := p.queueChPool.Get().(chan UdpTask)
 		q = &UdpTaskQueue{
+			key:       key,
 			ch:        ch,
 			timer:     nil,
 			agingTime: DefaultNatTimeout,
@@ -73,6 +77,12 @@ func (p *UdpTaskPool) EmitTask(key string, task UdpTask) {
 			freed:     make(chan struct{}),
 		}
 		q.timer = time.AfterFunc(q.agingTime, func() {
+			// This func may be invoked twice due to concurrent Reset.
+			select {
+			case <-q.closed:
+				return
+			default:
+			}
 			p.mu.Lock()
 			defer p.mu.Unlock()
 			if p.m[key] == q {