Setting SO_TIMEOUT in Akka I/O

[Mario Camou]

Hi,

I’m using Akka I/O and am trying to set the SO_TIMEOUT socket value.

Looking at the docs, I see that the Tcp.Bind message receives a Traversable[SocketOption]. However, looking at the docs for akka.io.Inet.SocketOption, the only subclasses are Broadcast, KeepAlive, OOBInline, ReceiveBufferSize, ReuseAddress, SendBufferSize, TcpNoDelay and TrafficClass. The superclass for the only SoTimeout class I see (akka.actor.IO.SoTimeout) is akka.actor.IO.SocketOption which I assume is part of the old (Iteratee-based) Akka I/O.

So, how to set the SO_TIMEOUT using the new Akka I/O? Also, how to reset the SO_TIMEOUT once the socket has been opened?

In my use case I need to set a relatively short SO_TIMEOUT value when the socket is first connected, and bump it up a bit after receiving the first actual message.

Thanks,
-Mario
—
I want to change the world, but they won’t give me the source code

[Akka Team]

Hi Mario,

The place for TCP socket options is the object akka.io.Tcp.SO which contains case classes for the relevant socket options. However in your case, I am not sure why you want SO_TIMEOUT since you can set the connection timeout in the Connect message:

  case class Connect(..., timeout: Option[FiniteDuration] = None) extends Command

Plus Akka IO is nonblocking, so there won't be any receive timeouts. If you want to have some timeout in your protocol, you have to implement it yourself.

-Endre

--
>>>>>>>>>>      Read the docs: http://akka.io/docs/
>>>>>>>>>>      Check the FAQ: http://akka.io/faq/
>>>>>>>>>>      Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/groups/opt_out.

--

Akka Team

Typesafe - The software stack for applications that scale

Blog: letitcrash.com
Twitter: @akkateam

[Mario Camou]

Hi Endre,

Thanks for your response. A bit of background would probably be useful.

I am writing a daemon to which an embedded device connects. I’m using the Akka I/O server API (I should have stated that at the beginning).

When the device connects I have no idea which device it is, until it sends its first message (which contains the device ID). One problem we’ve had is that there are times when the device might power-cycle repeatedly, so it establishes the connection but never has a chance to send its first message before power-cycling. This might happen several (tens) of times in a row, so I get a whole lot of open connections until they time out. Although with Akka this shouldn’t be a problem thread-wise, we do have all of those open sockets at the OS level. What I’d like to do is, set a short SO_TIMEOUT at the beginning so that the connection will time out if it doesn’t receive the first message before, say, 15 seconds.

Once the device has identified, it might be turned off. I need to have almost-real-time information (well, with no more than 5 minutes delay) when this happens, so I can update the rest of the system on the status. One puzzling thing is, I’ve tested shutting down the device and waiting… and waiting… and waiting (even leaving it overnight) but the connection is never closed on the server side. I’ve checked at the OS level with netstat, and the connection still appears as ESTABLISHED even 12 hours after the device has been turned off. Even if the device is then turned back on, it establishes a new connection but the old one still appears as ESTABLISHED. What I want to do is, after receiving the first message, reset the SO_TIMEOUT to, say, 5 minutes. I’ve thought of using a timer to shut down the actor if it doesn’t receive a message after a while, but the device might go 15-30 minutes without sending anything and I really need to know if it’s still on before that. That’s why I thought of using TCP’s built-in flow control to correct that.

Modifying the firmware in the device is not currently an option, so I have to fix that on my side.

So, while using some sort of timer would work in the first case (the device being powered down before sending the first message), it wouldn’t work on the second case (the device is powered down after sending the first message).

Any ideas?

Thanks!

-Mario

—

I want to change the world, but they won’t give me the source code

[Roland Kuhn]

Hi Mario,

SO_TIMEOUT is only used for blocking socket operations, which Akka does not use. In an event-driven system like Akka I/O you’ll instead schedule messages to be sent to your connection handler actor at specific delays, e.g. 15 seconds after the connection is established, and then have that actor shut down the connection unless something was received in the meantime.

On the TCP level there is no feature which is generally available/enabled which tells the server that the client went away—unless packets are exchanged. If you cannot modify the protocol to enable the detection of a client silently going away, then you’ll have to look into SO_KEEPALIVE, but check how to enable it in your operating system as well and if your client supports that.

Regards,

Roland

Dr. Roland Kuhn
Akka Tech Lead
Typesafe – Reactive apps on the JVM.
twitter: @rolandkuhn

[Mario Camou]

Hi again,

Sorry for taking so long to reply. Things have been hectic.

I was thinking of SO_TIMEOUT because that’s what we used in our old implementation, I hadn’t thought through the implications of sockets being non-blocking in Akka I/O.

In the end I used the Akka Scheduler to send a short message every minute to keep traffic flowing. That, plus tuning tcp_retries2 in the Linux kernel did the trick.

Thanks!

-Mario

—

I want to change the world, but they won’t give me the source code

[Roland Kuhn]

Good to hear, thanks for sharing!

Regards,

Roland

[Mario Camou]

Hi again,

Sorry to reopen this thread after so long, but I'm in another project with a similar problem.

One thing I'm noticing is that, even after the OS closes the connection (i.e., it doesn't appear in netstat -an) my handler isn't getting any of the Tcp.ConnectionClosed messages. Is that normal? It looks bizarre to me.

What I'm doing is, starting a client that makes a connection, then doing a kill -9 to simulate the client crashing (or the client machine turning off). The socket stays there for a while (again checking via netstat -an) but eventually the OS closes it. Akka is not sending any message to indicate that the socket has been closed.

Any ideas short of doing what Roiland suggested earlier? (i.e., modifying the protocol to periodically send some data).

Thanks,

-Mario.

[Roland Kuhn]

Hi Mario,

just to get this straight: you have a connection from process A to process B, possibly on different machines. You kill -9 process A and process B is an Akka application using IO actors. If that is right then there should definitely be a ConnectionClosed event if at the time of the kill a TCP connection was indeed established.

Regards,

Roland

-- 
>>>>>>>>>> Read the docs: http://akka.io/docs/

>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/current/additional/faq.html

>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.

For more options, visit https://groups.google.com/d/optout.

[Mario Camou]

Hi Roland,

It's not happening...

If you do it through localhost it works fine, but in the case where the client is on a different machine from the Akka server, it doesn't happen. I've tried with kill -9 and also pulling the power plug on the client box.

As I said before, when doing a netstat -an on the server side the socket eventually disappears, but Akka doesn't find out about it until you actually try to write data to the socket. I've been Googling around and diving into the akka.io source code, and all I've found is people saying that the Java ServerChannel doesn't tell you when the connection is closed until you try to send something (which sounds like a really bad lack in Java if that is true).

For the moment I'm starting to do the same thing I did in the other project back when this thread started: defining a new "Heartbeat" message type and having a Scheduler periodically (I'm thinking every 30 seconds) send it, but everybody else on my team to whome I've talked about this finds it really strange.

Any ideas? Of course, I'm always open to it being a case of PEBKAC....

Cheers,

-Mario.

-Mario.

--
I want to change the world but they won't give me the source code.

[Akka Team]

Hi Mario,

On Wed, Aug 27, 2014 at 9:12 PM, Mario Camou <mca...@tecnoguru.com> wrote:

Hi Roland,

It's not happening...

If you do it through localhost it works fine, but in the case where the client is on a different machine from the Akka server, it doesn't happen. I've tried with kill -9 and also pulling the power plug on the client box.

As I said before, when doing a netstat -an on the server side the socket eventually disappears, but Akka doesn't find out about it until you actually try to write data to the socket. I've been Googling around and diving into the akka.io source code, and all I've found is people saying that the Java ServerChannel doesn't tell you when the connection is closed until you try to send something (which sounds like a really bad lack in Java if that is true).

It would be really sad indeed, if this is true. Have you tried SO_KEEPALIVE as an option?

 

For the moment I'm starting to do the same thing I did in the other project back when this thread started: defining a new "Heartbeat" message type and having a Scheduler periodically (I'm thinking every 30 seconds) send it, but everybody else on my team to whome I've talked about this finds it really strange.

Any ideas? Of course, I'm always open to it being a case of PEBKAC....

This is something we need to look into. It might be a bug in Akka, but it looks like something stranger. I will open a ticket and try to reproduce it.

-Endre

[Akka Team]

And the ticket is here: https://github.com/akka/akka/issues/15766

[Roland Kuhn]

Hi Mario,

re-reading the docs it does sound like the channel should be woken up if it has either OP_READ or OP_WRITE registered. The latter only is registered when you try to write something without success, but the former should always be registered unless you suspended reading (or did not resume it for pollMode). Are you certain that the socket is ready for reading when you make this test?

Regards,

Roland

[Akka Team]

On Wed, Aug 27, 2014 at 9:25 PM, Roland Kuhn <goo...@rkuhn.info> wrote:

Hi Mario,

re-reading the docs it does sound like the channel should be woken up if it has either OP_READ or OP_WRITE registered. The latter only is registered when you try to write something without success, but the former should always be registered unless you suspended reading (or did not resume it for pollMode). Are you certain that the socket is ready for reading when you make this test?

Fair point, Roland.

-Endre

--

[Mario Camou]

Hi Roland, Endre and the rest :),

Here is an excerpted/simplified version of what we're doing (hopefully I didn't mess things up too badly during sanitization):

class ConnectionListener(endpoint: InetSocketAddress) extends Actor with ActorLogging {

  import akka.io.{ IO, Tcp }

  implicit val system = context.system

  IO(Tcp) ! Tcp.Bind(self, endpoint)

  log.info("listening on {}", endpoint)

  override def receive = {

    case Tcp.Connected(remote, _) ⇒ handleConnected(remote, sender())

    case Terminated(handler)      ⇒ log info s"Handler terminated: $handler"

  }

  // The pipeline is built using Google Protobuf

  private def pipelineStages =

    new ProtoBufStage >>

      new LengthFieldFrame(maxSize = maxFrameSize) >>

      new TcpReadWriteAdapter

  private def handleConnected(remote: InetSocketAddress, connection: ActorRef): Unit = {

    log info s"Client connected from IP $remote"

    val init = TcpPipelineHandler withLogger (log, pipelineStages)

    val handler = context.actorOf(Props(classOf[ConnectionHandler], init, connection)),

      s"gateway-connection-handler-${UUID.randomUUID().toString}"

    )

    val pipeline = context.actorOf(TcpPipelineHandler.props(init, connection, handler).withDeploy(Deploy.local))

    connection ! Tcp.Register(pipeline)

    context watch handler

  }

}

class ConnectionHandler(

    init: Init[WithinActorContext, ServerMessage, ClientMessage],

    connection: ActorRef) extends Actor with ActorLogging {

  var pipeline: Option[ActorRef] = None

  def receive = LoggingReceive {

    case init.Event(c: ClientMessage) ⇒

      pipeline = Some(sender)

      processMessage(c)

    case s: ServerMessage         ⇒ pipeline foreach { _ ! init.Command(s) }

    case _: Tcp.ConnectionClosed  ⇒ handleClosedConnection

    case Terminated(`connection`) ⇒ handleClosedConnection

  }

  private def processMessage(c: ClientMessage): Unit = {

    log info s"Processing message $c"

  }

  private def handleClosedConnection: Unit = {

    log info s"Client has disconnected"

  }

}

What we see is, the ConnectionHandler is waiting for messages from the client (via the pipeline - this was started in 2.2 when pipelines were a thing) or to send to the client. The thing is, if the server isn't sending anything and the client dies in an unexpected fashion (kill -9 or power loss), we never get the Tcp.ConnectionClosed message.

Endre mentioned SO_KEEPALIVE. I've already tried adding options = List(KeepAlive(true))to the Tcp.Bind() message, however, according to the TCP specification:

Keep-alive packets MUST only be sent when no data or acknowledgement packets have been received for the connection within an interval. This interval MUST be configurable and MUST default to no less than two hours.

2 hours is much too long for our application, we actually need it within a few minutes at the latest. The problem is that lowering this at the OS level affects all processes running on the server, which I've found that can break other stuff.

As I said, what I find most weird about this, is that the socket IS actually being closed (checking by netstat -an) but Akka is not getting a notification.

One more thing. If you do it over localhost, it will actually work fine, since localhost DOES correctly detect the other process dying. It's only when you are running on 2 separate hosts that the problem arises.

Hoping we can sort this out,

-Mario.

--
I want to change the world but they won't give me the source code.

[Akka Team]

Hi Mario,

Thanks for the reproducer. Since you don't use PullMode (it was introduced in 2.3.x I think, but it is not used by default even there) I assume that an OP_READ should be registered on the selector, so this might be a bug.

Can you verify that this exists on 2.3.5?

-Endre

[√iktor Ҡlang]

This may be relevant:

https://community.oracle.com/message/8640564

http://stackoverflow.com/questions/14010194/detecting-socket-disconnection

I don't think OP_READ will help in the case of a severed connection.

The RFC says:

"Keep-alive packets MUST only be sent when no data or

            acknowledgement packets have been received for the
            connection within an interval.  This interval MUST be
            configurable and MUST default to no less than two hours."

http://www.ietf.org/rfc/rfc1122.txt

What happens if you enable keepalive and set the interval to something short?

Cheers,

√