PaddlePaddle
diff --git a/‎paddle/fluid/operators/pull_sparse_op.cc
Copy file name to clipboard
-150Lines changed: 0 additions & 150 deletions b/‎paddle/fluid/operators/pull_sparse_op.cc
Copy file name to clipboard
-150Lines changed: 0 additions & 150 deletions
diff --git a/‎paddle/fluid/operators/pull_sparse_op.h
Copy file name to clipboard
-85Lines changed: 0 additions & 85 deletions b/‎paddle/fluid/operators/pull_sparse_op.h
Copy file name to clipboard
-85Lines changed: 0 additions & 85 deletions
diff --git a/‎paddle/fluid/operators/pull_sparse_v2_op.h
Copy file name to clipboard
+42-1Lines changed: 42 additions & 1 deletion b/‎paddle/fluid/operators/pull_sparse_v2_op.h
Copy file name to clipboard
+42-1Lines changed: 42 additions & 1 deletion
diff --git a/‎python/paddle/base/device_worker.py
Copy file name to clipboard
-1Lines changed: 0 additions & 1 deletion b/‎python/paddle/base/device_worker.py
Copy file name to clipboard
-1Lines changed: 0 additions & 1 deletion
diff --git a/‎python/paddle/incubate/distributed/fleet/parameter_server/pslib/__init__.py
Copy file name to clipboard
-89Lines changed: 0 additions & 89 deletions b/‎python/paddle/incubate/distributed/fleet/parameter_server/pslib/__init__.py
Copy file name to clipboard
-89Lines changed: 0 additions & 89 deletions
@@ -20,11 +20,52 @@
 #include "paddle/fluid/framework/fleet/fleet_wrapper.h"
 #include "paddle/fluid/framework/op_registry.h"
 #include "paddle/fluid/framework/tensor.h"
-#include "paddle/fluid/operators/pull_sparse_op.h"
 
 namespace paddle {
 namespace operators {
 
+template <typename T>
+void PullSparseFunctor(const framework::ExecutionContext& ctx) {
+  auto inputs = ctx.MultiInput<phi::DenseTensor>("Ids");
+  auto outputs = ctx.MultiOutput<phi::DenseTensor>("Out");
+  uint32_t fea_dim = static_cast<uint32_t>(ctx.Attr<int>("EmbeddingDim"));
+  uint64_t padding_id = static_cast<uint64_t>(ctx.Attr<int>("PaddingId"));
+  auto table_id = static_cast<uint32_t>(ctx.Attr<int>("TableId"));
+  // note: GetInstance() is not thread-safe
+  // we assume FleetWrapper has been already initialized
+  auto fleet_ptr = framework::FleetWrapper::GetInstance();
+  fleet_ptr->PullSparseToTensorSync(
+      table_id, fea_dim, padding_id, ctx.GetPlace(), &inputs, &outputs);
+}
+
+template <typename T>
+void PushSparseFunctor(const framework::ExecutionContext& ctx) {
+  auto inputs = ctx.MultiInput<phi::DenseTensor>("Ids");
+  auto grads = ctx.MultiInput<phi::DenseTensor>(framework::GradVarName("Out"));
+  uint32_t fea_dim = static_cast<uint32_t>(ctx.Attr<int>("EmbeddingDim"));
+  std::string accessor = ctx.Attr<std::string>("AccessorClass");
+  bool scale_sparse = ctx.Attr<bool>("ScaleSparseGrad");
+  uint64_t padding_id = static_cast<uint64_t>(ctx.Attr<int>("PaddingId"));
+  const std::string& label_name = ctx.Attr<std::string>("CtrLabelName");
+  const framework::Scope& scope = ctx.scope();
+  auto input_names = ctx.Attr<std::vector<std::string>>("InputNames");
+  auto table_id = static_cast<uint32_t>(ctx.Attr<int>("TableId"));
+  // note: GetInstance() is not thread-safe
+  // we assume FleetWrapper has been already initialized
+  auto fleet_ptr = framework::FleetWrapper::GetInstance();
+  fleet_ptr->PushSparseFromTensorWithLabelAsync(scope,
+                                                table_id,
+                                                fea_dim,
+                                                padding_id,
+                                                scale_sparse,
+                                                accessor,
+                                                label_name,
+                                                ctx.GetPlace(),
+                                                input_names,
+                                                &inputs,
+                                                &grads);
+}
+
 template <typename T, typename DeviceContext>
 class PullSparseV2CPUKernel : public framework::OpKernel<T> {
  public:
 
@@ -92,7 +92,6 @@ def _gen_worker_desc(self, trainer_desc):
             trainer_desc.hogwild_param.skip_ops.extend(
                 [
                     "feed",
-                    "push_sparse",
                     "push_sparse_v2",
                     "push_dense",
                     "distributed_push_sparse",
 
@@ -1045,95 +1045,6 @@ def _prepare_params(
         raise ValueError("dtype must be float32")
 
 
-def _fleet_embedding(
-    input,
-    size,
-    is_sparse=False,
-    is_distributed=False,
-    padding_idx=None,
-    param_attr=None,
-    dtype='float32',
-):
-    """
-    Add fleet embedding, this interface is not for users.
-
-    Args:
-        input (Variable|list of Variable): Input is a Tensor<int64> Variable.
-        size (list[int]): The embedding dim.
-        is_sparse (bool, optional): Whether input is sparse ids. Default is False.
-        is_distributed (bool, optional): Whether in distributed mode. Default is False.
-        padding_idx (int, optional): Padding idx of input. Default is None.
-        param_attr (ParamAttr, optional): To specify the weight parameter property. Default is None.
-        dtype (str, optional): Data type of output. Default is 'float32'.
-    """
-
-    def _pull_sparse(
-        input,
-        size,
-        table_id,
-        accessor_class,
-        name="embedding",
-        ctr_label_name="",
-        padding_id=0,
-        dtype='float32',
-        scale_sparse_grad=True,
-    ):
-        helper = LayerHelper(name, **locals())
-        inputs = helper.multiple_input()
-        outs = [helper.create_variable_for_type_inference(dtype)]
-        input_names = [i.name for i in inputs]
-        attrs = {
-            'EmbeddingDim': size,
-            'TableId': table_id,
-            'AccessorClass': accessor_class,
-            'CtrLabelName': ctr_label_name,
-            'PaddingId': padding_id,
-            'ScaleSparseGrad': scale_sparse_grad,
-            'InputNames': input_names,
-            # this is only for compatible with embedding op
-            'is_distributed': True,
-        }
-        # this is only for compatible with embedding op
-        w, _ = helper.create_or_get_global_variable(
-            name=name,
-            shape=[size],
-            dtype=dtype,
-            is_bias=False,
-            persistable=True,
-        )
-        helper.append_op(
-            type='pull_sparse',
-            inputs={'Ids': inputs, 'W': w},
-            outputs={'Out': outs},
-            attrs=attrs,
-        )
-        if len(outs) == 1:
-            return outs[0]
-        return outs
-
-    # check and set params
-    _prepare_params(
-        input, size, is_sparse, is_distributed, padding_idx, param_attr, dtype
-    )
-    name = param_attr.name
-    size = size[-1]
-    if padding_idx is None:
-        padding_idx = 0
-    global FLEET_GLOBAL_DICT
-
-    return _pull_sparse(
-        input=input,
-        size=size,
-        table_id=FLEET_GLOBAL_DICT["emb_to_table"][name],
-        accessor_class=FLEET_GLOBAL_DICT["emb_to_accessor"][name],
-        name=name,
-        ctr_label_name=FLEET_GLOBAL_DICT["click_name"],
-        padding_id=padding_idx,
-        dtype=dtype,
-        scale_sparse_grad=FLEET_GLOBAL_DICT["scale_sparse_grad"],
-    )
-
-
 def _fleet_embedding_v2(
     input,
     size,
Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,6 @@ def _gen_worker_desc(self, trainer_desc):`
`92`	`92`	`trainer_desc.hogwild_param.skip_ops.extend(`
`93`	`93`	`[`
`94`	`94`	`"feed",`
`95`		`- "push_sparse",`
`96`	`95`	`"push_sparse_v2",`
`97`	`96`	`"push_dense",`
`98`	`97`	`"distributed_push_sparse",`