solve pslib core in stop worker (#21263)

* general table

* add sparse table
test=develop

* no cvm
test=develop

* add no_cvm
test=develop

* add note
test=develop

* code style
test=develop

* code style
test=develop

* code style
test=develop

* code style
test=develop

* code style
test=develop

* add key of optimizer
test=develop

* solve pslib stop core
test=develop

* barrier
test=develop

* add notes
test=develop

paddle/fluid/framework/fleet/fleet_wrapper.cc

@@ -91,6 +91,13 @@ void FleetWrapper::StopServer() {
 #endif
 }
 
+void FleetWrapper::FinalizeWorker() {
+#ifdef PADDLE_WITH_PSLIB
+  VLOG(3) << "Going to finalize worker";
+  pslib_ptr_->finalize_worker();
+#endif
+}
+
 uint64_t FleetWrapper::RunServer() {
 #ifdef PADDLE_WITH_PSLIB
   VLOG(3) << "Going to run server";

paddle/fluid/framework/fleet/fleet_wrapper.h

@@ -147,6 +147,8 @@ class FleetWrapper {
                   int index);
   // stop server
   void StopServer();
+  // finalize worker to make worker can be stop
+  void FinalizeWorker();
   // run server
   uint64_t RunServer();
   // gather server ip

paddle/fluid/pybind/fleet_wrapper_py.cc

@@ -55,6 +55,7 @@ void BindFleetWrapper(py::module* m) {
       .def("load_model", &framework::FleetWrapper::LoadModel)
       .def("clear_model", &framework::FleetWrapper::ClearModel)
       .def("stop_server", &framework::FleetWrapper::StopServer)
+      .def("finalize_worker", &framework::FleetWrapper::FinalizeWorker)
       .def("gather_servers", &framework::FleetWrapper::GatherServers)
       .def("gather_clients", &framework::FleetWrapper::GatherClients)
       .def("get_clients_info", &framework::FleetWrapper::GetClientsInfo)

python/paddle/fluid/incubate/fleet/parameter_server/pslib/__init__.py

@@ -182,6 +182,10 @@ class PSLib(Fleet):
             destroyed when stop() is called.
         """
         self._role_maker._barrier_worker()
+        # all worker should be finalize first
+        if self._role_maker.is_worker():
+            self._fleet_ptr.finalize_worker()
+        self._role_maker._barrier_worker()
         if self._role_maker.is_first_worker():
             self._fleet_ptr.stop_server()
         self._role_maker._barrier_worker()