Author Archives: Lance

s_mach.concurrent: Futures utility library

Versions

Scala 2.11.0
s_mach.concurrent 0.1

Code

https://github.com/S-Mach/s_mach.concurrent

Overview

s_mach.concurrent is an open-source Scala utility library that extends the standard scala.concurrent library. It adds new types and functions for easily controlling concurrent execution flow and helps to fix a fundamental flaw in the standard scala.concurrent.Future implementation.

Imports

All code examples assume the following imports:

import scala.util._
import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import s_mach.concurrent._
import s_mach.concurrent.util._

case class Item(id: String, value: Int, relatedItemId: String)
def read(id: String) : Future[Item] = Future { Thread.sleep(1000); println(id); Item(id,id.toInt,(id.toInt+1).toString) }
def readFail(id: String) : Future[Item] = Future { Thread.sleep(1000); println(id); throw new RuntimeException(id.toString) }
def longRead(id: String) : Future[Item] = Future { Thread.sleep(2000); println(id); Item(id,id.toInt,(id.toInt+1).toString) }
def write(id: String, item: Item) : Future[Boolean] = Future { Thread.sleep(1000); println(id); true }
def writeFail(id: String, item: Item) : Future[Boolean] = Future { Thread.sleep(1000); println(id); throw new RuntimeException(id.toString) }

import scala.util._

import scala.concurrent._

import scala.concurrent.ExecutionContext.Implicits.global

import scala.concurrent.duration._

import s_mach.concurrent._

import s_mach.concurrent.util._

case class Item(id: String, value: Int, relatedItemId: String)

def read(id: String) : Future[Item] = Future { Thread.sleep(1000); println(id); Item(id,id.toInt,(id.toInt+1).toString) }

def readFail(id: String) : Future[Item] = Future { Thread.sleep(1000); println(id); throw new RuntimeException(id.toString) }

def longRead(id: String) : Future[Item] = Future { Thread.sleep(2000); println(id); Item(id,id.toInt,(id.toInt+1).toString) }

def write(id: String, item: Item) : Future[Boolean] = Future { Thread.sleep(1000); println(id); true }

def writeFail(id: String, item: Item) : Future[Boolean] = Future { Thread.sleep(1000); println(id); throw new RuntimeException(id.toString) }

Concurrently

When first using Future with a for-comprehension, it is natural to assume the following will produce concurrent operation:

Example 1: Incorrect Future concurrency

for {
  i1 <- read("1")
  i2 <- read("2")
  i3 <- read("3")
} yield (i1,i2,i3)

for {

i1 <- read("1")

i2 <- read("2")

i3 <- read("3")

} yield (i1,i2,i3)

Sadly, this code will compile and run just fine, but it will not execute concurrently. To correctly implement concurrent operation, the following standard pattern is used:

Example 2: Correct Future concurrency:

val f1 = read("1")
val f2 = read("2")
val f3 = read("3")
val future = { // necessary for pasting into repl
  for {
    i1 <- f1
    i2 <- f2
    i3 <- f3
  } yield (i1,i2,i3)
}

val f1 = read("1")

val f2 = read("2")

val f3 = read("3")

val future = { // necessary for pasting into repl

for {

i1 <- f1

i2 <- f2

i3 <- f3

} yield (i1,i2,i3)

}

To get concurrent operation, all of the futures must be started before the for-comprehension. The for-comprehension is a monadic workflow. It captures commands that must take place in a specific sequential order. The pattern in example 2 is necessary because Scala lacks an applicative workflow: a workflow that captures commands that may be run in any order. s_mach.concurrent adds an applicative workflow method for futures: concurrently. This method can more concisely express the pattern above:

Example 3: New concurrently method

for {
  (i1,i2,i3) <- concurrently(read("1"), read("2"), read("3"))
} yield (i1,i2,i3)

for {

(i1,i2,i3) <- concurrently(read("1"), read("2"), read("3"))

} yield (i1,i2,i3)

In the example above, all futures are started at the same time and fed to the concurrently method. The method returns a Future[(Int,Int,Int)] which completes once all supplied futures complete. After this returned Future completes, the tuple value results can be extracted using normal Scala idioms. The concurrently method also fixes problems with scala.concurrent exception handling (see the Under the hood: Merge section below).

Transforming and traversing collections serially and concurrently

A common task when working with futures is either transforming or traversing a collection that will call a method that returns a future. The standard idiom for performing this task only provides methods for concurrent operation and, with enough nesting, leads to difficult to read code:

Example 4: Transform and traverse collections, standard method

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList
val future = { // necessary for pasting into repl
  for {
    oomItem <- {
      println("Reading...")
      oomItemIdBatch
        // Serially perform read of each batch
        .foldLeft(Future.successful(List[Item]())) { (facc, idBatch) =>
          for {
            acc <- facc
            // Concurrently read batch
            oomItem <- Future.sequence(idBatch.map(read))
          } yield acc ::: oomItem
        }
    }
    _ = println("Computing...")
    oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(2).toList
    oomResult <- {
      println("Writing...")
      oomNewItemBatch
        // Serially perform write of each batch
        .foldLeft(Future.successful(List[Boolean]())) { (facc, itemBatch) =>
          for {
            acc <- facc
            // Concurrently write batch
            oomResult <- Future.sequence(itemBatch.map(item => write(item.id, item)))
          } yield acc ::: oomResult
        }
    }
  } yield oomResult.forall(_ == true)
}

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList

val future = { // necessary for pasting into repl

for {

oomItem <- {

println("Reading...")

oomItemIdBatch

// Serially perform read of each batch

.foldLeft(Future.successful(List[Item]())) { (facc, idBatch) =>

for {

acc <- facc

// Concurrently read batch

oomItem <- Future.sequence(idBatch.map(read))

} yield acc ::: oomItem

}

_ = println("Computing...")

oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(2).toList

oomResult <- {

println("Writing...")

oomNewItemBatch

// Serially perform write of each batch

.foldLeft(Future.successful(List[Boolean]())) { (facc, itemBatch) =>

for {

acc <- facc

// Concurrently write batch

oomResult <- Future.sequence(itemBatch.map(item => write(item.id, item)))

} yield acc ::: oomResult

}

} yield oomResult.forall(_ == true)

}

The same code, rewritten using s_mach.concurrent:

Example 5: Using s_mach.concurrent to serially or concurrently transform and traverse collections:

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList
val future = { // necessary for pasting into repl
  for {
    oomItem <- {
      println("Reading...")
      oomItemIdBatch.serially.flatMap(_.concurrently.map(read))
    }
    _ = println("Computing...")
    oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(10).toVector
    oomResult <- {
      println("Writing...")
      oomNewItemBatch.serially.flatMap(_.concurrently.map(item => write(item.id, item)))
    }
  } yield oomResult.forall(_ == true)
}

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList

val future = { // necessary for pasting into repl

for {

oomItem <- {

println("Reading...")

oomItemIdBatch.serially.flatMap(_.concurrently.map(read))

}

_ = println("Computing...")

oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(10).toVector

oomResult <- {

println("Writing...")

oomNewItemBatch.serially.flatMap(_.concurrently.map(item => write(item.id, item)))

}

} yield oomResult.forall(_ == true)

}

Transforming and traversing collections using workers

Example 6: Using s_mach.concurrent workers to transform and traverse collections:

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList
val future = { // necessary for pasting into repl
  for {
    oomItem <- {
      println("Reading...")
      oomItemIdBatch.workers(2).flatMap(_.workers(4).map(read))
    }
    _ = println("Computing...")
    oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(10).toVector
    oomResult <- {
      println("Writing...")
      oomNewItemBatch.workers(2).flatMap(_.workers(4).map(item => write(item.id, item)))
    }
  } yield oomResult.forall(_ == true)
}

val oomItemIdBatch = (1 to 10).toList.map(_.toString).grouped(2).toList

val future = { // necessary for pasting into repl

for {

oomItem <- {

println("Reading...")

oomItemIdBatch.workers(2).flatMap(_.workers(4).map(read))

}

_ = println("Computing...")

oomNewItemBatch = oomItem.map(item => item.copy(value = item.value + 1)).grouped(10).toVector

oomResult <- {

println("Writing...")

oomNewItemBatch.workers(2).flatMap(_.workers(4).map(item => write(item.id, item)))

}

} yield oomResult.forall(_ == true)

}

Under the hood: Merge method

Powering both the general concurrently method and the collection .concurrently.map, .concurrently.flatMap and .concurrently.foreach methods are the merge and flatMerge methods. The merge method performs the same function as Future.sequence (it calls Future.sequence internally) but it ensures that the returned future completes immediately after an exception occurs in any of the futures. Because Future.sequence waits on all futures in left to right order before completing, an exception thrown at the beginning of the computation by a future at the far right will not be detected until after all other futures have completed. For long running computations, this can mean a significant amount of wasted time waiting on futures to complete whose results will be discarded. Also, while the scala parallel collections correctly handle multiple concurrent exceptions, Future.sequence only returns the first exception encountered. In Future.sequence, all further exceptions past the first are discarded. The merge and flatMerge methods fixes these problems by throwing ConcurrentThrowable. ConcurrentThrowable has a member method to access both the first exception thrown and a future of all exceptions thrown during the computation.

Example 7: Future.sequence gets stuck waiting on longRead to complete and only returns the first exception:

scala> val t = Future.sequence(Vector(longRead("1"),readFail("2"),readFail("3"),read("4"))).getTry
3
4
2
1
t: scala.util.Try[scala.collection.immutable.Vector[Item]] = Failure(java.lang.RuntimeException: 2)

scala>

scala> val t = Future.sequence(Vector(longRead("1"),readFail("2"),readFail("3"),read("4"))).getTry

t: scala.util.Try[scala.collection.immutable.Vector[Item]] = Failure(java.lang.RuntimeException: 2)

scala>

Example 8: merge method fails immediately on the first exception and throws ConcurrentThrowable, which can retrieve all exceptions:

scala> val t = Vector(longRead("1"),readFail("2"),readFail("3"),read("4")).merge.getTry
2
t: scala.util.Try[scala.collection.immutable.Vector[Item]] = Failure(ConcurrentThrowable(java.lang.RuntimeException: 2))
3

scala> 4
1

scala> val allFailures = t.failed.get.asInstanceOf[ConcurrentThrowable].futAllFailure.get
allFailures: Vector[Throwable] = Vector(java.lang.RuntimeException: 2, java.lang.RuntimeException: 3)

scala> val t = Vector(longRead("1"),readFail("2"),readFail("3"),read("4")).merge.getTry

t: scala.util.Try[scala.collection.immutable.Vector[Item]] = Failure(ConcurrentThrowable(java.lang.RuntimeException: 2))

scala> 4

scala> val allFailures = t.failed.get.asInstanceOf[ConcurrentThrowable].futAllFailure.get

allFailures: Vector[Throwable] = Vector(java.lang.RuntimeException: 2, java.lang.RuntimeException: 3)

Concurrent Semaphore

TODO

Example 9: Semaphore

val s = Semaphore(10)

val promise = Promise[Int]()

val f1 = s.acquire(10) { () => println(1);promise.future }
val f2 = s.acquire(8) { () => println(2);Thread.sleep(1000);2.future }
val f3 = s.acquire(2) { () => println(3);Thread.sleep(1000);3.future }
val f4 = s.acquire(1) { () => println(4);4.future }

promise.success(1)

val s = Semaphore(10)

val promise = Promise[Int]()

val f1 = s.acquire(10) { () => println(1);promise.future }

val f2 = s.acquire(8) { () => println(2);Thread.sleep(1000);2.future }

val f3 = s.acquire(2) { () => println(3);Thread.sleep(1000);3.future }

val f4 = s.acquire(1) { () => println(4);4.future }

promise.success(1)

Concurrent Lock

TODO

Example 9: Lock

val lock = Lock()

val promise = Promise[Int]()
val f1 = lock { () => println("1");promise.future }
val f2 = lock { () => println("2");2.future }
val f3 = lock { () => println("3");3.future }

promise.success(1)

val lock = Lock()

val promise = Promise[Int]()

val f1 = lock { () => println("1");promise.future }

val f2 = lock { () => println("2");2.future }

val f3 = lock { () => println("3");3.future }

promise.success(1)

scala> :paste
// Entering paste mode (ctrl-D to finish)

val lock = Lock()

val promise = Promise[Int]()
val f1 = lock { () => println(1);promise.future }
val f2 = lock { () => println(2);2.future }
val f3 = lock { () => println(3);3.future }


// Exiting paste mode, now interpreting.

1
lock: s_mach.concurrent.util.Lock = s_mach.concurrent.util.Lock$LockImpl@2578c32a
promise: scala.concurrent.Promise[Int] = scala.concurrent.impl.Promise$DefaultPromise@1b550d9e
f1: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@1b550d9e
f2: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@6cf10e16
f3: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@45d3fe3a

scala> promise.success(1)
2
res2: promise.type = scala.concurrent.impl.Promise$DefaultPromise@37af4505
3

scala> :paste

// Entering paste mode (ctrl-D to finish)

val lock = Lock()

val promise = Promise[Int]()

val f1 = lock { () => println(1);promise.future }

val f2 = lock { () => println(2);2.future }

val f3 = lock { () => println(3);3.future }

// Exiting paste mode, now interpreting.

lock: s_mach.concurrent.util.Lock = s_mach.concurrent.util.Lock$LockImpl@2578c32a

promise: scala.concurrent.Promise[Int] = scala.concurrent.impl.Promise$DefaultPromise@1b550d9e

f1: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@1b550d9e

f2: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@6cf10e16

f3: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@45d3fe3a

scala> promise.success(1)

res2: promise.type = scala.concurrent.impl.Promise$DefaultPromise@37af4505

ConcurrentQueue

s_mach.concurrent provides a basic concurrent queue trait ConcurrentQueue that allows for asynchronous buffering operations, including operations on collections of items. Currently only one implementation, ConcurrentListQueue is provided.

Example 11: ConcurrentListQue

val q = new ConcurrentListQueue[String]()

// Completes when input is available
val f1 = q.poll()
f1 foreach println

// Completes the future above
q.offer("some input")

// Complete when all input is available
val f2 = q.poll(3)
f2 foreach println

// Completes the future above
q.offer("input1")
q.offer(List("input2","input3","input4"))

// If input is available now, future completes immediately:
q.poll() foreach println

val q = new ConcurrentListQueue[String]()

// Completes when input is available

val f1 = q.poll()

f1 foreach println

// Completes the future above

q.offer("some input")

// Complete when all input is available

val f2 = q.poll(3)

f2 foreach println

// Completes the future above

q.offer("input1")

q.offer(List("input2","input3","input4"))

// If input is available now, future completes immediately:

q.poll() foreach println

Utility methods

s_mach.concurrent provides a few utility methods for writing more concise and DRY code when working with Future:

Example 12: Utility methods

val fr : Future[Item] = read("1")

// Convert a Future[A] to Future[Try[A]] that always succeeds
val tt : Future[Try[Item]] = fr.toTry

// Convert a Future[A] to a Future[X] that always succeeds
val f : Future[String] = fr.fold({ i:Item => i.id.toString }, { t:Throwable => t.toString})

// Convert a Future[A] to a Future[Future[X]] that is flattened to Future[X] and that always succeeds
val f : Future[String] = fr.flatFold(
  { i:Item => Future.successful(i.id.toString) },
  { t:Throwable => Future.successful(t.toString) }
)

// Throw away the result, but ensure exceptions are reported to ExecutionContext.reportFailure
fr.discard

val fr : Future[Item] = read("1")

// Convert a Future[A] to Future[Try[A]] that always succeeds

val tt : Future[Try[Item]] = fr.toTry

// Convert a Future[A] to a Future[X] that always succeeds

val f : Future[String] = fr.fold({ i:Item => i.id.toString }, { t:Throwable => t.toString})

// Convert a Future[A] to a Future[Future[X]] that is flattened to Future[X] and that always succeeds

val f : Future[String] = fr.flatFold(

{ i:Item => Future.successful(i.id.toString) },

{ t:Throwable => Future.successful(t.toString) }

)

// Throw away the result, but ensure exceptions are reported to ExecutionContext.reportFailure

fr.discard

Sugar methods

s_mach.concurrent also provides a number of syntatic-sugar methods for writing more concise and DRY code when working with Future:

Example 13: Sugar methods

val vf = Vector(read("1"), read("2"), read("3"))
val fr = read("4")

// Future.sequence(vf)
val fv : Future[Vector[Item] = vf.sequence

// Await.result(fr, Duration.Inf)
val g : Item = fr.get

// Await.result(fr, 5.seconds)
val g : Int = fr.get(5.seconds)

// Await.ready(fr, Duration.Inf).value.get
val gt : Future[Try[Int]] = fr.getTry

// Await.ready(fr, Duration.Inf).value.get
val gt : Future[Try[Int]] = fr.getTry(5.seconds)

val vf = Vector(read("1"), read("2"), read("3"))

val fr = read("4")

// Future.sequence(vf)

val fv : Future[Vector[Item] = vf.sequence

// Await.result(fr, Duration.Inf)

val g : Item = fr.get

// Await.result(fr, 5.seconds)

val g : Int = fr.get(5.seconds)

// Await.ready(fr, Duration.Inf).value.get

val gt : Future[Try[Int]] = fr.getTry

// Await.ready(fr, Duration.Inf).value.get

val gt : Future[Try[Int]] = fr.getTry(5.seconds)

Configure IntelliJ IDEA For Scala and SBT

Leave a reply

This post has moved. See IntelliJ IDEA: Configure for Scala and SBT

Scala: Type-classes

Leave a reply

Versions

Scala 2.11.0

What is a type-class?

A type-class a type that is used to add new behaviors to a “primary” type, without having to extend or modify the primary type¹. In Scala, there is no native support for type-classes. Instead type-classes are implemented by following a design pattern². The most common design pattern is to create a Trait for the type-class that accepts one type parameter as the primary type (though more are supported⁶) and to define one or more abstract methods that add new behaviors to the primary type.

Example

trait Printable[A] { // <b><1></b>
  def print(a: A) : String // <b><2></b>
}

trait Printable[A] { // <1>

def print(a: A) : String // <2>

}

Printable accepts one primary type parameter A
Printable adds a new print method to A

A type is added as a member of a type-class by implementing the type-class for that type:

implicit object PrintableInt extends Printable[Int] {
  def print(i: Int) = "Int -> " + i.toString
}
...
scala> implicitly[Printable[Int]].print(1)
res0: String = Int -> 1

implicit object PrintableInt extends Printable[Int] {

def print(i: Int) = "Int -> " + i.toString

}

...

scala> implicitly[Printable[Int]].print(1)

res0: String = Int -> 1

Note	Simple type-class implementations can often be automatically created using macros or implicits.

In Scala, some “semantic sugar” can be used to make using type-classes easier:

implicit class PrintableSugar[A](val self: A) extends AnyVal {
  def print(implicit printable:Printable[A]) = printable.print(self)
}
....
scala> 1.print
res1: String = Int -> 1

implicit class PrintableSugar[A](val self: A) extends AnyVal {

def print(implicit printable:Printable[A]) = printable.print(self)

}

....

scala> 1.print

res1: String = Int -> 1

So what happens if we invoke print on a type that doesn’t belong to the Printable type-class?

scala> val gigawatts = 1.21
gigawatts: Double = 1.21

scala> gigawatts.print
<console>:12: error: could not find implicit value for parameter printable: Printable[Double]
              gigawatts.print

scala>

scala> val gigawatts = 1.21

gigawatts: Double = 1.21

scala> gigawatts.print

<console>:12: error: could not find implicit value for parameter printable: Printable[Double]

gigawatts.print

scala>

Compiler error! Scala code that doesn’t properly implement a type-class will never compile.

Why use type-classes?

Type-classes are used to create “ad hoc” polymorphism. They allow adding types as members of a type-class at any time, including after the original definition of the type. Object Oriented Design (OOD) uses inheritance to create polymorphism, but this requires knowing all possible desired instances of polymorphism at design time (though the Visitor Pattern can be used to alleviate this problem somewhat). If further polymorphism is required, most likely many types will need to be refactored.

Type-classes are also highly modular. Users that aren’t interested in adding the new behavior can simply ignore the type-class. This can significantly reduce the complexity of the primary type. Often when using type-classes and case classes in Scala, the case class doesn’t need methods at all! Also, users of type-class libraries can select specific implementations of a type-class for the primary type that best suits their specific needs at that moment. In Scala, this is controlled by importing the desired type-class implementation:

trait Printable[A] {
  def print(a: A) : String
}
object PrintableImplicits {
  implicit object PrintableInt extends Printable[Int] { // <b><1></b>
    def print(i: Int) = "Int -> " + i.toString
  }
}
implicit class PrintableSugar[A](val self: A) extends AnyVal {
  def print(implicit printable:Printable[A]) = printable.print(self)
}
object Example {
  def foo {
    import PrintableImplicits._
    println(1.print)
  }
  def bar {
    implicit val myPrintableInt = new Printable[Int] { // <b><2></b>
      def print(i: Int) = "%03d" format i
    }
    println(1.print)
  }
}
scala> Example.foo
Int -> 1
scala> Example.bar
001

trait Printable[A] {

def print(a: A) : String

}

object PrintableImplicits {

implicit object PrintableInt extends Printable[Int] { // <1>

def print(i: Int) = "Int -> " + i.toString

}

implicit class PrintableSugar[A](val self: A) extends AnyVal {

def print(implicit printable:Printable[A]) = printable.print(self)

}

object Example {

def foo {

import PrintableImplicits._

println(1.print)

}

def bar {

implicit val myPrintableInt = new Printable[Int] { // <2>

def print(i: Int) = "%03d" format i

}

println(1.print)

}

scala> Example.foo

Int -> 1

scala> Example.bar

001

A wrapper object is used here to allow for better control of where the implicit is imported
Defining a custom Printable for Int

Note	If you pasted the previous examples into the console, you will need to restart the console to remove the global implicit for Printable[Int].

An Extended Example

When I first read about type-classes, I found it difficult to understand what marginal value they added over classic OOD polymorphism. But overtime, I’ve grown to love type-classes. For me it took encountering the many painful refactorings that ultimately result from overuse of classic OOD polymorphism. Refactoring gets old quickly. I’ve built this example to help illustrate this idea.

Once upon a time, I created a very basic inheritance structure for modeling the tools in my shed:

trait Part
trait Screw extends Part
trait Tool
class Hammer extends Tool {
  def pound(something: Any) : Unit = ???
}
class Screwdriver extends Tool {
  def turn(screw: Screw) : Unit = ???
}
class Rake extends Tool {
  def gather(something: Any) : Unit = ???
}

trait Part

trait Screw extends Part

trait Tool

class Hammer extends Tool {

def pound(something: Any) : Unit = ???

}

class Screwdriver extends Tool {

def turn(screw: Screw) : Unit = ???

}

class Rake extends Tool {

def gather(something: Any) : Unit = ???

}

This classic OOD model suited my needs and got the job done for a long, long time.

But one day I realize I can’t find my hammer. I’m working on my new IKEA shelf and I just have some finishing nails that I need to hammer in to finish. I poke around my shed and realize my favorite screwdriver (“big bertha”) could probably get the job done! I awkwardly pound my finishing nails in using bertha, but my post-IKEA-assembly-bliss is cut short. I have a problem: Screwdrivers can pound! I’m in a hurry to get my new shelf into my house, so I quickly refactor my model:

trait Tool {
  def pound(something: Any) : Unit
}
class Hammer extends Tool {
  def pound(something: Any) : Unit = ???
}
class Screwdriver extends Tool {
  def pound(something: Any) : Unit = ???
  def turn(screw: Screw) : Unit = ???
}
class Rake extends Tool {
  def pound(something: Any) : Unit = throw new UnsupportedOperationException
  def gather(something: Any) : Unit = ???
}

trait Tool {

def pound(something: Any) : Unit

}

class Hammer extends Tool {

def pound(something: Any) : Unit = ???

}

class Screwdriver extends Tool {

def pound(something: Any) : Unit = ???

def turn(screw: Screw) : Unit = ???

}

class Rake extends Tool {

def pound(something: Any) : Unit = throw new UnsupportedOperationException

def gather(something: Any) : Unit = ???

}

This is far from ideal, but I’m in a hurry, so I commit my code and call it a day. Later that night, I’m restless in bed. I realize that if I were to loan my tools to a neighbor, he might assume that because my tool model has the pound method, he can pound things with any of my tools. This might break my rake but I made sure he can’t do that. But my model shouldn’t give him that idea at all. The next morning, I refactor again:

trait Tool
class ToolThatPounds extends Tool {
  def pound(something: Any) : Unit = ???
}
class Hammer extends ToolThatPounds {
  def pound(something: Any) : Unit = ???
}
class Screwdriver extends ToolThatPounds {
  def pound(something: Any) : Unit = ???
  def turn(screw: Screw) : Unit = ???
}
class Rake extends Tool {
  def gather(something: Any) : Unit = ???
}

trait Tool

class ToolThatPounds extends Tool {

def pound(something: Any) : Unit = ???

}

class Hammer extends ToolThatPounds {

def pound(something: Any) : Unit = ???

}

class Screwdriver extends ToolThatPounds {

def pound(something: Any) : Unit = ???

def turn(screw: Screw) : Unit = ???

}

class Rake extends Tool {

def gather(something: Any) : Unit = ???

}

Much better! My neighbor will no longer assume he can use my rake to pound things. Though I’ve created a class that doesn’t really represent anything real. Also, the more I think about the stuff in my shed, the more I realize there is a ton of stuff in there that could pound things. I could have used some of my spare piping to pound things as well! If I want to represent this I will have to refactor again!

Luckily, I spend some time searching the web and discover the pattern to end all this nasty refactoring: type-classes. I refactor one final time:

trait Part
trait Screw extends Part
trait Piping extends Part
trait Tool
class Hammer extends Tool
class Screwdriver extends Tool
class Rake extends Tool

trait Pound[A] {
  def pound(a: A, something: Any) : Unit
}
trait TurnScrew[A] {
  def turn(a: A, screw: Screw) : Unit
}
trait Gather[A] {
  def gather(a: A, something: Any) : Unit
}

implicit val PoundWithHammer : Pound[Hammer] = ???
implicit val PoundWithScrewdriver : Pound[Screwdriver] = ???
implicit val PoundWithPiping : Pound[Piping] = ???
implicit val TurnScrewWithScrewdriver : TurnScrew[Screwdriver] = ???
implicit val GatherWithRake : Gather[Rake] = ???

trait Part

trait Screw extends Part

trait Piping extends Part

trait Tool

class Hammer extends Tool

class Screwdriver extends Tool

class Rake extends Tool

trait Pound[A] {

def pound(a: A, something: Any) : Unit

}

trait TurnScrew[A] {

def turn(a: A, screw: Screw) : Unit

}

trait Gather[A] {

def gather(a: A, something: Any) : Unit

}

implicit val PoundWithHammer : Pound[Hammer] = ???

implicit val PoundWithScrewdriver : Pound[Screwdriver] = ???

implicit val PoundWithPiping : Pound[Piping] = ???

implicit val TurnScrewWithScrewdriver : TurnScrew[Screwdriver] = ???

implicit val GatherWithRake : Gather[Rake] = ???

Perfection! No refactoring needed ever again. As I find things around my shed that can pound, I simply add a new type-class implementation. Also, I can do the same for things that could turn screws or gather leaves. Super flexible!

When to use OOD polymorphism

Some folks might want you to think that you should always use type-classes. But in Scala they require significantly more boilerplate to implement. Also, because Scala doesn’t natively support type-classes, code readers must know the Scala type-class pattern to understand how they work.

I’ve found that the best time to use OOD polymorphism over type-classes is when all of the possible polymorphic methods are known up front and expansion to future use cases is unlikely. A great example of this is the Scala collections library. It is very unlikely that a new method will be added to IndexedSeq or that Traversable will suddenly need the ability to get a value by its index. On the flip side, implementing the collections library with only type-classes would introduce a ton of complexity. Each method on Traversable would need its own type-class. That’s at least 50 type-classes for 50 methods! (Though this number could be reduced significantly by grouping related methods into a few type-classes. See StringOps and StringLike for examples.)

When to use type-classes

In choosing to use type-classes, I’ve found that the clearest use case for them is when I might need to add a behavior to almost any type. The best example of this is for serialization/marshalling/binding etc. Converting to and from JSON, BSON, XML, etc is something that is commonly needed for most every type. Also, sometimes I like to swap out implementations based on what I’m doing. I might have a different JSON serializer depending on the recipient of the JSON.

In many cases, the choice of OOD inheritance or type-classes to achieve polymorphism can be somewhat arbitrary. Scala gives me a ton of flexibility and the downside of all of that choice is that many times, at least within the context of Scala, the question is simply one of what color to paint my shed.

Sources

Scala: Cardinality prefix naming convention

Leave a reply

Naming Option variables

Many times while writing Scala, I find myself having to look up the type of a variable that I’m working with. Often it is because I’m unsure whether the variable is an Option. This means having to depend on my IDE shortcut to show me the type or having to navigate to the source file.

val hotel : Hotel = ???
val stateCode = hotel.stateCode // <b><1></b>

1 2	val hotel : Hotel = ??? val stateCode = hotel.stateCode // <b><1></b>

Is stateCode an Option?

Other times, I’m transforming a given Option variable into its base type and I’m forced to make up an awkward name for the inner name of the value contained in the Option:

def getRawPhoneNumber(...) : Option[String] = ???

def doSomething(s: String) : Future[Unit] = ???

val phoneNumber = svc.getRawPhoneNumber(...)
phoneNumber match {
  case Some(gotAPhoneNumber) => doSomething(gotAPhoneNumber) // <b><1></b>
  case None => Future.succcessful(())
}

def getRawPhoneNumber(...) : Option[String] = ???

def doSomething(s: String) : Future[Unit] = ???

val phoneNumber = svc.getRawPhoneNumber(...)

phoneNumber match {

case Some(gotAPhoneNumber) => doSomething(gotAPhoneNumber) // <1>

case None => Future.succcessful(())

}

Ugly! What to name this value?!?

While this example is trivial, it gets complicated fast once you have 15 or so fields to deal with at once:

case class Hotel(
  id: Int,
  chainCode: Option[String],
  name: String,
  addressProvider: AddressProviderEnum,
  addressLine1: String,
  addressLine2: Option[String],
  city: String,
  stateCode: Option[String],
  postalCode: Option[String],
  countryCode: String,
  phoneNumber: String,
  location: PlaceGeocode,
  providerMapping: Set[ProviderHotelId],
  propertyType: PropertyTypeEnum
)

case class Hotel(

id: Int,

chainCode: Option[String],

name: String,

addressProvider: AddressProviderEnum,

addressLine1: String,

addressLine2: Option[String],

city: String,

stateCode: Option[String],

postalCode: Option[String],

countryCode: String,

phoneNumber: String,

location: PlaceGeocode,

providerMapping: Set[ProviderHotelId],

propertyType: PropertyTypeEnum

)

Without having to flip back to this source file, I don’t know which fields are Option!

I’ve started using a super-simple naming convention that explicitly declares the cardinality of the variable to both avoid the naming problem and to increase readability of the code. For Option, I simply add the prefix “opt” to the Option variable:

val optPhoneNumber = svc.getRawPhoneNumber(...)

optPhoneNumber match {
  case Some(phoneNumber) => doSomething(phoneNumber) // <b><1></b>
  case None => Future.succcessful(())
}

val optPhoneNumber = svc.getRawPhoneNumber(...)

optPhoneNumber match {

case Some(phoneNumber) => doSomething(phoneNumber) // <1>

case None => Future.succcessful(())

}

Ahhhh so much better. I know optPhoneNumber is an Option AND I have an obvious (and highly readable) name for the inner variable.

Naming collection variables

There is also another problem that I encounter when naming things: what the heck do I name collection variables? For plural words in English, the convention is mostly to just append s. And this mostly works:

val hotels : Vector[Hotel] = ???
val names : Vector[String] = ???
val phoneNumbers : Vector[String] = ???

val hotels : Vector[Hotel] = ???

val names : Vector[String] = ???

val phoneNumbers : Vector[String] = ???

Except when it doesn’t work:

val amenities : Vector[String] = ??? // <b><1></b>
val amenitys : Vector[String] = ??? // <b><2></b>

1 2	val amenities : Vector[String] = ??? // <b><1></b> val amenitys : Vector[String] = ??? // <b><2></b>

Ack. Ugly!!
Ugly and people think I don’t know basic English grammar. They won’t understand my need for regularity.

Also, there is a more serious problem here. If a collection has at least one item, it is always safe to call head. This condition can often be guaranteed at compile time, but how would anyone know? This leaves me with the same problem I had with null in Java: I must either defensively test everywhere for the condition OR I must depend on it being in the documentation.

val hotels : Vector[Hotel] = ???

val unsafeFirst = hotels.head // <b><1></b>
val optSafeFirst = hotels.size match { // <b><2></b>
  case 0 => None
  case _ => Some(hotels.head)
}

val hotels : Vector[Hotel] = ???

val unsafeFirst = hotels.head // <1>

val optSafeFirst = hotels.size match { // <2>

case 0 => None

case _ => Some(hotels.head)

}

Maybe this is wrong? Hopefully, documentation guarantees that hotels is non-empty.
Or I can be safe about this. But now I have an Option and my code grows ever more complex (for possibly no good reason if hotels is guaranteed to have at least one item).

In code I write now, I fix both of these problems by keeping the variable name in the English singular, and prefixing a plural variable prefix to the variable name:

“zom”: Read as “zero or more”
“oom”: Read as “one or more”
“all”: Read as “all of them” (Most likely a very large collection, but technically equivalent to “zom”)

Example:

val amenities : Vector[String] = ??? // <b><1></b>

val zomAmenity : Vector[String] = ??? // <b><2></b>

val oomAmenity : Vector[String] = ??? // <b><3></b>
val amenity = oomAmenity.head

val allAmenity : Vector[String] = ??? // <b><4></b>

allAmenity.foreach { amenity => // <b><5></b>
  ...
}

val amenities : Vector[String] = ??? // <1>

val zomAmenity : Vector[String] = ??? // <2>

val oomAmenity : Vector[String] = ??? // <3>

val amenity = oomAmenity.head

val allAmenity : Vector[String] = ??? // <4>

allAmenity.foreach { amenity => // <5>

...

}

Ugly. Also is head safe?
I know head is not safe here
I know there is at least one amenity. head is safe.
I know there should be at least one amenity. head is probably safe. More importantly I should avoid excessive transforms on the collection, all those copies will take up a lot of memory!
Bonus! Like Option above, I have an easy and readable decision for what to name the inner function parameter.

Cardinality Prefix naming convention

Here is the full listing of the cardinality prefix naming convention:

Name all variables in the English singular form
- Ex: hotel, phoneNumber, amenity
If the variable is an option, prefix with “opt”
- Ex: optHotel, optPhoneNumber, optAmenity
If the variable is a collection AND guarantees at least one member, prefix with “oom”
- Ex: oomHotel, oomPhoneNumber, oomAmenity
If the variable is a collection AND does not guarantee at least one member, prefix with “zom”
- Ex: zomHotel, zomPhoneNumber, zomAmenity
If the variable is a collection of all of the values (very large collection), prefix with “all”
- Ex: allHotel, allPhoneNumber, allAmenity

Example

The large case class from above, after applying the naming convention:

case class Hotel(
  id: Int,
  optChainCode: Option[String],
  name: String,
  addressProvider: AddressProviderEnum,
  addressLine1: String,
  optAddressLine2: Option[String],
  city: String,
  optStateCode: Option[String], // <b><1></b>
  optPostalCode: Option[String],
  countryCode: String,
  phoneNumber: String,
  location: PlaceGeocode,
  oomProviderMapping: Set[ProviderHotelId], // <b><2></b>
  propertyType: PropertyTypeEnum
)

case class Hotel(

id: Int,

optChainCode: Option[String],

name: String,

addressProvider: AddressProviderEnum,

addressLine1: String,

optAddressLine2: Option[String],

city: String,

optStateCode: Option[String], // <1>

optPostalCode: Option[String],

countryCode: String,

phoneNumber: String,

location: PlaceGeocode,

oomProviderMapping: Set[ProviderHotelId], // <2>

propertyType: PropertyTypeEnum

)

I don’t need to read the source to know optStateCode is an Option
I know oomProviderMapping has at least one item and that head is safe to call

MongoDB: Get a random record

Leave a reply

Get a random record in Mongo

A quick and simple function that picks a random record from a cursor:

var getRnd = function(q) {
  var count = q.count()
  var o = q.limit(-1).skip(Math.random() * q.count()).next()
  printjson(o)
}

var getRnd = function(q) {

var count = q.count()

var o = q.limit(-1).skip(Math.random() * q.count()).next()

printjson(o)

}

Example:

> var getRnd = function(q) {
...   var count = q.count()
...   var o = q.limit(-1).skip(Math.random() * q.count()).next()
...   printjson(o)
... }
> use hotel
switched to db hotel
> getRnd(db.hotelWithDetail.find())
{
        "_id" : ObjectId("538f5fa6d9805dc69ea218b5"),
        "hotel" : {
                "id" : 88173,
                "name" : "Pantai Inn",
                "addressProvider" : "GoogleMaps",
                "addressLine1" : "1003 Coast Boulevard",
                "city" : "San Diego",
                "countryCode" : "US",
                "phoneNumber" : "+1-858-224-7600",
                "location" : {
                        "latitude" : 32.8485673,
                        "longitude" : -117.275847,
                        "geocodeProvider" : "GoogleMaps",
                        "geocodeAccuracy" : "Vicinity",
                        "created" : ISODate("2014-04-19T17:48:45.101Z")
                },
                "oomProviderMapping" : [
                        {
                                "provider" : "TP",
                                "providerId" : "71990"
                        },
                        {
                                "provider" : "GIA",
                                "providerId" : "250629"
                        }
                ],
                "propertyType" : "Unknown",
                "optPostalCode" : "92037",
                "optStateCode" : "CA",
                "optChainCode" : "WV"
        },
        "hotelDetail" : {
                "id" : 88173,
                "description" : "The Pantai Inn is a Balinesian inspired hotel perched atop La Jollas jagged cliffs. Boutique-lined streets gemmed with art gifts and other shopping destinations. Indulgent soft sand beaches with year-round vacation sunshine and the Pacific Oceans watercolor sunsets. From the village to the cove La Jolla truly is the jewel of Southern California. - 11 ROOMS - 2 LEVELS - INTERIOR CORRIDORS",
                "zomProviderOomAmenity" : [ ],
                "optTpCityCode" : "SAN",
                "optCheckInTime" : "1500"
        }
}

> var getRnd = function(q) {

... var count = q.count()

... var o = q.limit(-1).skip(Math.random() * q.count()).next()

... printjson(o)

... }

> use hotel

switched to db hotel

> getRnd(db.hotelWithDetail.find())

{

"_id" : ObjectId("538f5fa6d9805dc69ea218b5"),

"hotel" : {

"id" : 88173,

"name" : "Pantai Inn",

"addressProvider" : "GoogleMaps",

"addressLine1" : "1003 Coast Boulevard",

"city" : "San Diego",

"countryCode" : "US",

"phoneNumber" : "+1-858-224-7600",

"location" : {

"latitude" : 32.8485673,

"longitude" : -117.275847,

"geocodeProvider" : "GoogleMaps",

"geocodeAccuracy" : "Vicinity",

"created" : ISODate("2014-04-19T17:48:45.101Z")

"oomProviderMapping" : [

{

"provider" : "TP",

"providerId" : "71990"

{

"provider" : "GIA",

"providerId" : "250629"

}

"propertyType" : "Unknown",

"optPostalCode" : "92037",

"optStateCode" : "CA",

"optChainCode" : "WV"

"hotelDetail" : {

"id" : 88173,

"description" : "The Pantai Inn is a Balinesian inspired hotel perched atop La Jollas jagged cliffs. Boutique-lined streets gemmed with art gifts and other shopping destinations. Indulgent soft sand beaches with year-round vacation sunshine and the Pacific Oceans watercolor sunsets. From the village to the cove La Jolla truly is the jewel of Southern California. - 11 ROOMS - 2 LEVELS - INTERIOR CORRIDORS",

"zomProviderOomAmenity" : [ ],

"optTpCityCode" : "SAN",

"optCheckInTime" : "1500"

}

Scala: Collections of Futures

Leave a reply

Version

Scala 2.11.1

Overview

One of my favorite things about Scala is the amazing collections library. Scala’s collections library combines the best of standard functional idioms with an OOP call style. This makes for some down right beautiful code. However — after working with Scala for a bit over a year now, it has become very apparent to me that the Scala collection’s library was not written with asyncronous/reactive programming in mind.

Note

The Scala/Akka standard library Future is not lazy. Once a Future is constructed, it is “hot” — executing immediately. There are other libraries whose Future implementation is lazy (such as Scalaz²). Lazy futures allow building up an execution “plan” which is eventually run by the ultimate caller. This approach has many advantages, but sadly is not the direction Scala/Akka took. This article focuses exclusively on the Scala/Akka standard library Future¹.

Note

I’ve tried to make sure all of the code examples here can be pasted directly into the Scala console. Some boilerplate is required to make these work:

import scala.language.higherKinds
import scala.collection.generic._
import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits.global

import scala.language.higherKinds

import scala.collection.generic._

import scala.concurrent._

import scala.concurrent.duration._

import ExecutionContext.Implicits.global

Make sure to paste this into the console before trying the code below.

Note	Scala has wonderful type inference which ordinarinally makes it unnecessary to explicitly declare the type of vals. However, for maximum readability in my examples, I explicitly label the type of vals.

Calling a method that returns a Future N times

Once I decided to start using futures, they started to bleed into function return-type signatures everywhere. One place in particular that they started to show up was in the service layer:

Listing 1

trait MyService {
  def doSomething(i: Int) : Future[Unit]
}
class MyServiceImpl extends MyService {
 def doSomething(i: Int) : Future[Unit] = Future { Thread.sleep(500);println(i) }
}

trait MyService {

def doSomething(i: Int) : Future[Unit]

}

class MyServiceImpl extends MyService {

def doSomething(i: Int) : Future[Unit] = Future { Thread.sleep(500);println(i) }

}

Using a service layer like this was straightforward, until I needed to call it N times:

Listing 2

val svc : MyService = new MyServiceImpl
val someInts : List[Int] = (1 to 20).toList
val result : Unit = someInts.foreach(svc.doSomething _)

val svc : MyService = new MyServiceImpl

val someInts : List[Int] = (1 to 20).toList

val result : Unit = someInts.foreach(svc.doSomething _)

This was my first intuition about how to call the service N times. It seemed straightforward to me at the time and it even compiles and runs! But what this actually does is to immediately create a List[Future[Unit]] of 20 “hot” futures — all 20 futures have been submitted for execution! While this might be ok for 20 futures, it’s not ok for 1,000, 10,000 or 100,000+ futures.

Internally, the executor stores futures in a queue and executes as many futures as it has workers simulatenously. Dumping too many futures into the executor queue at once will starve other code that uses the same executor and can cause an out of memory error. Definitely not what I wanted.

Also, there is another problem here: each future returned by svc.doSomething is discarded by assignment to Unit. Not only am I not properly waiting on my futures to complete, but by assigning my Future to Unit, I’m throwing away any exception that might be thrown! Also not what I want.

Tip

Assigning a Future to Unit is always an error. Perhaps in the future the compiler will emit a warning about this, but for now the only way to discover this is by code review of programmers who may be new to futures. Even worse, this code may not fail at runtime. For actions returned in futures that complete quickly and are 99% exception free, this bug might go unnoticed for sometime.

Never assign Future to Unit

So what can I do? I need to stop assigning Future to Unit. I can do this by using map instead of foreach. Also, I need a way to properly wait for all my futures to complete:

Listing 3

val futResult : Future[List[Unit]] =
  Future.sequence {
    someInts.map(svc.doSomething _) // # <b><1></b>
  } // # <b><2></b>
val result : Unit = Await.result(futResult,Duration.Inf) // # <b><3></b>

val futResult : Future[List[Unit]] =

Future.sequence {

someInts.map(svc.doSomething _) // # <1>

} // # <2>

val result : Unit = Await.result(futResult,Duration.Inf) // # <3>

Use map instead of foreach to ensure I don’t discard any futures
Use Future.sequence to convert a List[Future[Unit]] to Future[List[Unit]]
Properly wait on all my futures to complete. Also, now that I’ve waited on all my futures, I can safely discard my List[Unit] because no exceptions were thrown

If any exceptions are thrown during svc.doSomething calls, the exception will percolate up through Await.result. But how can I stop dumping all futures into the executor at once?

Future.sequence and the functional “sequence” idiom

Most of this is standard Scala collections stuff, but what is Future.sequence? Future.sequence is a Future specific version of the “sequence” functional idiom (not to be confused with the Scala collection type Sequence). There is much more to know about “sequence”, but it is basically a way to invert the nesting of two monads, i.e. G[F[_]] to F[G[_]] (given certain properties of types G and F). Unfortunately, standard Scala requires some compiler magic³ to generically implement sequence. Without resorting to this magic, I have to copy the sequence pattern for different monad types, such as has been done for Future.sequence.

In addition to making Future the outer monad, Future.sequence also takes care of waiting for all of our futures to complete. In listing 3, futResult is of type Future[List[Unit]], which is completed once all of my inner futures complete

Controlling the flow of Future execution

I’ve solved the problem of discarding futures, but I still need to somehow regulate the flow of how many futures go “hot” simulatenously.

Listing 4

val result : Unit =
    someInts
      .grouped(3) // # <b><1></b>
      .toList
      .map { group =>
        val innerFutResult : Future[List[Unit]] = // # <b><2></b>
          Future.sequence {
            group.map(svc.doSomething _)
          }
        Await.result(innerFutResult, Duration.Inf) // # <b><3></b>
      }
      .flatten // # <b><4></b>

val result : Unit =

someInts

.grouped(3) // # <1>

.toList

.map { group =>

val innerFutResult : Future[List[Unit]] = // # <2>

Future.sequence {

group.map(svc.doSomething _)

}

Await.result(innerFutResult, Duration.Inf) // # <3>

}

.flatten // # <4>

Group someInts into a group size that I want to execute simulatenously
For each group create a Future[List[Unit]]
Use Await.result inside the map to wait for each group to complete
Because I divided someInts into groups, I need to flatten the results (Note: this isn’t strictly necessary since result is Unit in this example. I’m going to discard List[Unit] anyway, but if result wasn’t Unit it would be necessary to flatten)

Ok this works. I’ve ensured that no more than N svc.doSomething calls are happening at once and exceptions are never discarded. However, this pattern has a fatal flaw. It does not pass a future back as a result. For the purposes of writing example code, this kind of thing gets the job done. However, when writing code that will live in an asyncronous eco-system, I must make my result a Future.

Tip	When writing a method that calls other functions or methods that return Future, I need to make sure to return a Future to callers of my method. This allows callers to use the Future of my method’s return value in the same way that I did when I called other methods that returned me a Future.

Returning a Future to callers

This has gotten complicated fast! But I feel like I’m almost there, so I will keep going. I’m modifying Listing 3 to ensure my result is a Future:

Pattern 1.0

val futResult : Future[List[Unit]] =
    someInts
      .grouped(3)
      .toList
      .foldLeft(Future.successful(List[Unit]())) { (futAccumulator,group) => // # <b><1></b>
        futAccumulator.flatMap { accumulator => // # <b><2></b>
          val futInnerResult : Future[List[Unit]] =
            Future.sequence {
              group.map(svc.doSomething _)
            }
          futInnerResult.map(innerResult => accumulator ::: innerResult) // # <b><3></b>
        }
      }
val result : Unit = Await.result(futResult,Duration.Inf)

val futResult : Future[List[Unit]] =

someInts

.grouped(3)

.toList

.foldLeft(Future.successful(List[Unit]())) { (futAccumulator,group) => // # <1>

futAccumulator.flatMap { accumulator => // # <2>

val futInnerResult : Future[List[Unit]] =

Future.sequence {

group.map(svc.doSomething _)

}

futInnerResult.map(innerResult => accumulator ::: innerResult) // # <3>

}

val result : Unit = Await.result(futResult,Duration.Inf)

I’ve replaced map with foldLeft. This will ensure that each group is processed one at a time, from left to right and will accumulate the Future[List[Unit]] result after each group completes. The accumulator is initialized with already completed Future of an empty List[Unit].
Future.flatMap is used here instead of Future.map to flatten the inner return type of Future[List[Unit]] over the entire collection (If Future.map had been used, it would return Future[Future[List[Unit]]]).
After a group completes, the result accumulates

Ok this is much better. I’m ensuring that I don’t discard exceptions, I control the flow of futures AND now I return a Future to callers. But I call a Future returning method N times in many places. This is a pretty tedious pattern to have to repeat everywhere. Scala gives me some amazing utilities for cleaning up complexity like this.

Pimp My Future: Pattern 1.1

I’m going to cleanup Pattern 1.0 using a for-comprehension⁷ and the Pimp-My-Library Pattern⁵ with the Scala Value Class⁶. The pimp-my-library pattern allows creating an implicit wrapper class that can “add” a method to an existing class, essentially making an OOP style call convention for the new method. The Value Class (added in Scala 2.10) makes the implicit wrapper class free — the compiler optimizes away the wrapper in emitted bytecode.

Listing 4

  implicit class Future_PimpMyFuture[T](val self: Future[T]) extends AnyVal {
    def get : T = Await.result(self, Duration.Inf)
  }
  implicit class Future_PimpMyTraversableOnceOfFutures[A, M[AA] <: TraversableOnce[AA]](val self: M[Future[A]]) extends AnyVal {
    /** @return a Future of M[A] completes once all futures have completed */
    def sequence(implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], ec: ExecutionContext) : Future[M[A]] =
      Future.sequence(self)
  }

implicit class Future_PimpMyFuture[T](val self: Future[T]) extends AnyVal {

def get : T = Await.result(self, Duration.Inf)

}

implicit class Future_PimpMyTraversableOnceOfFutures[A, M[AA] <: TraversableOnce[AA]](val self: M[Future[A]]) extends AnyVal {

/** @return a Future of M[A] completes once all futures have completed */

def sequence(implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], ec: ExecutionContext) : Future[M[A]] =

Future.sequence(self)

}

Pattern 1.1

val futResult : Future[List[Unit]] =
    someInts
      .grouped(3)
      .toList
      .foldLeft(Future.successful(List[Unit]())) { (futAccumulator,group) =>
        for { // # <b><1></b>
          accumulator <- futAccumulator
          innerResult <- group.map(svc.doSomething _).sequence // # <b><2></b>
        } yield accumulator ::: innerResult
      }

val result : Unit = futResult.get // # <b><3></b>

val futResult : Future[List[Unit]] =

someInts

.grouped(3)

.toList

.foldLeft(Future.successful(List[Unit]())) { (futAccumulator,group) =>

for { // # <1>

accumulator <- futAccumulator

innerResult <- group.map(svc.doSomething _).sequence // # <2>

} yield accumulator ::: innerResult

}

val result : Unit = futResult.get // # <3>

Replaced Future.flatMap and nested Future.map with a much cleaner more readable for-comprehension⁷
Replaced Future.sequence with sugar method
Replace Await.result with sugar method

I like the OOP style call convention, but this pattern is still tedious. Perhaps, I can make this even simpler?

Pimp My Future: Pattern 1.2

I’m going to further cleanup Pattern 1.1 by creating another pimp-my-library method on a new value class Future_PimpMyTraversableOnce.

Listing 5

  implicit class Future_PimpMyTraversableOnce[A, M[AA] <: TraversableOnce[AA]](val self: M[A]) extends AnyVal {
    /** @return a Future of M[B] that completes once all futures have completed */
    def mapAsync[B](groupSize: Int)(f: A => Future[B])(implicit
      cbf: CanBuildFrom[M[Future[A]], A, M[A]],
      cbf2: CanBuildFrom[Nothing, B, M[B]],
      ec: ExecutionContext) : Future[M[B]] = {
      self
       .toList // # <b><1></b>
       .grouped(groupSize)
       .foldLeft(Future.successful(List[B]())) { (futAccumulator,group) =>
         for {
           accumulator <- futAccumulator
           innerResult <- group.map(f).sequence
         } yield accumulator ::: innerResult
       }
       .map(_.to[M]) // # <b><2></b>
    }
  }

implicit class Future_PimpMyTraversableOnce[A, M[AA] <: TraversableOnce[AA]](val self: M[A]) extends AnyVal {

/** @return a Future of M[B] that completes once all futures have completed */

def mapAsync[B](groupSize: Int)(f: A => Future[B])(implicit

cbf: CanBuildFrom[M[Future[A]], A, M[A]],

cbf2: CanBuildFrom[Nothing, B, M[B]],

ec: ExecutionContext) : Future[M[B]] = {

self

.toList // # <1>

.grouped(groupSize)

.foldLeft(Future.successful(List[B]())) { (futAccumulator,group) =>

for {

accumulator <- futAccumulator

innerResult <- group.map(f).sequence

} yield accumulator ::: innerResult

}

.map(_.to[M]) // # <2>

}

Convert to List here for efficient accumulation of results (and no grouped method on TraversableOnce)
Convert back to desired collection

Pattern 1.2

val futResult : Future[List[Unit]] = someInts.mapAsync(3)(svc.doSomething _)
val result : Unit = futResult.get

1 2	val futResult : Future[List[Unit]] = someInts.mapAsync(3)(svc.doSomething _) val result : Unit = futResult.get

Much better! My code is now simple, readable, idiomatic, doesn’t discard exceptions, doesn’t flood the executor with futures and returns a future to the caller!

Further Exploration: Problems with Pattern 1

Pattern 1 solves the problem of discarding exceptions, regulating the flow of “hot” futures and returning a future to callers, but it isn’t the most efficient way of handling this problem. Because it has to wait for each group to complete, one of the svc.doSomething calls could take an extra long time. If it does, even though the other futures in its group have completed, I have to wait for that one long call to complete before moving on to the next group. Ideally, I should make it so that there are always N futures running simultaneously instead of grouping them. Work for another day!

Sources

CentOS 6: Install RPM Forge Repo

Leave a reply

Versions

CentOS 6.5 x86_64

Install

Note

replace $VERSION with latest release. As of this writing it is 0.5.3-1

wget http://pkgs.repoforge.org/rpmforge-release/rpmforge-release-$VERSION.el6.rf.x86_64.rpm
sudo yum localinstall rpmforge-release-$VERSION.el6.rf.x86_64.rpm

1 2	wget http://pkgs.repoforge.org/rpmforge-release/rpmforge-release-$VERSION.el6.rf.x86_64.rpm sudo yum localinstall rpmforge-release-$VERSION.el6.rf.x86_64.rpm

wget http://pkgs.repoforge.org/rpmforge-release/rpmforge-release-0.5.3-1.el6.rf.x86_64.rpm
sudo yum localinstall rpmforge-release-0.5.3-1.el6.rf.x86_64.rpm

1 2	wget http://pkgs.repoforge.org/rpmforge-release/rpmforge-release-0.5.3-1.el6.rf.x86_64.rpm sudo yum localinstall rpmforge-release-0.5.3-1.el6.rf.x86_64.rpm

To discover latest major/minor, visit: http://http://pkgs.repoforge.org/rpmforge-release/

Verify

$ head /etc/yum.repos.d/epel.repo
### Name: RPMforge RPM Repository for RHEL 6 - dag
### URL: http://rpmforge.net/
[rpmforge]

$ head /etc/yum.repos.d/epel.repo

### Name: RPMforge RPM Repository for RHEL 6 - dag

### URL: http://rpmforge.net/

[rpmforge]

Scala: Use Option instead of null to represent not set

Leave a reply

In Java, null is commonly used to represent an Object value that is not-set:

class Foo {
  private String bar = null; // null indicates not-set; not-set by default

  // may return null to indicate that bar is not-set!
  public String getBar() { return bar; }

  // _bar may be null to set bar to not-set
  public void setBar(String _bar) { bar = _bar; }
}

class Foo {

private String bar = null; // null indicates not-set; not-set by default

// may return null to indicate that bar is not-set!

public String getBar() { return bar; }

// _bar may be null to set bar to not-set

public void setBar(String _bar) { bar = _bar; }

}

While Java developers are forced to depend on method documentation to understand the semantics of null, Scala developers can instead code the semantics directly into the type system using Option:

class Foo(val bar : Option[String] = None)

1	class Foo(val bar : Option[String] = None)

This Scala snippet not only eliminates the need to explicitly declare the getter/setter but also removes the need to document nullability. From this declaration, it can be safely assumed that bar is never null and that bar may always be safely dereferenced. By using Option, users of the class are forced to deal with the possibility of bar being not set. In Java, documentation can easily be overlooked. The Java compiler will have no complaints about the following:

...
Foo foo = new Foo();
...
System.out.print(foo.getBar()); // forgot about null!

...

Foo foo = new Foo();

...

System.out.print(foo.getBar()); // forgot about null!

This snippet is problematic not just because it might dereference null, but because a NullPointerException doesn’t give the user, administrator or even later developers any indication of why the error has occurred. Only a later inspection of the source code and documentation of the getBar method would reveal that the null(not-set) value was not taken into account. If this code was buried in an edge case that is rarely or never exercised, the bug might not surface for years. Scala Option forces users of the Foo class to deal with the not-set state before accessing the value:

...
val foo = new Foo();
...
foo.bar match {
  case Some(value) => println(value)
  case None => println("not set!")
}

...

val foo = new Foo();

...

foo.bar match {

case Some(value) => println(value)

case None => println("not set!")

}

JVM: Memory Settings

Leave a reply

Versions

Java Oracle JDK 1.6.X

Heap

On program startup, Java allocates a certain amount of memory for the heap. This memory is used to fulfill requests for memory allocation by the program or libraries it calls. The Java heap space also has an upper limit that if reached, results in the java.lang.OutOfMemoryError[3] being thrown. Generally once running, there is no action a program can take to resolve this error. Instead, either set the maximum heap space higher when running the program or use a Java memory profiler to analyze memory usage (there might be a memory leak). It is also possible to redesign the algorithm to use the disk as a temporary swap space, but this should be avoided unless absolutely necessary.

Java allows setting the minimum and maximum heap settings as follows:[1]

-Xms$MIN_VALUE
-Xmx$MAX_VALUE

Values are specified as multiples of 1024 bytes and must be greater than 1 MB. Append k for kilobytes, m for megabytes or g for gigabytes.

Examples:

-Xms16k
-Xmx32m
-Xmx2g

If these values are not specified at program startup, Java will automatically determine the values to scale with the available system memory.[2]

PermGen

Java uses a separate pre-allocated memory region, called the PermGen, to store class files, string constants and other resources located in jar files. With the exception of very large programs, the default value for PermGen is sufficient. Issues with PermGen are more common when running an application server such as GlassFish or Tomcat that loads jar files on-demand. To fix issues with PermGen, simply raise the default value for the PermGen space. Note that it is possible to create a classloader leak[5] with on-demand jar file loading.

-XX:MaxPermSize

Examples:

-XX:MaxPermSize=128m

Application Server Tuning

It is recommended to run application servers such as GlassFish or Tomcat with the following two settings to optimize garbage collection of unused jar resources in PermGen when an application is undeployed.[7]

-XX:+CMSClassUnloadingEnabled
-XX:+CMSPermGenSweepingEnabled

Sources

Scala: Use Option or Either to replace null as a failure indicator

Leave a reply

While certainly not a recommended practice in Java, null is occasionally used to represent the failure of a simple operation:

class Foo {
  // may return null to indicate failure!
  public Integer readInt() { ... }
}

class Foo {

// may return null to indicate failure!

public Integer readInt() { ... }

}

Java developers must depend on documentation to explain if null is a possible return value, and if it is, what the meaning of null is. In this case, null is used to indicate the operation of readInt failed. Scala developers have a couple of options here:

class Foo {
  // None indicates failure!
  def readInt() : Option[Int] = { ... }
}

class Foo {

// None indicates failure!

def readInt() : Option[Int] = { ... }

}

In this Scala snippet, it is explicit that the function may not return a value. However, it is still not clear that None indicates failure and this should still be documented. Generally, I like to have as much information as possible about failures. This final Scala snippet uses Either to remove any ambiguity:

class Foo {
  def readInt() : Either[Exception, Int] = { ... }
}

class Foo {

def readInt() : Either[Exception, Int] = { ... }

}

In this snippet, no documentation is required since it is explicit that this function either returns an Exception or a value. Callers of this function are forced to deal with the possible failure of the operation before accessing the return value. If the operation did fail, callers are provided with a possibly informative Exception to explain the failure.