This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new a054f9cef4 fix: preserve StageActor demand across fused streams (#3461)
a054f9cef4 is described below
commit a054f9cef404c71d1c1770d9ae95f5039b5ac3ce
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Thu Aug 27 14:55:06 2026 +0800
fix: preserve StageActor demand across fused streams (#3461)
Motivation:
Batched StageActor callbacks can observe stale outlet demand before
interpreter events propagate across a fused graph. Fixed nested execute limits
are topology-dependent and bypass the shell processing budget.
Modification:
Share the shell event quota with lazy StageActor dispatch. Drain or park
callback-generated interpreter work before the next message, resume after the
queue becomes idle, preserve active-stage ownership, and stop queued dispatch
after completion or failure. Skip empty interpreter executions on callbacks
that enqueue no graph work. Add directional coverage for deep fusion, a
processing limit of one, completion, and failure.
Result:
Demand replenishment works for arbitrary fused topology depths without a
magic event limit, while interpreter fairness and failure semantics remain
effective. The direct StageActor benchmark finds no measurable throughput
regression against main.
Tests:
- FusingSpec and related StageActor/stream specs: 80/80 passed
- FusingSpec with fuzzing mode: 24/24 passed
- LifecycleInterpreterSpec: 13/13 passed
- stream MiMa: passed on Scala 2.13 and Scala 3.3.8
- scalafmt and git diff --check: passed
- StageActorRefBenchmark: main 26.286M +/- 6.002M, PR 27.988M +/- 3.124M
ops/s
References:
Fixes #3459
---
.../scala/org/apache/pekko/stream/FusingSpec.scala | 166 ++++++++++++++++++++-
.../stream/impl/fusing/ActorGraphInterpreter.scala | 84 ++++++++---
.../stream/impl/fusing/GraphInterpreter.scala | 48 ++++++
.../org/apache/pekko/stream/stage/GraphStage.scala | 114 ++++++++++----
4 files changed, 362 insertions(+), 50 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
index 1610f1c3ff..e5ada5fedf 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
@@ -13,7 +13,7 @@
package org.apache.pekko.stream
-import scala.concurrent.{ duration, Await, Promise }
+import scala.concurrent.{ duration, Await, Future, Promise }
import duration._
@@ -23,7 +23,7 @@ import pekko.stream.QueueOfferResult
import pekko.stream.impl.UnfoldResourceSource
import pekko.stream.impl.fusing.GraphInterpreter
import pekko.stream.scaladsl._
-import pekko.stream.stage.GraphStage
+import pekko.stream.stage.{ GraphStage, GraphStageLogic,
GraphStageWithMaterializedValue, OutHandler }
import pekko.stream.testkit.TestPublisher
import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.Utils.TE
@@ -249,6 +249,168 @@ class FusingSpec extends StreamSpec {
.expectError(ex)
}
+ // Regression tests for https://github.com/apache/pekko/issues/3459
+ // Matches the UDP connector pattern: getStageActor receives messages from
an external
+ // actor, pushes only when isAvailable(out) is true, drops otherwise.
+ def externalActorSource(
+ dropped: java.util.concurrent.atomic.AtomicInteger,
+ ready: Promise[Done],
+ beforePush: GraphStageLogic => Unit = (_: GraphStageLogic) => ()):
Source[Int, Future[pekko.actor.ActorRef]] = {
+ val stageActorPromise = Promise[pekko.actor.ActorRef]()
+ Source.fromGraph(new GraphStageWithMaterializedValue[SourceShape[Int],
Future[pekko.actor.ActorRef]] {
+ val out = Outlet[Int]("external.out")
+ override val shape = SourceShape(out)
+ override def createLogicAndMaterializedValue(
+ inheritedAttributes: Attributes): (GraphStageLogic,
Future[pekko.actor.ActorRef]) = {
+ val logic = new GraphStageLogic(shape) {
+ override def preStart(): Unit =
+ stageActorPromise.success(getStageActor {
+ case (_, elem: Int) =>
+ beforePush(this)
+ if (isAvailable(out)) push(out, elem)
+ else dropped.incrementAndGet()
+ case _ => // ignore non-Int messages
+ }.ref)
+ setHandler(out,
+ new OutHandler {
+ // Let tests wait until initial demand reaches the source port
before sending a burst.
+ override def onPull(): Unit = ready.trySuccess(Done)
+ })
+ }
+ (logic, stageActorPromise.future)
+ }
+ })
+ }
+
+ def sendBurst(
+ stageActorFuture: Future[pekko.actor.ActorRef],
+ ready: Promise[Done],
+ downstream: pekko.stream.testkit.TestSubscriber.Probe[Int],
+ elementCount: Int): Unit = {
+ downstream.request(elementCount)
+ val stageActor = Await.result(stageActorFuture, 3.seconds)
+ Await.result(ready.future, 3.seconds)
+ (1 to elementCount).foreach(stageActor ! _)
+ within(5.seconds) {
+ downstream.expectNextN(elementCount.toLong)
+ }
+ downstream.cancel()
+ }
+
+ "replenish demand per-element for external async sources across an async
boundary" in {
+ val elementCount = 100
+ val dropped = new java.util.concurrent.atomic.AtomicInteger(0)
+ val ready = Promise[Done]()
+
+ val (stageActorFuture, downstream) = externalActorSource(dropped, ready)
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ sendBurst(stageActorFuture, ready, downstream, elementCount)
+ dropped.get() should ===(0)
+ }
+
+ "replenish demand independently of fused topology depth" in {
+ val elementCount = 8
+ val dropped = new java.util.concurrent.atomic.AtomicInteger(0)
+ val ready = Promise[Done]()
+ val deepSource = (1 to 256).foldLeft(externalActorSource(dropped,
ready)) {
+ case (source, _) => source.map(identity)
+ }
+
+ val (stageActorFuture, downstream) = deepSource
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ sendBurst(stageActorFuture, ready, downstream, elementCount)
+ dropped.get() should ===(0)
+ }
+
+ "replenish demand when the shell sync processing limit is one" in {
+ val elementCount = 8
+ val dropped = new java.util.concurrent.atomic.AtomicInteger(0)
+ val ready = Promise[Done]()
+ val deepSource = (1 to 32).foldLeft(externalActorSource(dropped, ready))
{
+ case (source, _) => source.map(identity)
+ }
+
+ val (stageActorFuture, downstream) = deepSource
+ .async
+ .addAttributes(ActorAttributes.syncProcessingLimit(1) and
asyncBoundaryInputBuffer)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ sendBurst(stageActorFuture, ready, downstream, elementCount)
+ dropped.get() should ===(0)
+ }
+
+ "leave activeStage cleared when a stage actor callback completes its
stage" in {
+ val interpreterPromise = Promise[GraphInterpreter]()
+ val stageActorPromise = Promise[pekko.actor.ActorRef]()
+ val ready = Promise[Done]()
+
+ val source =
+ Source.fromGraph(new GraphStageWithMaterializedValue[SourceShape[Int],
Future[pekko.actor.ActorRef]] {
+ val out = Outlet[Int]("completion.out")
+ override val shape = SourceShape(out)
+ override def createLogicAndMaterializedValue(
+ inheritedAttributes: Attributes): (GraphStageLogic,
Future[pekko.actor.ActorRef]) = {
+ val logic = new GraphStageLogic(shape) {
+ override def preStart(): Unit = {
+ interpreterPromise.success(interpreter)
+ stageActorPromise.success(getStageActor { case _ =>
completeStage() }.ref)
+ }
+ setHandler(out,
+ new OutHandler {
+ override def onPull(): Unit = ready.trySuccess(Done)
+ })
+ }
+ (logic, stageActorPromise.future)
+ }
+ })
+
+ val (stageActorFuture, done) = source.toMat(Sink.ignore)(Keep.both).run()
+ Await.result(ready.future, 3.seconds)
+ Await.result(stageActorFuture, 3.seconds) ! "complete"
+ Await.result(done, 3.seconds) should ===(Done)
+ Await.result(interpreterPromise.future, 3.seconds).activeStage should
be(null)
+ }
+
+ "stop a lazy stage actor dispatch after its handler fails" in {
+ val failure = TE("stage actor handler failed")
+ val invocations = new java.util.concurrent.atomic.AtomicInteger(0)
+ val entered = new java.util.concurrent.CountDownLatch(1)
+ val release = new java.util.concurrent.CountDownLatch(1)
+ val ready = Promise[Done]()
+ val dropped = new java.util.concurrent.atomic.AtomicInteger(0)
+ val source = externalActorSource(
+ dropped,
+ ready,
+ _ => {
+ invocations.incrementAndGet()
+ entered.countDown()
+ release.await(3, java.util.concurrent.TimeUnit.SECONDS)
+ throw failure
+ })
+
+ val (stageActorFuture, done) = source.toMat(Sink.ignore)(Keep.both).run()
+ Await.result(ready.future, 3.seconds)
+ val stageActor = Await.result(stageActorFuture, 3.seconds)
+ stageActor ! 1
+ try {
+ entered.await(3, java.util.concurrent.TimeUnit.SECONDS) should be(true)
+ (2 to 100).foreach(stageActor ! _)
+ } finally release.countDown()
+
+ Await.result(done.failed, 3.seconds) should ===(failure)
+ invocations.get() should ===(1)
+ dropped.get() should ===(0)
+ }
+
"use multiple actors when there are asynchronous boundaries in the
subflows (manual)" in {
val async = Flow[Int].map(x => { testActor ! actorRunningStage; x
}).async
Source(0 to 9)
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
index 8110807300..b2435e792f 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
@@ -629,12 +629,17 @@ import org.reactivestreams.Subscription
@InternalStableApi
override def execute(eventLimit: Int): Int = {
if (!shell.waitingForShutdown) {
- shell.interpreter.runAsyncInput(logic, evt, promise, handler)
- if (eventLimit == 1 && shell.interpreter.isSuspended) {
- shell.flushOutputs()
- shell.sendResume(true)
- 0
- } else shell.runBatch(eventLimit - 1)
+ handler match {
+ case eventLimitHandler: GraphInterpreter.EventLimitHandler =>
+ shell.runEventLimitHandler(logic, evt, promise, eventLimitHandler,
eventLimit)
+ case _ =>
+ shell.interpreter.runAsyncInput(logic, evt, promise, handler)
+ if (eventLimit == 1 && shell.interpreter.isSuspended) {
+ shell.flushOutputs()
+ shell.sendResume(true)
+ 0
+ } else shell.runBatch(eventLimit - 1)
+ }
} else {
eventLimit
}
@@ -719,6 +724,7 @@ import org.reactivestreams.Subscription
// TODO: Better heuristic here
private val abortLimit = shellEventLimit * 2
private var resumeScheduled = false
+ private var waitingEventLimitHandlers:
util.ArrayDeque[GraphInterpreter.EventLimitHandler] = _
def isInitialized: Boolean = self != null
def init(
@@ -798,22 +804,27 @@ import org.reactivestreams.Subscription
else enqueueToShortCircuit(resume)
}
- def runBatch(actorEventLimit: Int): Int = {
+ def runEventLimitHandler(
+ logic: GraphStageLogic,
+ evt: Any,
+ promise: Promise[Done],
+ handler: GraphInterpreter.EventLimitHandler,
+ actorEventLimit: Int): Int = {
try {
- val usingShellLimit = shellEventLimit < actorEventLimit
- val remainingQuota = interpreter.execute(Math.min(actorEventLimit,
shellEventLimit))
- flushOutputs()
- if (interpreter.isCompleted) {
- // Cannot stop right away if not completely subscribed
- if (canShutDown) interpreterCompleted = true
- else {
- waitingForShutdown = true
- val subscriptionTimeout =
attributes.mandatoryAttribute[ActorAttributes.StreamSubscriptionTimeout].timeout
- mat.scheduleOnce(subscriptionTimeout, () => self ! new
Abort(GraphInterpreterShell.this))
- }
- } else if (interpreter.isSuspended && !resumeScheduled)
sendResume(!usingShellLimit)
+ val actorLimitAfterAsyncInput = actorEventLimit - 1
+ val interpreterLimit = Math.min(actorLimitAfterAsyncInput,
shellEventLimit)
+ val handlerRemaining = interpreter.runAsyncInput(logic, evt, promise,
handler, interpreterLimit)
+ val remaining = interpreter.execute(handlerRemaining)
+
+ if (handler.isWaitingForInterpreter) {
+ if (interpreter.isSuspended) {
+ if (waitingEventLimitHandlers eq null)
+ waitingEventLimitHandlers = new
util.ArrayDeque[GraphInterpreter.EventLimitHandler]()
+ waitingEventLimitHandlers.addLast(handler)
+ } else handler.onInterpreterIdle()
+ }
- if (usingShellLimit) actorEventLimit - shellEventLimit + remainingQuota
else remainingQuota
+ finishBatch(actorLimitAfterAsyncInput - interpreterLimit + remaining)
} catch {
case NonFatal(e) =>
tryAbort(e)
@@ -821,6 +832,39 @@ import org.reactivestreams.Subscription
}
}
+ def runBatch(actorEventLimit: Int): Int = {
+ try {
+ val interpreterLimit = Math.min(actorEventLimit, shellEventLimit)
+ val remaining = interpreter.execute(interpreterLimit)
+ finishBatch(actorEventLimit - interpreterLimit + remaining)
+ } catch {
+ case NonFatal(e) =>
+ tryAbort(e)
+ actorEventLimit - 1
+ }
+ }
+
+ private def finishBatch(remainingQuota: Int): Int = {
+ flushOutputs()
+ if (interpreter.isCompleted) {
+ // Cannot stop right away if not completely subscribed
+ if (canShutDown) interpreterCompleted = true
+ else {
+ waitingForShutdown = true
+ val subscriptionTimeout =
attributes.mandatoryAttribute[ActorAttributes.StreamSubscriptionTimeout].timeout
+ mat.scheduleOnce(subscriptionTimeout, () => self ! new
Abort(GraphInterpreterShell.this))
+ }
+ } else if (interpreter.isSuspended && !resumeScheduled)
sendResume(remainingQuota == 0)
+
+ if (!interpreter.isSuspended && (waitingEventLimitHandlers ne null)) {
+ val handlers = waitingEventLimitHandlers
+ waitingEventLimitHandlers = null
+ while (!handlers.isEmpty) handlers.removeFirst().onInterpreterIdle()
+ }
+
+ remainingQuota
+ }
+
private def flushOutputs(): Unit = {
val outputBoundaries = outputs
val count = outputBoundaries.length
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
index 6d79526528..e91c25c508 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
@@ -60,6 +60,19 @@ import pekko.stream.stage._
final val KeepGoingFlag = 0x4000000
final val KeepGoingMask = 0x3FFFFFF
+ /**
+ * INTERNAL API
+ *
+ * An async input handler that shares the interpreter event budget of the
enclosing shell. The handler returns the
+ * unused part of `eventLimit`. If it has to stop while interpreter events
are still pending, the shell invokes
+ * `onInterpreterIdle` after those events have been processed.
+ */
+ @InternalApi private[stream] trait EventLimitHandler extends (Any => Unit) {
+ def apply(event: Any, eventLimit: Int): Int
+ def isWaitingForInterpreter: Boolean
+ def onInterpreterIdle(): Unit
+ }
+
/**
* Marker object that indicates that a port holds no element since it was
already grabbed. The port is still pullable,
* but there is no more element to grab.
@@ -593,6 +606,41 @@ import pekko.stream.stage._
} finally currentInterpreterHolder(0) = previousInterpreter
}
+ /** INTERNAL API */
+ @InternalApi private[stream] def runAsyncInput(
+ logic: GraphStageLogic,
+ evt: Any,
+ promise: Promise[Done],
+ handler: EventLimitHandler,
+ eventLimit: Int): Int = {
+ var eventsRemaining = eventLimit
+ if (!isStageCompleted(logic)) {
+ if (GraphInterpreter.Debug) println(s"$Name ASYNC $evt ($handler)
[$logic]")
+ val currentInterpreterHolder = _currentInterpreter.get()
+ val previousInterpreter = currentInterpreterHolder(0)
+ currentInterpreterHolder(0) = this
+ try {
+ activeStage = logic
+ try {
+ eventsRemaining = handler(evt, eventLimit)
+ if (promise ne GraphStageLogic.NoPromise) {
+ promise.success(Done)
+ logic.onFeedbackDispatched(promise)
+ }
+ } catch {
+ case NonFatal(ex) =>
+ if (promise ne GraphStageLogic.NoPromise) {
+ promise.failure(ex)
+ logic.onFeedbackDispatched(promise)
+ }
+ logic.failStage(ex)
+ }
+ afterStageHasRun(logic)
+ } finally currentInterpreterHolder(0) = previousInterpreter
+ }
+ eventsRemaining
+ }
+
// Decodes and processes a single event for the given connection
@InternalStableApi
private def processEvent(connection: Connection): Unit = {
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
index bae071f83a..8f2764cc1f 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
@@ -344,6 +344,7 @@ object GraphStageLogic {
private final val SchedStateIdle: Int = 0
private final val SchedStateScheduled: Int = 1
+ private final val SchedStateStopped: Int = 2
/**
* Lazy-path dispatch: producers enqueue into a Vyukov MPSC queue and
elect a single drain via
@@ -375,7 +376,7 @@ object GraphStageLogic {
handler: Any => Unit,
drainBatchSize: Int)
extends AbstractNodeQueue[(ActorRef, Any)]
- with (Any => Unit) {
+ with GraphInterpreter.EventLimitHandler {
// IDLE/SCHEDULED election state. VarHandle avoids per-instance
AtomicInteger.
@volatile var state: Int = SchedStateIdle
@@ -388,12 +389,13 @@ object GraphStageLogic {
private def casState(expect: Int, update: Int): Boolean =
LazyDispatch.stateHandle.compareAndSet(this, expect, update)
+ private var waitingForInterpreter = false
+
// null msg = drain signal from onAsyncInput; non-null = (ActorRef, Any)
tuple from FunctionRef.
override def apply(msg: Any): Unit =
- if (msg == null) drain()
- else {
+ if (msg != null) {
val pair = msg.asInstanceOf[(ActorRef, Any)]
- if (!interpreter.isStageCompleted(logic)) {
+ if (!interpreter.isStageCompleted(logic) && getState() !=
SchedStateStopped) {
add(pair)
if (getState() == SchedStateIdle && casState(SchedStateIdle,
SchedStateScheduled)) {
if (interpreter.isStageCompleted(logic)) setState(SchedStateIdle)
@@ -404,36 +406,92 @@ object GraphStageLogic {
private def scheduleDrain(): Unit =
// 1 AsyncInput + 1 Envelope per drain batch (amortized across up to
drainBatchSize tells).
- // `this` serves as the drain callback (Any => Unit); onAsyncInput
calls apply(null).
+ // `this` serves as both the producer callback and the
event-limit-aware drain callback.
interpreter.onAsyncInput(logic, null, NoPromise, this)
- private def drain(): Unit = {
- val limit = drainBatchSize
- var processed = 0
- while (processed < limit) {
- if (interpreter.isStageCompleted(logic)) {
- while (poll() ne null) ()
- setState(SchedStateIdle)
- return
+ override def apply(event: Any, eventLimit: Int): Int = {
+ require(event == null, "LazyDispatch drain event must be null")
+ drain(eventLimit)
+ }
+
+ override def isWaitingForInterpreter: Boolean = waitingForInterpreter
+
+ override def onInterpreterIdle(): Unit = {
+ waitingForInterpreter = false
+ if (interpreter.isStageCompleted(logic)) stopDispatch()
+ else publishIdleOrSchedule()
+ }
+
+ private def drain(eventLimit: Int): Int = {
+ var eventsRemaining = eventLimit
+ waitingForInterpreter = false
+ try {
+ // BoundaryEvents can overtake a shell resume when the
shell-specific limit is reached. Finish any
+ // previously queued interpreter work before dispatching another
StageActor message.
+ if (interpreter.isSuspended) {
+ eventsRemaining = interpreter.execute(eventsRemaining)
+ if (interpreter.isSuspended) {
+ waitingForInterpreter = true
+ return eventsRemaining
+ }
}
- val item = poll()
- if (item eq null) {
- setState(SchedStateIdle)
- // Recheck race: a producer may have added between `poll == null`
and the IDLE publish above.
- if (!isEmpty && casState(SchedStateIdle, SchedStateScheduled))
- scheduleDrain()
- return
+
+ val limit = drainBatchSize
+ var processed = 0
+ // The AsyncInput itself accounts for the first callback. Even when
it consumed the last actor quota,
+ // dispatch one message and defer any interpreter work it creates to
the next shell resume.
+ while (processed < limit && (eventsRemaining > 0 || processed == 0))
{
+ if (interpreter.isStageCompleted(logic)) {
+ stopDispatch()
+ return eventsRemaining
+ }
+ val item = poll()
+ if (item eq null) {
+ publishIdleOrSchedule()
+ return eventsRemaining
+ }
+
+ // execute() changes activeStage while propagating the element.
Set it explicitly for every
+ // StageActor callback; never restore a logic that execute() may
have finalized and released.
+ interpreter.activeStage = logic
+ handler(item)
+ // Most StageActor callbacks only update local stage state. Avoid
entering execute() unless the
+ // callback actually enqueued interpreter work; demand-producing
callbacks still drain that work
+ // before another message is dispatched.
+ if (interpreter.isSuspended) eventsRemaining =
interpreter.execute(eventsRemaining)
+ processed += 1
+
+ if (interpreter.isStageCompleted(logic)) {
+ stopDispatch()
+ return eventsRemaining
+ }
+ if (interpreter.isSuspended) {
+ waitingForInterpreter = true
+ return eventsRemaining
+ }
}
- handler(item)
- processed += 1
+
+ publishIdleOrSchedule()
+ eventsRemaining
+ } catch {
+ case ex: Throwable =>
+ stopDispatch()
+ throw ex
}
- // The last handler(item) may have completed the stage; check before
re-scheduling.
- if (interpreter.isStageCompleted(logic)) {
- while (poll() ne null) ()
+ }
+
+ private def publishIdleOrSchedule(): Unit = {
+ if (isEmpty) {
setState(SchedStateIdle)
- return
- }
- scheduleDrain()
+ // Recheck race: a producer may have added between observing empty
and publishing IDLE above.
+ if (!isEmpty && casState(SchedStateIdle, SchedStateScheduled))
scheduleDrain()
+ } else scheduleDrain()
+ }
+
+ private def stopDispatch(): Unit = {
+ waitingForInterpreter = false
+ setState(SchedStateStopped)
+ while (poll() ne null) ()
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
