Concurrent map/foreach in scala

Question

I have an iteration vals: Iterable[T] and a long-running function without any relevant side effects: f: (T => Unit). Right now this is applied to vals in the obvious way:

vals.foreach(f)

I would like the calls to f to be done concurrently (within reasonable limits). Is there an obvious function somewhere in the Scala base library? Something like:

Concurrent.foreach(8 /* Number of threads. */)(vals, f)

While f is reasonably long running, it is short enough that I don't want the overhead of invoking a thread for each call, so I am looking for something based on a thread pool.

Kei-ven · Accepted Answer · 2016-07-04 01:50:07Z

24

Many of the answers from 2009 still use the old scala.actors.Futures._, which are no longer in the newer Scala. While Akka is the preferred way, a much more readable way is to just use parallel (.par) collections:

vals.foreach { v => f(v) }

becomes

vals.par.foreach { v => f(v) }

Alternatively, using parMap can appear more succinct though with the caveat that you need to remember to import the usual Scalaz*. As usual, there's more than one way to do the same thing in Scala!

edited Jul 4, 2016 at 1:50

answered Sep 30, 2014 at 20:50

Kei-ven

4323 silver badges10 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Apocalisp · Accepted Answer · 2010-07-16 00:23:19Z

13

Scalaz has parMap. You would use it as follows:

import scalaz.Scalaz._
import scalaz.concurrent.Strategy.Naive

This will equip every functor (including Iterable) with a parMap method, so you can just do:

vals.parMap(f)

You also get parFlatMap, parZipWith, etc.

edited Jul 16, 2010 at 0:23

answered Nov 19, 2009 at 20:10

Apocalisp

35.1k8 gold badges109 silver badges158 bronze badges

Comments

Daniel Spiewak · Accepted Answer · 2009-11-18 02:24:55Z

10

I like the Futures answer. However, while it will execute concurrently, it will also return asynchronously, which is probably not what you want. The correct approach would be as follows:

import scala.actors.Futures._

vals map { x => future { f(x) } } foreach { _() }

answered Nov 18, 2009 at 2:24

Daniel Spiewak

55.2k14 gold badges113 silver badges121 bronze badges

6 Comments

Ken Bloom Over a year ago

Be careful that vals is a strict collection -- if it's lazy (and in Scala 2.7 this includes the Range class), the futures won't be created until each one is needed by foreach, and nothing will happen in parallel.

Daniel Spiewak Over a year ago

I suppose we could solve that problem by injecting another foreach call between the map and the current foreach. Thus: vals map { x => future { f(x) } } foreach { x => x } foreach { _() }

David Crawshaw Over a year ago

That would be a map we have to inject, not another foreach? And it is not clear to me that the map of a lazy collection is strict. The safest way may be to call toArray.

Daniel Spiewak Over a year ago

You're right, foreach was (obviously) the wrong thing to inject since it returns Unit. My bad! :-) The map function on lazy collections is almost always non-strict, so we can either call toList (or toArray), or we can project and then force: (vals map { x => future { f(x) } } projection).force foreach { _() }. I don't know whether that's better than simply toList, but it is certainly different.

Jus12 Over a year ago

What does it mean when you say it "returns asynchronously"? Does it imply that it is non-blocking? (and why would that be a problem?)

|

Jonathan Graehl · Accepted Answer · 2009-11-17 21:36:17Z

3

I had some issues using scala.actors.Futures in Scala 2.8 (it was buggy when I checked). Using java libs directly worked for me, though:

final object Parallel {
  val cpus=java.lang.Runtime.getRuntime().availableProcessors
  import java.util.{Timer,TimerTask}
  def afterDelay(ms: Long)(op: =>Unit) = new Timer().schedule(new TimerTask {override def run = op},ms)
  def repeat(n: Int,f: Int=>Unit) = {
    import java.util.concurrent._
    val e=Executors.newCachedThreadPool //newFixedThreadPool(cpus+1)
    (0 until n).foreach(i=>e.execute(new Runnable {def run = f(i)}))
    e.shutdown
    e.awaitTermination(Math.MAX_LONG, TimeUnit.SECONDS)
  }
}

answered Nov 17, 2009 at 21:36

Jonathan Graehl

9,33138 silver badges40 bronze badges

Comments

Daniel C. Sobral · Accepted Answer · 2009-11-17 22:16:24Z

2

I'd use scala.actors.Futures:

vals.foreach(t => scala.actors.Futures.future(f(t)))

answered Nov 17, 2009 at 22:16

Daniel C. Sobral

298k88 gold badges508 silver badges688 bronze badges

Comments

Apocalisp · Accepted Answer · 2010-07-15 16:09:53Z

2

The latest release of Functional Java has some higher-order concurrency features that you can use.

import fjs.F._
import fj.control.parallel.Strategy._
import fj.control.parallel.ParModule._
import java.util.concurrent.Executors._

val pool = newCachedThreadPool
val par = parModule(executorStrategy[Unit](pool))

And then...

par.parMap(vals, f)

Remember to shutdown the pool.

edited Jul 15, 2010 at 16:09

answered Nov 19, 2009 at 22:05

Apocalisp

35.1k8 gold badges109 silver badges158 bronze badges

Comments

Roland · Accepted Answer · 2014-05-14 07:38:28Z

0

You can use the Parallel Collections from the Scala standard library. They're just like ordinary collections, but their operations run in parallel. You just need to put a par call before you invoke some collections operation.

import scala.collection._

val array = new Array[String](10000)
for (i <- (0 until 10000).par) array(i) = i.toString

answered May 14, 2014 at 7:38

Roland

212 bronze badges

Collectives™ on Stack Overflow

Concurrent map/foreach in scala

7 Answers 7

Comments

Comments

6 Comments

Comments

Comments

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

7 Answers 7

Comments

Comments

6 Comments

Comments

Comments

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related