nim-libp2p/libp2p/muxers/mplex/lpchannel.nim

## Nim-LibP2P
## Copyright (c) 2019 Status Research & Development GmbH
## Licensed under either of
##  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
##  * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.

import oids, deques
import chronos, chronicles
import types,
       coder,
       nimcrypto/utils,
       ../../stream/bufferstream,
       ../../stream/lpstream,
       ../../connection,
       ../../utility,
       ../../errors

export lpstream

logScope:
  topics = "mplexchannel"

## Channel half-closed states
##
## | State    | Closed local      | Closed remote
## |=============================================
## | Read     | Yes (until EOF)   | No
## | Write    | No                | Yes
##

# TODO: this is one place where we need to use
# a proper state machine, but I've opted out of
# it for now for two reasons:
#
# 1) we don't have that many states to manage
# 2) I'm not sure if adding the state machine
# would have simplified or complicated the code
#
# But now that this is in place, we should perhaps
# reconsider reworking it again, this time with a
# more formal approach.
#

type
  LPChannel* = ref object of BufferStream
    id*: uint64                   # channel id
    name*: string                 # name of the channel (for debugging)
    conn*: Connection             # wrapped connection used to for writing
    initiator*: bool              # initiated remotely or locally flag
    isLazy*: bool                 # is channel lazy
    isOpen*: bool                 # has channel been opened (only used with isLazy)
    closedLocal*: bool            # has channel been closed locally
    msgCode*: MessageType         # cached in/out message code
    closeCode*: MessageType       # cached in/out close code
    resetCode*: MessageType       # cached in/out reset code

proc open*(s: LPChannel) {.async, gcsafe.}

template withWriteLock(lock: AsyncLock, body: untyped): untyped =
  try:
    await lock.acquire()
    body
  finally:
    if not(isNil(lock)) and lock.locked:
      lock.release()

template withEOFExceptions(body: untyped): untyped =
  try:
      body
  except LPStreamEOFError as exc:
    trace "muxed connection EOF", exc = exc.msg
  except LPStreamClosedError as exc:
    trace "muxed connection closed", exc = exc.msg
  except LPStreamIncompleteError as exc:
    trace "incomplete message", exc = exc.msg

method reset*(s: LPChannel) {.base, async, gcsafe.}

proc newChannel*(id: uint64,
                 conn: Connection,
                 initiator: bool,
                 name: string = "",
                 size: int = DefaultBufferSize,
                 lazy: bool = false): LPChannel =
  new result
  result.id = id
  result.name = name
  result.conn = conn
  result.initiator = initiator
  result.msgCode = if initiator: MessageType.MsgOut else: MessageType.MsgIn
  result.closeCode = if initiator: MessageType.CloseOut else: MessageType.CloseIn
  result.resetCode = if initiator: MessageType.ResetOut else: MessageType.ResetIn
  result.isLazy = lazy

  let chan = result
  proc writeHandler(data: seq[byte]): Future[void] {.async, gcsafe.} =
    try:
      if chan.isLazy and not(chan.isOpen):
        await chan.open()

      # writes should happen in sequence
      trace "sending data", data = data.shortLog,
                            id = chan.id,
                            initiator = chan.initiator,
                            name = chan.name,
                            oid = chan.oid

      try:
        await conn.writeMsg(chan.id,
                            chan.msgCode,
                            data).wait(2.minutes) # write header
      except AsyncTimeoutError:
        trace "timeout writing channel, resetting"
        asyncCheck chan.reset()
    except CatchableError as exc:
      trace "unable to write in bufferstream handler", exc = exc.msg

  result.initBufferStream(writeHandler, size)
  when chronicles.enabledLogLevel == LogLevel.TRACE:
    result.name = if result.name.len > 0: result.name else: $result.oid

  trace "created new lpchannel", id = result.id,
                                 oid = result.oid,
                                 initiator = result.initiator,
                                 name = result.name

proc closeMessage(s: LPChannel) {.async.} =
  ## send close message - this will not raise
  ## on EOF or Closed
  withEOFExceptions:
    withWriteLock(s.writeLock):
      trace "sending close message", id = s.id,
                                     initiator = s.initiator,
                                     name = s.name,
                                     oid = s.oid

      await s.conn.writeMsg(s.id, s.closeCode) # write close

proc resetMessage(s: LPChannel) {.async.} =
  ## send reset message - this will not raise
  withEOFExceptions:
    withWriteLock(s.writeLock):
      trace "sending reset message", id = s.id,
                                     initiator = s.initiator,
                                     name = s.name,
                                     oid = s.oid

      await s.conn.writeMsg(s.id, s.resetCode) # write reset

proc open*(s: LPChannel) {.async, gcsafe.} =
  ## NOTE: Don't call withExcAndLock or withWriteLock,
  ## because this already gets called from writeHandler
  ## which is locked
  withEOFExceptions:
    await s.conn.writeMsg(s.id, MessageType.New, s.name)
    trace "opened channel", oid = s.oid,
                             name = s.name,
                             initiator = s.initiator
    s.isOpen = true

proc closeRemote*(s: LPChannel) {.async.} =
  trace "got EOF, closing channel", id = s.id,
                                    initiator = s.initiator,
                                    name = s.name,
                                    oid = s.oid

  # wait for all data in the buffer to be consumed
  while s.len > 0:
    await s.dataReadEvent.wait()
    s.dataReadEvent.clear()

  s.isEof = true # set EOF immediately to prevent further reads
  await s.close() # close local end
  # call to avoid leaks
  await procCall BufferStream(s).close() # close parent bufferstream

  trace "channel closed on EOF", id = s.id,
                                 initiator = s.initiator,
                                 oid = s.oid,
                                 name = s.name

method closed*(s: LPChannel): bool =
  ## this emulates half-closed behavior
  ## when closed locally writing is
  ## disabled - see the table in the
  ## header of the file
  s.closedLocal

method reset*(s: LPChannel) {.base, async, gcsafe.} =
  # we asyncCheck here because the other end
  # might be dead already - reset is always
  # optimistic
  asyncCheck s.resetMessage()
  await procCall BufferStream(s).close()
  s.isEof = true
  s.closedLocal = true

method close*(s: LPChannel) {.async, gcsafe.} =
  if s.closedLocal:
    trace "channel already closed", id = s.id,
                                    initiator = s.initiator,
                                    name = s.name,
                                    oid = s.oid
    return

  proc closeRemote() {.async.} =
    try:
      trace "closing local lpchannel", id = s.id,
                                       initiator = s.initiator,
                                       name = s.name,
                                       oid = s.oid
      await s.closeMessage().wait(2.minutes)
      s.closedLocal = true
      if s.atEof: # already closed by remote close parent buffer immediately
        await procCall BufferStream(s).close()
    except AsyncTimeoutError:
      trace "close timed out, reset channel"
      asyncCheck s.reset() # reset on timeout
    except CatchableError as exc:
      trace "exception closing channel"

    trace "lpchannel closed local", id = s.id,
                                    initiator = s.initiator,
                                    name = s.name,
                                    oid = s.oid

  asyncCheck closeRemote()