ANE KernelDMA 1 MiB prefetch-notch profiler (M1 Max / M3 Ultra / M4 / M5). eval_us, not wall clock. No ANEC 1MiB DMA-split flag.

README.md

KernelDMA 1 MiB prefetch notch — portable profile

Post: Getting 50 GB/s Back Out of the ANE (Eileen Yoon, 2026-08-10). This repo’s Bonsai 27B export is not on the notch. Use this note plus the profiler to check other chips (M1 Max, M3 Ultra, M4, M5, …) and to confirm ANEC has no 1 MiB KernelDMA-split flag.

Gist (same scripts): https://gist.github.com/Anemll/39f657dc48b402747bdd96458edd415f

Combined results (2026-09-11)

Stock pair: Cout=4096, Cin 2016 vs 2048, FP16, native eval_us (not wall clock).

Chip	Host	2016	2048	ratio	class
M1 Max	MacBookPro18,2	458 µs / 36.1 GB/s	1381 µs / 12.2 GB/s	3.02×	m3_class_notch
M4 Pro	Mac16,11	201 / 82.0	649 / 25.8	3.22×	m3_class_notch
article M3	—	~44.5 GB/s	~16.9 GB/s	2.63×	m3_class_notch
M5	Mac17,2	356 / 46.3	746 / 22.5	2.09×	m3_class_notch
M5 (prior)	portable host	354 / 46.7	785 / 21.4	2.22×	m3_class_notch
M1	MacBookPro17,1	520 / 31.7	493 / 34.0	0.95×	no_notch
M5 Max	Mac17,6	189 / 87.5	190 / 88.4	1.01×	no_notch

Do not put M3 Ultra on this 16-core pair. That shape is 0.5 MiB/core on 2×16 ANE. Ultra’s result is the dual lattice:

Chip	Host	lattice	off → on	ratio	class
M3 Ultra	Mac15,14 --preset dual	8192 × 2016/2048	66 → 42 GB/s	1.61×	milder_1mib_slowdown
M3 Ultra	Mac15,14 --preset dual	4096 × 8128/8192	89 → 51 GB/s	1.77×	milder_1mib_slowdown
M3 Ultra	same, Cin=4096 at 1 MiB	4096 × 4032/4096	—	1.05×	miss
M5 Max	Mac17,6 dual lattice	same shapes	~112–130 GB/s	~1.00×	flat

Notch: M1 Max, M3, M4 Pro, M5. Milder (32-core): M3 Ultra. Flat: M1, M5 Max. Ultra: --preset dual. ANEC: no 1 MiB KernelDMA-split flag (M1, M1 Max, M4 Pro, M5, M5 Max, M3 Ultra).

---

For dense FP16 on 16 ANE cores (one cluster), each core’s KernelDMA payload is

bytes/core = (Cout / 16) × Cin × 2

When that product is exactly k × 1 MiB, M3 streaming drops from ~45–60 GB/s to 17–19 GB/s (suspected 14-bit / 0x4000-line prefetch wrap). Recovery is ±256 lines (16 KiB). The trigger is the product, not “hidden = 2048” by itself.

At Cout=4096, Cin=2048 is 1.0 MiB/core; Cin=2016 is 0.984375 MiB and recovers. If both dims are multiples of 1024, the extra test is (Cin/1024)×(Cout/1024) is a multiple of 8. Hidden % 1024 == 0 is not sufficient (2048×2048 misses; 5120×17408 misses).

Dual cluster (Ultra)

M1/M2/M3 Ultra is 2×16 ANE cores. If Cout is striped across both clusters, per-core bytes are

bytes/core = (Cout / 32) × Cin × 2

The stock article pair Cout=4096 × Cin=2048 is then 0.5 MiB/core — off the lattice — and will report no_notch even when the chip still has the bug. Probe the 32-core lattice (--preset dual):

lattice	Cout × Cin off/on	32c MiB/core	M3 Ultra 2026-09-11
16c stock	4096 × 2016/2048	0.49 / 0.5	off lattice (~32 GB/s both; ignore)
32c 1 MiB	8192 × 2016/2048	0.98 / 1.0	milder 1.61× (66 vs 42 GB/s)
32c 1 MiB	4096 × 4032/4096	0.98 / 1.0	miss (1.05×) — same bytes, Cin=4096
32c 2 MiB	4096 × 8128/8192	1.98 / 2.0	milder 1.77× (89 vs 51 GB/s)
32c 2 MiB	8192 × 4032/4096	1.97 / 2.0	miss (1.03×)

Ultra’s cliff is real but milder than single-cluster M3/M4/M5 (~2.1–3.2×). It is not “any k×1 MiB/core”: Cin=2048 or 8192 hit; Cin=4096 at the same byte count did not. Pad those hitting GEMMs (e.g. 2016 or 8128), not the stock 4096×2048 pair.

--preset auto selects dual when guessed_chip contains Ultra (or --label m3u). --cores 32 only changes MiB/core math; it does not by itself scan extra shapes — use --preset dual.

M5 Max (Mac17,6) ran the same dual lattice flat (~112–130 GB/s, ratio ~1.00). Treat Max as no_notch until a hitting tile shows up; do not assume Ultra == Max.

Single-cluster receipts

The 16-core table is at the top. M3 Ultra is not in that table; use --preset dual (1.61× / 1.77× milder).

Use native eval_us, not wall clock. Wall-clock ~58 ms both is invalid.

Bonsai Tail43 windows are 2.8–64 KiB, not 1 MiB. Do not split Tail43 for this bug.

Run

Python 3.9, coremltools, Apple Silicon. Never hidden=64.

# host-only math + chip + compiler inventory
python3.9 profile_kernel_dma_mib.py --host-only --label m4p

# native 16-core article pair
python3.9 profile_kernel_dma_mib.py --out ./kernel_dma_mib_m4p --label m4p --preset stock

# Ultra / dual-cluster scan (writes summary.json + one subdir per lattice)
python3.9 profile_kernel_dma_mib.py --out ./kernel_dma_mib_m3u --label m3u --preset dual

# ANEC split-flag probe
python3.9 profile_kernel_dma_mib.py --probe-anec

--preset auto (default): dual on Ultra, stock otherwise.

How to read class_name (ratio = last Cin / first Cin eval_us):

ratio	median eval_us	class
≥ 2.0	hundreds of µs	m3_class_notch
≥ 1.3	hundreds of µs	milder_1mib_slowdown
~1.0	tens of ms	host_dominated_invalid — you timed the host
~1.0	hundreds of µs	no_notch on that tile

Do not compare wall-clock predict(). Re-run if aned restarts.

Does ANEC split 1 MiB KernelDMA?

No flag found (M1, M1 Max, M4 Pro, M5, M5 Max, M3 Ultra).

ANECCreateCompilerOptionsCFString serializes several split keys. They are spatial / batch / channel / HWX section / LUT-palette splits, not “keep KernelDMA off k × 1 MiB”:

Dict key	CLI	What it actually is
SplitKernelSection	--split-kernel-section	__KERN section packing
EnableSegmentAwareKernelSectionSplit	--enable-segment-aware-kernel-section-split	same family
EnableSpatialSplitInX	--fspatial-split-in-x	activation spatial split
EnableKernelSplitForMultiPaletteLUT	--enable-kernel-split-for-multi-palette-lut	palettized LUT
EnableL2BatchSplitting	--enable-l2-batch-splitting	batch
EnableGlobalChannelSplitting	--enable-global-channel-splitting	channel
EnableForcedMaximalBondedSplit	--enable-forced-maximal-bonded-split	bonded nets
DisableCachePrefetchMask	--disable-cache-prefetch-mask	CacheDMA prefetch (occupancy), not 1 MiB KernelDMA

Fabricated keys SplitKernelDMA, EnableKernelDMASplit, MaxKernelDMASize, AvoidOneMiBKernelDMA, SplitCoeffBuffer are dropped (do not appear in the serialized option string).

The workaround is still graph/TD level: split or pad the GEMM so each core’s coeff window is not k × 1 MiB. On Ultra, pad the 32-core hitting tiles (8192×2048, 4096×8192), not 4096×2048. That is not an ANEC toggle. Do not vault-patch TDs in this tree.

Models that hit (dense FP16, 16-core)

Hit iff Cin × Cout = k × 8,388,608. Typical hits: Llama 3.2 1B FFN 2048×8192, Llama 3.2 3B FFN 3072×8192, Llama 3.1 8B Q 4096×4096 and FFN 4096×14336, Qwen3-8B Q and FFN. Miss: Qwen3-0.6B/1.7B, Gemma 3 270M/1B/4B, Bonsai 5120×17408, packed Tail43.

On Ultra, re-evaluate with Cout/32: 4096×2048 misses; 8192×2048 and 4096×8192 are the measured hits.

Files

File	Role
profile_kernel_dma_mib.py	portable CLI (--preset stock|dual|auto)
kernel_dma_mib_host.m	_ANEClient eval_us
anec_options_probe.m	ANEC option serializer
test_profile_kernel_dma_mib.py	host tests

In this tree they live under Bonsai/target/ (doc: Bonsai/KERNEL_DMA_MIB.md). Gist: https://gist.github.com/Anemll/39f657dc48b402747bdd96458edd415f

# this repo
PYTHONPATH=Bonsai:Bonsai/target python3.9 \
  Bonsai/target/profile_kernel_dma_mib.py --out Bonsai/generated/kernel_dma_mib_m3u \
  --label m3u --preset dual

anec_options_probe.m

// Serialize ANECCompile flags through ANECCreateCompilerOptionsCFString.
// Compile only: no aned, no vault, no ANE evaluate.
#import <Foundation/Foundation.h>
#import <dlfcn.h>
int main(int argc, char **argv) {
  if (argc < 2) {
    fprintf(stderr, "usage: %s <flags.json>\n", argv[0]);
    return 2;
  }
  @autoreleasepool {
    NSData *raw = [NSData dataWithContentsOfFile:[NSString stringWithUTF8String:argv[1]]];
    if (raw == nil) {
      fprintf(stderr, "flags json missing\n");
      return 2;
    }
    NSError *err = nil;
    id parsed = [NSJSONSerialization JSONObjectWithData:raw options:0 error:&err];
    if (![parsed isKindOfClass:[NSDictionary class]]) {
      NSLog(@"flags json %@", err);
      return 2;
    }
    typedef CFStringRef (*OptsFn)(CFDictionaryRef);
    OptsFn fn = (OptsFn)dlsym(RTLD_DEFAULT, "ANECCreateCompilerOptionsCFString");
    if (fn == NULL) {
      printf("symbol=missing dlerror=%s\n", dlerror());
      return 1;
    }
    CFStringRef text = fn((__bridge CFDictionaryRef)parsed);
    if (text == NULL) {
      printf("symbol=ok serialized=null\n");
      return 0;
    }
    NSLog(@"%@", (__bridge NSString *)text);
    printf("symbol=ok serialized=%s\n",
           [(__bridge NSString *)text UTF8String]);
    return 0;
  }
}

kernel_dma_mib_host.m

// Compact ANE eval_us host for the KernelDMA 1 MiB notch profiler.
// Ordinary compiler MIL only. No custom HWX, no vault, no 0-4 dump.
// Usage: kernel_dma_mib_host <model.mlmodelc> <warm> <timed>
#import <CoreVideo/CoreVideo.h>
#import <Foundation/Foundation.h>
#import <IOSurface/IOSurfaceObjc.h>
#import <objc/runtime.h>
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

@interface _ANEClient : NSObject
+ (id)sharedConnection;
- (bool)compileModel:(id)model options:(id)options qos:(unsigned int)qos error:(id *)error;
- (bool)loadModel:(id)model options:(id)options qos:(unsigned int)qos error:(id *)error;
- (_Bool)evaluateWithModel:(id)model options:(id)options request:(id)request
                       qos:(unsigned int)qos error:(id *)error;
@end

@interface _ANEModel : NSObject
@property(retain, nonatomic) NSDictionary *modelAttributes;
+ (id)modelAtURL:(id)url key:(id)key;
@end

@interface _ANEIOSurfaceObject : NSObject
+ (id)objectWithIOSurface:(IOSurface *)iosurface;
@end

@interface _ANERequest : NSObject
+ (id)requestWithInputs:(id)inputs inputIndices:(id)indices outputs:(id)outputs
          outputIndices:(id)indices2 perfStats:(id)stats procedureIndex:(id)index;
@end

static NSDictionary *opts(void) {
  const char *ident = getenv("ANE_IDENTITY");
  NSString *identity = (ident && ident[0]) ? [NSString stringWithUTF8String:ident]
                                          : @"kernel_dma_mib_probe";
  return @{@"kANEFModelIdentityStrKey" : identity, @"kANEFModelType" : @"kANEFModelMIL"};
}

static IOSurface *surface_from_dict(NSDictionary *dict, NSString *name) {
  NSDictionary *props = @{
    IOSurfacePropertyKeyWidth : dict[@"Width"] ?: @1,
    IOSurfacePropertyKeyHeight : dict[@"Height"] ?: @1,
    IOSurfacePropertyKeyBytesPerElement : @2,
    IOSurfacePropertyKeyPixelFormat : @(kCVPixelFormatType_OneComponent16Half),
    IOSurfacePropertyKeyName : name,
  };
  if (dict[@"Batches"]) {
    NSNumber *tensor_size = @([dict[@"Batches"] intValue] * [dict[@"BatchStride"] intValue]);
    props = @{
      IOSurfacePropertyKeyWidth : tensor_size,
      IOSurfacePropertyKeyHeight : @1,
      IOSurfacePropertyKeyName : name,
      IOSurfacePropertyKeyPixelFormat : @(kCVPixelFormatType_OneComponent16Half),
    };
  }
  return [[IOSurface alloc] initWithProperties:props];
}

static void fill_constant(IOSurface *ios, NSDictionary *d, float value) {
  NSUInteger b = MAX(1, [d[@"Batches"] integerValue]);
  NSUInteger c = MAX(1, [d[@"Channels"] integerValue]);
  NSUInteger h = MAX(1, [d[@"Height"] integerValue]);
  NSUInteger w = MAX(1, [d[@"Width"] integerValue]);
  NSUInteger rs = [d[@"RowStride"] unsignedIntegerValue] ?: w * 2;
  NSUInteger ps = [d[@"PlaneStride"] unsignedIntegerValue] ?: h * rs;
  NSUInteger bs = [d[@"BatchStride"] unsignedIntegerValue] ?: c * ps;
  IOSurfaceRef ref = (__bridge IOSurfaceRef)ios;
  IOSurfaceLock(ref, 0, NULL);
  uint8_t *base = IOSurfaceGetBaseAddress(ref);
  memset(base, 0, IOSurfaceGetAllocSize(ref));
  __fp16 half = (__fp16)value;
  for (NSUInteger nb = 0; nb < b; nb++)
    for (NSUInteger nc = 0; nc < c; nc++)
      for (NSUInteger nh = 0; nh < h; nh++)
        for (NSUInteger nw = 0; nw < w; nw++)
          memcpy(base + nb * bs + nc * ps + nh * rs + nw * 2, &half, 2);
  IOSurfaceUnlock(ref, 0, NULL);
}

int main(int argc, char **argv) {
  if (argc != 4) {
    fprintf(stderr, "usage: %s model.mlmodelc warm timed\n", argv[0]);
    return 2;
  }
  int warm = atoi(argv[2]), timed = atoi(argv[3]);
  if (warm < 0 || timed < 1 || timed > 128) return 2;
  @autoreleasepool {
    NSError *err = nil;
    NSString *path = [NSString stringWithUTF8String:argv[1]];
    id client = [_ANEClient sharedConnection];
    id model = [_ANEModel modelAtURL:[NSURL fileURLWithPath:path] key:@"ANE_model"];
    if (![client compileModel:model options:opts() qos:21 error:&err] || err) {
      NSLog(@"compile failed %@", err);
      return 3;
    }
    err = nil;
    if (![client loadModel:model options:opts() qos:21 error:&err] || err) {
      NSLog(@"load failed %@", err);
      return 4;
    }
    NSDictionary *attr = [model modelAttributes];
    NSArray *procedures = attr[@"ANEFModelDescription"][@"ANEFModelProcedures"];
    NSArray *networks = attr[@"NetworkStatusList"];
    if (procedures.count != 1 || networks.count != 1) return 5;
    NSArray *inputs = networks[0][@"LiveInputList"];
    NSArray *outputs = networks[0][@"LiveOutputList"];
    if (inputs.count != 1 || outputs.count != 1) return 5;
    IOSurface *in = surface_from_dict(inputs[0], inputs[0][@"Name"]);
    IOSurface *out = surface_from_dict(outputs[0], outputs[0][@"Name"]);
    if (!in || !out) return 6;
    fill_constant(in, inputs[0], 0.01f);
    id request = [_ANERequest
        requestWithInputs:@[ [_ANEIOSurfaceObject objectWithIOSurface:in] ]
             inputIndices:procedures[0][@"ANEFModelInputSymbolIndexArray"]
                  outputs:@[ [_ANEIOSurfaceObject objectWithIOSurface:out] ]
            outputIndices:procedures[0][@"ANEFModelOutputSymbolIndexArray"]
                perfStats:@[]
          procedureIndex:@0];
    NSMutableArray *samples = [NSMutableArray array];
    int total = warm + timed;
    for (int i = 0; i < total; i++) {
      err = nil;
      uint64_t start = clock_gettime_nsec_np(CLOCK_UPTIME_RAW);
      BOOL ok = [client evaluateWithModel:model
                                  options:@{@"kANEFDisableIOFencesUseSharedEventsKey" : @0}
                                  request:request
                                      qos:0x21
                                    error:&err];
      double us = (clock_gettime_nsec_np(CLOCK_UPTIME_RAW) - start) / 1000.0;
      if (!ok || err) {
        NSLog(@"evaluate failed %@", err);
        return 8;
      }
      if (i >= warm) [samples addObject:@(us)];
    }
    NSMutableString *json = [NSMutableString stringWithString:@"{\"eval_us\":["];
    for (NSUInteger i = 0; i < samples.count; i++) {
      if (i) [json appendString:@","];
      [json appendFormat:@"%.3f", [samples[i] doubleValue]];
    }
    [json appendString:@"]}"];
    printf("DMA_JSON %s\n", json.UTF8String);
    return 0;
  }
}

profile_kernel_dma_mib.py

"""Portable KernelDMA 1 MiB prefetch-notch profiler (M1 / M3 / M4 / M5).

Article: https://eiln.github.io/posts/ane-dma.html

Single-cluster (16 ANE cores) — default:

  python3.9 profile_kernel_dma_mib.py --host-only --label m4p
  python3.9 profile_kernel_dma_mib.py --out ./kernel_dma_mib_m4p --label m4p

Dual-cluster Ultra / 32-core lattice (M3 Ultra, M1/M2 Ultra):

  python3.9 profile_kernel_dma_mib.py --host-only --preset dual --label m3u
  python3.9 profile_kernel_dma_mib.py --out ./kernel_dma_mib_m3u --preset dual --label m3u

Wall-clock Core ML predict() is host-dominated and invalid. This script
times `_ANEClient evaluateWithModel` (eval_us). Do not dump layers 0-4.
"""
from __future__ import annotations

import argparse
import copy
import json
import os
import platform
import statistics
import subprocess
import sys
from pathlib import Path

try:
    import numpy as np
except ImportError:
    np = None

try:
    import coremltools as ct
    from coremltools.converters.mil import Builder as mb
    from coremltools.converters.mil.mil import types
    from coremltools.models.utils import compile_model
except ImportError:
    ct = None
    mb = None
    types = None
    compile_model = None

try:
    from dspark_runtime import acquire_native_lock
except ImportError:
    acquire_native_lock = None

HERE = Path(__file__).resolve().parent
HOST_M = HERE / 'kernel_dma_mib_host.m'
MIB = 1 << 20
LINE = 64
CORES = 16  # single ANE cluster
DEFAULT_COUT = 4096
DEFAULT_CINS = (2016, 2048)
ARTICLE = 'https://eiln.github.io/posts/ane-dma.html'

SERIALIZED_SPLIT_RELATED = (
    '--enable-global-channel-splitting',
    '--enable-forced-maximal-bonded-split=true',
    '--fspatial-split-in-x',
    '--fkernel-rewind=enabled',
    '--split-kernel-section=false',
    '--disable-cache-prefetch-mask=0',
    '--global-refinement-in-spatial-split=true',
    '--enable-segment-aware-kernel-section-split=true',
    '--enable-l2-batch-splitting=true',
    '--enable-kernel-split-for-multi-palette-lut=true',
)
DROPPED_DMA_SPLIT_CANDIDATES = (
    'SplitKernelDMA',
    'EnableKernelDMASplit',
    'MaxKernelDMASize',
    'AvoidOneMiBKernelDMA',
    'SplitCoeffBuffer',
)
NOT_ONE_MIB_DMA_SPLIT = (
    'SplitKernelSection',
    'SpatialSplit',
    'EnableSpatialSplitInX',
    'EnableKernelSplitForMultiPaletteLUT',
    'EnableSegmentAwareKernelSectionSplit',
    'EnableGlobalChannelSplitting',
    'EnableL2BatchSplitting',
    'EnableForcedMaximalBondedSplit',
    'DisableCachePrefetchMask',
)

# Dual-cluster Ultra: 2x16 cores. Per-core bytes use cores=32 if Cout is
# striped across both clusters. Stock 16-core 1 MiB pair is then 0.5 MiB
# and is the wrong probe. Measured 2026-09-11 on M3 Ultra (Mac15,14).
DUAL_LATTICE = (
    dict(name='16c_stock_1mib', cores=16, cout=4096, cins=(2016, 2048),
         note='single-cluster article pair; Ultra off-lattice (0.5 MiB/core)'),
    dict(name='32c_1mib_cin2048', cores=32, cout=8192, cins=(2016, 2048),
         note='32-core 1 MiB; Ultra milder_1mib_slowdown ~1.61x'),
    dict(name='32c_1mib_cin4096', cores=32, cout=4096, cins=(4032, 4096),
         note='same 1 MiB/32-core via Cin=4096; Ultra miss'),
    dict(name='32c_2mib_cin8192', cores=32, cout=4096, cins=(8128, 8192),
         note='32-core 2 MiB; Ultra milder_1mib_slowdown ~1.77x'),
    dict(name='32c_2mib_cin4096', cores=32, cout=8192, cins=(4032, 4096),
         note='same 2 MiB/32-core via Cin=4096; Ultra miss'),
)


def fp16_bytes_per_core(cin, cout, cores=CORES):
    if cin <= 0 or cout <= 0 or cores <= 0 or cout % cores:
        raise ValueError('positive Cin and Cout divisible by core count required')
    return (cout // cores) * cin * 2


def total_fp16_bytes(cin, cout):
    return cin * cout * 2


def on_prefetch_notch(nbytes):
    if nbytes <= 0:
        raise ValueError('empty transfer')
    k = max(1, int(round(nbytes / MIB)))
    return abs(nbytes - k * MIB) == 0


def classify_ratio(ratio, median_us_on=None):
    """Map on/off eval_us ratio to a notch class. Wall-clock ms is invalid."""
    if median_us_on is not None and median_us_on >= 5000:
        return 'host_dominated_invalid'
    if ratio >= 2.0:
        return 'm3_class_notch'
    if ratio >= 1.3:
        return 'milder_1mib_slowdown'
    return 'no_notch'


def sysctl(name):
    run = subprocess.run(['sysctl', '-n', name], capture_output=True, text=True, timeout=5)
    return run.stdout.strip() if run.returncode == 0 else ''


def guess_chip(hw_model, brand):
    brand_l = (brand or '').lower()
    model = hw_model or ''
    text = f'{brand_l} {model.lower()}'
    for name in ('m5', 'm4', 'm3', 'm2', 'm1'):
        if name in text:
            extra = ''
            for suf in ('ultra', 'max', 'pro'):
                if suf in brand_l:
                    extra = ' ' + suf.title()
                    break
            return ('Apple ' + name.upper() + extra).strip()
    if hw_model.startswith('Mac17'):
        return 'Apple M5-class (hw.model)'
    if hw_model.startswith('Mac16'):
        return 'Apple M4-class (hw.model)'
    if hw_model.startswith('Mac15'):
        return 'Apple M3-class (hw.model)'
    if hw_model.startswith('Mac14'):
        return 'Apple M2-class (hw.model)'
    if hw_model.startswith('Mac13') or hw_model.startswith('MacBookPro18'):
        return 'Apple M1-class (hw.model)'
    return 'unknown'


def guess_ane_cores(chip_name):
    text = (chip_name or '').lower()
    if 'ultra' in text:
        return 32
    return 16


def chip_info(label=None):
    brand = sysctl('machdep.cpu.brand_string')
    model = sysctl('hw.model')
    guessed = guess_chip(model, brand)
    cores = guess_ane_cores(guessed)
    return dict(
        hw_model=model,
        brand=brand,
        platform=platform.platform(),
        guessed_chip=guessed,
        guessed_ane_cores=cores,
        dual_cluster=cores >= 32,
        label=label or guessed,
        python=sys.version.split()[0],
    )


def dense_hits(cin, cout, cores=CORES):
    bpc = fp16_bytes_per_core(cin, cout, cores)
    row = dict(
        cin=cin, cout=cout, cores=cores,
        bytes_per_core=bpc, mib_per_core=bpc / MIB,
        on_prefetch_notch=on_prefetch_notch(bpc),
        total_fp16_bytes=total_fp16_bytes(cin, cout),
    )
    if cout % 16 == 0:
        bpc16 = fp16_bytes_per_core(cin, cout, 16)
        row['mib_per_core_16'] = bpc16 / MIB
        row['on_prefetch_notch_16'] = on_prefetch_notch(bpc16)
    if cout % 32 == 0:
        bpc32 = fp16_bytes_per_core(cin, cout, 32)
        row['mib_per_core_32'] = bpc32 / MIB
        row['on_prefetch_notch_32'] = on_prefetch_notch(bpc32)
    return row


def default_preset(label=None):
    chip = chip_info(label)
    if chip['dual_cluster'] or (label or '').lower() in ('m3u', 'm1u', 'm2u', 'ultra'):
        return 'dual'
    return 'stock'


def host_report(cins=DEFAULT_CINS, cout=DEFAULT_COUT, label=None, cores=None):
    chip = chip_info(label)
    cores = int(cores or chip['guessed_ane_cores'])
    points = [dense_hits(cin, cout, cores) for cin in cins]
    return dict(
        article=ARTICLE,
        chip=chip,
        cores=cores,
        cout=cout,
        points=points,
        dual_lattice=list(DUAL_LATTICE),
        compiler=dict(
            has_1mib_kernel_dma_split_flag=False,
            serialized_split_related=list(SERIALIZED_SPLIT_RELATED),
            dropped_dma_split_candidates=list(DROPPED_DMA_SPLIT_CANDIDATES),
            not_one_mib_dma_split=list(NOT_ONE_MIB_DMA_SPLIT),
            note=('ANEC split flags cover spatial/batch/channel/section/LUT. '
                  'Fabricated KernelDMA 1 MiB keys are dropped by '
                  'ANECCreateCompilerOptionsCFString. Workaround is pad/split '
                  'the GEMM so each cluster core is not k*1 MiB. Ultra dual-cluster '
                  'needs the 32-core lattice; stock 16-core 4096x2048 is 0.5 MiB.')),
        reference_m5_evalus=dict(
            d2016_us=353.8, d2048_us=785.5, ratio=2.22, gbps=(46.7, 21.4),
            class_name='m3_class_notch', cores=16,
            host='kernel_dma_mib_host.m'),
        reference_m3_article=dict(
            d2016_gbps=44.5, d2048_gbps=16.93, ratio=2.63,
            class_name='m3_class_notch', cores=16),
        reference_m3_ultra_20260911=dict(
            note=('Mac15,14 2x16 ANE. Do not cite stock 4096x2048 (0.5 MiB/core). '
                  'Hits: 8192x2048 ~1.61x (66 vs 42 GB/s); 4096x8192 ~1.77x (89 vs 51 GB/s). '
                  'Same byte-count with Cin=4096 missed. M5 Max (Mac17,6) flat on dual lattice.')),
    )


def build_host(dest):
    dest = Path(dest)
    run = subprocess.run(
        ['clang', '-O2', '-fobjc-arc', str(HOST_M), '-o', str(dest),
         '-framework', 'Foundation', '-framework', 'CoreVideo',
         '-framework', 'IOSurface',
         '-F/System/Library/PrivateFrameworks', '-framework', 'AppleNeuralEngine'],
        capture_output=True, text=True, timeout=30)
    if run.returncode:
        raise RuntimeError(run.stderr[-4000:])
    return dest


def convert_conv(cin, cout, dest):
    if ct is None or np is None:
        raise RuntimeError('coremltools and numpy required for native compile')
    dest = Path(dest)
    dest.mkdir(parents=True, exist_ok=True)
    pkg, mlc = dest / 'model.mlpackage', dest / 'model.mlmodelc'
    weight = np.full((cout, cin, 1, 1), np.float16(0.001))
    specs = [mb.TensorSpec(shape=(1, cin, 1, 1), dtype=types.fp16)]

    def conv_main(x):
        y = mb.conv(x=x, weight=weight, pad_type='valid', strides=[1, 1])
        return mb.identity(x=y, name='y')

    prog = mb.program(input_specs=specs, opset_version=ct.target.iOS18)(conv_main)
    pipeline = copy.deepcopy(ct.PassPipeline.DEFAULT)
    pipeline.remove_passes(['common::fuse_conv_scale', 'common::fuse_conv_bias'])
    model = ct.convert(
        prog, convert_to='mlprogram', compute_precision=ct.precision.FLOAT16,
        minimum_deployment_target=ct.target.iOS18, pass_pipeline=pipeline,
        skip_model_load=True)
    model.save(str(pkg))
    compile_model(str(pkg), destination_path=str(mlc))
    return mlc


def time_mlmodelc(host, mlc, warm, timed):
    env = dict(os.environ, ANE_IDENTITY='kernel_dma_mib_probe')
    for name in ('ANE_INMEM_HWX', 'ANE_INPUT_SPLIT'):
        env.pop(name, None)
    run = subprocess.run(
        [str(host), str(mlc), str(warm), str(timed)],
        capture_output=True, text=True, timeout=120, env=env)
    if run.returncode:
        raise RuntimeError((run.stderr or run.stdout)[-4000:])
    line = [row for row in run.stdout.splitlines() if row.startswith('DMA_JSON ')]
    if not line:
        raise RuntimeError('native host printed no DMA_JSON')
    payload = json.loads(line[-1][9:])
    samples = [float(x) for x in payload['eval_us']]
    return dict(eval_us_samples=samples, median_eval_us=statistics.median(samples))


def probe_anec(out):
    """Re-serialize split-related keys on this machine's ANECompiler."""
    out = Path(out)
    out.mkdir(parents=True, exist_ok=True)
    src = HERE / 'anec_options_probe.m'
    binary = out / 'anec_options_probe'
    flags = {
        'SplitKernelSection': False,
        'EnableSpatialSplitInX': True,
        'EnableKernelSplitForMultiPaletteLUT': True,
        'GlobalRefinementInSpatialSplit': True,
        'EnableSegmentAwareKernelSectionSplit': True,
        'EnableGlobalChannelSplitting': True,
        'EnableL2BatchSplitting': True,
        'EnableForcedMaximalBondedSplit': True,
        'DisableCachePrefetchMask': 0,
        'EnableKernelRewind': True,
        'SplitKernelDMA': True,
        'EnableKernelDMASplit': True,
        'MaxKernelDMASize': 1048575,
        'AvoidOneMiBKernelDMA': True,
        'SplitCoeffBuffer': True,
    }
    flags_path = out / 'flags.json'
    flags_path.write_text(json.dumps(flags) + '\n')
    if not src.is_file():
        return dict(status='skipped_no_probe_source', path=str(src))
    build = subprocess.run(
        ['clang', '-O2', '-fobjc-arc', str(src), '-o', str(binary),
         '-framework', 'Foundation',
         '-F/System/Library/PrivateFrameworks', '-framework', 'ANECompiler'],
        capture_output=True, text=True, timeout=30)
    if build.returncode:
        return dict(status='probe_build_failed', stderr=build.stderr[-2000:])
    run = subprocess.run(
        [str(binary), str(flags_path)], capture_output=True, text=True, timeout=30)
    serialized = (run.stdout or '') + (run.stderr or '')
    has_dma_split = any(
        token in serialized.lower() for token in (
            'split-kernel-dma', 'kernel-dma-split', 'max-kernel-dma', 'one-mib'))
    return dict(
        status='probed' if run.returncode == 0 else 'probe_failed',
        returncode=run.returncode,
        serialized=serialized.strip()[-2000:],
        fabricated_keys_dropped=not has_dma_split,
        has_1mib_kernel_dma_split_flag=bool(has_dma_split),
    )


def _payload_gbps(cin, cout, median_us):
    return (total_fp16_bytes(cin, cout) / 1e9) / (median_us / 1e6)


def run_native(out, cins=DEFAULT_CINS, cout=DEFAULT_COUT, warm=3, timed=12,
               label=None, cores=None, host=None):
    out = Path(out).resolve()
    if out.exists():
        raise FileExistsError('fresh kernel-DMA profile directory required: %s' % out)
    out.mkdir(parents=True)
    report = host_report(cins, cout, label, cores)
    cores = report['cores']
    report.update(status='preparing', passed=False, metric='native_eval_us')
    (out / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
    lock = None
    try:
        if acquire_native_lock is not None:
            lock = acquire_native_lock(timeout=1)
        if host is None:
            host = build_host(out / 'kernel_dma_mib_host')
        measured = []
        for cin in cins:
            dest = out / ('d%d' % cin)
            print('DMA_NOTCH_COMPILE', cin, 'cout', cout, 'cores', cores, flush=True)
            mlc = convert_conv(cin, cout, dest)
            timed_row = time_mlmodelc(host, mlc, warm, timed)
            gbps = _payload_gbps(cin, cout, timed_row['median_eval_us'])
            row = dense_hits(cin, cout, cores)
            row.update(gbps_weight_payload=gbps, **timed_row)
            measured.append(row)
            print('DMA_NOTCH_TIME', cin, row['median_eval_us'], gbps, flush=True)
        off, on = measured[0], measured[-1]
        slow = on['median_eval_us'] / off['median_eval_us']
        class_name = classify_ratio(slow, on['median_eval_us'])
        report.update(
            status='passed_kernel_dma_profile', passed=True, points=measured,
            ratio_on_over_off=slow,
            ratio_2048_over_2016=slow if set(cins) >= {2016, 2048} else None,
            class_name=class_name,
            m3_class_notch=class_name == 'm3_class_notch',
            milder_1mib_slowdown=class_name == 'milder_1mib_slowdown')
        (out / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
        print('DMA_NOTCH_RESULT', report['status'], slow, class_name, flush=True)
        return report
    except Exception as exc:
        report.update(status='stopped_no_retry', error='%s: %s' % (type(exc).__name__, exc))
        (out / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
        raise
    finally:
        if lock is not None:
            lock.close()


def run_preset(out, preset, warm, timed, label, cores_override=None):
    out = Path(out).resolve()
    if out.exists():
        raise FileExistsError('fresh kernel-DMA profile directory required: %s' % out)
    out.mkdir(parents=True)
    lattices = DUAL_LATTICE if preset == 'dual' else (DUAL_LATTICE[0],)
    host = build_host(out / 'kernel_dma_mib_host')
    summary = []
    for spec in lattices:
        sub = out / spec['name']
        cores = cores_override or spec['cores']
        try:
            report = run_native(
                sub, spec['cins'], spec['cout'], warm, timed, label, cores, host=host)
            summary.append(dict(
                name=spec['name'], note=spec['note'], cores=cores,
                cout=spec['cout'], cins=list(spec['cins']),
                class_name=report.get('class_name'),
                ratio_on_over_off=report.get('ratio_on_over_off'),
                points=[dict(
                    cin=p['cin'], median_eval_us=p.get('median_eval_us'),
                    gbps_weight_payload=p.get('gbps_weight_payload'),
                    mib_per_core=p.get('mib_per_core'),
                    on_prefetch_notch=p.get('on_prefetch_notch'))
                    for p in report.get('points', [])],
            ))
        except Exception as exc:
            summary.append(dict(name=spec['name'], error='%s: %s' % (type(exc).__name__, exc)))
    blob = dict(preset=preset, chip=chip_info(label), scans=summary)
    (out / 'summary.json').write_text(json.dumps(blob, indent=2) + '\n')
    print('DMA_NOTCH_SUMMARY', json.dumps(blob['scans'], indent=2), flush=True)
    return blob


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--host-only', action='store_true')
    parser.add_argument('--out', type=Path)
    parser.add_argument('--probe-anec', action='store_true')
    parser.add_argument('--label', help='override chip label, e.g. m1m / m3u / m4 / m5m')
    parser.add_argument('--cout', type=int, default=DEFAULT_COUT)
    parser.add_argument('--cins', default='2016,2048')
    parser.add_argument('--cores', type=int, default=0,
                        help='ANE cores for MiB/core math (0=auto: 32 if Ultra else 16)')
    parser.add_argument('--preset', choices=('stock', 'dual', 'auto'), default='auto',
                        help='stock=16-core article pair; dual=16c+32c lattice; auto=dual on Ultra')
    parser.add_argument('--warm', type=int, default=3)
    parser.add_argument('--timed', type=int, default=12)
    args = parser.parse_args(argv)
    cins = tuple(int(x) for x in args.cins.split(','))
    cores = args.cores or None
    preset = args.preset
    if preset == 'auto':
        preset = default_preset(args.label)
    if args.host_only or args.out is None:
        report = host_report(cins, args.cout, args.label, cores)
        report['preset'] = preset
        print(json.dumps(report, indent=2))
        if args.out is None and not args.probe_anec:
            return 0
    if args.probe_anec:
        dest = args.out / 'anec_probe' if args.out else Path('anec_dma_split_probe')
        print(json.dumps(probe_anec(dest), indent=2))
    if args.out is not None and not args.host_only:
        if preset == 'dual':
            run_preset(args.out, 'dual', args.warm, args.timed, args.label, cores)
        else:
            run_native(args.out, cins, args.cout, args.warm, args.timed, args.label, cores)
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

test_profile_kernel_dma_mib.py

"""Host-only KernelDMA 1 MiB profiler tests. No native compile."""
from profile_kernel_dma_mib import (
    DEFAULT_CINS, DROPPED_DMA_SPLIT_CANDIDATES, DUAL_LATTICE, MIB,
    NOT_ONE_MIB_DMA_SPLIT, SERIALIZED_SPLIT_RELATED, classify_ratio,
    dense_hits, fp16_bytes_per_core, guess_ane_cores, guess_chip, host_report,
    on_prefetch_notch)


def test_article_pair_is_exact_one_mib_at_2048():
    recovered = dense_hits(2016, 4096, cores=16)
    notch = dense_hits(2048, 4096, cores=16)
    assert recovered['mib_per_core'] == 0.984375
    assert recovered['on_prefetch_notch'] is False
    assert notch['bytes_per_core'] == MIB
    assert notch['on_prefetch_notch'] is True


def test_dual_cluster_stock_pair_is_half_mib():
    row = dense_hits(2048, 4096, cores=32)
    assert row['mib_per_core'] == 0.5
    assert row['on_prefetch_notch'] is False
    assert row['on_prefetch_notch_32'] is False
    assert row['on_prefetch_notch_16'] is True


def test_dual_cluster_8192x2048_is_one_mib():
    off = dense_hits(2016, 8192, cores=32)
    on = dense_hits(2048, 8192, cores=32)
    assert on['mib_per_core'] == 1.0
    assert on['on_prefetch_notch'] is True
    assert off['on_prefetch_notch'] is False


def test_classify_ratio_matches_m3_and_m5_receipts():
    assert classify_ratio(2.63, 996) == 'm3_class_notch'
    assert classify_ratio(2.22, 785) == 'm3_class_notch'
    assert classify_ratio(1.88, 882) == 'milder_1mib_slowdown'
    assert classify_ratio(1.61, 804) == 'milder_1mib_slowdown'
    assert classify_ratio(1.003, 58_000) == 'host_dominated_invalid'
    assert classify_ratio(1.003, 470) == 'no_notch'


def test_guess_chip_and_ane_cores():
    assert guess_chip('MacBookPro18,2', 'Apple M1 Max') == 'Apple M1 Max'
    assert guess_chip('MacBookPro17,1', 'Apple M1') == 'Apple M1'
    assert guess_chip('Mac15,14', 'Apple M3 Ultra') == 'Apple M3 Ultra'
    assert guess_chip('Mac16,7', 'Apple M4') == 'Apple M4'
    assert guess_chip('Mac17,2', 'Apple M5') == 'Apple M5'
    assert guess_ane_cores('Apple M3 Ultra') == 32
    assert guess_ane_cores('Apple M5 Max') == 16
    assert guess_ane_cores('Apple M4 Pro') == 16


def test_host_report_denies_compiler_dma_split_flag():
    report = host_report(DEFAULT_CINS, 4096, label='m5', cores=16)
    assert report['compiler']['has_1mib_kernel_dma_split_flag'] is False
    assert 'SplitKernelDMA' in DROPPED_DMA_SPLIT_CANDIDATES
    assert 'SplitKernelSection' in NOT_ONE_MIB_DMA_SPLIT
    assert '--split-kernel-section=false' in SERIALIZED_SPLIT_RELATED
    assert on_prefetch_notch(fp16_bytes_per_core(4096, 4096, 16)) is True
    assert report['chip']['label'] == 'm5'
    assert len(report['dual_lattice']) == len(DUAL_LATTICE)


def test_bonsai_ffn_is_off_the_lattice():
    row = dense_hits(5120, 17408, cores=16)
    assert row['mib_per_core'] == 10.625
    assert row['on_prefetch_notch'] is False

bash <<'BASH'
set -euo pipefail

workdir="$(mktemp -d "$HOME/kernel-dma-mib.XXXXXX")"
cd "$workdir"

base="https://gist.githubusercontent.com/Anemll/39f657dc48b402747bdd96458edd415f/raw"

for file in profile_kernel_dma_mib.py kernel_dma_mib_host.m anec_options_probe.m; do
  curl -fSL --retry 3 "$base/$file" -o "$file"
done

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install coremltools numpy

python profile_kernel_dma_mib.py \
  --preset auto --out "$workdir/results" 2>&1 | tee run.log

python profile_kernel_dma_mib.py \
  --probe-anec 2>&1 | tee anec-probe.log

echo "Results and logs: $workdir"
BASH

Apple M3 Air 15-inch 2024, 24 GB

Successfully installed PyYAML-6.0.3 attrs-26.1.0 cattrs-26.2.0 coremltools-9.0 mpmath-1.3.0 numpy-2.5.3 packaging-26.3 protobuf-7.36.1 pyaml-26.7.0 sympy-1.14.0 tqdm-4.70.1 typing-extensions-4.16.0
DMA_NOTCH_COMPILE 2016 cout 4096 cores 16
Running MIL frontend_milinternal pipeline: 0 passes [00:00, ? passes/s]
Running MIL default pipeline: 100%|██████████| 91/91 [00:01<00:00, 84.87 passes/s]
Running MIL backend_mlprogram pipeline: 100%|██████████| 12/12 [00:00<00:00, 14961.85 passes/s]
DMA_NOTCH_TIME 2016 382.29200000000003 43.20015067016835
DMA_NOTCH_COMPILE 2048 cout 4096 cores 16
Running MIL frontend_milinternal pipeline: 0 passes [00:00, ? passes/s]
Running MIL default pipeline: 100%|██████████| 91/91 [00:00<00:00, 2575.87 passes/s]
Running MIL backend_mlprogram pipeline: 100%|██████████| 12/12 [00:00<00:00, 22519.75 passes/s]
DMA_NOTCH_TIME 2048 903.3330000000001 18.572570690985494
DMA_NOTCH_RESULT passed_kernel_dma_profile 2.3629398470279264 m3_class_notch