빠르게 메시지를 전파하고 싶어

type Node[T any] struct {
	Value  T
	Next   *Node[T]
	Closed bool
}

type AsyncLinkedList[T any] struct {
	head               atomic.Pointer[Node[T]]
	tail               atomic.Pointer[Node[T]]
	bufferLength       atomic.Int64
	maxBuf             int64
	lock               sync.Locker
	cond               *sync.Cond
	wheel              *timingwheel.TimingWheel
	latestNotification time.Time
}

func New[T any](maxBufferSize int64) *AsyncLinkedList[T] {
	l := FakeLock{}
	tw := timingwheel.NewTimingWheel(50*time.Millisecond, 60)
	all := &AsyncLinkedList[T]{
		maxBuf: maxBufferSize,
		lock:   l,
		cond:   sync.NewCond(l),
		wheel:  tw,
	}
	tw.ScheduleFunc(all, func() {
		if all.latestNotification.Add(10 * time.Millisecond).Before(time.Now()) {
			all.latestNotification = time.Now()
			all.cond.Broadcast()
		}
	})
	tw.Start()
	return all
}

func (all *AsyncLinkedList[T]) Push(value T) {
	all.push(value, false)
}

func (all *AsyncLinkedList[T]) Close() {
	all.push(*new(T), true)
}

func (all *AsyncLinkedList[T]) push(value T, closed bool) {
	node := all.getNode()
	node.Value = value
	node.Closed = closed

outer:
	for {
		tail := all.tail.Load()
		if all.tail.CompareAndSwap(tail, node) {
			switch tail {
			case nil:
				all.head.Store(node)
			default:
				tail.Next = node
			}
			all.latestNotification = time.Now()
			all.cond.Broadcast()
			if all.bufferLength.Add(1) > all.maxBuf {
				all.bufferLength.Add(-1)
			inner:
				for {
					head := all.head.Load()
					if all.head.CompareAndSwap(head, head.Next) {
						all.putNode(head)
						break inner
					}
					fmt.Println("failed to swap head")
				}
			}
			break outer
		}
	}
}

func (all *AsyncLinkedList[T]) Subscribe(callback func(value T, closed bool)) <-chan struct{} {
	cursor := all.head.Load()
	last := (*Node[T])(nil)
	closed := make(chan struct{}, 1)

	go func() {
		for {
			all.lock.Lock()
			all.cond.Wait()
			all.lock.Unlock()

			switch last {
			case nil:
				cursor = all.head.Load()
			default:
				cursor = last.Next
			}

			for cursor != nil {
				last = cursor
				callback(cursor.Value, cursor.Closed)
				switch cursor.Closed {
				case true:
					close(closed)
					return
				}
				cursor = cursor.Next
			}
		}
	}()

	return closed
}

package all_test

import (
	"sync"
	"testing"

	"github.com/snowmerak/all"
)

const (
	Subscribers = 1000
	Messages    = 10000
	Buffer      = 100
)

func BenchmarkAllLinkedList(b *testing.B) {
	ll := all.New[[]byte](Buffer)

	wg := sync.WaitGroup{}
	for i := 0; i < Subscribers; i++ {
		wg.Add(1)
		ll.Subscribe(func(value []byte, closed bool) {
			if closed {
				wg.Done()
			}

			mesg := value
			_ = mesg
		})
	}

	for i := 0; i < Messages; i++ {
		ll.Push([]byte("hello"))
	}
	ll.Close()

	wg.Wait()
}

func BenchmarkChannel(b *testing.B) {
	chList := make([]chan []byte, Subscribers)

	wg := sync.WaitGroup{}
	for i := 0; i < Subscribers; i++ {
		wg.Add(1)
		chList[i] = make(chan []byte, Buffer)
		go func(ch chan []byte) {
			defer wg.Done()
			for mesg := range ch {
				_ = mesg
			}
		}(chList[i])
	}

	for i := 0; i < Messages; i++ {
		for j := 0; j < Subscribers; j++ {
			chList[j] <- []byte("hello")
		}
	}
	for i := 0; i < Subscribers; i++ {
		close(chList[i])
	}

	wg.Wait()
}