The RPC Framework core package.

This is the second generation of the old base package (framework.base)
and implements the translation between the new links and base
interfaces.

21 files changed, 2767 insertions, 14 deletions

diff --git a/src/python/grpcio/grpc/framework/core/__init__.py b/src/python/grpcio/grpc/framework/core/__init__.py
new file mode 100644
index 0000000000..7086519106
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/__init__.py
@@ -0,0 +1,30 @@
+# 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.
+
+
diff --git a/src/python/grpcio/grpc/framework/core/_constants.py b/src/python/grpcio/grpc/framework/core/_constants.py
new file mode 100644
index 0000000000..d3be3a4c4a
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_constants.py
@@ -0,0 +1,59 @@
+# 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.
+
+"""Private constants for the package."""
+
+from grpc.framework.interfaces.base import base
+from grpc.framework.interfaces.links import links
+
+TICKET_SUBSCRIPTION_FOR_BASE_SUBSCRIPTION_KIND = {
+    base.Subscription.Kind.NONE: links.Ticket.Subscription.NONE,
+    base.Subscription.Kind.TERMINATION_ONLY:
+        links.Ticket.Subscription.TERMINATION,
+    base.Subscription.Kind.FULL: links.Ticket.Subscription.FULL,
+    }
+
+# Mapping from abortive operation outcome to ticket termination to be
+# sent to the other side of the operation, or None to indicate that no
+# ticket should be sent to the other side in the event of such an
+# outcome.
+ABORTION_OUTCOME_TO_TICKET_TERMINATION = {
+    base.Outcome.CANCELLED: links.Ticket.Termination.CANCELLATION,
+    base.Outcome.EXPIRED: links.Ticket.Termination.EXPIRATION,
+    base.Outcome.LOCAL_SHUTDOWN: links.Ticket.Termination.SHUTDOWN,
+    base.Outcome.REMOTE_SHUTDOWN: None,
+    base.Outcome.RECEPTION_FAILURE: links.Ticket.Termination.RECEPTION_FAILURE,
+    base.Outcome.TRANSMISSION_FAILURE: None,
+    base.Outcome.LOCAL_FAILURE: links.Ticket.Termination.LOCAL_FAILURE,
+    base.Outcome.REMOTE_FAILURE: links.Ticket.Termination.REMOTE_FAILURE,
+}
+
+INTERNAL_ERROR_LOG_MESSAGE = ':-( RPC Framework (Core) internal error! )-:'
+TERMINATION_CALLBACK_EXCEPTION_LOG_MESSAGE = (
+    'Exception calling termination callback!')
diff --git a/src/python/grpcio/grpc/framework/core/_context.py b/src/python/grpcio/grpc/framework/core/_context.py
new file mode 100644
index 0000000000..24a12b612e
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_context.py
@@ -0,0 +1,92 @@
+# 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.
+
+"""State and behavior for operation context."""
+
+import time
+
+# _interfaces is referenced from specification in this module.
+from grpc.framework.core import _interfaces  # pylint: disable=unused-import
+from grpc.framework.interfaces.base import base
+
+
+class OperationContext(base.OperationContext):
+  """An implementation of interfaces.OperationContext."""
+
+  def __init__(
+      self, lock, termination_manager, transmission_manager,
+      expiration_manager):
+    """Constructor.
+
+    Args:
+      lock: The operation-wide lock.
+      termination_manager: The _interfaces.TerminationManager for the operation.
+      transmission_manager: The _interfaces.TransmissionManager for the
+        operation.
+      expiration_manager: The _interfaces.ExpirationManager for the operation.
+    """
+    self._lock = lock
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._expiration_manager = expiration_manager
+
+  def _abort(self, outcome):
+    with self._lock:
+      if self._termination_manager.outcome is None:
+        self._termination_manager.abort(outcome)
+        self._transmission_manager.abort(outcome)
+        self._expiration_manager.terminate()
+
+  def outcome(self):
+    """See base.OperationContext.outcome for specification."""
+    with self._lock:
+      return self._termination_manager.outcome
+
+  def add_termination_callback(self, callback):
+    """See base.OperationContext.add_termination_callback."""
+    with self._lock:
+      if self._termination_manager.outcome is None:
+        self._termination_manager.add_callback(callback)
+        return None
+      else:
+        return self._termination_manager.outcome
+
+  def time_remaining(self):
+    """See base.OperationContext.time_remaining for specification."""
+    with self._lock:
+      deadline = self._expiration_manager.deadline()
+    return max(0.0, deadline - time.time())
+
+  def cancel(self):
+    """See base.OperationContext.cancel for specification."""
+    self._abort(base.Outcome.CANCELLED)
+
+  def fail(self, exception):
+    """See base.OperationContext.fail for specification."""
+    self._abort(base.Outcome.LOCAL_FAILURE)
diff --git a/src/python/grpcio/grpc/framework/core/_emission.py b/src/python/grpcio/grpc/framework/core/_emission.py
new file mode 100644
index 0000000000..7c702ab2ce
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_emission.py
@@ -0,0 +1,97 @@
+# 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.
+
+"""State and behavior for handling emitted values."""
+
+from grpc.framework.core import _interfaces
+from grpc.framework.interfaces.base import base
+
+
+class EmissionManager(_interfaces.EmissionManager):
+  """An EmissionManager implementation."""
+
+  def __init__(
+      self, lock, termination_manager, transmission_manager,
+      expiration_manager):
+    """Constructor.
+
+    Args:
+      lock: The operation-wide lock.
+      termination_manager: The _interfaces.TerminationManager for the operation.
+      transmission_manager: The _interfaces.TransmissionManager for the
+        operation.
+      expiration_manager: The _interfaces.ExpirationManager for the operation.
+    """
+    self._lock = lock
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._expiration_manager = expiration_manager
+    self._ingestion_manager = None
+
+    self._initial_metadata_seen = False
+    self._payload_seen = False
+    self._completion_seen = False
+
+  def set_ingestion_manager(self, ingestion_manager):
+    """Sets the ingestion manager with which this manager will cooperate.
+
+    Args:
+      ingestion_manager: The _interfaces.IngestionManager for the operation.
+    """
+    self._ingestion_manager = ingestion_manager
+
+  def advance(
+      self, initial_metadata=None, payload=None, completion=None,
+      allowance=None):
+    initial_metadata_present = initial_metadata is not None
+    payload_present = payload is not None
+    completion_present = completion is not None
+    allowance_present = allowance is not None
+    with self._lock:
+      if self._termination_manager.outcome is None:
+        if (initial_metadata_present and (
+                self._initial_metadata_seen or self._payload_seen or
+                self._completion_seen) or
+            payload_present and self._completion_seen or
+            completion_present and self._completion_seen or
+            allowance_present and allowance <= 0):
+          self._termination_manager.abort(base.Outcome.LOCAL_FAILURE)
+          self._transmission_manager.abort(base.Outcome.LOCAL_FAILURE)
+          self._expiration_manager.terminate()
+        else:
+          self._initial_metadata_seen |= initial_metadata_present
+          self._payload_seen |= payload_present
+          self._completion_seen |= completion_present
+          if completion_present:
+            self._termination_manager.emission_complete()
+            self._ingestion_manager.local_emissions_done()
+          self._transmission_manager.advance(
+              initial_metadata, payload, completion, allowance)
+          if allowance_present:
+            self._ingestion_manager.add_local_allowance(allowance)
diff --git a/src/python/grpcio/grpc/framework/core/_end.py b/src/python/grpcio/grpc/framework/core/_end.py
new file mode 100644
index 0000000000..fb2c532df6
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_end.py
@@ -0,0 +1,251 @@
+# 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.
+
+"""Implementation of base.End."""
+
+import abc
+import enum
+import threading
+import uuid
+
+from grpc.framework.core import _operation
+from grpc.framework.core import _utilities
+from grpc.framework.foundation import callable_util
+from grpc.framework.foundation import later
+from grpc.framework.foundation import logging_pool
+from grpc.framework.interfaces.base import base
+from grpc.framework.interfaces.links import links
+from grpc.framework.interfaces.links import utilities
+
+_IDLE_ACTION_EXCEPTION_LOG_MESSAGE = 'Exception calling idle action!'
+
+
+class End(base.End, links.Link):
+  """A bridge between base.End and links.Link.
+
+  Implementations of this interface translate arriving tickets into
+  calls on application objects implementing base interfaces and
+  translate calls from application objects implementing base interfaces
+  into tickets sent to a joined link.
+  """
+  __metaclass__ = abc.ABCMeta
+
+
+class _Cycle(object):
+  """State for a single start-stop End lifecycle."""
+
+  def __init__(self, pool):
+    self.pool = pool
+    self.grace = False
+    self.futures = []
+    self.operations = {}
+    self.idle_actions = []
+
+
+def _abort(operations):
+  for operation in operations:
+    operation.abort(base.Outcome.LOCAL_SHUTDOWN)
+
+
+def _cancel_futures(futures):
+  for future in futures:
+    futures.cancel()
+
+
+def _future_shutdown(lock, cycle, event):
+  def in_future():
+    with lock:
+      _abort(cycle.operations.values())
+      _cancel_futures(cycle.futures)
+      pool = cycle.pool
+    cycle.pool.shutdown(wait=True)
+  return in_future
+
+
+def _termination_action(lock, stats, operation_id, cycle):
+  """Constructs the termination action for a single operation.
+
+  Args:
+    lock: A lock to hold during the termination action.
+    states: A mapping from base.Outcome values to integers to increment with
+      the outcome given to the termination action.
+    operation_id: The operation ID for the termination action.
+    cycle: A _Cycle value to be updated during the termination action.
+
+  Returns:
+    A callable that takes an operation outcome as its sole parameter and that
+      should be used as the termination action for the operation associated
+      with the given operation ID.
+  """
+  def termination_action(outcome):
+    with lock:
+      stats[outcome] += 1
+      cycle.operations.pop(operation_id, None)
+      if not cycle.operations:
+        for action in cycle.idle_actions:
+          cycle.pool.submit(action)
+        cycle.idle_actions = []
+        if cycle.grace:
+          _cancel_futures(cycle.futures)
+  return termination_action
+
+
+class _End(End):
+  """An End implementation."""
+
+  def __init__(self, servicer_package):
+    """Constructor.
+
+    Args:
+      servicer_package: A _ServicerPackage for servicing operations or None if
+        this end will not be used to service operations.
+    """
+    self._lock = threading.Condition()
+    self._servicer_package = servicer_package
+
+    self._stats = {outcome: 0 for outcome in base.Outcome}
+
+    self._mate = None
+
+    self._cycle = None
+
+  def start(self):
+    """See base.End.start for specification."""
+    with self._lock:
+      if self._cycle is not None:
+        raise ValueError('Tried to start a not-stopped End!')
+      else:
+        self._cycle = _Cycle(logging_pool.pool(1))
+
+  def stop(self, grace):
+    """See base.End.stop for specification."""
+    with self._lock:
+      if self._cycle is None:
+        event = threading.Event()
+        event.set()
+        return event
+      elif not self._cycle.operations:
+        event = threading.Event()
+        self._cycle.pool.submit(event.set)
+        self._cycle.pool.shutdown(wait=False)
+        self._cycle = None
+        return event
+      else:
+        self._cycle.grace = True
+        event = threading.Event()
+        self._cycle.idle_actions.append(event.set)
+        if 0 < grace:
+          future = later.later(
+              grace, _future_shutdown(self._lock, self._cycle, event))
+          self._cycle.futures.append(future)
+        else:
+          _abort(self._cycle.operations.values())
+        return event
+
+  def operate(
+      self, group, method, subscription, timeout, initial_metadata=None,
+      payload=None, completion=None):
+    """See base.End.operate for specification."""
+    operation_id = uuid.uuid4()
+    with self._lock:
+      if self._cycle is None or self._cycle.grace:
+        raise ValueError('Can\'t operate on stopped or stopping End!')
+      termination_action = _termination_action(
+          self._lock, self._stats, operation_id, self._cycle)
+      operation = _operation.invocation_operate(
+          operation_id, group, method, subscription, timeout, initial_metadata,
+          payload, completion, self._mate.accept_ticket, termination_action,
+          self._cycle.pool)
+      self._cycle.operations[operation_id] = operation
+      return operation.context, operation.operator
+
+  def operation_stats(self):
+    """See base.End.operation_stats for specification."""
+    with self._lock:
+      return dict(self._stats)
+
+  def add_idle_action(self, action):
+    """See base.End.add_idle_action for specification."""
+    with self._lock:
+      if self._cycle is None:
+        raise ValueError('Can\'t add idle action to stopped End!')
+      action_with_exceptions_logged = callable_util.with_exceptions_logged(
+          action, _IDLE_ACTION_EXCEPTION_LOG_MESSAGE)
+      if self._cycle.operations:
+        self._cycle.idle_actions.append(action_with_exceptions_logged)
+      else:
+        self._cycle.pool.submit(action_with_exceptions_logged)
+
+  def accept_ticket(self, ticket):
+    """See links.Link.accept_ticket for specification."""
+    with self._lock:
+      if self._cycle is not None and not self._cycle.grace:
+        operation = self._cycle.operations.get(ticket.operation_id)
+        if operation is not None:
+          operation.handle_ticket(ticket)
+        elif self._servicer_package is not None:
+          termination_action = _termination_action(
+              self._lock, self._stats, ticket.operation_id, self._cycle)
+          operation = _operation.service_operate(
+              self._servicer_package, ticket, self._mate.accept_ticket,
+              termination_action, self._cycle.pool)
+          if operation is not None:
+            self._cycle.operations[ticket.operation_id] = operation
+
+  def join_link(self, link):
+    """See links.Link.join_link for specification."""
+    with self._lock:
+      self._mate = utilities.NULL_LINK if link is None else link
+
+
+def serviceless_end_link():
+  """Constructs an End usable only for invoking operations.
+
+  Returns:
+    An End usable for translating operations into ticket exchange.
+  """
+  return _End(None)
+
+
+def serviceful_end_link(servicer, default_timeout, maximum_timeout):
+  """Constructs an End capable of servicing operations.
+
+  Args:
+    servicer: An interfaces.Servicer for servicing operations.
+    default_timeout: A length of time in seconds to be used as the default
+      time alloted for a single operation.
+    maximum_timeout: A length of time in seconds to be used as the maximum
+      time alloted for a single operation.
+
+  Returns:
+    An End capable of servicing the operations requested of it through ticket
+      exchange.
+  """
+  return _End(
+      _utilities.ServicerPackage(servicer, default_timeout, maximum_timeout))
diff --git a/src/python/grpcio/grpc/framework/core/_expiration.py b/src/python/grpcio/grpc/framework/core/_expiration.py
new file mode 100644
index 0000000000..d94bdf2d2b
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_expiration.py
@@ -0,0 +1,152 @@
+# 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.
+
+"""State and behavior for operation expiration."""
+
+import time
+
+from grpc.framework.core import _interfaces
+from grpc.framework.foundation import later
+from grpc.framework.interfaces.base import base
+
+
+class _ExpirationManager(_interfaces.ExpirationManager):
+  """An implementation of _interfaces.ExpirationManager."""
+
+  def __init__(
+      self, commencement, timeout, maximum_timeout, lock, termination_manager,
+      transmission_manager):
+    """Constructor.
+
+    Args:
+      commencement: The time in seconds since the epoch at which the operation
+        began.
+      timeout: A length of time in seconds to allow for the operation to run.
+      maximum_timeout: The maximum length of time in seconds to allow for the
+        operation to run despite what is requested via this object's
+        change_timout method.
+      lock: The operation-wide lock.
+      termination_manager: The _interfaces.TerminationManager for the operation.
+      transmission_manager: The _interfaces.TransmissionManager for the
+        operation.
+    """
+    self._lock = lock
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._commencement = commencement
+    self._maximum_timeout = maximum_timeout
+
+    self._timeout = timeout
+    self._deadline = commencement + timeout
+    self._index = None
+    self._future = None
+
+  def _expire(self, index):
+    def expire():
+      with self._lock:
+        if self._future is not None and index == self._index:
+          self._future = None
+          self._termination_manager.expire()
+          self._transmission_manager.abort(base.Outcome.EXPIRED)
+    return expire
+
+  def start(self):
+    self._index = 0
+    self._future = later.later(self._timeout, self._expire(0))
+
+  def change_timeout(self, timeout):
+    if self._future is not None and timeout != self._timeout:
+      self._future.cancel()
+      new_timeout = min(timeout, self._maximum_timeout)
+      new_index = self._index + 1
+      self._timeout = new_timeout
+      self._deadline = self._commencement + new_timeout
+      self._index = new_index
+      delay = self._deadline - time.time()
+      self._future = later.later(delay, self._expire(new_index))
+      if new_timeout != timeout:
+        self._transmission_manager.timeout(new_timeout)
+
+  def deadline(self):
+    return self._deadline
+
+  def terminate(self):
+    if self._future:
+      self._future.cancel()
+      self._future = None
+    self._deadline_index = None
+
+
+def invocation_expiration_manager(
+    timeout, lock, termination_manager, transmission_manager):
+  """Creates an _interfaces.ExpirationManager appropriate for front-side use.
+
+  Args:
+    timeout: A length of time in seconds to allow for the operation to run.
+    lock: The operation-wide lock.
+    termination_manager: The _interfaces.TerminationManager for the operation.
+    transmission_manager: The _interfaces.TransmissionManager for the
+      operation.
+
+  Returns:
+    An _interfaces.ExpirationManager appropriate for invocation-side use.
+  """
+  expiration_manager = _ExpirationManager(
+      time.time(), timeout, timeout, lock, termination_manager,
+      transmission_manager)
+  expiration_manager.start()
+  return expiration_manager
+
+
+def service_expiration_manager(
+    timeout, default_timeout, maximum_timeout, lock, termination_manager,
+    transmission_manager):
+  """Creates an _interfaces.ExpirationManager appropriate for back-side use.
+
+  Args:
+    timeout: A length of time in seconds to allow for the operation to run. May
+      be None in which case default_timeout will be used.
+    default_timeout: The default length of time in seconds to allow for the
+      operation to run if the front-side customer has not specified such a value
+      (or if the value they specified is not yet known).
+    maximum_timeout: The maximum length of time in seconds to allow for the
+      operation to run.
+    lock: The operation-wide lock.
+    termination_manager: The _interfaces.TerminationManager for the operation.
+    transmission_manager: The _interfaces.TransmissionManager for the
+      operation.
+
+  Returns:
+    An _interfaces.ExpirationManager appropriate for service-side use.
+  """
+  expiration_manager = _ExpirationManager(
+      time.time(), default_timeout if timeout is None else timeout,
+      maximum_timeout, lock, termination_manager, transmission_manager)
+  expiration_manager.start()
+  return expiration_manager
diff --git a/src/python/grpcio/grpc/framework/core/_ingestion.py b/src/python/grpcio/grpc/framework/core/_ingestion.py
new file mode 100644
index 0000000000..59f7f8adc8
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_ingestion.py
@@ -0,0 +1,410 @@
+# 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.
+
+"""State and behavior for ingestion during an operation."""
+
+import abc
+import collections
+
+from grpc.framework.core import _constants
+from grpc.framework.core import _interfaces
+from grpc.framework.foundation import abandonment
+from grpc.framework.foundation import callable_util
+from grpc.framework.interfaces.base import base
+
+_CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE = 'Exception initializing ingestion!'
+_INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!'
+
+
+class _SubscriptionCreation(collections.namedtuple(
+    '_SubscriptionCreation', ('subscription', 'remote_error', 'abandoned'))):
+  """A sum type for the outcome of ingestion initialization.
+
+  Either subscription will be non-None, remote_error will be True, or abandoned
+  will be True.
+
+  Attributes:
+    subscription: A base.Subscription describing the customer's interest in
+      operation values from the other side.
+    remote_error: A boolean indicating that the subscription could not be
+      created due to an error on the remote side of the operation.
+    abandoned: A boolean indicating that subscription creation was abandoned.
+  """
+
+
+class _SubscriptionCreator(object):
+  """Common specification of subscription-creating behavior."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def create(self, group, method):
+    """Creates the base.Subscription of the local customer.
+
+    Any exceptions raised by this method should be attributed to and treated as
+    defects in the customer code called by this method.
+
+    Args:
+      group: The group identifier of the operation.
+      method: The method identifier of the operation.
+
+    Returns:
+      A _SubscriptionCreation describing the result of subscription creation.
+    """
+    raise NotImplementedError()
+
+
+class _ServiceSubscriptionCreator(_SubscriptionCreator):
+  """A _SubscriptionCreator appropriate for service-side use."""
+
+  def __init__(self, servicer, operation_context, output_operator):
+    """Constructor.
+
+    Args:
+      servicer: The base.Servicer that will service the operation.
+      operation_context: A base.OperationContext for the operation to be passed
+        to the customer.
+      output_operator: A base.Operator for the operation to be passed to the
+        customer and to be called by the customer to accept operation data
+        emitted by the customer.
+    """
+    self._servicer = servicer
+    self._operation_context = operation_context
+    self._output_operator = output_operator
+
+  def create(self, group, method):
+    try:
+      subscription = self._servicer.service(
+          group, method, self._operation_context, self._output_operator)
+    except base.NoSuchMethodError:
+      return _SubscriptionCreation(None, True, False)
+    except abandonment.Abandoned:
+      return _SubscriptionCreation(None, False, True)
+    else:
+      return _SubscriptionCreation(subscription, False, False)
+
+
+def _wrap(behavior):
+  def wrapped(*args, **kwargs):
+    try:
+      behavior(*args, **kwargs)
+    except abandonment.Abandoned:
+      return False
+    else:
+      return True
+  return wrapped
+
+
+class _IngestionManager(_interfaces.IngestionManager):
+  """An implementation of _interfaces.IngestionManager."""
+
+  def __init__(
+      self, lock, pool, subscription, subscription_creator, termination_manager,
+      transmission_manager, expiration_manager):
+    """Constructor.
+
+    Args:
+      lock: The operation-wide lock.
+      pool: A thread pool in which to execute customer code.
+      subscription: A base.Subscription describing the customer's interest in
+        operation values from the other side. May be None if
+        subscription_creator is not None.
+      subscription_creator: A _SubscriptionCreator wrapping the portion of
+        customer code that when called returns the base.Subscription describing
+        the customer's interest in operation values from the other side. May be
+        None if subscription is not None.
+      termination_manager: The _interfaces.TerminationManager for the operation.
+      transmission_manager: The _interfaces.TransmissionManager for the
+        operation.
+      expiration_manager: The _interfaces.ExpirationManager for the operation.
+    """
+    self._lock = lock
+    self._pool = pool
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._expiration_manager = expiration_manager
+
+    if subscription is None:
+      self._subscription_creator = subscription_creator
+      self._wrapped_operator = None
+    elif subscription.kind is base.Subscription.Kind.FULL:
+      self._subscription_creator = None
+      self._wrapped_operator = _wrap(subscription.operator.advance)
+    else:
+      # TODO(nathaniel): Support other subscriptions.
+      raise ValueError('Unsupported subscription "%s"!' % subscription.kind)
+    self._pending_initial_metadata = None
+    self._pending_payloads = []
+    self._pending_completion = None
+    self._local_allowance = 1
+    # A nonnegative integer or None, with None indicating that the local
+    # customer is done emitting anyway so there's no need to bother it by
+    # informing it that the remote customer has granted it further permission to
+    # emit.
+    self._remote_allowance = 0
+    self._processing = False
+
+  def _abort_internal_only(self):
+    self._subscription_creator = None
+    self._wrapped_operator = None
+    self._pending_initial_metadata = None
+    self._pending_payloads = None
+    self._pending_completion = None
+
+  def _abort_and_notify(self, outcome):
+    self._abort_internal_only()
+    self._termination_manager.abort(outcome)
+    self._transmission_manager.abort(outcome)
+    self._expiration_manager.terminate()
+
+  def _operator_next(self):
+    """Computes the next step for full-subscription ingestion.
+
+    Returns:
+      An initial_metadata, payload, completion, allowance, continue quintet
+        indicating what operation values (if any) are available to pass into
+        customer code and whether or not there is anything immediately
+        actionable to call customer code to do.
+    """
+    if self._wrapped_operator is None:
+      return None, None, None, None, False
+    else:
+      initial_metadata, payload, completion, allowance, action = [None] * 5
+      if self._pending_initial_metadata is not None:
+        initial_metadata = self._pending_initial_metadata
+        self._pending_initial_metadata = None
+        action = True
+      if self._pending_payloads and 0 < self._local_allowance:
+        payload = self._pending_payloads.pop(0)
+        self._local_allowance -= 1
+        action = True
+      if not self._pending_payloads and self._pending_completion is not None:
+        completion = self._pending_completion
+        self._pending_completion = None
+        action = True
+      if self._remote_allowance is not None and 0 < self._remote_allowance:
+        allowance = self._remote_allowance
+        self._remote_allowance = 0
+        action = True
+      return initial_metadata, payload, completion, allowance, bool(action)
+
+  def _operator_process(
+      self, wrapped_operator, initial_metadata, payload,
+      completion, allowance):
+    while True:
+      advance_outcome = callable_util.call_logging_exceptions(
+          wrapped_operator, _INGESTION_EXCEPTION_LOG_MESSAGE,
+          initial_metadata=initial_metadata, payload=payload,
+          completion=completion, allowance=allowance)
+      if advance_outcome.exception is None:
+        if advance_outcome.return_value:
+          with self._lock:
+            if self._termination_manager.outcome is not None:
+              return
+            if completion is not None:
+              self._termination_manager.ingestion_complete()
+            initial_metadata, payload, completion, allowance, moar = (
+                self._operator_next())
+            if not moar:
+              self._processing = False
+              return
+        else:
+          with self._lock:
+            if self._termination_manager.outcome is None:
+              self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+            return
+      else:
+        with self._lock:
+          if self._termination_manager.outcome is None:
+            self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+          return
+
+  def _operator_post_create(self, subscription):
+    wrapped_operator = _wrap(subscription.operator.advance)
+    with self._lock:
+      if self._termination_manager.outcome is not None:
+        return
+      self._wrapped_operator = wrapped_operator
+      self._subscription_creator = None
+      metadata, payload, completion, allowance, moar = self._operator_next()
+      if not moar:
+        self._processing = False
+        return
+    self._operator_process(
+        wrapped_operator, metadata, payload, completion, allowance)
+
+  def _create(self, subscription_creator, group, name):
+    outcome = callable_util.call_logging_exceptions(
+        subscription_creator.create, _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE,
+        group, name)
+    if outcome.return_value is None:
+      with self._lock:
+        if self._termination_manager.outcome is None:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+    elif outcome.return_value.abandoned:
+      with self._lock:
+        if self._termination_manager.outcome is None:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+    elif outcome.return_value.remote_error:
+      with self._lock:
+        if self._termination_manager.outcome is None:
+          self._abort_and_notify(base.Outcome.REMOTE_FAILURE)
+    elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL:
+      self._operator_post_create(outcome.return_value.subscription)
+    else:
+      # TODO(nathaniel): Support other subscriptions.
+      raise ValueError(
+          'Unsupported "%s"!' % outcome.return_value.subscription.kind)
+
+  def _store_advance(self, initial_metadata, payload, completion, allowance):
+    if initial_metadata is not None:
+      self._pending_initial_metadata = initial_metadata
+    if payload is not None:
+      self._pending_payloads.append(payload)
+    if completion is not None:
+      self._pending_completion = completion
+    if allowance is not None and self._remote_allowance is not None:
+      self._remote_allowance += allowance
+
+  def _operator_advance(self, initial_metadata, payload, completion, allowance):
+    if self._processing:
+      self._store_advance(initial_metadata, payload, completion, allowance)
+    else:
+      action = False
+      if initial_metadata is not None:
+        action = True
+      if payload is not None:
+        if 0 < self._local_allowance:
+          self._local_allowance -= 1
+          action = True
+        else:
+          self._pending_payloads.append(payload)
+          payload = False
+      if completion is not None:
+        if self._pending_payloads:
+          self._pending_completion = completion
+        else:
+          action = True
+      if allowance is not None and self._remote_allowance is not None:
+        allowance += self._remote_allowance
+        self._remote_allowance = 0
+        action = True
+      if action:
+        self._pool.submit(
+            callable_util.with_exceptions_logged(
+                self._operator_process, _constants.INTERNAL_ERROR_LOG_MESSAGE),
+            self._wrapped_operator, initial_metadata, payload, completion,
+            allowance)
+
+  def set_group_and_method(self, group, method):
+    """See _interfaces.IngestionManager.set_group_and_method for spec."""
+    if self._subscription_creator is not None and not self._processing:
+      self._pool.submit(
+          callable_util.with_exceptions_logged(
+              self._create, _constants.INTERNAL_ERROR_LOG_MESSAGE),
+          self._subscription_creator, group, method)
+      self._processing = True
+
+  def add_local_allowance(self, allowance):
+    """See _interfaces.IngestionManager.add_local_allowance for spec."""
+    if any((self._subscription_creator, self._wrapped_operator,)):
+      self._local_allowance += allowance
+      if not self._processing:
+        initial_metadata, payload, completion, allowance, moar = (
+            self._operator_next())
+        if moar:
+          self._pool.submit(
+              callable_util.with_exceptions_logged(
+                  self._operator_process,
+                  _constants.INTERNAL_ERROR_LOG_MESSAGE),
+              initial_metadata, payload, completion, allowance)
+
+  def local_emissions_done(self):
+    self._remote_allowance = None
+
+  def advance(self, initial_metadata, payload, completion, allowance):
+    """See _interfaces.IngestionManager.advance for specification."""
+    if self._subscription_creator is not None:
+      self._store_advance(initial_metadata, payload, completion, allowance)
+    elif self._wrapped_operator is not None:
+      self._operator_advance(initial_metadata, payload, completion, allowance)
+
+
+def invocation_ingestion_manager(
+    subscription, lock, pool, termination_manager, transmission_manager,
+    expiration_manager):
+  """Creates an IngestionManager appropriate for invocation-side use.
+
+  Args:
+    subscription: A base.Subscription indicating the customer's interest in the
+      data and results from the service-side of the operation.
+    lock: The operation-wide lock.
+    pool: A thread pool in which to execute customer code.
+    termination_manager: The _interfaces.TerminationManager for the operation.
+    transmission_manager: The _interfaces.TransmissionManager for the
+      operation.
+    expiration_manager: The _interfaces.ExpirationManager for the operation.
+
+  Returns:
+    An IngestionManager appropriate for invocation-side use.
+  """
+  return _IngestionManager(
+      lock, pool, subscription, None, termination_manager, transmission_manager,
+      expiration_manager)
+
+
+def service_ingestion_manager(
+    servicer, operation_context, output_operator, lock, pool,
+    termination_manager, transmission_manager, expiration_manager):
+  """Creates an IngestionManager appropriate for service-side use.
+
+  The returned IngestionManager will require its set_group_and_name method to be
+  called before its advance method may be called.
+
+  Args:
+    servicer: A base.Servicer for servicing the operation.
+    operation_context: A base.OperationContext for the operation to be passed to
+      the customer.
+    output_operator: A base.Operator for the operation to be passed to the
+      customer and to be called by the customer to accept operation data output
+      by the customer.
+    lock: The operation-wide lock.
+    pool: A thread pool in which to execute customer code.
+    termination_manager: The _interfaces.TerminationManager for the operation.
+    transmission_manager: The _interfaces.TransmissionManager for the
+      operation.
+    expiration_manager: The _interfaces.ExpirationManager for the operation.
+
+  Returns:
+    An IngestionManager appropriate for service-side use.
+  """
+  subscription_creator = _ServiceSubscriptionCreator(
+      servicer, operation_context, output_operator)
+  return _IngestionManager(
+      lock, pool, None, subscription_creator, termination_manager,
+      transmission_manager, expiration_manager)
diff --git a/src/python/grpcio/grpc/framework/core/_interfaces.py b/src/python/grpcio/grpc/framework/core/_interfaces.py
new file mode 100644
index 0000000000..a626b9f767
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_interfaces.py
@@ -0,0 +1,308 @@
+# 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.
+
+"""Package-internal interfaces."""
+
+import abc
+
+from grpc.framework.interfaces.base import base
+
+
+class TerminationManager(object):
+  """An object responsible for handling the termination of an operation.
+
+  Attributes:
+    outcome: None if the operation is active or a base.Outcome value if it has
+      terminated.
+  """
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def add_callback(self, callback):
+    """Registers a callback to be called on operation termination.
+
+    If the operation has already terminated the callback will not be called.
+
+    Args:
+      callback: A callable that will be passed an interfaces.Outcome value.
+
+    Returns:
+      None if the operation has not yet terminated and the passed callback will
+        be called when it does, or a base.Outcome value describing the operation
+        termination if the operation has terminated and the callback will not be
+        called as a result of this method call.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def emission_complete(self):
+    """Indicates that emissions from customer code have completed."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def transmission_complete(self):
+    """Indicates that transmissions to the remote end are complete.
+
+    Returns:
+      True if the operation has terminated or False if the operation remains
+        ongoing.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def reception_complete(self):
+    """Indicates that reception from the other side is complete."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def ingestion_complete(self):
+    """Indicates that customer code ingestion of received values is complete."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def expire(self):
+    """Indicates that the operation must abort because it has taken too long."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def abort(self, outcome):
+    """Indicates that the operation must abort for the indicated reason.
+
+    Args:
+      outcome: An interfaces.Outcome indicating operation abortion.
+    """
+    raise NotImplementedError()
+
+
+class TransmissionManager(object):
+  """A manager responsible for transmitting to the other end of an operation."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def kick_off(
+      self, group, method, timeout, initial_metadata, payload, completion,
+      allowance):
+    """Transmits the values associated with operation invocation."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def advance(self, initial_metadata, payload, completion, allowance):
+    """Accepts values for transmission to the other end of the operation.
+
+    Args:
+      initial_metadata: An initial metadata value to be transmitted to the other
+        side of the operation. May only ever be non-None once.
+      payload: A payload value.
+      completion: A base.Completion value. May only ever be non-None in the last
+        transmission to be made to the other side.
+      allowance: A positive integer communicating the number of additional
+        payloads allowed to be transmitted from the other side to this side of
+        the operation, or None if no additional allowance is being granted in
+        this call.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def timeout(self, timeout):
+    """Accepts for transmission to the other side a new timeout value.
+
+    Args:
+      timeout: A positive float used as the new timeout value for the operation
+        to be transmitted to the other side.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def allowance(self, allowance):
+    """Indicates to this manager that the remote customer is allowing payloads.
+
+    Args:
+      allowance: A positive integer indicating the number of additional payloads
+        the remote customer is allowing to be transmitted from this side of the
+        operation.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def remote_complete(self):
+    """Indicates to this manager that data from the remote side is complete."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def abort(self, outcome):
+    """Indicates that the operation has aborted.
+
+    Args:
+      outcome: An interfaces.Outcome for the operation. If None, indicates that
+        the operation abortion should not be communicated to the other side of
+        the operation.
+    """
+    raise NotImplementedError()
+
+
+class ExpirationManager(object):
+  """A manager responsible for aborting the operation if it runs out of time."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def change_timeout(self, timeout):
+    """Changes the timeout allotted for the operation.
+
+    Operation duration is always measure from the beginning of the operation;
+    calling this method changes the operation's allotted time to timeout total
+    seconds, not timeout seconds from the time of this method call.
+
+    Args:
+      timeout: A length of time in seconds to allow for the operation.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def deadline(self):
+    """Returns the time until which the operation is allowed to run.
+
+    Returns:
+      The time (seconds since the epoch) at which the operation will expire.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def terminate(self):
+    """Indicates to this manager that the operation has terminated."""
+    raise NotImplementedError()
+
+
+class EmissionManager(base.Operator):
+  """A manager of values emitted by customer code."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def advance(
+      self, initial_metadata=None, payload=None, completion=None,
+      allowance=None):
+    """Accepts a value emitted by customer code.
+
+    This method should only be called by customer code.
+
+    Args:
+      initial_metadata: An initial metadata value emitted by the local customer
+        to be sent to the other side of the operation.
+      payload: A payload value emitted by the local customer to be sent to the
+        other side of the operation.
+      completion: A Completion value emitted by the local customer to be sent to
+        the other side of the operation.
+      allowance: A positive integer indicating an additional number of payloads
+        that the local customer is willing to accept from the other side of the
+        operation.
+    """
+    raise NotImplementedError()
+
+
+class IngestionManager(object):
+  """A manager responsible for executing customer code.
+
+  This name of this manager comes from its responsibility to pass successive
+  values from the other side of the operation into the code of the local
+  customer.
+  """
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def set_group_and_method(self, group, method):
+    """Communicates to this IngestionManager the operation group and method.
+
+    Args:
+      group: The group identifier of the operation.
+      method: The method identifier of the operation.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def add_local_allowance(self, allowance):
+    """Communicates to this IngestionManager that more payloads may be ingested.
+
+    Args:
+      allowance: A positive integer indicating an additional number of payloads
+        that the local customer is willing to ingest.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def local_emissions_done(self):
+    """Indicates to this manager that local emissions are done."""
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def advance(self, initial_metadata, payload, completion, allowance):
+    """Advances the operation by passing values to the local customer."""
+    raise NotImplementedError()
+
+
+class ReceptionManager(object):
+  """A manager responsible for receiving tickets from the other end."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def receive_ticket(self, ticket):
+    """Handle a ticket from the other side of the operation.
+
+    Args:
+      ticket: An interfaces.BackToFrontTicket or interfaces.FrontToBackTicket
+        appropriate to this end of the operation and this object.
+    """
+    raise NotImplementedError()
+
+
+class Operation(object):
+  """An ongoing operation.
+
+  Attributes:
+    context: A base.OperationContext object for the operation.
+    operator: A base.Operator object for the operation for use by the customer
+      of the operation.
+  """
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def handle_ticket(self, ticket):
+    """Handle a ticket from the other side of the operation.
+
+    Args:
+      ticket: A links.Ticket from the other side of the operation.
+    """
+    raise NotImplementedError()
+
+  @abc.abstractmethod
+  def abort(self, outcome):
+    """Aborts the operation.
+
+    Args:
+      outcome: A base.Outcome value indicating operation abortion.
+    """
+    raise NotImplementedError()
diff --git a/src/python/grpcio/grpc/framework/core/_operation.py b/src/python/grpcio/grpc/framework/core/_operation.py
new file mode 100644
index 0000000000..d20e40a53d
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_operation.py
@@ -0,0 +1,192 @@
+# 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.
+
+"""Implementation of operations."""
+
+import threading
+
+# _utilities is referenced from specification in this module.
+from grpc.framework.core import _context
+from grpc.framework.core import _emission
+from grpc.framework.core import _expiration
+from grpc.framework.core import _ingestion
+from grpc.framework.core import _interfaces
+from grpc.framework.core import _reception
+from grpc.framework.core import _termination
+from grpc.framework.core import _transmission
+from grpc.framework.core import _utilities  # pylint: disable=unused-import
+
+
+class _EasyOperation(_interfaces.Operation):
+  """A trivial implementation of interfaces.Operation."""
+
+  def __init__(
+      self, lock, termination_manager, transmission_manager, expiration_manager,
+      context, operator, reception_manager):
+    """Constructor.
+
+    Args:
+      lock: The operation-wide lock.
+      termination_manager: The _interfaces.TerminationManager for the operation.
+      transmission_manager: The _interfaces.TransmissionManager for the
+        operation.
+      expiration_manager: The _interfaces.ExpirationManager for the operation.
+      context: A base.OperationContext for use by the customer during the
+        operation.
+      operator: A base.Operator for use by the customer during the operation.
+      reception_manager: The _interfaces.ReceptionManager for the operation.
+    """
+    self._lock = lock
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._expiration_manager = expiration_manager
+    self._reception_manager = reception_manager
+
+    self.context = context
+    self.operator = operator
+
+  def handle_ticket(self, ticket):
+    with self._lock:
+      self._reception_manager.receive_ticket(ticket)
+
+  def abort(self, outcome):
+    with self._lock:
+      if self._termination_manager.outcome is None:
+        self._termination_manager.abort(outcome)
+        self._transmission_manager.abort(outcome)
+        self._expiration_manager.terminate()
+
+
+def invocation_operate(
+    operation_id, group, method, subscription, timeout, initial_metadata,
+    payload, completion, ticket_sink, termination_action, pool):
+  """Constructs objects necessary for front-side operation management.
+
+  Args:
+    operation_id: An object identifying the operation.
+    group: The group identifier of the operation.
+    method: The method identifier of the operation.
+    subscription: A base.Subscription describing the customer's interest in the
+      results of the operation.
+    timeout: A length of time in seconds to allow for the operation.
+    initial_metadata: An initial metadata value to be sent to the other side of
+      the operation. May be None if the initial metadata will be passed later or
+      if there will be no initial metadata passed at all.
+    payload: The first payload value to be transmitted to the other side. May be
+      None if there is no such value or if the customer chose not to pass it at
+      operation invocation.
+    completion: A base.Completion value indicating the end of values passed to
+      the other side of the operation.
+    ticket_sink: A callable that accepts links.Tickets and delivers them to the
+      other side of the operation.
+    termination_action: A callable that accepts the outcome of the operation as
+      a base.Outcome value to be called on operation completion.
+    pool: A thread pool with which to do the work of the operation.
+
+  Returns:
+    An _interfaces.Operation for the operation.
+  """
+  lock = threading.Lock()
+  with lock:
+    termination_manager = _termination.invocation_termination_manager(
+        termination_action, pool)
+    transmission_manager = _transmission.TransmissionManager(
+        operation_id, ticket_sink, lock, pool, termination_manager)
+    expiration_manager = _expiration.invocation_expiration_manager(
+        timeout, lock, termination_manager, transmission_manager)
+    operation_context = _context.OperationContext(
+        lock, termination_manager, transmission_manager, expiration_manager)
+    emission_manager = _emission.EmissionManager(
+        lock, termination_manager, transmission_manager, expiration_manager)
+    ingestion_manager = _ingestion.invocation_ingestion_manager(
+        subscription, lock, pool, termination_manager, transmission_manager,
+        expiration_manager)
+    reception_manager = _reception.ReceptionManager(
+        termination_manager, transmission_manager, expiration_manager,
+        ingestion_manager)
+
+    termination_manager.set_expiration_manager(expiration_manager)
+    transmission_manager.set_expiration_manager(expiration_manager)
+    emission_manager.set_ingestion_manager(ingestion_manager)
+
+    transmission_manager.kick_off(
+        group, method, timeout, initial_metadata, payload, completion, None)
+
+  return _EasyOperation(
+      lock, termination_manager, transmission_manager, expiration_manager,
+      operation_context, emission_manager, reception_manager)
+
+
+def service_operate(
+    servicer_package, ticket, ticket_sink, termination_action, pool):
+  """Constructs an Operation for service of an operation.
+
+  Args:
+    servicer_package: A _utilities.ServicerPackage to be used servicing the
+      operation.
+    ticket: The first links.Ticket received for the operation.
+    ticket_sink: A callable that accepts links.Tickets and delivers them to the
+      other side of the operation.
+    termination_action: A callable that accepts the outcome of the operation as
+      a base.Outcome value to be called on operation completion.
+    pool: A thread pool with which to do the work of the operation.
+
+  Returns:
+    An _interfaces.Operation for the operation.
+  """
+  lock = threading.Lock()
+  with lock:
+    termination_manager = _termination.service_termination_manager(
+        termination_action, pool)
+    transmission_manager = _transmission.TransmissionManager(
+        ticket.operation_id, ticket_sink, lock, pool, termination_manager)
+    expiration_manager = _expiration.service_expiration_manager(
+        ticket.timeout, servicer_package.default_timeout,
+        servicer_package.maximum_timeout, lock, termination_manager,
+        transmission_manager)
+    operation_context = _context.OperationContext(
+        lock, termination_manager, transmission_manager, expiration_manager)
+    emission_manager = _emission.EmissionManager(
+        lock, termination_manager, transmission_manager, expiration_manager)
+    ingestion_manager = _ingestion.service_ingestion_manager(
+        servicer_package.servicer, operation_context, emission_manager, lock,
+        pool, termination_manager, transmission_manager, expiration_manager)
+    reception_manager = _reception.ReceptionManager(
+        termination_manager, transmission_manager, expiration_manager,
+        ingestion_manager)
+
+    termination_manager.set_expiration_manager(expiration_manager)
+    transmission_manager.set_expiration_manager(expiration_manager)
+    emission_manager.set_ingestion_manager(ingestion_manager)
+
+    reception_manager.receive_ticket(ticket)
+
+  return _EasyOperation(
+      lock, termination_manager, transmission_manager, expiration_manager,
+      operation_context, emission_manager, reception_manager)
diff --git a/src/python/grpcio/grpc/framework/core/_reception.py b/src/python/grpcio/grpc/framework/core/_reception.py
new file mode 100644
index 0000000000..b64faf8146
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_reception.py
@@ -0,0 +1,137 @@
+# 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.
+
+"""State and behavior for ticket reception."""
+
+from grpc.framework.core import _interfaces
+from grpc.framework.interfaces.base import base
+from grpc.framework.interfaces.base import utilities
+from grpc.framework.interfaces.links import links
+
+_REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME = {
+    links.Ticket.Termination.CANCELLATION: base.Outcome.CANCELLED,
+    links.Ticket.Termination.EXPIRATION: base.Outcome.EXPIRED,
+    links.Ticket.Termination.SHUTDOWN: base.Outcome.REMOTE_SHUTDOWN,
+    links.Ticket.Termination.RECEPTION_FAILURE: base.Outcome.RECEPTION_FAILURE,
+    links.Ticket.Termination.TRANSMISSION_FAILURE:
+        base.Outcome.TRANSMISSION_FAILURE,
+    links.Ticket.Termination.LOCAL_FAILURE: base.Outcome.REMOTE_FAILURE,
+}
+
+
+class ReceptionManager(_interfaces.ReceptionManager):
+  """A ReceptionManager based around a _Receiver passed to it."""
+
+  def __init__(
+      self, termination_manager, transmission_manager, expiration_manager,
+      ingestion_manager):
+    """Constructor.
+
+    Args:
+      termination_manager: The operation's _interfaces.TerminationManager.
+      transmission_manager: The operation's _interfaces.TransmissionManager.
+      expiration_manager: The operation's _interfaces.ExpirationManager.
+      ingestion_manager: The operation's _interfaces.IngestionManager.
+    """
+    self._termination_manager = termination_manager
+    self._transmission_manager = transmission_manager
+    self._expiration_manager = expiration_manager
+    self._ingestion_manager = ingestion_manager
+
+    self._lowest_unseen_sequence_number = 0
+    self._out_of_sequence_tickets = {}
+    self._aborted = False
+
+  def _abort(self, outcome):
+    self._aborted = True
+    self._termination_manager.abort(outcome)
+    self._transmission_manager.abort(outcome)
+    self._expiration_manager.terminate()
+
+  def _sequence_failure(self, ticket):
+    """Determines a just-arrived ticket's sequential legitimacy.
+
+    Args:
+      ticket: A just-arrived ticket.
+
+    Returns:
+      True if the ticket is sequentially legitimate; False otherwise.
+    """
+    if ticket.sequence_number < self._lowest_unseen_sequence_number:
+      return True
+    elif ticket.sequence_number in self._out_of_sequence_tickets:
+      return True
+    else:
+      return False
+
+  def _process_one(self, ticket):
+    if ticket.sequence_number == 0:
+      self._ingestion_manager.set_group_and_method(ticket.group, ticket.method)
+    if ticket.timeout is not None:
+      self._expiration_manager.change_timeout(ticket.timeout)
+    if ticket.termination is None:
+      completion = None
+    else:
+      completion = utilities.completion(
+          ticket.terminal_metadata, ticket.code, ticket.message)
+    self._ingestion_manager.advance(
+        ticket.initial_metadata, ticket.payload, completion, ticket.allowance)
+    if ticket.allowance is not None:
+      self._transmission_manager.allowance(ticket.allowance)
+
+  def _process(self, ticket):
+    """Process those tickets ready to be processed.
+
+    Args:
+      ticket: A just-arrived ticket the sequence number of which matches this
+        _ReceptionManager's _lowest_unseen_sequence_number field.
+    """
+    while True:
+      self._process_one(ticket)
+      next_ticket = self._out_of_sequence_tickets.pop(
+          ticket.sequence_number + 1, None)
+      if next_ticket is None:
+        self._lowest_unseen_sequence_number = ticket.sequence_number + 1
+        return
+      else:
+        ticket = next_ticket
+
+  def receive_ticket(self, ticket):
+    """See _interfaces.ReceptionManager.receive_ticket for specification."""
+    if self._aborted:
+      return
+    elif self._sequence_failure(ticket):
+      self._abort(base.Outcome.RECEPTION_FAILURE)
+    elif ticket.termination not in (None, links.Ticket.Termination.COMPLETION):
+      outcome = _REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME[ticket.termination]
+      self._abort(outcome)
+    elif ticket.sequence_number == self._lowest_unseen_sequence_number:
+      self._process(ticket)
+    else:
+      self._out_of_sequence_tickets[ticket.sequence_number] = ticket
diff --git a/src/python/grpcio/grpc/framework/core/_termination.py b/src/python/grpcio/grpc/framework/core/_termination.py
new file mode 100644
index 0000000000..ad9f6123d8
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_termination.py
@@ -0,0 +1,212 @@
+# 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.
+
+"""State and behavior for operation termination."""
+
+import abc
+
+from grpc.framework.core import _constants
+from grpc.framework.core import _interfaces
+from grpc.framework.foundation import callable_util
+from grpc.framework.interfaces.base import base
+
+
+def _invocation_completion_predicate(
+    unused_emission_complete, unused_transmission_complete,
+    unused_reception_complete, ingestion_complete):
+  return ingestion_complete
+
+
+def _service_completion_predicate(
+    unused_emission_complete, transmission_complete, unused_reception_complete,
+    unused_ingestion_complete):
+  return transmission_complete
+
+
+class TerminationManager(_interfaces.TerminationManager):
+  """A _interfaces.TransmissionManager on which another manager may be set."""
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def set_expiration_manager(self, expiration_manager):
+    """Sets the expiration manager with which this manager will interact.
+
+    Args:
+      expiration_manager: The _interfaces.ExpirationManager associated with the
+        current operation.
+    """
+    raise NotImplementedError()
+
+
+class _TerminationManager(TerminationManager):
+  """An implementation of TerminationManager."""
+
+  def __init__(self, predicate, action, pool):
+    """Constructor.
+
+    Args:
+      predicate: One of _invocation_completion_predicate or
+        _service_completion_predicate to be used to determine when the operation
+        has completed.
+      action: A behavior to pass the operation outcome on operation termination.
+      pool: A thread pool.
+    """
+    self._predicate = predicate
+    self._action = action
+    self._pool = pool
+    self._expiration_manager = None
+
+    self.outcome = None
+    self._callbacks = []
+
+    self._emission_complete = False
+    self._transmission_complete = False
+    self._reception_complete = False
+    self._ingestion_complete = False
+
+  def set_expiration_manager(self, expiration_manager):
+    self._expiration_manager = expiration_manager
+
+  def _terminate_internal_only(self, outcome):
+    """Terminates the operation.
+
+    Args:
+      outcome: A base.Outcome describing the outcome of the operation.
+    """
+    self.outcome = outcome
+    callbacks = list(self._callbacks)
+    self._callbacks = None
+
+    act = callable_util.with_exceptions_logged(
+        self._action, _constants.INTERNAL_ERROR_LOG_MESSAGE)
+
+    if outcome is base.Outcome.LOCAL_FAILURE:
+      self._pool.submit(act, outcome)
+    else:
+      def call_callbacks_and_act(callbacks, outcome):
+        for callback in callbacks:
+          callback_outcome = callable_util.call_logging_exceptions(
+              callback, _constants.TERMINATION_CALLBACK_EXCEPTION_LOG_MESSAGE,
+              outcome)
+          if callback_outcome.exception is not None:
+            outcome = base.Outcome.LOCAL_FAILURE
+            break
+        act(outcome)
+
+      self._pool.submit(
+          callable_util.with_exceptions_logged(
+              call_callbacks_and_act, _constants.INTERNAL_ERROR_LOG_MESSAGE),
+          callbacks, outcome)
+
+  def _terminate_and_notify(self, outcome):
+    self._terminate_internal_only(outcome)
+    self._expiration_manager.terminate()
+
+  def _perhaps_complete(self):
+    if self._predicate(
+        self._emission_complete, self._transmission_complete,
+        self._reception_complete, self._ingestion_complete):
+      self._terminate_and_notify(base.Outcome.COMPLETED)
+      return True
+    else:
+      return False
+
+  def is_active(self):
+    """See _interfaces.TerminationManager.is_active for specification."""
+    return self.outcome is None
+
+  def add_callback(self, callback):
+    """See _interfaces.TerminationManager.add_callback for specification."""
+    if self.outcome is None:
+      self._callbacks.append(callback)
+      return None
+    else:
+      return self.outcome
+
+  def emission_complete(self):
+    """See superclass method for specification."""
+    if self.outcome is None:
+      self._emission_complete = True
+      self._perhaps_complete()
+
+  def transmission_complete(self):
+    """See superclass method for specification."""
+    if self.outcome is None:
+      self._transmission_complete = True
+      return self._perhaps_complete()
+    else:
+      return False
+
+  def reception_complete(self):
+    """See superclass method for specification."""
+    if self.outcome is None:
+      self._reception_complete = True
+      self._perhaps_complete()
+
+  def ingestion_complete(self):
+    """See superclass method for specification."""
+    if self.outcome is None:
+      self._ingestion_complete = True
+      self._perhaps_complete()
+
+  def expire(self):
+    """See _interfaces.TerminationManager.expire for specification."""
+    self._terminate_internal_only(base.Outcome.EXPIRED)
+
+  def abort(self, outcome):
+    """See _interfaces.TerminationManager.abort for specification."""
+    self._terminate_and_notify(outcome)
+
+
+def invocation_termination_manager(action, pool):
+  """Creates a TerminationManager appropriate for invocation-side use.
+
+  Args:
+    action: An action to call on operation termination.
+    pool: A thread pool in which to execute the passed action and any
+      termination callbacks that are registered during the operation.
+
+  Returns:
+    A TerminationManager appropriate for invocation-side use.
+  """
+  return _TerminationManager(_invocation_completion_predicate, action, pool)
+
+
+def service_termination_manager(action, pool):
+  """Creates a TerminationManager appropriate for service-side use.
+
+  Args:
+    action: An action to call on operation termination.
+    pool: A thread pool in which to execute the passed action and any
+      termination callbacks that are registered during the operation.
+
+  Returns:
+    A TerminationManager appropriate for service-side use.
+  """
+  return _TerminationManager(_service_completion_predicate, action, pool)
diff --git a/src/python/grpcio/grpc/framework/core/_transmission.py b/src/python/grpcio/grpc/framework/core/_transmission.py
new file mode 100644
index 0000000000..01894d398d
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_transmission.py
@@ -0,0 +1,294 @@
+# 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.
+
+"""State and behavior for ticket transmission during an operation."""
+
+from grpc.framework.core import _constants
+from grpc.framework.core import _interfaces
+from grpc.framework.foundation import callable_util
+from grpc.framework.interfaces.base import base
+from grpc.framework.interfaces.links import links
+
+_TRANSMISSION_EXCEPTION_LOG_MESSAGE = 'Exception during transmission!'
+
+
+def _explode_completion(completion):
+  if completion is None:
+    return None, None, None, None
+  else:
+    return (
+        completion.terminal_metadata, completion.code, completion.message,
+        links.Ticket.Termination.COMPLETION)
+
+
+class TransmissionManager(_interfaces.TransmissionManager):
+  """An _interfaces.TransmissionManager that sends links.Tickets."""
+
+  def __init__(
+      self, operation_id, ticket_sink, lock, pool, termination_manager):
+    """Constructor.
+
+    Args:
+      operation_id: The operation's ID.
+      ticket_sink: A callable that accepts tickets and sends them to the other
+        side of the operation.
+      lock: The operation-servicing-wide lock object.
+      pool: A thread pool in which the work of transmitting tickets will be
+        performed.
+      termination_manager: The _interfaces.TerminationManager associated with
+        this operation.
+    """
+    self._lock = lock
+    self._pool = pool
+    self._ticket_sink = ticket_sink
+    self._operation_id = operation_id
+    self._termination_manager = termination_manager
+    self._expiration_manager = None
+
+    self._lowest_unused_sequence_number = 0
+    self._remote_allowance = 1
+    self._remote_complete = False
+    self._timeout = None
+    self._local_allowance = 0
+    self._initial_metadata = None
+    self._payloads = []
+    self._completion = None
+    self._aborted = False
+    self._abortion_outcome = None
+    self._transmitting = False
+
+  def set_expiration_manager(self, expiration_manager):
+    """Sets the ExpirationManager with which this manager will cooperate."""
+    self._expiration_manager = expiration_manager
+
+  def _next_ticket(self):
+    """Creates the next ticket to be transmitted.
+
+    Returns:
+      A links.Ticket to be sent to the other side of the operation or None if
+        there is nothing to be sent at this time.
+    """
+    if self._aborted:
+      if self._abortion_outcome is None:
+        return None
+      else:
+        termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
+            self._abortion_outcome]
+        if termination is None:
+          return None
+        else:
+          self._abortion_outcome = None
+          return links.Ticket(
+              self._operation_id, self._lowest_unused_sequence_number, None,
+              None, None, None, None, None, None, None, None, None,
+              termination)
+
+    action = False
+    # TODO(nathaniel): Support other subscriptions.
+    local_subscription = links.Ticket.Subscription.FULL
+    timeout = self._timeout
+    if timeout is not None:
+      self._timeout = None
+      action = True
+    if self._local_allowance <= 0:
+      allowance = None
+    else:
+      allowance = self._local_allowance
+      self._local_allowance = 0
+      action = True
+    initial_metadata = self._initial_metadata
+    if initial_metadata is not None:
+      self._initial_metadata = None
+      action = True
+    if not self._payloads or self._remote_allowance <= 0:
+      payload = None
+    else:
+      payload = self._payloads.pop(0)
+      self._remote_allowance -= 1
+      action = True
+    if self._completion is None or self._payloads:
+      terminal_metadata, code, message, termination = None, None, None, None
+    else:
+      terminal_metadata, code, message, termination = _explode_completion(
+          self._completion)
+      self._completion = None
+      action = True
+
+    if action:
+      ticket = links.Ticket(
+          self._operation_id, self._lowest_unused_sequence_number, None, None,
+          local_subscription, timeout, allowance, initial_metadata, payload,
+          terminal_metadata, code, message, termination)
+      self._lowest_unused_sequence_number += 1
+      return ticket
+    else:
+      return None
+
+  def _transmit(self, ticket):
+    """Commences the transmission loop sending tickets.
+
+    Args:
+      ticket: A links.Ticket to be sent to the other side of the operation.
+    """
+    def transmit(ticket):
+      while True:
+        transmission_outcome = callable_util.call_logging_exceptions(
+            self._ticket_sink, _TRANSMISSION_EXCEPTION_LOG_MESSAGE, ticket)
+        if transmission_outcome.exception is None:
+          with self._lock:
+            if ticket.termination is links.Ticket.Termination.COMPLETION:
+              self._termination_manager.transmission_complete()
+            ticket = self._next_ticket()
+            if ticket is None:
+              self._transmitting = False
+              return
+        else:
+          with self._lock:
+            if self._termination_manager.outcome is None:
+              self._termination_manager.abort(base.Outcome.TRANSMISSION_FAILURE)
+              self._expiration_manager.terminate()
+            return
+
+    self._pool.submit(callable_util.with_exceptions_logged(
+        transmit, _constants.INTERNAL_ERROR_LOG_MESSAGE), ticket)
+    self._transmitting = True
+
+  def kick_off(
+      self, group, method, timeout, initial_metadata, payload, completion,
+      allowance):
+    """See _interfaces.TransmissionManager.kickoff for specification."""
+    # TODO(nathaniel): Support other subscriptions.
+    subscription = links.Ticket.Subscription.FULL
+    terminal_metadata, code, message, termination = _explode_completion(
+        completion)
+    self._remote_allowance = 1 if payload is None else 0
+    ticket = links.Ticket(
+        self._operation_id, 0, group, method, subscription, timeout, allowance,
+        initial_metadata, payload, terminal_metadata, code, message,
+        termination)
+    self._lowest_unused_sequence_number = 1
+    self._transmit(ticket)
+
+  def advance(self, initial_metadata, payload, completion, allowance):
+    """See _interfaces.TransmissionManager.advance for specification."""
+    effective_initial_metadata = initial_metadata
+    effective_payload = payload
+    effective_completion = completion
+    if allowance is not None and not self._remote_complete:
+      effective_allowance = allowance
+    else:
+      effective_allowance = None
+    if self._transmitting:
+      if effective_initial_metadata is not None:
+        self._initial_metadata = effective_initial_metadata
+      if effective_payload is not None:
+        self._payloads.append(effective_payload)
+      if effective_completion is not None:
+        self._completion = effective_completion
+      if effective_allowance is not None:
+        self._local_allowance += effective_allowance
+    else:
+      if effective_payload is not None:
+        if 0 < self._remote_allowance:
+          ticket_payload = effective_payload
+          self._remote_allowance -= 1
+        else:
+          self._payloads.append(effective_payload)
+          ticket_payload = None
+      else:
+        ticket_payload = None
+      if effective_completion is not None and not self._payloads:
+        ticket_completion = effective_completion
+      else:
+        self._completion = effective_completion
+        ticket_completion = None
+      if any(
+          (effective_initial_metadata, ticket_payload, ticket_completion,
+           effective_allowance)):
+        terminal_metadata, code, message, termination = _explode_completion(
+            completion)
+        ticket = links.Ticket(
+            self._operation_id, self._lowest_unused_sequence_number, None, None,
+            None, None, allowance, effective_initial_metadata, ticket_payload,
+            terminal_metadata, code, message, termination)
+        self._lowest_unused_sequence_number += 1
+        self._transmit(ticket)
+
+  def timeout(self, timeout):
+    """See _interfaces.TransmissionManager.timeout for specification."""
+    if self._transmitting:
+      self._timeout = timeout
+    else:
+      ticket = links.Ticket(
+          self._operation_id, self._lowest_unused_sequence_number, None, None,
+          None, timeout, None, None, None, None, None, None, None)
+      self._lowest_unused_sequence_number += 1
+      self._transmit(ticket)
+
+  def allowance(self, allowance):
+    """See _interfaces.TransmissionManager.allowance for specification."""
+    if self._transmitting or not self._payloads:
+      self._remote_allowance += allowance
+    else:
+      self._remote_allowance += allowance - 1
+      payload = self._payloads.pop(0)
+      if self._payloads:
+        completion = None
+      else:
+        completion = self._completion
+        self._completion = None
+      terminal_metadata, code, message, termination = _explode_completion(
+          completion)
+      ticket = links.Ticket(
+          self._operation_id, self._lowest_unused_sequence_number, None, None,
+          None, None, None, None, payload, terminal_metadata, code, message,
+          termination)
+      self._lowest_unused_sequence_number += 1
+      self._transmit(ticket)
+
+  def remote_complete(self):
+    """See _interfaces.TransmissionManager.remote_complete for specification."""
+    self._remote_complete = True
+    self._local_allowance = 0
+
+  def abort(self, outcome):
+    """See _interfaces.TransmissionManager.abort for specification."""
+    if self._transmitting:
+      self._aborted, self._abortion_outcome = True, outcome
+    else:
+      self._aborted = True
+      if outcome is not None:
+        termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
+            outcome]
+        if termination is not None:
+          ticket = links.Ticket(
+              self._operation_id, self._lowest_unused_sequence_number, None,
+              None, None, None, None, None, None, None, None, None,
+              termination)
+          self._transmit(ticket)
diff --git a/src/python/grpcio/grpc/framework/core/_utilities.py b/src/python/grpcio/grpc/framework/core/_utilities.py
new file mode 100644
index 0000000000..5b0d798751
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/_utilities.py
@@ -0,0 +1,46 @@
+# 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.
+
+"""Package-internal utilities."""
+
+import collections
+
+
+class ServicerPackage(
+    collections.namedtuple(
+        'ServicerPackage', ('servicer', 'default_timeout', 'maximum_timeout'))):
+  """A trivial bundle class.
+
+  Attributes:
+    servicer: A base.Servicer.
+    default_timeout: A float indicating the length of time in seconds to allow
+      for an operation invoked without a timeout.
+    maximum_timeout: A float indicating the maximum length of time in seconds to
+      allow for an operation.
+  """
diff --git a/src/python/grpcio/grpc/framework/core/implementations.py b/src/python/grpcio/grpc/framework/core/implementations.py
new file mode 100644
index 0000000000..364a7faed4
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/core/implementations.py
@@ -0,0 +1,62 @@
+# 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.
+
+"""Entry points into the ticket-exchange-based base layer implementation."""
+
+# base and links are referenced from specification in this module.
+from grpc.framework.core import _end
+from grpc.framework.interfaces.base import base  # pylint: disable=unused-import
+from grpc.framework.interfaces.links import links  # pylint: disable=unused-import
+
+
+def invocation_end_link():
+  """Creates a base.End-links.Link suitable for operation invocation.
+
+  Returns:
+    An object that is both a base.End and a links.Link, that supports operation
+      invocation, and that translates operation invocation into ticket exchange.
+  """
+  return _end.serviceless_end_link()
+
+
+def service_end_link(servicer, default_timeout, maximum_timeout):
+  """Creates a base.End-links.Link suitable for operation service.
+
+  Args:
+    servicer: A base.Servicer for servicing operations.
+    default_timeout: A length of time in seconds to be used as the default
+      time alloted for a single operation.
+    maximum_timeout: A length of time in seconds to be used as the maximum
+      time alloted for a single operation.
+
+  Returns:
+    An object that is both a base.End and a links.Link and that services
+      operations that arrive at it through ticket exchange.
+  """
+  return _end.serviceful_end_link(servicer, default_timeout, maximum_timeout)
diff --git a/src/python/grpcio/grpc/framework/interfaces/base/base.py b/src/python/grpcio/grpc/framework/interfaces/base/base.py
index 9d1651daac..76e0a5bdae 100644
--- a/src/python/grpcio/grpc/framework/interfaces/base/base.py
+++ b/src/python/grpcio/grpc/framework/interfaces/base/base.py
@@ -27,10 +27,20 @@
 # (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 base interface of RPC Framework."""
+"""The base interface of RPC Framework.
 
+Implementations of this interface support the conduct of "operations":
+exchanges between two distinct ends of an arbitrary number of data payloads
+and metadata such as a name for the operation, initial and terminal metadata
+in each direction, and flow control. These operations may be used for transfers
+of data, remote procedure calls, status indication, or anything else
+applications choose.
+"""
+
+# threading is referenced from specification in this module.
 import abc
 import enum
+import threading
 
 # abandonment is referenced from specification in this module.
 from grpc.framework.foundation import abandonment  # pylint: disable=unused-import
@@ -208,19 +218,26 @@ class End(object):
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def stop_gracefully(self):
-    """Gracefully stops this object's service of operations.
+  def stop(self, grace):
+    """Stops this object's service of operations.
 
-    Operations in progress will be allowed to complete, and this method blocks
-    until all of them have.
-    """
-    raise NotImplementedError()
+    This object will refuse service of new operations as soon as this method is
+    called but operations under way at the time of the call may be given a
+    grace period during which they are allowed to finish.
 
-  @abc.abstractmethod
-  def stop_immediately(self):
-    """Immediately stops this object's service of operations.
+    Args:
+      grace: A duration of time in seconds to allow ongoing operations to
+        terminate before being forcefully terminated by the stopping of this
+        End. May be zero to terminate all ongoing operations and immediately
+        stop.
 
-    Operations in progress will not be allowed to complete.
+    Returns:
+      A threading.Event that will be set to indicate all operations having
+        terminated and this End having completely stopped. The returned event
+        may not be set until after the full grace period (if some ongoing
+        operation continues for the full length of the period) or it may be set
+        much sooner (if for example this End had no operations in progress at
+        the time its stop method was called).
     """
     raise NotImplementedError()
 
diff --git a/src/python/grpcio/grpc/framework/interfaces/links/links.py b/src/python/grpcio/grpc/framework/interfaces/links/links.py
index 5ebbac8a6f..069ff024dd 100644
--- a/src/python/grpcio/grpc/framework/interfaces/links/links.py
+++ b/src/python/grpcio/grpc/framework/interfaces/links/links.py
@@ -98,7 +98,7 @@ class Ticket(
     COMPLETION = 'completion'
     CANCELLATION = 'cancellation'
     EXPIRATION = 'expiration'
-    LOCAL_SHUTDOWN = 'local shutdown'
+    SHUTDOWN = 'shutdown'
     RECEPTION_FAILURE = 'reception failure'
     TRANSMISSION_FAILURE = 'transmission failure'
     LOCAL_FAILURE = 'local failure'
diff --git a/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py
new file mode 100644
index 0000000000..72b1ae5642
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py
@@ -0,0 +1,165 @@
+# 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.
+
+"""Tests the RPC Framework Core's implementation of the Base interface."""
+
+import collections
+import logging
+import random
+import time
+import unittest
+
+from grpc._adapter import _intermediary_low
+from grpc._links import invocation
+from grpc._links import service
+from grpc.framework.core import implementations
+from grpc.framework.interfaces.base import utilities
+from grpc_test import test_common as grpc_test_common
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.base import test_cases
+from grpc_test.framework.interfaces.base import test_interfaces
+
+_INVOCATION_INITIAL_METADATA = ((b'0', b'abc'), (b'1', b'def'), (b'2', b'ghi'),)
+_SERVICE_INITIAL_METADATA = ((b'3', b'jkl'), (b'4', b'mno'), (b'5', b'pqr'),)
+_SERVICE_TERMINAL_METADATA = ((b'6', b'stu'), (b'7', b'vwx'), (b'8', b'yza'),)
+_CODE = _intermediary_low.Code.OK
+_MESSAGE = b'test message'
+
+
+class _SerializationBehaviors(
+    collections.namedtuple(
+        '_SerializationBehaviors',
+        ('request_serializers', 'request_deserializers', 'response_serializers',
+         'response_deserializers',))):
+  pass
+
+
+class _Links(
+    collections.namedtuple(
+        '_Links',
+        ('invocation_end_link', 'invocation_grpc_link', 'service_grpc_link',
+         'service_end_link'))):
+  pass
+
+
+def _serialization_behaviors_from_serializations(serializations):
+  request_serializers = {}
+  request_deserializers = {}
+  response_serializers = {}
+  response_deserializers = {}
+  for (group, method), serialization in serializations.iteritems():
+    request_serializers[group, method] = serialization.serialize_request
+    request_deserializers[group, method] = serialization.deserialize_request
+    response_serializers[group, method] = serialization.serialize_response
+    response_deserializers[group, method] = serialization.deserialize_response
+  return _SerializationBehaviors(
+      request_serializers, request_deserializers, response_serializers,
+      response_deserializers)
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def instantiate(self, serializations, servicer):
+    serialization_behaviors = _serialization_behaviors_from_serializations(
+        serializations)
+    invocation_end_link = implementations.invocation_end_link()
+    service_end_link = implementations.service_end_link(
+        servicer, test_constants.DEFAULT_TIMEOUT,
+        test_constants.MAXIMUM_TIMEOUT)
+    service_grpc_link = service.service_link(
+        serialization_behaviors.request_deserializers,
+        serialization_behaviors.response_serializers)
+    port = service_grpc_link.add_port(0, None)
+    channel = _intermediary_low.Channel('localhost:%d' % port, None)
+    invocation_grpc_link = invocation.invocation_link(
+        channel, b'localhost',
+        serialization_behaviors.request_serializers,
+        serialization_behaviors.response_deserializers)
+
+    invocation_end_link.join_link(invocation_grpc_link)
+    invocation_grpc_link.join_link(invocation_end_link)
+    service_end_link.join_link(service_grpc_link)
+    service_grpc_link.join_link(service_end_link)
+    invocation_grpc_link.start()
+    service_grpc_link.start()
+    return invocation_end_link, service_end_link, (
+        invocation_grpc_link, service_grpc_link)
+
+  def destantiate(self, memo):
+    invocation_grpc_link, service_grpc_link = memo
+    invocation_grpc_link.stop()
+    service_grpc_link.stop_gracefully()
+
+  def invocation_initial_metadata(self):
+    return _INVOCATION_INITIAL_METADATA
+
+  def service_initial_metadata(self):
+    return _SERVICE_INITIAL_METADATA
+
+  def invocation_completion(self):
+    return utilities.completion(None, None, None)
+
+  def service_completion(self):
+    return utilities.completion(_SERVICE_TERMINAL_METADATA, _CODE, _MESSAGE)
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return original_metadata is None or grpc_test_common.metadata_transmitted(
+        original_metadata, transmitted_metadata)
+
+  def completion_transmitted(self, original_completion, transmitted_completion):
+    if (original_completion.terminal_metadata is not None and
+        not grpc_test_common.metadata_transmitted(
+            original_completion.terminal_metadata,
+            transmitted_completion.terminal_metadata)):
+        return False
+    elif original_completion.code is not transmitted_completion.code:
+      return False
+    elif original_completion.message != transmitted_completion.message:
+      return False
+    else:
+      return True
+
+
+def setUpModule():
+  logging.warn('setUpModule!')
+
+
+def tearDownModule():
+  logging.warn('tearDownModule!')
+
+
+def load_tests(loader, tests, pattern):
+  return unittest.TestSuite(
+      tests=tuple(
+          loader.loadTestsFromTestCase(test_case_class)
+          for test_case_class in test_cases.test_cases(_Implementation())))
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/framework/core/__init__.py b/src/python/grpcio_test/grpc_test/framework/core/__init__.py
new file mode 100644
index 0000000000..7086519106
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/core/__init__.py
@@ -0,0 +1,30 @@
+# 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.
+
+
diff --git a/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py b/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py
new file mode 100644
index 0000000000..8d72f131d5
--- /dev/null
+++ b/src/python/grpcio_test/grpc_test/framework/core/_base_interface_test.py
@@ -0,0 +1,96 @@
+# 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.
+
+"""Tests the RPC Framework Core's implementation of the Base interface."""
+
+import logging
+import random
+import time
+import unittest
+
+from grpc.framework.core import implementations
+from grpc.framework.interfaces.base import utilities
+from grpc_test.framework.common import test_constants
+from grpc_test.framework.interfaces.base import test_cases
+from grpc_test.framework.interfaces.base import test_interfaces
+
+
+class _Implementation(test_interfaces.Implementation):
+
+  def __init__(self):
+    self._invocation_initial_metadata = object()
+    self._service_initial_metadata = object()
+    self._invocation_terminal_metadata = object()
+    self._service_terminal_metadata = object()
+
+  def instantiate(self, serializations, servicer):
+    invocation = implementations.invocation_end_link()
+    service = implementations.service_end_link(
+        servicer, test_constants.DEFAULT_TIMEOUT,
+        test_constants.MAXIMUM_TIMEOUT)
+    invocation.join_link(service)
+    service.join_link(invocation)
+    return invocation, service, None
+
+  def destantiate(self, memo):
+    pass
+
+  def invocation_initial_metadata(self):
+    return self._invocation_initial_metadata
+
+  def service_initial_metadata(self):
+    return self._service_initial_metadata
+
+  def invocation_completion(self):
+    return utilities.completion(self._invocation_terminal_metadata, None, None)
+
+  def service_completion(self):
+    return utilities.completion(self._service_terminal_metadata, None, None)
+
+  def metadata_transmitted(self, original_metadata, transmitted_metadata):
+    return transmitted_metadata is original_metadata
+
+  def completion_transmitted(self, original_completion, transmitted_completion):
+    return (
+        (original_completion.terminal_metadata is
+         transmitted_completion.terminal_metadata) and
+        original_completion.code is transmitted_completion.code and
+        original_completion.message is transmitted_completion.message
+    )
+
+
+def load_tests(loader, tests, pattern):
+  return unittest.TestSuite(
+      tests=tuple(
+          loader.loadTestsFromTestCase(test_case_class)
+          for test_case_class in test_cases.test_cases(_Implementation())))
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
index dd332fe5dd..5c8b176da4 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
@@ -211,8 +211,10 @@ class _OperationTest(unittest.TestCase):
       elif instruction.kind is _control.Instruction.Kind.CONCLUDE:
         break
 
-    invocation_end.stop_gracefully()
-    service_end.stop_gracefully()
+    invocation_stop_event = invocation_end.stop(0)
+    service_stop_event = service_end.stop(0)
+    invocation_stop_event.wait()
+    service_stop_event.wait()
     invocation_stats = invocation_end.operation_stats()
     service_stats = service_end.operation_stats()
 
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
index 6c2e3346aa..a2bd7107c1 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/links/test_utilities.py
@@ -29,9 +29,42 @@
 
 """State and behavior appropriate for use in tests."""
 
+import logging
 import threading
+import time
 
 from grpc.framework.interfaces.links import links
+from grpc.framework.interfaces.links import utilities
+
+# A more-or-less arbitrary limit on the length of raw data values to be logged.
+_UNCOMFORTABLY_LONG = 48
+
+
+def _safe_for_log_ticket(ticket):
+  """Creates a safe-for-printing-to-the-log ticket for a given ticket.
+
+  Args:
+    ticket: Any links.Ticket.
+
+  Returns:
+    A links.Ticket that is as much as can be equal to the given ticket but
+      possibly features values like the string "<payload of length 972321>" in
+      place of the actual values of the given ticket.
+  """
+  if isinstance(ticket.payload, (basestring,)):
+    payload_length = len(ticket.payload)
+  else:
+    payload_length = -1
+  if payload_length < _UNCOMFORTABLY_LONG:
+    return ticket
+  else:
+    return links.Ticket(
+        ticket.operation_id, ticket.sequence_number,
+        ticket.group, ticket.method, ticket.subscription, ticket.timeout,
+        ticket.allowance, ticket.initial_metadata,
+        '<payload of length {}>'.format(payload_length),
+        ticket.terminal_metadata, ticket.code, ticket.message,
+        ticket.termination)
 
 
 class RecordingLink(links.Link):
@@ -64,3 +97,71 @@ class RecordingLink(links.Link):
     """Returns a copy of the list of all tickets received by this Link."""
     with self._condition:
       return tuple(self._tickets)
+
+
+class _Pipe(object):
+  """A conduit that logs all tickets passed through it."""
+
+  def __init__(self, name):
+    self._lock = threading.Lock()
+    self._name = name
+    self._left_mate = utilities.NULL_LINK
+    self._right_mate = utilities.NULL_LINK
+
+  def accept_left_to_right_ticket(self, ticket):
+    with self._lock:
+      logging.warning(
+          '%s: moving left to right through %s: %s', time.time(), self._name,
+          _safe_for_log_ticket(ticket))
+      try:
+        self._right_mate.accept_ticket(ticket)
+      except Exception as e:  # pylint: disable=broad-except
+        logging.exception(e)
+
+  def accept_right_to_left_ticket(self, ticket):
+    with self._lock:
+      logging.warning(
+          '%s: moving right to left through %s: %s', time.time(), self._name,
+          _safe_for_log_ticket(ticket))
+      try:
+        self._left_mate.accept_ticket(ticket)
+      except Exception as e:  # pylint: disable=broad-except
+        logging.exception(e)
+
+  def join_left_mate(self, left_mate):
+    with self._lock:
+      self._left_mate = utilities.NULL_LINK if left_mate is None else left_mate
+
+  def join_right_mate(self, right_mate):
+    with self._lock:
+      self._right_mate = (
+          utilities.NULL_LINK if right_mate is None else right_mate)
+
+
+class _Facade(links.Link):
+
+  def __init__(self, accept, join):
+    self._accept = accept
+    self._join = join
+
+  def accept_ticket(self, ticket):
+    self._accept(ticket)
+
+  def join_link(self, link):
+    self._join(link)
+
+
+def logging_links(name):
+  """Creates a conduit that logs all tickets passed through it.
+
+  Args:
+    name: A name to use for the conduit to identify itself in logging output.
+
+  Returns:
+    Two links.Links, the first of which is the "left" side of the conduit
+      and the second of which is the "right" side of the conduit.
+  """
+  pipe = _Pipe(name)
+  left_facade = _Facade(pipe.accept_left_to_right_ticket, pipe.join_left_mate)
+  right_facade = _Facade(pipe.accept_right_to_left_ticket, pipe.join_right_mate)
+  return left_facade, right_facade