From 18d34ed47f0f42cc53d6d388ed4db7cfe0553553 Mon Sep 17 00:00:00 2001
From: anancds <anancds@163.com>
Date: Tue, 17 Nov 2020 16:49:10 +0800
Subject: [PATCH] added libevent pthread

---
 mindspore/ccsrc/ps/core/http_server.cc | 28 +++++++++++++++++++-------
 mindspore/ccsrc/ps/core/http_server.h  | 10 ++++++++-
 mindspore/ccsrc/ps/core/tcp_server.cc  |  5 ++++-
 3 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/mindspore/ccsrc/ps/core/http_server.cc b/mindspore/ccsrc/ps/core/http_server.cc
index 548e6ec1c6..56c7852c4f 100644
--- a/mindspore/ccsrc/ps/core/http_server.cc
+++ b/mindspore/ccsrc/ps/core/http_server.cc
@@ -49,6 +49,13 @@ bool HttpServer::InitServer() {
     MS_LOG(EXCEPTION) << "The http server ip:" << server_address_ << " is illegal!";
   }
 
+  int result = evthread_use_pthreads();
+  if (result != 0) {
+    MS_LOG(EXCEPTION) << "Use event pthread failed!";
+  }
+
+  is_stop_ = false;
+
   event_base_ = event_base_new();
   MS_EXCEPTION_IF_NULL(event_base_);
   event_http_ = evhttp_new(event_base_);
@@ -146,13 +153,20 @@ bool HttpServer::Start() {
 
 void HttpServer::Stop() {
   MS_LOG(INFO) << "Stop http server!";
-  if (event_http_) {
-    evhttp_free(event_http_);
-    event_http_ = nullptr;
-  }
-  if (event_base_) {
-    event_base_free(event_base_);
-    event_base_ = nullptr;
+  if (!is_stop_.load()) {
+    int ret = event_base_loopbreak(event_base_);
+    if (ret != 0) {
+      MS_LOG(EXCEPTION) << "event base loop break failed!";
+    }
+    if (event_http_) {
+      evhttp_free(event_http_);
+      event_http_ = nullptr;
+    }
+    if (event_base_) {
+      event_base_free(event_base_);
+      event_base_ = nullptr;
+    }
+    is_stop_ = true;
   }
 }
 
diff --git a/mindspore/ccsrc/ps/core/http_server.h b/mindspore/ccsrc/ps/core/http_server.h
index acea23db65..6e85e83ce2 100644
--- a/mindspore/ccsrc/ps/core/http_server.h
+++ b/mindspore/ccsrc/ps/core/http_server.h
@@ -25,6 +25,7 @@
 #include <event2/keyvalq_struct.h>
 #include <event2/listener.h>
 #include <event2/util.h>
+#include <event2/thread.h>
 
 #include <cstdio>
 #include <cstdlib>
@@ -32,6 +33,7 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <atomic>
 
 namespace mindspore {
 namespace ps {
@@ -55,7 +57,12 @@ class HttpServer {
  public:
   // Server address only support IPV4 now, and should be in format of "x.x.x.x"
   explicit HttpServer(const std::string &address, std::uint16_t port)
-      : server_address_(address), server_port_(port), event_base_(nullptr), event_http_(nullptr), is_init_(false) {}
+      : server_address_(address),
+        server_port_(port),
+        event_base_(nullptr),
+        event_http_(nullptr),
+        is_init_(false),
+        is_stop_(true) {}
 
   ~HttpServer();
 
@@ -84,6 +91,7 @@ class HttpServer {
   struct event_base *event_base_;
   struct evhttp *event_http_;
   bool is_init_;
+  std::atomic<bool> is_stop_;
 };
 
 }  // namespace core
diff --git a/mindspore/ccsrc/ps/core/tcp_server.cc b/mindspore/ccsrc/ps/core/tcp_server.cc
index 9a5c1b5987..9093eec627 100644
--- a/mindspore/ccsrc/ps/core/tcp_server.cc
+++ b/mindspore/ccsrc/ps/core/tcp_server.cc
@@ -198,7 +198,10 @@ void TcpServer::ListenerCallback(struct evconnlistener *, evutil_socket_t fd, st
   struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
   if (!bev) {
     MS_LOG(ERROR) << "Error constructing buffer event!";
-    event_base_loopbreak(base);
+    int ret = event_base_loopbreak(base);
+    if (ret != 0) {
+      MS_LOG(EXCEPTION) << "event base loop break failed!";
+    }
     return;
   }