GitHub PullRequestReviewCommentEvent at 2014-05-22 08:13:53
Repository: akka
Actor: drewhk
Payload: {"comment":{"url":"https://api.github.com/repos/akka/akka/pulls/comments/12939128","id":12939128,"diff_hunk":"@@ -0,0 +1,126 @@\n+/**\n+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>\n+ */\n+package akka.stream.impl\n+\n+import akka.stream.MaterializerSettings\n+import akka.OverflowStrategy\n+\n+class ConflateImpl(_settings: MaterializerSettings, seed: Any ⇒ Any, aggregate: (Any, Any) ⇒ Any) extends ActorProcessorImpl(_settings) {\n+  var conflated: Any = null\n+\n+  val waitNextZero: TransferPhase = TransferPhase(primaryInputs.NeedsInput) { () ⇒\n+    conflated = seed(primaryInputs.dequeueInputElement())\n+    nextPhase(conflateThenEmit)\n+  }\n+\n+  val conflateThenEmit: TransferPhase = TransferPhase(primaryInputs.NeedsInput || primaryOutputs.NeedsDemand) { () ⇒\n+    if (primaryInputs.inputsAvailable) conflated = aggregate(conflated, primaryInputs.dequeueInputElement())\n+    if (primaryOutputs.demandAvailable) {\n+      primaryOutputs.enqueueOutputElement(conflated)\n+      conflated = null\n+      nextPhase(waitNextZero)\n+    }\n+  }\n+\n+  nextPhase(waitNextZero)\n+}\n+\n+class ExpandImpl(_settings: MaterializerSettings, seed: Any ⇒ Any, extrapolate: Any ⇒ (Any, Any)) extends ActorProcessorImpl(_settings) {\n+  var extrapolateState: Any = null\n+\n+  val waitFirst: TransferPhase = TransferPhase(primaryInputs.NeedsInput) { () ⇒\n+    extrapolateState = seed(primaryInputs.dequeueInputElement())\n+    nextPhase(emitFirst)\n+  }\n+\n+  val emitFirst: TransferPhase = TransferPhase(primaryOutputs.NeedsDemand) { () ⇒\n+    emitExtrapolate()\n+    nextPhase(extrapolateOrReset)\n+  }\n+\n+  val extrapolateOrReset: TransferPhase = TransferPhase(primaryInputs.NeedsInputOrComplete || primaryOutputs.NeedsDemand) { () ⇒\n+    if (primaryInputs.inputsDepleted) nextPhase(completedPhase)\n+    else if (primaryInputs.inputsAvailable) {\n+      extrapolateState = seed(primaryInputs.dequeueInputElement())\n+      nextPhase(emitFirst)\n+    } else emitExtrapolate()\n+  }\n+\n+  def emitExtrapolate(): Unit = {\n+    val (emit, nextState) = extrapolate(extrapolateState)\n+    primaryOutputs.enqueueOutputElement(emit)\n+    extrapolateState = nextState\n+  }\n+\n+  nextPhase(waitFirst)\n+}\n+\n+class BufferImpl(_settings: MaterializerSettings, size: Int, overflowStrategy: OverflowStrategy) extends ActorProcessorImpl(_settings) {\n+  import OverflowStrategy._\n+  var readIdx = 0\n+  var writeIdx = 0\n+  var remainingCapacity = size\n+  val buffer = Array.fill[Any](size)(null)\n+\n+  def enqueueInternal(elem: Any): Unit = {\n+    buffer(writeIdx) = elem\n+    writeIdx = (writeIdx + 1) % size\n+    remainingCapacity -= 1\n+  }\n+\n+  def dequeueInternal(): Any = {\n+    val result = buffer(readIdx)\n+    dropHead()\n+    result\n+  }\n+\n+  def clearInternal(): Unit = {\n+    java.util.Arrays.fill(buffer.asInstanceOf[Array[Object]], null)\n+    readIdx = 0\n+    writeIdx = 0\n+    remainingCapacity = size\n+  }\n+\n+  def dropHead(): Unit = {\n+    buffer(readIdx) = null\n+    readIdx = (readIdx + 1) % size\n+    remainingCapacity += 1\n+  }\n+\n+  def dropTail(): Unit = {\n+    // Nothing to null out here, since writeIdx == readIdx when buffer is full\n+    writeIdx = (writeIdx + size - 1) % size\n+    remainingCapacity += 1\n+  }","path":"akka-stream/src/main/scala/akka/stream/impl/RateDetachedProcessors.scala","position":95,"original_position":95,"commit_id":"76c5780d055cb68d6a50758444709a2057f5f241","original_commit_id":"76c5780d055cb68d6a50758444709a2057f5f241","user":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?","gravatar_id":"e00d3476c1ca077dd6ba6897e724655d","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"body":"But then I cannot see it from the test. OTOH extracting this buffer means that I can create an alternative power-of-two sized one, and switch in that one if applicable to avoid division. Sold :)","created_at":"2014-05-22T08:13:53Z","updated_at":"2014-05-22T08:13:53Z","html_url":"https://github.com/akka/akka/pull/15244#discussion_r12939128","pull_request_url":"https://api.github.com/repos/akka/akka/pulls/15244","_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/comments/12939128"},"html":{"href":"https://github.com/akka/akka/pull/15244#discussion_r12939128"},"pull_request":{"href":"https://api.github.com/repos/akka/akka/pulls/15244"}}}}