curl添加全局锁

cpp/httpclientpool.cc

@@ -8,6 +8,8 @@
 #include <curl/curl.h>
 #include <curl/easy.h>
 
+#include <openssl/err.h>
+
 #include <a8/list.h>
 #include <a8/timer.h>
 #include <a8/timer_attacher.h>
@@ -426,8 +428,28 @@ namespace f8
 #else
         AsyncCurl* async_curl_ = nullptr;
 #endif
+        pthread_mutex_t* mutex_buf = nullptr;
+        size_t mutex_buf_size = 0;
     };
 
+    static void locking_function(int mode, int n, const char *file, int line)
+    {
+        if (n != f8::HttpClientPool::Instance()->Impl()->mutex_buf_size) {
+            abort();
+        }
+        pthread_mutex_t* mutex = &f8::HttpClientPool::Instance()->Impl()->mutex_buf[n];
+        if(mode & CRYPTO_LOCK) {
+            pthread_mutex_lock(mutex);
+        } else {
+            pthread_mutex_unlock(mutex);
+        }
+    }
+
+    static unsigned long id_function(void)
+    {
+        return pthread_self();
+    }
+
     void HttpClientPool::Init(int thread_num, int sys_num, int user_num)
     {
         if (thread_num < 0 || sys_num < 0 || user_num < 0) {
@@ -439,6 +461,9 @@ namespace f8
         this->thread_num = thread_num;
         this->sys_num = sys_num;
         this->user_num = user_num;
+        #if 1
+        curl_global_init(CURL_GLOBAL_ALL);
+        #endif
         impl_ = new HttpClientPoolImpl();
         impl_->Init();
         MsgQueue::Instance()->RegisterCallBack(impl_->exec_async_http_msgid,
@@ -456,13 +481,42 @@ namespace f8
                                                }
                                                );
         impl_->SetThreadNum(thread_num);
+        #if 1
+        {
+            if (CRYPTO_num_locks() > 0) {
+                impl_->mutex_buf = (pthread_mutex_t*)malloc(CRYPTO_num_locks() * sizeof(pthread_mutex_t));
+                impl_->mutex_buf_size = CRYPTO_num_locks();
+                for (size_t i = 0; i < impl_->mutex_buf_size; ++i) {
+                    pthread_mutex_init(&impl_->mutex_buf[i], nullptr);
+                }
+            }
+            CRYPTO_set_id_callback(id_function);
+            CRYPTO_set_locking_callback(locking_function);
+        }
+        #endif
     }
 
     void HttpClientPool::UnInit()
     {
         impl_->UnInit();
+        #if 1
+        {
+            CRYPTO_set_id_callback(nullptr);
+            CRYPTO_set_locking_callback(nullptr);
+            if (impl_->mutex_buf) {
+                for (size_t i = 0; i < impl_->mutex_buf_size; ++i) {
+                    pthread_mutex_destroy(&impl_->mutex_buf[i]);
+                }
+                free(impl_->mutex_buf);
+                impl_->mutex_buf = nullptr;
+            }
+        }
+        #endif
         delete impl_;
         impl_ = nullptr;
+        #if 1
+        curl_global_cleanup();
+        #endif
     }
 
     void HttpClientPool::HttpGet(a8::XParams param, AsyncHttpOnOkFunc on_ok, AsyncHttpOnErrorFunc on_error,
@@ -485,4 +539,9 @@ namespace f8
         return impl_->pending_num;
     }
 
+    HttpClientPoolImpl* HttpClientPool::Impl()
+    {
+        return impl_;
+    }
+
 }

cpp/httpclientpool.h

@@ -34,6 +34,7 @@ namespace f8
                       long long hash_code, a8::XObject* headers = nullptr);
         //获取执行中的数量
         long long GetPendingNum();
+        HttpClientPoolImpl* Impl();
 
     private:
         HttpClientPoolImpl* impl_ = nullptr;