Combinar el marco serializar operaciones asincrónicas

8

¿Cómo consigo que las canalizaciones asíncronas que constituyen el marco de trabajo de Combine se alineen sincrónicamente (en serie)?

Supongamos que tengo 50 URL de las que quiero descargar los recursos correspondientes, y digamos que quiero hacerlo de uno en uno. Sé cómo hacerlo con Operation / OperationQueue, por ejemplo, utilizando una subclase de Operation que no se declara terminada hasta que se completa la descarga. ¿Cómo haría lo mismo con Combine?

Por el momento, todo lo que se me ocurre es mantener una lista global de las URL restantes y abrir una, configurar esa tubería para una descarga, hacer la descarga y, en sinkla tubería, repetir. Eso no parece muy combinado.

Intenté hacer una matriz de las URL y asignarla a una variedad de editores. Sé que puedo "producir" un editor y hacer que se publique en el futuro utilizando flatMap. Pero aún sigo haciendo todas las descargas simultáneamente. No hay ninguna forma combinada de recorrer la matriz de manera controlada, ¿o sí?

(También imaginé hacer algo con Future, pero me confundí irremediablemente. No estoy acostumbrado a esta forma de pensar).

mate
fuente

Respuestas:

2

Solo probé brevemente esto, pero a primera vista parece que cada solicitud espera a que finalice la solicitud anterior antes de comenzar.

Estoy publicando esta solución en busca de comentarios. Sea crítico si esta no es una buena solución.

extension Collection where Element: Publisher {

    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        // If the collection is empty, we can't just create an arbititary publisher
        // so we return nil to indicate that we had nothing to serialize.
        if isEmpty { return nil }

        // We know at this point that it's safe to grab the first publisher.
        let first = self.first!

        // If there was only a single publisher then we can just return it.
        if count == 1 { return first.eraseToAnyPublisher() }

        // We're going to build up the output starting with the first publisher.
        var output = first.eraseToAnyPublisher()

        // We iterate over the rest of the publishers (skipping over the first.)
        for publisher in self.dropFirst() {
            // We build up the output by appending the next publisher.
            output = output.append(publisher).eraseToAnyPublisher()
        }

        return output
    }
}

Una versión más concisa de esta solución (proporcionada por @matt):

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            $0.append($1).eraseToAnyPublisher()
        }
    }
}
Clay Ellis
fuente
Excelente, gracias. appendes exactamente lo que estaba buscando - Su código se puede ajustar considerablemente; en particular, no hay necesidad de regresar prematuramente en el caso donde count == 1, porque en ese caso dropFirstestará vacío y simplemente no haremos un bucle. Y no hay necesidad de mantener la outputvariable, porque podemos usarla en reducelugar de for...in. Vea mi respuesta para una representación más ajustada.
mate
3

Puede crear un Suscriptor personalizado donde reciba Subscribers.Demand.max (1). En ese caso, el suscriptor solicitará el siguiente valor solo cuando reciba uno. El ejemplo es para Int.publisher, pero un retraso aleatorio en el mapa imita el tráfico de red :-)

import PlaygroundSupport
import SwiftUI
import Combine

class MySubscriber: Subscriber {
  typealias Input = String
  typealias Failure = Never

  func receive(subscription: Subscription) {
    print("Received subscription", Thread.current.isMainThread)
    subscription.request(.max(1))
  }

  func receive(_ input: Input) -> Subscribers.Demand {
    print("Received input: \(input)", Thread.current.isMainThread)
    return .max(1)
  }

  func receive(completion: Subscribers.Completion<Never>) {
    DispatchQueue.main.async {
        print("Received completion: \(completion)", Thread.current.isMainThread)
        PlaygroundPage.current.finishExecution()
    }
  }
}

(110...120)
    .publisher.receive(on: DispatchQueue.global())
    .map {
        print(Thread.current.isMainThread, Thread.current)
        usleep(UInt32.random(in: 10000 ... 1000000))
        return String(format: "%02x", $0)
    }
    .subscribe(on: DispatchQueue.main)
    .subscribe(MySubscriber())

print("Hello")

PlaygroundPage.current.needsIndefiniteExecution = true

Impresión del patio ...

Hello
Received subscription true
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 6e false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 6f false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 70 false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 71 false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 72 false
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 73 false
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 74 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 75 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 76 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 77 false
false <NSThread: 0x600000053400>{number = 3, name = (null)}
Received input: 78 false
Received completion: finished true

ACTUALIZACIÓN finalmente encontré .flatMap(maxPublishers: ), lo que me obliga a actualizar este interesante tema con un enfoque un poco diferente. Por favor, vea que estoy usando la cola global para la programación, no solo un retraso aleatorio, solo para asegurarme de que recibir una transmisión serializada no sea un comportamiento "aleatorio" o "afortunado" :-)

import PlaygroundSupport
import Combine
import Foundation

PlaygroundPage.current.needsIndefiniteExecution = true

let A = (1 ... 9)
    .publisher
    .flatMap(maxPublishers: .max(1)) { value in
        [value].publisher
            .flatMap { value in
                Just(value)
                    .delay(for: .milliseconds(Int.random(in: 0 ... 100)), scheduler: DispatchQueue.global())
        }
}
.sink { value in
    print(value, "A")
}

let B = (1 ... 9)
    .publisher
    .flatMap { value in
        [value].publisher
            .flatMap { value in
                Just(value)
                    .delay(for: .milliseconds(Int.random(in: 0 ... 100)), scheduler: RunLoop.main)
        }
}
.sink { value in
    print("     ",value, "B")
}

huellas dactilares

1 A
      4 B
      5 B
      7 B
      1 B
      2 B
      8 B
      6 B
2 A
      3 B
      9 B
3 A
4 A
5 A
6 A
7 A
8 A
9 A

Basado en escrito aquí

.publicar por fascículos()?

definido por Clay Ellis la respuesta aceptada podría ser reemplazada por

.publisher.flatMap (maxPublishers: .max (1)) {$ 0}

mientras que la versión "no serializada" debe usar

.publisher.flatMap {$ 0}

"ejemplo del mundo real"

import PlaygroundSupport
import Foundation
import Combine

let path = "postman-echo.com/get"
let urls: [URL] = "... which proves the downloads are happening serially .-)".map(String.init).compactMap { (parameter) in
    var components = URLComponents()
    components.scheme = "https"
    components.path = path
    components.queryItems = [URLQueryItem(name: parameter, value: nil)]
    return components.url
}
//["https://postman-echo.com/get?]
struct Postman: Decodable {
    var args: [String: String]
}


let collection = urls.compactMap { value in
        URLSession.shared.dataTaskPublisher(for: value)
        .tryMap { data, response -> Data in
            return data
        }
        .decode(type: Postman.self, decoder: JSONDecoder())
        .catch {_ in
            Just(Postman(args: [:]))
    }
}

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            return $0.append($1).eraseToAnyPublisher()
        }
    }
}

var streamA = ""
let A = collection
    .publisher.flatMap{$0}

    .sink(receiveCompletion: { (c) in
        print(streamA, "     ", c, "    .publisher.flatMap{$0}")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamA)
    })


var streamC = ""
let C = collection
    .serialize()?

    .sink(receiveCompletion: { (c) in
        print(streamC, "     ", c, "    .serialize()?")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamC)
    })

var streamD = ""
let D = collection
    .publisher.flatMap(maxPublishers: .max(1)){$0}

    .sink(receiveCompletion: { (c) in
        print(streamD, "     ", c, "    .publisher.flatMap(maxPublishers: .max(1)){$0}")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamD)
    })

PlaygroundPage.current.needsIndefiniteExecution = true

huellas dactilares

.w.h i.c hporves ht edownloadsa erh appeninsg eriall y.-)       finished     .publisher.flatMap{$0}
... which proves the downloads are happening serially .-)       finished     .publisher.flatMap(maxPublishers: .max(1)){$0}
... which proves the downloads are happening serially .-)       finished     .serialize()?

Me parece muy útil en otros escenarios también. Intente utilizar el valor predeterminado de maxPublishers en el siguiente fragmento y compare los resultados :-)

import Combine

let sequencePublisher = Publishers.Sequence<Range<Int>, Never>(sequence: 0..<Int.max)
let subject = PassthroughSubject<String, Never>()

let handle = subject
    .zip(sequencePublisher.print())
    //.publish
    .flatMap(maxPublishers: .max(1), { (pair)  in
        Just(pair)
    })
    .print()
    .sink { letters, digits in
        print(letters, digits)
    }

"Hello World!".map(String.init).forEach { (s) in
    subject.send(s)
}
subject.send(completion: .finished)
usuario3441734
fuente
@matt sink no funciona de manera diferente, solo en recibir Subsribers.Demand.unlimited devuelto ... Puede estar usando el instrumento adecuado, como cola en serie y Data.init? (contenido de url: URL) es la mejor opción en su escenario . Si necesita hacer una suma de dos Int, ¿lo hace como [lhs: Int, rhs: Int] .reduce .... ??? Usaré Data.init? (Contenido de url: URL) dentro de recibir (_ input :) de MySerialDownloaderSubscriber.
user3441734
@matt por favor, vea la respuesta actualizada. Combinar es emocionante, pero (al menos para mí) muy difícil de entender ...
user3441734
¡Si lo veo! Con el maxPublishersparámetro, podemos agregar contrapresión. Esto va con lo que dije en mi pregunta: "Sé que puedo" producir "un editor y hacer que se publique en el futuro usando flatMap. Pero aún así estoy haciendo todas las descargas simultáneamente". Pues bien, con el maxPublishersparámetro, que son no simultánea.
mate
@matt sí, hundir el suscriptor del editor con Subscribers.Demand.unlimited, flatMap tiene el mismo efecto que establecer el propio suscriptor del editor con un valor diferente, en nuestro caso de uso .max (1). Solo agrego otro ejemplo con un escenario diferente, donde es tan útil.
user3441734
2

En todos los demás marcos reactivos esto es realmente fácil; solo se usa concatpara concatenar y aplanar los resultados en un solo paso y luego se pueden obtener reducelos resultados en una matriz final. Apple hace esto difícil porque Publisher.Concatenateno tiene sobrecarga que acepte una variedad de editores. Hay rarezas similares con Publisher.Merge. Tengo la sensación de que esto tiene que ver con el hecho de que devuelven editores genéricos anidados en lugar de devolver un solo tipo genérico como rx Observable. Supongo que puedes llamar a Concatenateen un bucle y luego reducir los resultados concatenados en una sola matriz, pero realmente espero que aborden este problema en la próxima versión. Ciertamente, es necesario concatenar a más de 2 publicadores y fusionar más de 4 publicadores (y las sobrecargas para estos dos operadores ni siquiera son consistentes, lo cual es extraño).

EDITAR:

Volví a esto y descubrí que de hecho puedes concatenar una variedad arbitraria de editores y emitirán en secuencia. No tengo idea de por qué no hay una función como ConcatenateManyhacer esto por usted, pero parece que siempre y cuando esté dispuesto a usar un editor borrado no es tan difícil escribir uno. Este ejemplo muestra que la fusión emite en orden temporal mientras que concat emite en el orden de combinación:

import PlaygroundSupport
import SwiftUI
import Combine

let p = Just<Int>(1).append(2).append(3).delay(for: .seconds(0.25), scheduler: RunLoop.main).eraseToAnyPublisher()
let q = Just<Int>(4).append(5).append(6).eraseToAnyPublisher()
let r = Just<Int>(7).append(8).append(9).delay(for: .seconds(0.5), scheduler: RunLoop.main).eraseToAnyPublisher()
let concatenated: AnyPublisher<Int, Never> = [q,r].reduce(p) { total, next in
  total.append(next).eraseToAnyPublisher()
}

var subscriptions = Set<AnyCancellable>()

concatenated
  .sink(receiveValue: { v in
    print("concatenated: \(v)")
  }).store(in: &subscriptions)

Publishers
  .MergeMany([p,q,r])
  .sink(receiveValue: { v in
    print("merge: \(v)")
  }).store(in: &subscriptions)
Josh Homann
fuente
Sí, probablemente adivinaste que elegí un gran número como 50 intencionalmente.
mate
Hay un MergeMany. No entiendo por qué no hay un ConcatenateMany. Rx swift tiene Observable.concat y Reactive Swift tiene flatMap (.concat), así que esto es extraño; Tal vez me estoy perdiendo algo. Seguiré buscando developer.apple.com/documentation/combine/publishers/mergemany
Josh Homann
Sería concatserializar (en los otros marcos reactivos)?
mate
Si. Para una secuencia de secuencias, solo tiene una forma de aplanar, es decir, colocar los elementos de una secuencia interna después de otra al igual que Sequence.flatMap en forma rápida. Cuando tiene una secuencia asincrónica, debe tener en cuenta la dimensión temporal al aplanar. Por lo tanto, puede emitir los elementos de todas las secuencias internas en orden temporal (fusión) o puede emitir los elementos de cada secuencia interna en orden de las secuencias (concat). Vea el diagrama de mármol: rxmarbles.com/#concat vs rxmarbles.com/#merge
Josh Homann
Tenga en cuenta que .appendes un operador que crea un Publisher.Concatenate.
Rob Mayoff
2

De la pregunta original:

Intenté hacer una matriz de las URL y asignarla a una variedad de editores. Sé que puedo "producir" un editor y hacer que se publique en el futuro utilizando flatMap. Pero aún sigo haciendo todas las descargas simultáneamente. No hay ninguna forma combinada de recorrer la matriz de manera controlada, ¿o sí?


Aquí hay un ejemplo de juguete para reemplazar el problema real:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
collection.publisher
    .flatMap() {$0}
    .sink {print($0)}.store(in:&self.storage)

Esto emite los enteros del 1 al 10 en orden aleatorio llegando a tiempos aleatorios. El objetivo es hacer algo con collectioneso hará que emita los enteros del 1 al 10 en orden.


Ahora vamos a cambiar solo una cosa: en la línea

.flatMap {$0}

agregamos el maxPublishersparámetro:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
collection.publisher
    .flatMap(maxPublishers:.max(1)) {$0}
    .sink {print($0)}.store(in:&self.storage)

Presto, ahora hacer emiten los números enteros de 1 a 10, en orden, con intervalos aleatorios entre ellos.


Apliquemos esto al problema original. Para demostrarlo, necesito una conexión a Internet bastante lenta y un recurso bastante grande para descargar. Primero, lo haré con ordinario .flatMap:

let eph = URLSessionConfiguration.ephemeral
let session = URLSession(configuration: eph)
let url = "https://photojournal.jpl.nasa.gov/tiff/PIA23172.tif"
let collection = [url, url, url]
    .map {URL(string:$0)!}
    .map {session.dataTaskPublisher(for: $0)
        .eraseToAnyPublisher()
}
collection.publisher.setFailureType(to: URLError.self)
    .handleEvents(receiveOutput: {_ in print("start")})
    .flatMap() {$0}
    .map {$0.data}
    .sink(receiveCompletion: {comp in
        switch comp {
        case .failure(let err): print("error", err)
        case .finished: print("finished")
        }
    }, receiveValue: {_ in print("done")})
    .store(in:&self.storage)

El resultado es

start
start
start
done
done
done
finished

lo que muestra que estamos haciendo las tres descargas simultáneamente. Ok, ahora cambia

    .flatMap() {$0}

a

    .flatMap(maxPublishers:.max(1) {$0}

El resultado ahora es:

start
done
start
done
start
done
finished

Así que ahora estamos descargando en serie, que es el problema originalmente resuelto.


adjuntar

De acuerdo con el principio de TIMTOWTDI, podemos encadenar a los editores appendpara serializarlos:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
let pub = collection.dropFirst().reduce(collection.first!) {
    return $0.append($1).eraseToAnyPublisher()
}

El resultado es una editorial que serializa a las editoriales retrasadas en la colección original. Probémoslo suscribiéndolo:

pub.sink {print($0)}.store(in:&self.storage)

Efectivamente, los enteros ahora llegan en orden (con intervalos aleatorios entre).


Podemos encapsular la creación de pubuna colección de editores con una extensión en Colección, como lo sugiere Clay Ellis:

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            return $0.append($1).eraseToAnyPublisher()
        }
    }
}
mate
fuente
1

Aquí hay una página de código de juegos que muestra un posible enfoque. La idea principal es transformar las llamadas API asíncronas en una cadena de Futureeditores, haciendo así una canalización en serie.

Entrada: rango de int de 1 a 10 que asíncronamente en la cola de fondo convertida en cadenas

Demostración de llamada directa a API asíncrona:

let group = DispatchGroup()
inputValues.map {
    group.enter()
    asyncCall(input: $0) { (output, _) in
        print(">> \(output), in \(Thread.current)")
        group.leave()
    }
}
group.wait()

Salida:

>> 1, in <NSThread: 0x7fe76264fff0>{number = 4, name = (null)}
>> 3, in <NSThread: 0x7fe762446b90>{number = 3, name = (null)}
>> 5, in <NSThread: 0x7fe7624461f0>{number = 5, name = (null)}
>> 6, in <NSThread: 0x7fe762461ce0>{number = 6, name = (null)}
>> 10, in <NSThread: 0x7fe76246a7b0>{number = 7, name = (null)}
>> 4, in <NSThread: 0x7fe764c37d30>{number = 8, name = (null)}
>> 7, in <NSThread: 0x7fe764c37cb0>{number = 9, name = (null)}
>> 8, in <NSThread: 0x7fe76246b540>{number = 10, name = (null)}
>> 9, in <NSThread: 0x7fe7625164b0>{number = 11, name = (null)}
>> 2, in <NSThread: 0x7fe764c37f50>{number = 12, name = (null)}

Demostración de la tubería combinada:

Salida:

>> got 1
>> got 2
>> got 3
>> got 4
>> got 5
>> got 6
>> got 7
>> got 8
>> got 9
>> got 10
>>>> finished with true

Código:

import Cocoa
import Combine
import PlaygroundSupport

// Assuming there is some Asynchronous API with
// (eg. process Int input value during some time and generates String result)
func asyncCall(input: Int, completion: @escaping (String, Error?) -> Void) {
    DispatchQueue.global(qos: .background).async {
            sleep(.random(in: 1...5)) // wait for random Async API output
            completion("\(input)", nil)
        }
}

// There are some input values to be processed serially
let inputValues = Array(1...10)

// Prepare one pipeline item based on Future, which trasform Async -> Sync
func makeFuture(input: Int) -> AnyPublisher<Bool, Error> {
    Future<String, Error> { promise in
        asyncCall(input: input) { (value, error) in
            if let error = error {
                promise(.failure(error))
            } else {
                promise(.success(value))
            }
        }
    }
    .receive(on: DispatchQueue.main)
    .map {
        print(">> got \($0)") // << sideeffect of pipeline item
        return true
    }
    .eraseToAnyPublisher()
}

// Create pipeline trasnforming input values into chain of Future publishers
var subscribers = Set<AnyCancellable>()
let pipeline =
    inputValues
    .reduce(nil as AnyPublisher<Bool, Error>?) { (chain, value) in
        if let chain = chain {
            return chain.flatMap { _ in
                makeFuture(input: value)
            }.eraseToAnyPublisher()
        } else {
            return makeFuture(input: value)
        }
    }

// Execute pipeline
pipeline?
    .sink(receiveCompletion: { _ in
        // << do something on completion if needed
    }) { output in
        print(">>>> finished with \(output)")
    }
    .store(in: &subscribers)

PlaygroundPage.current.needsIndefiniteExecution = true
Asperi
fuente
0

Usar flatMap(maxPublishers:transform:)con .max(1), p. Ej.

func imagesPublisher(for urls: [URL]) -> AnyPublisher<UIImage, URLError> {
    Publishers.Sequence(sequence: urls.map { self.imagePublisher(for: $0) })
        .flatMap(maxPublishers: .max(1)) { $0 }
        .eraseToAnyPublisher()
}

Dónde

func imagePublisher(for url: URL) -> AnyPublisher<UIImage, URLError> {
    URLSession.shared.dataTaskPublisher(for: url)
        .compactMap { UIImage(data: $0.data) }
        .receive(on: RunLoop.main)
        .eraseToAnyPublisher()
}

y

var imageRequests: AnyCancellable?

func fetchImages() {
    imageRequests = imagesPublisher(for: urls).sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("done")
        case .failure(let error):
            print("failed", error)
        }
    }, receiveValue: { image in
        // do whatever you want with the images as they come in
    })
}

Eso resultó en:

de serie

Pero debemos reconocer que recibes un gran éxito de rendimiento haciéndolos secuencialmente, así. Por ejemplo, si lo aumento hasta 6 a la vez, es más del doble de rápido:

concurrente

Personalmente, recomendaría descargar solo secuencialmente si es absolutamente necesario (lo cual, al descargar una serie de imágenes / archivos, casi seguramente no es el caso). Sí, realizar solicitudes simultáneamente puede hacer que no terminen en un orden particular, pero solo usamos una estructura que es independiente del orden (por ejemplo, un diccionario en lugar de una matriz simple), pero las ganancias de rendimiento son tan significativas que generalmente vale la pena.

Pero, si desea que se descarguen secuencialmente, el maxPublishersparámetro puede lograrlo.

Robar
fuente
Sí, eso es lo que mi respuesta ya dice: stackoverflow.com/a/59889993/341994 , así como la respuesta que otorgué la recompensa a stackoverflow.com/a/59889174/341994
mate
Y vea también ahora mi libro apeth.com/UnderstandingCombine/operators/…
mate
Por cierto, hablando de forma secuencial, he hecho un gran uso de su operación asincrónica secuencial para una tarea diferente, gracias por escribirla
mate
@matt - Lol. Confieso que no vi que había encontrado la maxPublishersopción. Y no me habría molestado sobre "no hacer serial" si me hubiera dado cuenta de que eras tú (ya que sé que entiendes completamente los pros y los contras de serial versus concurrente). Literalmente solo vi "Quiero descargar un archivo a la vez", recientemente me topé con la maxPublishersopción de otra cosa que estaba haciendo (a saber, proporcionar una solución moderna a esta pregunta ), y pensé en compartir la solución Combinar. se le ocurrió. No quise ser tan derivado.
Rob
1
Sí, era la solución mencionada en stackoverflow.com/a/48104095/1271826 de la que estaba hablando antes; Lo encontré muy útil.
mate