func (ps *PubSub[T, M]) Subscribe(mask M, subCh chan T, doneCh <-chan struct{}, filter func(entry T) bool) error {
	totalSubs := atomic.AddInt32(&ps.numSubscribers, 1)
	if ps.maxSubscribers > 0 && totalSubs > ps.maxSubscribers {
		atomic.AddInt32(&ps.numSubscribers, -1)
		return fmt.Errorf("the limit of `%d` subscribers is reached", ps.maxSubscribers)
	}
	ps.Lock()
	defer ps.Unlock()

	for i := 0; i < numIterations; i++ {
		if rwm.GetRLock(context.Background(), "", "", time.Second) {
			n := atomic.AddInt32(activity, 1)
			if n < 1 || n >= 10000 {
				panic(fmt.Sprintf("wlock(%d)\n", n))
			}
			for i := 0; i < 100; i++ {
			}
			atomic.AddInt32(activity, -1)
			rwm.RUnlock()
		}
	}
	cdone <- true
}

// Borrowed from rwmutex_test.go

func abigen_sync_atomic_SwapUint64(addr *uint64, new uint64) (old uint64)

//go:linkname abigen_sync_atomic_AddInt32 sync/atomic.AddInt32
func abigen_sync_atomic_AddInt32(addr *int32, delta int32) (new int32)

//go:linkname abigen_sync_atomic_AddUint32 sync/atomic.AddUint32
func abigen_sync_atomic_AddUint32(addr *uint32, delta uint32) (new uint32)

//go:linkname abigen_sync_atomic_AddInt64 sync/atomic.AddInt64

package par

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

func TestWork(t *testing.T) {
	var w Work[int]

	const N = 10000
	n := int32(0)
	w.Add(N)
	w.Do(100, func(i int) {
		atomic.AddInt32(&n, 1)
		if i >= 2 {
			w.Add(i - 1)
			w.Add(i - 2)
		}
		w.Add(i >> 1)
		w.Add((i >> 1) ^ 1)
	})
	if n != N+1 {
		t.Fatalf("ran %d items, expected %d", n, N+1)
	}
}

	go func() {
		x1 = 1
		if atomic.AddInt32(&s, 1) == 2 {
			x2 = 1
		}
		ch <- true
	}()
	go func() {
		x2 = 1
		if atomic.AddInt32(&s, 1) == 2 {
			x1 = 1
		}
		ch <- true
	}()
	<-ch
	<-ch
}

func TestNoRaceAtomicLoadAddInt32(t *testing.T) {
	var x int64
	_ = x
	var s int32
	go func() {
		x = 2
		atomic.AddInt32(&s, 1)
	}()
	for atomic.LoadInt32(&s) != 1 {

)

func TestExecute(t *testing.T) {
	testVal := int32(0)
	wg := sync.WaitGroup{}
	wg.Add(5)
	queue := CreateWorkerQueue(func(ctx context.Context, fireAt time.Time, args *WorkArgs) error {
		atomic.AddInt32(&testVal, 1)
		wg.Done()
		return nil
	})
	now := time.Now()
	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, now)
	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, now)

	// (for arm64HasATOMICS), while ARM64 v8.1 has only atomic and no dynamic check.
	// arm64/v8.0:"LDADDALW"
	// arm64/v8.1:"LDADDALW"
	// arm64/v8.0:".*arm64HasATOMICS"
	// arm64/v8.1:-".*arm64HasATOMICS"
	atomic.AddInt32(&c.count, 1)

	for i := 0; i < num_iterations; i++ {
		rwm.RLock()
		n := atomic.AddInt32(activity, 1)
		if n < 1 || n >= 10000 {
			rwm.RUnlock()
			panic(fmt.Sprintf("wlock(%d)\n", n))
		}
		for i := 0; i < 100; i++ {
		}
		atomic.AddInt32(activity, -1)
		rwm.RUnlock()
	}
	cdone <- true
}

	plt.AddUint32(ctxt.Arch, 0x030ec023)               // subu  $24, $24, $14
	plt.AddUint32(ctxt.Arch, 0x03e07825)               // move  $15, $31
	plt.AddUint32(ctxt.Arch, 0x0018c0c2)               // srl   $24, $24, 3
	plt.AddUint32(ctxt.Arch, 0x0320f809)               // jalr  $25
	plt.AddUint32(ctxt.Arch, 0x2718fffe)               // subu  $24, $24, 2

// confuse the scheduler.

package main

import "sync/atomic"

func main() {
	f()
}

var x int32

type T [10]int
var sink *T

func f() (t T) {
	atomic.AddInt32(&x, 1)
	g(42, 42, 42, 42, 42, &t) // use int values that is invalid pointer to smash the stack slot of return value of runtime.newobject
	return
}

//go:noinline
func g(a, b, c, d, e int, p *T) {