Unify train_local and train_with_distributed_readers
Unify train_local and train_with_distributed_readers

Summary:
Outline of changes:

- add single-operator support to Caffe2-Flow integration (based on Alisson's suggestions)
- because of above support we can move graph construction to the main workflow body and pass the job to the Flow operator doing running, similarly to the distributed case
- after that it's easy to unify code even more
- there's some trickery required to make sure model exporting doesn't pollute Cluster info (as TaskGroup.to_task() creates new tasks)

Important: this diff changes train_local behavior by introducing queue between preprocessing and trainer (before we did everything on trainer thread). It doesn't seem to impact perf much (even slightly positive), so I guess it's fine. It also allows for better unification.

I'll follow up with a separate diff that moves max_examples gating to multi_reader (including train_local) and then we can enable checkpointing.

Reviewed By: xianjiec

Differential Revision: D4526079

fbshipit-source-id: 8c44044f45e7738e9b13e5b3acfbb994bc5a3d72

pietern/caffe2,pietern/caffe2,Yangqing/caffe2,bwasti/caffe2,sf-wind/caffe2,xzturn/caffe2,bwasti/caffe2,xzturn/caffe2,davinwang/caffe2,Yangqing/caffe2,pietern/caffe2,Yangqing/caffe2,davinwang/caffe2,davinwang/caffe2,pietern/caffe2,bwasti/caffe2,davinwang/caffe2,davinwang/caffe2,xzturn/caffe2,pietern/caffe2,caffe2/caffe2,sf-wind/caffe2,bwasti/caffe2,xzturn/caffe2,Yangqing/caffe2,sf-wind/caffe2,sf-wind/caffe2,bwasti/caffe2,xzturn/caffe2,sf-wind/caffe2,Yangqing/caffe2