Restore Sources custom types per extension.

This was removed to support the multiprocess engine which relies on
pickling.  The dynamic class objects generated by `Sources.of` were not
pickle compatible.  Restore `Sources.of` memoized Sources subtypes
per-extension by implementing `__reduce__` and exposing the `Sources`
subtypes at the module level.

Testing Done:
REPL testing to both check quickly pickle/unpicle now worked and type
identity was maintained.  This was backed up by the engine test which
pickles and unpickles `Sources` as part of running the multiprocess
engine.

Also re-rendered the simple codegen plan, make more sense now that the
Sources product type carries the extension - see attached and compare
to the image in https://rbcommons.com/s/twitter/r/3010/:
```
$ ./pants run src/python/pants/engine/exp/examples:viz -- \
  tests/python/pants_test/engine/exp/examples/scheduler_inputs \
  compile src/java/codegen/simple
```

CI went green here:
  https://travis-ci.org/pantsbuild/pants/builds/86753888

Bugs closed: 2423

Reviewed at https://rbcommons.com/s/twitter/r/3011/

src/python/pants/engine/exp/examples/BUILD

@@ -13,6 +13,7 @@ python_library(
     'src/python/pants/engine/exp:parsers',
     'src/python/pants/engine/exp:scheduler',
     'src/python/pants/engine/exp:targets',
+    'src/python/pants/util:memo',
   ]
 )
 

src/python/pants/engine/exp/examples/planners.py

@@ -6,6 +6,7 @@
                         unicode_literals, with_statement)
 
 import functools
+import sys
 
 from twitter.common.collections import OrderedSet
 
@@ -17,6 +18,7 @@
 from pants.engine.exp.scheduler import GlobalScheduler, Plan, Planners, Subject, Task, TaskPlanner
 from pants.engine.exp.targets import Sources as AddressableSources
 from pants.engine.exp.targets import Target
+from pants.util.memo import memoized
 
 
 class PrintingTask(Task):
@@ -88,18 +90,31 @@ def execute(self, jars):
     return super(IvyResolve, self).execute(jars=jars)
 
 
+def _create_sources(ext):
+  # A pickle-compatible top-level function for custom unpickling of Sources per-extension types.
+  return Sources.of(ext)
+
+
 class Sources(object):
-  def __init__(self, ext):
-    self.ext = ext
+  @classmethod
+  @memoized
+  def of(cls, ext):
+    type_name = b'Sources({!r})'.format(ext)
 
-  def __hash__(self):
-    return hash(self.ext)
+    class_dict = {'ext': ext,
+                  # We need custom serialization for the dynamic class type.
+                  '__reduce__': lambda self: (_create_sources, ext)}
 
-  def __eq__(self, other):
-    return isinstance(other, Sources) and self.ext == other.ext
+    ext_type = type(type_name, (cls,), class_dict)
 
-  def __ne__(self, other):
-    return not (self == other)
+    # Expose the custom class type at the module level to be pickle compatible.
+    setattr(sys.modules[cls.__module__], type_name, ext_type)
+
+    return ext_type
+
+  @classmethod
+  def ext(cls):
+    raise NotImplementedError()
 
   def __repr__(self):
     return 'Sources(ext={!r})'.format(self.ext)
@@ -129,7 +144,7 @@ class ApacheThriftPlanner(TaskPlanner):
   def __init__(self):
     # This will come via an option default.
     # TODO(John Sirois): once the options system is plumbed, make the languages configurable.
-    self._product_type_by_lang = {'java': Sources('.java'), 'py': Sources('.py')}
+    self._product_type_by_lang = {'java': Sources.of('.java'), 'py': Sources.of('.py')}
 
   @property
   def goal_name(self):
@@ -178,7 +193,7 @@ def execute(self, sources, rev, gen, strict):
 
 class JavacPlanner(TaskPlanner):
   # Product type
-  JavaSources = Sources('.java')
+  JavaSources = Sources.of('.java')
 
   @property
   def goal_name(self):

src/python/pants/engine/exp/examples/visualizer.py

@@ -5,7 +5,6 @@
 from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
                         unicode_literals, with_statement)
 
-import inspect
 import os
 import subprocess
 import sys
@@ -19,18 +18,14 @@
 
 
 def create_digraph(execution_graph):
-  def product_type_name(product_type):
-    return product_type.__name__ if inspect.isclass(product_type) else type(product_type).__name__
-
   def format_subject(subject):
     return subject.primary.address.spec if subject.primary.address else repr(subject.primary)
 
   def format_promise(promise):
-    return '{}({})'.format(product_type_name(promise._product_type),
-                           format_subject(promise.subject))
+    return '{}({})'.format(promise._product_type.__name__, format_subject(promise.subject))
 
   def format_label(product_type, plan):
-    return '{}:{}'.format(plan._task_type.__name__, product_type_name(product_type))
+    return '{}:{}'.format(plan._task_type.__name__, product_type.__name__)
 
   colorscheme = 'set312'
   colors = {}

tests/python/pants_test/engine/exp/test_scheduler.py

@@ -61,7 +61,7 @@ def test_gen(self):
     plans = list(execution_graph.walk())
     self.assertEqual(1, len(plans))
 
-    self.assertEqual((Sources('.java'),
+    self.assertEqual((Sources.of('.java'),
                       Plan(task_type=ApacheThrift,
                            subjects=[self.thrift],
                            strict=True,
@@ -84,7 +84,7 @@ def test_codegen_simple(self):
     jars = [self.guava] + thrift_jars
 
     # Independent leaves 1st
-    self.assertEqual({(Sources('.java'),
+    self.assertEqual({(Sources.of('.java'),
                        Plan(task_type=ApacheThrift,
                             subjects=[self.thrift],
                             strict=True,
@@ -98,7 +98,7 @@ def test_codegen_simple(self):
     self.assertEqual((Classpath,
                       Plan(task_type=JavacTask,
                            subjects=[self.thrift],
-                           sources=Promise(Sources('.java'), self.thrift),
+                           sources=Promise(Sources.of('.java'), self.thrift),
                            classpath=[Promise(Classpath, jar) for jar in thrift_jars])),
                      plans[2])