I am trying to achieve “It allows latency-sensitive programs to run in a hot-loop pinned to a thread on an isolated core in order to achieve low latency and jitter.”.
Usually I do this by calling LockOSThread and SetSchedulerAffinity (single CPU), but for some reason this introduces significantly latency spikes (>10ms) compared to running it without pinning it. go trace confirms that the Runner is locked to a single thread. Message handlers are not doing a lot more than setting on a poller diode.
Nothing changes when I:
- don't call SetSchedulerAffinity
- remove the core from
isolcpus
- remove the core from
nohz_full
- remove the core from
rcu_nocbs
What is the recommended approach to do this?
type Runner struct {
commandsDiode diodes.Diode
pollingEnabled bool
cpuAffinityManager *cpuaffinity.Manager
l *logger.Logger
}
func NewRunner(
pollingEnabled bool,
cpuAffinityManager *cpuaffinity.Manager,
l *logger.Logger,
) (*Runner, error) {
zl := l.With().Str("topic", "SonicRunner").Logger()
l = &logger.Logger{Logger: &zl}
commandsDiode := diodes.NewPoller(diodes.NewManyToOne(2048, diodes.AlertFunc(func(missed int) {
l.Error().Msgf("dropped %d messages", missed)
})))
return &Runner{
commandsDiode,
pollingEnabled,
cpuAffinityManager,
l,
}, nil
}
func (r *Runner) Run() {
ioc := sonic.MustIO()
defer ioc.Close()
if r.pollingEnabled {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
if err := r.cpuAffinityManager.AcquireCPU(); err != nil {
switch {
case errors.Is(err, cpuaffinity.ErrOSNotSupported):
r.l.Info().Err(err).Msgf("could not acquire CPU affinity")
default:
r.l.Fatal().Err(err).Msgf("could not acquire CPU affinity")
}
}
defer func() {
if err := r.cpuAffinityManager.ReleaseCPU(); err != nil {
r.l.Error().Err(err).Msgf("could not release CPU affinity")
}
}()
}
i := 0
for {
i += 1
if i%1000 == 0 {
r.processCommands(ioc)
i = 0
}
if _, err := ioc.PollOne(); err != nil && !errors.Is(err, sonicerrors.ErrTimeout) {
r.l.Error().Err(err).Msg("could not poll")
}
}
}
func (r *Runner) CreateStream(url string, resultCh chan interface{}) {
r.commandsDiode.Set(diodes.GenericDataType(newCreateStreamCommand(url, resultCh)))
}
func (r *Runner) ReadStream(stream *websocket.WebsocketStream, onRead websocket.AsyncMessageHandler, b []byte) {
r.commandsDiode.Set(diodes.GenericDataType(newReadStreamCommand(stream, onRead, b)))
}
func (r *Runner) CloseStream(s *websocket.WebsocketStream, resultCh chan interface{}) {
r.commandsDiode.Set(diodes.GenericDataType(newCloseStreamCommand(s, resultCh)))
}
func (r *Runner) processCommands(ioc *sonic.IO) {
// read channel may be notified once for multiple messages if we can't keep up
for {
p, ok := r.commandsDiode.TryNext()
if !ok {
break
}
c := (*command)(p)
switch c.commandType {
case commandTypeCreateStream:
r.processCreateStreamCommand(ioc, c)
case commandTypeReadStream:
r.processReadStreamCommand(c)
case commandTypeCloseStream:
r.processCloseStreamCommand(c)
default:
panic("unhandled default case")
}
}
}
func (r *Runner) processCreateStreamCommand(ioc *sonic.IO, c *command) {
stream, err := websocket.NewWebsocketStream(ioc, &tls.Config{}, websocket.RoleClient)
if err != nil {
c.resultCh <- fmt.Errorf("could not create websocket stream: %w", err)
return
}
stream.AsyncHandshake(c.url, func(err error) {
if err != nil {
c.resultCh <- fmt.Errorf("could not handshake stream: %w", err)
return
}
c.resultCh <- stream
})
}
func (r *Runner) processReadStreamCommand(c *command) {
c.stream.AsyncNextMessage(c.b, c.onRead)
}
func (r *Runner) processCloseStreamCommand(c *command) {
c.stream.AsyncClose(1000, "", func(err error) {
if err != nil {
c.resultCh <- fmt.Errorf("could not close stream: %w", err)
return
}
c.resultCh <- nil
})
}
I am trying to achieve “It allows latency-sensitive programs to run in a hot-loop pinned to a thread on an isolated core in order to achieve low latency and jitter.”.
Usually I do this by calling LockOSThread and SetSchedulerAffinity (single CPU), but for some reason this introduces significantly latency spikes (>10ms) compared to running it without pinning it.
go traceconfirms that the Runner is locked to a single thread. Message handlers are not doing a lot more than setting on a poller diode.Nothing changes when I:
isolcpusnohz_fullrcu_nocbsWhat is the recommended approach to do this?