Continue the problem of eqh first event SESSION_CLOSED

[cwk32]

@rosenbaumalex @katyakats HELLO!

We modificate the code according to your performance. Sadly we got kref_get warning when eqh is reusing. So we now only reuse the msgpool for sending messages. it works okay.

It happens sometimes that the ConnectionRdma will receive a SESSION_CLOSED as the first event, not SESSION_ESTABLISHED. we make sure that all eqhs we used are new, not reusing the former eqh.

14/10/17 07:33:06 WARN TaskSetManager: Lost TID 1195 (task 0.0:145)
14/10/17 07:33:06 WARN TaskSetManager: Loss was due to java.net.ConnectException
java.net.ConnectException: org.apache.spark.network.ConnectionRdma@6f03feb6 could not connect to 12.12.12.7 on port 53089, got SESSION_CLOSED
at org.apache.spark.network.ConnectionRdma.connect(ConnectionRdma.scala:34)
at org.apache.spark.network.ConnectionManagerRdma.sendMessage(ConnectionManagerRdma.scala:301)
at org.apache.spark.network.ConnectionManagerRdma.sendMessageReliably(ConnectionManagerRdma.scala:381)
at ....

A simple version of this problemd boy - - ConnectionRdma is as followed

please give some ideas

and, after this one fails, the following can still run good.

we make an estimation, that this problem only happens in small possibility.

private[spark] class ConnectionRdma(val uri: URI, val size: Int) extends Logging {
  var eqh: EventQueueHandler = null
  var msgPool: MsgPool = null

  var cs: ClientSession = null
  var thread: Thread = null

  var established = false
  var connectErrorType: EventName = null
  var close = false

  def connect() {
    eqh = new EventQueueHandler(null)

    msgPool = MyResourcesManager.getMsgPool(size)

    val startTime = System.nanoTime
    cs = new ClientSession(eqh, uri, new ClientCallbacks)
    eqh.runEventLoop(1, -1) // session established event

    if (!established) {
      throw new ConnectException(this.toString + " could not connect to " + uri.getHost + " on port "
        + uri.getPort + ", got " + connectErrorType)
    }

    Runtime.getRuntime.addShutdownHook(new Thread() {
      override def run() = {
        close = true
      }
    })

    val endTime = System.nanoTime

    logDebug(cs.toString + " session established with host " + uri.getHost + ", time taken to open: "
      + (endTime - startTime))
  }

  class ClientCallbacks extends ClientSession.Callbacks {
    def onMsgError(msg: Msg, reason: EventReason) {
      if (reason != EventReason.MSG_FLUSHED) {
        logError(this.toString + " onMsgErrorCallback, " + reason)
      }
      msgPool.releaseMsg(msg)
    }

    def onSessionEstablished() {
      established = true
    }

    def onSessionEvent(event: EventName, reason: EventReason) {
      logInfo(this.toString + " onSessionEvent " + event)
      if (event == EventName.SESSION_CLOSED || event == EventName.SESSION_ERROR
        || event == EventName.SESSION_REJECT) {
        // normal exit
        connectErrorType = event;
        close = true
        eqh.breakEventLoop
      }
    }

    def onResponse(m: Msg) {
      if (m.getIn.limit != 0) {
        println("Time: R " + m.getIn.getInt + " " + System.currentTimeMillis)
      }
      msgPool.releaseMsg(m)
    }
  }
}

[cwk32]

and an additional good message:

14/10/17 07:33:25 INFO Executor: Sending result for 1187 directly to driver
14/10/17 07:33:25 INFO Executor: Finished task ID 1187
14/10/17 07:33:25 INFO Executor: Serialized size of result for 1167 is 411311
14/10/17 07:33:25 INFO Executor: Sending result for 1167 directly to driver
14/10/17 07:33:25 INFO Executor: Finished task ID 1167
14/10/17 07:33:25 WARN EventQueueHandler: jxio.EQH[2b50366e3c70]: eventable with id 2b5034290d90 was not found in map
14/10/17 07:33:26 INFO Executor: Serialized size of result for 1170 is 405628
14/10/17 07:33:26 INFO Executor: Sending result for 1170 directly to driver
14/10/17 07:33:26 INFO Executor: Finished task ID 1170

and in this WARN. EQH[2b50366e3c70] is a totally new eqh. 2b5034290d90 appears only one time in log

[cwk32]

in the other side, there are two

14/10/17 07:33:56 WARN EventQueueHandler: jxio.EQH[2b40f67825b0]: eventable with id 2b40f6792190 was not found in map
14/10/17 07:33:56 WARN EventQueueHandler: jxio.EQH[2b40f67825b0]: eventable with id 2b40f6792190 was not found in map

[cwk32]

the eqh eventable not found(reply 1 and 2) may be related to receiving part, not must be related to ConnectionRdma

  def rdmaConnect(hostname: String, port: Int): Int = {
    var uri: URI = null
    var newport: Int = 0
    try {
      uri = new URI("rdma://" + hostname + ":" + port + "/")
    } catch {
      case e: URISyntaxException => e.printStackTrace()
    }

    routerPortal = new ServerPortal(listenEqh, uri, new PortalCallbacks(this))

    uri = routerPortal.getUri
    newport = uri.getPort

    logDebug("Waiting for RDMA incoming connections")
    thread = new Thread(new Runnable {
      def run() {
        listenEqh.runEventLoop(-1, -1)

        listenEqh.close
        releaseResources
        rspClosed = true
      }
    })
    Runtime.getRuntime.addShutdownHook(new Thread() {
      override def run() = {
        if(!rspClosed) routerPortal.close
      }
    })
    thread.start
    return newport
  }

  def releaseResources() {
    if (released == false) {
      released = true

      for ((ind, worker) <- workers) {
        worker.disconnect
      }

      MyResourcesManager.cleanCache
    }
  }

  class SessionWorker(size: Int, uri: URI) {
    val mp = new MsgPool(ConfigUtils.MSGPOOL_SERVER_CAPACITY, size, ConfigUtils.MSGPOOL_RESPONSE_SIZE)
    val eqh = new EventQueueHandler(new Callbacks {
      override def getAdditionalMsgPool(i: Int, i1: Int): MsgPool = {
        new MsgPool(ConfigUtils.MSGPOOL_SERVER_CAPACITY, size, ConfigUtils.MSGPOOL_RESPONSE_SIZE)
      }
    })
    eqh.bindMsgPool(mp)

    var server: ServerPortal = new ServerPortal(eqh, uri)

    var close = false

    thread = new Thread(new Runnable {
        def run() {
          eqh.runEventLoop(-1, -1)
          eqh.close
          mp.deleteMsgPool
        }
    })
    thread.start

    def disconnect() {
        if (close == false) {
          logDebug(this.toString + "disconnecting,got all msgs back, closing session")

          server.close
          eqh.breakEventLoop
        }
    }
  }

  class PortalCallbacks(connectionManager: ConnectionManagerRdma) extends ServerPortal.Callbacks {
    def onSessionNew(ses: ServerSession.SessionKey, srcIP: String, hint: Worker) {
      logDebug("new session coming")
      val uri = new URI(ses.getUri)
      val i = uri.getQuery().indexOf("=")
      val size = Integer.parseInt(uri.getQuery().substring(i + 1))
      val connectionManagerId = ConnectionManagerId.fromUri(uri)
      val worker = workers.getOrElseUpdate(size, {
        new SessionWorker(size, routerPortal.getUriForServer)
      })

      val container = new SessionContainer(null)

      val session = new ServerSession(ses, new sessionCallbacks(container, ConnectionManagerId.fromUri(uri)))

      container.session = session

      routerPortal.forward(worker.server, session)
    }

    def onSessionEvent(event: EventName, reason: EventReason) {
      logDebug(this.toString + " GOT EVENT " + event.toString + "because of "
        + reason.toString)
      if (event == EventName.PORTAL_CLOSED) {
          listenEqh.breakEventLoop
      }
    }
  }

  class sessionCallbacks(container: SessionContainer, connectionManagerId: ConnectionManagerId)
    extends ServerSession.Callbacks {

    def onRequest(msg: Msg) {
      val buffer = msg.getIn

      //......................omit too many code, response will be written to msg
      sendResponse(msg)
    }

    def sendResponse(msg: Msg){
        if(container.session != null){
            container.session.sendResponse(msg)
        }else{
            throw new Exception("no session for this msg")
        }
    }

    def onSessionEvent(event: EventName, reason: EventReason) {
      logDebug(this.toString + " got event " + event.toString + ", the reason is "
        + reason.toString)
    }

    def onMsgError(msg: Msg, reason: EventReason): Boolean = {
      logError("RDMA: [ERROR] onMsgErrorCallback. reason=" + reason)
      true
    }
  }

[yuyang733]

I also encountered the same issue.