This is an automated email from the ASF dual-hosted git repository.
chl501 pushed a commit to branch componentization
in repository https://gitbox.apache.org/repos/asf/hama.git
The following commit(s) were added to refs/heads/componentization by this push:
new ee26bce 1. Upgrade aeron to version 1.17.0. 2. Switch to plain ZIO
functions first, fixing Aeron.addSubscription() not being called.
ee26bce is described below
commit ee26bce5729501ab764192aa514ee947b9c058fd
Author: Chiahung Lin <[email protected]>
AuthorDate: Wed May 1 19:02:14 2019 +0200
1. Upgrade aeron to version 1.17.0.
2. Switch to plain ZIO functions first, fixing Aeron.addSubscription() not
being called.
---
membership/build.sbt | 2 +-
.../org/apache/hama/membership/Communicator.scala | 172 ++++++++++++---------
.../apache/hama/membership/CommunicatorSpec.scala | 45 ++++--
3 files changed, 128 insertions(+), 91 deletions(-)
diff --git a/membership/build.sbt b/membership/build.sbt
index dd0d2ff..47e5cd1 100644
--- a/membership/build.sbt
+++ b/membership/build.sbt
@@ -19,7 +19,7 @@ name := "membership"
libraryDependencies ++= Seq (
"com.google.guava" % "guava" % "27.0-jre",
- "io.aeron" % "aeron-all" % "1.11.3",
+ "io.aeron" % "aeron-all" % "1.17.0",
"org.apache.curator" % "curator-framework" % "4.0.1",
"org.apache.curator" % "curator-test" % "4.0.1" % Test
)
diff --git
a/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
b/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
index 5705baa..ba6a3db 100644
--- a/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
+++ b/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
@@ -41,11 +41,13 @@ import scala.concurrent.duration.DurationInt
import scala.util.Either
import scalaz.zio._
+import scalaz.zio.duration.{ Duration => ZDuration }
-trait Communicator
object Communicator {
- val localhost = "localhost"
+ case class Channel(host: String = localhost, port: Int = 12345) {
+ override def toString(): String = s"aeron:udp?endpoint=$host:$port"
+ }
final case object BackPressured extends RuntimeException (
"Failure due to back pressure!"
@@ -71,94 +73,118 @@ object Communicator {
"Failure due to unknown reason!"
)
- case class Channel(host: String = localhost, port: Int = 12345) {
- override def toString(): String = s"aeron:udp?endpoint=$host:$port"
+ val localhost = "localhost"
+
+ def driver(barrier: ShutdownSignalBarrier): Task[MediaDriver] = ZIO.effect {
+ val context = new MediaDriver.Context().
+ threadingMode(DEDICATED).
+ conductorIdleStrategy(new BusySpinIdleStrategy).
+ receiverIdleStrategy(new BusySpinIdleStrategy).
+ senderIdleStrategy(new BusySpinIdleStrategy)
+ val driver = MediaDriver.launch(context)
+ barrier.await
+ driver
}
- implicit class SubscriberOps(aeron: Task[Aeron]) {
- def subscriber(channel: Channel = Channel(), streamId: Int = 10):
- Task[Subscription] = aeron.flatMap { a =>
- Task.effect(a.addSubscription(channel.toString, streamId))
- }
+ def aeron = ZIO.effect {
+ val context = new Aeron.Context()
+ Aeron.connect(context)
}
- implicit class ReceiveOps(subscriber: Task[Subscription]) {
- def receive(f: (DirectBuffer, Int, Int, Header) => Boolean) = ZIO.effect {
- var continuous = true
- val handler = new FragmentAssembler({
- (buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
- val shouldContinuous = f(buffer, offset, length, header)
- continuous = shouldContinuous
- })
- val idleStrategy = new BusySpinIdleStrategy()
- while(continuous) subscriber.flatMap { s =>
- val read = s.poll(handler, 10)
- idleStrategy.idle(read)
- ZIO.unit
- }
- this
+ def subscriber(aeron: Task[Aeron]) = aeron.flatMap { a =>
+ ZIO.effect { a.addSubscription(Channel().toString, 10) }
+ }
+
+ def receive (
+ sub: Task[Subscription],
+ f: (DirectBuffer, Int, Int, Header) => Boolean
+ ) = sub.flatMap { s =>
+ var continuous = true
+ val handler = new FragmentAssembler({
+ (buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
+ val shouldContinuous = f(buffer, offset, length, header)
+ continuous = shouldContinuous
+ })
+ val idleStrategy = new BusySpinIdleStrategy()
+ while(continuous) {
+ val read = s.poll(handler, 10)
+ idleStrategy.idle(read)
}
+ sub
+ }
- def close() = subscriber.flatMap { sub => sub.close; ZIO.unit }
-
+ def publisher(aeron: Task[Aeron]) = aeron.flatMap { a =>
+ ZIO.effect { a.addPublication(Channel().toString, 10) }
}
- implicit class PublisherOps(aeron: Task[Aeron]) {
- def publisher(channel: Channel = Channel(), streamId: Int = 10):
- Task[Publication] = aeron.flatMap { a =>
- Task.effect(a.addPublication(channel.toString, streamId))
+ def publish(
+ publisher: Task[Publication],
+ messageBytes: Array[Byte],
+ bufferCapacity: Int = 512,
+ boundaryAlighment: Int = CACHE_LINE_LENGTH,
+ countLimit: Int = 3,
+ deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ sleep: () => Unit = { () => Thread.sleep(1) }
+ ) = {
+
+ def isConnected (publication: Publication)(
+ implicit _deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ _countLimit: Int = 3,
+ _sleep: () => Unit = { () => Thread.sleep(1) }
+ ): Boolean = {
+ @tailrec
+ def _isConnected(pub: Publication, times: Int = 0): Boolean = {
+ val connected = pub.isConnected
+ if(_countLimit > times && (false == connected)) {
+ if(System.nanoTime >= _deadline) false else {
+ _sleep()
+ _isConnected(pub, times + 1)
+ }
+ } else connected
}
- }
+ _isConnected(publication)
+ }
- implicit class PublishOps(publisher: Task[Publication]) {
- def publish (
+ def _publish (
messageBytes: Array[Byte],
bufferCapacity: Int = 512,
boundaryAlighment: Int = CACHE_LINE_LENGTH,
+ countLimit: Int = 3,
deadline: Long = (System.nanoTime + 3.seconds.toNanos),
sleep: () => Unit = { () => Thread.sleep(1) }
- ): ZIO[Any, Throwable, PublishOps] = publisher.flatMap { pub =>
- while(!pub.isConnected) {
- if(System.nanoTime >= deadline) return ZIO.fail(new RuntimeException (
- s"Publication can't connect: ${pub.toString}"
- ))
- sleep()
- }
- val buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned (
- bufferCapacity, boundaryAlighment
- ))
- buffer.putBytes(0, messageBytes)
- val result = pub.offer(buffer, 0, messageBytes.length)
- result match {
- case res if res < 0L && BACK_PRESSURED == res =>
ZIO.fail(BackPressured)
- case res if res < 0L && NOT_CONNECTED == res => ZIO.fail(NotConnected)
- case res if res < 0L && ADMIN_ACTION == res => ZIO.fail(AdminAction)
- case res if res < 0L && CLOSED == res => ZIO.fail(Closed)
- case res if res < 0L && MAX_POSITION_EXCEEDED == res => ZIO.fail(
- MaxPositionExceeded
- )
- case res if res < 0L => ZIO.fail(UnknownReason)
- case _ => ZIO.succeed(this)
+ ) = publisher.flatMap { p =>
+ implicit val _deadline = deadline
+ implicit val _countLimit = countLimit
+ implicit val _sleep = sleep
+ isConnected(p) match {
+ case true =>
+ val buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned (
+ bufferCapacity, boundaryAlighment
+ ))
+ buffer.putBytes(0, messageBytes)
+ val result = p.offer(buffer, 0, messageBytes.length)
+ result match {
+ case res if res < 0L && BACK_PRESSURED == res =>
ZIO.fail(BackPressured)
+ case res if res < 0L && NOT_CONNECTED == res =>
ZIO.fail(NotConnected)
+ case res if res < 0L && ADMIN_ACTION == res =>
ZIO.fail(AdminAction)
+ case res if res < 0L && CLOSED == res => ZIO.fail(Closed)
+ case res if res < 0L && MAX_POSITION_EXCEEDED == res => ZIO.fail(
+ MaxPositionExceeded
+ )
+ case res if res < 0L => ZIO.fail(UnknownReason)
+ case _ => ZIO.succeed(this)
+ }
+ case false => ZIO.fail(new RuntimeException("Not Connected to
Subscriber!"))
}
}
-
- def close() = publisher.flatMap { pub => pub.close; ZIO.unit }
- }
-
- def driver(barrier: ShutdownSignalBarrier): Task[MediaDriver] = ZIO.effect {
- val context = new MediaDriver.Context().
- threadingMode(DEDICATED).
- conductorIdleStrategy(new BusySpinIdleStrategy).
- receiverIdleStrategy(new BusySpinIdleStrategy).
- senderIdleStrategy(new BusySpinIdleStrategy)
- val driver = MediaDriver.launch(context)
- barrier.await
- driver
- }
-
- def aeron: Task[Aeron] = ZIO.effect {
- val context = new Aeron.Context()
- Aeron.connect(context)
+ _publish (
+ messageBytes = messageBytes,
+ bufferCapacity = bufferCapacity,
+ boundaryAlighment = boundaryAlighment,
+ countLimit = countLimit,
+ deadline = deadline,
+ sleep = sleep
+ )
}
}
diff --git
a/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
index 49740b8..f4fe826 100644
---
a/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
+++
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
@@ -21,37 +21,48 @@ import io.aeron.logbuffer.Header
import org.scalatest._
import org.agrona.DirectBuffer
import org.agrona.concurrent.ShutdownSignalBarrier
+import scala.concurrent.duration.Duration
+import scala.concurrent.duration.DurationInt
import scalaz.zio._
+import scalaz.zio.duration.{ Duration => ZDuration}
class CommunicatorSpec extends FlatSpec with Matchers {
import Communicator._
- "Communicator" should "" in {
-
+ "Communicator" should "send and receive a message." in {
+ var hello = ""
val barrier = new ShutdownSignalBarrier
- val dio = driver(barrier)
- val a = aeron
- val subio = a.subscriber().receive {
+ val dio = driver(barrier).delay(ZDuration.fromScala(0 seconds))
+ val sub = subscriber(aeron)
+ val subio = receive(sub, {
(buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
val data = new Array[Byte](length)
buffer.getBytes(offset, data)
- val hello = new String(data)
+ hello = new String(data)
+ println("Received? "+hello)
false
- }.flatMap(_.close)
- val pubio = a.publisher().publish(messageBytes =
"hello".getBytes).flatMap(_.close)
+ })
+ val pub = publisher(aeron)
+ val pubio = publish (
+ publisher = pub,
+ messageBytes = "hello".getBytes,
+ countLimit = 1024,
+ deadline = (System.nanoTime + 10.seconds.toNanos),
+ sleep = { () => Thread.sleep(1 * 1000) }
+ )
val handler = for {
- forked_d <- dio.fork
- forked_sub <- subio.fork
- forked_pub <- pubio.fork
- forked_b <- ZIO.effect { barrier.signal; ZIO.unit }.fork
- _ <- forked_b.join
- p <- forked_pub.join
- s <- forked_sub.join
- d <- forked_d.join
- } yield (p, s, d)
+ forked_driver <- dio.fork
+ forked_sub <- subio.delay(ZDuration.fromScala(3 seconds)).fork
+ forked_pub <- pubio.delay(ZDuration.fromScala(5 seconds)).fork
+ forked_barrier <- ZIO.effect { barrier.signal; ZIO.unit }.fork
+ _ <- forked_barrier.join
+ pub <- forked_pub.join
+ sub <- forked_sub.join
+ driver <- forked_driver.join
+ } yield (pub, sub, driver)
val runtime = new DefaultRuntime {}
runtime.unsafeRun { handler.run }
