@@ -185,6 +185,117 @@ public final class ValkeyClusterClient: Sendable {
185185 throw ValkeyClusterError . clientRequestCancelled
186186 }
187187
188+ /// Pipeline a series of commands to nodes in the Valkey cluster
189+ ///
190+ /// This function splits up the array of commands into smaller arrays containing
191+ /// the commands that should be run on each node in the cluster. It then runs a
192+ /// pipelined execute using these smaller arrays on each node concurrently.
193+ ///
194+ /// Once all the responses for the commands have been received the function converys
195+ /// them to their expected Response type.
196+ ///
197+ /// Because the commands are split across nodes it is not possible to guarantee
198+ /// the order that commands will run in. The only way to guarantee the order is to
199+ /// only pipeline commands that use keys from the same HashSlot. If a key has a
200+ /// substring between brackets `{}` then that substring is used to calculate the
201+ /// HashSlot. That substring is called the hash tag. Using this you can ensure two
202+ /// keys are in the same hash slot, by giving them the same hash tag eg `user:{123}`
203+ /// and `profile:{123}`.
204+ ///
205+ /// - Parameter commands: Parameter pack of ValkeyCommands
206+ /// - Returns: Parameter pack holding the responses of all the commands
207+ @inlinable
208+ public func execute< each Command : ValkeyCommand > (
209+ _ commands: repeat each Command
210+ ) async throws -> sending ( repeat Result < ( each Command ) . Response, Error > ) {
211+ func convert< Response: RESPTokenDecodable > ( _ result: Result < RESPToken , Error > , to: Response . Type ) -> Result < Response , Error > {
212+ result. flatMap {
213+ do {
214+ return try . success( Response ( fromRESP: $0) )
215+ } catch {
216+ return . failure( error)
217+ }
218+ }
219+ }
220+ let results = try await self . execute ( [ any ValkeyCommand ] ( commands: repeat each commands) )
221+ var index = AutoIncrementingInteger ( )
222+ return ( repeat convert( results [ index. next ( ) ] , to: ( each Command) . Response. self) )
223+ }
224+
225+ /// Results from pipeline and index for each result
226+ @usableFromInline
227+ struct NodePipelineResult : Sendable {
228+ @usableFromInline
229+ let indices : [ [ any ValkeyCommand ] . Index]
230+ @usableFromInline
231+ let results : [ Result < RESPToken , Error > ]
232+
233+ @inlinable
234+ init ( indices: [ [ any ValkeyCommand ] . Index] , results: [ Result < RESPToken , Error > ] ) {
235+ self . indices = indices
236+ self . results = results
237+ }
238+ }
239+
240+ /// Pipeline a series of commands to nodes in the Valkey cluster
241+ ///
242+ /// This function splits up the array of commands into smaller arrays containing
243+ /// the commands that should be run on each node in the cluster. It then runs a
244+ /// pipelined execute using these smaller arrays on each node concurrently.
245+ ///
246+ /// Once all the responses for the commands have been received the function returns
247+ /// an array of RESPToken Results, one for each command.
248+ ///
249+ /// Because the commands are split across nodes it is not possible to guarantee
250+ /// the order that commands will run in. The only way to guarantee the order is to
251+ /// only pipeline commands that use keys from the same HashSlot. If a key has a
252+ /// substring between brackets `{}` then that substring is used to calculate the
253+ /// HashSlot. That substring is called the hash tag. Using this you can ensure two
254+ /// keys are in the same hash slot, by giving them the same hash tag eg `user:{123}`
255+ /// and `profile:{123}`.
256+ ///
257+ /// - Parameter commands: Parameter pack of ValkeyCommands
258+ /// - Returns: Array holding the RESPToken responses of all the commands
259+ @inlinable
260+ public func execute(
261+ _ commands: [ any ValkeyCommand ]
262+ ) async throws -> sending [ Result < RESPToken , Error > ] {
263+ guard commands. count > 0 else { return [ ] }
264+ // get a list of nodes and the commands that should be run on them
265+ let nodes = try await self . splitCommandsAcrossNodes ( commands: commands)
266+ // if this list has one element, then just run the pipeline on that single node
267+ if nodes. count == 1 {
268+ do {
269+ return try await self . execute ( node: nodes [ nodes. startIndex] . node, commands: commands)
270+ } catch {
271+ return . init( repeating: . failure( error) , count: commands. count)
272+ }
273+ }
274+ return await withTaskGroup ( of: NodePipelineResult . self) { group in
275+ // run generated pipelines concurrently
276+ for node in nodes {
277+ let indices = node. commandIndices
278+ group. addTask {
279+ do {
280+ let results = try await self . execute ( node: node. node, commands: IndexedSubCollection ( commands, indices: indices) )
281+ return . init( indices: indices, results: results)
282+ } catch {
283+ return NodePipelineResult ( indices: indices, results: . init( repeating: . failure( error) , count: indices. count) )
284+ }
285+ }
286+ }
287+ var results = [ Result < RESPToken , Error > ] ( repeating: . failure( ValkeyClusterError . pipelinedResultNotReturned) , count: commands. count)
288+ // get results for each node
289+ while let taskResult = await group. next ( ) {
290+ precondition ( taskResult. indices. count == taskResult. results. count)
291+ for index in 0 ..< taskResult. indices. count {
292+ results [ taskResult. indices [ index] ] = taskResult. results [ index]
293+ }
294+ }
295+ return results
296+ }
297+ }
298+
188299 struct Redirection {
189300 let node : ValkeyNodeClient
190301 let ask : Bool
@@ -347,6 +458,74 @@ public final class ValkeyClusterClient: Sendable {
347458 return hashSlot
348459 }
349460
461+ /// Node and list of indices into command array
462+ @usableFromInline
463+ struct NodeAndCommands : Sendable {
464+ @usableFromInline
465+ let node : ValkeyNodeClient
466+ @usableFromInline
467+ var commandIndices : [ Int ]
468+
469+ @usableFromInline
470+ internal init ( node: ValkeyNodeClient , commandIndices: [ Int ] ) {
471+ self . node = node
472+ self . commandIndices = commandIndices
473+ }
474+ }
475+
476+ /// Split command array into multiple arrays of indices into the original array.
477+ ///
478+ /// These array of indices are then used to create collections of commands to
479+ /// run on each node
480+ @usableFromInline
481+ func splitCommandsAcrossNodes( commands: [ any ValkeyCommand ] ) async throws -> some Collection < NodeAndCommands > {
482+ var nodeMap : [ ValkeyServerAddress : NodeAndCommands ] = [ : ]
483+ var index = commands. startIndex
484+ var prevAddress : ValkeyServerAddress ? = nil
485+ // iterate through commands until you reach one that affects a key
486+ while index < commands. endIndex {
487+ let command = commands [ index]
488+ index += 1
489+ let keysAffected = command. keysAffected
490+ if keysAffected. count > 0 {
491+ // Get hash slot for key and add all the commands you have iterated through so far to the
492+ // node associated with that key and break out of loop
493+ let hashSlot = try self . hashSlot ( for: keysAffected)
494+ let node = try await self . nodeClient ( for: hashSlot. map { [ $0] } ?? [ ] )
495+ let address = node. serverAddress
496+ let nodeAndCommands = NodeAndCommands ( node: node, commandIndices: . init( commands. startIndex..< index) )
497+ nodeMap [ address] = nodeAndCommands
498+ prevAddress = address
499+ break
500+ }
501+ }
502+ // If we found a key while iterating through the commands iterate through the remaining commands
503+ if var prevAddress {
504+ while index < commands. endIndex {
505+ let command = commands [ index]
506+ let keysAffected = command. keysAffected
507+ if keysAffected. count > 0 {
508+ // If command affects a key get hash slot for key and add command to the node associated with that key
509+ let hashSlot = try self . hashSlot ( for: keysAffected)
510+ let node = try await self . nodeClient ( for: hashSlot. map { [ $0] } ?? [ ] )
511+ prevAddress = node. serverAddress
512+ nodeMap [ prevAddress, default: . init( node: node, commandIndices: [ ] ) ] . commandIndices. append ( index)
513+ } else {
514+ // if command doesn't affect a key then use the node the previous command used
515+ nodeMap [ prevAddress] !. commandIndices. append ( index)
516+ }
517+ index += 1
518+ }
519+ } else {
520+ // if none of the commands affect any keys then choose a random node
521+ let node = try await self . nodeClient ( for: [ ] )
522+ let address = node. serverAddress
523+ let nodeAndCommands = NodeAndCommands ( node: node, commandIndices: . init( commands. startIndex..< index) )
524+ nodeMap [ address] = nodeAndCommands
525+ }
526+ return nodeMap. values
527+ }
528+
350529 @usableFromInline
351530 enum RetryAction {
352531 case redirect( ValkeyClusterRedirectionError )
@@ -791,3 +970,18 @@ public final class ValkeyClusterClient: Sendable {
791970/// This allows the cluster client to be used anywhere a `ValkeyClientProtocol` is expected.
792971@available ( valkeySwift 1 . 0 , * )
793972extension ValkeyClusterClient : ValkeyClientProtocol { }
973+
974+ extension Array where Element == any ValkeyCommand {
975+ /// Initializer used internally in cluster client and tests for constructing an array
976+ /// of commands from a parameter pack of commands
977+ @inlinable
978+ init < each Command : ValkeyCommand > (
979+ commands: repeat each Command
980+ ) {
981+ var commandArray : [ any ValkeyCommand ] = [ ]
982+ for command in repeat each commands {
983+ commandArray. append ( command)
984+ }
985+ self = commandArray
986+ }
987+ }
0 commit comments