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(), nil, id, source, Options{Timeout: 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(context.Background())
		}
	}
	cdone <- true
}

// Borrowed from rwmutex_test.go

	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

			w.Header().Set(RetryAfter, "60")

			w.WriteHeader(http.StatusServiceUnavailable)
			w.Write([]byte(http.ErrServerClosed.Error()))
			return
		}

		atomic.AddInt32(&srv.requestCount, 1)
		defer atomic.AddInt32(&srv.requestCount, -1)

		// Handle request using passed handler.
		handler.ServeHTTP(w, r)
	})

	srv.listenerMutex.Lock()
	srv.Handler = wrappedHandler
	srv.listener = listener

				case errors.Is(err, errFileNotFound):
					atomic.StoreInt32(&missingPartsHeal, 1)
				case errors.Is(err, errFileCorrupt):
					atomic.StoreInt32(&bitrotHeal, 1)
				case errors.Is(err, errDiskNotFound):
					atomic.AddInt32(&disksNotFound, 1)
				}

				// This will be communicated upstream.
				p.orgReaders[bufIdx] = nil
				if br, ok := p.readers[i].(io.Closer); ok {
					br.Close()
				}
				p.readers[i] = nil