AWS S3性能上不去？你以为的的异步可能不是你以为的

// init 和 unint aws接口
Aws::InitAPI(Aws::SDKOptions);
Aws::ShutdownAPI(options);
// 同步上传与下载接口
PutObjectOutcome S3Client::PutObject(const PutObjectRequest& request);
GetObjectOutcome S3Client::GetObject(const GetObjectRequest& request);
// 异步上传与下载接口
void S3Client::PutObjectAsync(const PutObjectRequest& request, const PutObjectResponseReceivedHandler& handler, const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context);
void S3Client::GetObjectAsync(const GetObjectRequest& request, const GetObjectResponseReceivedHandler& handler, const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context)

void S3Client::PutObjectAsync(const PutObjectRequest& request, const PutObjectResponseReceivedHandler& handler, const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context) const
{
  // 从这里我们可以看到把PutObject()作为任务丢到了aws执行器的执行队列里
  m_executor->Submit( [this, request, handler, context](){ this->PutObjectAsyncHelper( request, handler, context ); } );
}
void S3Client::PutObjectAsyncHelper(const PutObjectRequest& request, const PutObjectResponseReceivedHandler& handler, const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context) const
{
  handler(this, request, PutObject(request), context);
}

namespace Aws
{
    namespace Utils
    {
        namespace Threading
        {
            class ThreadTask;
            // 如果想实现一个新的执行器，继承该类，并实现SubmitToThread函数
            /**
            * Interface for implementing an Executor, to implement a custom thread execution strategy, inherit from this class
            * and override SubmitToThread().
            */
            class AWS_CORE_API Executor
            {
            public:                
                virtual ~Executor() = default;
                /**
                 * Send function and its arguments to the SubmitToThread function.
                 */
                template<class Fn, class ... Args>
                bool Submit(Fn&& fn, Args&& ... args)
                {
                    std::function<void()> callable{ std::bind(std::forward<Fn>(fn), std::forward<Args>(args)...) };
                    return SubmitToThread(std::move(callable));
                }
            protected:
                /**
                * To implement your own executor implementation, then simply subclass Executor and implement this method.
                */
                virtual bool SubmitToThread(std::function<void()>&&) = 0;
            };
            /**
            * Default Executor implementation. Simply spawns a thread and detaches it.
            */
            class AWS_CORE_API DefaultExecutor : public Executor
            {
            public:
                DefaultExecutor() : m_state(State::Free) {}
                ~DefaultExecutor();
            protected:
                enum class State
                {
                    Free, Locked, Shutdown
                };
                bool SubmitToThread(std::function<void()>&&) override;
                void Detach(std::thread::id id);
                std::atomic<State> m_state;
                Aws::UnorderedMap<std::thread::id, std::thread> m_threads;
            };
            enum class OverflowPolicy
            {
                QUEUE_TASKS_EVENLY_ACCROSS_THREADS,
                REJECT_IMMEDIATELY
            };
            /**
            * Thread Pool Executor implementation.
            */
            class AWS_CORE_API PooledThreadExecutor : public Executor
            {
            public:
                PooledThreadExecutor(size_t poolSize, OverflowPolicy overflowPolicy = OverflowPolicy::QUEUE_TASKS_EVENLY_ACCROSS_THREADS);
                ~PooledThreadExecutor();
                /**
                * Rule of 5 stuff.
                * Don't copy or move
                */
                PooledThreadExecutor(const PooledThreadExecutor&) = delete;
                PooledThreadExecutor& operator =(const PooledThreadExecutor&) = delete;
                PooledThreadExecutor(PooledThreadExecutor&&) = delete;
                PooledThreadExecutor& operator =(PooledThreadExecutor&&) = delete;
            protected:
                bool SubmitToThread(std::function<void()>&&) override;
            private:
                Aws::Queue<std::function<void()>*> m_tasks;
                std::mutex m_queueLock;
                Aws::Utils::Threading::Semaphore m_sync;
                Aws::Vector<ThreadTask*> m_threadTaskHandles;
                size_t m_poolSize;
                OverflowPolicy m_overflowPolicy;
                /**
                 * Once you call this, you are responsible for freeing the memory pointed to by task.
                 */
                std::function<void()>* PopTask();
                bool HasTasks();
                friend class ThreadTask;
            };
        } // namespace Threading
    } // namespace Utils
} // namespace Aws

bool DefaultExecutor::SubmitToThread(std::function<void()>&&  fx)
{
    auto main = [fx, this] { 
        // 调用传进来的任务，如果是上面的调用，那么便是PutObject
        fx(); 
        Detach(std::this_thread::get_id()); 
    };
    State expected;
    do
    {
        expected = State::Free;
        if(m_state.compare_exchange_strong(expected, State::Locked))
        {
            // 启动线程执行上面的main中内容
            std::thread t(main);
            const auto id = t.get_id(); // copy the id before we std::move the thread
            m_threads.emplace(id, std::move(t));
            m_state = State::Free;
            return true;
        }
    }
    while(expected != State::Shutdown);
    return false;
}

bool PooledThreadExecutor::SubmitToThread(std::function<void()>&& fn)
{
    //avoid the need to do copies inside the lock. Instead lets do a pointer push.
    std::function<void()>* fnCpy = Aws::New<std::function<void()>>(POOLED_CLASS_TAG, std::forward<std::function<void()>>(fn));
    {
        std::lock_guard<std::mutex> locker(m_queueLock);
    /*
   
        当任务队列中的请求数量大于线程池中线程数量时，有两种策略：
        QUEUE_TASKS_EVENLY_ACROSS_THREADS：不管，一直丢就好(即使内存爆了也无所谓，所以如果使用这种策略，业务自己需  要保证内存的消耗控制)
        REJECT_IMMEDIATELY：直接拒绝该task
        
        - QUEUE_TASKS_EVENLY_ACROSS_THREADS
        
            will allow you to push as many tasks as you want to the executor. It will make sure tasks are queued and pulled by each thread as quickly as possible. For most cases, QUEUE_TASKS_EVENLY_ACROSS_THREADS is the preferred option.
        - REJECT_IMMEDIATELY
          
          will reject the task submission if the queued task length ever exceeds the size of the thread pool.
    */ 
        if (m_overflowPolicy == OverflowPolicy::REJECT_IMMEDIATELY && m_tasks.size() >= m_poolSize)
        {
            Aws::Delete(fnCpy);
            return false;
        }
        // 把任务丢进队列
        m_tasks.push(fnCpy);
    }
    m_sync.Release();
    return true;
}
// 从队列中取出一个任务
std::function<void()>* PooledThreadExecutor::PopTask()
{
    std::lock_guard<std::mutex> locker(m_queueLock);
    if (m_tasks.size() > 0)
    {
        std::function<void()>* fn = m_tasks.front();
        if (fn)
        {           
            m_tasks.pop();
            return fn;
        }
    }
    return nullptr;
}
// 队列中是否有任务
bool PooledThreadExecutor::HasTasks()
{
    std::lock_guard<std::mutex> locker(m_queueLock);
    return m_tasks.size() > 0;
}
// 创建时根据传递进来的线程池数量poolSize创建对应数量的
PooledThreadExecutor::PooledThreadExecutor(size_t poolSize, OverflowPolicy overflowPolicy) :
    m_sync(0, poolSize), m_poolSize(poolSize), m_overflowPolicy(overflowPolicy)
{
    for (size_t index = 0; index < m_poolSize; ++index)
    {
        m_threadTaskHandles.push_back(Aws::New<ThreadTask>(POOLED_CLASS_TAG, *this));
    }
}

PutObjectOutcome S3Client::PutObject(const PutObjectRequest& request) const
{
    Aws::Http::URI uri = computeEndpointOutcome.GetResult().first;
  Aws::StringStream ss;
  ss << "/";
  ss << request.GetKey();
  uri.SetPath(uri.GetPath() + ss.str());
  // 发送请求
  XmlOutcome outcome = MakeRequest(uri, request, Aws::Http::HttpMethod::HTTP_PUT, Aws::Auth::SIGV4_SIGNER, computeEndpointOutcome.GetResult().second.c_str() /*signerRegionOverride*/);
  // 构建返回值返给用户
  return PutObjectOutcome(PutObjectResult(outcome.GetResult()));
}
// 接下 MakeRequest的调用栈如下:
MakeRequest（aws/aws-cpp-sdk-core/source/client/AWSClient.cpp)
--> AttemptExhaustively
  --> AttemptOneRequest
    --> CurlHttpClient::MakeRequest (aws/aws-cpp-sdk-core/source/http/curl/CurlHttpClient.cpp)
      --> MakeRequestInternal
void CurlHttpClient::MakeRequestInternal(HttpRequest& request,
        std::shared_ptr<StandardHttpResponse>& response,
        Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
        Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const
{
    URI uri = request.GetUri();
    Aws::String url = uri.GetURIString();
    
    // curl相关参数设置
    curl_easy_setopt(connectionHandle, CURLOPT_URL, url.c_str());
    curl_easy_setopt(connectionHandle, CURLOPT_WRITEFUNCTION, WriteData);
    curl_easy_setopt(connectionHandle, CURLOPT_WRITEDATA, &writeContext);
    curl_easy_setopt(connectionHandle, CURLOPT_HEADERFUNCTION, WriteHeader);
    curl_easy_setopt(connectionHandle, CURLOPT_HEADERDATA, response.get());
        
    ......
    
     // 可以看到底层调用的是curl函数
     // curl_easy_perform是一个同步阻塞函数(阻塞直到收到curl的返回为止)
     CURLcode curlResponseCode = curl_easy_perform(connectionHandle);
     
     if (curlResponseCode != CURLE_OK && shouldContinueRequest)
      
      {
       response->SetClientErrorType(CoreErrors::NETWORK_CONNECTION);
} else {
          
          response->SetResponseCode(static_cast<HttpResponseCode>(responseCode));
      }
}

MaxBW = 线程池线程数量 * 一个线程每秒处理请求个数 * 请求大小
 
MaxBW = 32 * （1000/100）* 128KB =  40MB