add executor.prepare

paddle/fluid/framework/executor.cc

@@ -14,17 +14,12 @@ limitations under the License. */
 
 #include "paddle/fluid/framework/executor.h"
 
-#include <set>
-
-#include "gflags/gflags.h"
 #include "paddle/fluid/framework/channel.h"
 #include "paddle/fluid/framework/feed_fetch_method.h"
-#include "paddle/fluid/framework/feed_fetch_type.h"
 #include "paddle/fluid/framework/lod_rank_table.h"
 #include "paddle/fluid/framework/lod_tensor_array.h"
 #include "paddle/fluid/framework/op_registry.h"
 #include "paddle/fluid/framework/reader.h"
-#include "paddle/fluid/platform/place.h"
 
 DECLARE_bool(benchmark);
 DEFINE_bool(check_nan_inf, false,
@@ -34,14 +29,9 @@ DEFINE_bool(check_nan_inf, false,
 namespace paddle {
 namespace framework {
 
-struct ExecutorPrepareContext {
-  ExecutorPrepareContext(const framework::ProgramDesc& prog, size_t block_id)
-      : prog_(prog), block_id_(block_id) {}
-
-  framework::ProgramDesc prog_;
-  size_t block_id_;
-  std::vector<std::unique_ptr<OperatorBase>> ops_;
-};
+ExecutorPrepareContext::ExecutorPrepareContext(
+    const framework::ProgramDesc& prog, size_t block_id)
+    : prog_(prog), block_id_(block_id) {}
 
 Executor::Executor(const platform::Place& place) : place_(place) {}
 
@@ -269,6 +259,10 @@ ExecutorPrepareContext* Executor::Prepare(const ProgramDesc& program,
   return ctx;
 }
 
+void Executor::DeletePreparedContext(ExecutorPrepareContext* ctx) {
+  delete ctx;
+}
+
 void Executor::RunPreparedContext(ExecutorPrepareContext* ctx, Scope* scope,
                                   bool create_local_scope, bool create_vars) {
   auto& block = ctx->prog_.Block(ctx->block_id_);

paddle/fluid/framework/executor.h

@@ -22,7 +22,15 @@ limitations under the License. */
 
 namespace paddle {
 namespace framework {
-struct ExecutorPrepareContext;
+
+struct ExecutorPrepareContext {
+  ExecutorPrepareContext(const framework::ProgramDesc& prog, size_t block_id);
+
+  framework::ProgramDesc prog_;
+  size_t block_id_;
+  std::vector<std::unique_ptr<OperatorBase>> ops_;
+};
+
 class Executor {
  public:
   // TODO(dzhwinter) : Do not rely on this function, it will be removed
@@ -50,6 +58,8 @@ class Executor {
   static ExecutorPrepareContext* Prepare(const ProgramDesc& program,
                                          int block_id);
 
+  static void DeletePreparedContext(ExecutorPrepareContext* ctx);
+
   void RunPreparedContext(ExecutorPrepareContext* ctx, Scope* scope,
                           bool create_local_scope = true,
                           bool create_vars = true);

paddle/fluid/pybind/pybind.cc

@@ -401,6 +401,24 @@ All parameter, weight, gradient are variables in Paddle.
 
   py::class_<framework::Executor>(m, "Executor")
       .def(py::init<const platform::Place &>())
+      .def_static(
+          "prepare",
+          [](const ProgramDesc &pdesc, int block_id) -> void * {
+            return static_cast<void *>(Executor::Prepare(pdesc, block_id));
+          },
+          py::return_value_policy::reference)
+      .def_static("delete_prepared_ctx",
+                  [](void *handle) {
+                    Executor::DeletePreparedContext(
+                        static_cast<ExecutorPrepareContext *>(handle));
+                  })
+      .def("run_prepared_ctx",
+           [](Executor &self, void *handle, Scope *scope,
+              bool create_local_scope, bool create_vars) {
+             self.RunPreparedContext(
+                 static_cast<ExecutorPrepareContext *>(handle), scope,
+                 create_local_scope, create_vars);
+           })
       .def("run",
            (void (Executor::*)(const ProgramDesc &, Scope *, int, bool, bool)) &
                Executor::Run);

python/paddle/fluid/executor.py

@@ -179,6 +179,16 @@ def to_name_str(var):
     return str(feed_var_names + fetch_var_names)
 
 
+class PreparedContext(object):
+    def __init__(self, handle, program, fetch_list, feed_var_name,
+                 fetch_var_name):
+        self.handle = handle
+        self.program = program
+        self.fetch_list = fetch_list
+        self.feed_var_name = feed_var_name
+        self.fetch_var_name = fetch_var_name
+
+
 class Executor(object):
     def __init__(self, places):
         if not isinstance(places, list) and not isinstance(places, tuple):
@@ -235,6 +245,122 @@ def parselod(data):
             tensor.set_lod(lod)
             return tensor
 
+    def _get_program_cache(self, feed, fetch_list):
+        program_cache_key = get_program_cache_key(feed, fetch_list)
+        program_cache = self.program_caches.get(program_cache_key, None)
+        return program_cache
+
+    def _add_program_cache(self, feed, fetch_list, program):
+        program_cache_key = get_program_cache_key(feed, fetch_list)
+        self.program_caches[program_cache_key] = program
+
+    def _add_feed_fetch_ops(self, program, feed, fetch_list, feed_var_name,
+                            fetch_var_name):
+        tmp_program = program.clone()
+
+        global_block = tmp_program.global_block()
+
+        if feed_var_name in global_block.vars:
+            feed_var = global_block.var(feed_var_name)
+        else:
+            feed_var = global_block.create_var(
+                name=feed_var_name,
+                type=core.VarDesc.VarType.FEED_MINIBATCH,
+                persistable=True)
+
+        if fetch_var_name in global_block.vars:
+            fetch_var = global_block.var(fetch_var_name)
+        else:
+            fetch_var = global_block.create_var(
+                name=fetch_var_name,
+                type=core.VarDesc.VarType.FETCH_LIST,
+                persistable=True)
+
+        # prepend feed operators
+        if not has_feed_operators(global_block, feed, feed_var_name):
+            for i, name in enumerate(feed):
+                out = global_block.var(name)
+                global_block.prepend_op(
+                    type='feed',
+                    inputs={'X': [feed_var]},
+                    outputs={'Out': [out]},
+                    attrs={'col': i})
+
+        # append fetch_operators
+        if not has_fetch_operators(global_block, fetch_list, fetch_var_name):
+            for i, var in enumerate(fetch_list):
+                assert isinstance(var, Variable) or isinstance(var, str), (
+                    "Wrong type for fetch_list[%s]: %s" % (i, type(var)))
+                global_block.append_op(
+                    type='fetch',
+                    inputs={'X': [var]},
+                    outputs={'Out': [fetch_var]},
+                    attrs={'col': i})
+
+        return tmp_program
+
+    def feed_data(self, program, feed, feed_var_name, scope):
+        # feed var to framework
+        for op in program.global_block().ops:
+            if op.desc.type() == 'feed':
+                feed_target_name = op.desc.output('Out')[0]
+                cur_feed = feed[feed_target_name]
+                if not isinstance(cur_feed, core.LoDTensor):
+                    cur_feed = self.aslodtensor(cur_feed)
+                idx = op.desc.attr('col')
+                core.set_feed_variable(scope, cur_feed, feed_var_name, idx)
+            else:
+                break
+
+    def fetch_data(self, fetch_list, fetch_var_name, scope):
+        outs = [
+            core.get_fetch_variable(scope, fetch_var_name, i)
+            for i in xrange(len(fetch_list))
+        ]
+        return outs
+
+    def prepare(self,
+                program=None,
+                feed=None,
+                fetch_list=None,
+                feed_var_name='feed',
+                fetch_var_name='fetch'):
+        if feed is None:
+            feed = {}
+        if not isinstance(feed, dict):
+            raise TypeError("feed should be a map")
+        if fetch_list is None:
+            fetch_list = []
+        if program is None:
+            program = default_main_program()
+
+        if not isinstance(program, Program):
+            raise TypeError()
+
+        program = self._add_feed_fetch_ops(
+            program=program,
+            feed=feed,
+            fetch_list=fetch_list,
+            feed_var_name=feed_var_name,
+            fetch_var_name=fetch_var_name)
+        handle = self.executor.prepare(program.desc, 0)
+        return PreparedContext(handle, program, fetch_list, feed_var_name,
+                               fetch_var_name)
+
+    def run_prepared_ctx(self, ctx, feed=None, scope=None, return_numpy=True):
+        if scope is None:
+            scope = global_scope()
+
+        self.feed_data(ctx.program, feed, ctx.feed_var_name, scope)
+        self.executor.run_prepared_ctx(ctx.handle, scope, True, True)
+        outs = self.fetch_data(ctx.fetch_list, ctx.fetch_var_name, scope)
+        if return_numpy:
+            outs = as_numpy(outs)
+        return outs
+
+    def delete_prepared_ctx(self, ctx):
+        self.executor.delete_prepared_ctx(ctx.handle)
+
     def run(self,
             program=None,
             feed=None,
@@ -268,7 +394,6 @@ def run(self,
             raise TypeError("feed should be a map")
         if fetch_list is None:
             fetch_list = []
-
         if program is None:
             program = default_main_program()
 
@@ -278,79 +403,28 @@ def run(self,
         if scope is None:
             scope = global_scope()
 
-        program_cache = None
-        program_cache_key = get_program_cache_key(feed, fetch_list)
-
         if use_program_cache:
-            # find program cache by cache_key
-            program_cache = self.program_caches.get(program_cache_key, None)
-            # TODO(qiao): Should check program_cache and program are exactly the same.
+            cached_program = self._get_program_cache(feed, fetch_list)
+            if cached_program is None:
+                cached_program = self._add_feed_fetch_ops(
+                    program=program,
+                    feed=feed,
+                    fetch_list=fetch_list,
+                    feed_var_name=feed_var_name,
+                    fetch_var_name=fetch_var_name)
+                self._add_program_cache(feed, fetch_list, cached_program)
+            program = cached_program
         else:
-            self.program_caches.pop(program_cache_key, None)
-
-        if program_cache is None:
-            program_cache = program.clone()
-
-            if use_program_cache:
-                self.program_caches[program_cache_key] = program_cache
-
-            global_block = program_cache.global_block()
-
-            if feed_var_name in global_block.vars:
-                feed_var = global_block.var(feed_var_name)
-            else:
-                feed_var = global_block.create_var(
-                    name=feed_var_name,
-                    type=core.VarDesc.VarType.FEED_MINIBATCH,
-                    persistable=True)
-
-            if fetch_var_name in global_block.vars:
-                fetch_var = global_block.var(fetch_var_name)
-            else:
-                fetch_var = global_block.create_var(
-                    name=fetch_var_name,
-                    type=core.VarDesc.VarType.FETCH_LIST,
-                    persistable=True)
-
-            # prepend feed operators
-            if not has_feed_operators(global_block, feed, feed_var_name):
-                for i, name in enumerate(feed):
-                    out = global_block.var(name)
-                    global_block.prepend_op(
-                        type='feed',
-                        inputs={'X': [feed_var]},
-                        outputs={'Out': [out]},
-                        attrs={'col': i})
-
-            # append fetch_operators
-            if not has_fetch_operators(global_block, fetch_list,
-                                       fetch_var_name):
-                for i, var in enumerate(fetch_list):
-                    assert isinstance(var, Variable) or isinstance(var, str), (
-                        "Wrong type for fetch_list[%s]: %s" % (i, type(var)))
-                    global_block.append_op(
-                        type='fetch',
-                        inputs={'X': [var]},
-                        outputs={'Out': [fetch_var]},
-                        attrs={'col': i})
-
-        # feed var to framework
-        for op in program_cache.global_block().ops:
-            if op.desc.type() == 'feed':
-                feed_target_name = op.desc.output('Out')[0]
-                cur_feed = feed[feed_target_name]
-                if not isinstance(cur_feed, core.LoDTensor):
-                    cur_feed = self.aslodtensor(cur_feed)
-                idx = op.desc.attr('col')
-                core.set_feed_variable(scope, cur_feed, feed_var_name, idx)
-            else:
-                break
-
-        self.executor.run(program_cache.desc, scope, 0, True, True)
-        outs = [
-            core.get_fetch_variable(scope, fetch_var_name, i)
-            for i in xrange(len(fetch_list))
-        ]
+            program = self._add_feed_fetch_ops(
+                program=program,
+                feed=feed,
+                fetch_list=fetch_list,
+                feed_var_name=feed_var_name,
+                fetch_var_name=fetch_var_name)
+
+        self.feed_data(program, feed, feed_var_name, scope)
+        self.executor.run(program.desc, scope, 0, True, True)
+        outs = self.fetch_data(fetch_list, fetch_var_name, scope)
         if return_numpy:
             outs = as_numpy(outs)
         return outs

python/paddle/fluid/tests/unittests/test_executor_and_mul.py

@@ -39,6 +39,52 @@ def test_mul(self):
         self.assertEqual((100, 100), out.shape)
         self.assertTrue(numpy.allclose(out, numpy.dot(a_np, b_np)))
 
+    def test_prepare_then_run(self):
+        a = data(name='a', shape=[784], dtype='float32')
+        b = data(
+            name='b',
+            shape=[784, 100],
+            dtype='float32',
+            append_batch_size=False)
+        c = data(
+            name='c', shape=[100, 10], dtype='float32', append_batch_size=False)
+        out = mul(x=a, y=b)
+        place = core.CPUPlace()
+        a_np = numpy.random.random((100, 784)).astype('float32')
+        b_np = numpy.random.random((784, 100)).astype('float32')
+        c_np = numpy.random.random((100, 10)).astype('float32')
+        exe = Executor(place)
+        feed = {'a': a_np, 'b': b_np, 'c': c_np}
+
+        prepared_ctx = exe.prepare(feed=feed, fetch_list=[out])
+        for _ in range(2):
+            outs = exe.run_prepared_ctx(ctx=prepared_ctx, feed=feed)
+            out_np = outs[0]
+            self.assertEqual((100, 100), out_np.shape)
+            self.assertTrue(numpy.allclose(out_np, numpy.dot(a_np, b_np)))
+
+        new_out = mul(x=out, y=c)
+        new_prepared_ctx = exe.prepare(feed=feed, fetch_list=[new_out])
+
+        handle_equal = (prepared_ctx.handle == new_prepared_ctx.handle)
+        self.assertFalse(handle_equal, "handle should not be equal")
+
+        for _ in range(2):
+            outs = exe.run_prepared_ctx(ctx=new_prepared_ctx, feed=feed)
+            out_np = outs[0]
+            self.assertEqual((100, 10), out_np.shape)
+            self.assertTrue(
+                numpy.allclose(out_np, numpy.dot(numpy.dot(a_np, b_np), c_np)))
+
+        for _ in range(2):
+            outs = exe.run_prepared_ctx(ctx=prepared_ctx, feed=feed)
+            out_np = outs[0]
+            self.assertEqual((100, 100), out_np.shape)
+            self.assertTrue(numpy.allclose(out_np, numpy.dot(a_np, b_np)))
+
+        exe.delete_prepared_ctx(prepared_ctx)
+        exe.delete_prepared_ctx(new_prepared_ctx)
+
 
 if __name__ == '__main__':
     unittest.main()