Buffered forwarder for #15 #50

laher · 2017-03-27T11:20:33Z

@peterbourgon please see this experimental implementation for #15 , although it's a bit racey for the time being. Please don't consider this complete yet.

It basically works - when I kill my ingeststore the forwarder buffers some messages, and then after restarting the ingeststore it reconnects & forwards the buffer 👍 ... but sometimes a couple of messages get sent twice - it needs work still.
I've been testing it by piping date once per second onto oklog forward -buf -bufsize=5 localhost, then repeatedly querying oklog query -from 30s in another window, and then killing/restarting my ingeststore for 5-15 seconds. It's pretty easy to see what's going on.
For the buffer I tried using container/ring from the standard library. I added a mutex and a couple of other fields to maintain state. Seems OKish, but not very simple really. Any advice on how you envisioned this?
I chose messages as the unit for buffering, rather than bytes. I figured it fits the forwarding code reasonably well.

It's late now so I just thought I'd elicit some feedback for now, as you may have had something very different in mind.
Cheers

… bit racey.

laher · 2017-03-29T03:51:15Z

OK did a bit more this morning, and I'm happier with the approach now - some work still needed, but it's working nicely and the approach feels more sane. The ring buffer now implements a textScanner interface, to act as a drop-in replacement for the bufio.Scanner.

I'll report back once I've got it 100%.

peterbourgon

Some relatively straightforward changes to the calling scope, but I think the ringbuf is not usable as-is.

peterbourgon · 2017-03-31T08:34:22Z

cmd/oklog/forward.go

+		apiAddr     = flagset.String("api", "", "listen address for forward API (and metrics)")
+		prefixes    = stringslice{}
+		ringBuf     = flagset.Bool("buf", false, "use a ring buffer to drop old messages when e.g. the ingestion engine fails")
+		ringBufSize = flagset.Int("bufsize", 1024, "ring buffer size (messages)")


I would want the flag to describe the forwarder backpressure strategy (the user intent) rather than a lower-level detail like enabling a ringbuffer (the implementation detail). How about...

backpressure = flagset.String("backpressure", "block", "block, buffer") bufferSize = flagset.Int("buffer-size", 1024, "in -backpressure=buffer mode, ringbuffer size in records")

And it's important to validate the string when you use it down below, e.g.

var s textScanner switch strings.ToLower(*backpressure) { case "block": s = bufio.NewScanner(os.Stdin) case "buffer": // e.g. from below rb := newRingBuffer(*bufferSize) go rb.Consume(os.Stdin) s = rb default: level.Error(logger).Log("backpressure", *backpressure, "err", "invalid backpressure option") os.Exit(1) }

Yup, sure. Sounds like you'd like to keep options open for other alternatives in future.

peterbourgon · 2017-03-31T08:37:30Z

cmd/oklog/forward_buffer.go

+	Scan() bool
+	Text() string
+	Err() error
+}


Please define this at the point of use, i.e. in func main.

Also, please comment that e.g.

// textScanner models bufio.Scanner, so we can provide // an alternate ringbuffered implementation.

peterbourgon · 2017-03-31T08:39:11Z

cmd/oklog/forward.go

+					level.Info(logger).Log("stdin", "exhausted", "due_to", bufioScanner.Err())
+				}
+			}
+		}()


What do you think about making this code a method on the ringBuffer?

So the callsite would look like

rb := newRingBuffer(*bufferSize) go rb.Consume(os.Stdin) s = rb

peterbourgon · 2017-03-31T16:54:20Z

cmd/oklog/forward.go

+		rb := newRingBuffer(*ringBufSize)
+		go func() {
+			for {
+				bufioScanner := bufio.NewScanner(os.Stdin)


bs := bufio.NewScanner(os.Stdin)

peterbourgon · 2017-03-31T16:54:55Z

cmd/oklog/forward.go

+					ok = bufioScanner.Scan()
+				}
+				if !ok {
+					level.Info(logger).Log("stdin", "exhausted", "due_to", bufioScanner.Err())


I guess we should use this error from stdin to shut down the ringbuffer somehow. That shutdown signal should propagate thru to terminate the existing for ok loop.

peterbourgon · 2017-03-31T17:11:26Z

cmd/oklog/forward_buffer.go

+
+func (bf *ringBuffer) Scan() bool {
+	return true
+}


As noted above, we're gonna want to propagate a shutdown signal thru somehow to terminate this.

peterbourgon · 2017-03-31T17:12:58Z

cmd/oklog/forward_buffer.go

+	bf.r = bf.r.Next()
+	bf.inc(-1)
+	return v
+}


Unless I'm really misunderstanding something, this method has a number of problems…

It's burning the CPU calling bf.Remaining over and over

Once that loop exits, the bf.r.Value is not guaranteed to be there, another goroutine calling Get could have stolen the work item — a data race

bf.r is written without synchronization — a data race

I think using a remaining counter as the synchronization point is not really viable, at least not as implemented here...

Agreed. It's not great. I'm not loving container/ring TBH. I'd like to grok it better with some more investigation into the Link/Unlink methods, but I'm also thinking about alternatives.

I'd like to try a buffered channel implementation (2 channels, similar to this https://content.pivotal.io/blog/a-channel-based-ring-buffer-in-go ). It would be safe from races but I'm not sure how performance would compare to locking approaches. I guess a benchmark would help confirm or otherwise.

Check the comments on that Pivotal buffer, it's racy as printed.

OK, I'll see if I can make container/ring work without the counter and then look at other options from there. Cheers

…needs fixing.

laher · 2017-03-31T20:22:26Z

I've applied the straightforward changes, thanks for your guidance. I'll work on the ring buffer itself in due course and add appropriate tests/benchmarks. Cheers

… slice, not a ring buffer

laher · 2017-04-02T09:58:31Z

@peterbourgon I've rewritten the buffer now, so that Get() blocks on a channel whilst data is unavailable.

So, I tried starting from scratch: 2 new struct types, composed together with an interface, and backed by a slice and a channel.

bufferedScanner is forwarding messages to a 'buffer' type, implementing some methods which are shared with io.Scanner.
sliceBuffer is the only implementation that I've added.
sliceBuffer.Get() blocks on a channel whenever the buffer is empty, and no data is available.
sliceBuffer.Put() attempts to deliver messages on the channel. If the channel is not being listened to, Put adds the messages onto a bounded buffer (a slice). Once the maximum size is reached, further messages are discarded.

Hoping this change is closer to the mark.
Thanks

laher · 2017-04-02T21:20:44Z

cmd/oklog/forward_slicebuffer.go

+	}
+	b.mutex.RUnlock()
+	b.mutex.Lock()
+	record, b.buf = b.buf[0], b.buf[1:]


This could be expensive in terms of GC.

Happy to switch out the slice for a container/list.List or ring.Ring instead. Incidentally I was experimenting with a list.List, found with a benchmark that the slice was marginally faster, but maybe GC is more important.

Hmm, I think maybe the best bet is to keep it as a slice but to use it like a ring-buffer to reduce the number of allocations. I'll have a play with it and try some -memprofile analysis

laher · 2017-04-03T09:02:25Z

cmd/oklog/forward_slicebuffer.go

+func (b *sliceBuffer) Put(record string) {
+	select {
+	case b.ch <- record:
+	default:


I've used the select/default on the send side only, for one-way blocking. We always want Get to wait for data, but we want Put to buffer instead.

…dd them to a package in 'pkg')

laher · 2017-04-15T12:31:30Z

Please see #62 for a proposed direction I'd like to take this feature.

(If 62 gets rejected, perhaps we can still come back to this)

…iven test

…esn't have any locks to worry about

laher · 2017-04-28T11:43:58Z

OK I think I'm done with this buffered-channel implementation of the ring buffer. I think it's nice and simple.

I fancy you'll have some more thoughts on the bufferedScanner. I have amended the API a little to make it a little simpler IMO. Just one call to Next() instead of Scan()/Text()/Error(). It makes more sense to me.

Still, I'm good for now. I'll go over and close the other PR. Cheers.

laher · 2017-05-23T21:40:27Z

@peterbourgon have you had a chance to look at the most recent implementation with the buffered channel?

Ta

peterbourgon · 2017-05-25T12:15:17Z

Not yet, sorry! I will find some time on the weekend 🤞

laher and others added 4 commits March 27, 2017 23:48

Buffered forwarder, using container.Ring(). Initial implementation, A…

351f3ce

… bit racey.

rename cancel() to stop()

b3d6ea7

v2 - implementing a textScanner interface

1371768

Brute force it to work for my integration test with a time.Sleep()

32a71cc

laher added 2 commits March 29, 2017 20:43

bugfix on scanner

d62d5dc

Always return true from 'Scan()'

5532f4c

peterbourgon suggested changes Mar 31, 2017

View reviewed changes

laher added 3 commits April 1, 2017 09:13

Just the straightforward changes from Code Review. Ring buffer still …

f3be503

…needs fixing.

Merge upstream/master

9f78fda

Merge upstream/master

175ae64

New bufferedScanner blocks when awaiting data. Inner buffer is just a…

a1811ed

… slice, not a ring buffer

laher commented Apr 3, 2017

View reviewed changes

laher added 2 commits April 3, 2017 21:36

Use 2 integers to mark the start and end of the slice-based ring buffer

310006b

Tidy up and doc comments for ring-buffer behaviour

d0a29be

laher changed the title ~~Buffered forwarder, using container.Ring() for #15~~ Buffered forwarder for #15 Apr 3, 2017

laher and others added 4 commits April 4, 2017 08:40

Rename sliceBuffer -> RingBuffer. Make both types public (Hoping to a…

54b1410

…dd them to a package in 'pkg')

Simplify synchronization to avoid some races

5e5a92d

Merge branch 'master' into feature/buffered-forwarder

cb9ecca

Move types to pkg/forward. Improve tests and benchmarks for RingBuffer

6f61cb1

laher changed the title ~~Buffered forwarder for #15~~ [WIP] Buffered forwarder for #15 Apr 15, 2017

laher mentioned this pull request Apr 15, 2017

Enable programmatic forwarding from Go apps (relates to #15 and #50) #62

Closed

laher added 5 commits April 16, 2017 20:19

Remove 'drop' behaviour and fix related bug in buffer. Add a table-dr…

59b2b1a

…iven test

Remove 'drop' behaviour and fix related bug in buffer. Add a table-dr…

474540d

…iven test

Fix receive-during-lock. Remove print statements

10140a6

No longer skip the buffer when Get is awaiting

3086753

Remove print statement

3480e64

laher added 5 commits April 16, 2017 21:37

correct a log statement (function not variable)

31d80e1

RingBufferBCH: Buffered-channel implementation works just fine and do…

e89ba30

…esn't have any locks to worry about

Add appropriate comments to RingBufferBCH

bd4e75c

forward: refactor to exclusively use channel-based implementation

8145707

forward: refactor the forward package. Expose fewer types.

f98e91b

laher changed the title ~~[WIP] Buffered forwarder for #15~~ Buffered forwarder for #15 Apr 28, 2017

laher added a commit to laher/oklog that referenced this pull request Apr 28, 2017

forward: fix conflicts and accommodate ringbuffer changes from oklog#50

8670805

peterbourgon closed this Mar 28, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Buffered forwarder for #15 #50

Buffered forwarder for #15 #50

laher commented Mar 27, 2017

laher commented Mar 29, 2017

peterbourgon left a comment

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017 •

edited

laher Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

peterbourgon Mar 31, 2017

laher Mar 31, 2017

peterbourgon Mar 31, 2017

laher Mar 31, 2017

laher commented Mar 31, 2017

laher commented Apr 2, 2017

laher Apr 2, 2017

laher Apr 3, 2017

laher Apr 3, 2017

laher commented Apr 15, 2017

laher commented Apr 28, 2017

laher commented May 23, 2017

peterbourgon commented May 25, 2017

Buffered forwarder for #15 #50

Buffered forwarder for #15 #50

Conversation

laher commented Mar 27, 2017

laher commented Mar 29, 2017

peterbourgon left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

peterbourgon Mar 31, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

laher commented Mar 31, 2017

laher commented Apr 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

laher commented Apr 15, 2017

laher commented Apr 28, 2017

laher commented May 23, 2017

peterbourgon commented May 25, 2017

peterbourgon Mar 31, 2017 •

edited