フランチェスキーニソートで配列を並び替える
フランチェスキーニソートを使用する
フランチェスキーニソート (Franceschini sort) は比較回数・要素移動・補助記憶を同時に漸近最適へ近づけるインプレース整列の系統である。
古典的な未解決問題——最悪でも比較 \(O(n \log n)\)・移動 \(O(n)\)・補助記憶 \(O(1)\) を両立できるか——に対し、肯定的な構成を与えたことで知られる。
安定版は同値の相対順序も保ちつつ同じ資源境界を狙う。実用上は定数倍と実装の重さからウィキソートやグレイルソートほどは使われないが、理論上の到達点として重要である。
下記のデモと計測コードは、フランチェスキーニソートが提案された論文の骨格を次のように簡略化した版である。
- バッファの切り出し: 順位おおよそ
n/4の要素をピボットにし、厳密に小さい要素を先頭へ集める。左側（アクティブ）は約n/4、右側（バッファ）は約3n/4になる。 - バッファ付き部分整列: アクティブ区間をバッファ先頭と交換し、そこを高い分岐数の d 分木ヒープソートで整える。分岐数をおよそ
n^{1/4}に取るとヒープの高さが定数に近く、要素あたりの移動が抑えられる。整列後、再び交換してアクティブ位置へ戻す。 - 残りへの再帰: ピボット以上の未整列側へ同じ手順を繰り返す。左はすでに整っており、かつ右のどの要素より小さいので、連結した配列全体が昇順になる。
- 小さな入力: 長さが小さいときは挿入ソート、または同じ d 分木ヒープへフォールバックする。
論文本体では、さらに標本とセグメント構造・ビット符号化（最小／最大要素ブロックの交換でポインタビットを作る）などで移動回数を \(O(n)\) に押し込む。計測コードはその外側の「四分割＋バッファ＋高分岐ヒープ」までを実装している。
procedure dary_heap_sort(A)
d = roughly length(A)^(1/4)
build_max_heap_with_branching_d(A)
for end from length(A)-1 down to 1
swap A[0] with A[end]
sift_down(A, 0, end-1, d)
procedure sort_with_buffer(Active, Buffer)
// |Buffer| >= |Active|
swap Active with Buffer[0 .. |Active|)
dary_heap_sort(Buffer[0 .. |Active|))
swap back
procedure franceschini_sort(A)
n = length(A)
if n is small then
insertion_or_dary_heap_sort(A)
return
pivot = select_kth(A, floor(n/4))
split = stable_gather of elements strictly < pivot to the front
if split = 0 or split > n - split then
dary_heap_sort(A)
return
sort_with_buffer(A[0 .. split), A[split .. n))
franceschini_sort(A[split .. n))
デモでは小さな配列向けに分岐数と閾値を下げている。本番の計測コードはより大きい入力で同じ骨格を動かす。
類似アルゴリズムとの相違点
ウィキソート・グレイルソート・コタソートも原地安定な \(O(n \log n)\) を狙うブロックマージ系だが、内部バッファやキータグで隣接ランをマージする。フランチェスキーニソートは順位分割でバッファ領域を切り出し、高分岐ヒープなどで移動回数そのものを漸近的に減らす点が異なる。
ヒープソートの二分ヒープは移動が \(\Theta(n \log n)\) になりやすい。こちらは分岐数を大きくして高さを抑え、論文の「移動 \(O(n)\)」側の直感に寄せている。
時間計算量および空間計算量を計測する
| Size | Average time (s) | Maximum time (s) | Average memory (KiB) | Maximum memory (KiB) |
|---|---|---|---|---|
| 256 | 0.000007 | 0.000051 | 0 | 0 |
| 512 | 0.000017 | 0.000053 | 0 | 0 |
| 1024 | 0.000035 | 0.000080 | 0 | 0 |
| 2048 | 0.000077 | 0.000185 | 0 | 0 |
| 4096 | 0.000168 | 0.000570 | 0 | 0 |
| 8192 | 0.000360 | 0.000784 | 0 | 0 |
| 16384 | 0.000791 | 0.002214 | 0 | 0 |
| 32768 | 0.001845 | 0.003016 | 0 | 0 |
| 65536 | 0.004259 | 0.006351 | 0 | 0 |
| 131072 | 0.009808 | 0.014092 | 0 | 0 |
| 262144 | 0.022445 | 0.035735 | 0 | 0 |
計測に使用したコードを表示する
set -euo pipefail
WORKDIR="$(mktemp -d)"
trap 'rm -rf "$WORKDIR"' EXIT
cat > "$WORKDIR/Dockerfile" <<'EOF'
FROM swift:6.0
WORKDIR /app
RUN cat > alloc_track.c <<'ALLOC'
#define _GNU_SOURCE
#include <dlfcn.h>
#include <malloc.h>
#include <stdatomic.h>
#include <stddef.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
static atomic_size_t live_bytes = 0;
static atomic_size_t peak_bytes = 0;
static void *(*real_malloc)(size_t) = NULL;
static void *(*real_calloc)(size_t, size_t) = NULL;
static void *(*real_realloc)(void *, size_t) = NULL;
static void (*real_free)(void *) = NULL;
static void init_reals(void) {
if (real_malloc) {
return;
}
real_malloc = (void *(*)(size_t))dlsym(RTLD_NEXT, "malloc");
real_calloc = (void *(*)(size_t, size_t))dlsym(RTLD_NEXT, "calloc");
real_realloc = (void *(*)(void *, size_t))dlsym(RTLD_NEXT, "realloc");
real_free = (void (*)(void *))dlsym(RTLD_NEXT, "free");
}
static void record_alloc(size_t size) {
size_t live = atomic_fetch_add(&live_bytes, size) + size;
size_t peak = atomic_load(&peak_bytes);
while (live > peak) {
if (atomic_compare_exchange_weak(&peak_bytes, &peak, live)) {
break;
}
}
}
void alloc_track_reset_peak(void) {
atomic_store(&peak_bytes, atomic_load(&live_bytes));
}
size_t alloc_track_live(void) { return atomic_load(&live_bytes); }
size_t alloc_track_peak(void) { return atomic_load(&peak_bytes); }
void *malloc(size_t size) {
init_reals();
void *p = real_malloc(size);
if (p) {
record_alloc(malloc_usable_size(p));
}
return p;
}
void *calloc(size_t nmemb, size_t size) {
init_reals();
void *p = real_calloc(nmemb, size);
if (p) {
record_alloc(malloc_usable_size(p));
}
return p;
}
void *realloc(void *ptr, size_t size) {
init_reals();
size_t old_size = 0;
if (ptr) {
old_size = malloc_usable_size(ptr);
}
void *p = real_realloc(ptr, size);
if (p) {
atomic_fetch_sub(&live_bytes, old_size);
record_alloc(malloc_usable_size(p));
} else if (size == 0) {
atomic_fetch_sub(&live_bytes, old_size);
}
return p;
}
void free(void *ptr) {
init_reals();
if (ptr) {
atomic_fetch_sub(&live_bytes, malloc_usable_size(ptr));
real_free(ptr);
}
}
ALLOC
RUN cat > main.swift <<'SWIFT'
import Foundation
#if canImport(Glibc)
import Glibc
#elseif canImport(Darwin)
import Darwin
#endif
@_silgen_name("alloc_track_live") func alloc_track_live() -> Int
@_silgen_name("alloc_track_peak") func alloc_track_peak() -> Int
@_silgen_name("alloc_track_reset_peak") func alloc_track_reset_peak()
extension UnsafeMutableBufferPointer where Element == Int {
func swapAt(_ i: Int, _ j: Int) {
let t = self[i]; self[i] = self[j]; self[j] = t
}
}
let MIN_POWER: Int = 8
let MAX_POWER: Int = 18
let RUNS: Int = 8192
private func franceschini_branch_factor(_ len: Int) -> Int {
if len <= 2 {
return 2
}
var d = 2
while d * d * d * d < len {
d += 1
if d > 64 {
break
}
}
return max(d, 2)
}
private func franceschini_child(_ parent: Int, _ which: Int, _ d: Int) -> Int {
parent * d + 1 + which
}
private func franceschini_sift_down(
_ a: UnsafeMutableBufferPointer<Int>,
_ root: Int,
_ end: Int,
_ d: Int
) {
var root = root
while true {
let first = franceschini_child(root, 0, d)
if first > end {
break
}
var best = first
let last = min(first + d - 1, end)
if first + 1 <= last {
for child in (first + 1)...last {
if a[child] > a[best] {
best = child
}
}
}
if a[root] >= a[best] {
break
}
a.swapAt(root, best)
root = best
}
}
private func franceschini_dary_heap_sort(_ a: UnsafeMutableBufferPointer<Int>) {
let n = a.count
if n <= 1 {
return
}
let d = franceschini_branch_factor(n)
let last_parent = (n - 2) / d
for start in stride(from: last_parent, through: 0, by: -1) {
franceschini_sift_down(a, start, n - 1, d)
}
for end in stride(from: n - 1, through: 1, by: -1) {
a.swapAt(0, end)
if end > 1 {
franceschini_sift_down(a, 0, end - 1, d)
}
}
}
private func franceschini_insertion_sort(_ a: UnsafeMutableBufferPointer<Int>) {
for i in 1..<a.count {
let key = a[i]
var j = i
while j > 0 && a[j - 1] > key {
a[j] = a[j - 1]
j -= 1
}
a[j] = key
}
}
private func franceschini_partition_at(
_ a: UnsafeMutableBufferPointer<Int>,
_ left: Int,
_ right: Int,
_ pivot_index: Int
) -> Int {
a.swapAt(pivot_index, right)
let pivot = a[right]
var store = left
for i in left..<right {
if a[i] < pivot {
a.swapAt(store, i)
store += 1
}
}
a.swapAt(store, right)
return store
}
private func franceschini_quickselect(
_ a: UnsafeMutableBufferPointer<Int>,
_ left: Int,
_ right: Int,
_ k: Int
) {
var left = left
var right = right
while left < right {
let mid = left + (right - left) / 2
if a[right] < a[left] {
a.swapAt(left, right)
}
if a[mid] < a[left] {
a.swapAt(left, mid)
}
if a[right] < a[mid] {
a.swapAt(mid, right)
}
let pivot_index = franceschini_partition_at(a, left, right, mid)
if k == pivot_index {
return
} else if k < pivot_index {
if pivot_index == 0 {
return
}
right = pivot_index - 1
} else {
left = pivot_index + 1
}
}
}
private func franceschini_sort_with_buffer(
_ active: UnsafeMutableBufferPointer<Int>,
_ buffer: UnsafeMutableBufferPointer<Int>
) {
let m = active.count
if m == 0 {
return
}
for i in 0..<m {
let t = active[i]; active[i] = buffer[i]; buffer[i] = t
}
if m <= 32 {
franceschini_insertion_sort(UnsafeMutableBufferPointer(rebasing: buffer[0..<m]))
} else {
franceschini_dary_heap_sort(UnsafeMutableBufferPointer(rebasing: buffer[0..<m]))
}
for i in 0..<m {
let t = active[i]; active[i] = buffer[i]; buffer[i] = t
}
}
private func franceschini_rec(_ a: UnsafeMutableBufferPointer<Int>) {
let n = a.count
if n <= 1 {
return
}
if n <= 64 {
if n <= 32 {
franceschini_insertion_sort(a)
} else {
franceschini_dary_heap_sort(a)
}
return
}
let rank = n / 4
franceschini_quickselect(a, 0, n - 1, rank)
let pivot = a[rank]
var split = 0
for i in 0..<n {
if a[i] < pivot {
a.swapAt(split, i)
split += 1
}
}
if split == 0 || split > n - split {
franceschini_dary_heap_sort(a)
return
}
franceschini_sort_with_buffer(
UnsafeMutableBufferPointer(rebasing: a[0..<split]),
UnsafeMutableBufferPointer(rebasing: a[split..<n])
)
franceschini_rec(UnsafeMutableBufferPointer(rebasing: a[split..<n]))
}
func franceschini_sort(_ a: inout [Int]) {
a.withUnsafeMutableBufferPointer { franceschini_sort($0) }
}
func franceschini_sort(_ a: UnsafeMutableBufferPointer<Int>) {
franceschini_rec(a)
}
func benchmark_sort(_ array: inout [Int]) {
franceschini_sort(&array)
}
func is_non_decreasing(_ a: [Int]) -> Bool {
guard a.count >= 2 else { return true }
for i in 1..<a.count {
if a[i - 1] > a[i] { return false }
}
return true
}
func same_multiset(_ a: [Int], _ b: [Int]) -> Bool {
if a.count != b.count {
return false
}
var left = a
var right = b
left.sort()
right.sort()
return left == right
}
func check_correctness_case(_ label: String, _ input: [Int]) {
var input = input
let original = input
benchmark_sort(&input)
if !is_non_decreasing(input) {
fatalError("correctness case \(label): output is not sorted")
}
if !same_multiset(input, original) {
fatalError("correctness case \(label): elements were lost or added")
}
}
// Skip cases larger than the algorithm's measured size cap (MAX_POWER). That
// cap exists because larger inputs are impractically slow; forcing them here
// would stall the published measurement script before any table rows print.
func check_correctness_case_within_limit(_ label: String, _ input: [Int]) {
if input.count > (1 << MAX_POWER) {
return
}
check_correctness_case(label, input)
}
func few_unique_values(_ size: Int, _ unique: Int, _ seed: UInt64) -> [Int] {
var state = seed
var result = [Int]()
result.reserveCapacity(size)
for _ in 0..<size {
state ^= state << 13
state ^= state >> 7
state ^= state << 17
result.append(Int(state % UInt64(unique)) + 1)
}
return result
}
func run_correctness_checks() {
check_correctness_case("empty", [])
check_correctness_case("single", [42])
check_correctness_case("duplicates", [3, 1, 3, 2, 1, 2])
check_correctness_case("sorted", [1, 2, 3, 4, 5])
check_correctness_case("reverse", [5, 4, 3, 2, 1])
check_correctness_case("all_equal", [7, 7, 7, 7])
check_correctness_case("skewed_range", [1_000_000, 2, 1_000_001, 1, 999_999])
// Static-buffer Grail skips the in-buffer build when key collection is sparse
// (ideal_buffer = false). Exercising that path catches regressions in buffer gating.
check_correctness_case(
"few_keys_len16",
[2, 2, 2, 2, 2, 2, 2, 2, 4, 3, 1, 2, 3, 4, 1, 4]
)
// Seed 0 is a fixed point of the xorshift below, so it would degenerate into
// yet another all-equal case instead of a 4-value mix. Start at 1.
for seed in 1...32 {
check_correctness_case(
"few_keys_len32_seed_\(seed)",
few_unique_values(32, 4, UInt64(seed))
)
}
// Small-input cutoffs (insertion sort below 32 elements, etc.) hide duplicate-key
// bugs in the recursive path, so repeat the duplicate cases at the smallest
// benchmark size, which every algorithm must handle within reasonable time.
check_correctness_case("all_equal_len256", [Int](repeating: 7, count: 256))
for seed in 1...4 {
check_correctness_case(
"few_keys_len256_seed_\(seed)",
few_unique_values(256, 4, UInt64(seed))
)
}
// Blit's equal-key second sweep used to copy the whole range into a fixed
// 512-element swap; lengths above that must still sort without panicking.
// Respect MAX_POWER so algorithms with a low measured-size cap (slow,
// sleep) do not hang here for minutes or months.
check_correctness_case_within_limit("all_equal_len600", [Int](repeating: 7, count: 600))
for seed in 1...4 {
check_correctness_case_within_limit(
"few_keys_len2048_seed_\(seed)",
few_unique_values(2048, 4, UInt64(seed))
)
}
}
func shuffled(_ size: Int, seed: UInt64) -> [Int] {
guard size > 0 else { return [] }
var v = Array(1...size)
var state = seed
if size > 1 {
for i in stride(from: size - 1, through: 1, by: -1) {
state ^= state << 13
state ^= state >> 7
state ^= state << 17
let j = Int(state % UInt64(i + 1))
v.swapAt(i, j)
}
}
return v
}
func micros(_ d: Duration) -> UInt64 {
let c = d.components
let fromSeconds = UInt64(c.seconds) * 1_000_000
let fromAttos = UInt64(max(0, c.attoseconds / 1_000_000_000_000))
return fromSeconds + fromAttos
}
func padLeft(_ value: String, _ width: Int) -> String {
if value.count >= width {
return value
}
return String(repeating: " ", count: width - value.count) + value
}
func formatSeconds(_ micros: UInt64) -> String {
let whole = micros / 1_000_000
let frac = micros % 1_000_000
let fracStr = padLeft(String(frac), 6).replacingOccurrences(of: " ", with: "0")
return "\(whole).\(fracStr)"
}
func input_array(_ size: Int, seed: UInt64) -> [Int] {
shuffled(size, seed: seed)
}
/// Peak heap growth during `benchmark_sort`, in bytes (explicit buffers such as swap).
/// Kept in bytes so the parent can average before rounding; converting to KiB here
/// would truncate sub-KiB buffers to 0 in every run and hide them from the average.
func run_once(size: Int, seed: Int) -> (UInt64, Int) {
var array = input_array(size, seed: UInt64(seed))
let baseBytes = alloc_track_live()
alloc_track_reset_peak()
let start = ContinuousClock.now
benchmark_sort(&array)
let elapsed = ContinuousClock.now - start
let peakBytes = alloc_track_peak()
let auxBytes = max(0, peakBytes - baseBytes)
let expected: [Int] = size > 0 ? Array(1...size) : []
if array != expected {
fatalError("sort failed with seed \(seed) for size \(size)")
}
return (micros(elapsed), auxBytes)
}
func run_child(_ args: [String]) {
let size = Int(args[2])!
let seed = Int(args[3])!
let (elapsedUs, mem) = run_once(size: size, seed: seed)
print("\(elapsedUs) \(mem)")
}
let args = CommandLine.arguments
if args.count > 1 && args[1] == "--run-once" {
run_child(args)
} else {
run_correctness_checks()
print(
"| \(padLeft("Size", 10)) | \(padLeft("Average time (s)", 16)) | \(padLeft("Maximum time (s)", 16)) | \(padLeft("Average memory (KiB)", 20)) | \(padLeft("Maximum memory (KiB)", 20)) |"
)
print("|----------:|----------------:|----------------:|--------------------:|--------------------:|")
for power in MIN_POWER...MAX_POWER {
let size = 1 << power
var totalTime: UInt64 = 0
var maxTime: UInt64 = 0
var totalMem = 0
var maxMem = 0
for seed in 1...RUNS {
let process = Process()
process.executableURL = URL(fileURLWithPath: args[0])
process.arguments = ["--run-once", "\(size)", "\(seed)"]
let stdout = Pipe()
let stderr = Pipe()
process.standardOutput = stdout
process.standardError = stderr
do {
try process.run()
} catch {
fatalError("failed to run benchmark child process: \(error)")
}
process.waitUntilExit()
if process.terminationStatus != 0 {
let err = String(data: stderr.fileHandleForReading.readDataToEndOfFile(), encoding: .utf8) ?? ""
fatalError("benchmark child process failed: \(err)")
}
let data = stdout.fileHandleForReading.readDataToEndOfFile()
let stdoutText = String(data: data, encoding: .utf8) ?? ""
let fields = stdoutText.split(whereSeparator: \.isWhitespace)
guard fields.count >= 2,
let elapsedUs = UInt64(fields[0]),
let auxMem = Int(fields[1]) else {
fatalError("invalid child process output: \(stdoutText)")
}
totalTime += elapsedUs
if elapsedUs > maxTime {
maxTime = elapsedUs
}
totalMem += auxMem
if auxMem > maxMem {
maxMem = auxMem
}
}
let avgTime = totalTime / UInt64(RUNS)
// Memory is summed in bytes and converted to KiB once, after averaging.
let avgMemKb = totalMem / RUNS / 1024
let maxMemKb = maxMem / 1024
print(
"| \(padLeft(String(size), 10)) | \(padLeft(formatSeconds(avgTime), 16)) | \(padLeft(formatSeconds(maxTime), 16)) | \(padLeft(String(avgMemKb), 20)) | \(padLeft(String(maxMemKb), 20)) |"
)
}
}
SWIFT
RUN clang -O2 -fPIC -shared alloc_track.c -o liballoc_track.so -ldl
RUN swiftc -Ounchecked -whole-module-optimization \
main.swift \
-o swift-benchmark \
-L. -lalloc_track \
-Xlinker -rpath -Xlinker /app
ENV LD_PRELOAD=/app/liballoc_track.so
CMD ["./swift-benchmark"]
EOF
docker build -t swift-benchmark "$WORKDIR"
docker run --rm --init swift-benchmark