func (s *Sorter) Close() error {
	if len(s.recs) > 0 || s.totalRecords == 0 {
		s.startGenerateSortShard()
	}
	close(s.bgSorterCh)
	s.wg.Wait()
	if s.err.Err() == nil {
		s.mergeShards(s.shards, s.header, s.outPath)
	}
	for _, path := range s.shards {
		if err := os.Remove(path); err != nil {
			vlog.Errorf("sort %v: failed to remove sorter tmp file: %v (%v)", path, err, s.err.Err())
		}
	}
	return s.err.Err()
}
Close must be called after adding all the records. It blocks the caller until
the shard file is generated. After Close, Sorter becomes invalid.
package sorter
import (
	"bytes"
	"crypto/sha256"
	"encoding/binary"
	"fmt"
	"io/ioutil"
	"os"
	"runtime"
	"sort"
	"sync"
	"github.com/biogo/store/llrb"
	"github.com/grailbio/base/errors"
	"github.com/grailbio/base/file"
	"github.com/grailbio/base/vcontext"
	gbam "github.com/grailbio/bio/encoding/bam"
	"github.com/grailbio/hts/bam"
	"github.com/grailbio/hts/bgzf"
	"github.com/grailbio/hts/sam"
	"v.io/x/lib/vlog"
)
const DefaultSortBatchSize = 1 << 20
const DefaultParallelism = 2
type SortOptions struct {
	ShardIndex uint32
	SortBatchSize int
	Parallelism int
	NoCompressTmpFiles bool
	TmpDir string
}
type recCoord uint64
type sortBatch struct {
	sortTieBreaker uint64
	recs           []sortEntry
}
type sortEntry struct {
	coord recCoord
	body  []byte  
}
func (k sortEntry) String() string {
	refid, pos, reverse := parseCoord(k.coord)
	return fmt.Sprintf("(%d,%d,%v,%d)", refid, pos, reverse, len(k.body))
}
func (k sortEntry) compare(other sortEntry) int {
	if k.coord < other.coord {
		return -1
	}
	if k.coord > other.coord {
		return 1
	}
	return bytes.Compare(k.body, other.body)
}
func coordFromRecord(rec *sam.Record) (key recCoord) {
	if rec.Ref == nil {
		key = unmappedCoord
	} else {
		key = (recCoord(rec.Ref.ID()) << 33) | recCoord(rec.Pos)<<1
	}
	if (rec.Flags & sam.Reverse) != 0 {
		key |= recCoord(1)
	}
	return
}
func (r recCoord) String() string {
	refid, pos, reverse := parseCoord(r)
	return fmt.Sprintf("(%d,%d,%v)", refid, pos, reverse)
}
func makeCoord(refid, pos int, reverse bool) (key recCoord) {
	if refid < 0 {
		key = unmappedCoord
	} else {
		key = (recCoord(refid) << 33) | recCoord(pos)<<1
	}
	if reverse {
		key |= recCoord(1)
	}
	return
}
func parseCoord(coord recCoord) (refid, pos int, reverse bool) {
	if (coord & unmappedCoord) == unmappedCoord {
		refid = -1
		pos = -1
	} else {
		refid = int(int32(coord >> 33))
		pos = int(int32((coord & 0x1ffffffff) >> 1))
	}
	if (coord & 1) != 0 {
		reverse = true
	}
	return
}
const invalidCoord recCoord = 0xfffffffffffffffe
const infinityCoord = invalidCoord
const unmappedCoord recCoord = 0x7ffffffffffffffe
type Sorter struct {
	options       SortOptions
	outPath       string
	header        *sam.Header
	smallPool     *sync.Pool           
	sortBlockPool *sortShardBlockPool  
	totalRecords  uint32
	recs          []sortEntry
	err           errors.Once
	bgSorterCh    chan sortBatch
	wg     sync.WaitGroup
	mu     sync.Mutex
	shards []string  
}
type mergeLeaf struct {
	seq    int
	name   string  
	reader *sortShardReader
	done   bool  
	err    *errors.Once
}
func newMergeLeaf(seq int, reader *sortShardReader,
	pool *sortShardBlockPool, errorReporter *errors.Once) *mergeLeaf {
	leaf := mergeLeaf{
		seq:    seq,
		name:   reader.path,
		reader: reader,
		err:    errorReporter,
	}
	if !leaf.reader.scan() {
		return nil
	}
	return &leaf
}
func (l *mergeLeaf) Compare(c1 llrb.Comparable) int {
	l1 := c1.(*mergeLeaf)
	k0 := l.reader.key()
	k1 := l1.reader.key()
	if c := k0.compare(k1); c != 0 {
		return c
	}
	return l.seq - l1.seq
}
func (s *Sorter) mergeShards(paths []string, header *sam.Header, outPath string) {
	ctx := vcontext.Background()
	out, err := file.Create(ctx, outPath)
	if err != nil {
		s.err.Set(err)
		return
	}
	shardReaders := make([]*sortShardReader, len(paths))
	for i, path := range paths {
		shardReaders[i] = newSortShardReader(path, s.sortBlockPool, &s.err)
	}
	writer := newSortShardWriter(out.Writer(ctx), !s.options.NoCompressTmpFiles, true, header,
		s.sortBlockPool, &s.err)
	callback := func(key sortEntry) bool {
		writer.add(key)
		return true
	}
	internalMergeShards(shardReaders, callback, s.sortBlockPool, &s.err)
	writer.finish()
	s.err.Set(out.Close(ctx))
}
func internalMergeShards(
	shards []*sortShardReader,
	readCallback func(key sortEntry) bool,
	pool *sortShardBlockPool,
	errReporter *errors.Once) {
	leafs := llrb.Tree{}
	for i, shard := range shards {
		if c := newMergeLeaf(i, shard, pool, errReporter); c != nil {
			vlog.VI(1).Infof("Leaf %v created", c.name)
			leafs.Insert(c)
		}
	}
	vlog.VI(1).Infof("Merging %d shards, %d leafs active", len(shards), leafs.Len())
	done := false
	for !done && leafs.Len() > 0 {
		nthiter := 0
		var top, next *mergeLeaf
		leafs.Do(func(item llrb.Comparable) bool {
			nthiter++
			switch nthiter {
			case 1:
				top = item.(*mergeLeaf)
				return false
			case 2:
				next = item.(*mergeLeaf)
				return true
			default:
				vlog.Fatal(nthiter)
				return false
			}
		})
		for {
			if !readCallback(top.reader.key()) {
				done = true
				break
			}
			top.done = !top.reader.scan()
			if top.done || (next != nil && next.reader.key().compare(top.reader.key()) < 0) {
				break
			}
		}
		lenBefore := leafs.Len()
		leafs.DeleteMin()
		if !top.done {
			leafs.Insert(top)
			if lenAfter := leafs.Len(); lenBefore != lenAfter {
				vlog.Fatalf("Leaf size decreased from %d -> %d", lenBefore, lenAfter)
			}
			continue
		}
	}
	for _, shard := range shards {
		shard.drain()
	}
}
func NewSorter(outPath string, header *sam.Header, optList ...SortOptions) *Sorter {
	options := SortOptions{}
	if len(optList) > 0 {
		if len(optList) > 1 {
			vlog.Fatalf("More than options specified: %v", optList)
		}
		options = optList[0]
	}
	if options.ShardIndex == 0 {
		hash := sha256.Sum224([]byte(outPath))
		options.ShardIndex = binary.LittleEndian.Uint32(hash[:])
	}
	if options.SortBatchSize <= 0 {
		options.SortBatchSize = DefaultSortBatchSize
	}
	if options.Parallelism <= 0 {
		options.Parallelism = DefaultParallelism
	}
	vlog.VI(1).Infof("New Sorter: %v, %+v", outPath, options)
	sorter := &Sorter{
		options:       options,
		outPath:       outPath,
		header:        header,
		smallPool:     &sync.Pool{New: func() interface{} { return bytes.Buffer{} }},
		sortBlockPool: newSortShardBlockPool(),
		bgSorterCh:    make(chan sortBatch, options.Parallelism),
	}
	for i := 0; i < options.Parallelism; i++ {
		sorter.wg.Add(1)
		go func() {
			for batch := range sorter.bgSorterCh {
				path := sorter.sortRecords(batch.recs, batch.sortTieBreaker)
				sorter.mu.Lock()
				sorter.shards = append(sorter.shards, path)
				sorter.mu.Unlock()
			}
			sorter.wg.Done()
		}()
	}
	return sorter
}
func (s *Sorter) AddRecord(rec *sam.Record) {
	s.totalRecords++
	var buf bytes.Buffer
	err := bam.Marshal(rec, &buf)
	if err != nil {
		s.err.Set(err)
		return
	}
	s.recs = append(s.recs, sortEntry{coordFromRecord(rec), buf.Bytes()})
	if len(s.recs) >= s.options.SortBatchSize {
		s.startGenerateSortShard()
	}
}
func (s *Sorter) startGenerateSortShard() {
	s.bgSorterCh <- sortBatch{
		recs:           s.recs,
		sortTieBreaker: (uint64(s.options.ShardIndex) << 32) + uint64(s.totalRecords),
	}
	s.recs = nil
}
func (s *Sorter) sortRecords(records []sortEntry, sortTieBreaker uint64) string {
	vlog.VI(1).Infof("Sorting %d records, tiebreaker %x", len(records), sortTieBreaker)
	temp, err := ioutil.TempFile(s.options.TmpDir, "bamsort")
	if err != nil {
		s.err.Set(err)
		return ""
	}
	sort.SliceStable(records, func(i, j int) bool {
		return records[i].compare(records[j]) < 0
	})
	writer := newSortShardWriter(temp, !s.options.NoCompressTmpFiles, false, nil, s.sortBlockPool, &s.err)
	for _, key := range records {
		writer.add(key)
	}
	writer.finish()
	s.err.Set(temp.Close())
	return temp.Name()
}