extsort

A high-performance external sorting library for Go that enables sorting of arbitrarily large datasets that don't fit entirely in memory. The library operates on channels and uses temporary disk storage to handle datasets larger than available RAM.

Features

• Memory Efficient: Sorts datasets of any size using configurable memory limits
• High Performance: Optimized for throughput with parallel sorting and merging
• Generic Support: Modern Go generics for type-safe operations
• Legacy Compatible: Maintains backward compatibility with interface-based API
• Cross-Platform: Works on Unix, Linux, macOS, and Windows
• Channel-Based: Integrates seamlessly with Go's concurrency patterns

Installation

go get github.com/lanrat/extsort

Quick Start

IMPORTANT: All input channels must be closed after all data has been sent. The Sort() method will continue reading from the input channel until it is closed. Failure to close the input channel will cause the sorting process to hang indefinitely.

Generic API (Recommended)

The modern generic API provides type safety and improved performance:

import (
"context""fmt""math/rand""github.com/lanrat/extsort"
)
funcmain() {
// Create input channel with unsorted integersinputChan:=make(chanint, 100)
gofunc() {
deferclose(inputChan)
fori:=0; i<1000000; i++ {
inputChan<-rand.Int()
}
}()
// Sort using the generic APIsorter, outputChan, errChan:=extsort.Ordered(inputChan, nil)
// Start sorting in backgroundgosorter.Sort(context.Background())
// Process sorted resultsforvalue:=rangeoutputChan {
fmt.Println(value)
}
// Check for errorsiferr:=<-errChan; err!=nil {
panic(err)
}
}

String Sorting

For string data, use the optimized string sorter:

import (
"context""fmt""github.com/lanrat/extsort"
)
funcmain() {
words:= []string{"zebra", "apple", "banana", "cherry"}
inputChan:=make(chanstring, len(words))
for_, word:=rangewords {
inputChan<-word
}
close(inputChan)
sorter, outputChan, errChan:=extsort.Strings(inputChan, nil)
gosorter.Sort(context.Background())
fmt.Println("Sorted words:")
forword:=rangeoutputChan {
fmt.Println(word)
}
iferr:=<-errChan; err!=nil {
panic(err)
}
}

Custom Types with Generic API

import (
"bytes""context""encoding/gob""fmt""github.com/lanrat/extsort"
)
typePersonstruct {
NamestringAgeint
}
funcpersonToBytes(pPerson) ([]byte, error) {
varbuf bytes.Bufferenc:=gob.NewEncoder(&buf)
err:=enc.Encode(p)
returnbuf.Bytes(), err
}
funcpersonFromBytes(data []byte) (Person, error) {
varpPersonbuf:=bytes.NewReader(data)
dec:=gob.NewDecoder(buf)
err:=dec.Decode(&p)
returnp, err
}
funccomparePersonsByAge(a, bPerson) int {
// Sort by ageifa.Age!=b.Age {
ifa.Age<b.Age {
return-1
}
return1
}
return0
}
funcmain() {
people:= []Person{
{"Alice", 30},
{"Bob", 25},
{"Charlie", 35},
}
inputChan:=make(chanPerson, len(people))
for_, person:=rangepeople {
inputChan<-person
}
close(inputChan)
sorter, outputChan, errChan:=extsort.Generic(
inputChan,
personFromBytes,
personToBytes,
comparePersonsByAge,
nil,
)
gosorter.Sort(context.Background())
fmt.Println("People sorted by age:")
forperson:=rangeoutputChan {
fmt.Printf("%s (age %d)\n", person.Name, person.Age)
}
iferr:=<-errChan; err!=nil {
panic(err)
}
}

Configuration

Customize sorting behavior with the Config struct:

config:=&extsort.Config{
ChunkSize: 500000, // Records per chunk (default: 1M)NumWorkers: 4, // Parallel sorting/merging workers (default: 2)ChanBuffSize: 10, // Channel buffer size (default: 1)SortedChanBuffSize: 1000, // Output channel buffer (default: 1000)TempFilesDir: "/var/tmp", // Temporary files directory (default: intelligent selection)
}
sorter, outputChan, errChan:=extsort.Ordered(inputChan, config)

Temporary Directory Selection

When TempFilesDir is empty (default), the library intelligently selects temporary directories that prefer disk-backed locations over potentially memory-backed filesystems. On Linux systems where /tmp may be mounted as tmpfs (memory-backed), this helps prevent out-of-memory issues when sorting datasets larger than available RAM.

For production use with large datasets, it's recommended to explicitly set TempFilesDir to a known disk-backed directory (such as /var/tmp on Unix systems) to ensure optimal performance and avoid memory limitations.

Legacy Interface-Based API

The library maintains backward compatibility with the original interface-based API:

import (
"context""encoding/binary""fmt""math/rand""github.com/lanrat/extsort"
)
typesortIntstruct {
valueint64
}
func (ssortInt) ToBytes() []byte {
buf:=make([]byte, 8)
binary.LittleEndian.PutUint64(buf, uint64(s.value))
returnbuf
}
funcsortIntFromBytes(data []byte) extsort.SortType {
value:=int64(binary.LittleEndian.Uint64(data))
returnsortInt{value: value}
}
funccompareSortInt(a, b extsort.SortType) bool {
returna.(sortInt).value<b.(sortInt).value
}
funcmain() {
inputChan:=make(chan extsort.SortType, 100)
gofunc() {
deferclose(inputChan)
fori:=0; i<100000; i++ {
inputChan<-sortInt{value: rand.Int63()}
}
}()
sorter, outputChan, errChan:=extsort.New(
inputChan,
sortIntFromBytes,
compareSortInt,
nil,
)
gosorter.Sort(context.Background())
foritem:=rangeoutputChan {
fmt.Println(item.(sortInt).value)
}
iferr:=<-errChan; err!=nil {
panic(err)
}
}

Diff Sub-Package

The diff sub-package provides functionality for comparing two sorted data streams and identifying differences. It's particularly useful for comparing large datasets efficiently.

Basic String Diff

import (
"context""fmt""github.com/lanrat/extsort/diff"
)
funcmain() {
// Create two sorted string channelsstreamA:=make(chanstring, 5)
streamB:=make(chanstring, 5)
// Populate stream Agofunc() {
deferclose(streamA)
for_, item:=range []string{"apple", "banana", "cherry", "elderberry"} {
streamA<-item
}
}()
// Populate stream Bgofunc() {
deferclose(streamB)
for_, item:=range []string{"banana", "cherry", "date", "fig"} {
streamB<-item
}
}()
// Create error channelserrA:=make(chanerror, 1)
errB:=make(chanerror, 1)
close(errA)
close(errB)
// Process differencesresult, err:=diff.Strings(
context.Background(),
streamA, streamB,
errA, errB,
func(delta diff.Delta, itemstring) error {
switchdelta {
casediff.OLD:
fmt.Printf("Only in A: %s\n", item)
casediff.NEW:
fmt.Printf("Only in B: %s\n", item)
}
returnnil
},
)
iferr!=nil {
panic(err)
}
fmt.Printf("Summary: %d items only in A, %d items only in B, %d common items\n",
result.ExtraA, result.ExtraB, result.Common)
}

Generic Diff for Custom Types

import (
"context""fmt""github.com/lanrat/extsort/diff"
)
funcmain() {
// Create channels with integer datastreamA:=make(chanint, 5)
streamB:=make(chanint, 5)
errA:=make(chanerror, 1)
errB:=make(chanerror, 1)
// Populate streamsgofunc() {
deferclose(streamA)
deferclose(errA)
for_, num:=range []int{1, 3, 5, 7, 9} {
streamA<-num
}
}()
gofunc() {
deferclose(streamB)
deferclose(errB)
for_, num:=range []int{2, 4, 5, 6, 8} {
streamB<-num
}
}()
// Compare using generic diffcompareFunc:=func(a, bint) int {
ifa<b {
return-1
}
ifa>b {
return1
}
return0
}
resultFunc:=func(delta diff.Delta, itemint) error {
symbol:=map[diff.Delta]string{diff.OLD: "<", diff.NEW: ">"}[delta]
fmt.Printf("%s %d\n", symbol, item)
returnnil
}
result, err:=diff.Generic(
context.Background(),
streamA, streamB,
errA, errB,
compareFunc,
resultFunc,
)
iferr!=nil {
panic(err)
}
fmt.Printf("Differences found: %d\n", result.ExtraA+result.ExtraB)
}

Parallel Diff Processing

import (
"context""fmt""sync""github.com/lanrat/extsort/diff"
)
funcmain() {
streamA:=make(chanstring, 100)
streamB:=make(chanstring, 100)
errA:=make(chanerror, 1)
errB:=make(chanerror, 1)
// Populate streams with test datagofunc() {
deferclose(streamA)
deferclose(errA)
fori:=0; i<50; i+=2 {
streamA<-fmt.Sprintf("item_%03d", i)
}
}()
gofunc() {
deferclose(streamB)
deferclose(errB)
fori:=1; i<50; i+=2 {
streamB<-fmt.Sprintf("item_%03d", i)
}
}()
// Use channel-based result processing for parallel handlingresultFunc, resultChan:=diff.StringResultChan()
varwg sync.WaitGroupwg.Add(1)
// Process results in parallelgofunc() {
deferwg.Done()
forresult:=rangeresultChan {
fmt.Printf("Difference: %s %s\n", result.D, result.S)
}
}()
// Start diff operationgofunc() {
deferclose(resultChan)
_, err:=diff.Strings(
context.Background(),
streamA, streamB,
errA, errB,
resultFunc,
)
iferr!=nil {
fmt.Printf("Diff error: %v\n", err)
}
}()
wg.Wait()
fmt.Println("Diff processing complete")
}

Performance Considerations

• Memory Usage: Configure ChunkSize based on available memory (larger chunks = less I/O, more memory)
• Parallelism: Increase NumWorkers on multi-core systems
• Temporary Storage:
  • Explicitly set TempFilesDir to a known disk-backed directory for large datasets
  • On Linux, prefer /var/tmp over /tmp (which may be tmpfs/memory-backed)
  • Use fast storage (SSD recommended) for temporary files
• Channel Buffers: Tune buffer sizes based on your producer/consumer patterns

Error Handling

The library uses Go's standard error handling patterns. Always check the error channel:

sorter, outputChan, errChan:=extsort.Ordered(inputChan, nil)
gosorter.Sort(context.Background())
foritem:=rangeoutputChan {
// Process sorted item
}
iferr:=<-errChan; err!=nil {
// Handle errorlog.Fatal(err)
}

Limitations

• Not Stable: The sort is not stable (equal elements may change relative order)
• Disk Space: Requires temporary disk space approximately equal to input data size
• Memory: Minimum memory usage depends on chunk size configuration

License

This project is licensed under the Apache License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.