Skip to content
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.

Sign up for GitHub

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

Jump to bottom

Enable command pipelining #552

Merged
merged 11 commits into from
Apr 29, 2021
Merged

Enable command pipelining #552

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
ebc1d20
Enable command pipelining
Davidde94 Apr 22, 2021
168be14
Sanity
Davidde94 Apr 23, 2021
572fb32
Add authenticate test
Davidde94 Apr 26, 2021
e08db08
Fix special cases
Davidde94 Apr 28, 2021
855bef1
Soundness
Davidde94 Apr 28, 2021
1dab4a9
Reword comment
Davidde94 Apr 29, 2021
e5ed391
Partial fix reentrancy
Davidde94 Apr 29, 2021
bcda339
Add commented test to show re-entrancy
Davidde94 Apr 29, 2021
417f1ee
Soundness
Davidde94 Apr 29, 2021
01e21eb
Fix not cascading errors to promising
Davidde94 Apr 29, 2021
c7bd886
Soundness
Davidde94 Apr 29, 2021
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
78 changes: 51 additions & 27 deletions Sources/NIOIMAP/Coders/IMAPClientHandler.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,9 @@ public final class IMAPClientHandler: ChannelDuplexHandler {
public typealias OutboundOut = ByteBuffer

private let decoder: NIOSingleStepByteToMessageProcessor<ResponseDecoder>
private var bufferedWrites: MarkedCircularBuffer<(_EncodeBuffer, EventLoopPromise<Void>?)> =
MarkedCircularBuffer(initialCapacity: 4)

private var currentEncodeBuffer: (_EncodeBuffer, EventLoopPromise<Void>?)?
private var bufferedCommands: MarkedCircularBuffer<(CommandStream, EventLoopPromise<Void>?)> = .init(initialCapacity: 4)

public struct UnexpectedContinuationRequest: Error {}

Expand Down Expand Up @@ -183,33 +184,60 @@ public final class IMAPClientHandler: ChannelDuplexHandler {
}

private func writeNextChunks(context: ChannelHandlerContext) {
assert(self.bufferedWrites.hasMark)
assert(self.bufferedCommands.hasMark || self.bufferedCommands.isEmpty)
defer {
// Note, we can `flush` here because this is already flushed (or else the we wouldn't have a mark).
context.flush()
}
repeat {
let next = self.bufferedWrites[self.bufferedWrites.startIndex].0.nextChunk()

if next.waitForContinuation {
guard let bufferPromise = self.currentEncodeBuffer else {
preconditionFailure("No current buffer to continue writing")
}
var currentBuffer = bufferPromise.0
let currentPromise = bufferPromise.1

// first flush whatever command we currently have buffered
Davidde94 marked this conversation as resolved.
Show resolved Hide resolved
repeat {
let nextChunk = currentBuffer.nextChunk()
if nextChunk.waitForContinuation {
assert(self.state == .expectingResponses || self.state == .expectingLiteralContinuationRequest)
self.state = .expectingLiteralContinuationRequest
context.write(self.wrapOutboundOut(next.bytes), promise: nil)
context.write(self.wrapOutboundOut(nextChunk.bytes), promise: nil)
Davidde94 marked this conversation as resolved.
Show resolved Hide resolved
self.currentEncodeBuffer = (currentBuffer, currentPromise)
weissi marked this conversation as resolved.
Show resolved Hide resolved
return
} else {
assert(self.state == .expectingLiteralContinuationRequest)
self.state = .expectingResponses
let promise = self.bufferedWrites.removeFirst().1
context.write(self.wrapOutboundOut(next.bytes), promise: promise)
context.write(self.wrapOutboundOut(nextChunk.bytes), promise: currentPromise)
self.currentEncodeBuffer = nil
weissi marked this conversation as resolved.
Show resolved Hide resolved
}
} while self.bufferedWrites.hasMark
} while self.currentEncodeBuffer != nil

// continue writing commands until we find a mark, or need a continuation
repeat {
self.writeNextCommand(context: context)
} while self.bufferedCommands.hasMark && self.currentEncodeBuffer == nil
}

public func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
precondition(self.bufferedWrites.isEmpty, "Sorry, we only allow one command at a time right now. We're working on it. Issue #528")
let command = self.unwrapOutboundIn(data)
var encoder = CommandEncodeBuffer(buffer: context.channel.allocator.buffer(capacity: 1024), options: self.encodingOptions)
encoder.writeCommandStream(command)
self.bufferedCommands.append((command, promise))
if self.currentEncodeBuffer == nil {
self.writeNextCommand(context: context)
}
}

public func writeNextCommand(context: ChannelHandlerContext) {
assert(self.currentEncodeBuffer == nil)
guard let (command, promise) = self.bufferedCommands.popFirst() else {
return
}

var commandEncoder = CommandEncodeBuffer(
buffer: context.channel.allocator.buffer(capacity: 512),
options: self.encodingOptions
)
commandEncoder.writeCommandStream(command)

switch command {
case .command(let command):
Expand All @@ -231,24 +259,20 @@ public final class IMAPClientHandler: ChannelDuplexHandler {
break
}

if self.bufferedWrites.isEmpty {
let next = encoder._buffer.nextChunk()

if next.waitForContinuation {
assert(self.state == .expectingResponses)
self.state = .expectingLiteralContinuationRequest
context.write(self.wrapOutboundOut(next.bytes), promise: nil)
// fall through to append below
} else {
context.write(self.wrapOutboundOut(next.bytes), promise: promise)
return
}
let next = commandEncoder._buffer.nextChunk()
Davidde94 marked this conversation as resolved.
Show resolved Hide resolved
if next.waitForContinuation {
self.currentEncodeBuffer = (commandEncoder._buffer, promise)
Davidde94 marked this conversation as resolved.
Show resolved Hide resolved
assert(self.state == .expectingResponses)
self.state = .expectingLiteralContinuationRequest
context.write(self.wrapOutboundOut(next.bytes), promise: nil)
Copy link
Member

Choose a reason for hiding this comment

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

I think this isn't right. I think if this context.write fails here, we'll never fail the user promise (and will therefore crash). If the user gave us a promise, you'll need to .cascadeFailure into the user's promise.

Copy link
Member

Choose a reason for hiding this comment

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

& test please

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

The code didn't change here again, but I can add the cascade

} else {
self.currentEncodeBuffer = nil
context.write(self.wrapOutboundOut(next.bytes), promise: promise)
Copy link
Member

Choose a reason for hiding this comment

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

isn't there a lot of logic duplicated here from above?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

Yeh this was already here but I can factor it out

Copy link
Member

Choose a reason for hiding this comment

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

Yes please

}
self.bufferedWrites.append((encoder._buffer, promise))
}

public func flush(context: ChannelHandlerContext) {
self.bufferedWrites.mark()
self.bufferedCommands.mark()
context.flush()
}
}
120 changes: 93 additions & 27 deletions Tests/NIOIMAPTests/Coders/IMAPClientHandlerTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,33 +87,99 @@ class IMAPClientHandlerTests: XCTestCase {
state: .ok(.init(code: nil, text: "ok")))))
}

// TODO: Make a new state machine that can handle pipelined commands and uncomment this test, issue #528
// func testTwoContReqCommandsEnqueued() {
// let f1 = self.writeOutbound(CommandStream.command(TaggedCommand(tag: "x",
// command: .rename(from: .init("\\"),
// to: .init("to"),
// params: [:]))),
// wait: false)
// let f2 = self.writeOutbound(CommandStream.command(TaggedCommand(tag: "y",
// command: .rename(from: .init("from"),
// to: .init("\\"),
// params: [:]))),
// wait: false)
// self.assertOutboundString("x RENAME {1}\r\n")
// self.writeInbound("+ OK\r\n")
// XCTAssertNoThrow(try f1.wait())
// self.assertOutboundString("\\ \"to\"\r\n")
// self.assertOutboundString("y RENAME \"from\" {1}\r\n")
// self.writeInbound("+ OK\r\n")
// XCTAssertNoThrow(try f2.wait())
// self.assertOutboundString("\\\r\n")
// self.writeInbound("x OK ok\r\n")
// self.assertInbound(.taggedResponse(.init(tag: "x",
// state: .ok(.init(code: nil, text: "ok")))))
// self.writeInbound("y OK ok\r\n")
// self.assertInbound(.taggedResponse(.init(tag: "y",
// state: .ok(.init(code: nil, text: "ok")))))
// }
func testTwoContReqCommandsEnqueued() {
let f1 = self.writeOutbound(CommandStream.command(TaggedCommand(tag: "x",
command: .rename(from: .init("\\"),
to: .init("to"),
params: [:]))),
wait: false)
let f2 = self.writeOutbound(CommandStream.command(TaggedCommand(tag: "y",
command: .rename(from: .init("from"),
to: .init("\\"),
params: [:]))),
wait: false)
self.assertOutboundString("x RENAME {1}\r\n")
self.writeInbound("+ OK\r\n")
XCTAssertNoThrow(try f1.wait())
self.assertOutboundString("\\ \"to\"\r\n")
self.assertOutboundString("y RENAME \"from\" {1}\r\n")
self.writeInbound("+ OK\r\n")
XCTAssertNoThrow(try f2.wait())
self.assertOutboundString("\\\r\n")
self.writeInbound("x OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "x",
state: .ok(.init(code: nil, text: "ok")))))
self.writeInbound("y OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "y",
state: .ok(.init(code: nil, text: "ok")))))
}

// This makes sure that we successfully switch from responding to continuation
// requests back to "simple" commands that can be written in one shot. This was a bug
// in a previous implementation, so this test prevents regression.
func testThreeContReqCommandsEnqueuedFollowedBy2BasicOnes() {
let f1 = self.writeOutbound(.command(.init(tag: "1", command: .create(.init("\\"), []))), wait: false)
let f2 = self.writeOutbound(.command(.init(tag: "2", command: .create(.init("\\"), []))), wait: false)
let f3 = self.writeOutbound(.command(.init(tag: "3", command: .create(.init("\\"), []))), wait: false)
let f4 = self.writeOutbound(.command(.init(tag: "4", command: .create(.init("a"), []))), wait: false)
let f5 = self.writeOutbound(.command(.init(tag: "5", command: .create(.init("b"), []))), wait: false)

self.assertOutboundString("1 CREATE {1}\r\n")
self.writeInbound("+ OK\r\n")
XCTAssertNoThrow(try f1.wait())
self.assertOutboundString("\\\r\n")

self.assertOutboundString("2 CREATE {1}\r\n")
self.writeInbound("+ OK\r\n")
XCTAssertNoThrow(try f2.wait())
self.assertOutboundString("\\\r\n")

self.assertOutboundString("3 CREATE {1}\r\n")
self.writeInbound("+ OK\r\n")
XCTAssertNoThrow(try f3.wait())
self.assertOutboundString("\\\r\n")

self.assertOutboundString("4 CREATE \"a\"\r\n")
XCTAssertNoThrow(try f4.wait())
self.assertOutboundString("5 CREATE \"b\"\r\n")
XCTAssertNoThrow(try f5.wait())

self.writeInbound("1 OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "1",
state: .ok(.init(code: nil, text: "ok")))))
self.writeInbound("2 OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "2",
state: .ok(.init(code: nil, text: "ok")))))

self.writeInbound("3 OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "3",
state: .ok(.init(code: nil, text: "ok")))))

self.writeInbound("4 OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "4",
state: .ok(.init(code: nil, text: "ok")))))

self.writeInbound("5 OK ok\r\n")
self.assertInbound(.taggedResponse(.init(tag: "5",
state: .ok(.init(code: nil, text: "ok")))))
}

func testContinueRequestCommandFollowedByAuthenticate() {
self.writeOutbound(.command(.init(tag: "1", command: .move(.lastCommand, .init("\\")))), wait: false)
self.writeOutbound(.command(.init(tag: "2", command: .authenticate(mechanism: .gssAPI, initialResponse: nil))), wait: false)

// send the move command
self.assertOutboundString("1 MOVE $ {1}\r\n")
self.writeInbound("+ OK\r\n")

// respond to the continuation, move straight to authentication
self.assertOutboundString("\\\r\n")
self.assertOutboundString("2 AUTHENTICATE GSSAPI\r\n")

// server sends an auth challenge
self.writeInbound("+\r\n")
self.assertInbound(.authenticationChallenge(""))
}

func testUnexpectedContinuationRequest() {
let f = self.writeOutbound(CommandStream.command(TaggedCommand(tag: "x",
Expand Down