connection updates
This commit is contained in:
+79
-30
@@ -437,40 +437,78 @@ private final class PersistentMySQLSessions {
|
||||
let connection: MySQLWireConnection
|
||||
}
|
||||
|
||||
private let queue = DispatchQueue(label: "DBSyn.MySQLSessions")
|
||||
private var connections: [UUID: Entry] = [:]
|
||||
// Each profile has its own serial executor. An unreachable server cannot
|
||||
// block requests to any other saved connection.
|
||||
private final class SessionSlot {
|
||||
let queue: DispatchQueue
|
||||
var entry: Entry?
|
||||
// Only accessed on queue except generation, which is lock protected.
|
||||
private let generationLock = NSLock()
|
||||
private var currentGeneration: UInt64 = 0
|
||||
|
||||
init(id: UUID) {
|
||||
queue = DispatchQueue(label: "DBSyn.MySQLSession.\(id.uuidString)")
|
||||
}
|
||||
func generation() -> UInt64 {
|
||||
generationLock.lock(); defer { generationLock.unlock() }
|
||||
return currentGeneration
|
||||
}
|
||||
func invalidate() {
|
||||
generationLock.lock(); currentGeneration &+= 1; generationLock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
private let slotsLock = NSLock()
|
||||
private var slots: [UUID: SessionSlot] = [:]
|
||||
|
||||
private func slot(for id: UUID) -> SessionSlot {
|
||||
slotsLock.lock(); defer { slotsLock.unlock() }
|
||||
if let existing = slots[id] { return existing }
|
||||
let created = SessionSlot(id: id)
|
||||
slots[id] = created
|
||||
return created
|
||||
}
|
||||
|
||||
func execute(profile: ConnectionProfile, database: String,
|
||||
password: String, sql: String) throws -> QueryResult {
|
||||
try queue.sync {
|
||||
// A changed connection profile or password requires a new session.
|
||||
if let previous = connections[profile.id],
|
||||
(previous.profile != profile || previous.password != password) {
|
||||
connections.removeValue(forKey: profile.id)
|
||||
let session = slot(for: profile.id)
|
||||
let expectedGeneration = session.generation()
|
||||
return try session.queue.sync {
|
||||
guard session.generation() == expectedGeneration else {
|
||||
throw MySQLWireError(message: "Connection request superseded")
|
||||
}
|
||||
|
||||
if connections[profile.id] == nil {
|
||||
if let previous = session.entry,
|
||||
(previous.profile != profile || previous.password != password) {
|
||||
session.entry = nil
|
||||
DBSynSSHTunnels.shared.stop(profile.id)
|
||||
}
|
||||
if session.entry == nil {
|
||||
let endpoint = try DBSynSSHTunnels.shared.endpoint(for: profile)
|
||||
let connection = try MySQLWireConnection(host: endpoint.0, port: endpoint.1, requireTLS: !profile.useSSHTunnel && profile.useTLS && !["127.0.0.1", "localhost", "::1"].contains(endpoint.0.lowercased()))
|
||||
let connection = try MySQLWireConnection(
|
||||
host: endpoint.0, port: endpoint.1,
|
||||
requireTLS: !profile.useSSHTunnel && profile.useTLS &&
|
||||
!["127.0.0.1", "localhost", "::1"].contains(endpoint.0.lowercased())
|
||||
)
|
||||
connection.markHost(endpoint.0)
|
||||
try connection.authenticate(user: profile.username,
|
||||
password: password, database: database)
|
||||
connections[profile.id] = Entry(profile: profile,
|
||||
password: password,
|
||||
connection: connection)
|
||||
guard session.generation() == expectedGeneration else {
|
||||
throw MySQLWireError(message: "Connection request superseded")
|
||||
}
|
||||
session.entry = Entry(profile: profile, password: password,
|
||||
connection: connection)
|
||||
}
|
||||
|
||||
guard let entry = connections[profile.id] else {
|
||||
guard let entry = session.entry else {
|
||||
throw MySQLWireError(message: "MySQL session was not created")
|
||||
}
|
||||
do {
|
||||
// Crucially, no new authentication takes place here. USE,
|
||||
// session variables and transactions survive between queries.
|
||||
return try entry.connection.query(sql)
|
||||
let result = try entry.connection.query(sql)
|
||||
guard session.generation() == expectedGeneration else {
|
||||
throw MySQLWireError(message: "Connection request superseded")
|
||||
}
|
||||
return result
|
||||
} catch {
|
||||
// The wire protocol can be out of sync after a failed query.
|
||||
// Discard it rather than silently issuing commands on a bad socket.
|
||||
connections.removeValue(forKey: profile.id)
|
||||
session.entry = nil
|
||||
throw error
|
||||
}
|
||||
}
|
||||
@@ -479,23 +517,36 @@ private final class PersistentMySQLSessions {
|
||||
func stream(profile: ConnectionProfile, database: String,
|
||||
password: String, sql: String,
|
||||
onRow: (String) throws -> Void) throws -> Int64 {
|
||||
try queue.sync {
|
||||
guard let entry = connections[profile.id],
|
||||
let session = slot(for: profile.id)
|
||||
let expectedGeneration = session.generation()
|
||||
return try session.queue.sync {
|
||||
guard session.generation() == expectedGeneration,
|
||||
let entry = session.entry,
|
||||
entry.profile == profile, entry.password == password else {
|
||||
throw MySQLWireError(message: "Export session is no longer connected")
|
||||
}
|
||||
do {
|
||||
return try entry.connection.streamFirstColumn(sql, onRow: onRow)
|
||||
return try entry.connection.streamFirstColumn(sql, onRow: { row in
|
||||
guard session.generation() == expectedGeneration else {
|
||||
throw MySQLWireError(message: "Export cancelled: connection switched")
|
||||
}
|
||||
try onRow(row)
|
||||
})
|
||||
} catch {
|
||||
connections.removeValue(forKey: profile.id)
|
||||
session.entry = nil
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func disconnect(profileID: UUID) {
|
||||
queue.sync { _ = connections.removeValue(forKey: profileID) }
|
||||
DBSynSSHTunnels.shared.stop(profileID)
|
||||
let session = slot(for: profileID)
|
||||
// Invalidate immediately without waiting for a blocked network read.
|
||||
session.invalidate()
|
||||
session.queue.async {
|
||||
session.entry = nil
|
||||
DBSynSSHTunnels.shared.stop(profileID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -526,8 +577,6 @@ enum NativeMySQLRunner {
|
||||
// Connection shutdown may wait behind a slow network operation on the
|
||||
// session's serial queue. Never make the SwiftUI main thread wait for it.
|
||||
static func disconnect(profileID: UUID) {
|
||||
Task.detached(priority: .utility) {
|
||||
PersistentMySQLSessions.shared.disconnect(profileID: profileID)
|
||||
}
|
||||
PersistentMySQLSessions.shared.disconnect(profileID: profileID)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user