26

The short version: Is there a way to empty a go channel without recreating it, or looping through it?

The why: I'm using two channels to send and receive data, and I have an extra channel to signal that a reconnect is needed.

Now when the transport has been reset/reconnected I want to 'empty' the extra channel to make sure that there is not any lingering other reset requests which would cause the thing to reconnect again.

5
  • possible duplicate of periodically flushing channel in golang
    – Floegipoky
    Oct 1, 2014 at 14:07
  • Can you be more specific about your use case? Specifically about what you're reconnecting to and why. Without all the details it sounds to me like the real issue that needs to be addressed is putting junk data into your channel, not removing it after-the-fact.
    – Floegipoky
    Oct 1, 2014 at 14:12
  • @Floegipoky: The 2 channels is to send and receive data over a socket where the data is parsed in a different part of the code which handles the protocol. In case the protocol finds errors I want to be able to signal the transport layer that a reconnect is needed. Thats why I needed the extra channel
    – Toad
    Oct 1, 2014 at 14:29
  • 1
    @dewy broto: I thought it would be nice to have a dataflow going where every step is a bit higher level than the next. The lowest is the transport, then the next in chain is the actual protocol, and one step further might be something acting upon the protocol. That way you can swap out parts without the other bits knowing about it (e.g. removing the transport and sending known byte sequences for testing)
    – Toad
    Oct 1, 2014 at 14:54
  • @Toad your coworkers will thank you for separating out unrelated logic. Socket IO does not belong with protocol-specific message parsing.
    – Floegipoky
    Oct 1, 2014 at 15:34

5 Answers 5

41

It is not possible to empty a channel without a loop. If you don't have any concurrent receivers, then you can use this simple loop:

for len(ch) > 0 {
  <-ch
}

If you do have concurrent receivers, then use the loop:

L:
for {
    select {
    case <-c:
    default:
       break L
    }
}
2
  • 1
    I'd think the risk of doing it this way is that if someone else is also reading that channel, it might block. (In my case there is not an extra reader but it is a caveat worth noting)
    – Toad
    Oct 1, 2014 at 14:27
  • if chan closed, this got infinite loop
    – yurenchen
    Sep 18, 2022 at 21:14
7

What you're describing is inherently racy, since there may be legitimate requests to reconnect on the channel. Rather than trying to drain the channel, I would recommend keeping track of timing.

On your reconnect channel, post the time. When finished with the reconnect, note the time. While consuming the reconnect channel, throw away any messages that are older than your last reconnect.

Another more lockstep solution to achieve this is to make the reconnect channel a bool. Post "true" to reconnect. When the reconnect completes, post "false". Then consume the channel until you find "false."

1
  • 2
    I like the timestamp approach. It's still cycling through the channel, but in this case you are right that at least I can't accidentally remove any newer requests
    – Toad
    Oct 1, 2014 at 14:26
2

Another approach is using sync.Cond and atomic, something along the lines of:

type Server struct {
    s     chan int
    r     chan int
    c     *sync.Cond
    state uint32
}

const (
    sNormal       = 0
    sQuitting     = 1
    sReconnecting = 2
)

func New() *Server {
    s := &Server{
        s: make(chan int),
        r: make(chan int),
        c: sync.NewCond(&sync.Mutex{}),
    }
    go s.sender()
    // go s.receiver()
    return s
}
func (s *Server) sender() {
    //
    for {
        select {
        case data := <-s.s:
        //do stuff with data
        default:
            s.c.L.Lock()
        L:
            for {
                switch atomic.LoadUint32(&s.state) {
                case sNormal:
                    break L
                case sReconnecting:
                case sQuitting:
                    s.c.L.Unlock()
                    return
                }
                s.c.Wait()
            }
            s.c.L.Unlock()
        }
    }
}

//repeat for receiver

func (s *Server) Reconnect() {
    var cannotReconnect bool
    atomic.StoreUint32(&s.state, sReconnecting)
    //keep trying to reconnect
    if cannotReconnect {
        atomic.StoreUint32(&s.state, sQuitting)
    } else {
        atomic.StoreUint32(&s.state, sNormal)
    }
    s.c.Broadcast()
}

playground

1
  • @Toad yes but you don't really have that many options, you have to use some sort of locking / atomic mechanism when dealing with concurrency¸this is the only way I can think of without adding an extra channel like Floegipoky suggested. Also channels use their own locks, so you won't gain much doingthat.
    – OneOfOne
    Oct 8, 2014 at 10:46
2

can't edit Simon Fox 's answer, so write here:

  • if chan closed, that will got infinite loop:
    cause <-ch immediately return err.

More safer version (handle closed chan):

L:
    for {
        select {
        case _, ok := <-ch:
            if !ok { //ch is closed //immediately return err
                break L
            }
        default: //all other case not-ready: means nothing in ch for now
            break L
        }
    }
0

It sounds like instead of a reset channel, you want a reset goroutine. It would have an input from the side sending the reset signal, and an output to the receiver. When this goroutine receives a request to reconnect, it passes it to the receiver. Then it waits to receive an acknowledgement back from the receiver on a third channel, throwing away any reconnect requests it receives in the meantime. So 3 channels total, 1 input, 1 output, 1 ack.

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.