src/python/src/grpc/_adapter/fore.py

# Copyright 2015, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#     * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

"""The RPC-service-side bridge between RPC Framework and GRPC-on-the-wire."""

import enum
import logging
import threading
import time

from grpc._adapter import _common
from grpc._adapter import _low
from grpc.framework.base import interfaces as base_interfaces
from grpc.framework.base import null
from grpc.framework.foundation import activated
from grpc.framework.foundation import logging_pool

_THREAD_POOL_SIZE = 100


@enum.unique
class _LowWrite(enum.Enum):
  """The possible categories of low-level write state."""

  OPEN = 'OPEN'
  ACTIVE = 'ACTIVE'
  CLOSED = 'CLOSED'


def _write(call, rpc_state, payload):
  serialized_payload = rpc_state.serializer(payload)
  if rpc_state.write.low is _LowWrite.OPEN:
    call.write(serialized_payload, call)
    rpc_state.write.low = _LowWrite.ACTIVE
  else:
    rpc_state.write.pending.append(serialized_payload)


def _status(call, rpc_state):
  call.status(_low.Status(_low.Code.OK, ''), call)
  rpc_state.write.low = _LowWrite.CLOSED


class ForeLink(base_interfaces.ForeLink, activated.Activated):
  """A service-side bridge between RPC Framework and the C-ish _low code."""

  def __init__(
      self, pool, request_deserializers, response_serializers,
      root_certificates, key_chain_pairs, port=None):
    """Constructor.

    Args:
      pool: A thread pool.
      request_deserializers: A dict from RPC method names to request object
        deserializer behaviors.
      response_serializers: A dict from RPC method names to response object
        serializer behaviors.
      root_certificates: The PEM-encoded client root certificates as a
        bytestring or None.
      key_chain_pairs: A sequence of PEM-encoded private key-certificate chain
        pairs.
      port: The port on which to serve, or None to have a port selected
        automatically.
    """
    self._condition = threading.Condition()
    self._pool = pool
    self._request_deserializers = request_deserializers
    self._response_serializers = response_serializers
    self._root_certificates = root_certificates
    self._key_chain_pairs = key_chain_pairs
    self._requested_port = port

    self._rear_link = null.NULL_REAR_LINK
    self._completion_queue = None
    self._server = None
    self._rpc_states = {}
    self._spinning = False
    self._port = None

  def _on_stop_event(self):
    self._spinning = False
    self._condition.notify_all()

  def _on_service_acceptance_event(self, event, server):
    """Handle a service invocation event."""
    service_acceptance = event.service_acceptance
    if service_acceptance is None:
      return

    call = service_acceptance.call
    call.accept(self._completion_queue, call)
    # TODO(nathaniel): Metadata support.
    call.premetadata()
    call.read(call)
    method = service_acceptance.method

    self._rpc_states[call] = _common.CommonRPCState(
        _common.WriteState(_LowWrite.OPEN, _common.HighWrite.OPEN, []), 1,
        self._request_deserializers[method],
        self._response_serializers[method])

    ticket = base_interfaces.FrontToBackTicket(
        call, 0, base_interfaces.FrontToBackTicket.Kind.COMMENCEMENT, method,
        base_interfaces.ServicedSubscription.Kind.FULL, None, None,
        service_acceptance.deadline - time.time())
    self._rear_link.accept_front_to_back_ticket(ticket)

    server.service(None)

  def _on_read_event(self, event):
    """Handle data arriving during an RPC."""
    call = event.tag
    rpc_state = self._rpc_states.get(call, None)
    if rpc_state is None:
      return

    sequence_number = rpc_state.sequence_number
    rpc_state.sequence_number += 1
    if event.bytes is None:
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.COMPLETION, None, None, None,
          None, None)
    else:
      call.read(call)
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.CONTINUATION, None, None,
          None, rpc_state.deserializer(event.bytes), None)

    self._rear_link.accept_front_to_back_ticket(ticket)

  def _on_write_event(self, event):
    call = event.tag
    rpc_state = self._rpc_states.get(call, None)
    if rpc_state is None:
      return

    if rpc_state.write.pending:
      serialized_payload = rpc_state.write.pending.pop(0)
      call.write(serialized_payload, call)
    elif rpc_state.write.high is _common.HighWrite.CLOSED:
      _status(call, rpc_state)
    else:
      rpc_state.write.low = _LowWrite.OPEN

  def _on_complete_event(self, event):
    if not event.complete_accepted:
      logging.error('Complete not accepted! %s', (event,))
      call = event.tag
      rpc_state = self._rpc_states.pop(call, None)
      if rpc_state is None:
        return

      sequence_number = rpc_state.sequence_number
      rpc_state.sequence_number += 1
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.TRANSMISSION_FAILURE, None,
          None, None, None, None)
      self._rear_link.accept_front_to_back_ticket(ticket)

  def _on_finish_event(self, event):
    """Handle termination of an RPC."""
    call = event.tag
    rpc_state = self._rpc_states.pop(call, None)
    if rpc_state is None:
      return

    code = event.status.code
    if code is _low.Code.OK:
      return

    sequence_number = rpc_state.sequence_number
    rpc_state.sequence_number += 1
    if code is _low.Code.CANCELLED:
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.CANCELLATION, None, None,
          None, None, None)
    elif code is _low.Code.EXPIRED:
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.EXPIRATION, None, None, None,
          None, None)
    else:
      # TODO(nathaniel): Better mapping of codes to ticket-categories
      ticket = base_interfaces.FrontToBackTicket(
          call, sequence_number,
          base_interfaces.FrontToBackTicket.Kind.TRANSMISSION_FAILURE, None,
          None, None, None, None)
    self._rear_link.accept_front_to_back_ticket(ticket)

  def _spin(self, completion_queue, server):
    while True:
      event = completion_queue.get(None)

      with self._condition:
        if event.kind is _low.Event.Kind.STOP:
          self._on_stop_event()
          return
        elif self._server is None:
          continue
        elif event.kind is _low.Event.Kind.SERVICE_ACCEPTED:
          self._on_service_acceptance_event(event, server)
        elif event.kind is _low.Event.Kind.READ_ACCEPTED:
          self._on_read_event(event)
        elif event.kind is _low.Event.Kind.WRITE_ACCEPTED:
          self._on_write_event(event)
        elif event.kind is _low.Event.Kind.COMPLETE_ACCEPTED:
          self._on_complete_event(event)
        elif event.kind is _low.Event.Kind.FINISH:
          self._on_finish_event(event)
        else:
          logging.error('Illegal event! %s', (event,))

  def _continue(self, call, payload):
    rpc_state = self._rpc_states.get(call, None)
    if rpc_state is None:
      return

    _write(call, rpc_state, payload)

  def _complete(self, call, payload):
    """Handle completion of the writes of an RPC."""
    rpc_state = self._rpc_states.get(call, None)
    if rpc_state is None:
      return

    if rpc_state.write.low is _LowWrite.OPEN:
      if payload is None:
        _status(call, rpc_state)
      else:
        _write(call, rpc_state, payload)
    elif rpc_state.write.low is _LowWrite.ACTIVE:
      if payload is not None:
        rpc_state.write.pending.append(rpc_state.serializer(payload))
    else:
      raise ValueError('Called to complete after having already completed!')
    rpc_state.write.high = _common.HighWrite.CLOSED

  def _cancel(self, call):
    call.cancel()
    self._rpc_states.pop(call, None)

  def join_rear_link(self, rear_link):
    """See base_interfaces.ForeLink.join_rear_link for specification."""
    self._rear_link = null.NULL_REAR_LINK if rear_link is None else rear_link

  def _start(self):
    """Starts this ForeLink.

    This method must be called before attempting to exchange tickets with this
    object.
    """
    with self._condition:
      address = '[::]:%d' % (
          0 if self._requested_port is None else self._requested_port)
      self._completion_queue = _low.CompletionQueue()
      if self._root_certificates is None and not self._key_chain_pairs:
        self._server = _low.Server(self._completion_queue)
        self._port = self._server.add_http2_addr(address)
      else:
        server_credentials = _low.ServerCredentials(
          self._root_certificates, self._key_chain_pairs)
        self._server = _low.Server(self._completion_queue)
        self._port = self._server.add_secure_http2_addr(
            address, server_credentials)
      self._server.start()

      self._server.service(None)

      self._pool.submit(self._spin, self._completion_queue, self._server)
      self._spinning = True

      return self

  # TODO(nathaniel): Expose graceful-shutdown semantics in which this object
  # enters a state in which it finishes ongoing RPCs but refuses new ones.
  def _stop(self):
    """Stops this ForeLink.

    This method must be called for proper termination of this object, and no
    attempts to exchange tickets with this object may be made after this method
    has been called.
    """
    with self._condition:
      self._server.stop()
      # TODO(nathaniel): Yep, this is weird. Deleting a server shouldn't have a
      # behaviorally significant side-effect.
      self._server = None
      self._completion_queue.stop()

      while self._spinning:
        self._condition.wait()

      self._port = None

  def __enter__(self):
    """See activated.Activated.__enter__ for specification."""
    return self._start()

  def __exit__(self, exc_type, exc_val, exc_tb):
    """See activated.Activated.__exit__ for specification."""
    self._stop()
    return False

  def start(self):
    """See activated.Activated.start for specification."""
    return self._start()

  def stop(self):
    """See activated.Activated.stop for specification."""
    self._stop()

  def port(self):
    """Identifies the port on which this ForeLink is servicing RPCs.

    Returns:
      The number of the port on which this ForeLink is servicing RPCs, or None
        if this ForeLink is not currently activated and servicing RPCs.
    """
    with self._condition:
      return self._port

  def accept_back_to_front_ticket(self, ticket):
    """See base_interfaces.ForeLink.accept_back_to_front_ticket for spec."""
    with self._condition:
      if self._server is None:
        return

      if ticket.kind is base_interfaces.BackToFrontTicket.Kind.CONTINUATION:
        self._continue(ticket.operation_id, ticket.payload)
      elif ticket.kind is base_interfaces.BackToFrontTicket.Kind.COMPLETION:
        self._complete(ticket.operation_id, ticket.payload)
      else:
        self._cancel(ticket.operation_id)