monitor: Introduce channel to buffer notifications and listeners #2933
Conversation
tgraf
added
kind/bug
tgraf
force-pushed
the
buffer-monitor
branch
from
9b5f617
to
833f26e
Compare
No functional change of code Signed-off-by: Thomas Graf <thomas@cilium.io>
tgraf
force-pushed
the
buffer-monitor
branch
from
833f26e
to
20d6702
Compare
monitor/launch/launcher.go
Outdated
| @@ -96,3 +121,76 @@ func (nm *NodeMonitor) setState(state *models.MonitorStatus) { | |||
| nm.state = state | |||
| nm.Mutex.Unlock() | |||
| } | |||
|
|
|||
| func (nm *NodeMonitor) SendEvent(typ int, event interface{}) error { | |||
There was a problem hiding this comment.
exported method NodeMonitor.SendEvent should have comment or be unexported
* Makes SendEvent() lockless and non-blocking. Notifications are enqueued to a channel and in case the channel is full, the notification is dropped. * Lost notifications are accounted for and reported * The writing to the pipe is made via a single Write() call to maximise the chances that either none of the message buffer or all of it is written to the pipe. Signed-off-by: Thomas Graf <thomas@cilium.io>
Signed-off-by: Thomas Graf <thomas@cilium.io>
This decouples multiple readers and avoids blocking other readers if one of the readers can't keep it up. Signed-off-by: Thomas Graf <thomas@cilium.io>
tgraf
force-pushed
the
buffer-monitor
branch
from
20d6702
to
2061e70
Compare
|
test-me-please |
monitor/launch/launcher.go
Outdated
| return fmt.Errorf("Unable to encode metadata: %s", err) | ||
| } | ||
|
|
||
| msgBuf := append(metaBuf, payloadBuf...) |
There was a problem hiding this comment.
Can you add a comment into this code? I'm failing to understand why you are appending payloadBuf to metaBuf
There was a problem hiding this comment.
The meta header comes before the payload. I'll move it into a func.
|
|
||
| var ( | ||
| mutex lock.Mutex | ||
| listeners = list.New() | ||
| listeners = make(map[*monitorListener]struct{}) |
There was a problem hiding this comment.
Can't you use map[monitorListener]chan []byte instead?
There was a problem hiding this comment.
What's the advantage of that? It's not extendable.
There was a problem hiding this comment.
I was trying to solve the problem of using a pointer as a key of a map. That can give problems when doing something like
func Foo(m monitorListener){
_, exists := listeners[&m]
// exists will always be false
}
There was a problem hiding this comment.
Sure, because it's a different address.
Signed-off-by: Thomas Graf <thomas@cilium.io>
|
test-me-please |
| } | ||
|
|
||
| select { | ||
| case nm.queue <- append([]byte{byte(typ)}, buf.Bytes()...): |
There was a problem hiding this comment.
Isn't typ already in the event object? In either case, doesn't this break the API we're presenting current clients?
There was a problem hiding this comment.
Oh, nevermind. This typ is different than the one you replaced.
The current monitor code has two drawbacks:
sendEvent()blocks until the notification has been read, this can block fast path operations such as policy regeneration or even processing of L7 requests.Changes:
to a channel and in case the channel is full, the notification is dropped.
chances that either none of the message buffer or all of it is written to
the pipe.