perf: mux - consolidate wire, sc, mu slices into muxwire struct

[arbhalerao]

fixes: #862

[arbhalerao]

Hi @rueian, any update on this PR? Thanks!

[rueian]

Hi @arbhalerao, the change looks good to me. However, I haven't had a chance to make sure there is no performane degration.

[rueian]

Or could you help doing a simulated benchmark on this critical path:

rueidis/mux.go

Lines 155 to 158 in dad42a2

	func (m *mux) _pipe(ctx context.Context, i uint16) (w wire, err error) {
	if w = m.wire[i].Load().(wire); w != m.init {
	return w, nil
	}

to show that it won't degrade with your change?

[arbhalerao]

Hi @rueian,
Absolutely, I'll put together a benchmark for that path and follow up soon with the results.

[rueian]

Thanks @arbhalerao. A simulated benchmark on similar structures is enough. I don't need a real benchmark on the real mux structure.

[arbhalerao]

Hi @rueian,

I've created a benchmark testing the critical path performance as requested. The benchmark compares:

• Current: m.wire[i].Load()
• Proposed: m.muxwires[i].wire.Load()

Results:

BenchmarkCurrentWireLoad-22         	861622592	         1.350 ns/op	       0 B/op	       0 allocs/op
BenchmarkProposedWireLoad-22        	879522333	         1.385 ns/op	       0 B/op	       0 allocs/op

BenchmarkCurrentCriticalPath-22     	442859290	         2.755 ns/op	       0 B/op	       0 allocs/op
BenchmarkProposedCriticalPath-22    	439302198	         2.728 ns/op	       0 B/op	       0 allocs/op

package main

import (
	"sync"
	"sync/atomic"
	"testing"
)

type wire interface {
	ID() int
}

type mockWire struct {
	id int
}

func (m *mockWire) ID() int {
	return m.id
}

type singleconnect struct {
	w wire
	e error
	g sync.WaitGroup
}

var initWire = &mockWire{id: -1}

type currentMux struct {
	wire []atomic.Value
	sc   []*singleconnect
	mu   []sync.Mutex
}

type muxwire struct {
	wire atomic.Value
	sc   *singleconnect
	mu   sync.Mutex
}

type proposedMux struct {
	muxwires []muxwire
}

func newCurrentMux(size int) *currentMux {
	m := &currentMux{
		wire: make([]atomic.Value, size),
		mu:   make([]sync.Mutex, size),
		sc:   make([]*singleconnect, size),
	}
	for i := 0; i < size; i++ {
		m.wire[i].Store(initWire)
	}
	return m
}

func newProposedMux(size int) *proposedMux {
	m := &proposedMux{
		muxwires: make([]muxwire, size),
	}
	for i := 0; i < size; i++ {
		m.muxwires[i].wire.Store(initWire)
	}
	return m
}

func (m *currentMux) wireLoad(i uint16) wire {
	return m.wire[i].Load().(wire)
}

func (m *proposedMux) wireLoad(i uint16) wire {
	return m.muxwires[i].wire.Load().(wire)
}

func BenchmarkCurrentWireLoad(b *testing.B) {
	m := newCurrentMux(16)
	for i := 0; i < b.N; i++ {
		m.wireLoad(uint16(i % 16))
	}
}

func BenchmarkProposedWireLoad(b *testing.B) {
	m := newProposedMux(16)
	for i := 0; i < b.N; i++ {
		m.wireLoad(uint16(i % 16))
	}
}

func BenchmarkCurrentCriticalPath(b *testing.B) {
	m := newCurrentMux(16)
	for i := 0; i < b.N; i++ {
		slot := uint16(i % 16)
		if w := m.wireLoad(slot); w != initWire {
			continue
		}
	}
}

func BenchmarkProposedCriticalPath(b *testing.B) {
	m := newProposedMux(16)
	for i := 0; i < b.N; i++ {
		slot := uint16(i % 16)
		if w := m.wireLoad(slot); w != initWire {
			continue
		}
	}
}

[rueian]

Thanks @arbhalerao