func FirstValueReducer(ctx context.Context, input ReducerInput) ([]byte, error) {
	if !input.Next() {
		return nil, fmt.Errorf("no inputs to reducer")
	}
	if err := input.Err(); err != nil {
		return nil, fmt.Errorf("reduce: %w", err)
	}
	return input.Value(), nil
}
FirstValueReducer outputs its first value and quits.
package mr
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"path"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/shiblon/entroq"
	"golang.org/x/sync/errgroup"
)
const (
	claimDuration = 5 * time.Second
	claimWait     = 20 * time.Second
	shuffleWait   = 5 * time.Second
)
func Fingerprint64(key []byte) uint64 {
	h := fnv.New64()
	h.Write(key)
	return h.Sum64()
}
func ShardForKey(key []byte, n int) int {
	return int(Fingerprint64(key) % uint64(n))
}
type MapEmitFunc func(ctx context.Context, key, value []byte) error
type MapEmitter interface {
	Emit(ctx context.Context, key, value []byte) error
	AsModifyArgs(prefix string, additional ...entroq.ModifyArg) ([]entroq.ModifyArg, error)
}
type CollectingMapEmitter struct {
	NumShards int
	shards    [][]*KV
}
func NewCollectingMapEmitter(numShards int) *CollectingMapEmitter {
	if numShards < 1 {
		numShards = 1
	}
	return &CollectingMapEmitter{
		NumShards: numShards,
		shards:    make([][]*KV, numShards),
	}
}
func (e *CollectingMapEmitter) Emit(_ context.Context, key, value []byte) error {
	shard := ShardForKey(key, e.NumShards)
	e.shards[shard] = append(e.shards[shard], NewKV(key, value))
	return nil
}
func (e *CollectingMapEmitter) AsModifyArgs(qPrefix string, additional ...entroq.ModifyArg) ([]entroq.ModifyArg, error) {
	var args []entroq.ModifyArg
	for shard, kvs := range e.shards {
		if len(kvs) == 0 {
			continue
		}
		sort.Sort(byKey(kvs))
		value, err := json.Marshal(kvs)
		if err != nil {
			return nil, fmt.Errorf("emit to modify args: %w", err)
		}
		queue := path.Join(qPrefix, fmt.Sprint(shard))
		args = append(args, entroq.InsertingInto(queue, entroq.WithValue(value)))
	}
	return append(args, additional...), nil
}
type reducingProxyMapEmitter struct {
	sync.Mutex
	target MapEmitter
	reduce Reducer
	collection []*KV
	emitCtx    context.Context  
}
func newReducingProxyMapEmitter(target MapEmitter, reduce Reducer) *reducingProxyMapEmitter {
	return &reducingProxyMapEmitter{
		target: target,
		reduce: reduce,
	}
}
func (e *reducingProxyMapEmitter) reduceAndEmit(ctx context.Context, threshold int) error {
	var (
		kvs  []*KV
		err  error
		bail bool
	)
	func() {
		e.Lock()
		defer e.Unlock()
		if len(e.collection) < threshold {
			bail = true
			return
		}
		sort.Sort(byKey(e.collection))
		kvs, err = reduceSortedKVs(ctx, e.reduce, e.collection)
		if err != nil {
			err = fmt.Errorf("reduce and emit error: %w", err)
		}
	}()
	if bail {
		return nil
	}
	for _, kv := range kvs {
		if err := e.target.Emit(ctx, kv.Key, kv.Value); err != nil {
			return fmt.Errorf("proxy emit: %w", err)
		}
	}
	e.Lock()
	defer e.Unlock()
	e.collection = kvs
	return nil
}
func (e *reducingProxyMapEmitter) Emit(ctx context.Context, key, value []byte) error {
	func() {
		e.Lock()
		defer e.Unlock()
		e.emitCtx = ctx
		e.collection = append(e.collection, NewKV(key, value))
	}()
	if err := e.reduceAndEmit(ctx, 100); err != nil {
		return fmt.Errorf("reducing proxy emit: %w", err)
	}
	return nil
}
func (e *reducingProxyMapEmitter) AsModifyArgs(prefix string, additional ...entroq.ModifyArg) ([]entroq.ModifyArg, error) {
	if err := e.reduceAndEmit(e.emitCtx, 1); err != nil {
		return nil, fmt.Errorf("proxy modify args: %w", err)
	}
	return e.target.AsModifyArgs(prefix, additional...)
}
type Mapper func(ctx context.Context, key, value []byte, emit MapEmitFunc) error
func IdentityMapper(ctx context.Context, key, value []byte, emit MapEmitFunc) error {
	return emit(ctx, key, value)
}
func WordCountMapper(ctx context.Context, key, value []byte, emit MapEmitFunc) error {
	words := make(map[string]int)
	for _, w := range strings.Fields(string(value)) {
		words[w]++
	}
	numEmitted := 0
	for word, count := range words {
		if (numEmitted+1)%1000 == 0 {
			select {
			case <-ctx.Done():
				return fmt.Errorf("canceled map operation: %w", ctx.Err())
			default:
			}
		}
		if err := emit(ctx, []byte(word), []byte(fmt.Sprint(count))); err != nil {
			return fmt.Errorf("word count output error: %w", err)
		}
		numEmitted++
	}
	return nil
}
type KV struct {
	Key   []byte `json:"key"`
	Key2  []byte `json:"key2"`  
	Value []byte `json:"value"`
}
func NewKV(key, value []byte) *KV {
	return &KV{Key: key, Value: value}
}
func (kv *KV) String() string {
	if len(kv.Key2) > 0 {
		return fmt.Sprintf("(%s,%s)=%s", string(kv.Key), string(kv.Key2), string(kv.Value))
	}
	return fmt.Sprintf("(%s)=%s", string(kv.Key), string(kv.Value))
}
type byKey []*KV
func (b byKey) Less(i, j int) bool {
	cmp := bytes.Compare(b[i].Key, b[j].Key)
	if cmp == 0 {
		return bytes.Compare(b[i].Key2, b[j].Key2) < 0
	}
	return cmp < 0
}
func (b byKey) Swap(i, j int) { b[i], b[j] = b[j], b[i] }
func (b byKey) Len() int      { return len(b) }
type MapWorker struct {
	client     *entroq.EntroQ
	newEmitter func() MapEmitter
	Name string
	InputQueue   string
	OutputPrefix string
	Map          Mapper
	EarlyReduce  Reducer
}
type MapWorkerOption func(*MapWorker)
func WithMapper(m Mapper) MapWorkerOption {
	return func(mw *MapWorker) {
		if m == nil {
			return
		}
		mw.Map = m
	}
}
func WithEarlyReducer(r Reducer) MapWorkerOption {
	return func(mw *MapWorker) {
		mw.EarlyReduce = r
	}
}
func MapAsName(name string) MapWorkerOption {
	return func(mw *MapWorker) {
		mw.Name = name
	}
}
func MapToOutputPrefix(p string) MapWorkerOption {
	return func(mw *MapWorker) {
		if p == "" {
			return
		}
		mw.OutputPrefix = p
	}
}
func NewMapWorker(eq *entroq.EntroQ, inQueue string, newEmitter func() MapEmitter, opts ...MapWorkerOption) *MapWorker {
	w := &MapWorker{
		client:       eq,
		newEmitter:   newEmitter,
		InputQueue:   inQueue,
		OutputPrefix: inQueue,
		Map:          IdentityMapper,
	}
	for _, opt := range opts {
		opt(w)
	}
	return w
}
func (w *MapWorker) Run(ctx context.Context) error {
	return w.client.NewWorker(w.InputQueue).Run(ctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
		emitter := w.newEmitter()
		if w.EarlyReduce != nil {
			emitter = newReducingProxyMapEmitter(emitter, w.EarlyReduce)
		}
		kv := new(KV)
		if err := json.Unmarshal(task.Value, kv); err != nil {
			return nil, fmt.Errorf("map run json: %w", err)
		}
		if err := w.Map(ctx, kv.Key, kv.Value, emitter.Emit); err != nil {
			return nil, fmt.Errorf("map run map: %w", err)
		}
		return emitter.AsModifyArgs(w.OutputPrefix, task.AsDeletion())
	})
}
type ReducerInput interface {
	Key() []byte
	Value() []byte
	Err() error
	Next() bool
}
type Reducer func(ctx context.Context, input ReducerInput) ([]byte, error)