-
Notifications
You must be signed in to change notification settings - Fork 328
Prototype pipe handling without dispatch IO #2315
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -58,7 +58,7 @@ public final class JSONRPCConnection: Connection { | |
/// The queue on which we send data. | ||
private let sendQueue: DispatchQueue = DispatchQueue(label: "jsonrpc-send-queue", qos: .userInitiated) | ||
|
||
private let receiveIO: DispatchIO | ||
private let inFD: FileHandle | ||
private let sendIO: DispatchIO | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think that we should replace this to be symmetric with the read side. |
||
private let messageRegistry: MessageRegistry | ||
|
||
|
@@ -86,7 +86,6 @@ public final class JSONRPCConnection: Connection { | |
/// Buffer of received bytes that haven't been parsed. | ||
/// | ||
/// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are | ||
/// - The `receiveIO` handler: This is synchronized on `queue`. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We now have a corresponding use in /// - The `inFD.readabilityHandler`: This is synchronized on `queue`. |
||
/// - `requestBufferIsEmpty`: Also synchronized on `queue`. | ||
private nonisolated(unsafe) var requestBuffer: [UInt8] = [] | ||
|
||
|
@@ -140,23 +139,12 @@ public final class JSONRPCConnection: Connection { | |
|
||
#if os(Windows) | ||
let rawInFD = dispatch_fd_t(bitPattern: inFD._handle) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we still need |
||
self.inFD = inFD | ||
#else | ||
let rawInFD = inFD.fileDescriptor | ||
#endif | ||
|
||
ioGroup.enter() | ||
receiveIO = DispatchIO( | ||
type: .stream, | ||
fileDescriptor: rawInFD, | ||
queue: queue, | ||
cleanupHandler: { (error: Int32) in | ||
if error != 0 { | ||
logger.fault("IO error \(error)") | ||
} | ||
ioGroup.leave() | ||
} | ||
) | ||
|
||
#if os(Windows) | ||
let rawOutFD = dispatch_fd_t(bitPattern: outFD._handle) | ||
#else | ||
|
@@ -188,10 +176,6 @@ public final class JSONRPCConnection: Connection { | |
} | ||
} | ||
|
||
// We cannot assume the client will send us bytes in packets of any particular size, so set the lower limit to 1. | ||
receiveIO.setLimit(lowWater: 1) | ||
receiveIO.setLimit(highWater: Int.max) | ||
|
||
sendIO.setLimit(lowWater: 1) | ||
sendIO.setLimit(highWater: Int.max) | ||
} | ||
|
@@ -293,27 +277,17 @@ public final class JSONRPCConnection: Connection { | |
state = .running | ||
self.receiveHandler = receiveHandler | ||
self.closeHandler = closeHandler | ||
} | ||
|
||
receiveIO.read(offset: 0, length: Int.max, queue: queue) { done, data, errorCode in | ||
guard errorCode == 0 else { | ||
#if !os(Windows) | ||
if errorCode != POSIXError.ECANCELED.rawValue { | ||
logger.fault("IO error reading \(errorCode)") | ||
} | ||
#endif | ||
if done { self.closeAssumingOnQueue() } | ||
return | ||
} | ||
|
||
if done { | ||
self.closeAssumingOnQueue() | ||
return | ||
} | ||
|
||
guard let data = data, !data.isEmpty else { | ||
self.inFD.readabilityHandler = { fileHandle in | ||
let data = (try? fileHandle.read(upToCount: Int.max)) ?? Data() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Instead of |
||
if data.isEmpty { | ||
fileHandle.readabilityHandler = nil | ||
self.close() | ||
return | ||
} | ||
} | ||
|
||
self.queue.sync { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
orLog("Writing input mirror file") { | ||
try self.inputMirrorFile?.write(contentsOf: data) | ||
} | ||
|
@@ -660,7 +634,6 @@ public final class JSONRPCConnection: Connection { | |
|
||
logger.log("Closing JSONRPCConnection...") | ||
// Attempt to close the reader immediately; we do not need to accept remaining inputs. | ||
receiveIO.close(flags: .stop) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should we set the |
||
// Close the writer after it finishes outstanding work. | ||
sendIO.close() | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we match the previous naming and call this
receiveFD
?inFD
can get confusing if you eg. set up aJSONRPCConnection
from SourceKit-LSP to a BSP server, in which case thereceiveFD
will be stdout of the BSP server process and thusreceiveFD
wouldn’t be stdin.