job_subscribe.go

package main

import (
	"context"
	"encoding/binary"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

func (jm *JobManager) CurrentJob() *Job {
	jm.mu.RLock()
	defer jm.mu.RUnlock()
	return jm.curJob
}

func (jm *JobManager) currentJobAndLongPollID() (*Job, string) {
	jm.mu.RLock()
	defer jm.mu.RUnlock()
	return jm.curJob, jm.longPollID
}

func (jm *JobManager) Ready() bool {
	jm.mu.RLock()
	defer jm.mu.RUnlock()
	return jm.curJob != nil
}

func (jm *JobManager) NextExtranonce1() []byte {
	id := atomic.AddUint32(&jm.extraID, 1)
	var buf [4]byte // Use fixed-size array instead of slice allocation
	binary.BigEndian.PutUint32(buf[:], id)
	return buf[:]
}

func (jm *JobManager) nextJobID() string {
	id := (atomic.AddUint64(&jm.jobIDCounter, 1) - 1) % jobIDRolloverModulo
	return encodeBase58Uint64(id)
}

func (jm *JobManager) Subscribe() chan *Job {
	ch := make(chan *Job, jobSubscriberBuffer)
	jm.subsMu.Lock()
	jm.subs[ch] = struct{}{}
	jm.subsMu.Unlock()

	return ch
}

func (jm *JobManager) Unsubscribe(ch chan *Job) {
	jm.subsMu.Lock()
	delete(jm.subs, ch)
	close(ch)
	jm.subsMu.Unlock()
}

func (jm *JobManager) ActiveMiners() int {
	jm.subsMu.Lock()
	defer jm.subsMu.Unlock()
	return len(jm.subs)
}

func (jm *JobManager) broadcastJob(job *Job) {
	// Queue the job for ordered async distribution. If the queue is full, replace
	// its oldest pending update with this newest job. Broadcasting synchronously
	// here would overtake already-queued jobs and allow a stale job to arrive last.
	select {
	case jm.notifyQueue <- job:
		return
	default:
	}

	dropped := false
	select {
	case <-jm.notifyQueue:
		dropped = true
	default:
	}
	select {
	case jm.notifyQueue <- job:
		logger.Warn("notification queue full; replaced oldest pending job", "dropped", dropped)
	default:
		logger.Warn("notification queue full; newest job could not be queued")
	}
}

// sendJobNonBlocking attempts to deliver the latest job to a subscriber channel
// without blocking. If the channel is full, it drops one pending job and retries
// so the subscriber converges to the newest template.
func sendJobNonBlocking(ch chan *Job, job *Job) (dropped bool) {
	select {
	case ch <- job:
		return false
	default:
	}

	// Channel full: drop one stale job, then retry once.
	select {
	case <-ch:
		dropped = true
	default:
	}
	select {
	case ch <- job:
	default:
		dropped = true
	}
	return dropped
}

// broadcastJobSync performs synchronous job notification (fallback only)
func (jm *JobManager) broadcastJobSync(job *Job) {
	jm.subsMu.Lock()
	dropped := 0
	subscribers := len(jm.subs)
	for ch := range jm.subs {
		if sendJobNonBlocking(ch, job) {
			dropped++
		}
	}
	jm.subsMu.Unlock()

	if dropped > 0 {
		logger.Warn("job broadcast dropped stale updates (sync)", "subscribers", subscribers, "dropped", dropped)
	}
}

const (
	jobFanoutMaxShards           = 8
	jobFanoutSubscribersPerShard = 256
)

func jobFanoutShardCount(subscribers int) int {
	if subscribers <= jobFanoutSubscribersPerShard {
		return 1
	}
	shards := (subscribers + jobFanoutSubscribersPerShard - 1) / jobFanoutSubscribersPerShard
	if procs := runtime.GOMAXPROCS(0); shards > procs {
		shards = procs
	}
	if shards > jobFanoutMaxShards {
		shards = jobFanoutMaxShards
	}
	if shards < 1 {
		return 1
	}
	return shards
}

// distributeJobSharded keeps subsMu held until every shard completes. This
// prevents Unsubscribe from closing a channel during delivery. The caller does
// not dequeue the next job until this barrier finishes, preserving per-miner
// FIFO order while parallelizing large fanouts.
func (jm *JobManager) distributeJobSharded(job *Job) (subscribers, dropped, shards int) {
	jm.subsMu.Lock()
	defer jm.subsMu.Unlock()

	subscribers = len(jm.subs)
	shards = jobFanoutShardCount(subscribers)
	if shards == 1 {
		for ch := range jm.subs {
			if sendJobNonBlocking(ch, job) {
				dropped++
			}
		}
		return subscribers, dropped, shards
	}

	buckets := make([][]chan *Job, shards)
	i := 0
	for ch := range jm.subs {
		bucket := i % shards
		buckets[bucket] = append(buckets[bucket], ch)
		i++
	}
	droppedByShard := make([]int, shards)
	deliver := func(shard int) {
		for _, ch := range buckets[shard] {
			if sendJobNonBlocking(ch, job) {
				droppedByShard[shard]++
			}
		}
	}

	var wg sync.WaitGroup
	wg.Add(shards - 1)
	for shard := 1; shard < shards; shard++ {
		go func(shard int) {
			defer wg.Done()
			deliver(shard)
		}(shard)
	}
	deliver(0)
	wg.Wait()
	for _, count := range droppedByShard {
		dropped += count
	}
	return subscribers, dropped, shards
}

// notificationWorker is the single FIFO ingress for job notifications. Each
// job may fan out in parallel, but the next job waits for the current barrier.
func (jm *JobManager) notificationWorker(ctx context.Context, workerID int) {
	defer jm.notifyWg.Done()

	for {
		select {
		case <-ctx.Done():
			return
		case job, ok := <-jm.notifyQueue:
			if !ok {
				return
			}

			subscribers, dropped, shards := jm.distributeJobSharded(job)

			if dropped > 0 {
				logger.Warn("job broadcast dropped stale updates", "worker", workerID, "subscribers", subscribers, "dropped", dropped, "shards", shards)
			}
			if jm.metrics != nil {
				if job.FastEmpty && !job.fastEmptyTriggeredAt.IsZero() {
					jm.metrics.ObserveFastEmptyNotifyLatency(time.Since(job.fastEmptyTriggeredAt))
				} else if !job.templateReceivedAt.IsZero() {
					jm.metrics.ObserveGBTApplyNotifyLatency(time.Since(job.templateReceivedAt))
				}
			}
		}
	}
}