Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions Libraries/MLXLMCommon/ConcurrentError.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
import Foundation
import MLX

package final class ThreadSafeError: @unchecked Sendable {
package let lock = NSLock()
package var error: Swift.Error?

package init() {}

package func catchError(_ block: () throws -> Void) {
do {
try MLX.withError {
try block()
}
} catch {
lock.withLock {
if self.error == nil { self.error = error }
}
}
}

package func check() {
if let error = error {
fatalError("MLX SSD Streaming Error: \(error.localizedDescription). (The model safetensors file may be corrupted, truncated, or incomplete).")
}
Comment on lines +22 to +25

Copilot AI Apr 27, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PR description says the new wrapper will trigger a single clean crash "on the main thread", but check() calls fatalError directly on whichever thread invokes check(). If the main-thread behavior is required, check() (or its callers) should explicitly dispatch/ensure main-thread execution; otherwise the PR description should be updated to match the actual behavior.

Copilot uses AI. Check for mistakes.
}
}
28 changes: 28 additions & 0 deletions Libraries/MLXLMCommon/SwitchLayers.swift
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
if !specTargets.isEmpty {
let bpe = _stackedBytesPerExpert
let downBpe = _stackedDownBytesPerExpert
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: specTargets.count * 3) { [specTargets] i in
errState.catchError {
let mIdx = i / 3
let proj = i % 3
let info = specTargets[mIdx]
Expand Down Expand Up @@ -301,7 +303,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadIntoOffset(self._stackedDown!, safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(info.expertId), dstOffset: info.slot * downBpe)
}
}
}
errState.check()
}

if idx.size == 0 {
Expand Down Expand Up @@ -372,7 +376,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
if !missesNeedingPread.isEmpty {
let bpe = _stackedBytesPerExpert
let downBpe = _stackedDownBytesPerExpert
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: missesNeedingPread.count * 3) { [missesNeedingPread] i in
errState.catchError {
let mIdx = i / 3
let proj = i % 3
let info = missesNeedingPread[mIdx]
Expand All @@ -399,7 +405,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadIntoOffset(self._stackedDown!, safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(info.expertId), dstOffset: info.slot * downBpe)
}
}
}
errState.check()
}
_previousExpertIds = ranges.map { $0.id }

Expand Down Expand Up @@ -596,7 +604,9 @@ public class SwitchGLU: Module, @unchecked Sendable {

// Full concurrent pread (baseline path)
let totalReads = ranges.count * 3
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: totalReads) { [ranges] i in
errState.catchError {
let expertIdx = i / 3
let projIdx = i % 3
let r = ranges[expertIdx]
Expand All @@ -611,7 +621,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadInto(self._persistentDown![expertIdx], safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(r.id))
}
}
}
errState.check()

// Store routing for next token's predictions
_previousExpertIds = ranges.map { $0.id }
Expand Down Expand Up @@ -640,7 +652,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
if let prevIds = _previousExpertIds {
let specCount = min(prevIds.count, maxBuffers)
let specReads = specCount * 3
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: specReads) { i in
errState.catchError {
let slot = i / 3
let proj = i % 3
let expertId = prevIds[slot]
Expand All @@ -655,7 +669,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadInto(self._persistentDown![slot], safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(expertId))
}
}
}
errState.check()
}

// Sync on idx (blocks until GPU finishes attention + router)
Expand Down Expand Up @@ -719,7 +735,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
// Pread only misses (~30% of experts, ~6 reads at QD=6)
if !missInfo.isEmpty {
let totalMissReads = missInfo.count * 3
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: totalMissReads) { [missInfo] i in
errState.catchError {
let mIdx = i / 3
let proj = i % 3
let info = missInfo[mIdx]
Expand All @@ -740,7 +758,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
tensorName: downSSD.tensorName,
expertIndex: UInt32(info.expertId))
}
}
}
errState.check()
}
} else {
// No predictions available — full pread fallback
Expand All @@ -750,7 +770,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
usedDown.append(_persistentDown![i])
}
let totalReads = ranges.count * 3
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: totalReads) { [ranges] i in
errState.catchError {
let expertIdx = i / 3
let projIdx = i % 3
let r = ranges[expertIdx]
Expand All @@ -765,7 +787,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadInto(self._persistentDown![expertIdx], safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(r.id))
}
}
}
errState.check()
}

// Update routing for next token's predictions
Expand Down Expand Up @@ -814,7 +838,9 @@ public class SwitchGLU: Module, @unchecked Sendable {

// Concurrent pread (same as fast path)
let totalReads = ranges.count * 3
let errState = ThreadSafeError()
DispatchQueue.concurrentPerform(iterations: totalReads) { [ranges] i in
errState.catchError {
let expertIdx = i / 3
let projIdx = i % 3
let r = ranges[expertIdx]
Expand All @@ -829,7 +855,9 @@ public class SwitchGLU: Module, @unchecked Sendable {
MLXFast.preadInto(downBuffers[expertIdx], safetensorsPath: downSSD.path,
tensorName: downSSD.tensorName, expertIndex: UInt32(r.id))
}
}
}
errState.check()

// Lazy compute (no eval — next layer forces it)
let xGate = qGate.computeExperts(x, buffers: gateBuffers, ranges: ranges)
Expand Down
52 changes: 52 additions & 0 deletions Tests/MLXLMTests/CorruptSafetensorsTests.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
import Foundation
import MLX
import MLXLMCommon
import Testing

@Suite
struct CorruptSafetensorsTests {
@Test
func testDeadlock() throws {
let tempDir = FileManager.default.temporaryDirectory
let pathUrl = tempDir.appendingPathComponent("dummy_corrupt_\(UUID().uuidString).safetensors")
let path = pathUrl.path

let dict: [String: MLXArray] = ["weight": MLXArray.zeros([256, 1024])]
try MLX.save(arrays: dict, url: pathUrl)

defer {
try? FileManager.default.removeItem(at: pathUrl)
}

let fd = open(path, O_RDWR)
#expect(fd != -1)
guard fd != -1 else { return }

defer {
let closeResult = close(fd)
#expect(closeResult == 0)
}

let truncateResult = ftruncate(fd, 100)
#expect(truncateResult == 0)
guard truncateResult == 0 else { return }

let dst = MLXArray.zeros([256, 1024])
dst.eval()

struct SendableArray: @unchecked Sendable {
let array: MLXArray
}
let sendableDst = SendableArray(array: dst)

let errState = ThreadSafeError()

DispatchQueue.concurrentPerform(iterations: 16) { i in
errState.catchError {
MLXFast.preadIntoOffset(sendableDst.array, safetensorsPath: path, tensorName: "weight", expertIndex: UInt32(i), dstOffset: i * 1024 * 4)
}
}

#expect(errState.error != nil)
}
}
Loading