@rules as coroutines (#5580)

### Problem

The "complex" `Selector` types used to select `@rule` inputs for subjects other than the current subject of an `@rule` have proven to be entirely too complicated. Additionally, their lack of flexibility meant that we needed to introduce many intermediate types to provide links between `@rules` in cases where imperative logic was needed for filtering or conditionals.

The only remaining motivating factor for these complex selectors was that they allow for closed-world compilation of the `@rule` graph, which avoids the need for error checking in rule bodies and provides for easier composition of sets of `@rules`.

### Solution

After experimenting with using coroutines (ie: functions containing `yield` statements that receive a result on their left hand side), it became clear both that:

1. Coroutines were universally easier to understand than the Selector DSL
2. The dependency requests made by coroutines could still easily be statically checked via a small amount of runtime introspection.

And so, this change removes all usage of `SelectProjection` and most usage of `SelectDependencies` in favor of `@rule` coroutines. It does not yet remove the implementation of those `Selectors` (although doing so will shed a massive amount of complexity from the codebase).

### Result

`@rules` are still statically checked, but are now much easier to write. Performance is unaffected.

src/python/pants/engine/BUILD

@@ -182,6 +182,7 @@ python_library(
   sources=['native.py'],
   dependencies=[
     ':native_engine_shared_library',
+    ':selectors',
     '3rdparty/python:cffi',
     '3rdparty/python:setuptools',
     'src/python/pants/binaries:binary_util',

src/python/pants/engine/build_files.py

@@ -20,10 +20,10 @@
 from pants.engine.mapper import AddressFamily, AddressMap, AddressMapper, ResolveError
 from pants.engine.objects import Locatable, SerializableFactory, Validatable
 from pants.engine.rules import RootRule, SingletonRule, TaskRule, rule
-from pants.engine.selectors import Select, SelectDependencies, SelectProjection
+from pants.engine.selectors import Get, Select, SelectDependencies
 from pants.engine.struct import Struct
 from pants.util.dirutil import fast_relpath_optional
-from pants.util.objects import Collection, datatype
+from pants.util.objects import datatype
 
 
 class ResolvedTypeMismatchError(ResolveError):
@@ -35,47 +35,26 @@ def _key_func(entry):
   return key
 
 
-class BuildDirs(datatype('BuildDirs', ['dependencies'])):
-  """A list of Stat objects for directories containing build files."""
-
-
-class BuildFiles(datatype('BuildFiles', ['files_content'])):
-  """The FileContents of BUILD files in some directory"""
-
-
-class BuildFileGlobs(datatype('BuildFilesGlobs', ['path_globs'])):
-  """A wrapper around PathGlobs that are known to match a build file pattern."""
-
-
-@rule(BuildFiles,
-      [SelectProjection(FilesContent, PathGlobs, 'path_globs', BuildFileGlobs)])
-def build_files(files_content):
-  return BuildFiles(files_content)
-
-
-@rule(BuildFileGlobs, [Select(AddressMapper), Select(Dir)])
-def buildfile_path_globs_for_dir(address_mapper, directory):
-  patterns = tuple(join(directory.path, p) for p in address_mapper.build_patterns)
-  return BuildFileGlobs(PathGlobs.create('',
-                                         include=patterns,
-                                         exclude=address_mapper.build_ignore_patterns))
-
-
-@rule(AddressFamily, [Select(AddressMapper), Select(Dir), Select(BuildFiles)])
-def parse_address_family(address_mapper, path, build_files):
-  """Given the contents of the build files in one directory, return an AddressFamily.
+@rule(AddressFamily, [Select(AddressMapper), Select(Dir)])
+def parse_address_family(address_mapper, directory):
+  """Given an AddressMapper and a directory, return an AddressFamily.
 
   The AddressFamily may be empty, but it will not be None.
   """
-  files_content = build_files.files_content.dependencies
+  patterns = tuple(join(directory.path, p) for p in address_mapper.build_patterns)
+  path_globs = PathGlobs.create('',
+                                include=patterns,
+                                exclude=address_mapper.build_ignore_patterns)
+  files_content = yield Get(FilesContent, PathGlobs, path_globs)
+
   if not files_content:
-    raise ResolveError('Directory "{}" does not contain build files.'.format(path))
+    raise ResolveError('Directory "{}" does not contain build files.'.format(directory.path))
   address_maps = []
-  for filecontent_product in files_content:
+  for filecontent_product in files_content.dependencies:
     address_maps.append(AddressMap.parse(filecontent_product.path,
                                          filecontent_product.content,
                                          address_mapper.parser))
-  return AddressFamily.create(path.path, address_maps)
+  yield AddressFamily.create(directory.path, address_maps)
 
 
 class UnhydratedStruct(datatype('UnhydratedStruct', ['address', 'struct', 'dependencies'])):
@@ -107,17 +86,16 @@ def _raise_did_you_mean(address_family, name):
                      .format(name, address_family.namespace, possibilities))
 
 
-@rule(UnhydratedStruct,
-      [Select(AddressMapper),
-       SelectProjection(AddressFamily, Dir, 'spec_path', Address),
-       Select(Address)])
-def resolve_unhydrated_struct(address_mapper, address_family, address):
-  """Given an Address and its AddressFamily, resolve an UnhydratedStruct.
+@rule(UnhydratedStruct, [Select(AddressMapper), Select(Address)])
+def resolve_unhydrated_struct(address_mapper, address):
+  """Given an AddressMapper and an Address, resolve an UnhydratedStruct.
 
   Recursively collects any embedded addressables within the Struct, but will not walk into a
-  dependencies field, since those are requested explicitly by tasks using SelectDependencies.
+  dependencies field, since those should be requested explicitly by rules.
   """
 
+  address_family = yield Get(AddressFamily, Dir(address.spec_path))
+
   struct = address_family.addressables.get(address)
   addresses = address_family.addressables
   if not struct or address not in addresses:
@@ -148,7 +126,7 @@ def collect_dependencies(item):
 
   collect_dependencies(struct)
 
-  return UnhydratedStruct(
+  yield UnhydratedStruct(
     filter(lambda build_address: build_address == address, addresses)[0], struct, dependencies)
 
 
@@ -222,17 +200,18 @@ def _hydrate(item_type, spec_path, **kwargs):
   return item
 
 
-@rule(BuildFileAddresses,
-      [Select(AddressMapper),
-       SelectDependencies(AddressFamily, BuildDirs, field_types=(Dir,)),
-       Select(Specs)])
-def addresses_from_address_families(address_mapper, address_families, specs):
-  """Given a list of AddressFamilies matching a list of Specs, return matching Addresses.
+@rule(BuildFileAddresses, [Select(AddressMapper), Select(Specs)])
+def addresses_from_address_families(address_mapper, specs):
+  """Given an AddressMapper and list of Specs, return matching BuildFileAddresses.
 
   Raises a AddressLookupError if:
      - there were no matching AddressFamilies, or
      - the Spec matches no addresses for SingleAddresses.
   """
+  # Capture a Snapshot covering all paths for these Specs, then group by directory.
+  snapshot = yield Get(Snapshot, PathGlobs, _spec_to_globs(address_mapper, specs))
+  dirnames = set(dirname(f.stat.path) for f in snapshot.files)
+  address_families = yield [Get(AddressFamily, Dir(d)) for d in dirnames]
 
   # NB: `@memoized` does not work on local functions.
   def by_directory():
@@ -294,20 +273,11 @@ def include(address_families, predicate=None):
     else:
       raise ValueError('Unrecognized Spec type: {}'.format(spec))
 
-  return BuildFileAddresses(addresses)
+  yield BuildFileAddresses(addresses)
 
 
-@rule(BuildDirs, [Select(AddressMapper), Select(Snapshot)])
-def filter_build_dirs(address_mapper, snapshot):
-  """Given a Snapshot matching a build pattern, return parent directories as BuildDirs."""
-  dirnames = set(dirname(f.stat.path) for f in snapshot.files)
-  return BuildDirs(tuple(Dir(d) for d in dirnames))
-
-
-@rule(PathGlobs, [Select(AddressMapper), Select(Specs)])
-def spec_to_globs(address_mapper, specs):
-  """Given a Spec object, return a PathGlobs object for the build files that it matches.
-  """
+def _spec_to_globs(address_mapper, specs):
+  """Given a Specs object, return a PathGlobs object for the build files that it matches."""
   patterns = set()
   for spec in specs.dependencies:
     if type(spec) is DescendantAddresses:
@@ -340,9 +310,6 @@ def _recursive_dirname(f):
   yield ''
 
 
-BuildFilesCollection = Collection.of(BuildFiles)
-
-
 def create_graph_rules(address_mapper, symbol_table):
   """Creates tasks used to parse Structs from BUILD files.
 
@@ -351,9 +318,6 @@ def create_graph_rules(address_mapper, symbol_table):
   """
   symbol_table_constraint = symbol_table.constraint()
   return [
-    TaskRule(BuildFilesCollection,
-             [SelectDependencies(BuildFiles, BuildDirs, field_types=(Dir,))],
-             BuildFilesCollection),
     # A singleton to provide the AddressMapper.
     SingletonRule(AddressMapper, address_mapper),
     # Support for resolving Structs from Addresses.
@@ -367,13 +331,9 @@ def create_graph_rules(address_mapper, symbol_table):
     resolve_unhydrated_struct,
     # BUILD file parsing.
     parse_address_family,
-    build_files,
-    buildfile_path_globs_for_dir,
     # Spec handling: locate directories that contain build files, and request
     # AddressFamilies for each of them.
     addresses_from_address_families,
-    filter_build_dirs,
-    spec_to_globs,
     # Root rules representing parameters that might be provided via root subjects.
     RootRule(Address),
     RootRule(BuildFileAddress),

src/python/pants/engine/legacy/address_mapper.py

@@ -13,7 +13,6 @@
 from pants.build_graph.address_lookup_error import AddressLookupError
 from pants.build_graph.address_mapper import AddressMapper
 from pants.engine.addressable import BuildFileAddresses
-from pants.engine.build_files import BuildFilesCollection
 from pants.engine.mapper import ResolveError
 from pants.engine.nodes import Throw
 from pants.util.dirutil import fast_relpath
@@ -33,14 +32,10 @@ def __init__(self, scheduler, build_root):
     self._build_root = build_root
 
   def scan_build_files(self, base_path):
-    specs = (DescendantAddresses(base_path),)
-    build_files_collection, = self._scheduler.product_request(BuildFilesCollection, [Specs(specs)])
+    build_file_addresses = self._internal_scan_specs([DescendantAddresses(base_path)],
+                                                     missing_is_fatal=False)
 
-    build_files_set = set()
-    for build_files in build_files_collection.dependencies:
-      build_files_set.update(f.path for f in build_files.files_content.dependencies)
-
-    return build_files_set
+    return {bfa.rel_path for bfa in build_file_addresses}
 
   @staticmethod
   def any_is_declaring_file(address, file_paths):

src/python/pants/engine/legacy/graph.py

@@ -23,7 +23,7 @@
 from pants.engine.fs import PathGlobs, Snapshot
 from pants.engine.legacy.structs import BundleAdaptor, BundlesField, SourcesField, TargetAdaptor
 from pants.engine.rules import TaskRule, rule
-from pants.engine.selectors import Select, SelectDependencies, SelectProjection
+from pants.engine.selectors import Get, Select, SelectDependencies
 from pants.source.wrapped_globs import EagerFilesetWithSpec, FilesetRelPathWrapper
 from pants.util.dirutil import fast_relpath
 from pants.util.objects import Collection, datatype
@@ -299,18 +299,19 @@ class HydratedTargets(Collection.of(HydratedTarget)):
   """An intransitive set of HydratedTarget objects."""
 
 
-@rule(TransitiveHydratedTargets, [SelectDependencies(TransitiveHydratedTarget,
-                                                     BuildFileAddresses,
-                                                     field_types=(Address,),
-                                                     field='addresses')])
-def transitive_hydrated_targets(transitive_hydrated_targets):
-  """Kicks off recursion on expansion of TransitiveHydratedTarget objects.
+@rule(TransitiveHydratedTargets, [Select(BuildFileAddresses)])
+def transitive_hydrated_targets(build_file_addresses):
+  """Given BuildFileAddresses, kicks off recursion on expansion of TransitiveHydratedTargets.
 
   The TransitiveHydratedTarget struct represents a structure-shared graph, which we walk
   and flatten here. The engine memoizes the computation of TransitiveHydratedTarget, so
   when multiple TransitiveHydratedTargets objects are being constructed for multiple
   roots, their structure will be shared.
   """
+
+  transitive_hydrated_targets = yield [Get(TransitiveHydratedTarget, Address, a)
+                                       for a in build_file_addresses.addresses]
+
   closure = set()
   to_visit = deque(transitive_hydrated_targets)
 
@@ -321,25 +322,20 @@ def transitive_hydrated_targets(transitive_hydrated_targets):
     closure.add(tht.root)
     to_visit.extend(tht.dependencies)
 
-  return TransitiveHydratedTargets(tuple(tht.root for tht in transitive_hydrated_targets), closure)
+  yield TransitiveHydratedTargets(tuple(tht.root for tht in transitive_hydrated_targets), closure)
 
 
-@rule(TransitiveHydratedTarget, [Select(HydratedTarget),
-                                 SelectDependencies(TransitiveHydratedTarget,
-                                                    HydratedTarget,
-                                                    field_types=(Address,),
-                                                    field='addresses')])
-def transitive_hydrated_target(root, dependencies):
-  return TransitiveHydratedTarget(root, dependencies)
+@rule(TransitiveHydratedTarget, [Select(HydratedTarget)])
+def transitive_hydrated_target(root):
+  dependencies = yield [Get(TransitiveHydratedTarget, Address, d) for d in root.dependencies]
+  yield TransitiveHydratedTarget(root, dependencies)
 
 
-@rule(HydratedTargets, [SelectDependencies(HydratedTarget,
-                                           BuildFileAddresses,
-                                           field_types=(Address,),
-                                           field='addresses')])
-def hydrated_targets(targets):
-  """Requests HydratedTarget instances."""
-  return HydratedTargets(targets)
+@rule(HydratedTargets, [Select(BuildFileAddresses)])
+def hydrated_targets(build_file_addresses):
+  """Requests HydratedTarget instances for BuildFileAddresses."""
+  targets = yield [Get(HydratedTarget, Address, a) for a in build_file_addresses.addresses]
+  yield HydratedTargets(targets)
 
 
 class HydratedField(datatype('HydratedField', ['name', 'value'])):
@@ -372,22 +368,22 @@ def _eager_fileset_with_spec(spec_path, filespec, snapshot, include_dirs=False):
                               files_hash=snapshot.fingerprint)
 
 
-@rule(HydratedField,
-      [Select(SourcesField),
-       SelectProjection(Snapshot, PathGlobs, 'path_globs', SourcesField)])
-def hydrate_sources(sources_field, snapshot):
-  """Given a SourcesField and a Snapshot for its path_globs, create an EagerFilesetWithSpec."""
+@rule(HydratedField, [Select(SourcesField)])
+def hydrate_sources(sources_field):
+  """Given a SourcesField, request a Snapshot for its path_globs and create an EagerFilesetWithSpec."""
+
+  snapshot = yield Get(Snapshot, PathGlobs, sources_field.path_globs)
   fileset_with_spec = _eager_fileset_with_spec(sources_field.address.spec_path,
                                                sources_field.filespecs,
                                                snapshot)
-  return HydratedField(sources_field.arg, fileset_with_spec)
+  yield HydratedField(sources_field.arg, fileset_with_spec)
+
 
+@rule(HydratedField, [Select(BundlesField)])
+def hydrate_bundles(bundles_field):
+  """Given a BundlesField, request Snapshots for each of its filesets and create BundleAdaptors."""
+  snapshot_list = yield [Get(Snapshot, PathGlobs, pg) for pg in bundles_field.path_globs_list]
 
-@rule(HydratedField,
-      [Select(BundlesField),
-       SelectDependencies(Snapshot, BundlesField, 'path_globs_list', field_types=(PathGlobs,))])
-def hydrate_bundles(bundles_field, snapshot_list):
-  """Given a BundlesField and a Snapshot for each of its filesets create a list of BundleAdaptors."""
   bundles = []
   zipped = zip(bundles_field.bundles,
                bundles_field.filespecs_list,
@@ -403,7 +399,7 @@ def hydrate_bundles(bundles_field, snapshot_list):
                                                  snapshot,
                                                  include_dirs=True)
     bundles.append(BundleAdaptor(**kwargs))
-  return HydratedField('bundles', bundles)
+  yield HydratedField('bundles', bundles)
 
 
 def create_legacy_graph_tasks(symbol_table):