diff options
Diffstat (limited to 'src')
58 files changed, 2710 insertions, 1625 deletions
diff --git a/src/actors/scala/actors/remote/TcpService.scala b/src/actors/scala/actors/remote/TcpService.scala index 8163ae9fc6..ad78ff784c 100644 --- a/src/actors/scala/actors/remote/TcpService.scala +++ b/src/actors/scala/actors/remote/TcpService.scala @@ -14,7 +14,7 @@ package remote import java.io.{DataInputStream, DataOutputStream, IOException} import java.lang.{Thread, SecurityException} -import java.net.{InetAddress, ServerSocket, Socket, UnknownHostException} +import java.net.{InetAddress, InetSocketAddress, ServerSocket, Socket, SocketTimeoutException, UnknownHostException} import scala.collection.mutable import scala.util.Random @@ -60,6 +60,23 @@ object TcpService { portnum } + private val connectTimeoutMillis = { + val propName = "scala.actors.tcpSocket.connectTimeoutMillis" + val defaultTimeoutMillis = 0 + sys.props get propName flatMap { + timeout => + try { + val to = timeout.toInt + Debug.info("Using socket timeout $to") + Some(to) + } catch { + case e: NumberFormatException => + Debug.warning(s"""Could not parse $propName = "$timeout" as an Int""") + None + } + } getOrElse defaultTimeoutMillis + } + var BufSize: Int = 65536 } @@ -178,7 +195,15 @@ class TcpService(port: Int, cl: ClassLoader) extends Thread with Service { } def connect(n: Node): TcpServiceWorker = synchronized { - val socket = new Socket(n.address, n.port) + val socket = new Socket() + val start = System.nanoTime + try { + socket.connect(new InetSocketAddress(n.address, n.port), TcpService.connectTimeoutMillis) + } catch { + case e: SocketTimeoutException => + Debug.warning(f"Timed out connecting to $n after ${(System.nanoTime - start) / math.pow(10, 9)}%.3f seconds") + throw e + } val worker = new TcpServiceWorker(this, socket) worker.sendNode(n) worker.start() diff --git a/src/build/maven/scala-actors-pom.xml b/src/build/maven/scala-actors-pom.xml index e8c6649721..3d37ef8174 100644 --- a/src/build/maven/scala-actors-pom.xml +++ b/src/build/maven/scala-actors-pom.xml @@ -30,6 +30,9 @@ <system>JIRA</system> <url>https://issues.scala-lang.org/</url> </issueManagement> + <properties> + <info.apiURL>http://www.scala-lang.org/api/@VERSION@/</info.apiURL> + </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> diff --git a/src/build/maven/scala-library-pom.xml b/src/build/maven/scala-library-pom.xml index d1192b2dd8..fc9964ae92 100644 --- a/src/build/maven/scala-library-pom.xml +++ b/src/build/maven/scala-library-pom.xml @@ -30,6 +30,9 @@ <system>JIRA</system> <url>https://issues.scala-lang.org/</url> </issueManagement> + <properties> + <info.apiURL>http://www.scala-lang.org/api/@VERSION@/</info.apiURL> + </properties> <dependencies> <!--<dependency> <groupId>com.typesafe</groupId> diff --git a/src/build/maven/scala-reflect-pom.xml b/src/build/maven/scala-reflect-pom.xml index 7a1613f42c..56d2ffc57c 100644 --- a/src/build/maven/scala-reflect-pom.xml +++ b/src/build/maven/scala-reflect-pom.xml @@ -30,7 +30,9 @@ <system>JIRA</system> <url>https://issues.scala-lang.org/</url> </issueManagement> - + <properties> + <info.apiURL>http://www.scala-lang.org/api/@VERSION@/</info.apiURL> + </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> diff --git a/src/build/maven/scala-swing-pom.xml b/src/build/maven/scala-swing-pom.xml index 3df5db5b21..5099fe11dc 100644 --- a/src/build/maven/scala-swing-pom.xml +++ b/src/build/maven/scala-swing-pom.xml @@ -30,6 +30,9 @@ <system>JIRA</system> <url>https://issues.scala-lang.org/</url> </issueManagement> + <properties> + <info.apiURL>http://www.scala-lang.org/api/@VERSION@/</info.apiURL> + </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> diff --git a/src/compiler/scala/reflect/macros/util/Traces.scala b/src/compiler/scala/reflect/macros/util/Traces.scala index d16916b753..2dffc68745 100644 --- a/src/compiler/scala/reflect/macros/util/Traces.scala +++ b/src/compiler/scala/reflect/macros/util/Traces.scala @@ -6,8 +6,6 @@ trait Traces { val macroDebugLite = globalSettings.YmacrodebugLite.value val macroDebugVerbose = globalSettings.YmacrodebugVerbose.value - val macroTraceLite = scala.tools.nsc.util.trace when (macroDebugLite || macroDebugVerbose) - val macroTraceVerbose = scala.tools.nsc.util.trace when macroDebugVerbose @inline final def macroLogLite(msg: => Any) { if (macroDebugLite || macroDebugVerbose) println(msg) } @inline final def macroLogVerbose(msg: => Any) { if (macroDebugVerbose) println(msg) } } diff --git a/src/compiler/scala/tools/nsc/ast/parser/Parsers.scala b/src/compiler/scala/tools/nsc/ast/parser/Parsers.scala index 1dfb2f3bab..76fe76315f 100644 --- a/src/compiler/scala/tools/nsc/ast/parser/Parsers.scala +++ b/src/compiler/scala/tools/nsc/ast/parser/Parsers.scala @@ -1086,9 +1086,8 @@ self => * | symbol * | null * }}} - * @note The returned tree does not yet have a position */ - def literal(isNegated: Boolean = false, inPattern: Boolean = false): Tree = { + def literal(isNegated: Boolean = false, inPattern: Boolean = false, start: Int = in.offset): Tree = atPos(start) { def finish(value: Any): Tree = try newLiteral(value) finally in.nextToken() if (in.token == SYMBOLLIT) Apply(scalaDot(nme.Symbol), List(finish(in.strVal))) @@ -1282,12 +1281,12 @@ self => parseTry case WHILE => def parseWhile = { + val start = in.offset atPos(in.skipToken()) { - val lname: Name = freshTermName(nme.WHILE_PREFIX) val cond = condExpr() newLinesOpt() val body = expr() - makeWhile(lname.toTermName, cond, body) + makeWhile(start, cond, body) } } parseWhile @@ -1460,7 +1459,7 @@ self => atPos(in.offset) { val name = nme.toUnaryName(rawIdent().toTermName) if (name == nme.UNARY_- && isNumericLit) - simpleExprRest(atPos(in.offset)(literal(isNegated = true)), canApply = true) + simpleExprRest(literal(isNegated = true), canApply = true) else Select(stripParens(simpleExpr()), name) } @@ -1485,7 +1484,7 @@ self => def simpleExpr(): Tree = { var canApply = true val t = - if (isLiteral) atPos(in.offset)(literal()) + if (isLiteral) literal() else in.token match { case XMLSTART => xmlLiteral() @@ -1862,7 +1861,7 @@ self => case INTLIT | LONGLIT | FLOATLIT | DOUBLELIT => t match { case Ident(nme.MINUS) => - return atPos(start) { literal(isNegated = true, inPattern = true) } + return literal(isNegated = true, inPattern = true, start = start) case _ => } case _ => @@ -1880,7 +1879,7 @@ self => atPos(start, start) { Ident(nme.WILDCARD) } case CHARLIT | INTLIT | LONGLIT | FLOATLIT | DOUBLELIT | STRINGLIT | INTERPOLATIONID | SYMBOLLIT | TRUE | FALSE | NULL => - atPos(start) { literal(inPattern = true) } + literal(inPattern = true) case LPAREN => atPos(start)(makeParens(noSeq.patterns())) case XMLSTART => diff --git a/src/compiler/scala/tools/nsc/ast/parser/Scanners.scala b/src/compiler/scala/tools/nsc/ast/parser/Scanners.scala index c1a8b96cdf..6786d10036 100644 --- a/src/compiler/scala/tools/nsc/ast/parser/Scanners.scala +++ b/src/compiler/scala/tools/nsc/ast/parser/Scanners.scala @@ -467,6 +467,7 @@ trait Scanners extends ScannersCommon { if (ch == '\"') { nextRawChar() if (ch == '\"') { + offset += 3 nextRawChar() getStringPart(multiLine = true) sepRegions = STRINGPART :: sepRegions // indicate string part @@ -476,6 +477,7 @@ trait Scanners extends ScannersCommon { strVal = "" } } else { + offset += 1 getStringPart(multiLine = false) sepRegions = STRINGLIT :: sepRegions // indicate single line string part } diff --git a/src/compiler/scala/tools/nsc/ast/parser/TreeBuilder.scala b/src/compiler/scala/tools/nsc/ast/parser/TreeBuilder.scala index 5f361f32b5..f737fcc635 100644 --- a/src/compiler/scala/tools/nsc/ast/parser/TreeBuilder.scala +++ b/src/compiler/scala/tools/nsc/ast/parser/TreeBuilder.scala @@ -248,8 +248,13 @@ abstract class TreeBuilder { else CompoundTypeTree(Template(tps, emptyValDef, Nil)) /** Create tree representing a while loop */ - def makeWhile(lname: TermName, cond: Tree, body: Tree): Tree = { - val continu = atPos(o2p(body.pos pointOrElse wrappingPos(List(cond, body)).pos.endOrPoint)) { Apply(Ident(lname), Nil) } + def makeWhile(startPos: Int, cond: Tree, body: Tree): Tree = { + val lname = freshTermName(nme.WHILE_PREFIX) + def default = wrappingPos(List(cond, body)) match { + case p if p.isDefined => p.endOrPoint + case _ => startPos + } + val continu = atPos(o2p(body.pos pointOrElse default)) { Apply(Ident(lname), Nil) } val rhs = If(cond, Block(List(body), continu), Literal(Constant(()))) LabelDef(lname, Nil, rhs) } diff --git a/src/compiler/scala/tools/nsc/backend/icode/BasicBlocks.scala b/src/compiler/scala/tools/nsc/backend/icode/BasicBlocks.scala index 16ee40280c..26d10d2514 100644 --- a/src/compiler/scala/tools/nsc/backend/icode/BasicBlocks.scala +++ b/src/compiler/scala/tools/nsc/backend/icode/BasicBlocks.scala @@ -8,8 +8,7 @@ package backend package icode import scala.collection.{ mutable, immutable } -import mutable.{ ListBuffer, ArrayBuffer } -import scala.reflect.internal.util.{ Position, NoPosition } +import mutable.ListBuffer import backend.icode.analysis.ProgramPoint import scala.language.postfixOps @@ -17,8 +16,7 @@ trait BasicBlocks { self: ICodes => import opcodes._ - import global.{ settings, debuglog, log, nme } - import nme.isExceptionResultName + import global._ /** Override Array creation for efficiency (to not go through reflection). */ private implicit val instructionTag: scala.reflect.ClassTag[Instruction] = new scala.reflect.ClassTag[Instruction] { @@ -336,9 +334,9 @@ trait BasicBlocks { * put into ignore mode so we hear about it if there's a problem. */ instr match { - case JUMP(_) | RETURN(_) | THROW(_) | SCOPE_EXIT(_) => // ok - case STORE_LOCAL(local) if isExceptionResultName(local.sym.name) => // ok - case x => log("Ignoring instruction, possibly at our peril, at " + pos + ": " + x) + case JUMP(_) | RETURN(_) | THROW(_) | SCOPE_EXIT(_) => // ok + case STORE_LOCAL(local) if nme.isExceptionResultName(local.sym.name) => // ok + case x => log("Ignoring instruction, possibly at our peril, at " + pos + ": " + x) } } } @@ -416,7 +414,7 @@ trait BasicBlocks { enterIgnoreMode() } } - + /** * Same as killIf but with the logic of the condition reversed */ @@ -477,16 +475,17 @@ trait BasicBlocks { def directSuccessors: List[BasicBlock] = if (isEmpty) Nil else lastInstruction match { - case JUMP(whereto) => whereto :: Nil - case CJUMP(succ, fail, _, _) => fail :: succ :: Nil - case CZJUMP(succ, fail, _, _) => fail :: succ :: Nil - case SWITCH(_, labels) => labels - case RETURN(_) => Nil - case THROW(_) => Nil - case _ => + case JUMP(whereto) => whereto :: Nil + case CJUMP(succ, fail, _, _) => fail :: succ :: Nil + case CZJUMP(succ, fail, _, _) => fail :: succ :: Nil + case SWITCH(_, labels) => labels + case RETURN(_) => Nil + case THROW(_) => Nil + case _ => if (closed) - dumpClassesAndAbort("The last instruction is not a control flow instruction: " + lastInstruction) - else Nil + devWarning(s"$lastInstruction/${lastInstruction.getClass.getName} is not a control flow instruction") + + Nil } /** Returns the predecessors of this block. */ diff --git a/src/compiler/scala/tools/nsc/backend/icode/GenICode.scala b/src/compiler/scala/tools/nsc/backend/icode/GenICode.scala index 9c9c001f02..cc4d97253f 100644 --- a/src/compiler/scala/tools/nsc/backend/icode/GenICode.scala +++ b/src/compiler/scala/tools/nsc/backend/icode/GenICode.scala @@ -2165,8 +2165,7 @@ abstract class GenICode extends SubComponent { * by a real JUMP instruction when all labels are resolved. */ abstract class PseudoJUMP(label: Label) extends Instruction { - override def toString(): String = "PJUMP " + label.symbol - + override def toString = s"PJUMP(${label.symbol})" override def consumed = 0 override def produced = 0 diff --git a/src/compiler/scala/tools/nsc/javac/JavaParsers.scala b/src/compiler/scala/tools/nsc/javac/JavaParsers.scala index 786754ce4c..80b475d5c0 100644 --- a/src/compiler/scala/tools/nsc/javac/JavaParsers.scala +++ b/src/compiler/scala/tools/nsc/javac/JavaParsers.scala @@ -368,6 +368,9 @@ trait JavaParsers extends ast.parser.ParsersCommon with JavaScanners { case FINAL => flags |= Flags.FINAL in.nextToken() + case DEFAULT => + flags |= Flags.DEFAULTMETHOD + in.nextToken() case NATIVE => addAnnot(NativeAttr) in.nextToken() @@ -485,8 +488,9 @@ trait JavaParsers extends ast.parser.ParsersCommon with JavaScanners { val vparams = formalParams() if (!isVoid) rtpt = optArrayBrackets(rtpt) optThrows() + val bodyOk = !inInterface || (mods hasFlag Flags.DEFAULTMETHOD) val body = - if (!inInterface && in.token == LBRACE) { + if (bodyOk && in.token == LBRACE) { methodBody() } else { if (parentToken == AT && in.token == DEFAULT) { diff --git a/src/compiler/scala/tools/nsc/settings/ScalaSettings.scala b/src/compiler/scala/tools/nsc/settings/ScalaSettings.scala index c105dcbdff..fe9165203f 100644 --- a/src/compiler/scala/tools/nsc/settings/ScalaSettings.scala +++ b/src/compiler/scala/tools/nsc/settings/ScalaSettings.scala @@ -163,8 +163,6 @@ trait ScalaSettings extends AbsScalaSettings val Ystatistics = BooleanSetting ("-Ystatistics", "Print compiler statistics.") andThen (scala.reflect.internal.util.Statistics.enabled = _) val stopAfter = PhasesSetting ("-Ystop-after", "Stop after") withAbbreviation ("-stop") // backward compat val stopBefore = PhasesSetting ("-Ystop-before", "Stop before") - val refinementMethodDispatch - = ChoiceSetting ("-Ystruct-dispatch", "policy", "structural method dispatch policy", List("no-cache", "mono-cache", "poly-cache", "invoke-dynamic"), "poly-cache") val Yrangepos = BooleanSetting ("-Yrangepos", "Use range positions for syntax trees.") val Ymemberpos = StringSetting ("-Yshow-member-pos", "output style", "Show start and end positions of members", "") withPostSetHook (_ => Yrangepos.value = true) val Yreifycopypaste = BooleanSetting ("-Yreify-copypaste", "Dump the reified trees in copypasteable representation.") diff --git a/src/compiler/scala/tools/nsc/symtab/classfile/ClassfileParser.scala b/src/compiler/scala/tools/nsc/symtab/classfile/ClassfileParser.scala index 42bf0f6b41..9afa1ba10d 100644 --- a/src/compiler/scala/tools/nsc/symtab/classfile/ClassfileParser.scala +++ b/src/compiler/scala/tools/nsc/symtab/classfile/ClassfileParser.scala @@ -75,18 +75,14 @@ abstract class ClassfileParser { private def handleMissing(e: MissingRequirementError) = { if (settings.debug) e.printStackTrace - throw new IOException("Missing dependency '" + e.req + "', required by " + in.file) + throw new IOException(s"Missing dependency '${e.req}', required by ${in.file}") } private def handleError(e: Exception) = { if (settings.debug) e.printStackTrace() - throw new IOException("class file '%s' is broken\n(%s/%s)".format( - in.file, - e.getClass, - if (e.getMessage eq null) "" else e.getMessage) - ) + throw new IOException(s"class file '${in.file}' is broken\n(${e.getClass}/${e.getMessage})") } private def mismatchError(c: Symbol) = { - throw new IOException("class file '%s' has location not matching its contents: contains ".format(in.file) + c) + throw new IOException(s"class file '${in.file}' has location not matching its contents: contains $c") } private def parseErrorHandler[T]: PartialFunction[Throwable, T] = { @@ -95,8 +91,8 @@ abstract class ClassfileParser { } @inline private def pushBusy[T](sym: Symbol)(body: => T): T = { busy match { - case Some(`sym`) => throw new IOException("unsatisfiable cyclic dependency in '%s'".format(sym)) - case Some(sym1) => throw new IOException("illegal class file dependency between '%s' and '%s'".format(sym, sym1)) + case Some(`sym`) => throw new IOException(s"unsatisfiable cyclic dependency in '$sym'") + case Some(sym1) => throw new IOException(s"illegal class file dependency between '$sym' and '$sym1'") case _ => () } @@ -242,8 +238,6 @@ abstract class ClassfileParser { forceMangledName(tpe0.typeSymbol.name, module = false) val (name, tpe) = getNameAndType(in.getChar(start + 3), ownerTpe) -// println("new tpe: " + tpe + " at phase: " + phase) - if (name == nme.MODULE_INSTANCE_FIELD) { val index = in.getChar(start + 1) val name = getExternalName(in.getChar(starts(index) + 1)) @@ -254,14 +248,12 @@ abstract class ClassfileParser { } else { val origName = nme.unexpandedName(name) val owner = if (static) ownerTpe.typeSymbol.linkedClassOfClass else ownerTpe.typeSymbol -// println("\t" + owner.info.member(name).tpe.widen + " =:= " + tpe) f = owner.info.findMember(origName, 0, 0, stableOnly = false).suchThat(_.tpe.widen =:= tpe) if (f == NoSymbol) f = owner.info.findMember(newTermName(origName + nme.LOCAL_SUFFIX_STRING), 0, 0, stableOnly = false).suchThat(_.tpe =:= tpe) if (f == NoSymbol) { // if it's an impl class, try to find it's static member inside the class if (ownerTpe.typeSymbol.isImplClass) { -// println("impl class, member: " + owner.tpe.member(origName) + ": " + owner.tpe.member(origName).tpe) f = ownerTpe.findMember(origName, 0, 0, stableOnly = false).suchThat(_.tpe =:= tpe) } else { log("Couldn't find " + name + ": " + tpe + " inside: \n" + ownerTpe) @@ -272,11 +264,13 @@ abstract class ClassfileParser { f setInfo tpe log("created fake member " + f.fullName) } -// println("\townerTpe.decls: " + ownerTpe.decls) -// println("Looking for: " + name + ": " + tpe + " inside: " + ownerTpe.typeSymbol + "\n\tand found: " + ownerTpe.members) } } - assert(f != NoSymbol, "could not find\n " + name + ": " + tpe + "\ninside:\n " + ownerTpe.members.mkString(", ")) + assert(f != NoSymbol, + s"could not find $name: $tpe in $ownerTpe" + ( + if (settings.debug.value) ownerTpe.members.mkString(", members are:\n ", "\n ", "") else "" + ) + ) values(index) = f } f @@ -560,11 +554,9 @@ abstract class ClassfileParser { def addEnclosingTParams(clazz: Symbol) { var sym = clazz.owner while (sym.isClass && !sym.isModuleClass) { -// println("adding tparams of " + sym) - for (t <- sym.tpe.typeArgs) { -// println("\tadding " + (t.typeSymbol.name + "->" + t.typeSymbol)) + for (t <- sym.tpe.typeArgs) classTParams = classTParams + (t.typeSymbol.name -> t.typeSymbol) - } + sym = sym.owner } } @@ -838,8 +830,6 @@ abstract class ClassfileParser { val sig = pool.getExternalName(u2) val newType = sigToType(sym, sig) sym.setInfo(newType) - if (settings.debug && settings.verbose) - println("" + sym + "; signature = " + sig + " type = " + newType) } else in.skip(attrLen) case tpnme.SyntheticATTR => @@ -856,10 +846,10 @@ abstract class ClassfileParser { val c = pool.getConstant(u2) val c1 = convertTo(c, symtype) if (c1 ne null) sym.setInfo(ConstantType(c1)) - else println("failure to convert " + c + " to " + symtype); //debug + else debugwarn(s"failure to convert $c to $symtype") case tpnme.ScalaSignatureATTR => if (!isScalaAnnot) { - debuglog("warning: symbol " + sym.fullName + " has pickled signature in attribute") + debugwarn(s"symbol ${sym.fullName} has pickled signature in attribute") unpickler.unpickle(in.buf, in.bp, clazz, staticModule, in.file.name) } in.skip(attrLen) @@ -904,6 +894,12 @@ abstract class ClassfileParser { case pkg => pkg.fullName(File.separatorChar)+File.separator+srcfileLeaf } srcfile0 = settings.outputDirs.srcFilesFor(in.file, srcpath).find(_.exists) + case tpnme.CodeATTR => + if (sym.owner.isInterface) { + sym setFlag DEFAULTMETHOD + log(s"$sym in ${sym.owner} is a java8+ default method.") + } + in.skip(attrLen) case _ => in.skip(attrLen) } @@ -991,16 +987,18 @@ abstract class ClassfileParser { } if (hasError) None else Some(AnnotationInfo(attrType, List(), nvpairs.toList)) - } catch { - case f: FatalError => throw f // don't eat fatal errors, they mean a class was not found - case ex: Throwable => + } + catch { + case f: FatalError => throw f // don't eat fatal errors, they mean a class was not found + case ex: java.lang.Error => throw ex + case ex: Throwable => // We want to be robust when annotations are unavailable, so the very least // we can do is warn the user about the exception // There was a reference to ticket 1135, but that is outdated: a reference to a class not on // the classpath would *not* end up here. A class not found is signaled // with a `FatalError` exception, handled above. Here you'd end up after a NPE (for example), // and that should never be swallowed silently. - warning("Caught: " + ex + " while parsing annotations in " + in.file) + warning(s"Caught: $ex while parsing annotations in ${in.file}") if (settings.debug) ex.printStackTrace() None // ignore malformed annotations diff --git a/src/compiler/scala/tools/nsc/symtab/classfile/Pickler.scala b/src/compiler/scala/tools/nsc/symtab/classfile/Pickler.scala index 94880c4b2e..a5619871e6 100644 --- a/src/compiler/scala/tools/nsc/symtab/classfile/Pickler.scala +++ b/src/compiler/scala/tools/nsc/symtab/classfile/Pickler.scala @@ -308,7 +308,6 @@ abstract class Pickler extends SubComponent { putTree(definition) */ case Template(parents, self, body) => - writeNat(parents.length) putTrees(parents) putTree(self) putTrees(body) diff --git a/src/compiler/scala/tools/nsc/transform/CleanUp.scala b/src/compiler/scala/tools/nsc/transform/CleanUp.scala index d498949b03..d159408f55 100644 --- a/src/compiler/scala/tools/nsc/transform/CleanUp.scala +++ b/src/compiler/scala/tools/nsc/transform/CleanUp.scala @@ -47,15 +47,6 @@ abstract class CleanUp extends Transform with ast.TreeDSL { private var localTyper: analyzer.Typer = null - private object MethodDispatchType extends scala.Enumeration { - val NO_CACHE, MONO_CACHE, POLY_CACHE = Value - } - import MethodDispatchType.{ NO_CACHE, MONO_CACHE, POLY_CACHE } - private def dispatchType() = settings.refinementMethodDispatch.value match { - case "no-cache" => NO_CACHE - case "mono-cache" => MONO_CACHE - case "poly-cache" => POLY_CACHE - } private def typedWithPos(pos: Position)(tree: Tree) = localTyper.typedPos(pos)(tree) @@ -113,133 +104,65 @@ abstract class CleanUp extends Transform with ast.TreeDSL { def fromTypesToClassArrayLiteral(paramTypes: List[Type]): Tree = ArrayValue(TypeTree(ClassClass.tpe), paramTypes map LIT) - /* ... */ - def reflectiveMethodCache(method: String, paramTypes: List[Type]): Symbol = dispatchType() match { - case NO_CACHE => - - /* Implementation of the cache is as follows for method "def xyz(a: A, b: B)": - - var reflParams$Cache: Array[Class[_]] = Array[JClass](classOf[A], classOf[B]) - - def reflMethod$Method(forReceiver: JClass[_]): JMethod = - forReceiver.getMethod("xyz", reflParams$Cache) - - */ - - val reflParamsCacheSym: Symbol = - addStaticVariableToClass(nme.reflParamsCacheName, arrayType(ClassClass.tpe), fromTypesToClassArrayLiteral(paramTypes), true) - - addStaticMethodToClass((_, forReceiverSym) => - gen.mkMethodCall(REF(forReceiverSym), Class_getMethod, Nil, List(LIT(method), REF(reflParamsCacheSym))) - ) - - case MONO_CACHE => - - /* Implementation of the cache is as follows for method "def xyz(a: A, b: B)" - (but with a SoftReference wrapping reflClass$Cache, similarly in the poly Cache) : - - var reflParams$Cache: Array[Class[_]] = Array[JClass](classOf[A], classOf[B]) - - var reflMethod$Cache: JMethod = null - - var reflClass$Cache: JClass[_] = null - - def reflMethod$Method(forReceiver: JClass[_]): JMethod = { - if (reflClass$Cache != forReceiver) { - reflMethod$Cache = forReceiver.getMethod("xyz", reflParams$Cache) - reflClass$Cache = forReceiver - } - reflMethod$Cache - } - - */ - - val reflParamsCacheSym: Symbol = - addStaticVariableToClass(nme.reflParamsCacheName, arrayType(ClassClass.tpe), fromTypesToClassArrayLiteral(paramTypes), true) - - val reflMethodCacheSym: Symbol = - addStaticVariableToClass(nme.reflMethodCacheName, MethodClass.tpe, NULL, false) - - val reflClassCacheSym: Symbol = - addStaticVariableToClass(nme.reflClassCacheName, SoftReferenceClass.tpe, NULL, false) - - def isCacheEmpty(receiver: Symbol): Tree = - reflClassCacheSym.IS_NULL() OR (reflClassCacheSym.GET() OBJ_NE REF(receiver)) - - addStaticMethodToClass((_, forReceiverSym) => - BLOCK( - IF (isCacheEmpty(forReceiverSym)) THEN BLOCK( - REF(reflMethodCacheSym) === ((REF(forReceiverSym) DOT Class_getMethod)(LIT(method), REF(reflParamsCacheSym))) , - REF(reflClassCacheSym) === gen.mkSoftRef(REF(forReceiverSym)), - UNIT - ) ENDIF, - REF(reflMethodCacheSym) - ) - ) - - case POLY_CACHE => + def reflectiveMethodCache(method: String, paramTypes: List[Type]): Symbol = { + /* Implementation of the cache is as follows for method "def xyz(a: A, b: B)" + (SoftReference so that it does not interfere with classloader garbage collection, + see ticket #2365 for details): - /* Implementation of the cache is as follows for method "def xyz(a: A, b: B)" - (SoftReference so that it does not interfere with classloader garbage collection, see ticket - #2365 for details): + var reflParams$Cache: Array[Class[_]] = Array[JClass](classOf[A], classOf[B]) - var reflParams$Cache: Array[Class[_]] = Array[JClass](classOf[A], classOf[B]) + var reflPoly$Cache: SoftReference[scala.runtime.MethodCache] = new SoftReference(new EmptyMethodCache()) - var reflPoly$Cache: SoftReference[scala.runtime.MethodCache] = new SoftReference(new EmptyMethodCache()) + def reflMethod$Method(forReceiver: JClass[_]): JMethod = { + var methodCache: MethodCache = reflPoly$Cache.find(forReceiver) + if (methodCache eq null) { + methodCache = new EmptyMethodCache + reflPoly$Cache = new SoftReference(methodCache) + } + var method: JMethod = methodCache.find(forReceiver) + if (method ne null) + return method + else { + method = ScalaRunTime.ensureAccessible(forReceiver.getMethod("xyz", reflParams$Cache)) + reflPoly$Cache = new SoftReference(methodCache.add(forReceiver, method)) + return method + } + } + */ - def reflMethod$Method(forReceiver: JClass[_]): JMethod = { - var methodCache: MethodCache = reflPoly$Cache.find(forReceiver) - if (methodCache eq null) { - methodCache = new EmptyMethodCache - reflPoly$Cache = new SoftReference(methodCache) - } - var method: JMethod = methodCache.find(forReceiver) - if (method ne null) - return method - else { - method = ScalaRunTime.ensureAccessible(forReceiver.getMethod("xyz", reflParams$Cache)) - reflPoly$Cache = new SoftReference(methodCache.add(forReceiver, method)) - return method - } - } + val reflParamsCacheSym: Symbol = + addStaticVariableToClass(nme.reflParamsCacheName, arrayType(ClassClass.tpe), fromTypesToClassArrayLiteral(paramTypes), true) - */ + def mkNewPolyCache = gen.mkSoftRef(NEW(TypeTree(EmptyMethodCacheClass.tpe))) + val reflPolyCacheSym: Symbol = addStaticVariableToClass(nme.reflPolyCacheName, SoftReferenceClass.tpe, mkNewPolyCache, false) - val reflParamsCacheSym: Symbol = - addStaticVariableToClass(nme.reflParamsCacheName, arrayType(ClassClass.tpe), fromTypesToClassArrayLiteral(paramTypes), true) + def getPolyCache = gen.mkCast(fn(REF(reflPolyCacheSym), nme.get), MethodCacheClass.tpe) - def mkNewPolyCache = gen.mkSoftRef(NEW(TypeTree(EmptyMethodCacheClass.tpe))) - val reflPolyCacheSym: Symbol = ( - addStaticVariableToClass(nme.reflPolyCacheName, SoftReferenceClass.tpe, mkNewPolyCache, false) - ) - def getPolyCache = gen.mkCast(fn(REF(reflPolyCacheSym), nme.get), MethodCacheClass.tpe) + addStaticMethodToClass((reflMethodSym, forReceiverSym) => { + val methodCache = reflMethodSym.newVariable(mkTerm("methodCache"), ad.pos) setInfo MethodCacheClass.tpe + val methodSym = reflMethodSym.newVariable(mkTerm("method"), ad.pos) setInfo MethodClass.tpe - addStaticMethodToClass((reflMethodSym, forReceiverSym) => { - val methodCache = reflMethodSym.newVariable(mkTerm("methodCache"), ad.pos) setInfo MethodCacheClass.tpe - val methodSym = reflMethodSym.newVariable(mkTerm("method"), ad.pos) setInfo MethodClass.tpe + BLOCK( + VAL(methodCache) === getPolyCache, + IF (REF(methodCache) OBJ_EQ NULL) THEN BLOCK( + REF(methodCache) === NEW(TypeTree(EmptyMethodCacheClass.tpe)), + REF(reflPolyCacheSym) === gen.mkSoftRef(REF(methodCache)) + ) ENDIF, + VAL(methodSym) === (REF(methodCache) DOT methodCache_find)(REF(forReceiverSym)), + IF (REF(methodSym) OBJ_NE NULL) . + THEN (Return(REF(methodSym))) + ELSE { + def methodSymRHS = ((REF(forReceiverSym) DOT Class_getMethod)(LIT(method), REF(reflParamsCacheSym))) + def cacheRHS = ((REF(methodCache) DOT methodCache_add)(REF(forReceiverSym), REF(methodSym))) BLOCK( - VAL(methodCache) === getPolyCache, - IF (REF(methodCache) OBJ_EQ NULL) THEN BLOCK( - REF(methodCache) === NEW(TypeTree(EmptyMethodCacheClass.tpe)), - REF(reflPolyCacheSym) === gen.mkSoftRef(REF(methodCache)) - ) ENDIF, - - VAL(methodSym) === (REF(methodCache) DOT methodCache_find)(REF(forReceiverSym)), - IF (REF(methodSym) OBJ_NE NULL) . - THEN (Return(REF(methodSym))) - ELSE { - def methodSymRHS = ((REF(forReceiverSym) DOT Class_getMethod)(LIT(method), REF(reflParamsCacheSym))) - def cacheRHS = ((REF(methodCache) DOT methodCache_add)(REF(forReceiverSym), REF(methodSym))) - BLOCK( - REF(methodSym) === (REF(ensureAccessibleMethod) APPLY (methodSymRHS)), - REF(reflPolyCacheSym) === gen.mkSoftRef(cacheRHS), - Return(REF(methodSym)) - ) - } + REF(methodSym) === (REF(ensureAccessibleMethod) APPLY (methodSymRHS)), + REF(reflPolyCacheSym) === gen.mkSoftRef(cacheRHS), + Return(REF(methodSym)) ) - }) - + } + ) + }) } /* ### HANDLING METHODS NORMALLY COMPILED TO OPERATORS ### */ @@ -394,99 +317,75 @@ abstract class CleanUp extends Transform with ast.TreeDSL { } } - if (settings.refinementMethodDispatch.value == "invoke-dynamic") { -/* val guardCallSite: Tree = { - val cachedClass = addStaticVariableToClass("cachedClass", definitions.ClassClass.tpe, EmptyTree) - val tmpVar = currentOwner.newVariable(ad.pos, unit.freshTermName(ad.pos, "x")).setInfo(definitions.AnyRefClass.tpe) - atPos(ad.pos)(Block(List( - ValDef(tmpVar, transform(qual))), - If(Apply(Select(gen.mkAttributedRef(cachedClass), nme.EQ), List(getClass(Ident(tmpVar)))), - Block(List(Assign(gen.mkAttributedRef(cachedClass), getClass(Ident(tmpVar)))), - treeCopy.ApplyDynamic(ad, Ident(tmpVar), transformTrees(params))), - EmptyTree))) - } - //println(guardCallSite) -*/ - localTyper.typed(treeCopy.ApplyDynamic(ad, transform(qual), transformTrees(params))) - } - else { - - /* ### BODY OF THE TRANSFORMATION -> remember we're in case ad@ApplyDynamic(qual, params) ### */ - - /* This creates the tree that does the reflective call (see general comment - * on the apply-dynamic tree for its format). This tree is simply composed - * of three successive calls, first to getClass on the callee, then to - * getMethod on the class, then to invoke on the method. - * - getMethod needs an array of classes for choosing one amongst many - * overloaded versions of the method. This is provided by paramTypeClasses - * and must be done on the static type as Scala's dispatching is static on - * the parameters. - * - invoke needs an array of AnyRefs that are the method's arguments. The - * erasure phase guarantees that any parameter passed to a dynamic apply - * is compatible (through boxing). Boxed ints et al. is what invoke expects - * when the applied method expects ints, hence no change needed there. - * - in the end, the result of invoke must be fixed, again to deal with arrays. - * This is provided by fixResult. fixResult will cast the invocation's result - * to the method's return type, which is generally ok, except when this type - * is a value type (int et al.) in which case it must cast to the boxed version - * because invoke only returns object and erasure made sure the result is - * expected to be an AnyRef. */ - val t: Tree = { - val (mparams, resType) = ad.symbol.tpe match { - case MethodType(mparams, resType) => - assert(params.length == mparams.length, ((params, mparams))) - (mparams, resType) - case tpe @ OverloadedType(pre, alts) => - unit.warning(ad.pos, s"Overloaded type reached the backend! This is a bug in scalac.\n Symbol: ${ad.symbol}\n Overloads: $tpe\n Arguments: " + ad.args.map(_.tpe)) - alts filter (_.paramss.flatten.size == params.length) map (_.tpe) match { - case mt @ MethodType(mparams, resType) :: Nil => - unit.warning(NoPosition, "Only one overload has the right arity, proceeding with overload " + mt) - (mparams, resType) - case _ => - unit.error(ad.pos, "Cannot resolve overload.") - (Nil, NoType) - } - } - typedPos { - val sym = currentOwner.newValue(mkTerm("qual"), ad.pos) setInfo qual0.tpe - qual = REF(sym) - - BLOCK( - VAL(sym) === qual0, - callAsReflective(mparams map (_.tpe), resType) - ) - } + { + + /* ### BODY OF THE TRANSFORMATION -> remember we're in case ad@ApplyDynamic(qual, params) ### */ + + /* This creates the tree that does the reflective call (see general comment + * on the apply-dynamic tree for its format). This tree is simply composed + * of three successive calls, first to getClass on the callee, then to + * getMethod on the class, then to invoke on the method. + * - getMethod needs an array of classes for choosing one amongst many + * overloaded versions of the method. This is provided by paramTypeClasses + * and must be done on the static type as Scala's dispatching is static on + * the parameters. + * - invoke needs an array of AnyRefs that are the method's arguments. The + * erasure phase guarantees that any parameter passed to a dynamic apply + * is compatible (through boxing). Boxed ints et al. is what invoke expects + * when the applied method expects ints, hence no change needed there. + * - in the end, the result of invoke must be fixed, again to deal with arrays. + * This is provided by fixResult. fixResult will cast the invocation's result + * to the method's return type, which is generally ok, except when this type + * is a value type (int et al.) in which case it must cast to the boxed version + * because invoke only returns object and erasure made sure the result is + * expected to be an AnyRef. */ + val t: Tree = { + val (mparams, resType) = ad.symbol.tpe match { + case MethodType(mparams, resType) => + assert(params.length == mparams.length, ((params, mparams))) + (mparams, resType) + case tpe @ OverloadedType(pre, alts) => + unit.warning(ad.pos, s"Overloaded type reached the backend! This is a bug in scalac.\n Symbol: ${ad.symbol}\n Overloads: $tpe\n Arguments: " + ad.args.map(_.tpe)) + alts filter (_.paramss.flatten.size == params.length) map (_.tpe) match { + case mt @ MethodType(mparams, resType) :: Nil => + unit.warning(NoPosition, "Only one overload has the right arity, proceeding with overload " + mt) + (mparams, resType) + case _ => + unit.error(ad.pos, "Cannot resolve overload.") + (Nil, NoType) + } } + typedPos { + val sym = currentOwner.newValue(mkTerm("qual"), ad.pos) setInfo qual0.tpe + qual = REF(sym) - /* For testing purposes, the dynamic application's condition - * can be printed-out in great detail. Remove? */ - if (settings.debug) { - def paramsToString(xs: Any*) = xs map (_.toString) mkString ", " - val mstr = ad.symbol.tpe match { - case MethodType(mparams, resType) => - """| with - | - declared parameter types: '%s' - | - passed argument types: '%s' - | - result type: '%s'""" . - stripMargin.format( - paramsToString(mparams), - paramsToString(params), - resType.toString - ) - case _ => "" - } - log( - """Dynamically application '%s.%s(%s)' %s - resulting code: '%s'""".format( - qual, ad.symbol.name, paramsToString(params), mstr, t - ) + BLOCK( + VAL(sym) === qual0, + callAsReflective(mparams map (_.tpe), resType) ) } + } - /* We return the dynamic call tree, after making sure no other - * clean-up transformation are to be applied on it. */ - transform(t) + /* For testing purposes, the dynamic application's condition + * can be printed-out in great detail. Remove? */ + if (settings.debug) { + def paramsToString(xs: Any*) = xs map (_.toString) mkString ", " + val mstr = ad.symbol.tpe match { + case MethodType(mparams, resType) => + sm"""| with + | - declared parameter types: '${paramsToString(mparams)}' + | - passed argument types: '${paramsToString(params)}' + | - result type: '${resType.toString}'""" + case _ => "" + } + log(s"""Dynamically application '$qual.${ad.symbol.name}(${paramsToString(params)})' $mstr - resulting code: '$t'""") } - /* ### END OF DYNAMIC APPLY TRANSFORM ### */ + + /* We return the dynamic call tree, after making sure no other + * clean-up transformation are to be applied on it. */ + transform(t) + /* ### END OF DYNAMIC APPLY TRANSFORM ### */ + } } override def transform(tree: Tree): Tree = tree match { diff --git a/src/compiler/scala/tools/nsc/transform/patmat/MatchAnalysis.scala b/src/compiler/scala/tools/nsc/transform/patmat/MatchAnalysis.scala index 006c7bd85f..68883831a2 100644 --- a/src/compiler/scala/tools/nsc/transform/patmat/MatchAnalysis.scala +++ b/src/compiler/scala/tools/nsc/transform/patmat/MatchAnalysis.scala @@ -95,7 +95,9 @@ trait TreeAndTypeAnalysis extends Debugging { // TODO: when type tags are available, we will check -- when this is implemented, can we take that into account here? // similar to typer.infer.approximateAbstracts object typeArgsToWildcardsExceptArray extends TypeMap { - def apply(tp: Type): Type = tp match { + // SI-6771 dealias would be enough today, but future proofing with the dealiasWiden. + // See neg/t6771b.scala for elaboration + def apply(tp: Type): Type = tp.dealiasWiden match { case TypeRef(pre, sym, args) if args.nonEmpty && (sym ne ArrayClass) => TypeRef(pre, sym, args map (_ => WildcardType)) case _ => diff --git a/src/compiler/scala/tools/nsc/typechecker/ContextErrors.scala b/src/compiler/scala/tools/nsc/typechecker/ContextErrors.scala index e1e26cd8c9..61e4f6bb1e 100644 --- a/src/compiler/scala/tools/nsc/typechecker/ContextErrors.scala +++ b/src/compiler/scala/tools/nsc/typechecker/ContextErrors.scala @@ -1308,6 +1308,11 @@ trait ContextErrors { throw MacroBodyTypecheckException // don't call fail, because we don't need IS_ERROR } + def MacroDefIsQmarkQmarkQmark() = { + macroLogVerbose("typecheck terminated unexpectedly: macro is ???") + throw MacroBodyTypecheckException + } + def MacroFeatureNotEnabled() = { macroLogVerbose("typecheck terminated unexpectedly: language.experimental.macros feature is not enabled") fail() diff --git a/src/compiler/scala/tools/nsc/typechecker/Implicits.scala b/src/compiler/scala/tools/nsc/typechecker/Implicits.scala index 05db86635a..37257b4f4e 100644 --- a/src/compiler/scala/tools/nsc/typechecker/Implicits.scala +++ b/src/compiler/scala/tools/nsc/typechecker/Implicits.scala @@ -929,10 +929,23 @@ trait Implicits { */ if (DivergentImplicitRecovery.sym != null) { DivergingImplicitExpansionError(tree, pt, DivergentImplicitRecovery.sym)(context) - } else if (invalidImplicits.nonEmpty) + } + else if (invalidImplicits.nonEmpty) { + val sym = invalidImplicits.head + // We don't even dare look if errors are being buffered + // !sym.hasFlag(LOCKED) is a hail mary between SI-2206 and SI-7486 + def isSensibleAddendum = !sym.hasFlag(LOCKED) && (pt match { + case Function1(_, out) => out <:< sym.tpe.finalResultType + case _ => pt <:< sym.tpe.finalResultType + }) + // Don't pitch in with this theory unless it looks plausible that the + // implicit would have helped setAddendum(pos, () => - "\n Note: implicit "+invalidImplicits.head+" is not applicable here"+ - " because it comes after the application point and it lacks an explicit result type") + if (isSensibleAddendum) + s"\n Note: implicit $sym is not applicable here because it comes after the application point and it lacks an explicit result type" + else "" + ) + } } best diff --git a/src/compiler/scala/tools/nsc/typechecker/Macros.scala b/src/compiler/scala/tools/nsc/typechecker/Macros.scala index ecae55562b..252ad56b24 100644 --- a/src/compiler/scala/tools/nsc/typechecker/Macros.scala +++ b/src/compiler/scala/tools/nsc/typechecker/Macros.scala @@ -344,11 +344,14 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { } import SigGenerator._ - macroTraceVerbose("generating macroImplSigs for: ")(macroDef) - macroTraceVerbose("tparams are: ")(tparams) - macroTraceVerbose("vparamss are: ")(vparamss) - macroTraceVerbose("retTpe is: ")(retTpe) - macroTraceVerbose("macroImplSig is: ")((paramss, implReturnType)) + macroLogVerbose(sm""" + |generating macroImplSigs for: $macroDef + |tparams are: $tparams + |vparamss are: $vparamss + |retTpe is: $retTpe + |macroImplSig is: $paramss, $implReturnType + """.trim) + (paramss, implReturnType) } /** Verifies that the body of a macro def typechecks to a reference to a static public non-overloaded method, @@ -380,7 +383,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { // doesn't manifest itself as an error in the resulting tree val prevNumErrors = reporter.ERROR.count var rhs1 = typer.typed1(rhs, EXPRmode, WildcardType) - def rhsNeedsMacroExpansion = rhs1.symbol != null && rhs1.symbol.isMacro && !rhs1.symbol.isErroneous + def rhsNeedsMacroExpansion = rhs1.symbol != null && rhs1.symbol.isTermMacro && !rhs1.symbol.isErroneous while (rhsNeedsMacroExpansion) { rhs1 = macroExpand1(typer, rhs1) match { case Success(expanded) => @@ -414,6 +417,9 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { // Phase II: typecheck the right-hand side of the macro def val typed = typecheckRhs(macroDdef.rhs) typed match { + case MacroImplReference(_, meth, _) if meth == Predef_??? => + bindMacroImpl(macroDef, typed) + MacroDefIsQmarkQmarkQmark() case MacroImplReference(owner, meth, targs) => if (!meth.isMethod) MacroDefInvalidBodyError() if (!meth.isPublic) MacroImplNotPublicError() @@ -495,7 +501,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { type MacroRuntime = MacroArgs => Any private val macroRuntimesCache = perRunCaches.newWeakMap[Symbol, MacroRuntime]() private def macroRuntime(macroDef: Symbol): MacroRuntime = { - macroTraceVerbose("looking for macro implementation: ")(macroDef) + macroLogVerbose(s"looking for macro implementation: $macroDef") if (fastTrack contains macroDef) { macroLogVerbose("macro expansion is serviced by a fast track") fastTrack(macroDef) @@ -506,26 +512,30 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { val methName = binding.methName macroLogVerbose(s"resolved implementation as $className.$methName") - // I don't use Scala reflection here, because it seems to interfere with JIT magic - // whenever you instantiate a mirror (and not do anything with in, just instantiate), performance drops by 15-20% - // I'm not sure what's the reason - for me it's pure voodoo - // upd. my latest experiments show that everything's okay - // it seems that in 2.10.1 we can easily switch to Scala reflection - try { - macroTraceVerbose("loading implementation class: ")(className) - macroTraceVerbose("classloader is: ")(ReflectionUtils.show(macroClassloader)) - val implObj = ReflectionUtils.staticSingletonInstance(macroClassloader, className) - // relies on the fact that macro impls cannot be overloaded - // so every methName can resolve to at maximum one method - val implMeths = implObj.getClass.getDeclaredMethods.find(_.getName == methName) - val implMeth = implMeths getOrElse { throw new NoSuchMethodException(s"$className.$methName") } - macroLogVerbose("successfully loaded macro impl as (%s, %s)".format(implObj, implMeth)) - args => implMeth.invoke(implObj, ((args.c +: args.others) map (_.asInstanceOf[AnyRef])): _*) - } catch { - case ex: Exception => - macroTraceVerbose(s"macro runtime failed to load: ")(ex.toString) - macroDef setFlag IS_ERROR - null + if (binding.className == Predef_???.owner.fullName.toString && binding.methName == Predef_???.name.encoded) { + args => throw new AbortMacroException(args.c.enclosingPosition, "macro implementation is missing") + } else { + // I don't use Scala reflection here, because it seems to interfere with JIT magic + // whenever you instantiate a mirror (and not do anything with in, just instantiate), performance drops by 15-20% + // I'm not sure what's the reason - for me it's pure voodoo + // upd. my latest experiments show that everything's okay + // it seems that in 2.10.1 we can easily switch to Scala reflection + try { + macroLogVerbose(s"loading implementation class: $className") + macroLogVerbose(s"classloader is: ${ReflectionUtils.show(macroClassloader)}") + val implObj = ReflectionUtils.staticSingletonInstance(macroClassloader, className) + // relies on the fact that macro impls cannot be overloaded + // so every methName can resolve to at maximum one method + val implMeths = implObj.getClass.getDeclaredMethods.find(_.getName == methName) + val implMeth = implMeths getOrElse { throw new NoSuchMethodException(s"$className.$methName") } + macroLogVerbose(s"successfully loaded macro impl as ($implObj, $implMeth)") + args => implMeth.invoke(implObj, ((args.c +: args.others) map (_.asInstanceOf[AnyRef])): _*) + } catch { + case ex: Exception => + macroLogVerbose(s"macro runtime failed to load: ${ex.toString}") + macroDef setFlag IS_ERROR + null + } } }) } @@ -570,8 +580,8 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { if (argcDoesntMatch && !nullaryArgsEmptyParams) { typer.TyperErrorGen.MacroPartialApplicationError(expandee) } val argss: List[List[Any]] = exprArgs.toList - macroTraceVerbose("context: ")(context) - macroTraceVerbose("argss: ")(argss) + macroLogVerbose(s"context: $context") + macroLogVerbose(s"argss: $argss") val preparedArgss: List[List[Any]] = if (fastTrack contains macroDef) { @@ -597,7 +607,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { // whereas V won't be resolved by asSeenFrom and need to be loaded directly from `expandee` which needs to contain a TypeApply node // also, macro implementation reference may contain a regular type as a type argument, then we pass it verbatim val binding = loadMacroImplBinding(macroDef) - macroTraceVerbose("binding: ")(binding) + macroLogVerbose(s"binding: $binding") val tags = binding.signature filter (_ != -1) map (paramPos => { val targ = binding.targs(paramPos).tpe.typeSymbol val tpe = if (targ.isTypeParameterOrSkolem) { @@ -615,7 +625,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { targ.tpe context.WeakTypeTag(tpe) }) - macroTraceVerbose("tags: ")(tags) + macroLogVerbose(s"tags: $tags") // transforms argss taking into account varargness of paramss // note that typetag context bounds are only declared on macroImpls @@ -632,7 +642,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { } else as }) } - macroTraceVerbose("preparedArgss: ")(preparedArgss) + macroLogVerbose(s"preparedArgss: $preparedArgss") MacroArgs(context, preparedArgss.flatten) } @@ -825,7 +835,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { withInfoLevel(nodePrinters.InfoLevel.Quiet) { if (expandee.symbol.isErroneous || (expandee exists (_.isErroneous))) { val reason = if (expandee.symbol.isErroneous) "not found or incompatible macro implementation" else "erroneous arguments" - macroTraceVerbose("cancelled macro expansion because of %s: ".format(reason))(expandee) + macroLogVerbose(s"cancelled macro expansion because of $reason: $expandee") Cancel(typer.infer.setError(expandee)) } else try { @@ -900,7 +910,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { private def macroExpandWithoutRuntime(typer: Typer, expandee: Tree): MacroStatus = { import typer.TyperErrorGen._ val fallbackSym = expandee.symbol.nextOverriddenSymbol orElse MacroImplementationNotFoundError(expandee) - macroTraceLite("falling back to: ")(fallbackSym) + macroLogLite(s"falling back to: $fallbackSym") def mkFallbackTree(tree: Tree): Tree = { tree match { @@ -952,7 +962,7 @@ trait Macros extends scala.tools.reflect.FastTrack with Traces { undetparams --= undetNoMore map (_.id) if (undetparams.isEmpty) { hasPendingMacroExpansions = true - macroTraceVerbose("macro expansion is pending: ")(expandee) + macroLogVerbose(s"macro expansion is pending: $expandee") } case _ => // do nothing diff --git a/src/compiler/scala/tools/nsc/typechecker/RefChecks.scala b/src/compiler/scala/tools/nsc/typechecker/RefChecks.scala index bbba786c0c..53f296f801 100644 --- a/src/compiler/scala/tools/nsc/typechecker/RefChecks.scala +++ b/src/compiler/scala/tools/nsc/typechecker/RefChecks.scala @@ -406,7 +406,7 @@ abstract class RefChecks extends InfoTransform with scala.reflect.internal.trans overrideError("cannot be used here - classes can only override abstract types") } else if (other.isEffectivelyFinal) { // (1.2) overrideError("cannot override final member") - } else if (!other.isDeferred && !member.isAnyOverride && !member.isSynthetic) { // (*) + } else if (!other.isDeferred && !other.hasFlag(DEFAULTMETHOD) && !member.isAnyOverride && !member.isSynthetic) { // (*) // (*) Synthetic exclusion for (at least) default getters, fixes SI-5178. We cannot assign the OVERRIDE flag to // the default getter: one default getter might sometimes override, sometimes not. Example in comment on ticket. if (isNeitherInClass && !(other.owner isSubClass member.owner)) @@ -945,9 +945,13 @@ abstract class RefChecks extends InfoTransform with scala.reflect.internal.trans case _ => } + private def isObjectOrAnyComparisonMethod(sym: Symbol) = sym match { + case Object_eq | Object_ne | Object_== | Object_!= | Any_== | Any_!= => true + case _ => false + } def checkSensible(pos: Position, fn: Tree, args: List[Tree]) = fn match { - case Select(qual, name @ (nme.EQ | nme.NE | nme.eq | nme.ne)) if args.length == 1 => - def isReferenceOp = name == nme.eq || name == nme.ne + case Select(qual, name @ (nme.EQ | nme.NE | nme.eq | nme.ne)) if args.length == 1 && isObjectOrAnyComparisonMethod(fn.symbol) => + def isReferenceOp = fn.symbol == Object_eq || fn.symbol == Object_ne def isNew(tree: Tree) = tree match { case Function(_, _) | Apply(Select(New(_), nme.CONSTRUCTOR), _) => true diff --git a/src/compiler/scala/tools/nsc/typechecker/SuperAccessors.scala b/src/compiler/scala/tools/nsc/typechecker/SuperAccessors.scala index 3f7e6d7665..4e78cecd7d 100644 --- a/src/compiler/scala/tools/nsc/typechecker/SuperAccessors.scala +++ b/src/compiler/scala/tools/nsc/typechecker/SuperAccessors.scala @@ -82,32 +82,11 @@ abstract class SuperAccessors extends transform.Transform with transform.TypingT private def transformArgs(params: List[Symbol], args: List[Tree]) = { treeInfo.mapMethodParamsAndArgs(params, args) { (param, arg) => if (isByNameParamType(param.tpe)) - withInvalidOwner { checkPackedConforms(transform(arg), param.tpe.typeArgs.head) } + withInvalidOwner(transform(arg)) else transform(arg) } } - private def checkPackedConforms(tree: Tree, pt: Type): Tree = { - def typeError(typer: analyzer.Typer, pos: Position, found: Type, req: Type) { - if (!found.isErroneous && !req.isErroneous) { - val msg = analyzer.ErrorUtils.typeErrorMsg(found, req, typer.infer.isPossiblyMissingArgs(found, req)) - typer.context.error(pos, analyzer.withAddendum(pos)(msg)) - if (settings.explaintypes) - explainTypes(found, req) - } - } - - if (tree.tpe exists (_.typeSymbol.isExistentialSkolem)) { - val packed = localTyper.packedType(tree, NoSymbol) - if (!(packed <:< pt)) { - val errorContext = localTyper.context.make(localTyper.context.tree) - errorContext.setReportErrors() - typeError(analyzer.newTyper(errorContext), tree.pos, packed, pt) - } - } - tree - } - /** Check that a class and its companion object to not both define * a class or module with same name */ diff --git a/src/compiler/scala/tools/nsc/typechecker/Typers.scala b/src/compiler/scala/tools/nsc/typechecker/Typers.scala index bb1dc10f0b..bf5de1548d 100644 --- a/src/compiler/scala/tools/nsc/typechecker/Typers.scala +++ b/src/compiler/scala/tools/nsc/typechecker/Typers.scala @@ -4852,19 +4852,23 @@ trait Typers extends Adaptations with Tags { } def typedAppliedTypeTree(tree: AppliedTypeTree) = { - val tpt = tree.tpt - val args = tree.args - val tpt1 = typed1(tpt, mode | FUNmode | TAPPmode, WildcardType) + val tpt = tree.tpt + val args = tree.args + val tpt1 = typed1(tpt, mode | FUNmode | TAPPmode, WildcardType) + def isPoly = tpt1.tpe.isInstanceOf[PolyType] + def isComplete = tpt1.symbol.rawInfo.isComplete + if (tpt1.isErrorTyped) { tpt1 } else if (!tpt1.hasSymbolField) { AppliedTypeNoParametersError(tree, tpt1.tpe) } else { val tparams = tpt1.symbol.typeParams + if (sameLength(tparams, args)) { // @M: kind-arity checking is done here and in adapt, full kind-checking is in checkKindBounds (in Infer) val args1 = - if (!tpt1.symbol.rawInfo.isComplete) + if (!isComplete) args mapConserve (typedHigherKindedType(_, mode)) // if symbol hasn't been fully loaded, can't check kind-arity else map2Conserve(args, tparams) { (arg, tparam) => @@ -4873,19 +4877,40 @@ trait Typers extends Adaptations with Tags { } val argtypes = args1 map (_.tpe) - foreach2(args, tparams)((arg, tparam) => arg match { - // note: can't use args1 in selector, because Bind's got replaced - case Bind(_, _) => - if (arg.symbol.isAbstractType) - arg.symbol setInfo // XXX, feedback. don't trackSymInfo here! - TypeBounds( - lub(List(arg.symbol.info.bounds.lo, tparam.info.bounds.lo.subst(tparams, argtypes))), - glb(List(arg.symbol.info.bounds.hi, tparam.info.bounds.hi.subst(tparams, argtypes)))) - case _ => - }) + foreach2(args, tparams) { (arg, tparam) => + // note: can't use args1 in selector, because Binds got replaced + val asym = arg.symbol + def abounds = asym.info.bounds + def tbounds = tparam.info.bounds + def enhanceBounds(): Unit = { + val TypeBounds(lo0, hi0) = abounds + val TypeBounds(lo1, hi1) = tbounds.subst(tparams, argtypes) + val lo = lub(List(lo0, lo1)) + val hi = glb(List(hi0, hi1)) + if (!(lo =:= lo0 && hi =:= hi0)) + asym setInfo logResult(s"Updating bounds of ${asym.fullLocationString} in $tree from '$abounds' to")(TypeBounds(lo, hi)) + } + if (asym != null && asym.isAbstractType) { + // See pos/t1786 to follow what's happening here. + def canEnhanceIdent = ( + asym.hasCompleteInfo + && tparam.exists /* sometimes it is NoSymbol */ + && tparam.hasCompleteInfo /* SI-2940 */ + && !tparam.isFBounded /* SI-2251 */ + && !tparam.isHigherOrderTypeParameter + && !(abounds.hi <:< tbounds.hi) + && asym.isSynthetic /* this limits us to placeholder tparams, excluding named ones */ + ) + arg match { + case Bind(_, _) => enhanceBounds() + case Ident(name) if canEnhanceIdent => enhanceBounds() + case _ => + } + } + } val original = treeCopy.AppliedTypeTree(tree, tpt1, args1) val result = TypeTree(appliedType(tpt1.tpe, argtypes)) setOriginal original - if(tpt1.tpe.isInstanceOf[PolyType]) // did the type application (performed by appliedType) involve an unchecked beta-reduction? + if (isPoly) // did the type application (performed by appliedType) involve an unchecked beta-reduction? TypeTreeWithDeferredRefCheck(){ () => // wrap the tree and include the bounds check -- refchecks will perform this check (that the beta reduction was indeed allowed) and unwrap // we can't simply use original in refchecks because it does not contains types @@ -5094,10 +5119,9 @@ trait Typers extends Adaptations with Tags { def typedApplyDynamic(tree: ApplyDynamic) = { assert(phase.erasedTypes) - val reflectiveCalls = !(settings.refinementMethodDispatch.value == "invoke-dynamic") val qual1 = typed(tree.qual, AnyRefClass.tpe) - val args1 = tree.args mapConserve (arg => if (reflectiveCalls) typed(arg, AnyRefClass.tpe) else typed(arg)) - treeCopy.ApplyDynamic(tree, qual1, args1) setType (if (reflectiveCalls) AnyRefClass.tpe else tree.symbol.info.resultType) + val args1 = tree.args mapConserve (arg => typed(arg, AnyRefClass.tpe)) + treeCopy.ApplyDynamic(tree, qual1, args1) setType AnyRefClass.tpe } def typedReferenceToBoxed(tree: ReferenceToBoxed) = { @@ -5456,7 +5480,7 @@ trait Typers extends Adaptations with Tags { tree1 } - val isMacroBodyOkay = !tree.symbol.isErroneous && !(tree1 exists (_.isErroneous)) + val isMacroBodyOkay = !tree.symbol.isErroneous && !(tree1 exists (_.isErroneous)) && tree1 != EmptyTree val shouldInheritMacroImplReturnType = ddef.tpt.isEmpty if (isMacroBodyOkay && shouldInheritMacroImplReturnType) computeMacroDefTypeFromMacroImpl(ddef, tree1.symbol) else AnyClass.tpe } diff --git a/src/compiler/scala/tools/reflect/MacroImplementations.scala b/src/compiler/scala/tools/reflect/MacroImplementations.scala index 4e8f02084d..109c148b7e 100644 --- a/src/compiler/scala/tools/reflect/MacroImplementations.scala +++ b/src/compiler/scala/tools/reflect/MacroImplementations.scala @@ -3,6 +3,7 @@ package scala.tools.reflect import scala.reflect.macros.runtime.Context import scala.collection.mutable.ListBuffer import scala.collection.mutable.Stack +import scala.reflect.internal.util.OffsetPosition abstract class MacroImplementations { val c: Context @@ -25,16 +26,12 @@ abstract class MacroImplementations { c.abort(args(parts.length-1).pos, "too many arguments for interpolated string") } - val stringParts = parts map { - case Literal(Constant(s: String)) => s - case _ => throw new IllegalArgumentException("argument parts must be a list of string literals") - } - val pi = stringParts.iterator + val pi = parts.iterator val bldr = new java.lang.StringBuilder val evals = ListBuffer[ValDef]() val ids = ListBuffer[Ident]() - val argsStack = Stack(args : _*) + val argStack = Stack(args : _*) def defval(value: Tree, tpe: Type): Unit = { val freshName = newTermName(c.freshName("arg$")) @@ -81,50 +78,73 @@ abstract class MacroImplementations { } def copyString(first: Boolean): Unit = { - val str = StringContext.treatEscapes(pi.next()) + val strTree = pi.next() + val rawStr = strTree match { + case Literal(Constant(str: String)) => str + case _ => throw new IllegalArgumentException("internal error: argument parts must be a list of string literals") + } + val str = StringContext.treatEscapes(rawStr) val strLen = str.length val strIsEmpty = strLen == 0 - var start = 0 + def charAtIndexIs(idx: Int, ch: Char) = idx < strLen && str(idx) == ch + def isPercent(idx: Int) = charAtIndexIs(idx, '%') + def isConversion(idx: Int) = isPercent(idx) && !charAtIndexIs(idx + 1, 'n') && !charAtIndexIs(idx + 1, '%') var idx = 0 + def errorAtIndex(idx: Int, msg: String) = c.error(new OffsetPosition(strTree.pos.source, strTree.pos.point + idx), msg) + def wrongConversionString(idx: Int) = errorAtIndex(idx, "wrong conversion string") + def illegalConversionCharacter(idx: Int) = errorAtIndex(idx, "illegal conversion character") + def nonEscapedPercent(idx: Int) = errorAtIndex(idx, "percent signs not directly following splicees must be escaped") + + // STEP 1: handle argument conversion + // 1) "...${smth}" => okay, equivalent to "...${smth}%s" + // 2) "...${smth}blahblah" => okay, equivalent to "...${smth}%sblahblah" + // 3) "...${smth}%" => error + // 4) "...${smth}%n" => okay, equivalent to "...${smth}%s%n" + // 5) "...${smth}%%" => okay, equivalent to "...${smth}%s%%" + // 6) "...${smth}[%legalJavaConversion]" => okay, according to http://docs.oracle.com/javase/1.5.0/docs/api/java/util/Formatter.html + // 7) "...${smth}[%illegalJavaConversion]" => error if (!first) { - val arg = argsStack.pop() - if (strIsEmpty || (str charAt 0) != '%') { - bldr append "%s" - defval(arg, AnyTpe) - } else { + val arg = argStack.pop() + if (isConversion(0)) { // PRE str is not empty and str(0) == '%' // argument index parameter is not allowed, thus parse // [flags][width][.precision]conversion var pos = 1 - while(pos < strLen && isFlag(str charAt pos)) pos += 1 - while(pos < strLen && Character.isDigit(str charAt pos)) pos += 1 - if(pos < strLen && str.charAt(pos) == '.') { pos += 1 - while(pos < strLen && Character.isDigit(str charAt pos)) pos += 1 + while (pos < strLen && isFlag(str charAt pos)) pos += 1 + while (pos < strLen && Character.isDigit(str charAt pos)) pos += 1 + if (pos < strLen && str.charAt(pos) == '.') { + pos += 1 + while (pos < strLen && Character.isDigit(str charAt pos)) pos += 1 } - if(pos < strLen) { + if (pos < strLen) { conversionType(str charAt pos, arg) match { case Some(tpe) => defval(arg, tpe) - case None => c.error(arg.pos, "illegal conversion character") + case None => illegalConversionCharacter(pos) } } else { - // TODO: place error message on conversion string - c.error(arg.pos, "wrong conversion string") + wrongConversionString(pos - 1) } + idx = 1 + } else { + bldr append "%s" + defval(arg, AnyTpe) } - idx = 1 } + + // STEP 2: handle the rest of the text + // 1) %n tokens are left as is + // 2) %% tokens are left as is + // 3) other usages of percents are reported as errors if (!strIsEmpty) { - val len = str.length - while (idx < len) { - def notPercentN = str(idx) != '%' || (idx + 1 < len && str(idx + 1) != 'n') - if (str(idx) == '%' && notPercentN) { - bldr append (str substring (start, idx)) append "%%" - start = idx + 1 + while (idx < strLen) { + if (isPercent(idx)) { + if (isConversion(idx)) nonEscapedPercent(idx) + else idx += 1 // skip n and % in %n and %% } idx += 1 } - bldr append (str substring (start, idx)) + bldr append (str take idx) } } diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java index 8dbca6da4b..6578504155 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java @@ -5,12 +5,12 @@ */ package scala.concurrent.forkjoin; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; -import java.util.Random; import java.util.concurrent.AbstractExecutorService; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; @@ -18,10 +18,357 @@ import java.util.concurrent.Future; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RunnableFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.AbstractQueuedSynchronizer; -import java.util.concurrent.locks.Condition; + +/** + * @since 1.8 + * @author Doug Lea + */ +/*public*/ abstract class CountedCompleter<T> extends ForkJoinTask<T> { + private static final long serialVersionUID = 5232453752276485070L; + + /** This task's completer, or null if none */ + final CountedCompleter<?> completer; + /** The number of pending tasks until completion */ + volatile int pending; + + /** + * Creates a new CountedCompleter with the given completer + * and initial pending count. + * + * @param completer this task's completer, or {@code null} if none + * @param initialPendingCount the initial pending count + */ + protected CountedCompleter(CountedCompleter<?> completer, + int initialPendingCount) { + this.completer = completer; + this.pending = initialPendingCount; + } + + /** + * Creates a new CountedCompleter with the given completer + * and an initial pending count of zero. + * + * @param completer this task's completer, or {@code null} if none + */ + protected CountedCompleter(CountedCompleter<?> completer) { + this.completer = completer; + } + + /** + * Creates a new CountedCompleter with no completer + * and an initial pending count of zero. + */ + protected CountedCompleter() { + this.completer = null; + } + + /** + * The main computation performed by this task. + */ + public abstract void compute(); + + /** + * Performs an action when method {@link #tryComplete} is invoked + * and the pending count is zero, or when the unconditional + * method {@link #complete} is invoked. By default, this method + * does nothing. You can distinguish cases by checking the + * identity of the given caller argument. If not equal to {@code + * this}, then it is typically a subtask that may contain results + * (and/or links to other results) to combine. + * + * @param caller the task invoking this method (which may + * be this task itself) + */ + public void onCompletion(CountedCompleter<?> caller) { + } + + /** + * Performs an action when method {@link #completeExceptionally} + * is invoked or method {@link #compute} throws an exception, and + * this task has not otherwise already completed normally. On + * entry to this method, this task {@link + * ForkJoinTask#isCompletedAbnormally}. The return value of this + * method controls further propagation: If {@code true} and this + * task has a completer, then this completer is also completed + * exceptionally. The default implementation of this method does + * nothing except return {@code true}. + * + * @param ex the exception + * @param caller the task invoking this method (which may + * be this task itself) + * @return true if this exception should be propagated to this + * task's completer, if one exists + */ + public boolean onExceptionalCompletion(Throwable ex, CountedCompleter<?> caller) { + return true; + } + + /** + * Returns the completer established in this task's constructor, + * or {@code null} if none. + * + * @return the completer + */ + public final CountedCompleter<?> getCompleter() { + return completer; + } + + /** + * Returns the current pending count. + * + * @return the current pending count + */ + public final int getPendingCount() { + return pending; + } + + /** + * Sets the pending count to the given value. + * + * @param count the count + */ + public final void setPendingCount(int count) { + pending = count; + } + + /** + * Adds (atomically) the given value to the pending count. + * + * @param delta the value to add + */ + public final void addToPendingCount(int delta) { + int c; // note: can replace with intrinsic in jdk8 + do {} while (!U.compareAndSwapInt(this, PENDING, c = pending, c+delta)); + } + + /** + * Sets (atomically) the pending count to the given count only if + * it currently holds the given expected value. + * + * @param expected the expected value + * @param count the new value + * @return true if successful + */ + public final boolean compareAndSetPendingCount(int expected, int count) { + return U.compareAndSwapInt(this, PENDING, expected, count); + } + + /** + * If the pending count is nonzero, (atomically) decrements it. + * + * @return the initial (undecremented) pending count holding on entry + * to this method + */ + public final int decrementPendingCountUnlessZero() { + int c; + do {} while ((c = pending) != 0 && + !U.compareAndSwapInt(this, PENDING, c, c - 1)); + return c; + } + + /** + * Returns the root of the current computation; i.e., this + * task if it has no completer, else its completer's root. + * + * @return the root of the current computation + */ + public final CountedCompleter<?> getRoot() { + CountedCompleter<?> a = this, p; + while ((p = a.completer) != null) + a = p; + return a; + } + + /** + * If the pending count is nonzero, decrements the count; + * otherwise invokes {@link #onCompletion} and then similarly + * tries to complete this task's completer, if one exists, + * else marks this task as complete. + */ + public final void tryComplete() { + CountedCompleter<?> a = this, s = a; + for (int c;;) { + if ((c = a.pending) == 0) { + a.onCompletion(s); + if ((a = (s = a).completer) == null) { + s.quietlyComplete(); + return; + } + } + else if (U.compareAndSwapInt(a, PENDING, c, c - 1)) + return; + } + } + + /** + * Equivalent to {@link #tryComplete} but does not invoke {@link + * #onCompletion} along the completion path: If the pending count + * is nonzero, decrements the count; otherwise, similarly tries to + * complete this task's completer, if one exists, else marks this + * task as complete. This method may be useful in cases where + * {@code onCompletion} should not, or need not, be invoked for + * each completer in a computation. + */ + public final void propagateCompletion() { + CountedCompleter<?> a = this, s = a; + for (int c;;) { + if ((c = a.pending) == 0) { + if ((a = (s = a).completer) == null) { + s.quietlyComplete(); + return; + } + } + else if (U.compareAndSwapInt(a, PENDING, c, c - 1)) + return; + } + } + + /** + * Regardless of pending count, invokes {@link #onCompletion}, + * marks this task as complete and further triggers {@link + * #tryComplete} on this task's completer, if one exists. The + * given rawResult is used as an argument to {@link #setRawResult} + * before invoking {@link #onCompletion} or marking this task as + * complete; its value is meaningful only for classes overriding + * {@code setRawResult}. + * + * <p>This method may be useful when forcing completion as soon as + * any one (versus all) of several subtask results are obtained. + * However, in the common (and recommended) case in which {@code + * setRawResult} is not overridden, this effect can be obtained + * more simply using {@code quietlyCompleteRoot();}. + * + * @param rawResult the raw result + */ + public void complete(T rawResult) { + CountedCompleter<?> p; + setRawResult(rawResult); + onCompletion(this); + quietlyComplete(); + if ((p = completer) != null) + p.tryComplete(); + } + + + /** + * If this task's pending count is zero, returns this task; + * otherwise decrements its pending count and returns {@code + * null}. This method is designed to be used with {@link + * #nextComplete} in completion traversal loops. + * + * @return this task, if pending count was zero, else {@code null} + */ + public final CountedCompleter<?> firstComplete() { + for (int c;;) { + if ((c = pending) == 0) + return this; + else if (U.compareAndSwapInt(this, PENDING, c, c - 1)) + return null; + } + } + + /** + * If this task does not have a completer, invokes {@link + * ForkJoinTask#quietlyComplete} and returns {@code null}. Or, if + * this task's pending count is non-zero, decrements its pending + * count and returns {@code null}. Otherwise, returns the + * completer. This method can be used as part of a completion + * traversal loop for homogeneous task hierarchies: + * + * <pre> {@code + * for (CountedCompleter<?> c = firstComplete(); + * c != null; + * c = c.nextComplete()) { + * // ... process c ... + * }}</pre> + * + * @return the completer, or {@code null} if none + */ + public final CountedCompleter<?> nextComplete() { + CountedCompleter<?> p; + if ((p = completer) != null) + return p.firstComplete(); + else { + quietlyComplete(); + return null; + } + } + + /** + * Equivalent to {@code getRoot().quietlyComplete()}. + */ + public final void quietlyCompleteRoot() { + for (CountedCompleter<?> a = this, p;;) { + if ((p = a.completer) == null) { + a.quietlyComplete(); + return; + } + a = p; + } + } + + /** + * Supports ForkJoinTask exception propagation. + */ + void internalPropagateException(Throwable ex) { + CountedCompleter<?> a = this, s = a; + while (a.onExceptionalCompletion(ex, s) && + (a = (s = a).completer) != null && a.status >= 0) + a.recordExceptionalCompletion(ex); + } + + /** + * Implements execution conventions for CountedCompleters. + */ + protected final boolean exec() { + compute(); + return false; + } + + /** + * Returns the result of the computation. By default + * returns {@code null}, which is appropriate for {@code Void} + * actions, but in other cases should be overridden, almost + * always to return a field or function of a field that + * holds the result upon completion. + * + * @return the result of the computation + */ + public T getRawResult() { return null; } + + /** + * A method that result-bearing CountedCompleters may optionally + * use to help maintain result data. By default, does nothing. + * Overrides are not recommended. However, if this method is + * overridden to update existing objects or fields, then it must + * in general be defined to be thread-safe. + */ + protected void setRawResult(T t) { } + + // Unsafe mechanics + private static final sun.misc.Unsafe U; + private static final long PENDING; + static { + try { + U = getUnsafe(); + PENDING = U.objectFieldOffset + (CountedCompleter.class.getDeclaredField("pending")); + } catch (Exception e) { + throw new Error(e); + } + } + + /** + * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. + * Replace with a simple call to Unsafe.getUnsafe when integrating + * into a jdk. + * + * @return a sun.misc.Unsafe + */ + private static sun.misc.Unsafe getUnsafe() { + return scala.concurrent.util.Unsafe.instance; + } +} /** * An {@link ExecutorService} for running {@link ForkJoinTask}s. @@ -41,14 +388,22 @@ import java.util.concurrent.locks.Condition; * ForkJoinPool}s may also be appropriate for use with event-style * tasks that are never joined. * - * <p>A {@code ForkJoinPool} is constructed with a given target - * parallelism level; by default, equal to the number of available - * processors. The pool attempts to maintain enough active (or - * available) threads by dynamically adding, suspending, or resuming - * internal worker threads, even if some tasks are stalled waiting to - * join others. However, no such adjustments are guaranteed in the - * face of blocked IO or other unmanaged synchronization. The nested - * {@link ManagedBlocker} interface enables extension of the kinds of + * <p>A static {@link #commonPool()} is available and appropriate for + * most applications. The common pool is used by any ForkJoinTask that + * is not explicitly submitted to a specified pool. Using the common + * pool normally reduces resource usage (its threads are slowly + * reclaimed during periods of non-use, and reinstated upon subsequent + * use). + * + * <p>For applications that require separate or custom pools, a {@code + * ForkJoinPool} may be constructed with a given target parallelism + * level; by default, equal to the number of available processors. The + * pool attempts to maintain enough active (or available) threads by + * dynamically adding, suspending, or resuming internal worker + * threads, even if some tasks are stalled waiting to join + * others. However, no such adjustments are guaranteed in the face of + * blocked I/O or other unmanaged synchronization. The nested {@link + * ManagedBlocker} interface enables extension of the kinds of * synchronization accommodated. * * <p>In addition to execution and lifecycle control methods, this @@ -58,7 +413,7 @@ import java.util.concurrent.locks.Condition; * {@link #toString} returns indications of pool state in a * convenient form for informal monitoring. * - * <p> As is the case with other ExecutorServices, there are three + * <p>As is the case with other ExecutorServices, there are three * main task execution methods summarized in the following table. * These are designed to be used primarily by clients not already * engaged in fork/join computations in the current pool. The main @@ -93,22 +448,16 @@ import java.util.concurrent.locks.Condition; * </tr> * </table> * - * <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is - * used for all parallel task execution in a program or subsystem. - * Otherwise, use would not usually outweigh the construction and - * bookkeeping overhead of creating a large set of threads. For - * example, a common pool could be used for the {@code SortTasks} - * illustrated in {@link RecursiveAction}. Because {@code - * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon - * daemon} mode, there is typically no need to explicitly {@link - * #shutdown} such a pool upon program exit. - * - * <pre> {@code - * static final ForkJoinPool mainPool = new ForkJoinPool(); - * ... - * public void sort(long[] array) { - * mainPool.invoke(new SortTask(array, 0, array.length)); - * }}</pre> + * <p>The common pool is by default constructed with default + * parameters, but these may be controlled by setting three {@link + * System#getProperty system properties} with prefix {@code + * java.util.concurrent.ForkJoinPool.common}: {@code parallelism} -- + * an integer greater than zero, {@code threadFactory} -- the class + * name of a {@link ForkJoinWorkerThreadFactory}, and {@code + * exceptionHandler} -- the class name of a {@link + * java.lang.Thread.UncaughtExceptionHandler + * Thread.UncaughtExceptionHandler}. Upon any error in establishing + * these settings, default parameters are used. * * <p><b>Implementation notes</b>: This implementation restricts the * maximum number of running threads to 32767. Attempts to create @@ -196,21 +545,24 @@ public class ForkJoinPool extends AbstractExecutorService { * WorkQueues are also used in a similar way for tasks submitted * to the pool. We cannot mix these tasks in the same queues used * for work-stealing (this would contaminate lifo/fifo - * processing). Instead, we loosely associate submission queues + * processing). Instead, we randomly associate submission queues * with submitting threads, using a form of hashing. The * ThreadLocal Submitter class contains a value initially used as * a hash code for choosing existing queues, but may be randomly * repositioned upon contention with other submitters. In - * essence, submitters act like workers except that they never - * take tasks, and they are multiplexed on to a finite number of - * shared work queues. However, classes are set up so that future - * extensions could allow submitters to optionally help perform - * tasks as well. Insertion of tasks in shared mode requires a - * lock (mainly to protect in the case of resizing) but we use - * only a simple spinlock (using bits in field runState), because - * submitters encountering a busy queue move on to try or create - * other queues -- they block only when creating and registering - * new queues. + * essence, submitters act like workers except that they are + * restricted to executing local tasks that they submitted (or in + * the case of CountedCompleters, others with the same root task). + * However, because most shared/external queue operations are more + * expensive than internal, and because, at steady state, external + * submitters will compete for CPU with workers, ForkJoinTask.join + * and related methods disable them from repeatedly helping to + * process tasks if all workers are active. Insertion of tasks in + * shared mode requires a lock (mainly to protect in the case of + * resizing) but we use only a simple spinlock (using bits in + * field qlock), because submitters encountering a busy queue move + * on to try or create other queues -- they block only when + * creating and registering new queues. * * Management * ========== @@ -232,23 +584,26 @@ public class ForkJoinPool extends AbstractExecutorService { * and their negations (used for thresholding) to fit into 16bit * fields. * - * Field "runState" contains 32 bits needed to register and - * deregister WorkQueues, as well as to enable shutdown. It is - * only modified under a lock (normally briefly held, but - * occasionally protecting allocations and resizings) but even - * when locked remains available to check consistency. + * Field "plock" is a form of sequence lock with a saturating + * shutdown bit (similarly for per-queue "qlocks"), mainly + * protecting updates to the workQueues array, as well as to + * enable shutdown. When used as a lock, it is normally only very + * briefly held, so is nearly always available after at most a + * brief spin, but we use a monitor-based backup strategy to + * block when needed. * * Recording WorkQueues. WorkQueues are recorded in the - * "workQueues" array that is created upon pool construction and - * expanded if necessary. Updates to the array while recording - * new workers and unrecording terminated ones are protected from - * each other by a lock but the array is otherwise concurrently - * readable, and accessed directly. To simplify index-based - * operations, the array size is always a power of two, and all - * readers must tolerate null slots. Shared (submission) queues - * are at even indices, worker queues at odd indices. Grouping - * them together in this way simplifies and speeds up task - * scanning. + * "workQueues" array that is created upon first use and expanded + * if necessary. Updates to the array while recording new workers + * and unrecording terminated ones are protected from each other + * by a lock but the array is otherwise concurrently readable, and + * accessed directly. To simplify index-based operations, the + * array size is always a power of two, and all readers must + * tolerate null slots. Worker queues are at odd indices. Shared + * (submission) queues are at even indices, up to a maximum of 64 + * slots, to limit growth even if array needs to expand to add + * more workers. Grouping them together in this way simplifies and + * speeds up task scanning. * * All worker thread creation is on-demand, triggered by task * submissions, replacement of terminated workers, and/or @@ -309,24 +664,33 @@ public class ForkJoinPool extends AbstractExecutorService { * * Signalling. We create or wake up workers only when there * appears to be at least one task they might be able to find and - * execute. When a submission is added or another worker adds a - * task to a queue that previously had fewer than two tasks, they - * signal waiting workers (or trigger creation of new ones if - * fewer than the given parallelism level -- see signalWork). - * These primary signals are buttressed by signals during rescans; - * together these cover the signals needed in cases when more - * tasks are pushed but untaken, and improve performance compared - * to having one thread wake up all workers. + * execute. However, many other threads may notice the same task + * and each signal to wake up a thread that might take it. So in + * general, pools will be over-signalled. When a submission is + * added or another worker adds a task to a queue that has fewer + * than two tasks, they signal waiting workers (or trigger + * creation of new ones if fewer than the given parallelism level + * -- signalWork), and may leave a hint to the unparked worker to + * help signal others upon wakeup). These primary signals are + * buttressed by others (see method helpSignal) whenever other + * threads scan for work or do not have a task to process. On + * most platforms, signalling (unpark) overhead time is noticeably + * long, and the time between signalling a thread and it actually + * making progress can be very noticeably long, so it is worth + * offloading these delays from critical paths as much as + * possible. * * Trimming workers. To release resources after periods of lack of * use, a worker starting to wait when the pool is quiescent will - * time out and terminate if the pool has remained quiescent for - * SHRINK_RATE nanosecs. This will slowly propagate, eventually - * terminating all workers after long periods of non-use. + * time out and terminate if the pool has remained quiescent for a + * given period -- a short period if there are more threads than + * parallelism, longer as the number of threads decreases. This + * will slowly propagate, eventually terminating all workers after + * periods of non-use. * * Shutdown and Termination. A call to shutdownNow atomically sets - * a runState bit and then (non-atomically) sets each worker's - * runState status, cancels all unprocessed tasks, and wakes up + * a plock bit and then (non-atomically) sets each worker's + * qlock status, cancels all unprocessed tasks, and wakes up * all waiting workers. Detecting whether termination should * commence after a non-abrupt shutdown() call requires more work * and bookkeeping. We need consensus about quiescence (i.e., that @@ -354,13 +718,13 @@ public class ForkJoinPool extends AbstractExecutorService { * method tryCompensate() may create or re-activate a spare * thread to compensate for blocked joiners until they unblock. * - * A third form (implemented in tryRemoveAndExec and - * tryPollForAndExec) amounts to helping a hypothetical - * compensator: If we can readily tell that a possible action of a - * compensator is to steal and execute the task being joined, the - * joining thread can do so directly, without the need for a - * compensation thread (although at the expense of larger run-time - * stacks, but the tradeoff is typically worthwhile). + * A third form (implemented in tryRemoveAndExec) amounts to + * helping a hypothetical compensator: If we can readily tell that + * a possible action of a compensator is to steal and execute the + * task being joined, the joining thread can do so directly, + * without the need for a compensation thread (although at the + * expense of larger run-time stacks, but the tradeoff is + * typically worthwhile). * * The ManagedBlocker extension API can't use helping so relies * only on compensation in method awaitBlocker. @@ -382,12 +746,12 @@ public class ForkJoinPool extends AbstractExecutorService { * steals, rather than use per-task bookkeeping. This sometimes * requires a linear scan of workQueues array to locate stealers, * but often doesn't because stealers leave hints (that may become - * stale/wrong) of where to locate them. A stealHint is only a - * hint because a worker might have had multiple steals and the - * hint records only one of them (usually the most current). - * Hinting isolates cost to when it is needed, rather than adding - * to per-task overhead. (2) It is "shallow", ignoring nesting - * and potentially cyclic mutual steals. (3) It is intentionally + * stale/wrong) of where to locate them. It is only a hint + * because a worker might have had multiple steals and the hint + * records only one of them (usually the most current). Hinting + * isolates cost to when it is needed, rather than adding to + * per-task overhead. (2) It is "shallow", ignoring nesting and + * potentially cyclic mutual steals. (3) It is intentionally * racy: field currentJoin is updated only while actively joining, * which means that we miss links in the chain during long-lived * tasks, GC stalls etc (which is OK since blocking in such cases @@ -395,6 +759,12 @@ public class ForkJoinPool extends AbstractExecutorService { * to find work (see MAX_HELP) and fall back to suspending the * worker and if necessary replacing it with another. * + * Helping actions for CountedCompleters are much simpler: Method + * helpComplete can take and execute any task with the same root + * as the task being waited on. However, this still entails some + * traversal of completer chains, so is less efficient than using + * CountedCompleters without explicit joins. + * * It is impossible to keep exactly the target parallelism number * of threads running at any given time. Determining the * existence of conservatively safe helping targets, the @@ -416,30 +786,41 @@ public class ForkJoinPool extends AbstractExecutorService { * intractable) game with an opponent that may choose the worst * (for us) active thread to stall at any time. We take several * precautions to bound losses (and thus bound gains), mainly in - * methods tryCompensate and awaitJoin: (1) We only try - * compensation after attempting enough helping steps (measured - * via counting and timing) that we have already consumed the - * estimated cost of creating and activating a new thread. (2) We - * allow up to 50% of threads to be blocked before initially - * adding any others, and unless completely saturated, check that - * some work is available for a new worker before adding. Also, we - * create up to only 50% more threads until entering a mode that - * only adds a thread if all others are possibly blocked. All - * together, this means that we might be half as fast to react, - * and create half as many threads as possible in the ideal case, - * but present vastly fewer anomalies in all other cases compared - * to both more aggressive and more conservative alternatives. - * - * Style notes: There is a lot of representation-level coupling - * among classes ForkJoinPool, ForkJoinWorkerThread, and - * ForkJoinTask. The fields of WorkQueue maintain data structures - * managed by ForkJoinPool, so are directly accessed. There is - * little point trying to reduce this, since any associated future - * changes in representations will need to be accompanied by - * algorithmic changes anyway. Several methods intrinsically - * sprawl because they must accumulate sets of consistent reads of - * volatiles held in local variables. Methods signalWork() and - * scan() are the main bottlenecks, so are especially heavily + * methods tryCompensate and awaitJoin. + * + * Common Pool + * =========== + * + * The static common Pool always exists after static + * initialization. Since it (or any other created pool) need + * never be used, we minimize initial construction overhead and + * footprint to the setup of about a dozen fields, with no nested + * allocation. Most bootstrapping occurs within method + * fullExternalPush during the first submission to the pool. + * + * When external threads submit to the common pool, they can + * perform some subtask processing (see externalHelpJoin and + * related methods). We do not need to record whether these + * submissions are to the common pool -- if not, externalHelpJoin + * returns quickly (at the most helping to signal some common pool + * workers). These submitters would otherwise be blocked waiting + * for completion, so the extra effort (with liberally sprinkled + * task status checks) in inapplicable cases amounts to an odd + * form of limited spin-wait before blocking in ForkJoinTask.join. + * + * Style notes + * =========== + * + * There is a lot of representation-level coupling among classes + * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The + * fields of WorkQueue maintain data structures managed by + * ForkJoinPool, so are directly accessed. There is little point + * trying to reduce this, since any associated future changes in + * representations will need to be accompanied by algorithmic + * changes anyway. Several methods intrinsically sprawl because + * they must accumulate sets of consistent reads of volatiles held + * in local variables. Methods signalWork() and scan() are the + * main bottlenecks, so are especially heavily * micro-optimized/mangled. There are lots of inline assignments * (of form "while ((local = field) != 0)") which are usually the * simplest way to ensure the required read orderings (which are @@ -447,7 +828,8 @@ public class ForkJoinPool extends AbstractExecutorService { * declarations of these locals at the heads of methods or blocks. * There are several occurrences of the unusual "do {} while * (!cas...)" which is the simplest way to force an update of a - * CAS'ed variable. There are also other coding oddities that help + * CAS'ed variable. There are also other coding oddities (including + * several unnecessary-looking hoisted null checks) that help * some methods perform reasonably even when interpreted (not * compiled). * @@ -496,34 +878,31 @@ public class ForkJoinPool extends AbstractExecutorService { * Default ForkJoinWorkerThreadFactory implementation; creates a * new ForkJoinWorkerThread. */ - static class DefaultForkJoinWorkerThreadFactory + static final class DefaultForkJoinWorkerThreadFactory implements ForkJoinWorkerThreadFactory { - public ForkJoinWorkerThread newThread(ForkJoinPool pool) { + public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { return new ForkJoinWorkerThread(pool); } } /** - * A simple non-reentrant lock used for exclusion when managing - * queues and workers. We use a custom lock so that we can readily - * probe lock state in constructions that check among alternative - * actions. The lock is normally only very briefly held, and - * sometimes treated as a spinlock, but other usages block to - * reduce overall contention in those cases where locked code - * bodies perform allocation/resizing. + * Per-thread records for threads that submit to pools. Currently + * holds only pseudo-random seed / index that is used to choose + * submission queues in method externalPush. In the future, this may + * also incorporate a means to implement different task rejection + * and resubmission policies. + * + * Seeds for submitters and workers/workQueues work in basically + * the same way but are initialized and updated using slightly + * different mechanics. Both are initialized using the same + * approach as in class ThreadLocal, where successive values are + * unlikely to collide with previous values. Seeds are then + * randomly modified upon collisions using xorshifts, which + * requires a non-zero seed. */ - static final class Mutex extends AbstractQueuedSynchronizer { - public final boolean tryAcquire(int ignore) { - return compareAndSetState(0, 1); - } - public final boolean tryRelease(int ignore) { - setState(0); - return true; - } - public final void lock() { acquire(0); } - public final void unlock() { release(0); } - public final boolean isHeldExclusively() { return getState() == 1; } - public final Condition newCondition() { return new ConditionObject(); } + static final class Submitter { + int seed; + Submitter(int s) { seed = s; } } /** @@ -533,6 +912,7 @@ public class ForkJoinPool extends AbstractExecutorService { * actually do anything beyond having a unique identity. */ static final class EmptyTask extends ForkJoinTask<Void> { + private static final long serialVersionUID = -7721805057305804111L; EmptyTask() { status = ForkJoinTask.NORMAL; } // force done public final Void getRawResult() { return null; } public final void setRawResult(Void x) {} @@ -553,27 +933,31 @@ public class ForkJoinPool extends AbstractExecutorService { * * Field "top" is the index (mod array.length) of the next queue * slot to push to or pop from. It is written only by owner thread - * for push, or under lock for trySharedPush, and accessed by - * other threads only after reading (volatile) base. Both top and - * base are allowed to wrap around on overflow, but (top - base) - * (or more commonly -(base - top) to force volatile read of base - * before top) still estimates size. + * for push, or under lock for external/shared push, and accessed + * by other threads only after reading (volatile) base. Both top + * and base are allowed to wrap around on overflow, but (top - + * base) (or more commonly -(base - top) to force volatile read of + * base before top) still estimates size. The lock ("qlock") is + * forced to -1 on termination, causing all further lock attempts + * to fail. (Note: we don't need CAS for termination state because + * upon pool shutdown, all shared-queues will stop being used + * anyway.) Nearly all lock bodies are set up so that exceptions + * within lock bodies are "impossible" (modulo JVM errors that + * would cause failure anyway.) * * The array slots are read and written using the emulation of * volatiles/atomics provided by Unsafe. Insertions must in * general use putOrderedObject as a form of releasing store to * ensure that all writes to the task object are ordered before - * its publication in the queue. (Although we can avoid one case - * of this when locked in trySharedPush.) All removals entail a - * CAS to null. The array is always a power of two. To ensure - * safety of Unsafe array operations, all accesses perform - * explicit null checks and implicit bounds checks via - * power-of-two masking. + * its publication in the queue. All removals entail a CAS to + * null. The array is always a power of two. To ensure safety of + * Unsafe array operations, all accesses perform explicit null + * checks and implicit bounds checks via power-of-two masking. * * In addition to basic queuing support, this class contains * fields described elsewhere to control execution. It turns out - * to work better memory-layout-wise to include them in this - * class rather than a separate class. + * to work better memory-layout-wise to include them in this class + * rather than a separate class. * * Performance on most platforms is very sensitive to placement of * instances of both WorkQueues and their arrays -- we absolutely @@ -587,10 +971,10 @@ public class ForkJoinPool extends AbstractExecutorService { * trades off slightly slower average field access for the sake of * avoiding really bad worst-case access. (Until better JVM * support is in place, this padding is dependent on transient - * properties of JVM field layout rules.) We also take care in + * properties of JVM field layout rules.) We also take care in * allocating, sizing and resizing the array. Non-shared queue - * arrays are initialized (via method growArray) by workers before - * use. Others are allocated on first use. + * arrays are initialized by workers before use. Others are + * allocated on first use. */ static final class WorkQueue { /** @@ -613,16 +997,17 @@ public class ForkJoinPool extends AbstractExecutorService { */ static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M - volatile long totalSteals; // cumulative number of steals + // Heuristic padding to ameliorate unfortunate memory placements + volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06; + int seed; // for random scanning; initialize nonzero volatile int eventCount; // encoded inactivation count; < 0 if inactive int nextWait; // encoded record of next event waiter - int rescans; // remaining scans until block - int nsteals; // top-level task executions since last idle - final int mode; // lifo, fifo, or shared + int hint; // steal or signal hint (index) int poolIndex; // index of this queue in pool (or 0) - int stealHint; // index of most recent known stealer - volatile int runState; // 1: locked, -1: terminate; else 0 + final int mode; // 0: lifo, > 0: fifo, < 0: shared + int nsteals; // number of steals + volatile int qlock; // 1: locked, -1: terminate; else 0 volatile int base; // index of next slot for poll int top; // index of next slot for push ForkJoinTask<?>[] array; // the elements (initially unallocated) @@ -631,14 +1016,16 @@ public class ForkJoinPool extends AbstractExecutorService { volatile Thread parker; // == owner during call to park; else null volatile ForkJoinTask<?> currentJoin; // task being joined in awaitJoin ForkJoinTask<?> currentSteal; // current non-local task being executed - // Heuristic padding to ameliorate unfortunate memory placements - Object p00, p01, p02, p03, p04, p05, p06, p07; - Object p08, p09, p0a, p0b, p0c, p0d, p0e; - WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode) { - this.mode = mode; + volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17; + volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d; + + WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode, + int seed) { this.pool = pool; this.owner = owner; + this.mode = mode; + this.seed = seed; // Place indices in the center of array (that is not yet allocated) base = top = INITIAL_QUEUE_CAPACITY >>> 1; } @@ -651,7 +1038,7 @@ public class ForkJoinPool extends AbstractExecutorService { return (n >= 0) ? 0 : -n; // ignore transient negative } - /** + /** * Provides a more accurate estimate of whether this queue has * any tasks than does queueSize, by checking whether a * near-empty queue has at least one unclaimed task. @@ -663,62 +1050,63 @@ public class ForkJoinPool extends AbstractExecutorService { (n == -1 && ((a = array) == null || (m = a.length - 1) < 0 || - U.getObjectVolatile - (a, ((m & (s - 1)) << ASHIFT) + ABASE) == null))); + U.getObject + (a, (long)((m & (s - 1)) << ASHIFT) + ABASE) == null))); } /** - * Pushes a task. Call only by owner in unshared queues. + * Pushes a task. Call only by owner in unshared queues. (The + * shared-queue version is embedded in method externalPush.) * * @param task the task. Caller must ensure non-null. - * @throw RejectedExecutionException if array cannot be resized + * @throws RejectedExecutionException if array cannot be resized */ final void push(ForkJoinTask<?> task) { ForkJoinTask<?>[] a; ForkJoinPool p; int s = top, m, n; if ((a = array) != null) { // ignore if queue removed - U.putOrderedObject - (a, (((m = a.length - 1) & s) << ASHIFT) + ABASE, task); + int j = (((m = a.length - 1) & s) << ASHIFT) + ABASE; + U.putOrderedObject(a, j, task); if ((n = (top = s + 1) - base) <= 2) { if ((p = pool) != null) - p.signalWork(); + p.signalWork(this); } else if (n >= m) - growArray(true); + growArray(); } } - /** - * Pushes a task if lock is free and array is either big - * enough or can be resized to be big enough. - * - * @param task the task. Caller must ensure non-null. - * @return true if submitted + /** + * Initializes or doubles the capacity of array. Call either + * by owner or with lock held -- it is OK for base, but not + * top, to move while resizings are in progress. */ - final boolean trySharedPush(ForkJoinTask<?> task) { - boolean submitted = false; - if (runState == 0 && U.compareAndSwapInt(this, RUNSTATE, 0, 1)) { - ForkJoinTask<?>[] a = array; - int s = top; - try { - if ((a != null && a.length > s + 1 - base) || - (a = growArray(false)) != null) { // must presize - int j = (((a.length - 1) & s) << ASHIFT) + ABASE; - U.putObject(a, (long)j, task); // don't need "ordered" - top = s + 1; - submitted = true; - } - } finally { - runState = 0; // unlock - } + final ForkJoinTask<?>[] growArray() { + ForkJoinTask<?>[] oldA = array; + int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY; + if (size > MAXIMUM_QUEUE_CAPACITY) + throw new RejectedExecutionException("Queue capacity exceeded"); + int oldMask, t, b; + ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size]; + if (oldA != null && (oldMask = oldA.length - 1) >= 0 && + (t = top) - (b = base) > 0) { + int mask = size - 1; + do { + ForkJoinTask<?> x; + int oldj = ((b & oldMask) << ASHIFT) + ABASE; + int j = ((b & mask) << ASHIFT) + ABASE; + x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj); + if (x != null && + U.compareAndSwapObject(oldA, oldj, x, null)) + U.putObjectVolatile(a, j, x); + } while (++b != t); } - return submitted; + return a; } /** * Takes next task, if one exists, in LIFO order. Call only - * by owner in unshared queues. (We do not have a shared - * version of this method because it is never needed.) + * by owner in unshared queues. */ final ForkJoinTask<?> pop() { ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m; @@ -773,7 +1161,7 @@ public class ForkJoinPool extends AbstractExecutorService { else if (base == b) { if (b + 1 == top) break; - Thread.yield(); // wait for lagging update + Thread.yield(); // wait for lagging update (very rare) } } return null; @@ -800,6 +1188,7 @@ public class ForkJoinPool extends AbstractExecutorService { /** * Pops the given task only if it is at the current top. + * (A shared version is available only via FJP.tryExternalUnpush) */ final boolean tryUnpush(ForkJoinTask<?> t) { ForkJoinTask<?>[] a; int s; @@ -813,57 +1202,6 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * Polls the given task only if it is at the current base. - */ - final boolean pollFor(ForkJoinTask<?> task) { - ForkJoinTask<?>[] a; int b; - if ((b = base) - top < 0 && (a = array) != null) { - int j = (((a.length - 1) & b) << ASHIFT) + ABASE; - if (U.getObjectVolatile(a, j) == task && base == b && - U.compareAndSwapObject(a, j, task, null)) { - base = b + 1; - return true; - } - } - return false; - } - - /** - * Initializes or doubles the capacity of array. Call either - * by owner or with lock held -- it is OK for base, but not - * top, to move while resizings are in progress. - * - * @param rejectOnFailure if true, throw exception if capacity - * exceeded (relayed ultimately to user); else return null. - */ - final ForkJoinTask<?>[] growArray(boolean rejectOnFailure) { - ForkJoinTask<?>[] oldA = array; - int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY; - if (size <= MAXIMUM_QUEUE_CAPACITY) { - int oldMask, t, b; - ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size]; - if (oldA != null && (oldMask = oldA.length - 1) >= 0 && - (t = top) - (b = base) > 0) { - int mask = size - 1; - do { - ForkJoinTask<?> x; - int oldj = ((b & oldMask) << ASHIFT) + ABASE; - int j = ((b & mask) << ASHIFT) + ABASE; - x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj); - if (x != null && - U.compareAndSwapObject(oldA, oldj, x, null)) - U.putObjectVolatile(a, j, x); - } while (++b != t); - } - return a; - } - else if (!rejectOnFailure) - return null; - else - throw new RejectedExecutionException("Queue capacity exceeded"); - } - - /** * Removes and cancels all known tasks, ignoring any exceptions. */ final void cancelAll() { @@ -887,7 +1225,7 @@ public class ForkJoinPool extends AbstractExecutorService { return seed = r ^= r << 5; } - // Execution methods + // Specialized execution methods /** * Pops and runs tasks until empty. @@ -916,16 +1254,14 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * If present, removes from queue and executes the given task, or - * any other cancelled task. Returns (true) immediately on any CAS + * If present, removes from queue and executes the given task, + * or any other cancelled task. Returns (true) on any CAS * or consistency check failure so caller can retry. * - * @return 0 if no progress can be made, else positive - * (this unusual convention simplifies use with tryHelpStealer.) + * @return false if no progress can be made, else true */ - final int tryRemoveAndExec(ForkJoinTask<?> task) { - int stat = 1; - boolean removed = false, empty = true; + final boolean tryRemoveAndExec(ForkJoinTask<?> task) { + boolean stat = true, removed = false, empty = true; ForkJoinTask<?>[] a; int m, s, b, n; if ((a = array) != null && (m = a.length - 1) >= 0 && (n = (s = top) - (b = base)) > 0) { @@ -955,7 +1291,7 @@ public class ForkJoinPool extends AbstractExecutorService { } if (--n == 0) { if (!empty && base == b) - stat = 0; + stat = false; break; } } @@ -966,21 +1302,49 @@ public class ForkJoinPool extends AbstractExecutorService { } /** + * Polls for and executes the given task or any other task in + * its CountedCompleter computation. + */ + final boolean pollAndExecCC(ForkJoinTask<?> root) { + ForkJoinTask<?>[] a; int b; Object o; + outer: while ((b = base) - top < 0 && (a = array) != null) { + long j = (((a.length - 1) & b) << ASHIFT) + ABASE; + if ((o = U.getObject(a, j)) == null || + !(o instanceof CountedCompleter)) + break; + for (CountedCompleter<?> t = (CountedCompleter<?>)o, r = t;;) { + if (r == root) { + if (base == b && + U.compareAndSwapObject(a, j, t, null)) { + base = b + 1; + t.doExec(); + return true; + } + else + break; // restart + } + if ((r = r.completer) == null) + break outer; // not part of root computation + } + } + return false; + } + + /** * Executes a top-level task and any local tasks remaining * after execution. */ final void runTask(ForkJoinTask<?> t) { if (t != null) { - currentSteal = t; - t.doExec(); - if (top != base) { // process remaining local tasks + (currentSteal = t).doExec(); + currentSteal = null; + ++nsteals; + if (base - top < 0) { // process remaining local tasks if (mode == 0) popAndExecAll(); else pollAndExecAll(); } - ++nsteals; - currentSteal = null; } } @@ -990,8 +1354,7 @@ public class ForkJoinPool extends AbstractExecutorService { final void runSubtask(ForkJoinTask<?> t) { if (t != null) { ForkJoinTask<?> ps = currentSteal; - currentSteal = t; - t.doExec(); + (currentSteal = t).doExec(); currentSteal = ps; } } @@ -1008,74 +1371,29 @@ public class ForkJoinPool extends AbstractExecutorService { s != Thread.State.TIMED_WAITING); } - /** - * If this owned and is not already interrupted, try to - * interrupt and/or unpark, ignoring exceptions. - */ - final void interruptOwner() { - Thread wt, p; - if ((wt = owner) != null && !wt.isInterrupted()) { - try { - wt.interrupt(); - } catch (SecurityException ignore) { - } - } - if ((p = parker) != null) - U.unpark(p); - } - // Unsafe mechanics private static final sun.misc.Unsafe U; - private static final long RUNSTATE; + private static final long QLOCK; private static final int ABASE; private static final int ASHIFT; static { - int s; try { U = getUnsafe(); Class<?> k = WorkQueue.class; Class<?> ak = ForkJoinTask[].class; - RUNSTATE = U.objectFieldOffset - (k.getDeclaredField("runState")); + QLOCK = U.objectFieldOffset + (k.getDeclaredField("qlock")); ABASE = U.arrayBaseOffset(ak); - s = U.arrayIndexScale(ak); + int scale = U.arrayIndexScale(ak); + if ((scale & (scale - 1)) != 0) + throw new Error("data type scale not a power of two"); + ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) { throw new Error(e); } - if ((s & (s-1)) != 0) - throw new Error("data type scale not a power of two"); - ASHIFT = 31 - Integer.numberOfLeadingZeros(s); - } - } - /** - * Per-thread records for threads that submit to pools. Currently - * holds only pseudo-random seed / index that is used to choose - * submission queues in method doSubmit. In the future, this may - * also incorporate a means to implement different task rejection - * and resubmission policies. - * - * Seeds for submitters and workers/workQueues work in basically - * the same way but are initialized and updated using slightly - * different mechanics. Both are initialized using the same - * approach as in class ThreadLocal, where successive values are - * unlikely to collide with previous values. This is done during - * registration for workers, but requires a separate AtomicInteger - * for submitters. Seeds are then randomly modified upon - * collisions using xorshifts, which requires a non-zero seed. - */ - static final class Submitter { - int seed; - Submitter() { - int s = nextSubmitterSeed.getAndAdd(SEED_INCREMENT); - seed = (s == 0) ? 1 : s; // ensure non-zero } } - /** ThreadLocal class for Submitters */ - static final class ThreadSubmitter extends ThreadLocal<Submitter> { - public Submitter initialValue() { return new Submitter(); } - } - // static fields (initialized in static initializer below) /** @@ -1086,15 +1404,13 @@ public class ForkJoinPool extends AbstractExecutorService { defaultForkJoinWorkerThreadFactory; /** - * Generator for assigning sequence numbers as pool names. - */ - private static final AtomicInteger poolNumberGenerator; - - /** - * Generator for initial hashes/seeds for submitters. Accessed by - * Submitter class constructor. + * Per-thread submission bookkeeping. Shared across all pools + * to reduce ThreadLocal pollution and because random motion + * to avoid contention in one pool is likely to hold for others. + * Lazily initialized on first submission (but null-checked + * in other contexts to avoid unnecessary initialization). */ - static final AtomicInteger nextSubmitterSeed; + static final ThreadLocal<Submitter> submitters; /** * Permission required for callers of methods that may start or @@ -1103,37 +1419,56 @@ public class ForkJoinPool extends AbstractExecutorService { private static final RuntimePermission modifyThreadPermission; /** - * Per-thread submission bookeeping. Shared across all pools - * to reduce ThreadLocal pollution and because random motion - * to avoid contention in one pool is likely to hold for others. + * Common (static) pool. Non-null for public use unless a static + * construction exception, but internal usages null-check on use + * to paranoically avoid potential initialization circularities + * as well as to simplify generated code. */ - private static final ThreadSubmitter submitters; + static final ForkJoinPool common; + + /** + * Common pool parallelism. Must equal common.parallelism. + */ + static final int commonParallelism; + + /** + * Sequence number for creating workerNamePrefix. + */ + private static int poolNumberSequence; + + /** + * Returns the next sequence number. We don't expect this to + * ever contend, so use simple builtin sync. + */ + private static final synchronized int nextPoolId() { + return ++poolNumberSequence; + } // static constants /** - * The wakeup interval (in nanoseconds) for a worker waiting for a - * task when the pool is quiescent to instead try to shrink the - * number of workers. The exact value does not matter too - * much. It must be short enough to release resources during - * sustained periods of idleness, but not so short that threads - * are continually re-created. + * Initial timeout value (in nanoseconds) for the thread + * triggering quiescence to park waiting for new work. On timeout, + * the thread will instead try to shrink the number of + * workers. The value should be large enough to avoid overly + * aggressive shrinkage during most transient stalls (long GCs + * etc). + */ + private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec + + /** + * Timeout value when there are more threads than parallelism level */ - private static final long SHRINK_RATE = - 4L * 1000L * 1000L * 1000L; // 4 seconds + private static final long FAST_IDLE_TIMEOUT = 200L * 1000L * 1000L; /** - * The timeout value for attempted shrinkage, includes - * some slop to cope with system timer imprecision. + * Tolerance for idle timeouts, to cope with timer undershoots */ - private static final long SHRINK_TIMEOUT = SHRINK_RATE - (SHRINK_RATE / 10); + private static final long TIMEOUT_SLOP = 2000000L; /** * The maximum stolen->joining link depth allowed in method - * tryHelpStealer. Must be a power of two. This value also - * controls the maximum number of times to try to help join a task - * without any apparent progress or change in pool state before - * giving up and blocking (see awaitJoin). Depths for legitimate + * tryHelpStealer. Must be a power of two. Depths for legitimate * chains are unbounded, but we use a fixed constant to avoid * (otherwise unchecked) cycles and to bound staleness of * traversal parameters at the expense of sometimes blocking when @@ -1142,22 +1477,12 @@ public class ForkJoinPool extends AbstractExecutorService { private static final int MAX_HELP = 64; /** - * Secondary time-based bound (in nanosecs) for helping attempts - * before trying compensated blocking in awaitJoin. Used in - * conjunction with MAX_HELP to reduce variance due to different - * polling rates associated with different helping options. The - * value should roughly approximate the time required to create - * and/or activate a worker thread. - */ - private static final long COMPENSATION_DELAY = 1L << 18; // ~0.25 millisec - - /** * Increment for seed generators. See class ThreadLocal for * explanation. */ private static final int SEED_INCREMENT = 0x61c88647; - /** + /* * Bits and masks for control variables * * Field ctl is a long packed with: @@ -1185,14 +1510,14 @@ public class ForkJoinPool extends AbstractExecutorService { * scan for them to avoid queuing races. Note however that * eventCount updates lag releases so usage requires care. * - * Field runState is an int packed with: + * Field plock is an int packed with: * SHUTDOWN: true if shutdown is enabled (1 bit) - * SEQ: a sequence number updated upon (de)registering workers (30 bits) - * INIT: set true after workQueues array construction (1 bit) + * SEQ: a sequence lock, with PL_LOCK bit set if locked (30 bits) + * SIGNAL: set when threads may be waiting on the lock (1 bit) * * The sequence number enables simple consistency checks: * Staleness of read-only operations on the workQueues array can - * be checked by comparing runState before vs after the reads. + * be checked by comparing plock before vs after the reads. */ // bit positions/shifts for fields @@ -1204,7 +1529,8 @@ public class ForkJoinPool extends AbstractExecutorService { // bounds private static final int SMASK = 0xffff; // short bits private static final int MAX_CAP = 0x7fff; // max #workers - 1 - private static final int SQMASK = 0xfffe; // even short bits + private static final int EVENMASK = 0xfffe; // even short bits + private static final int SQMASK = 0x007e; // max 64 (even) slots private static final int SHORT_SIGN = 1 << 15; private static final int INT_SIGN = 1 << 31; @@ -1229,91 +1555,164 @@ public class ForkJoinPool extends AbstractExecutorService { private static final int E_MASK = 0x7fffffff; // no STOP_BIT private static final int E_SEQ = 1 << EC_SHIFT; - // runState bits + // plock bits private static final int SHUTDOWN = 1 << 31; + private static final int PL_LOCK = 2; + private static final int PL_SIGNAL = 1; + private static final int PL_SPINS = 1 << 8; // access mode for WorkQueue static final int LIFO_QUEUE = 0; static final int FIFO_QUEUE = 1; static final int SHARED_QUEUE = -1; + // bounds for #steps in scan loop -- must be power 2 minus 1 + private static final int MIN_SCAN = 0x1ff; // cover estimation slop + private static final int MAX_SCAN = 0x1ffff; // 4 * max workers + // Instance fields /* - * Field layout order in this class tends to matter more than one - * would like. Runtime layout order is only loosely related to + * Field layout of this class tends to matter more than one would + * like. Runtime layout order is only loosely related to * declaration order and may differ across JVMs, but the following * empirically works OK on current JVMs. */ + // Heuristic padding to ameliorate unfortunate memory placements + volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06; + + volatile long stealCount; // collects worker counts volatile long ctl; // main pool control - final int parallelism; // parallelism level - final int localMode; // per-worker scheduling mode - final int submitMask; // submit queue index bound - int nextSeed; // for initializing worker seeds - volatile int runState; // shutdown status and seq + volatile int plock; // shutdown status and seqLock + volatile int indexSeed; // worker/submitter index seed + final int config; // mode and parallelism level WorkQueue[] workQueues; // main registry - final Mutex lock; // for registration - final Condition termination; // for awaitTermination - final ForkJoinWorkerThreadFactory factory; // factory for new workers + final ForkJoinWorkerThreadFactory factory; final Thread.UncaughtExceptionHandler ueh; // per-worker UEH - final AtomicLong stealCount; // collect counts when terminated - final AtomicInteger nextWorkerNumber; // to create worker name string final String workerNamePrefix; // to create worker name string - // Creating, registering, and deregistering workers + volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17; + volatile Object pad18, pad19, pad1a, pad1b; + + /** + * Acquires the plock lock to protect worker array and related + * updates. This method is called only if an initial CAS on plock + * fails. This acts as a spinlock for normal cases, but falls back + * to builtin monitor to block when (rarely) needed. This would be + * a terrible idea for a highly contended lock, but works fine as + * a more conservative alternative to a pure spinlock. + */ + private int acquirePlock() { + int spins = PL_SPINS, r = 0, ps, nps; + for (;;) { + if (((ps = plock) & PL_LOCK) == 0 && + U.compareAndSwapInt(this, PLOCK, ps, nps = ps + PL_LOCK)) + return nps; + else if (r == 0) { // randomize spins if possible + Thread t = Thread.currentThread(); WorkQueue w; Submitter z; + if ((t instanceof ForkJoinWorkerThread) && + (w = ((ForkJoinWorkerThread)t).workQueue) != null) + r = w.seed; + else if ((z = submitters.get()) != null) + r = z.seed; + else + r = 1; + } + else if (spins >= 0) { + r ^= r << 1; r ^= r >>> 3; r ^= r << 10; // xorshift + if (r >= 0) + --spins; + } + else if (U.compareAndSwapInt(this, PLOCK, ps, ps | PL_SIGNAL)) { + synchronized (this) { + if ((plock & PL_SIGNAL) != 0) { + try { + wait(); + } catch (InterruptedException ie) { + try { + Thread.currentThread().interrupt(); + } catch (SecurityException ignore) { + } + } + } + else + notifyAll(); + } + } + } + } /** - * Tries to create and start a worker + * Unlocks and signals any thread waiting for plock. Called only + * when CAS of seq value for unlock fails. */ - private void addWorker() { - Throwable ex = null; - ForkJoinWorkerThread wt = null; - try { - if ((wt = factory.newThread(this)) != null) { - wt.start(); - return; - } - } catch (Throwable e) { - ex = e; - } - deregisterWorker(wt, ex); // adjust counts etc on failure + private void releasePlock(int ps) { + plock = ps; + synchronized (this) { notifyAll(); } } /** - * Callback from ForkJoinWorkerThread constructor to assign a - * public name. This must be separate from registerWorker because - * it is called during the "super" constructor call in - * ForkJoinWorkerThread. + * Tries to create and start one worker if fewer than target + * parallelism level exist. Adjusts counts etc on failure. */ - final String nextWorkerName() { - return workerNamePrefix.concat - (Integer.toString(nextWorkerNumber.addAndGet(1))); + private void tryAddWorker() { + long c; int u; + while ((u = (int)((c = ctl) >>> 32)) < 0 && + (u & SHORT_SIGN) != 0 && (int)c == 0) { + long nc = (long)(((u + UTC_UNIT) & UTC_MASK) | + ((u + UAC_UNIT) & UAC_MASK)) << 32; + if (U.compareAndSwapLong(this, CTL, c, nc)) { + ForkJoinWorkerThreadFactory fac; + Throwable ex = null; + ForkJoinWorkerThread wt = null; + try { + if ((fac = factory) != null && + (wt = fac.newThread(this)) != null) { + wt.start(); + break; + } + } catch (Throwable e) { + ex = e; + } + deregisterWorker(wt, ex); + break; + } + } } + // Registering and deregistering workers + /** - * Callback from ForkJoinWorkerThread constructor to establish its - * poolIndex and record its WorkQueue. To avoid scanning bias due - * to packing entries in front of the workQueues array, we treat - * the array as a simple power-of-two hash table using per-thread - * seed as hash, expanding as needed. + * Callback from ForkJoinWorkerThread to establish and record its + * WorkQueue. To avoid scanning bias due to packing entries in + * front of the workQueues array, we treat the array as a simple + * power-of-two hash table using per-thread seed as hash, + * expanding as needed. * - * @param w the worker's queue + * @param wt the worker thread + * @return the worker's queue */ - - final void registerWorker(WorkQueue w) { - Mutex lock = this.lock; - lock.lock(); + final WorkQueue registerWorker(ForkJoinWorkerThread wt) { + Thread.UncaughtExceptionHandler handler; WorkQueue[] ws; int s, ps; + wt.setDaemon(true); + if ((handler = ueh) != null) + wt.setUncaughtExceptionHandler(handler); + do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed, + s += SEED_INCREMENT) || + s == 0); // skip 0 + WorkQueue w = new WorkQueue(this, wt, config >>> 16, s); + if (((ps = plock) & PL_LOCK) != 0 || + !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) + ps = acquirePlock(); + int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); try { - WorkQueue[] ws = workQueues; - if (w != null && ws != null) { // skip on shutdown/failure - int rs, n = ws.length, m = n - 1; - int s = nextSeed += SEED_INCREMENT; // rarely-colliding sequence - w.seed = (s == 0) ? 1 : s; // ensure non-zero seed - int r = (s << 1) | 1; // use odd-numbered indices - if (ws[r &= m] != null) { // collision - int probes = 0; // step by approx half size - int step = (n <= 4) ? 2 : ((n >>> 1) & SQMASK) + 2; + if ((ws = workQueues) != null) { // skip if shutting down + int n = ws.length, m = n - 1; + int r = (s << 1) | 1; // use odd-numbered indices + if (ws[r &= m] != null) { // collision + int probes = 0; // step by approx half size + int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2; while (ws[r = (r + step) & m] != null) { if (++probes >= n) { workQueues = ws = Arrays.copyOf(ws, n <<= 1); @@ -1322,104 +1721,206 @@ public class ForkJoinPool extends AbstractExecutorService { } } } - w.eventCount = w.poolIndex = r; // establish before recording - ws[r] = w; // also update seq - runState = ((rs = runState) & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN); + w.eventCount = w.poolIndex = r; // volatile write orders + ws[r] = w; } } finally { - lock.unlock(); + if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) + releasePlock(nps); } + wt.setName(workerNamePrefix.concat(Integer.toString(w.poolIndex))); + return w; } /** * Final callback from terminating worker, as well as upon failure - * to construct or start a worker in addWorker. Removes record of - * worker from array, and adjusts counts. If pool is shutting - * down, tries to complete termination. + * to construct or start a worker. Removes record of worker from + * array, and adjusts counts. If pool is shutting down, tries to + * complete termination. * - * @param wt the worker thread or null if addWorker failed + * @param wt the worker thread or null if construction failed * @param ex the exception causing failure, or null if none */ final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { - Mutex lock = this.lock; WorkQueue w = null; if (wt != null && (w = wt.workQueue) != null) { - w.runState = -1; // ensure runState is set - stealCount.getAndAdd(w.totalSteals + w.nsteals); - int idx = w.poolIndex; - lock.lock(); - try { // remove record from array + int ps; + w.qlock = -1; // ensure set + long ns = w.nsteals, sc; // collect steal count + do {} while (!U.compareAndSwapLong(this, STEALCOUNT, + sc = stealCount, sc + ns)); + if (((ps = plock) & PL_LOCK) != 0 || + !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) + ps = acquirePlock(); + int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); + try { + int idx = w.poolIndex; WorkQueue[] ws = workQueues; if (ws != null && idx >= 0 && idx < ws.length && ws[idx] == w) ws[idx] = null; } finally { - lock.unlock(); + if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) + releasePlock(nps); } } - long c; // adjust ctl counts + long c; // adjust ctl counts do {} while (!U.compareAndSwapLong (this, CTL, c = ctl, (((c - AC_UNIT) & AC_MASK) | ((c - TC_UNIT) & TC_MASK) | (c & ~(AC_MASK|TC_MASK))))); - if (!tryTerminate(false, false) && w != null) { - w.cancelAll(); // cancel remaining tasks - if (w.array != null) // suppress signal if never ran - signalWork(); // wake up or create replacement - if (ex == null) // help clean refs on way out - ForkJoinTask.helpExpungeStaleExceptions(); + if (!tryTerminate(false, false) && w != null && w.array != null) { + w.cancelAll(); // cancel remaining tasks + WorkQueue[] ws; WorkQueue v; Thread p; int u, i, e; + while ((u = (int)((c = ctl) >>> 32)) < 0 && (e = (int)c) >= 0) { + if (e > 0) { // activate or create replacement + if ((ws = workQueues) == null || + (i = e & SMASK) >= ws.length || + (v = ws[i]) == null) + break; + long nc = (((long)(v.nextWait & E_MASK)) | + ((long)(u + UAC_UNIT) << 32)); + if (v.eventCount != (e | INT_SIGN)) + break; + if (U.compareAndSwapLong(this, CTL, c, nc)) { + v.eventCount = (e + E_SEQ) & E_MASK; + if ((p = v.parker) != null) + U.unpark(p); + break; + } + } + else { + if ((short)u < 0) + tryAddWorker(); + break; + } + } } - - if (ex != null) // rethrow + if (ex == null) // help clean refs on way out + ForkJoinTask.helpExpungeStaleExceptions(); + else // rethrow ForkJoinTask.rethrow(ex); } - // Submissions /** * Unless shutting down, adds the given task to a submission queue * at submitter's current queue index (modulo submission - * range). If no queue exists at the index, one is created. If - * the queue is busy, another index is randomly chosen. The - * submitMask bounds the effective number of queues to the - * (nearest power of two for) parallelism level. + * range). Only the most common path is directly handled in this + * method. All others are relayed to fullExternalPush. * * @param task the task. Caller must ensure non-null. */ - private void doSubmit(ForkJoinTask<?> task) { - Submitter s = submitters.get(); - for (int r = s.seed, m = submitMask;;) { - WorkQueue[] ws; WorkQueue q; - int k = r & m & SQMASK; // use only even indices - if (runState < 0 || (ws = workQueues) == null || ws.length <= k) - throw new RejectedExecutionException(); // shutting down - else if ((q = ws[k]) == null) { // create new queue - WorkQueue nq = new WorkQueue(this, null, SHARED_QUEUE); - Mutex lock = this.lock; // construct outside lock - lock.lock(); - try { // recheck under lock - int rs = runState; // to update seq - if (ws == workQueues && ws[k] == null) { - ws[k] = nq; - runState = ((rs & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN)); - } - } finally { - lock.unlock(); - } - } - else if (q.trySharedPush(task)) { - signalWork(); + final void externalPush(ForkJoinTask<?> task) { + WorkQueue[] ws; WorkQueue q; Submitter z; int m; ForkJoinTask<?>[] a; + if ((z = submitters.get()) != null && plock > 0 && + (ws = workQueues) != null && (m = (ws.length - 1)) >= 0 && + (q = ws[m & z.seed & SQMASK]) != null && + U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock + int b = q.base, s = q.top, n, an; + if ((a = q.array) != null && (an = a.length) > (n = s + 1 - b)) { + int j = (((an - 1) & s) << ASHIFT) + ABASE; + U.putOrderedObject(a, j, task); + q.top = s + 1; // push on to deque + q.qlock = 0; + if (n <= 2) + signalWork(q); return; } - else if (m > 1) { // move to a different index - r ^= r << 13; // same xorshift as WorkQueues + q.qlock = 0; + } + fullExternalPush(task); + } + + /** + * Full version of externalPush. This method is called, among + * other times, upon the first submission of the first task to the + * pool, so must perform secondary initialization. It also + * detects first submission by an external thread by looking up + * its ThreadLocal, and creates a new shared queue if the one at + * index if empty or contended. The plock lock body must be + * exception-free (so no try/finally) so we optimistically + * allocate new queues outside the lock and throw them away if + * (very rarely) not needed. + * + * Secondary initialization occurs when plock is zero, to create + * workQueue array and set plock to a valid value. This lock body + * must also be exception-free. Because the plock seq value can + * eventually wrap around zero, this method harmlessly fails to + * reinitialize if workQueues exists, while still advancing plock. + */ + private void fullExternalPush(ForkJoinTask<?> task) { + int r = 0; // random index seed + for (Submitter z = submitters.get();;) { + WorkQueue[] ws; WorkQueue q; int ps, m, k; + if (z == null) { + if (U.compareAndSwapInt(this, INDEXSEED, r = indexSeed, + r += SEED_INCREMENT) && r != 0) + submitters.set(z = new Submitter(r)); + } + else if (r == 0) { // move to a different index + r = z.seed; + r ^= r << 13; // same xorshift as WorkQueues r ^= r >>> 17; - s.seed = r ^= r << 5; + z.seed = r ^ (r << 5); + } + else if ((ps = plock) < 0) + throw new RejectedExecutionException(); + else if (ps == 0 || (ws = workQueues) == null || + (m = ws.length - 1) < 0) { // initialize workQueues + int p = config & SMASK; // find power of two table size + int n = (p > 1) ? p - 1 : 1; // ensure at least 2 slots + n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; + n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1; + WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ? + new WorkQueue[n] : null); + if (((ps = plock) & PL_LOCK) != 0 || + !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) + ps = acquirePlock(); + if (((ws = workQueues) == null || ws.length == 0) && nws != null) + workQueues = nws; + int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); + if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) + releasePlock(nps); + } + else if ((q = ws[k = r & m & SQMASK]) != null) { + if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) { + ForkJoinTask<?>[] a = q.array; + int s = q.top; + boolean submitted = false; + try { // locked version of push + if ((a != null && a.length > s + 1 - q.base) || + (a = q.growArray()) != null) { // must presize + int j = (((a.length - 1) & s) << ASHIFT) + ABASE; + U.putOrderedObject(a, j, task); + q.top = s + 1; + submitted = true; + } + } finally { + q.qlock = 0; // unlock + } + if (submitted) { + signalWork(q); + return; + } + } + r = 0; // move on failure + } + else if (((ps = plock) & PL_LOCK) == 0) { // create new queue + q = new WorkQueue(this, null, SHARED_QUEUE, r); + if (((ps = plock) & PL_LOCK) != 0 || + !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) + ps = acquirePlock(); + if ((ws = workQueues) != null && k < ws.length && ws[k] == null) + ws[k] = q; + int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); + if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) + releasePlock(nps); } else - Thread.yield(); // yield if no alternatives + r = 0; // try elsewhere while lock held } } @@ -1434,37 +1935,37 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * Tries to activate or create a worker if too few are active. + * Tries to create or activate a worker if too few are active. + * + * @param q the (non-null) queue holding tasks to be signalled */ - final void signalWork() { - long c; int u; - while ((u = (int)((c = ctl) >>> 32)) < 0) { // too few active - WorkQueue[] ws = workQueues; int e, i; WorkQueue w; Thread p; - if ((e = (int)c) > 0) { // at least one waiting - if (ws != null && (i = e & SMASK) < ws.length && + final void signalWork(WorkQueue q) { + int hint = q.poolIndex; + long c; int e, u, i, n; WorkQueue[] ws; WorkQueue w; Thread p; + while ((u = (int)((c = ctl) >>> 32)) < 0) { + if ((e = (int)c) > 0) { + if ((ws = workQueues) != null && ws.length > (i = e & SMASK) && (w = ws[i]) != null && w.eventCount == (e | INT_SIGN)) { long nc = (((long)(w.nextWait & E_MASK)) | ((long)(u + UAC_UNIT) << 32)); if (U.compareAndSwapLong(this, CTL, c, nc)) { + w.hint = hint; w.eventCount = (e + E_SEQ) & E_MASK; if ((p = w.parker) != null) - U.unpark(p); // activate and release + U.unpark(p); break; } + if (q.top - q.base <= 0) + break; } else break; } - else if (e == 0 && (u & SHORT_SIGN) != 0) { // too few total - long nc = (long)(((u + UTC_UNIT) & UTC_MASK) | - ((u + UAC_UNIT) & UAC_MASK)) << 32; - if (U.compareAndSwapLong(this, CTL, c, nc)) { - addWorker(); - break; - } - } - else + else { + if ((short)u < 0) + tryAddWorker(); break; + } } } @@ -1474,8 +1975,8 @@ public class ForkJoinPool extends AbstractExecutorService { * Top-level runloop for workers, called by ForkJoinWorkerThread.run. */ final void runWorker(WorkQueue w) { - w.growArray(false); // initialize queue array in this thread - do { w.runTask(scan(w)); } while (w.runState >= 0); + w.growArray(); // allocate queue + do { w.runTask(scan(w)); } while (w.qlock >= 0); } /** @@ -1486,116 +1987,122 @@ public class ForkJoinPool extends AbstractExecutorService { * contention, or state changes that indicate possible success on * re-invocation. * - * The scan searches for tasks across a random permutation of - * queues (starting at a random index and stepping by a random - * relative prime, checking each at least once). The scan - * terminates upon either finding a non-empty queue, or completing - * the sweep. If the worker is not inactivated, it takes and - * returns a task from this queue. On failure to find a task, we - * take one of the following actions, after which the caller will - * retry calling this method unless terminated. + * The scan searches for tasks across queues (starting at a random + * index, and relying on registerWorker to irregularly scatter + * them within array to avoid bias), checking each at least twice. + * The scan terminates upon either finding a non-empty queue, or + * completing the sweep. If the worker is not inactivated, it + * takes and returns a task from this queue. Otherwise, if not + * activated, it signals workers (that may include itself) and + * returns so caller can retry. Also returns for true if the + * worker array may have changed during an empty scan. On failure + * to find a task, we take one of the following actions, after + * which the caller will retry calling this method unless + * terminated. * * * If pool is terminating, terminate the worker. * - * * If not a complete sweep, try to release a waiting worker. If - * the scan terminated because the worker is inactivated, then the - * released worker will often be the calling worker, and it can - * succeed obtaining a task on the next call. Or maybe it is - * another worker, but with same net effect. Releasing in other - * cases as well ensures that we have enough workers running. - * * * If not already enqueued, try to inactivate and enqueue the * worker on wait queue. Or, if inactivating has caused the pool - * to be quiescent, relay to idleAwaitWork to check for - * termination and possibly shrink pool. + * to be quiescent, relay to idleAwaitWork to possibly shrink + * pool. * - * * If already inactive, and the caller has run a task since the - * last empty scan, return (to allow rescan) unless others are - * also inactivated. Field WorkQueue.rescans counts down on each - * scan to ensure eventual inactivation and blocking. + * * If already enqueued and none of the above apply, possibly + * park awaiting signal, else lingering to help scan and signal. * - * * If already enqueued and none of the above apply, park - * awaiting signal, + * * If a non-empty queue discovered or left as a hint, + * help wake up other workers before return. * * @param w the worker (via its WorkQueue) - * @return a task or null of none found + * @return a task or null if none found */ private final ForkJoinTask<?> scan(WorkQueue w) { - WorkQueue[] ws; // first update random seed - int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; - int rs = runState, m; // volatile read order matters - if ((ws = workQueues) != null && (m = ws.length - 1) > 0) { - int ec = w.eventCount; // ec is negative if inactive - int step = (r >>> 16) | 1; // relative prime - for (int j = (m + 1) << 2; ; r += step) { - WorkQueue q; ForkJoinTask<?> t; ForkJoinTask<?>[] a; int b; - if ((q = ws[r & m]) != null && (b = q.base) - q.top < 0 && - (a = q.array) != null) { // probably nonempty + WorkQueue[] ws; int m; + int ps = plock; // read plock before ws + if (w != null && (ws = workQueues) != null && (m = ws.length - 1) >= 0) { + int ec = w.eventCount; // ec is negative if inactive + int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; + w.hint = -1; // update seed and clear hint + int j = ((m + m + 1) | MIN_SCAN) & MAX_SCAN; + do { + WorkQueue q; ForkJoinTask<?>[] a; int b; + if ((q = ws[(r + j) & m]) != null && (b = q.base) - q.top < 0 && + (a = q.array) != null) { // probably nonempty int i = (((a.length - 1) & b) << ASHIFT) + ABASE; - t = (ForkJoinTask<?>)U.getObjectVolatile(a, i); + ForkJoinTask<?> t = (ForkJoinTask<?>) + U.getObjectVolatile(a, i); if (q.base == b && ec >= 0 && t != null && U.compareAndSwapObject(a, i, t, null)) { - q.base = b + 1; // specialization of pollAt - return t; + if ((q.base = b + 1) - q.top < 0) + signalWork(q); + return t; // taken } - else if (ec < 0 || j <= m) { - rs = 0; // mark scan as imcomplete - break; // caller can retry after release + else if ((ec < 0 || j < m) && (int)(ctl >> AC_SHIFT) <= 0) { + w.hint = (r + j) & m; // help signal below + break; // cannot take } } - if (--j < 0) - break; - } + } while (--j >= 0); - long c = ctl; int e = (int)c, a = (int)(c >> AC_SHIFT), nr, ns; - if (e < 0) // decode ctl on empty scan - w.runState = -1; // pool is terminating - else if (rs == 0 || rs != runState) { // incomplete scan - WorkQueue v; Thread p; // try to release a waiter - if (e > 0 && a < 0 && w.eventCount == ec && - (v = ws[e & m]) != null && v.eventCount == (e | INT_SIGN)) { - long nc = ((long)(v.nextWait & E_MASK) | - ((c + AC_UNIT) & (AC_MASK|TC_MASK))); - if (ctl == c && U.compareAndSwapLong(this, CTL, c, nc)) { + int h, e, ns; long c, sc; WorkQueue q; + if ((ns = w.nsteals) != 0) { + if (U.compareAndSwapLong(this, STEALCOUNT, + sc = stealCount, sc + ns)) + w.nsteals = 0; // collect steals and rescan + } + else if (plock != ps) // consistency check + ; // skip + else if ((e = (int)(c = ctl)) < 0) + w.qlock = -1; // pool is terminating + else { + if ((h = w.hint) < 0) { + if (ec >= 0) { // try to enqueue/inactivate + long nc = (((long)ec | + ((c - AC_UNIT) & (AC_MASK|TC_MASK)))); + w.nextWait = e; // link and mark inactive + w.eventCount = ec | INT_SIGN; + if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc)) + w.eventCount = ec; // unmark on CAS failure + else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK)) + idleAwaitWork(w, nc, c); + } + else if (w.eventCount < 0 && ctl == c) { + Thread wt = Thread.currentThread(); + Thread.interrupted(); // clear status + U.putObject(wt, PARKBLOCKER, this); + w.parker = wt; // emulate LockSupport.park + if (w.eventCount < 0) // recheck + U.park(false, 0L); // block + w.parker = null; + U.putObject(wt, PARKBLOCKER, null); + } + } + if ((h >= 0 || (h = w.hint) >= 0) && + (ws = workQueues) != null && h < ws.length && + (q = ws[h]) != null) { // signal others before retry + WorkQueue v; Thread p; int u, i, s; + for (int n = (config & SMASK) - 1;;) { + int idleCount = (w.eventCount < 0) ? 0 : -1; + if (((s = idleCount - q.base + q.top) <= n && + (n = s) <= 0) || + (u = (int)((c = ctl) >>> 32)) >= 0 || + (e = (int)c) <= 0 || m < (i = e & SMASK) || + (v = ws[i]) == null) + break; + long nc = (((long)(v.nextWait & E_MASK)) | + ((long)(u + UAC_UNIT) << 32)); + if (v.eventCount != (e | INT_SIGN) || + !U.compareAndSwapLong(this, CTL, c, nc)) + break; + v.hint = h; v.eventCount = (e + E_SEQ) & E_MASK; if ((p = v.parker) != null) U.unpark(p); + if (--n <= 0) + break; } } } - else if (ec >= 0) { // try to enqueue/inactivate - long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK)); - w.nextWait = e; - w.eventCount = ec | INT_SIGN; // mark as inactive - if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc)) - w.eventCount = ec; // unmark on CAS failure - else { - if ((ns = w.nsteals) != 0) { - w.nsteals = 0; // set rescans if ran task - w.rescans = (a > 0) ? 0 : a + parallelism; - w.totalSteals += ns; - } - if (a == 1 - parallelism) // quiescent - idleAwaitWork(w, nc, c); - } - } - else if (w.eventCount < 0) { // already queued - if ((nr = w.rescans) > 0) { // continue rescanning - int ac = a + parallelism; - if (((w.rescans = (ac < nr) ? ac : nr - 1) & 3) == 0) - Thread.yield(); // yield before block - } - else { - Thread.interrupted(); // clear status - Thread wt = Thread.currentThread(); - U.putObject(wt, PARKBLOCKER, this); - w.parker = wt; // emulate LockSupport.park - if (w.eventCount < 0) // recheck - U.park(false, 0L); - w.parker = null; - U.putObject(wt, PARKBLOCKER, null); - } - } } return null; } @@ -1603,8 +2110,8 @@ public class ForkJoinPool extends AbstractExecutorService { /** * If inactivating worker w has caused the pool to become * quiescent, checks for pool termination, and, so long as this is - * not the only worker, waits for event for up to SHRINK_RATE - * nanosecs. On timeout, if ctl has not changed, terminates the + * not the only worker, waits for event for up to a given + * duration. On timeout, if ctl has not changed, terminates the * worker, which will in turn wake up another worker to possibly * repeat this process. * @@ -1613,25 +2120,28 @@ public class ForkJoinPool extends AbstractExecutorService { * @param prevCtl the ctl value to restore if thread is terminated */ private void idleAwaitWork(WorkQueue w, long currentCtl, long prevCtl) { - if (w.eventCount < 0 && !tryTerminate(false, false) && - (int)prevCtl != 0 && !hasQueuedSubmissions() && ctl == currentCtl) { + if (w != null && w.eventCount < 0 && + !tryTerminate(false, false) && (int)prevCtl != 0 && + ctl == currentCtl) { + int dc = -(short)(currentCtl >>> TC_SHIFT); + long parkTime = dc < 0 ? FAST_IDLE_TIMEOUT: (dc + 1) * IDLE_TIMEOUT; + long deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP; Thread wt = Thread.currentThread(); - Thread.yield(); // yield before block while (ctl == currentCtl) { - long startTime = System.nanoTime(); Thread.interrupted(); // timed variant of version in scan() U.putObject(wt, PARKBLOCKER, this); w.parker = wt; if (ctl == currentCtl) - U.park(false, SHRINK_RATE); + U.park(false, parkTime); w.parker = null; U.putObject(wt, PARKBLOCKER, null); if (ctl != currentCtl) break; - if (System.nanoTime() - startTime >= SHRINK_TIMEOUT && + if (deadline - System.nanoTime() <= 0L && U.compareAndSwapLong(this, CTL, currentCtl, prevCtl)) { w.eventCount = (w.eventCount + E_SEQ) | E_MASK; - w.runState = -1; // shrink + w.hint = -1; + w.qlock = -1; // shrink break; } } @@ -1639,6 +2149,47 @@ public class ForkJoinPool extends AbstractExecutorService { } /** + * Scans through queues looking for work while joining a task; if + * any present, signals. May return early if more signalling is + * detectably unneeded. + * + * @param task return early if done + * @param origin an index to start scan + */ + private void helpSignal(ForkJoinTask<?> task, int origin) { + WorkQueue[] ws; WorkQueue w; Thread p; long c; int m, u, e, i, s; + if (task != null && task.status >= 0 && + (u = (int)(ctl >>> 32)) < 0 && (u >> UAC_SHIFT) < 0 && + (ws = workQueues) != null && (m = ws.length - 1) >= 0) { + outer: for (int k = origin, j = m; j >= 0; --j) { + WorkQueue q = ws[k++ & m]; + for (int n = m;;) { // limit to at most m signals + if (task.status < 0) + break outer; + if (q == null || + ((s = -q.base + q.top) <= n && (n = s) <= 0)) + break; + if ((u = (int)((c = ctl) >>> 32)) >= 0 || + (e = (int)c) <= 0 || m < (i = e & SMASK) || + (w = ws[i]) == null) + break outer; + long nc = (((long)(w.nextWait & E_MASK)) | + ((long)(u + UAC_UNIT) << 32)); + if (w.eventCount != (e | INT_SIGN)) + break outer; + if (U.compareAndSwapLong(this, CTL, c, nc)) { + w.eventCount = (e + E_SEQ) & E_MASK; + if ((p = w.parker) != null) + U.unpark(p); + if (--n <= 0) + break; + } + } + } + } + } + + /** * Tries to locate and execute tasks for a stealer of the given * task, or in turn one of its stealers, Traces currentSteal -> * currentJoin links looking for a thread working on a descendant @@ -1669,7 +2220,7 @@ public class ForkJoinPool extends AbstractExecutorService { } if ((ws = workQueues) == null || (m = ws.length - 1) <= 0) break restart; // shutting down - if ((v = ws[h = (j.stealHint | 1) & m]) == null || + if ((v = ws[h = (j.hint | 1) & m]) == null || v.currentSteal != subtask) { for (int origin = h;;) { // find stealer if (((h = (h + 2) & m) & 15) == 1 && @@ -1677,7 +2228,7 @@ public class ForkJoinPool extends AbstractExecutorService { continue restart; // occasional staleness check if ((v = ws[h]) != null && v.currentSteal == subtask) { - j.stealHint = h; // save hint + j.hint = h; // save hint break; } if (h == origin) @@ -1725,88 +2276,77 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * If task is at base of some steal queue, steals and executes it. + * Analog of tryHelpStealer for CountedCompleters. Tries to steal + * and run tasks within the target's computation. * - * @param joiner the joining worker - * @param task the task - */ - private void tryPollForAndExec(WorkQueue joiner, ForkJoinTask<?> task) { - WorkQueue[] ws; - if ((ws = workQueues) != null) { - for (int j = 1; j < ws.length && task.status >= 0; j += 2) { - WorkQueue q = ws[j]; - if (q != null && q.pollFor(task)) { - joiner.runSubtask(task); - break; + * @param task the task to join + * @param mode if shared, exit upon completing any task + * if all workers are active + */ + private int helpComplete(ForkJoinTask<?> task, int mode) { + WorkQueue[] ws; WorkQueue q; int m, n, s, u; + if (task != null && (ws = workQueues) != null && + (m = ws.length - 1) >= 0) { + for (int j = 1, origin = j;;) { + if ((s = task.status) < 0) + return s; + if ((q = ws[j & m]) != null && q.pollAndExecCC(task)) { + origin = j; + if (mode == SHARED_QUEUE && + ((u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0)) + break; } + else if ((j = (j + 2) & m) == origin) + break; } } + return 0; } /** * Tries to decrement active count (sometimes implicitly) and * possibly release or create a compensating worker in preparation * for blocking. Fails on contention or termination. Otherwise, - * adds a new thread if no idle workers are available and either - * pool would become completely starved or: (at least half - * starved, and fewer than 50% spares exist, and there is at least - * one task apparently available). Even though the availability - * check requires a full scan, it is worthwhile in reducing false - * alarms. - * - * @param task if non-null, a task being waited for - * @param blocker if non-null, a blocker being waited for - * @return true if the caller can block, else should recheck and retry - */ - final boolean tryCompensate(ForkJoinTask<?> task, ManagedBlocker blocker) { - int pc = parallelism, e; - long c = ctl; - WorkQueue[] ws = workQueues; - if ((e = (int)c) >= 0 && ws != null) { - int u, a, ac, hc; - int tc = (short)((u = (int)(c >>> 32)) >>> UTC_SHIFT) + pc; - boolean replace = false; - if ((a = u >> UAC_SHIFT) <= 0) { - if ((ac = a + pc) <= 1) - replace = true; - else if ((e > 0 || (task != null && - ac <= (hc = pc >>> 1) && tc < pc + hc))) { - WorkQueue w; - for (int j = 0; j < ws.length; ++j) { - if ((w = ws[j]) != null && !w.isEmpty()) { - replace = true; - break; // in compensation range and tasks available - } - } + * adds a new thread if no idle workers are available and pool + * may become starved. + */ + final boolean tryCompensate() { + int pc = config & SMASK, e, i, tc; long c; + WorkQueue[] ws; WorkQueue w; Thread p; + if ((ws = workQueues) != null && (e = (int)(c = ctl)) >= 0) { + if (e != 0 && (i = e & SMASK) < ws.length && + (w = ws[i]) != null && w.eventCount == (e | INT_SIGN)) { + long nc = ((long)(w.nextWait & E_MASK) | + (c & (AC_MASK|TC_MASK))); + if (U.compareAndSwapLong(this, CTL, c, nc)) { + w.eventCount = (e + E_SEQ) & E_MASK; + if ((p = w.parker) != null) + U.unpark(p); + return true; // replace with idle worker } } - if ((task == null || task.status >= 0) && // recheck need to block - (blocker == null || !blocker.isReleasable()) && ctl == c) { - if (!replace) { // no compensation - long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK); - if (U.compareAndSwapLong(this, CTL, c, nc)) - return true; - } - else if (e != 0) { // release an idle worker - WorkQueue w; Thread p; int i; - if ((i = e & SMASK) < ws.length && (w = ws[i]) != null) { - long nc = ((long)(w.nextWait & E_MASK) | - (c & (AC_MASK|TC_MASK))); - if (w.eventCount == (e | INT_SIGN) && - U.compareAndSwapLong(this, CTL, c, nc)) { - w.eventCount = (e + E_SEQ) & E_MASK; - if ((p = w.parker) != null) - U.unpark(p); + else if ((tc = (short)(c >>> TC_SHIFT)) >= 0 && + (int)(c >> AC_SHIFT) + pc > 1) { + long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK); + if (U.compareAndSwapLong(this, CTL, c, nc)) + return true; // no compensation + } + else if (tc + pc < MAX_CAP) { + long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); + if (U.compareAndSwapLong(this, CTL, c, nc)) { + ForkJoinWorkerThreadFactory fac; + Throwable ex = null; + ForkJoinWorkerThread wt = null; + try { + if ((fac = factory) != null && + (wt = fac.newThread(this)) != null) { + wt.start(); return true; } + } catch (Throwable rex) { + ex = rex; } - } - else if (tc < MAX_CAP) { // create replacement - long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); - if (U.compareAndSwapLong(this, CTL, c, nc)) { - addWorker(); - return true; - } + deregisterWorker(wt, ex); // clean up and return false } } } @@ -1821,25 +2361,25 @@ public class ForkJoinPool extends AbstractExecutorService { * @return task status on exit */ final int awaitJoin(WorkQueue joiner, ForkJoinTask<?> task) { - int s; - if ((s = task.status) >= 0) { + int s = 0; + if (joiner != null && task != null && (s = task.status) >= 0) { ForkJoinTask<?> prevJoin = joiner.currentJoin; joiner.currentJoin = task; - long startTime = 0L; - for (int k = 0;;) { - if ((s = (joiner.isEmpty() ? // try to help - tryHelpStealer(joiner, task) : - joiner.tryRemoveAndExec(task))) == 0 && + do {} while ((s = task.status) >= 0 && !joiner.isEmpty() && + joiner.tryRemoveAndExec(task)); // process local tasks + if (s >= 0 && (s = task.status) >= 0) { + helpSignal(task, joiner.poolIndex); + if ((s = task.status) >= 0 && + (task instanceof CountedCompleter)) + s = helpComplete(task, LIFO_QUEUE); + } + while (s >= 0 && (s = task.status) >= 0) { + if ((!joiner.isEmpty() || // try helping + (s = tryHelpStealer(joiner, task)) == 0) && (s = task.status) >= 0) { - if (k == 0) { - startTime = System.nanoTime(); - tryPollForAndExec(joiner, task); // check uncommon case - } - else if ((k & (MAX_HELP - 1)) == 0 && - System.nanoTime() - startTime >= - COMPENSATION_DELAY && - tryCompensate(task, null)) { - if (task.trySetSignal()) { + helpSignal(task, joiner.poolIndex); + if ((s = task.status) >= 0 && tryCompensate()) { + if (task.trySetSignal() && (s = task.status) >= 0) { synchronized (task) { if (task.status >= 0) { try { // see ForkJoinTask @@ -1856,13 +2396,8 @@ public class ForkJoinPool extends AbstractExecutorService { (this, CTL, c = ctl, c + AC_UNIT)); } } - if (s < 0 || (s = task.status) < 0) { - joiner.currentJoin = prevJoin; - break; - } - else if ((k++ & (MAX_HELP - 1)) == MAX_HELP >>> 1) - Thread.yield(); // for politeness } + joiner.currentJoin = prevJoin; } return s; } @@ -1874,46 +2409,49 @@ public class ForkJoinPool extends AbstractExecutorService { * * @param joiner the joining worker * @param task the task - * @return task status on exit */ - final int helpJoinOnce(WorkQueue joiner, ForkJoinTask<?> task) { + final void helpJoinOnce(WorkQueue joiner, ForkJoinTask<?> task) { int s; - while ((s = task.status) >= 0 && - (joiner.isEmpty() ? - tryHelpStealer(joiner, task) : - joiner.tryRemoveAndExec(task)) != 0) - ; - return s; + if (joiner != null && task != null && (s = task.status) >= 0) { + ForkJoinTask<?> prevJoin = joiner.currentJoin; + joiner.currentJoin = task; + do {} while ((s = task.status) >= 0 && !joiner.isEmpty() && + joiner.tryRemoveAndExec(task)); + if (s >= 0 && (s = task.status) >= 0) { + helpSignal(task, joiner.poolIndex); + if ((s = task.status) >= 0 && + (task instanceof CountedCompleter)) + s = helpComplete(task, LIFO_QUEUE); + } + if (s >= 0 && joiner.isEmpty()) { + do {} while (task.status >= 0 && + tryHelpStealer(joiner, task) > 0); + } + joiner.currentJoin = prevJoin; + } } /** * Returns a (probably) non-empty steal queue, if one is found - * during a random, then cyclic scan, else null. This method must - * be retried by caller if, by the time it tries to use the queue, - * it is empty. - */ - private WorkQueue findNonEmptyStealQueue(WorkQueue w) { - // Similar to loop in scan(), but ignoring submissions - int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; - int step = (r >>> 16) | 1; - for (WorkQueue[] ws;;) { - int rs = runState, m; - if ((ws = workQueues) == null || (m = ws.length - 1) < 1) - return null; - for (int j = (m + 1) << 2; ; r += step) { - WorkQueue q = ws[((r << 1) | 1) & m]; - if (q != null && !q.isEmpty()) - return q; - else if (--j < 0) { - if (runState == rs) - return null; - break; + * during a scan, else null. This method must be retried by + * caller if, by the time it tries to use the queue, it is empty. + * @param r a (random) seed for scanning + */ + private WorkQueue findNonEmptyStealQueue(int r) { + for (;;) { + int ps = plock, m; WorkQueue[] ws; WorkQueue q; + if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) { + for (int j = (m + 1) << 2; j >= 0; --j) { + if ((q = ws[(((r + j) << 1) | 1) & m]) != null && + q.base - q.top < 0) + return q; } } + if (plock == ps) + return null; } } - /** * Runs tasks until {@code isQuiescent()}. We piggyback on * active count ctl maintenance, but rather than blocking @@ -1922,36 +2460,34 @@ public class ForkJoinPool extends AbstractExecutorService { */ final void helpQuiescePool(WorkQueue w) { for (boolean active = true;;) { - ForkJoinTask<?> localTask; // exhaust local queue - while ((localTask = w.nextLocalTask()) != null) - localTask.doExec(); - WorkQueue q = findNonEmptyStealQueue(w); - if (q != null) { - ForkJoinTask<?> t; int b; + long c; WorkQueue q; ForkJoinTask<?> t; int b; + while ((t = w.nextLocalTask()) != null) { + if (w.base - w.top < 0) + signalWork(w); + t.doExec(); + } + if ((q = findNonEmptyStealQueue(w.nextSeed())) != null) { if (!active) { // re-establish active count - long c; active = true; do {} while (!U.compareAndSwapLong (this, CTL, c = ctl, c + AC_UNIT)); } - if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) + if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) { + if (q.base - q.top < 0) + signalWork(q); w.runSubtask(t); + } } - else { - long c; - if (active) { // decrement active count without queuing + else if (active) { // decrement active count without queuing + long nc = (c = ctl) - AC_UNIT; + if ((int)(nc >> AC_SHIFT) + (config & SMASK) == 0) + return; // bypass decrement-then-increment + if (U.compareAndSwapLong(this, CTL, c, nc)) active = false; - do {} while (!U.compareAndSwapLong - (this, CTL, c = ctl, c -= AC_UNIT)); - } - else - c = ctl; // re-increment on exit - if ((int)(c >> AC_SHIFT) + parallelism == 0) { - do {} while (!U.compareAndSwapLong - (this, CTL, c = ctl, c + AC_UNIT)); - break; - } } + else if ((int)((c = ctl) >> AC_SHIFT) + (config & SMASK) == 0 && + U.compareAndSwapLong(this, CTL, c, c + AC_UNIT)) + return; } } @@ -1965,27 +2501,75 @@ public class ForkJoinPool extends AbstractExecutorService { WorkQueue q; int b; if ((t = w.nextLocalTask()) != null) return t; - if ((q = findNonEmptyStealQueue(w)) == null) + if ((q = findNonEmptyStealQueue(w.nextSeed())) == null) return null; - if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) + if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) { + if (q.base - q.top < 0) + signalWork(q); return t; + } } } /** - * Returns the approximate (non-atomic) number of idle threads per - * active thread to offset steal queue size for method - * ForkJoinTask.getSurplusQueuedTaskCount(). - */ - final int idlePerActive() { - // Approximate at powers of two for small values, saturate past 4 - int p = parallelism; - int a = p + (int)(ctl >> AC_SHIFT); - return (a > (p >>>= 1) ? 0 : - a > (p >>>= 1) ? 1 : - a > (p >>>= 1) ? 2 : - a > (p >>>= 1) ? 4 : - 8); + * Returns a cheap heuristic guide for task partitioning when + * programmers, frameworks, tools, or languages have little or no + * idea about task granularity. In essence by offering this + * method, we ask users only about tradeoffs in overhead vs + * expected throughput and its variance, rather than how finely to + * partition tasks. + * + * In a steady state strict (tree-structured) computation, each + * thread makes available for stealing enough tasks for other + * threads to remain active. Inductively, if all threads play by + * the same rules, each thread should make available only a + * constant number of tasks. + * + * The minimum useful constant is just 1. But using a value of 1 + * would require immediate replenishment upon each steal to + * maintain enough tasks, which is infeasible. Further, + * partitionings/granularities of offered tasks should minimize + * steal rates, which in general means that threads nearer the top + * of computation tree should generate more than those nearer the + * bottom. In perfect steady state, each thread is at + * approximately the same level of computation tree. However, + * producing extra tasks amortizes the uncertainty of progress and + * diffusion assumptions. + * + * So, users will want to use values larger (but not much larger) + * than 1 to both smooth over transient shortages and hedge + * against uneven progress; as traded off against the cost of + * extra task overhead. We leave the user to pick a threshold + * value to compare with the results of this call to guide + * decisions, but recommend values such as 3. + * + * When all threads are active, it is on average OK to estimate + * surplus strictly locally. In steady-state, if one thread is + * maintaining say 2 surplus tasks, then so are others. So we can + * just use estimated queue length. However, this strategy alone + * leads to serious mis-estimates in some non-steady-state + * conditions (ramp-up, ramp-down, other stalls). We can detect + * many of these by further considering the number of "idle" + * threads, that are known to have zero queued tasks, so + * compensate by a factor of (#idle/#active) threads. + * + * Note: The approximation of #busy workers as #active workers is + * not very good under current signalling scheme, and should be + * improved. + */ + static int getSurplusQueuedTaskCount() { + Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q; + if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) { + int p = (pool = (wt = (ForkJoinWorkerThread)t).pool).config & SMASK; + int n = (q = wt.workQueue).top - q.base; + int a = (int)(pool.ctl >> AC_SHIFT) + p; + return n - (a > (p >>>= 1) ? 0 : + a > (p >>>= 1) ? 1 : + a > (p >>>= 1) ? 2 : + a > (p >>>= 1) ? 4 : + 8); + } + return 0; } // Termination @@ -2005,56 +2589,71 @@ public class ForkJoinPool extends AbstractExecutorService { * @return true if now terminating or terminated */ private boolean tryTerminate(boolean now, boolean enable) { - Mutex lock = this.lock; + int ps; + if (this == common) // cannot shut down + return false; + if ((ps = plock) >= 0) { // enable by setting plock + if (!enable) + return false; + if ((ps & PL_LOCK) != 0 || + !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) + ps = acquirePlock(); + int nps = ((ps + PL_LOCK) & ~SHUTDOWN) | SHUTDOWN; + if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) + releasePlock(nps); + } for (long c;;) { - if (((c = ctl) & STOP_BIT) != 0) { // already terminating - if ((short)(c >>> TC_SHIFT) == -parallelism) { - lock.lock(); // don't need try/finally - termination.signalAll(); // signal when 0 workers - lock.unlock(); + if (((c = ctl) & STOP_BIT) != 0) { // already terminating + if ((short)(c >>> TC_SHIFT) == -(config & SMASK)) { + synchronized (this) { + notifyAll(); // signal when 0 workers + } } return true; } - if (runState >= 0) { // not yet enabled - if (!enable) + if (!now) { // check if idle & no tasks + WorkQueue[] ws; WorkQueue w; + if ((int)(c >> AC_SHIFT) != -(config & SMASK)) return false; - lock.lock(); - runState |= SHUTDOWN; - lock.unlock(); - } - if (!now) { // check if idle & no tasks - if ((int)(c >> AC_SHIFT) != -parallelism || - hasQueuedSubmissions()) - return false; - // Check for unqueued inactive workers. One pass suffices. - WorkQueue[] ws = workQueues; WorkQueue w; - if (ws != null) { - for (int i = 1; i < ws.length; i += 2) { - if ((w = ws[i]) != null && w.eventCount >= 0) - return false; + if ((ws = workQueues) != null) { + for (int i = 0; i < ws.length; ++i) { + if ((w = ws[i]) != null) { + if (!w.isEmpty()) { // signal unprocessed tasks + signalWork(w); + return false; + } + if ((i & 1) != 0 && w.eventCount >= 0) + return false; // unqueued inactive worker + } } } } if (U.compareAndSwapLong(this, CTL, c, c | STOP_BIT)) { for (int pass = 0; pass < 3; ++pass) { - WorkQueue[] ws = workQueues; - if (ws != null) { - WorkQueue w; + WorkQueue[] ws; WorkQueue w; Thread wt; + if ((ws = workQueues) != null) { int n = ws.length; for (int i = 0; i < n; ++i) { if ((w = ws[i]) != null) { - w.runState = -1; + w.qlock = -1; if (pass > 0) { w.cancelAll(); - if (pass > 1) - w.interruptOwner(); + if (pass > 1 && (wt = w.owner) != null) { + if (!wt.isInterrupted()) { + try { + wt.interrupt(); + } catch (Throwable ignore) { + } + } + U.unpark(wt); + } } } } // Wake up workers parked on event queue int i, e; long cc; Thread p; while ((e = (int)(cc = ctl) & E_MASK) != 0 && - (i = e & SMASK) < n && + (i = e & SMASK) < n && i >= 0 && (w = ws[i]) != null) { long nc = ((long)(w.nextWait & E_MASK) | ((cc + AC_UNIT) & AC_MASK) | @@ -2062,7 +2661,7 @@ public class ForkJoinPool extends AbstractExecutorService { if (w.eventCount == (e | INT_SIGN) && U.compareAndSwapLong(this, CTL, cc, nc)) { w.eventCount = (e + E_SEQ) & E_MASK; - w.runState = -1; + w.qlock = -1; if ((p = w.parker) != null) U.unpark(p); } @@ -2073,6 +2672,135 @@ public class ForkJoinPool extends AbstractExecutorService { } } + // external operations on common pool + + /** + * Returns common pool queue for a thread that has submitted at + * least one task. + */ + static WorkQueue commonSubmitterQueue() { + ForkJoinPool p; WorkQueue[] ws; int m; Submitter z; + return ((z = submitters.get()) != null && + (p = common) != null && + (ws = p.workQueues) != null && + (m = ws.length - 1) >= 0) ? + ws[m & z.seed & SQMASK] : null; + } + + /** + * Tries to pop the given task from submitter's queue in common pool. + */ + static boolean tryExternalUnpush(ForkJoinTask<?> t) { + ForkJoinPool p; WorkQueue[] ws; WorkQueue q; Submitter z; + ForkJoinTask<?>[] a; int m, s; + if (t != null && + (z = submitters.get()) != null && + (p = common) != null && + (ws = p.workQueues) != null && + (m = ws.length - 1) >= 0 && + (q = ws[m & z.seed & SQMASK]) != null && + (s = q.top) != q.base && + (a = q.array) != null) { + long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE; + if (U.getObject(a, j) == t && + U.compareAndSwapInt(q, QLOCK, 0, 1)) { + if (q.array == a && q.top == s && // recheck + U.compareAndSwapObject(a, j, t, null)) { + q.top = s - 1; + q.qlock = 0; + return true; + } + q.qlock = 0; + } + } + return false; + } + + /** + * Tries to pop and run local tasks within the same computation + * as the given root. On failure, tries to help complete from + * other queues via helpComplete. + */ + private void externalHelpComplete(WorkQueue q, ForkJoinTask<?> root) { + ForkJoinTask<?>[] a; int m; + if (q != null && (a = q.array) != null && (m = (a.length - 1)) >= 0 && + root != null && root.status >= 0) { + for (;;) { + int s, u; Object o; CountedCompleter<?> task = null; + if ((s = q.top) - q.base > 0) { + long j = ((m & (s - 1)) << ASHIFT) + ABASE; + if ((o = U.getObject(a, j)) != null && + (o instanceof CountedCompleter)) { + CountedCompleter<?> t = (CountedCompleter<?>)o, r = t; + do { + if (r == root) { + if (U.compareAndSwapInt(q, QLOCK, 0, 1)) { + if (q.array == a && q.top == s && + U.compareAndSwapObject(a, j, t, null)) { + q.top = s - 1; + task = t; + } + q.qlock = 0; + } + break; + } + } while ((r = r.completer) != null); + } + } + if (task != null) + task.doExec(); + if (root.status < 0 || + (u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0) + break; + if (task == null) { + helpSignal(root, q.poolIndex); + if (root.status >= 0) + helpComplete(root, SHARED_QUEUE); + break; + } + } + } + } + + /** + * Tries to help execute or signal availability of the given task + * from submitter's queue in common pool. + */ + static void externalHelpJoin(ForkJoinTask<?> t) { + // Some hard-to-avoid overlap with tryExternalUnpush + ForkJoinPool p; WorkQueue[] ws; WorkQueue q, w; Submitter z; + ForkJoinTask<?>[] a; int m, s, n; + if (t != null && + (z = submitters.get()) != null && + (p = common) != null && + (ws = p.workQueues) != null && + (m = ws.length - 1) >= 0 && + (q = ws[m & z.seed & SQMASK]) != null && + (a = q.array) != null) { + int am = a.length - 1; + if ((s = q.top) != q.base) { + long j = ((am & (s - 1)) << ASHIFT) + ABASE; + if (U.getObject(a, j) == t && + U.compareAndSwapInt(q, QLOCK, 0, 1)) { + if (q.array == a && q.top == s && + U.compareAndSwapObject(a, j, t, null)) { + q.top = s - 1; + q.qlock = 0; + t.doExec(); + } + else + q.qlock = 0; + } + } + if (t.status >= 0) { + if (t instanceof CountedCompleter) + p.externalHelpComplete(q, t); + else + p.helpSignal(t, q.poolIndex); + } + } + } + // Exported methods // Constructors @@ -2089,7 +2817,7 @@ public class ForkJoinPool extends AbstractExecutorService { * java.lang.RuntimePermission}{@code ("modifyThread")} */ public ForkJoinPool() { - this(Runtime.getRuntime().availableProcessors(), + this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), defaultForkJoinWorkerThreadFactory, null, false); } @@ -2144,29 +2872,48 @@ public class ForkJoinPool extends AbstractExecutorService { throw new NullPointerException(); if (parallelism <= 0 || parallelism > MAX_CAP) throw new IllegalArgumentException(); - this.parallelism = parallelism; this.factory = factory; this.ueh = handler; - this.localMode = asyncMode ? FIFO_QUEUE : LIFO_QUEUE; + this.config = parallelism | (asyncMode ? (FIFO_QUEUE << 16) : 0); long np = (long)(-parallelism); // offset ctl counts this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); - // Use nearest power 2 for workQueues size. See Hackers Delight sec 3.2. - int n = parallelism - 1; - n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16; - int size = (n + 1) << 1; // #slots = 2*#workers - this.submitMask = size - 1; // room for max # of submit queues - this.workQueues = new WorkQueue[size]; - this.termination = (this.lock = new Mutex()).newCondition(); - this.stealCount = new AtomicLong(); - this.nextWorkerNumber = new AtomicInteger(); - int pn = poolNumberGenerator.incrementAndGet(); + int pn = nextPoolId(); StringBuilder sb = new StringBuilder("ForkJoinPool-"); sb.append(Integer.toString(pn)); sb.append("-worker-"); this.workerNamePrefix = sb.toString(); - lock.lock(); - this.runState = 1; // set init flag - lock.unlock(); + } + + /** + * Constructor for common pool, suitable only for static initialization. + * Basically the same as above, but uses smallest possible initial footprint. + */ + ForkJoinPool(int parallelism, long ctl, + ForkJoinWorkerThreadFactory factory, + Thread.UncaughtExceptionHandler handler) { + this.config = parallelism; + this.ctl = ctl; + this.factory = factory; + this.ueh = handler; + this.workerNamePrefix = "ForkJoinPool.commonPool-worker-"; + } + + /** + * Returns the common pool instance. This pool is statically + * constructed; its run state is unaffected by attempts to {@link + * #shutdown} or {@link #shutdownNow}. However this pool and any + * ongoing processing are automatically terminated upon program + * {@link System#exit}. Any program that relies on asynchronous + * task processing to complete before program termination should + * invoke {@code commonPool().}{@link #awaitQuiescence}, before + * exit. + * + * @return the common pool instance + * @since 1.8 + */ + public static ForkJoinPool commonPool() { + // assert common != null : "static init error"; + return common; } // Execution methods @@ -2190,7 +2937,7 @@ public class ForkJoinPool extends AbstractExecutorService { public <T> T invoke(ForkJoinTask<T> task) { if (task == null) throw new NullPointerException(); - doSubmit(task); + externalPush(task); return task.join(); } @@ -2205,7 +2952,7 @@ public class ForkJoinPool extends AbstractExecutorService { public void execute(ForkJoinTask<?> task) { if (task == null) throw new NullPointerException(); - doSubmit(task); + externalPush(task); } // AbstractExecutorService methods @@ -2223,7 +2970,7 @@ public class ForkJoinPool extends AbstractExecutorService { job = (ForkJoinTask<?>) task; else job = new ForkJoinTask.AdaptedRunnableAction(task); - doSubmit(job); + externalPush(job); } /** @@ -2238,7 +2985,7 @@ public class ForkJoinPool extends AbstractExecutorService { public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) { if (task == null) throw new NullPointerException(); - doSubmit(task); + externalPush(task); return task; } @@ -2249,7 +2996,7 @@ public class ForkJoinPool extends AbstractExecutorService { */ public <T> ForkJoinTask<T> submit(Callable<T> task) { ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task); - doSubmit(job); + externalPush(job); return job; } @@ -2260,7 +3007,7 @@ public class ForkJoinPool extends AbstractExecutorService { */ public <T> ForkJoinTask<T> submit(Runnable task, T result) { ForkJoinTask<T> job = new ForkJoinTask.AdaptedRunnable<T>(task, result); - doSubmit(job); + externalPush(job); return job; } @@ -2277,7 +3024,7 @@ public class ForkJoinPool extends AbstractExecutorService { job = (ForkJoinTask<?>) task; else job = new ForkJoinTask.AdaptedRunnableAction(task); - doSubmit(job); + externalPush(job); return job; } @@ -2289,27 +3036,23 @@ public class ForkJoinPool extends AbstractExecutorService { // In previous versions of this class, this method constructed // a task to run ForkJoinTask.invokeAll, but now external // invocation of multiple tasks is at least as efficient. - List<ForkJoinTask<T>> fs = new ArrayList<ForkJoinTask<T>>(tasks.size()); - // Workaround needed because method wasn't declared with - // wildcards in return type but should have been. - @SuppressWarnings({"unchecked", "rawtypes"}) - List<Future<T>> futures = (List<Future<T>>) (List) fs; + ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size()); boolean done = false; try { for (Callable<T> t : tasks) { ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t); - doSubmit(f); - fs.add(f); + futures.add(f); + externalPush(f); } - for (ForkJoinTask<T> f : fs) - f.quietlyJoin(); + for (int i = 0, size = futures.size(); i < size; i++) + ((ForkJoinTask<?>)futures.get(i)).quietlyJoin(); done = true; return futures; } finally { if (!done) - for (ForkJoinTask<T> f : fs) - f.cancel(false); + for (int i = 0, size = futures.size(); i < size; i++) + futures.get(i).cancel(false); } } @@ -2338,7 +3081,17 @@ public class ForkJoinPool extends AbstractExecutorService { * @return the targeted parallelism level of this pool */ public int getParallelism() { - return parallelism; + return config & SMASK; + } + + /** + * Returns the targeted parallelism level of the common pool. + * + * @return the targeted parallelism level of the common pool + * @since 1.8 + */ + public static int getCommonPoolParallelism() { + return commonParallelism; } /** @@ -2350,7 +3103,7 @@ public class ForkJoinPool extends AbstractExecutorService { * @return the number of worker threads */ public int getPoolSize() { - return parallelism + (short)(ctl >>> TC_SHIFT); + return (config & SMASK) + (short)(ctl >>> TC_SHIFT); } /** @@ -2360,7 +3113,7 @@ public class ForkJoinPool extends AbstractExecutorService { * @return {@code true} if this pool uses async mode */ public boolean getAsyncMode() { - return localMode != 0; + return (config >>> 16) == FIFO_QUEUE; } /** @@ -2391,7 +3144,7 @@ public class ForkJoinPool extends AbstractExecutorService { * @return the number of active threads */ public int getActiveThreadCount() { - int r = parallelism + (int)(ctl >> AC_SHIFT); + int r = (config & SMASK) + (int)(ctl >> AC_SHIFT); return (r <= 0) ? 0 : r; // suppress momentarily negative values } @@ -2407,7 +3160,7 @@ public class ForkJoinPool extends AbstractExecutorService { * @return {@code true} if all threads are currently idle */ public boolean isQuiescent() { - return (int)(ctl >> AC_SHIFT) + parallelism == 0; + return (int)(ctl >> AC_SHIFT) + (config & SMASK) == 0; } /** @@ -2422,12 +3175,12 @@ public class ForkJoinPool extends AbstractExecutorService { * @return the number of steals */ public long getStealCount() { - long count = stealCount.get(); + long count = stealCount; WorkQueue[] ws; WorkQueue w; if ((ws = workQueues) != null) { for (int i = 1; i < ws.length; i += 2) { if ((w = ws[i]) != null) - count += w.totalSteals; + count += w.nsteals; } } return count; @@ -2552,7 +3305,7 @@ public class ForkJoinPool extends AbstractExecutorService { public String toString() { // Use a single pass through workQueues to collect counts long qt = 0L, qs = 0L; int rc = 0; - long st = stealCount.get(); + long st = stealCount; long c = ctl; WorkQueue[] ws; WorkQueue w; if ((ws = workQueues) != null) { @@ -2563,14 +3316,14 @@ public class ForkJoinPool extends AbstractExecutorService { qs += size; else { qt += size; - st += w.totalSteals; + st += w.nsteals; if (w.isApparentlyUnblocked()) ++rc; } } } } - int pc = parallelism; + int pc = (config & SMASK); int tc = pc + (short)(c >>> TC_SHIFT); int ac = pc + (int)(c >> AC_SHIFT); if (ac < 0) // ignore transient negative @@ -2579,7 +3332,7 @@ public class ForkJoinPool extends AbstractExecutorService { if ((c & STOP_BIT) != 0) level = (tc == 0) ? "Terminated" : "Terminating"; else - level = runState < 0 ? "Shutting down" : "Running"; + level = plock < 0 ? "Shutting down" : "Running"; return super.toString() + "[" + level + ", parallelism = " + pc + @@ -2593,11 +3346,13 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * Initiates an orderly shutdown in which previously submitted - * tasks are executed, but no new tasks will be accepted. - * Invocation has no additional effect if already shut down. - * Tasks that are in the process of being submitted concurrently - * during the course of this method may or may not be rejected. + * Possibly initiates an orderly shutdown in which previously + * submitted tasks are executed, but no new tasks will be + * accepted. Invocation has no effect on execution state if this + * is the {@link #commonPool()}, and no additional effect if + * already shut down. Tasks that are in the process of being + * submitted concurrently during the course of this method may or + * may not be rejected. * * @throws SecurityException if a security manager exists and * the caller is not permitted to modify threads @@ -2610,14 +3365,16 @@ public class ForkJoinPool extends AbstractExecutorService { } /** - * Attempts to cancel and/or stop all tasks, and reject all - * subsequently submitted tasks. Tasks that are in the process of - * being submitted or executed concurrently during the course of - * this method may or may not be rejected. This method cancels - * both existing and unexecuted tasks, in order to permit - * termination in the presence of task dependencies. So the method - * always returns an empty list (unlike the case for some other - * Executors). + * Possibly attempts to cancel and/or stop all tasks, and reject + * all subsequently submitted tasks. Invocation has no effect on + * execution state if this is the {@link #commonPool()}, and no + * additional effect if already shut down. Otherwise, tasks that + * are in the process of being submitted or executed concurrently + * during the course of this method may or may not be + * rejected. This method cancels both existing and unexecuted + * tasks, in order to permit termination in the presence of task + * dependencies. So the method always returns an empty list + * (unlike the case for some other Executors). * * @return an empty list * @throws SecurityException if a security manager exists and @@ -2639,7 +3396,7 @@ public class ForkJoinPool extends AbstractExecutorService { public boolean isTerminated() { long c = ctl; return ((c & STOP_BIT) != 0L && - (short)(c >>> TC_SHIFT) == -parallelism); + (short)(c >>> TC_SHIFT) == -(config & SMASK)); } /** @@ -2647,7 +3404,7 @@ public class ForkJoinPool extends AbstractExecutorService { * commenced but not yet completed. This method may be useful for * debugging. A return of {@code true} reported a sufficient * period after shutdown may indicate that submitted tasks have - * ignored or suppressed interruption, or are waiting for IO, + * ignored or suppressed interruption, or are waiting for I/O, * causing this executor not to properly terminate. (See the * advisory notes for class {@link ForkJoinTask} stating that * tasks should not normally entail blocking operations. But if @@ -2658,7 +3415,7 @@ public class ForkJoinPool extends AbstractExecutorService { public boolean isTerminating() { long c = ctl; return ((c & STOP_BIT) != 0L && - (short)(c >>> TC_SHIFT) != -parallelism); + (short)(c >>> TC_SHIFT) != -(config & SMASK)); } /** @@ -2667,13 +3424,16 @@ public class ForkJoinPool extends AbstractExecutorService { * @return {@code true} if this pool has been shut down */ public boolean isShutdown() { - return runState < 0; + return plock < 0; } /** - * Blocks until all tasks have completed execution after a shutdown - * request, or the timeout occurs, or the current thread is - * interrupted, whichever happens first. + * Blocks until all tasks have completed execution after a + * shutdown request, or the timeout occurs, or the current thread + * is interrupted, whichever happens first. Because the {@link + * #commonPool()} never terminates until program shutdown, when + * applied to the common pool, this method is equivalent to {@link + * #awaitQuiescence} but always returns {@code false}. * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument @@ -2683,20 +3443,84 @@ public class ForkJoinPool extends AbstractExecutorService { */ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + if (Thread.interrupted()) + throw new InterruptedException(); + if (this == common) { + awaitQuiescence(timeout, unit); + return false; + } long nanos = unit.toNanos(timeout); - final Mutex lock = this.lock; - lock.lock(); - try { - for (;;) { - if (isTerminated()) - return true; - if (nanos <= 0) + if (isTerminated()) + return true; + long startTime = System.nanoTime(); + boolean terminated = false; + synchronized (this) { + for (long waitTime = nanos, millis = 0L;;) { + if (terminated = isTerminated() || + waitTime <= 0L || + (millis = unit.toMillis(waitTime)) <= 0L) + break; + wait(millis); + waitTime = nanos - (System.nanoTime() - startTime); + } + } + return terminated; + } + + /** + * If called by a ForkJoinTask operating in this pool, equivalent + * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise, + * waits and/or attempts to assist performing tasks until this + * pool {@link #isQuiescent} or the indicated timeout elapses. + * + * @param timeout the maximum time to wait + * @param unit the time unit of the timeout argument + * @return {@code true} if quiescent; {@code false} if the + * timeout elapsed. + */ + public boolean awaitQuiescence(long timeout, TimeUnit unit) { + long nanos = unit.toNanos(timeout); + ForkJoinWorkerThread wt; + Thread thread = Thread.currentThread(); + if ((thread instanceof ForkJoinWorkerThread) && + (wt = (ForkJoinWorkerThread)thread).pool == this) { + helpQuiescePool(wt.workQueue); + return true; + } + long startTime = System.nanoTime(); + WorkQueue[] ws; + int r = 0, m; + boolean found = true; + while (!isQuiescent() && (ws = workQueues) != null && + (m = ws.length - 1) >= 0) { + if (!found) { + if ((System.nanoTime() - startTime) > nanos) return false; - nanos = termination.awaitNanos(nanos); + Thread.yield(); // cannot block + } + found = false; + for (int j = (m + 1) << 2; j >= 0; --j) { + ForkJoinTask<?> t; WorkQueue q; int b; + if ((q = ws[r++ & m]) != null && (b = q.base) - q.top < 0) { + found = true; + if ((t = q.pollAt(b)) != null) { + if (q.base - q.top < 0) + signalWork(q); + t.doExec(); + } + break; + } } - } finally { - lock.unlock(); } + return true; + } + + /** + * Waits and/or attempts to assist performing tasks indefinitely + * until the {@link #commonPool()} {@link #isQuiescent}. + */ + static void quiesceCommonPool() { + common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS); } /** @@ -2795,19 +3619,37 @@ public class ForkJoinPool extends AbstractExecutorService { public static void managedBlock(ManagedBlocker blocker) throws InterruptedException { Thread t = Thread.currentThread(); - ForkJoinPool p = ((t instanceof ForkJoinWorkerThread) ? - ((ForkJoinWorkerThread)t).pool : null); - while (!blocker.isReleasable()) { - if (p == null || p.tryCompensate(null, blocker)) { - try { - do {} while (!blocker.isReleasable() && !blocker.block()); - } finally { - if (p != null) + if (t instanceof ForkJoinWorkerThread) { + ForkJoinPool p = ((ForkJoinWorkerThread)t).pool; + while (!blocker.isReleasable()) { // variant of helpSignal + WorkQueue[] ws; WorkQueue q; int m, u; + if ((ws = p.workQueues) != null && (m = ws.length - 1) >= 0) { + for (int i = 0; i <= m; ++i) { + if (blocker.isReleasable()) + return; + if ((q = ws[i]) != null && q.base - q.top < 0) { + p.signalWork(q); + if ((u = (int)(p.ctl >>> 32)) >= 0 || + (u >> UAC_SHIFT) >= 0) + break; + } + } + } + if (p.tryCompensate()) { + try { + do {} while (!blocker.isReleasable() && + !blocker.block()); + } finally { p.incrementActiveCount(); + } + break; } - break; } } + else { + do {} while (!blocker.isReleasable() && + !blocker.block()); + } } // AbstractExecutorService overrides. These rely on undocumented @@ -2828,32 +3670,80 @@ public class ForkJoinPool extends AbstractExecutorService { private static final long PARKBLOCKER; private static final int ABASE; private static final int ASHIFT; + private static final long STEALCOUNT; + private static final long PLOCK; + private static final long INDEXSEED; + private static final long QLOCK; static { - poolNumberGenerator = new AtomicInteger(); - nextSubmitterSeed = new AtomicInteger(0x55555555); - modifyThreadPermission = new RuntimePermission("modifyThread"); - defaultForkJoinWorkerThreadFactory = - new DefaultForkJoinWorkerThreadFactory(); - submitters = new ThreadSubmitter(); - int s; + // initialize field offsets for CAS etc try { U = getUnsafe(); Class<?> k = ForkJoinPool.class; - Class<?> ak = ForkJoinTask[].class; CTL = U.objectFieldOffset (k.getDeclaredField("ctl")); + STEALCOUNT = U.objectFieldOffset + (k.getDeclaredField("stealCount")); + PLOCK = U.objectFieldOffset + (k.getDeclaredField("plock")); + INDEXSEED = U.objectFieldOffset + (k.getDeclaredField("indexSeed")); Class<?> tk = Thread.class; PARKBLOCKER = U.objectFieldOffset (tk.getDeclaredField("parkBlocker")); + Class<?> wk = WorkQueue.class; + QLOCK = U.objectFieldOffset + (wk.getDeclaredField("qlock")); + Class<?> ak = ForkJoinTask[].class; ABASE = U.arrayBaseOffset(ak); - s = U.arrayIndexScale(ak); + int scale = U.arrayIndexScale(ak); + if ((scale & (scale - 1)) != 0) + throw new Error("data type scale not a power of two"); + ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) { throw new Error(e); } - if ((s & (s-1)) != 0) - throw new Error("data type scale not a power of two"); - ASHIFT = 31 - Integer.numberOfLeadingZeros(s); + + submitters = new ThreadLocal<Submitter>(); + ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory = + new DefaultForkJoinWorkerThreadFactory(); + modifyThreadPermission = new RuntimePermission("modifyThread"); + + /* + * Establish common pool parameters. For extra caution, + * computations to set up common pool state are here; the + * constructor just assigns these values to fields. + */ + + int par = 0; + Thread.UncaughtExceptionHandler handler = null; + try { // TBD: limit or report ignored exceptions? + String pp = System.getProperty + ("java.util.concurrent.ForkJoinPool.common.parallelism"); + String hp = System.getProperty + ("java.util.concurrent.ForkJoinPool.common.exceptionHandler"); + String fp = System.getProperty + ("java.util.concurrent.ForkJoinPool.common.threadFactory"); + if (fp != null) + fac = ((ForkJoinWorkerThreadFactory)ClassLoader. + getSystemClassLoader().loadClass(fp).newInstance()); + if (hp != null) + handler = ((Thread.UncaughtExceptionHandler)ClassLoader. + getSystemClassLoader().loadClass(hp).newInstance()); + if (pp != null) + par = Integer.parseInt(pp); + } catch (Exception ignore) { + } + + if (par <= 0) + par = Runtime.getRuntime().availableProcessors(); + if (par > MAX_CAP) + par = MAX_CAP; + commonParallelism = par; + long np = (long)(-par); // precompute initial ctl value + long ct = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); + + common = new ForkJoinPool(par, ct, fac, handler); } /** @@ -2866,5 +3756,4 @@ public class ForkJoinPool extends AbstractExecutorService { private static sun.misc.Unsafe getUnsafe() { return scala.concurrent.util.Unsafe.instance; } - } diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java index 839fd26b39..fd1e132b07 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java @@ -5,6 +5,7 @@ */ package scala.concurrent.forkjoin; + import java.io.Serializable; import java.util.Collection; import java.util.List; @@ -29,15 +30,18 @@ import java.lang.reflect.Constructor; * subtasks may be hosted by a small number of actual threads in a * ForkJoinPool, at the price of some usage limitations. * - * <p>A "main" {@code ForkJoinTask} begins execution when submitted - * to a {@link ForkJoinPool}. Once started, it will usually in turn - * start other subtasks. As indicated by the name of this class, - * many programs using {@code ForkJoinTask} employ only methods - * {@link #fork} and {@link #join}, or derivatives such as {@link + * <p>A "main" {@code ForkJoinTask} begins execution when it is + * explicitly submitted to a {@link ForkJoinPool}, or, if not already + * engaged in a ForkJoin computation, commenced in the {@link + * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or + * related methods. Once started, it will usually in turn start other + * subtasks. As indicated by the name of this class, many programs + * using {@code ForkJoinTask} employ only methods {@link #fork} and + * {@link #join}, or derivatives such as {@link * #invokeAll(ForkJoinTask...) invokeAll}. However, this class also * provides a number of other methods that can come into play in - * advanced usages, as well as extension mechanics that allow - * support of new forms of fork/join processing. + * advanced usages, as well as extension mechanics that allow support + * of new forms of fork/join processing. * * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}. * The efficiency of {@code ForkJoinTask}s stems from a set of @@ -51,7 +55,7 @@ import java.lang.reflect.Constructor; * minimize other blocking synchronization apart from joining other * tasks or using synchronizers such as Phasers that are advertised to * cooperate with fork/join scheduling. Subdividable tasks should also - * not perform blocking IO, and should ideally access variables that + * not perform blocking I/O, and should ideally access variables that * are completely independent of those accessed by other running * tasks. These guidelines are loosely enforced by not permitting * checked exceptions such as {@code IOExceptions} to be @@ -69,10 +73,11 @@ import java.lang.reflect.Constructor; * <p>It is possible to define and use ForkJoinTasks that may block, * but doing do requires three further considerations: (1) Completion * of few if any <em>other</em> tasks should be dependent on a task - * that blocks on external synchronization or IO. Event-style async - * tasks that are never joined often fall into this category. (2) To - * minimize resource impact, tasks should be small; ideally performing - * only the (possibly) blocking action. (3) Unless the {@link + * that blocks on external synchronization or I/O. Event-style async + * tasks that are never joined (for example, those subclassing {@link + * CountedCompleter}) often fall into this category. (2) To minimize + * resource impact, tasks should be small; ideally performing only the + * (possibly) blocking action. (3) Unless the {@link * ForkJoinPool.ManagedBlocker} API is used, or the number of possibly * blocked tasks is known to be less than the pool's {@link * ForkJoinPool#getParallelism} level, the pool cannot guarantee that @@ -121,13 +126,7 @@ import java.lang.reflect.Constructor; * other actions. Normally, a concrete ForkJoinTask subclass declares * fields comprising its parameters, established in a constructor, and * then defines a {@code compute} method that somehow uses the control - * methods supplied by this base class. While these methods have - * {@code public} access (to allow instances of different task - * subclasses to call each other's methods), some of them may only be - * called from within other ForkJoinTasks (as may be determined using - * method {@link #inForkJoinPool}). Attempts to invoke them in other - * contexts result in exceptions or errors, possibly including {@code - * ClassCastException}. + * methods supplied by this base class. * * <p>Method {@link #join} and its variants are appropriate for use * only when completion dependencies are acyclic; that is, the @@ -138,17 +137,16 @@ import java.lang.reflect.Constructor; * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that * may be of use in constructing custom subclasses for problems that * are not statically structured as DAGs. To support such usages a - * ForkJoinTask may be atomically <em>tagged</em> with a {@code - * short} value using {@link #setForkJoinTaskTag} or {@link + * ForkJoinTask may be atomically <em>tagged</em> with a {@code short} + * value using {@link #setForkJoinTaskTag} or {@link * #compareAndSetForkJoinTaskTag} and checked using {@link - * #getForkJoinTaskTag}. The ForkJoinTask implementation does not - * use these {@code protected} methods or tags for any purpose, but - * they may be of use in the construction of specialized subclasses. - * For example, parallel graph traversals can use the supplied methods - * to avoid revisiting nodes/tasks that have already been processed. - * Also, completion based designs can use them to record that subtasks - * have completed. (Method names for tagging are bulky in part to - * encourage definition of methods that reflect their usage patterns.) + * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use + * these {@code protected} methods or tags for any purpose, but they + * may be of use in the construction of specialized subclasses. For + * example, parallel graph traversals can use the supplied methods to + * avoid revisiting nodes/tasks that have already been processed. + * (Method names for tagging are bulky in part to encourage definition + * of methods that reflect their usage patterns.) * * <p>Most base support methods are {@code final}, to prevent * overriding of implementations that are intrinsically tied to the @@ -286,8 +284,9 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * @return status upon completion */ private int externalAwaitDone() { - boolean interrupted = false; int s; + ForkJoinPool.externalHelpJoin(this); + boolean interrupted = false; while ((s = status) >= 0) { if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { synchronized (this) { @@ -315,6 +314,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { int s; if (Thread.interrupted()) throw new InterruptedException(); + ForkJoinPool.externalHelpJoin(this); while ((s = status) >= 0) { if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { synchronized (this) { @@ -328,6 +328,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { return s; } + /** * Implementation for join, get, quietlyJoin. Directly handles * only cases of already-completed, external wait, and @@ -337,16 +338,12 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { */ private int doJoin() { int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; - if ((s = status) >= 0) { - if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) { - if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue). - tryUnpush(this) || (s = doExec()) >= 0) - s = wt.pool.awaitJoin(w, this); - } - else - s = externalAwaitDone(); - } - return s; + return (s = status) < 0 ? s : + ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (w = (wt = (ForkJoinWorkerThread)t).workQueue). + tryUnpush(this) && (s = doExec()) < 0 ? s : + wt.pool.awaitJoin(w, this) : + externalAwaitDone(); } /** @@ -356,14 +353,10 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { */ private int doInvoke() { int s; Thread t; ForkJoinWorkerThread wt; - if ((s = doExec()) >= 0) { - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) - s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, - this); - else - s = externalAwaitDone(); - } - return s; + return (s = doExec()) < 0 ? s : + ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) : + externalAwaitDone(); } // Exception table support @@ -411,11 +404,11 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } /** - * Records exception and sets exceptional completion. + * Records exception and sets status. * * @return status on exit */ - private int setExceptionalCompletion(Throwable ex) { + final int recordExceptionalCompletion(Throwable ex) { int s; if ((s = status) >= 0) { int h = System.identityHashCode(this); @@ -438,17 +431,25 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } s = setCompletion(EXCEPTIONAL); } - ForkJoinTask<?> p = internalGetCompleter(); // propagate - if (p != null && p.status >= 0) - p.setExceptionalCompletion(ex); return s; } /** - * Exception propagation support for tasks with completers. + * Records exception and possibly propagates. + * + * @return status on exit */ - ForkJoinTask<?> internalGetCompleter() { - return null; + private int setExceptionalCompletion(Throwable ex) { + int s = recordExceptionalCompletion(ex); + if ((s & DONE_MASK) == EXCEPTIONAL) + internalPropagateException(ex); + return s; + } + + /** + * Hook for exception propagation support for tasks with completers. + */ + void internalPropagateException(Throwable ex) { } /** @@ -467,7 +468,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } /** - * Removes exception node and clears status + * Removes exception node and clears status. */ private void clearExceptionalCompletion() { int h = System.identityHashCode(this); @@ -595,7 +596,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } } } - + /** * A version of "sneaky throw" to relay exceptions */ @@ -624,35 +625,35 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * Throws exception, if any, associated with the given status. */ private void reportException(int s) { - Throwable ex = ((s == CANCELLED) ? new CancellationException() : - (s == EXCEPTIONAL) ? getThrowableException() : - null); - if (ex != null) - ForkJoinTask.rethrow(ex); + if (s == CANCELLED) + throw new CancellationException(); + if (s == EXCEPTIONAL) + rethrow(getThrowableException()); } // public methods /** - * Arranges to asynchronously execute this task. While it is not - * necessarily enforced, it is a usage error to fork a task more - * than once unless it has completed and been reinitialized. - * Subsequent modifications to the state of this task or any data - * it operates on are not necessarily consistently observable by - * any thread other than the one executing it unless preceded by a - * call to {@link #join} or related methods, or a call to {@link - * #isDone} returning {@code true}. - * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * Arranges to asynchronously execute this task in the pool the + * current task is running in, if applicable, or using the {@link + * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}. While + * it is not necessarily enforced, it is a usage error to fork a + * task more than once unless it has completed and been + * reinitialized. Subsequent modifications to the state of this + * task or any data it operates on are not necessarily + * consistently observable by any thread other than the one + * executing it unless preceded by a call to {@link #join} or + * related methods, or a call to {@link #isDone} returning {@code + * true}. * * @return {@code this}, to simplify usage */ public final ForkJoinTask<V> fork() { - ((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this); + Thread t; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) + ((ForkJoinWorkerThread)t).workQueue.push(this); + else + ForkJoinPool.common.externalPush(this); return this; } @@ -702,12 +703,6 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * cancelled, completed normally or exceptionally, or left * unprocessed. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param t1 the first task * @param t2 the second task * @throws NullPointerException if any task is null @@ -733,12 +728,6 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * related methods to check if they have been cancelled, completed * normally or exceptionally, or left unprocessed. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param tasks the tasks * @throws NullPointerException if any task is null */ @@ -766,7 +755,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } } if (ex != null) - ForkJoinTask.rethrow(ex); + rethrow(ex); } /** @@ -782,12 +771,6 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * cancelled, completed normally or exceptionally, or left * unprocessed. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param tasks the collection of tasks * @return the tasks argument, to simplify usage * @throws NullPointerException if tasks or any element are null @@ -823,7 +806,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { } } if (ex != null) - ForkJoinTask.rethrow(ex); + rethrow(ex); return tasks; } @@ -996,8 +979,9 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { if (Thread.interrupted()) throw new InterruptedException(); // Messy in part because we measure in nanosecs, but wait in millisecs - int s; long ns, ms; - if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) { + int s; long ms; + long ns = unit.toNanos(timeout); + if ((s = status) >= 0 && ns > 0L) { long deadline = System.nanoTime() + ns; ForkJoinPool p = null; ForkJoinPool.WorkQueue w = null; @@ -1006,16 +990,18 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; p = wt.pool; w = wt.workQueue; - s = p.helpJoinOnce(w, this); // no retries on failure + p.helpJoinOnce(w, this); // no retries on failure } + else + ForkJoinPool.externalHelpJoin(this); boolean canBlock = false; boolean interrupted = false; try { while ((s = status) >= 0) { - if (w != null && w.runState < 0) + if (w != null && w.qlock < 0) cancelIgnoringExceptions(this); else if (!canBlock) { - if (p == null || p.tryCompensate(this, null)) + if (p == null || p.tryCompensate()) canBlock = true; } else { @@ -1083,17 +1069,15 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * be of use in designs in which many tasks are forked, but none * are explicitly joined, instead executing them until all are * processed. - * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. */ public static void helpQuiesce() { - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - wt.pool.helpQuiescePool(wt.workQueue); + Thread t; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { + ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; + wt.pool.helpQuiescePool(wt.workQueue); + } + else + ForkJoinPool.quiesceCommonPool(); } /** @@ -1146,23 +1130,19 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { /** * Tries to unschedule this task for execution. This method will - * typically succeed if this task is the most recently forked task - * by the current thread, and has not commenced executing in - * another thread. This method may be useful when arranging - * alternative local processing of tasks that could have been, but - * were not, stolen. - * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * typically (but is not guaranteed to) succeed if this task is + * the most recently forked task by the current thread, and has + * not commenced executing in another thread. This method may be + * useful when arranging alternative local processing of tasks + * that could have been, but were not, stolen. * * @return {@code true} if unforked */ public boolean tryUnfork() { - return ((ForkJoinWorkerThread)Thread.currentThread()) - .workQueue.tryUnpush(this); + Thread t; + return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) : + ForkJoinPool.tryExternalUnpush(this)); } /** @@ -1171,84 +1151,32 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * value may be useful for heuristic decisions about whether to * fork other tasks. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the number of tasks */ public static int getQueuedTaskCount() { - return ((ForkJoinWorkerThread) Thread.currentThread()) - .workQueue.queueSize(); + Thread t; ForkJoinPool.WorkQueue q; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) + q = ((ForkJoinWorkerThread)t).workQueue; + else + q = ForkJoinPool.commonSubmitterQueue(); + return (q == null) ? 0 : q.queueSize(); } /** * Returns an estimate of how many more locally queued tasks are * held by the current worker thread than there are other worker - * threads that might steal them. This value may be useful for + * threads that might steal them, or zero if this thread is not + * operating in a ForkJoinPool. This value may be useful for * heuristic decisions about whether to fork other tasks. In many * usages of ForkJoinTasks, at steady state, each worker should * aim to maintain a small constant surplus (for example, 3) of * tasks, and to process computations locally if this threshold is * exceeded. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the surplus number of tasks, which may be negative */ public static int getSurplusQueuedTaskCount() { - /* - * The aim of this method is to return a cheap heuristic guide - * for task partitioning when programmers, frameworks, tools, - * or languages have little or no idea about task granularity. - * In essence by offering this method, we ask users only about - * tradeoffs in overhead vs expected throughput and its - * variance, rather than how finely to partition tasks. - * - * In a steady state strict (tree-structured) computation, - * each thread makes available for stealing enough tasks for - * other threads to remain active. Inductively, if all threads - * play by the same rules, each thread should make available - * only a constant number of tasks. - * - * The minimum useful constant is just 1. But using a value of - * 1 would require immediate replenishment upon each steal to - * maintain enough tasks, which is infeasible. Further, - * partitionings/granularities of offered tasks should - * minimize steal rates, which in general means that threads - * nearer the top of computation tree should generate more - * than those nearer the bottom. In perfect steady state, each - * thread is at approximately the same level of computation - * tree. However, producing extra tasks amortizes the - * uncertainty of progress and diffusion assumptions. - * - * So, users will want to use values larger, but not much - * larger than 1 to both smooth over transient shortages and - * hedge against uneven progress; as traded off against the - * cost of extra task overhead. We leave the user to pick a - * threshold value to compare with the results of this call to - * guide decisions, but recommend values such as 3. - * - * When all threads are active, it is on average OK to - * estimate surplus strictly locally. In steady-state, if one - * thread is maintaining say 2 surplus tasks, then so are - * others. So we can just use estimated queue length. - * However, this strategy alone leads to serious mis-estimates - * in some non-steady-state conditions (ramp-up, ramp-down, - * other stalls). We can detect many of these by further - * considering the number of "idle" threads, that are known to - * have zero queued tasks, so compensate by a factor of - * (#idle/#active) threads. - */ - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - return wt.workQueue.queueSize() - wt.pool.idlePerActive(); + return ForkJoinPool.getSurplusQueuedTaskCount(); } // Extension methods @@ -1299,59 +1227,51 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { * primarily to support extensions, and is unlikely to be useful * otherwise. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask<?> peekNextLocalTask() { - return ((ForkJoinWorkerThread) Thread.currentThread()).workQueue.peek(); + Thread t; ForkJoinPool.WorkQueue q; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) + q = ((ForkJoinWorkerThread)t).workQueue; + else + q = ForkJoinPool.commonSubmitterQueue(); + return (q == null) ? null : q.peek(); } /** * Unschedules and returns, without executing, the next task - * queued by the current thread but not yet executed. This method - * is designed primarily to support extensions, and is unlikely to - * be useful otherwise. - * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * queued by the current thread but not yet executed, if the + * current thread is operating in a ForkJoinPool. This method is + * designed primarily to support extensions, and is unlikely to be + * useful otherwise. * * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask<?> pollNextLocalTask() { - return ((ForkJoinWorkerThread) Thread.currentThread()) - .workQueue.nextLocalTask(); + Thread t; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() : + null; } /** - * Unschedules and returns, without executing, the next task + * If the current thread is operating in a ForkJoinPool, + * unschedules and returns, without executing, the next task * queued by the current thread but not yet executed, if one is * available, or if not available, a task that was forked by some * other thread, if available. Availability may be transient, so a - * {@code null} result does not necessarily imply quiescence - * of the pool this task is operating in. This method is designed + * {@code null} result does not necessarily imply quiescence of + * the pool this task is operating in. This method is designed * primarily to support extensions, and is unlikely to be useful * otherwise. * - * <p>This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return a task, or {@code null} if none are available */ protected static ForkJoinTask<?> pollTask() { - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - return wt.pool.nextTaskFor(wt.workQueue); + Thread t; ForkJoinWorkerThread wt; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) : + null; } // tag operations @@ -1540,14 +1460,16 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable { // Unsafe mechanics private static final sun.misc.Unsafe U; private static final long STATUS; + static { exceptionTableLock = new ReentrantLock(); exceptionTableRefQueue = new ReferenceQueue<Object>(); exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY]; try { U = getUnsafe(); + Class<?> k = ForkJoinTask.class; STATUS = U.objectFieldOffset - (ForkJoinTask.class.getDeclaredField("status")); + (k.getDeclaredField("status")); } catch (Exception e) { throw new Error(e); } diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java index 90a0af5723..e62fc6eb71 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java @@ -25,10 +25,17 @@ public class ForkJoinWorkerThread extends Thread { * ForkJoinWorkerThreads are managed by ForkJoinPools and perform * ForkJoinTasks. For explanation, see the internal documentation * of class ForkJoinPool. + * + * This class just maintains links to its pool and WorkQueue. The + * pool field is set immediately upon construction, but the + * workQueue field is not set until a call to registerWorker + * completes. This leads to a visibility race, that is tolerated + * by requiring that the workQueue field is only accessed by the + * owning thread. */ - final ForkJoinPool.WorkQueue workQueue; // Work-stealing mechanics final ForkJoinPool pool; // the pool this thread works in + final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics /** * Creates a ForkJoinWorkerThread operating in the given pool. @@ -37,14 +44,10 @@ public class ForkJoinWorkerThread extends Thread { * @throws NullPointerException if pool is null */ protected ForkJoinWorkerThread(ForkJoinPool pool) { - super(pool.nextWorkerName()); - setDaemon(true); - Thread.UncaughtExceptionHandler ueh = pool.ueh; - if (ueh != null) - setUncaughtExceptionHandler(ueh); + // Use a placeholder until a useful name can be set in registerWorker + super("aForkJoinWorkerThread"); this.pool = pool; - pool.registerWorker(this.workQueue = new ForkJoinPool.WorkQueue - (pool, this, pool.localMode)); + this.workQueue = pool.registerWorker(this); } /** @@ -116,4 +119,3 @@ public class ForkJoinWorkerThread extends Thread { } } } - diff --git a/src/interactive/scala/tools/nsc/interactive/CompilerControl.scala b/src/interactive/scala/tools/nsc/interactive/CompilerControl.scala index c38c7c8257..d036a6e853 100644 --- a/src/interactive/scala/tools/nsc/interactive/CompilerControl.scala +++ b/src/interactive/scala/tools/nsc/interactive/CompilerControl.scala @@ -297,6 +297,11 @@ trait CompilerControl { self: Global => def implicitlyAdded = false private def accessible_s = if (accessible) "" else "[inaccessible] " + def forceInfoString = { + definitions.fullyInitializeSymbol(sym) + definitions.fullyInitializeType(tpe) + infoString + } def infoString = s"$accessible_s${sym.defStringSeenAs(tpe)}" } diff --git a/src/interactive/scala/tools/nsc/interactive/tests/core/CoreTestDefs.scala b/src/interactive/scala/tools/nsc/interactive/tests/core/CoreTestDefs.scala index 08d84af8f5..e28bf20745 100644 --- a/src/interactive/scala/tools/nsc/interactive/tests/core/CoreTestDefs.scala +++ b/src/interactive/scala/tools/nsc/interactive/tests/core/CoreTestDefs.scala @@ -27,7 +27,7 @@ private[tests] trait CoreTestDefs reporter.println("retrieved %d members".format(members.size)) compiler ask { () => val filtered = members.filterNot(member => (member.sym.name string_== "getClass") || member.sym.isConstructor) - reporter println (filtered.map(_.infoString).sorted mkString "\n") + reporter println (filtered.map(_.forceInfoString).sorted mkString "\n") } } } diff --git a/src/library/scala/PartialFunction.scala b/src/library/scala/PartialFunction.scala index 7ff5a33586..9ff648a05a 100644 --- a/src/library/scala/PartialFunction.scala +++ b/src/library/scala/PartialFunction.scala @@ -81,7 +81,7 @@ trait PartialFunction[-A, +B] extends (A => B) { self => override def andThen[C](k: B => C): PartialFunction[A, C] = new AndThen[A, B, C] (this, k) - /** Turns this partial function into an plain function returning an `Option` result. + /** Turns this partial function into a plain function returning an `Option` result. * @see Function.unlift * @return a function that takes an argument `x` to `Some(this(x))` if `this` * is defined for `x`, and to `None` otherwise. diff --git a/src/library/scala/collection/concurrent/BasicNode.java b/src/library/scala/collection/concurrent/BasicNode.java index a65d84bbf8..97b8870036 100644 --- a/src/library/scala/collection/concurrent/BasicNode.java +++ b/src/library/scala/collection/concurrent/BasicNode.java @@ -8,13 +8,8 @@ package scala.collection.concurrent; - - - - - public abstract class BasicNode { - + public abstract String string(int lev); - -}
\ No newline at end of file + +} diff --git a/src/library/scala/collection/concurrent/CNodeBase.java b/src/library/scala/collection/concurrent/CNodeBase.java index d6eb29c8df..2fce971b2b 100644 --- a/src/library/scala/collection/concurrent/CNodeBase.java +++ b/src/library/scala/collection/concurrent/CNodeBase.java @@ -8,28 +8,26 @@ package scala.collection.concurrent; - - import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +abstract class CNodeBase<K, V> extends MainNode<K, V> { + @SuppressWarnings("rawtypes") + public static final AtomicIntegerFieldUpdater<CNodeBase> updater = + AtomicIntegerFieldUpdater.newUpdater(CNodeBase.class, "csize"); -abstract class CNodeBase<K, V> extends MainNode<K, V> { - - public static final AtomicIntegerFieldUpdater<CNodeBase> updater = AtomicIntegerFieldUpdater.newUpdater(CNodeBase.class, "csize"); - public volatile int csize = -1; - + public boolean CAS_SIZE(int oldval, int nval) { return updater.compareAndSet(this, oldval, nval); } - + public void WRITE_SIZE(int nval) { updater.set(this, nval); } - + public int READ_SIZE() { return updater.get(this); } - + }
\ No newline at end of file diff --git a/src/library/scala/collection/concurrent/Gen.java b/src/library/scala/collection/concurrent/Gen.java index 331eeca16b..6019884683 100644 --- a/src/library/scala/collection/concurrent/Gen.java +++ b/src/library/scala/collection/concurrent/Gen.java @@ -8,11 +8,4 @@ package scala.collection.concurrent; - - - - - -final class Gen { -} - +final class Gen {} diff --git a/src/library/scala/collection/concurrent/INodeBase.java b/src/library/scala/collection/concurrent/INodeBase.java index cbe404edf6..2f2d203287 100644 --- a/src/library/scala/collection/concurrent/INodeBase.java +++ b/src/library/scala/collection/concurrent/INodeBase.java @@ -8,28 +8,26 @@ package scala.collection.concurrent; - - import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +abstract class INodeBase<K, V> extends BasicNode { + @SuppressWarnings("rawtypes") + public static final AtomicReferenceFieldUpdater<INodeBase, MainNode> updater = + AtomicReferenceFieldUpdater.newUpdater(INodeBase.class, MainNode.class, "mainnode"); -abstract class INodeBase<K, V> extends BasicNode { - - public static final AtomicReferenceFieldUpdater<INodeBase, MainNode> updater = AtomicReferenceFieldUpdater.newUpdater(INodeBase.class, MainNode.class, "mainnode"); - public static final Object RESTART = new Object(); - + public volatile MainNode<K, V> mainnode = null; - + public final Gen gen; - + public INodeBase(Gen generation) { gen = generation; } - + public BasicNode prev() { return null; } - + }
\ No newline at end of file diff --git a/src/library/scala/collection/concurrent/MainNode.java b/src/library/scala/collection/concurrent/MainNode.java index ffe535742e..adb9b59a3d 100644 --- a/src/library/scala/collection/concurrent/MainNode.java +++ b/src/library/scala/collection/concurrent/MainNode.java @@ -8,33 +8,32 @@ package scala.collection.concurrent; - - import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +abstract class MainNode<K, V> extends BasicNode { + @SuppressWarnings("rawtypes") + public static final AtomicReferenceFieldUpdater<MainNode, MainNode> updater = + AtomicReferenceFieldUpdater.newUpdater(MainNode.class, MainNode.class, "prev"); -abstract class MainNode<K, V> extends BasicNode { - - public static final AtomicReferenceFieldUpdater<MainNode, MainNode> updater = AtomicReferenceFieldUpdater.newUpdater(MainNode.class, MainNode.class, "prev"); - public volatile MainNode<K, V> prev = null; - + public abstract int cachedSize(Object ct); - + public boolean CAS_PREV(MainNode<K, V> oldval, MainNode<K, V> nval) { return updater.compareAndSet(this, oldval, nval); } - + public void WRITE_PREV(MainNode<K, V> nval) { updater.set(this, nval); } - + // do we need this? unclear in the javadocs... // apparently not - volatile reads are supposed to be safe // irregardless of whether there are concurrent ARFU updates + @Deprecated @SuppressWarnings("unchecked") public MainNode<K, V> READ_PREV() { return updater.get(this); } - + }
\ No newline at end of file diff --git a/src/library/scala/concurrent/ExecutionContext.scala b/src/library/scala/concurrent/ExecutionContext.scala index 8928724b34..68513f9c80 100644 --- a/src/library/scala/concurrent/ExecutionContext.scala +++ b/src/library/scala/concurrent/ExecutionContext.scala @@ -25,7 +25,7 @@ trait ExecutionContext { /** Reports that an asynchronous computation failed. */ - def reportFailure(t: Throwable): Unit + def reportFailure(@deprecatedName('t) cause: Throwable): Unit /** Prepares for the execution of a task. Returns the prepared * execution context. A valid implementation of `prepare` is one @@ -83,7 +83,7 @@ object ExecutionContext { /** The default reporter simply prints the stack trace of the `Throwable` to System.err. */ - def defaultReporter: Throwable => Unit = (t: Throwable) => t.printStackTrace() + def defaultReporter: Throwable => Unit = _.printStackTrace() } diff --git a/src/library/scala/concurrent/Future.scala b/src/library/scala/concurrent/Future.scala index c444050e3d..2e4c72cd71 100644 --- a/src/library/scala/concurrent/Future.scala +++ b/src/library/scala/concurrent/Future.scala @@ -133,8 +133,8 @@ trait Future[+T] extends Awaitable[T] { * $multipleCallbacks * $callbackInContext */ - def onFailure[U](callback: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Unit = onComplete { - case Failure(t) if NonFatal(t) && callback.isDefinedAt(t) => callback(t) + def onFailure[U](@deprecatedName('callback) pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Unit = onComplete { + case Failure(t) if NonFatal(t) && pf.isDefinedAt(t) => pf(t) case _ => }(executor) @@ -147,7 +147,7 @@ trait Future[+T] extends Awaitable[T] { * $multipleCallbacks * $callbackInContext */ - def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit + def onComplete[U](@deprecatedName('func) f: Try[T] => U)(implicit executor: ExecutionContext): Unit /* Miscellaneous */ @@ -303,21 +303,21 @@ trait Future[+T] extends Awaitable[T] { * Await.result(h, Duration.Zero) // throw a NoSuchElementException * }}} */ - def filter(pred: T => Boolean)(implicit executor: ExecutionContext): Future[T] = { - val p = Promise[T]() + def filter(@deprecatedName('pred) p: T => Boolean)(implicit executor: ExecutionContext): Future[T] = { + val promise = Promise[T]() onComplete { - case f: Failure[_] => p complete f.asInstanceOf[Failure[T]] + case f: Failure[_] => promise complete f.asInstanceOf[Failure[T]] case Success(v) => try { - if (pred(v)) p success v - else p failure new NoSuchElementException("Future.filter predicate is not satisfied") + if (p(v)) promise success v + else promise failure new NoSuchElementException("Future.filter predicate is not satisfied") } catch { - case NonFatal(t) => p failure t + case NonFatal(t) => promise failure t } }(executor) - p.future + promise.future } /** Used by for-comprehensions. @@ -565,16 +565,16 @@ object Future { * * @tparam T the type of the result * @param body the asychronous computation - * @param execctx the execution context on which the future is run + * @param executor the execution context on which the future is run * @return the `Future` holding the result of the computation */ - def apply[T](body: =>T)(implicit execctx: ExecutionContext): Future[T] = impl.Future(body) + def apply[T](body: =>T)(implicit @deprecatedName('execctx) executor: ExecutionContext): Future[T] = impl.Future(body) /** Simple version of `Futures.traverse`. Transforms a `TraversableOnce[Future[A]]` into a `Future[TraversableOnce[A]]`. * Useful for reducing many `Future`s into a single `Future`. */ def sequence[A, M[_] <: TraversableOnce[_]](in: M[Future[A]])(implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], executor: ExecutionContext): Future[M[A]] = { - in.foldLeft(Promise.successful(cbf(in)).future) { + in.foldLeft(successful(cbf(in))) { (fr, fa) => for (r <- fr; a <- fa.asInstanceOf[Future[A]]) yield (r += a) } map (_.result()) } @@ -592,15 +592,15 @@ object Future { /** Returns a `Future` that will hold the optional result of the first `Future` with a result that matches the predicate. */ - def find[T](futurestravonce: TraversableOnce[Future[T]])(predicate: T => Boolean)(implicit executor: ExecutionContext): Future[Option[T]] = { - val futures = futurestravonce.toBuffer - if (futures.isEmpty) Promise.successful[Option[T]](None).future + def find[T](@deprecatedName('futurestravonce) futures: TraversableOnce[Future[T]])(@deprecatedName('predicate) p: T => Boolean)(implicit executor: ExecutionContext): Future[Option[T]] = { + val futuresBuffer = futures.toBuffer + if (futuresBuffer.isEmpty) successful[Option[T]](None) else { val result = Promise[Option[T]]() - val ref = new AtomicInteger(futures.size) + val ref = new AtomicInteger(futuresBuffer.size) val search: Try[T] => Unit = v => try { v match { - case Success(r) => if (predicate(r)) result tryComplete Success(Some(r)) + case Success(r) if p(r) => result tryComplete Success(Some(r)) case _ => } } finally { @@ -609,7 +609,7 @@ object Future { } } - futures.foreach(_ onComplete search) + futuresBuffer.foreach(_ onComplete search) result.future } @@ -625,9 +625,9 @@ object Future { * val result = Await.result(Future.fold(futures)(0)(_ + _), 5 seconds) * }}} */ - def fold[T, R](futures: TraversableOnce[Future[T]])(zero: R)(foldFun: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = { - if (futures.isEmpty) Promise.successful(zero).future - else sequence(futures).map(_.foldLeft(zero)(foldFun)) + def fold[T, R](futures: TraversableOnce[Future[T]])(zero: R)(@deprecatedName('foldFun) op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = { + if (futures.isEmpty) successful(zero) + else sequence(futures).map(_.foldLeft(zero)(op)) } /** Initiates a fold over the supplied futures where the fold-zero is the result value of the `Future` that's completed first. @@ -638,7 +638,7 @@ object Future { * }}} */ def reduce[T, R >: T](futures: TraversableOnce[Future[T]])(op: (R, T) => R)(implicit executor: ExecutionContext): Future[R] = { - if (futures.isEmpty) Promise[R]().failure(new NoSuchElementException("reduce attempted on empty collection")).future + if (futures.isEmpty) failed(new NoSuchElementException("reduce attempted on empty collection")) else sequence(futures).map(_ reduceLeft op) } @@ -651,7 +651,7 @@ object Future { * }}} */ def traverse[A, B, M[_] <: TraversableOnce[_]](in: M[A])(fn: A => Future[B])(implicit cbf: CanBuildFrom[M[A], B, M[B]], executor: ExecutionContext): Future[M[B]] = - in.foldLeft(Promise.successful(cbf(in)).future) { (fr, a) => + in.foldLeft(successful(cbf(in))) { (fr, a) => val fb = fn(a.asInstanceOf[A]) for (r <- fr; b <- fb) yield (r += b) }.map(_.result()) diff --git a/src/library/scala/concurrent/Promise.scala b/src/library/scala/concurrent/Promise.scala index 8355a73a1f..f950b13b78 100644 --- a/src/library/scala/concurrent/Promise.scala +++ b/src/library/scala/concurrent/Promise.scala @@ -86,7 +86,7 @@ trait Promise[T] { * * $promiseCompletion */ - def success(v: T): this.type = complete(Success(v)) + def success(@deprecatedName('v) value: T): this.type = complete(Success(value)) /** Tries to complete the promise with a value. * @@ -104,7 +104,7 @@ trait Promise[T] { * * $promiseCompletion */ - def failure(t: Throwable): this.type = complete(Failure(t)) + def failure(@deprecatedName('t) cause: Throwable): this.type = complete(Failure(cause)) /** Tries to complete the promise with an exception. * @@ -112,7 +112,7 @@ trait Promise[T] { * * @return If the promise has already been completed returns `false`, or `true` otherwise. */ - def tryFailure(t: Throwable): Boolean = tryComplete(Failure(t)) + def tryFailure(@deprecatedName('t) cause: Throwable): Boolean = tryComplete(Failure(cause)) } diff --git a/src/library/scala/concurrent/impl/Future.scala b/src/library/scala/concurrent/impl/Future.scala index 055ce6e4fa..042d32c234 100644 --- a/src/library/scala/concurrent/impl/Future.scala +++ b/src/library/scala/concurrent/impl/Future.scala @@ -28,7 +28,7 @@ private[concurrent] object Future { def apply[T](body: =>T)(implicit executor: ExecutionContext): scala.concurrent.Future[T] = { val runnable = new PromiseCompletingRunnable(body) - executor.execute(runnable) + executor.prepare.execute(runnable) runnable.promise.future } } diff --git a/src/library/scala/concurrent/package.scala b/src/library/scala/concurrent/package.scala index 3e849f1722..50a66a622a 100644 --- a/src/library/scala/concurrent/package.scala +++ b/src/library/scala/concurrent/package.scala @@ -24,10 +24,10 @@ package object concurrent { * * @tparam T the type of the result * @param body the asynchronous computation - * @param execctx the execution context on which the future is run + * @param executor the execution context on which the future is run * @return the `Future` holding the result of the computation */ - def future[T](body: =>T)(implicit execctx: ExecutionContext): Future[T] = Future[T](body) + def future[T](body: =>T)(implicit @deprecatedName('execctx) executor: ExecutionContext): Future[T] = Future[T](body) /** Creates a promise object which can be completed with a value or an exception. * diff --git a/src/library/scala/sys/process/BasicIO.scala b/src/library/scala/sys/process/BasicIO.scala index 58517de402..b31bbf0540 100644 --- a/src/library/scala/sys/process/BasicIO.scala +++ b/src/library/scala/sys/process/BasicIO.scala @@ -162,21 +162,29 @@ object BasicIO { */ def processFully(processLine: String => Unit): InputStream => Unit = in => { val reader = new BufferedReader(new InputStreamReader(in)) - processLinesFully(processLine)(reader.readLine) - reader.close() + try processLinesFully(processLine)(reader.readLine) + finally reader.close() } /** Calls `processLine` with the result of `readLine` until the latter returns - * `null`. - */ + * `null` or the current thread is interrupted. + */ def processLinesFully(processLine: String => Unit)(readLine: () => String) { - def readFully() { - val line = readLine() - if (line != null) { - processLine(line) - readFully() + def working = (Thread.currentThread.isInterrupted == false) + def halting = { Thread.currentThread.interrupt(); null } + def readFully(): Unit = + if (working) { + val line = + try readLine() + catch { + case _: InterruptedException => halting + case e: IOException if !working => halting + } + if (line != null) { + processLine(line) + readFully() + } } - } readFully() } diff --git a/src/library/scala/sys/process/ProcessImpl.scala b/src/library/scala/sys/process/ProcessImpl.scala index 7a5fc4ef9b..2b7fcdeb73 100644 --- a/src/library/scala/sys/process/ProcessImpl.scala +++ b/src/library/scala/sys/process/ProcessImpl.scala @@ -223,8 +223,8 @@ private[process] trait ProcessImpl { p.exitValue() } override def destroy() = { - try{ - outputThreads foreach (_.stop()) + try { + outputThreads foreach (_.interrupt()) // on destroy, don't bother consuming any more output p.destroy() } finally inputThread.interrupt() diff --git a/src/partest/scala/tools/partest/CompilerTest.scala b/src/partest/scala/tools/partest/CompilerTest.scala index 7495f97efd..df4a81dee2 100644 --- a/src/partest/scala/tools/partest/CompilerTest.scala +++ b/src/partest/scala/tools/partest/CompilerTest.scala @@ -19,7 +19,7 @@ abstract class CompilerTest extends DirectTest { def check(source: String, unit: global.CompilationUnit): Unit lazy val global: Global = newCompiler() - lazy val units = compilationUnits(global)(sources: _ *) + lazy val units: List[global.CompilationUnit] = compilationUnits(global)(sources: _ *) import global._ import definitions.{ compilerTypeFromTag } diff --git a/src/partest/scala/tools/partest/DirectTest.scala b/src/partest/scala/tools/partest/DirectTest.scala index 46e9621b31..7f9ca3a321 100644 --- a/src/partest/scala/tools/partest/DirectTest.scala +++ b/src/partest/scala/tools/partest/DirectTest.scala @@ -6,7 +6,7 @@ package scala.tools.partest import scala.tools.nsc._ -import util.{BatchSourceFile, CommandLineParser} +import util.{ SourceFile, BatchSourceFile, CommandLineParser } import reporters.{Reporter, ConsoleReporter} /** A class for testing code which is embedded as a string. @@ -45,18 +45,32 @@ abstract class DirectTest extends App { def reporter(settings: Settings): Reporter = new ConsoleReporter(settings) - def newSources(sourceCodes: String*) = sourceCodes.toList.zipWithIndex map { - case (src, idx) => new BatchSourceFile("newSource" + (idx + 1), src) - } + private def newSourcesWithExtension(ext: String)(codes: String*): List[BatchSourceFile] = + codes.toList.zipWithIndex map { + case (src, idx) => new BatchSourceFile(s"newSource${idx + 1}.$ext", src) + } + + def newJavaSources(codes: String*) = newSourcesWithExtension("java")(codes: _*) + def newSources(codes: String*) = newSourcesWithExtension("scala")(codes: _*) + def compileString(global: Global)(sourceCode: String): Boolean = { withRun(global)(_ compileSources newSources(sourceCode)) !global.reporter.hasErrors } - def compilationUnits(global: Global)(sourceCodes: String*): List[global.CompilationUnit] = { - val units = withRun(global) { run => - run compileSources newSources(sourceCodes: _*) + + def javaCompilationUnits(global: Global)(sourceCodes: String*) = { + sourceFilesToCompiledUnits(global)(newJavaSources(sourceCodes: _*)) + } + + def sourceFilesToCompiledUnits(global: Global)(files: List[SourceFile]) = { + withRun(global) { run => + run compileSources files run.units.toList } + } + + def compilationUnits(global: Global)(sourceCodes: String*): List[global.CompilationUnit] = { + val units = sourceFilesToCompiledUnits(global)(newSources(sourceCodes: _*)) if (global.reporter.hasErrors) { global.reporter.flush() sys.error("Compilation failure.") diff --git a/src/reflect/scala/reflect/api/Mirrors.scala b/src/reflect/scala/reflect/api/Mirrors.scala index 4fd9943379..d702555ba6 100644 --- a/src/reflect/scala/reflect/api/Mirrors.scala +++ b/src/reflect/scala/reflect/api/Mirrors.scala @@ -134,9 +134,7 @@ package api * scala> fmX.get * res0: Any = 2 * - * scala> fmX.set(3) - * scala.ScalaReflectionException: cannot set an immutable field x - * ... + * scala> fmX.set(3) // NOTE: can set an underlying value of an immutable field! * * scala> val fieldY = typeOf[C].declaration(newTermName("y")).asTerm.accessed.asTerm * fieldY: reflect.runtime.universe.TermSymbol = variable y diff --git a/src/reflect/scala/reflect/internal/Flags.scala b/src/reflect/scala/reflect/internal/Flags.scala index 1d5fe3685c..b8e3407824 100644 --- a/src/reflect/scala/reflect/internal/Flags.scala +++ b/src/reflect/scala/reflect/internal/Flags.scala @@ -60,9 +60,9 @@ import scala.collection.{ mutable, immutable } // 42: VBRIDGE // 43: VARARGS // 44: TRIEDCOOKING -// 45: -// 46: -// 47: +// 45: SYNCHRONIZED/M +// 46: ARTIFACT +// 47: DEFAULTMETHOD/M // 48: // 49: // 50: @@ -118,6 +118,7 @@ class ModifierFlags { final val PRESUPER = 1L << 37 // value is evaluated before super call final val DEFAULTINIT = 1L << 41 // symbol is initialized to the default value: used by -Xcheckinit final val ARTIFACT = 1L << 46 // symbol should be ignored when typechecking; will be marked ACC_SYNTHETIC in bytecode + final val DEFAULTMETHOD = 1L << 47 // symbol is a java default method /** Symbols which are marked ARTIFACT. (Expand this list?) * @@ -253,7 +254,7 @@ class Flags extends ModifierFlags { */ final val ExplicitFlags = PRIVATE | PROTECTED | ABSTRACT | FINAL | SEALED | - OVERRIDE | CASE | IMPLICIT | ABSOVERRIDE | LAZY + OVERRIDE | CASE | IMPLICIT | ABSOVERRIDE | LAZY | DEFAULTMETHOD /** The two bridge flags */ final val BridgeFlags = BRIDGE | VBRIDGE @@ -439,7 +440,7 @@ class Flags extends ModifierFlags { case TRIEDCOOKING => "<triedcooking>" // (1L << 44) case SYNCHRONIZED => "<synchronized>" // (1L << 45) case ARTIFACT => "<artifact>" // (1L << 46) - case 0x800000000000L => "" // (1L << 47) + case DEFAULTMETHOD => "<defaultmethod>" // (1L << 47) case 0x1000000000000L => "" // (1L << 48) case 0x2000000000000L => "" // (1L << 49) case 0x4000000000000L => "" // (1L << 50) diff --git a/src/reflect/scala/reflect/internal/Kinds.scala b/src/reflect/scala/reflect/internal/Kinds.scala index 4612e65e7a..46a95c7d26 100644 --- a/src/reflect/scala/reflect/internal/Kinds.scala +++ b/src/reflect/scala/reflect/internal/Kinds.scala @@ -230,4 +230,180 @@ trait Kinds { } } } + + /** + * The data structure describing the kind of a given type. + * + * Proper types are represented using ProperTypeKind. + * + * Type constructors are reprented using TypeConKind. + */ + abstract class Kind { + import Kind.StringState + def description: String + def order: Int + def bounds: TypeBounds + + /** Scala syntax notation of this kind. + * Proper types are expresses as A. + * Type constructors are expressed as F[k1 >: lo <: hi, k2, ...] where k1, k2, ... are parameter kinds. + * If the bounds exists at any level, it preserves the type variable names. Otherwise, + * it uses prescribed letters for each level: A, F, X, Y, Z. + */ + def scalaNotation: String + + /** Kind notation used in http://adriaanm.github.com/files/higher.pdf. + * Proper types are expressed as *. + * Type constructors are expressed * -> *(lo, hi) -(+)-> *. + */ + def starNotation: String + + /** Contains bounds either as part of itself or its arguments. + */ + def hasBounds: Boolean = !bounds.isEmptyBounds + + private[internal] def buildState(sym: Symbol, v: Variance)(s: StringState): StringState + } + object Kind { + private[internal] sealed trait ScalaNotation + private[internal] sealed case class Head(order: Int, n: Option[Int], alias: Option[String]) extends ScalaNotation { + override def toString: String = { + alias getOrElse { + typeAlias(order) + n.map(_.toString).getOrElse("") + } + } + private def typeAlias(x: Int): String = + x match { + case 0 => "A" + case 1 => "F" + case 2 => "X" + case 3 => "Y" + case 4 => "Z" + case n if n < 12 => ('O'.toInt - 5 + n).toChar.toString + case _ => "V" + } + } + private[internal] sealed case class Text(value: String) extends ScalaNotation { + override def toString: String = value + } + private[internal] case class StringState(tokens: Seq[ScalaNotation]) { + override def toString: String = tokens.mkString + def append(value: String): StringState = StringState(tokens :+ Text(value)) + def appendHead(order: Int, sym: Symbol): StringState = { + val n = countByOrder(order) + 1 + val alias = if (sym eq NoSymbol) None + else Some(sym.nameString) + StringState(tokens :+ Head(order, Some(n), alias)) + } + def countByOrder(o: Int): Int = tokens count { + case Head(`o`, _, _) => true + case t => false + } + // Replace Head(o, Some(1), a) with Head(o, None, a) if countByOrder(o) <= 1, so F1[A] becomes F[A] + def removeOnes: StringState = { + val maxOrder = (tokens map { + case Head(o, _, _) => o + case _ => 0 + }).max + StringState((tokens /: (0 to maxOrder)) { (ts: Seq[ScalaNotation], o: Int) => + if (countByOrder(o) <= 1) + ts map { + case Head(`o`, _, a) => Head(o, None, a) + case t => t + } + else ts + }) + } + // Replace Head(o, n, Some(_)) with Head(o, n, None), so F[F] becomes F[A]. + def removeAlias: StringState = { + StringState(tokens map { + case Head(o, n, Some(_)) => Head(o, n, None) + case t => t + }) + } + } + private[internal] object StringState { + def empty: StringState = StringState(Seq()) + } + } + class ProperTypeKind(val bounds: TypeBounds) extends Kind { + import Kind.StringState + val description: String = "This is a proper type." + val order = 0 + private[internal] def buildState(sym: Symbol, v: Variance)(s: StringState): StringState = { + s.append(v.symbolicString).appendHead(order, sym).append(bounds.scalaNotation(_.toString)) + } + def scalaNotation: String = Kind.Head(order, None, None) + bounds.scalaNotation(_.toString) + def starNotation: String = "*" + bounds.starNotation(_.toString) + } + object ProperTypeKind { + def apply: ProperTypeKind = this(TypeBounds.empty) + def apply(bounds: TypeBounds): ProperTypeKind = new ProperTypeKind(bounds) + def unapply(ptk: ProperTypeKind): Some[TypeBounds] = Some(ptk.bounds) + } + + class TypeConKind(val bounds: TypeBounds, val args: Seq[TypeConKind.Argument]) extends Kind { + import Kind.StringState + val order = (args map {_.kind.order} max) + 1 + def description: String = + if (order == 1) "This is a type constructor: a 1st-order-kinded type." + else "This is a type constructor that takes type constructor(s): a higher-kinded type." + override def hasBounds: Boolean = super.hasBounds || args.exists(_.kind.hasBounds) + def scalaNotation: String = { + val s = buildState(NoSymbol, Variance.Invariant)(StringState.empty).removeOnes + val s2 = if (hasBounds) s + else s.removeAlias + s2.toString + } + private[internal] def buildState(sym: Symbol, v: Variance)(s0: StringState): StringState = { + var s: StringState = s0 + s = s.append(v.symbolicString).appendHead(order, sym).append("[") + args.zipWithIndex foreach { case (arg, i) => + s = arg.kind.buildState(arg.sym, arg.variance)(s) + if (i != args.size - 1) { + s = s.append(",") + } + } + s = s.append("]").append(bounds.scalaNotation(_.toString)) + s + } + def starNotation: String = { + import Variance._ + (args map { arg => + (if (arg.kind.order == 0) arg.kind.starNotation + else "(" + arg.kind.starNotation + ")") + + (if (arg.variance == Invariant) " -> " + else " -(" + arg.variance.symbolicString + ")-> ") + }).mkString + "*" + bounds.starNotation(_.toString) + } + } + object TypeConKind { + def apply(args: Seq[TypeConKind.Argument]): TypeConKind = this(TypeBounds.empty, args) + def apply(bounds: TypeBounds, args: Seq[TypeConKind.Argument]): TypeConKind = new TypeConKind(bounds, args) + def unapply(tck: TypeConKind): Some[(TypeBounds, Seq[TypeConKind.Argument])] = Some(tck.bounds, tck.args) + case class Argument(variance: Variance, kind: Kind)(val sym: Symbol) {} + } + + /** + * Starting from a Symbol (sym) or a Type (tpe), infer the kind that classifies it (sym.tpeHK/tpe). + */ + object inferKind { + import TypeConKind.Argument + + abstract class InferKind { + protected def infer(tpe: Type, owner: Symbol, topLevel: Boolean): Kind + protected def infer(sym: Symbol, topLevel: Boolean): Kind = infer(sym.tpeHK, sym.owner, topLevel) + def apply(sym: Symbol): Kind = infer(sym, true) + def apply(tpe: Type, owner: Symbol): Kind = infer(tpe, owner, true) + } + + def apply(pre: Type): InferKind = new InferKind { + protected def infer(tpe: Type, owner: Symbol, topLevel: Boolean): Kind = { + val bounds = if (topLevel) TypeBounds.empty + else tpe.asSeenFrom(pre, owner).bounds + if(!tpe.isHigherKinded) ProperTypeKind(bounds) + else TypeConKind(bounds, tpe.typeParams map { p => Argument(p.variance, infer(p, false))(p) }) + } + } + } } diff --git a/src/reflect/scala/reflect/internal/Types.scala b/src/reflect/scala/reflect/internal/Types.scala index 00fd20a46b..3956bbcdec 100644 --- a/src/reflect/scala/reflect/internal/Types.scala +++ b/src/reflect/scala/reflect/internal/Types.scala @@ -1435,13 +1435,27 @@ trait Types case TypeBounds(_, _) => that <:< this case _ => lo <:< that && that <:< hi } - private def lowerString = if (emptyLowerBound) "" else " >: " + lo - private def upperString = if (emptyUpperBound) "" else " <: " + hi private def emptyLowerBound = typeIsNothing(lo) || lo.isWildcard private def emptyUpperBound = typeIsAny(hi) || hi.isWildcard def isEmptyBounds = emptyLowerBound && emptyUpperBound - override def safeToString = lowerString + upperString + override def safeToString = scalaNotation(_.toString) + + /** Bounds notation used in Scala sytanx. + * For example +This <: scala.collection.generic.Sorted[K,This]. + */ + private[internal] def scalaNotation(typeString: Type => String): String = { + (if (emptyLowerBound) "" else " >: " + typeString(lo)) + + (if (emptyUpperBound) "" else " <: " + typeString(hi)) + } + /** Bounds notation used in http://adriaanm.github.com/files/higher.pdf. + * For example *(scala.collection.generic.Sorted[K,This]). + */ + private[internal] def starNotation(typeString: Type => String): String = { + if (emptyLowerBound && emptyUpperBound) "" + else if (emptyLowerBound) "(" + typeString(hi) + ")" + else "(%s, %s)" format (typeString(lo), typeString(hi)) + } override def kind = "TypeBoundsType" } diff --git a/src/reflect/scala/reflect/internal/util/Statistics.scala b/src/reflect/scala/reflect/internal/util/Statistics.scala index 5b1e8ffdf8..9c19159657 100644 --- a/src/reflect/scala/reflect/internal/util/Statistics.scala +++ b/src/reflect/scala/reflect/internal/util/Statistics.scala @@ -104,7 +104,7 @@ quant) r <- q :: q.children.toList if r.prefix.nonEmpty) yield r private def showPercent(x: Double, base: Double) = - if (base == 0) "" else f" (${x / base * 100}%2.1f%)" + if (base == 0) "" else f" (${x / base * 100}%2.1f%%)" /** The base trait for quantities. * Quantities with non-empty prefix are printed in the statistics info. diff --git a/src/reflect/scala/reflect/runtime/JavaMirrors.scala b/src/reflect/scala/reflect/runtime/JavaMirrors.scala index c46ec691ae..ca9cc85ab2 100644 --- a/src/reflect/scala/reflect/runtime/JavaMirrors.scala +++ b/src/reflect/scala/reflect/runtime/JavaMirrors.scala @@ -125,7 +125,6 @@ private[reflect] trait JavaMirrors extends internal.SymbolTable with api.JavaUni private def ErrorStaticModule(sym: Symbol) = abort(s"$sym is a static module, use reflectModule on a RuntimeMirror to obtain its ModuleMirror") private def ErrorNotMember(sym: Symbol, owner: Symbol) = abort(s"expected a member of $owner, you provided ${sym.kindString} ${sym.fullName}") private def ErrorNotField(sym: Symbol) = abort(s"expected a field or an accessor method symbol, you provided $sym") - private def ErrorSetImmutableField(sym: Symbol) = abort(s"cannot set an immutable field ${sym.name}") private def ErrorNotConstructor(sym: Symbol, owner: Symbol) = abort(s"expected a constructor of $owner, you provided $sym") private def ErrorFree(member: Symbol, freeType: Symbol) = abort(s"cannot reflect ${member.kindString} ${member.name}, because it's a member of a weak type ${freeType.name}") private def ErrorNonExistentField(sym: Symbol) = abort( @@ -269,7 +268,8 @@ private[reflect] trait JavaMirrors extends internal.SymbolTable with api.JavaUni lazy val jfield = ensureAccessible(fieldToJava(symbol)) def get = jfield get receiver def set(value: Any) = { - if (!symbol.isMutable) ErrorSetImmutableField(symbol) + // it appears useful to be able to set values of vals, therefore I'm disabling this check + // if (!symbol.isMutable) ErrorSetImmutableField(symbol) jfield.set(receiver, value) } def bind(newReceiver: Any) = new JavaFieldMirror(newReceiver, symbol) diff --git a/src/reflect/scala/reflect/runtime/ReflectionUtils.scala b/src/reflect/scala/reflect/runtime/ReflectionUtils.scala index 536bdc815d..53495e6ac8 100644 --- a/src/reflect/scala/reflect/runtime/ReflectionUtils.scala +++ b/src/reflect/scala/reflect/runtime/ReflectionUtils.scala @@ -29,15 +29,6 @@ private[scala] object ReflectionUtils { case ex if pf isDefinedAt unwrapThrowable(ex) => pf(unwrapThrowable(ex)) } - private def systemProperties: Iterator[(String, String)] = { - import scala.collection.JavaConverters._ - System.getProperties.asScala.iterator - } - - private def inferBootClasspath: String = ( - systemProperties find (_._1 endsWith ".boot.class.path") map (_._2) getOrElse "" - ) - def show(cl: ClassLoader): String = { import scala.language.reflectiveCalls @@ -52,7 +43,8 @@ private[scala] object ReflectionUtils { case cl if cl != null && isAbstractFileClassLoader(cl.getClass) => cl.asInstanceOf[{val root: scala.reflect.io.AbstractFile}].root.canonicalPath case null => - inferBootClasspath + val loadBootCp = (flavor: String) => scala.util.Properties.propOrNone(flavor + ".boot.class.path") + loadBootCp("sun") orElse loadBootCp("java") getOrElse "<unknown>" case _ => "<unknown>" } diff --git a/src/repl/scala/tools/nsc/interpreter/ExprTyper.scala b/src/repl/scala/tools/nsc/interpreter/ExprTyper.scala index 55182a4f95..9353215e1e 100644 --- a/src/repl/scala/tools/nsc/interpreter/ExprTyper.scala +++ b/src/repl/scala/tools/nsc/interpreter/ExprTyper.scala @@ -95,4 +95,19 @@ trait ExprTyper { } finally typeOfExpressionDepth -= 1 } + + // This only works for proper types. + def typeOfTypeString(typeString: String): Type = { + def asProperType(): Option[Type] = { + val name = freshInternalVarName() + val line = "def %s: %s = ???" format (name, typeString) + interpretSynthetic(line) match { + case IR.Success => + val sym0 = symbolOfTerm(name) + Some(sym0.asMethod.returnType) + case _ => None + } + } + beSilentDuring(asProperType()) getOrElse NoType + } } diff --git a/src/repl/scala/tools/nsc/interpreter/ILoop.scala b/src/repl/scala/tools/nsc/interpreter/ILoop.scala index 197e76d990..f4e06f6bfc 100644 --- a/src/repl/scala/tools/nsc/interpreter/ILoop.scala +++ b/src/repl/scala/tools/nsc/interpreter/ILoop.scala @@ -222,6 +222,7 @@ class ILoop(in0: Option[BufferedReader], protected val out: JPrintWriter) shCommand, nullary("silent", "disable/enable automatic printing of results", verbosity), cmd("type", "[-v] <expr>", "display the type of an expression without evaluating it", typeCommand), + cmd("kind", "[-v] <expr>", "display the kind of expression's type", kindCommand), nullary("warnings", "show the suppressed warnings from the most recent line which had any", warningsCommand) ) @@ -287,9 +288,15 @@ class ILoop(in0: Option[BufferedReader], protected val out: JPrintWriter) // Still todo: modules. private def typeCommand(line0: String): Result = { line0.trim match { - case "" => ":type [-v] <expression>" - case s if s startsWith "-v " => intp.typeCommandInternal(s stripPrefix "-v " trim, verbose = true) - case s => intp.typeCommandInternal(s, verbose = false) + case "" => ":type [-v] <expression>" + case s => intp.typeCommandInternal(s stripPrefix "-v " trim, verbose = s startsWith "-v ") + } + } + + private def kindCommand(expr: String): Result = { + expr.trim match { + case "" => ":kind [-v] <expression>" + case s => intp.kindCommandInternal(s stripPrefix "-v " trim, verbose = s startsWith "-v ") } } diff --git a/src/repl/scala/tools/nsc/interpreter/IMain.scala b/src/repl/scala/tools/nsc/interpreter/IMain.scala index 66129e0be8..3a71930383 100644 --- a/src/repl/scala/tools/nsc/interpreter/IMain.scala +++ b/src/repl/scala/tools/nsc/interpreter/IMain.scala @@ -597,11 +597,19 @@ class IMain(@BeanProperty val factory: ScriptEngineFactory, initialSettings: Set private class WrappedRequest(val req: Request) extends CompiledScript { var recorded = false + /** In Java we would have to wrap any checked exception in the declared + * ScriptException. Runtime exceptions and errors would be ok and would + * not need to be caught. So let us do the same in Scala : catch and + * wrap any checked exception, and let runtime exceptions and errors + * escape. We could have wrapped runtime exceptions just like other + * exceptions in ScriptException, this is a choice. + */ @throws(classOf[ScriptException]) def eval(context: ScriptContext): Object = { val result = req.lineRep.evalEither match { + case Left(e: RuntimeException) => throw e case Left(e: Exception) => throw new ScriptException(e) - case Left(_) => throw new ScriptException("run-time error") + case Left(e) => throw e case Right(result) => result.asInstanceOf[Object] } if (!recorded) { @@ -899,7 +907,7 @@ class IMain(@BeanProperty val factory: ScriptEngineFactory, initialSettings: Set val preamble = """ |object %s { | %s - | val %s: String = %s { + | lazy val %s: String = %s { | %s | ("" """.stripMargin.format( diff --git a/src/repl/scala/tools/nsc/interpreter/package.scala b/src/repl/scala/tools/nsc/interpreter/package.scala index 52a085080b..f82c38f5e7 100644 --- a/src/repl/scala/tools/nsc/interpreter/package.scala +++ b/src/repl/scala/tools/nsc/interpreter/package.scala @@ -10,6 +10,7 @@ import scala.reflect.{ classTag, ClassTag } import scala.reflect.runtime.{ universe => ru } import scala.reflect.{ClassTag, classTag} import scala.reflect.api.{Mirror, TypeCreator, Universe => ApiUniverse} +import scala.util.control.Exception.catching /** The main REPL related classes and values are as follows. * In addition to standard compiler classes Global and Settings, there are: @@ -127,6 +128,47 @@ package object interpreter extends ReplConfig with ReplStrings { "" } + def kindCommandInternal(expr: String, verbose: Boolean): Unit = { + val catcher = catching(classOf[MissingRequirementError], + classOf[ScalaReflectionException]) + def typeFromTypeString: Option[ClassSymbol] = catcher opt { + exprTyper.typeOfTypeString(expr).typeSymbol.asClass + } + def typeFromNameTreatedAsTerm: Option[ClassSymbol] = catcher opt { + val moduleClass = exprTyper.typeOfExpression(expr).typeSymbol + moduleClass.linkedClassOfClass.asClass + } + def typeFromFullName: Option[ClassSymbol] = catcher opt { + intp.global.rootMirror.staticClass(expr) + } + def typeOfTerm: Option[TypeSymbol] = replInfo(symbolOfLine(expr)).typeSymbol match { + case sym: TypeSymbol => Some(sym) + case _ => None + } + (typeFromTypeString orElse typeFromNameTreatedAsTerm orElse typeFromFullName orElse typeOfTerm) foreach { sym => + val (kind, tpe) = exitingTyper { + val tpe = sym.tpeHK + (intp.global.inferKind(NoPrefix)(tpe, sym.owner), tpe) + } + echoKind(tpe, kind, verbose) + } + } + + def echoKind(tpe: Type, kind: Kind, verbose: Boolean) { + def typeString(tpe: Type): String = { + tpe match { + case TypeRef(_, sym, _) => typeString(sym.typeSignature) + case RefinedType(_, _) => tpe.toString + case _ => tpe.typeSymbol.fullName + } + } + printAfterTyper(typeString(tpe) + "'s kind is " + kind.scalaNotation) + if (verbose) { + echo(kind.starNotation) + echo(kind.description) + } + } + /** TODO - * -n normalize * -l label with case class parameter names diff --git a/src/scaladoc/scala/tools/nsc/doc/Settings.scala b/src/scaladoc/scala/tools/nsc/doc/Settings.scala index e16b6be853..e5dbaa3fd5 100644 --- a/src/scaladoc/scala/tools/nsc/doc/Settings.scala +++ b/src/scaladoc/scala/tools/nsc/doc/Settings.scala @@ -238,10 +238,7 @@ class Settings(error: String => Unit, val printMsg: String => Unit = println(_)) } } - def appendIndex(url: String): String = { - val index = "/index.html" - if (url.endsWith(index)) url else url + index - } + def appendIndex(url: String): String = url.stripSuffix("index.html").stripSuffix("/") + "/index.html" lazy val extUrlMapping: Map[String, String] = docExternalDoc.value flatMap { s => val idx = s.indexOf("#") diff --git a/src/scaladoc/scala/tools/nsc/doc/html/resource/lib/index.js b/src/scaladoc/scala/tools/nsc/doc/html/resource/lib/index.js index 70073b272a..96689ae701 100644 --- a/src/scaladoc/scala/tools/nsc/doc/html/resource/lib/index.js +++ b/src/scaladoc/scala/tools/nsc/doc/html/resource/lib/index.js @@ -14,9 +14,9 @@ var title = $(document).attr('title'); var lastHash = ""; $(document).ready(function() { - $('body').layout({ + $('body').layout({ west__size: '20%', - center__maskContents: true + center__maskContents: true }); $('#browser').layout({ center__paneSelector: ".ui-west-center" @@ -342,11 +342,6 @@ function configureTextFilter() { if (event.keyCode == 27) { // escape input.attr("value", ""); } - if (event.keyCode == 9) { // tab - $("#template").contents().find("#mbrsel-input").focus(); - input.attr("value", ""); - return false; - } if (event.keyCode == 40) { // down arrow $(window).unbind("keydown"); keyboardScrolldownLeftPane(); @@ -354,6 +349,14 @@ function configureTextFilter() { } textFilter(); }); + input.bind('keydown', function(event) { + if (event.keyCode == 9) { // tab + $("#template").contents().find("#mbrsel-input").focus(); + input.attr("value", ""); + return false; + } + textFilter(); + }); input.focus(function(event) { input.select(); }); }); scheduler.add("init", function() { diff --git a/src/scaladoc/scala/tools/nsc/doc/model/diagram/DiagramFactory.scala b/src/scaladoc/scala/tools/nsc/doc/model/diagram/DiagramFactory.scala index ebac25bbe4..87d7ece8f2 100644 --- a/src/scaladoc/scala/tools/nsc/doc/model/diagram/DiagramFactory.scala +++ b/src/scaladoc/scala/tools/nsc/doc/model/diagram/DiagramFactory.scala @@ -116,18 +116,27 @@ trait DiagramFactory extends DiagramDirectiveParser { case d: TemplateEntity if ((!diagramFilter.hideInheritedNodes) || (d.inTemplate == pack)) => d } + def listSuperClasses(member: MemberTemplateImpl) = { + // TODO: Everyone should be able to use the @{inherit,content}Diagram annotation to add nodes to diagrams. + (pack.sym, member.sym) match { + case (ScalaPackage, NullClass) => + List(makeTemplate(AnyRefClass)) + case (ScalaPackage, NothingClass) => + (List(NullClass) ::: ScalaValueClasses) map { makeTemplate(_) } + case _ => + member.parentTypes map { + case (template, tpe) => template + } filter { + nodesAll.contains(_) + } + } + } + // for each node, add its subclasses for (node <- nodesAll if !classExcluded(node)) { node match { case dnode: MemberTemplateImpl => - var superClasses = dnode.parentTypes.map(_._1).filter(nodesAll.contains(_)) - - // TODO: Everyone should be able to use the @{inherit,content}Diagram annotation to add nodes to diagrams. - if (pack.sym == ScalaPackage) - if (dnode.sym == NullClass) - superClasses = List(makeTemplate(AnyRefClass)) - else if (dnode.sym == NothingClass) - superClasses = (List(NullClass) ::: ScalaValueClasses).map(makeTemplate(_)) + val superClasses = listSuperClasses(dnode) if (!superClasses.isEmpty) { nodesShown += dnode @@ -149,7 +158,14 @@ trait DiagramFactory extends DiagramDirectiveParser { None else { val nodes = nodesAll.filter(nodesShown.contains(_)).flatMap(mapNodes.get(_)) - val edges = edgesAll.map(pair => (mapNodes(pair._1), pair._2.map(mapNodes(_)))).filterNot(pair => pair._2.isEmpty) + val edges = edgesAll.map { + case (entity, superClasses) => { + (mapNodes(entity), superClasses flatMap { mapNodes.get(_) }) + } + } filterNot { + case (node, superClassNodes) => superClassNodes.isEmpty + } + val diagram = // TODO: Everyone should be able to use the @{inherit,content}Diagram annotation to change the diagrams. if (pack.sym == ScalaPackage) { |