Add parEvalMapPartitioned - concurrent eval with per-key ordering

Abierto
  • ¿Algo no está bien?
#3,800 2 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 3 días

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
35/100
Tipo de issue
Nueva funcionalidad
Claridad
Bastante claro
Estado de actividad
Activo
Stack tecnológico
scala
Área
backend, distributed-systems

Línea de trabajo

Read the prototype linked in the issue and compare it with fs2's existing parEvalMap implementation and tests. The proposed operation needs per-key ordering, concurrency and buffering limits, and ordered result emission; the issue also raises unresolved API choices around key equality, overloads, naming, and aliases. Done means those semantics and API questions are resolved and the implementation is covered by tests, including the stated ordered-mode hang.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

Elements with the same key run one at a time, in arrival order. Other keys run concurrently, and results are emitted in input order. Akka and Pekko have this as mapAsyncPartitioned.

records
  .parEvalMapPartitioned(maxConcurrent = 32, maxBuffered = 512)(_.record.key) {
    (committable, _) => processRecord(committable.record).as(committable.offset)
  }
  .through(commitBatchWithin(500, 15.seconds))

maxConcurrent limits running effects, for example to a connection pool's size. maxBuffered limits the elements held, including those waiting for their key or to be emitted. There is also a single-limit overload and parEvalMapPartitionedUnordered.

I have a prototype built on parEvalMap here. If an effect fails, later elements with that key are skipped. If an effect fails, the stream terminates with that error, as with parEvalMap. Until then, later elements with the same key aren't evaluated.
It inherits the hang from #3486 in ordered mode.

Questions:

  1. Would you want this in fs2 (I personally need this for processing records from kafka using fs2-kafka)?
  2. Eq or Hash for keys? The prototype uses Eq, with a linear scan over keys in progress.
  3. Is maxBuffered the right name, and should the single-limit overload stay?
  4. Should there be mapAsyncPartitioned aliases? The prototype has none, because Akka's second parameter is perPartition, so mapAsyncPartitioned(32, 40) would silently mean something different.
Lenguaje dominante
Scala
Estrellas
2.5k
Forks
633
Merge medio
3 d 21 h
PR fusionados (30 d)
13

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de typelevel/fs2

Todos los issues de typelevel/fs2

Issues similares

Más issues de Scala

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.