Cleaning up tasks and caching

This commit is contained in:
Mark Harrah 2009-12-11 18:56:09 -05:00
parent 14db8c2079
commit a301df2ff1
13 changed files with 180 additions and 132 deletions

29
cache/Cache.scala vendored
View File

@ -23,18 +23,23 @@ object Cache extends BasicCacheImplicits with SBinaryFormats with HListCacheImpl
def wrapOutputCache[O,DO](implicit convert: O => DO, reverse: DO => O, base: OutputCache[DO]): OutputCache[O] =
new WrappedOutputCache[O,DO](convert, reverse, base)
/* Note: Task[O] { type Input = I } is written out because ITask[I,O] did not work (type could not be inferred properly) with a task
* with an HList input.*/
def apply[I,O](task: Task[O] { type Input = I }, file: File)(implicit cache: Cache[I,O]): Task[O] { type Input = I } =
task match { case m: M[I,O,_] =>
new M[I,O,Result[O]](None)(m.dependencies)(m.extract)(computeWithCache(m, cache, file))
}
private def computeWithCache[I,O](m: M[I,O,_], cache: Cache[I,O], file: File)(in: I): Result[O] =
cache(file)(in) match
{
case Left(value) => Value(value)
case Right(store) => m.map { out => store(out); out }
}
def apply[I,O](file: File)(f: I => Task[O])(implicit cache: Cache[I,O]): I => Task[O] =
in =>
cache(file)(in) match
{
case Left(value) => Task(value)
case Right(store) => f(in) map { out => store(out); out }
}
def cached[I,O](file: File)(f: I => O)(implicit cache: Cache[I,O]): I => O =
in =>
cache(file)(in) match
{
case Left(value) => value
case Right(store) =>
val out = f(in)
store(out)
out
}
}
trait BasicCacheImplicits extends NotNull
{

View File

@ -1,18 +1,18 @@
package xsbt
import java.io.{InputStream,OutputStream}
import metascala.HLists.{HCons,HList,HNil}
import HLists._
class HNilInputCache extends NoInputCache[HNil]
class HConsInputCache[H,T <: HList](val headCache: InputCache[H], val tailCache: InputCache[T]) extends InputCache[HCons[H,T]]
{
def uptodate(in: HCons[H,T])(cacheStream: InputStream) =
{
lazy val headResult = headCache.uptodate(in.head)(cacheStream)
lazy val tailResult = tailCache.uptodate(in.tail)(cacheStream)
val headResult = headCache.uptodate(in.head)(cacheStream)
val tailResult = tailCache.uptodate(in.tail)(cacheStream)
new CacheResult
{
lazy val uptodate = headResult.uptodate && tailResult.uptodate
val uptodate = headResult.uptodate && tailResult.uptodate
def update(outputStream: OutputStream) =
{
headResult.update(outputStream)

View File

@ -4,19 +4,25 @@ import java.io.File
object CacheTest// extends Properties("Cache test")
{
val lengthCache = new File("/tmp/length-cache")
val cCache = new File("/tmp/c-cache")
import Task._
import Cache._
import FileInfo.hash._
def checkFormattable(file: File)
def test
{
val createTask = Task { new File("test") }
val lengthTask = createTask map { f => println("File length: " + f.length); f.length }
val cached = Cache(lengthTask, new File("/tmp/length-cache"))
val cTask = (createTask :: cached :: TNil) map { case (file :: len :: HNil) => println("File: " + file + " length: " + len); len :: file :: HNil }
val cachedC = Cache(cTask, new File("/tmp/c-cache"))
val length = (f: File) => { println("File length: " + f.length); f.length }
val cachedLength = cached(lengthCache) ( length )
try { TaskRunner(cachedC) }
val lengthTask = createTask map cachedLength
val c = (file: File, len: Long) => { println("File: " + file + ", length: " + len); len :: file :: HNil }
val cTask = (createTask :: lengthTask :: TNil) map cached(cCache) { case (file :: len :: HNil) => c(file, len) }
try { TaskRunner(cTask) }
catch { case TasksFailed(failures) => failures.foreach(_.exception.printStackTrace) }
}
}

View File

@ -24,7 +24,7 @@ class Changed[O](val task: Task[O], val cacheFile: File)(implicit input: InputCa
{
val clean = Clean(cacheFile)
def clear = Task.empty
def apply[O2](ifChanged: O => O2, ifUnchanged: O => O2): Task[O2] { type Input = O } =
def apply[O2](ifChanged: O => O2, ifUnchanged: O => O2): Task[O2] =
task map { value =>
val cache =
try { OpenResource.fileInputStream(cacheFile)(input.uptodate(value)) }

View File

@ -0,0 +1,27 @@
Copyright (c) 2009, Jesper Nordenberg
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
* Neither the name of the author nor the
names of its contributors may be used to endorse or promote products
derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY JESPER NORDENBERG ''AS IS'' AND ANY
EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL JESPER NORDENBERG BE LIABLE FOR ANY
DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

37
tasks/TList.scala Normal file
View File

@ -0,0 +1,37 @@
package xsbt
import Task.{bindTask, mapTask}
sealed trait TList
{
type Head
type Tail <: TList
type HListType <: HList
private[xsbt] def tasks: List[Task[_]]
private[xsbt] def get(results: Results): HListType
}
sealed class TNil extends TList
{
type Head = Nothing
type Tail = TNil
type HListType = HNil
def ::[A](t: Task[A]) = TCons[A,HNil,TNil](t, this)
private[xsbt] def tasks = Nil
private[xsbt] def get(results: Results) = HNil
}
final case class TCons[H, HL <: HList, T <: TList { type HListType = HL}](head: Task[H], tail: T) extends TList
{
type Head = H
type Tail = T
type HListType = HCons[H,HL]
type This = TCons[H, HL, T]
def ::[A](t: Task[A]) = TCons[A,HListType,This](t, this)
private[xsbt] def tasks = head :: tail.tasks
private[xsbt] def get(results: Results) = HCons(results(head), tail.get(results))
private def getF = get _
def map[X](f: HListType => X): Task[X] = mapTask(tasks: _*)(f compose getF)
def bind[X](f: HListType => Result[X]): Task[X] = bindTask(tasks: _*)(f compose getF)
def join: Task[HListType] = map(identity[HListType])
}
object TNil extends TNil

View File

@ -1,36 +1,35 @@
package xsbt
import Task.{mapTask,bindTask, ITask}
import Task.{mapTask,bindTask}
import scala.collection.{mutable,immutable}
sealed trait Result[O] extends NotNull
final case class Value[O](t: O) extends Result[O]
sealed abstract class Task[O] extends Identity with Result[O]
{
type Input
def dependencies: TreeHashSet[Task[_]] // IMPORTANT!! immutable.HashSet is NOT suitable. It has issues with multi-threaded access
def map[N](f: O => N): ITask[O,N]
def bind[N](f: O => Result[N]): ITask[O,N]
def dependsOn(addDependencies: Task[_]*): ITask[Input,O]
def named(s: String): ITask[Input,O]
def map[N](f: O => N): Task[N]
def bind[N](f: O => Result[N]): Task[N]
def dependsOn(addDependencies: Task[_]*): Task[O]
def named(s: String): Task[O]
}
private final class M[I,O,R <: Result[O]](name: Option[String])
(val dependencies: TreeHashSet[Task[_]])(val extract: Results => I)(val compute: I => R) extends Task[O]
private final class M[O,R <: Result[O]](name: Option[String])(val dependencies: TreeHashSet[Task[_]])(val compute: Results => R) extends Task[O]
{
type Input = I
def this(dependencies: Task[_]*)(extract: Results => I)(compute: I => R) =
this(None)(TreeHashSet(dependencies: _*))(extract)(compute)
def this(dependencies: Task[_]*)(compute: Results => R) =
this(None)(TreeHashSet(dependencies: _*))(compute)
final def dependsOn(addDependencies: Task[_]*) = new M(name)(dependencies ++ addDependencies)(extract)(compute)
final def map[N](f: O => N) = mapTask(this)(_(this))(f)
final def bind[N](f: O => Result[N]) = bindTask(this)(_(this))(f)
final def dependsOn(addDependencies: Task[_]*) = new M(name)(dependencies ++ addDependencies)(compute)
final def map[N](f: O => N) = mapTask(this)(f compose get)
final def bind[N](f: O => Result[N]) = bindTask(this)(f compose get)
final def named(s: String) =
name match
{
case Some(n) => error("Cannot rename task already named '" + n + "'. (Tried to rename it to '" + s + "')")
case None => new M(Some(s))(dependencies)(extract)(compute)
case None => new M(Some(s))(dependencies)(compute)
}
final override def toString = "Task " + name.getOrElse("<anon>")
final override def toString = "Task " + name.getOrElse("<anon$" + hashCode.toHexString + ">")
private def get: (Results => O) = _(this)
}
abstract class Identity extends NotNull
{
@ -49,22 +48,19 @@ object Task
{
val empty = Task(())
type ITask[I,O] = Task[O] { type Input = I }
import Function.tupled
def apply[O](o: => O): ITask[Unit,O] =
new M[Unit,O,Value[O]]()(r => ())( u => Value(o) )
def bindTask[I,O](dependencies: Task[_]*)(extract: Results => I)(compute: I => Result[O]): ITask[I,O] =
new M[I,O,Result[O]](dependencies : _*)(extract)(compute)
def mapTask[I,O](dependencies: Task[_]*)(extract: Results => I)(compute: I => O): ITask[I,O] =
new M[I,O,Value[O]](dependencies : _*)(extract)(in => Value(compute(in)))
def apply[O](o: => O): Task[O] = mapTask()( _ => o )
def mapTask[O](dependencies: Task[_]*)(compute: Results => O): Task[O] =
bindTask(dependencies : _*)(in => Value(compute(in)))
def bindTask[O](dependencies: Task[_]*)(compute: Results => Result[O]): Task[O] =
new M[O,Result[O]](dependencies : _*)(compute)
private[xsbt] def extract[I,O](t: ITask[I,O], results: Results): I = t match { case m: M[I,O,_] => m.extract(results) }
private[xsbt] def compute[I,O](t: ITask[I,O], input: I): Result[O] = t match { case m: M[I,O,_] => m.compute(input) }
private[xsbt] def compute[O](t: Task[O], results: Results): Result[O] = t match { case m: M[O,_] => m.compute(results) }
implicit def iterableToForkBuilder[A](t: Iterable[A]): ForkBuilderIterable[A] = new ForkBuilderIterable(t)
final class ForkBuilderIterable[A] private[Task](a: Iterable[A]) extends NotNull
{
def fork[X](f: A => X): Iterable[ITask[Unit,X]] = a.map(x => Task(f(x)))
def fork[X](f: A => X): Iterable[Task[X]] = forkTasks(x => Task(f(x)) )
def forkTasks[X](f: A => Task[X]): Iterable[Task[X]] = a.map(x => f(x))
def reduce(f: (A,A) => A): Task[A] = fork(x => x) reduce(f)
}
@ -72,11 +68,11 @@ object Task
implicit def iterableToBuilder[O](t: Iterable[Task[O]]): BuilderIterable[O] = new BuilderIterable(t)
final class BuilderIterable[O] private[Task](a: Iterable[Task[O]]) extends NotNull
{
//def mapBind[X](f: O => Task[_,X]): Iterable[Task[O,XO]] = a.map(_.bind(f))
//def mapBind[X](f: O => Task[X]): Iterable[Task[XO]] = a.map(_.bind(f))
def join: Task[Iterable[O]] = join(identity[O])
def joinIgnore: Task[Unit] = join.map(i => ())
def join[X](f: O => X): Task[Iterable[X]] = mapTask(a.toSeq: _*)( r => a.map(t => r(t)) )(_.map(f))
//def bindJoin[X](f: O => Task[_,X]): Task[Iterable[X],Iterable[X]] = mapBind(f).join
def joinIgnore: Task[Unit] = join.map(_ => ())
def join[X](f: O => X): Task[Iterable[X]] = mapTask(a.toSeq: _*)( r => a map (f compose r.apply[O]) )
//def bindJoin[X](f: O => Task[X]): Task[Iterable[X]] = mapBind(f).join
def reduce(f: (O,O) => O): Task[O] =
{
def reduce2(list: List[Task[O]], accumulate: List[Task[O]]): List[Task[O]] =
@ -97,59 +93,31 @@ object Task
}
}
import metascala.HLists.{HList,HNil,HCons}
sealed trait TList
{
type Head
type Tail <: TList
type HListType <: HList
def tasks: List[Task[_]]
def get(results: Results): HListType
}
final class TNil extends TList
{
type Head = Nothing
type Tail = TNil
type HListType = HNil
def ::[A](t: Task[A]) = TCons[A,HNil,TNil](t, this)
def tasks = Nil
def get(results: Results) = HNil
}
final case class TCons[H, HL <: HList, T <: TList { type HListType = HL}](head: Task[H], tail: T) extends TList
{
type Head = H
type Tail = T
type This = TCons[H,HL,T]
type HListType = HCons[H,HL]
def ::[A](t: Task[A]) = TCons[A,HListType,This](t, this)
def tasks = head :: tail.tasks
def get(results: Results) = HCons(results(head), tail.get(results))
def map[X](f: HListType => X): ITask[HListType,X] = mapTask(tasks: _*)(get)(f)
def bind[X](f: HListType => Result[X]): ITask[HListType,X] = bindTask(tasks: _*)(get)(f)
def join: ITask[HListType,HListType] = map(identity[HListType])
}
val TNil = new TNil
implicit def twoToBuilder[A,B](t: (Task[A], Task[B]) ): Builder2[A,B] =
t match { case (a,b) => new Builder2(a,b) }
final class Builder2[A,B] private[Task](a: Task[A], b: Task[B]) extends NotNull
{
def map[X](f: (A,B) => X): ITask[(A,B),X] = mapTask(a,b)(r => (r(a), r(b)))(tupled(f))
def bind[X](f: (A,B) => Result[X]): ITask[(A,B),X] = bindTask(a,b)( r => (r(a), r(b)) )(tupled(f))
private def extract = (r: Results) => (r(a), r(b))
private def compute[T](f: (A,B) => T) = tupled(f) compose extract
def map[X](f: (A,B) => X): Task[X] = mapTask(a,b)(compute(f))
def bind[X](f: (A,B) => Result[X]): Task[X] = bindTask(a,b)(compute(f))
}
implicit def threeToBuilder[A,B,C](t: (Task[A], Task[B], Task[C])): Builder3[A,B,C] = t match { case (a,b,c) => new Builder3(a,b,c) }
final class Builder3[A,B,C] private[Task](a: Task[A], b: Task[B], c: Task[C]) extends NotNull
{
def map[X](f: (A,B,C) => X): ITask[(A,B,C),X] = mapTask(a,b,c)( r => (r(a), r(b), r(c)) )(tupled(f))
def bind[X](f: (A,B,C) => Result[X]): ITask[(A,B,C),X] = bindTask(a,b,c)( r => (r(a), r(b), r(c)) )(tupled(f))
private def extract = (r: Results) => (r(a), r(b), r(c))
private def compute[T](f: (A,B,C) => T) = tupled(f) compose extract
def map[X](f: (A,B,C) => X): Task[X] = mapTask(a,b,c)(compute(f))
def bind[X](f: (A,B,C) => Result[X]): Task[X] = bindTask(a,b,c)(compute(f))
}
implicit def fourToBuilder[A,B,C,D](t: (Task[A], Task[B], Task[C], Task[D])): Builder4[A,B,C,D] = t match { case (a,b,c,d) => new Builder4(a,b,c,d) }
final class Builder4[A,B,C,D] private[Task](a: Task[A], b: Task[B], c: Task[C], d: Task[D]) extends NotNull
{
def map[X](f: (A,B,C,D) => X): ITask[(A,B,C,D),X] = mapTask(a,b,c,d)( r => (r(a), r(b), r(c), r(d)) )(tupled(f))
def bind[X](f: (A,B,C,D) => Result[X]): ITask[(A,B,C,D),X] = bindTask(a,b,c,d)( r => (r(a), r(b), r(c),r(d)) )(tupled(f))
private def extract = (r: Results) => (r(a), r(b), r(c), r(d))
private def compute[T](f: (A,B,C,D) => T) = tupled(f) compose extract
def map[X](f: (A,B,C,D) => X): Task[X] = mapTask(a,b,c,d)( compute(f) )
def bind[X](f: (A,B,C,D) => Result[X]): Task[X] = bindTask(a,b,c,d)( compute(f) )
}
}

View File

@ -8,10 +8,10 @@ object TaskRunner
def apply[T](node: Task[T], maximumTasks: Int): T =
{
require(maximumTasks > 0)
val compute = new Compute[Work.Job, Result] { def apply[A](w: Work.Job[A]) = w.apply }
val strategy = new SimpleStrategy[Work[_,_]]
val compute = new Compute[Work, Result] { def apply[A](w: Work[A]) = w.apply }
val strategy = new SimpleStrategy[Work[_]]
val scheduler = new TaskScheduler(node, strategy, new BasicTaskListener)
val distributor = new Distributor[ Either[ List[WorkFailure[Task[_]]], T ] , Work.Job, Result](scheduler, compute, maximumTasks)
val distributor = new Distributor[ Either[ List[WorkFailure[Task[_]]], T ] , Work, Result](scheduler, compute, maximumTasks)
distributor.run().fold(failures => throw new TasksFailed(failures), identity[T])
}
}

View File

@ -1,14 +1,13 @@
package xsbt
import scala.collection.{immutable,mutable}
import Task.ITask
final case class WorkFailure[D](work: D, exception: Throwable) extends NotNull
{
def map[C](f: D => C) = WorkFailure(f(work), exception)
}
private final class TaskScheduler[O](root: Task[O], strategy: ScheduleStrategy[Work[_,_]], newListener: => TaskListener)
extends Scheduler[ Either[ List[WorkFailure[Task[_]]], O ], Work.Job, Result]
private final class TaskScheduler[O](root: Task[O], strategy: ScheduleStrategy[Work[_]], newListener: => TaskListener)
extends Scheduler[ Either[ List[WorkFailure[Task[_]]], O ], Work, Result]
{
def run = new Run
{
@ -32,7 +31,7 @@ private final class TaskScheduler[O](root: Task[O], strategy: ScheduleStrategy[W
}
def isComplete = reverseDeps.isEmpty
def hasPending = strategyRun.hasReady || !forwardDeps.isEmpty
def complete[A](work: Work.Job[A], result: Either[Throwable,Result[A]]): Unit =
def complete[A](work: Work[A], result: Either[Throwable,Result[A]]): Unit =
{
val task = work.source
result match
@ -66,19 +65,14 @@ private final class TaskScheduler[O](root: Task[O], strategy: ScheduleStrategy[W
private def addReady[O](m: Task[O])
{
def add[I](m: ITask[I,O])
{
val input = Task.extract(m, completed)
strategyRun.workReady(new Work(m, input))
listener.runnable(m)
}
assert(!forwardDeps.contains(m), m)
assert(reverseDeps.contains(m), m)
assert(!completed.contains(m), m)
assert(!calls.isCalling(m), m)
assert(m.dependencies.forall(completed.contains), "Could not find result for dependency of ready task " + m)
add(m: ITask[_,O])
strategyRun.workReady(new Work(m, completed))
listener.runnable(m)
}
// context called node
private def addGraph(node: Task[_], context: Task[_]): Boolean =
@ -230,19 +224,16 @@ private final class CalledByMap extends NotNull
caller.asInstanceOf[Task[O]]
}
}
private final class ResultMap(private val map: mutable.HashMap[Task[_], Any]) extends Results
import java.util.concurrent.{ConcurrentHashMap => HashMap}
private final class ResultMap(private val map: HashMap[Task[_], Any]) extends Results
{
def this() = this(new mutable.HashMap)
def update[O](task: Task[O], value: O) { map(task) = value }
def apply[O](task: Task[O]): O = map(task).asInstanceOf[O]
def contains(task: Task[_]) = map.contains(task)
def this() = this(new HashMap)
def update[O](task: Task[O], value: O) { map.put(task, value) }
def apply[O](task: Task[O]): O = map.get(task).asInstanceOf[O]
def contains(task: Task[_]) = map.containsKey(task)
}
private final class Work[I,O](val source: ITask[I,O], input: I) extends Identity with NotNull
private final class Work[O](val source: Task[O], results: Results) extends Identity with NotNull
{
final def apply = Task.compute(source, input)
}
private object Work
{
type Job[A] = Work[_,A]
final def apply = Task.compute(source, results)
}

View File

@ -1,5 +1,5 @@
//import metascala.HLists._
import xsbt.{HLists,Task}
package xsbt
import HLists._
import Task._

View File

@ -12,7 +12,7 @@ object TaskRunnerSortTest extends Properties("TaskRunnerSort")
java.util.Arrays.sort(sorted)
("Workers: " + workers) |: ("Array: " + a.toList) |:
{
def result = TaskRunner( sort(a.toArray), workers)
def result = TaskRunner( sort(a.toArray), if(workers > 0) workers else 1)
checkResult(result.toList, sorted.toList)
}
}
@ -29,8 +29,8 @@ object TaskRunnerSortTest extends Properties("TaskRunnerSort")
}
final def sort(a: RandomAccessSeq[Int]): Task[RandomAccessSeq[Int]] =
{
if(a.length < 2)
Task(a)
if(a.length < 200)
Task(sortDirect(a))
else
{
Task(a) bind { a =>

View File

@ -1,15 +1,29 @@
package xsbt
import metascala.HLists.{HCons => metaHCons, HList => metaHList, HNil => metaHNil}
// stripped down version of http://trac.assembla.com/metascala/browser/src/metascala/HLists.scala
// Copyright (c) 2009, Jesper Nordenberg
// new BSD license, see licenses/MetaScala
object HLists extends HLists
// add an extractor to metascala.HLists and define aliases to the HList classes in the xsbt namespace
trait HLists extends NotNull
trait HLists
{
object :: { def unapply[H,T<:HList](list: HCons[H,T]) = Some((list.head,list.tail)) }
final val HNil = metaHNil
final type ::[H, T <: HList] = metaHCons[H, T]
final type HNil = metaHNil
final type HList = metaHList
final type HCons[H, T <: HList] = metaHCons[H, T]
type ::[H, T <: HList] = HCons[H, T]
}
object HNil extends HNil
sealed trait HList {
type Head
type Tail <: HList
}
sealed class HNil extends HList {
type Head = Nothing
type Tail = HNil
def ::[T](v : T) = HCons(v, this)
}
final case class HCons[H, T <: HList](head : H, tail : T) extends HList {
type Head = H
type Tail = T
def ::[T](v : T) = HCons(v, this)
}