0
\$\begingroup\$

I'm trying to just write logs out every N seconds or when the buffer gets too big, so that I don't send external log requests too large and that I limit data loss to at most N seconds. For the life of me, I cannot seem to find any sample code to do this. Most of the examples like in this question either have a timer but don't have a max size, or have a max size but allow for complete data loss (no timer).

I know that I can use timer.Tick to "timeout" on filling the buffer, but:

  • Do I just use a slice to implement the buffer? I feel like I'm re-inventing buffered channels then
  • There shouldn't be any race condition, right? e.g. Between the timer.Tick event and the "buffer capacity reached"?

This is my code but it feels very wrong:

package main

import (
    "fmt"
    "time"
)

func populateChannel(outp chan int) {
    i := 0

    // emit an item every second
    // but 10th item, dump out 200 items fast
    timerChan := time.Tick(1 * time.Second)
    for _ = range timerChan {
        count := 1
        if i%10 == 0 {
            count = 200 + 1
        }
        for j := 1; j <= count; j++ {
            outp <- i
            i++
        }
    }
}

func processBatches(inp chan []int) {
    for batch := range inp {
        fmt.Printf("Got new batch of len=%d, cap=%d: %v\n", len(batch), cap(batch), batch)
    }
}

func main() {
    maxBatchSize := 100

    // I feel like this totally defeats the purpose of channels
    currentBatch := make([]int, 0, maxBatchSize)

    unbufChan := make(chan int)
    go populateChannel(unbufChan)

    timerChan := time.Tick(5 * time.Second)
    batchedChan := make(chan []int, 1)

    go processBatches(batchedChan)

    for {
        select {
        case item := <-unbufChan:
            currentBatch = append(currentBatch, item)
            fmt.Printf("Got item %v. Batch size = %d\n", item, len(currentBatch))
            if len(currentBatch) >= maxBatchSize {
                fmt.Println("Max batch size reached. New batch")
                // TODO: refactor
                batch := currentBatch // what if the timer case hits at this exact moment? Could it even?
                currentBatch = make([]int, 0, maxBatchSize)
                batchedChan <- batch
            }
        case <-timerChan:
            fmt.Println("Max time reached. Current batch too stale. New batch")
            // TODO: refactor
            batch := currentBatch
            currentBatch = make([]int, 0, maxBatchSize)
            batchedChan <- batch
        }
    }

}
\$\endgroup\$
  • \$\begingroup\$ "This is my code but it feels very wrong:" Does it work as intended? \$\endgroup\$ – Mast Dec 19 '17 at 5:31
  • \$\begingroup\$ It works but I'm worried that I'm not using channels properly, and that I may have a race condition. \$\endgroup\$ – Bob Dec 19 '17 at 15:34
  • \$\begingroup\$ There is no race condition in your code. You can easily check this with the race-detector tool (see details here. To use it, just run go build -race or go run -race main.go \$\endgroup\$ – felix Jan 10 '18 at 7:01
1
\$\begingroup\$

Do I just use a slice to implement the buffer? I feel like I'm re-inventing buffered channels then

A buffered channel is just a way to allow goroutines to be more independent when they are producing/consuming a channel.

There shouldn't be any race condition, right? e.g. Between the timer.Tick event and the "buffer capacity reached"?

No, because the code you are talking about is not concurrent (the select statement will run only one of the two loop).


Is it expected, that even if some batches were sent in the meantime, a new batch will be send at 5 seconds? Shouldn't the timer be reset, when a batch is sent?

Anyway, I would recommend you to put this ingester in a struct, for instance like this:

type ingester struct {
    MaxSize  int
    MaxDelay time.Duration
}

func (ing ingester) Forward(in <-chan int, out chan<- []int) {
    defer close(out)
    buf := make([]int, 0, ing.MaxSize)

    deadline := time.After(ing.MaxDelay)

    sendBufferAndResetDeadline := func() {
        out <- buf
        buf = make([]int, 0, ing.MaxSize)
        deadline = time.After(ing.MaxDelay)
    }

    for {
        select {
        case i, ok := <-in:
            if !ok {
                if len(buf) > 0 {
                    sendBufferAndResetDeadline()
                }
                return
            }
            buf = append(buf, i)
            if len(buf) >= ing.MaxSize {
                sendBufferAndResetDeadline()
            }
        case <-deadline:
            sendBufferAndResetDeadline()
        }
    }
}

Your main would then look like this:

func main() {
    maxBatchSize := 100

    // I feel like this totally defeats the purpose of channels
    currentBatch := make([]int, 0, maxBatchSize)

    unbufChan := make(chan int)
    go populateChannel(unbufChan)

    batchedChan := make(chan []int, 1)

    go processBatches(batchedChan)

    ing := ingester{
        MaxSize:  maxBatchSize,
        MaxDelay: 5 * time.Second,
    }

    ing.Forward(unbufChan, batchedChan)
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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