Openpose的多线程测试

拖地先生 2019-11-22

2305

OpenPose人体姿态识别项目是美国卡耐基梅隆大学（CMU）基于卷积神经网络和监督学习并以caffe为框架开发的开源库。

可以实现人体动作、面部表情、手指运动等姿态估计。适用于单人和多人，具有极好的鲁棒性。是世界上首个基于深度学习的实时多人二维姿态估计应用，基于它的实例如雨后春笋般涌现。

人体姿态估计技术在体育健身、动作采集、3D试衣、舆情监测等领域具有广阔的应用前景，人们更加熟悉的应用就是抖音尬舞机。

OpenPose项目Github链接：https://github.com/CMU-Perceptual-Computing-Lab/openpose

要做为线上服务，性能估算是重要的环节，做一个简单的分析和测试。首先大致了解openpose的源码封装。

1. 默认开启多线程

// Implementation
#include <openpose/wrapper/wrapperAuxiliary.hpp>
namespace op
{
    template<typename TDatum, typename TDatums, typename TDatumsSP, typename TWorker>
    WrapperT<TDatum, TDatums, TDatumsSP, TWorker>::WrapperT(const ThreadManagerMode threadManagerMode) :
        mThreadManagerMode{threadManagerMode},
        mThreadManager{threadManagerMode},
        mMultiThreadEnabled{true}
    {
    }
    ...
}

2. demo的初始化出于顺序处理、调试或者减少时延，会根据编译宏禁用，注释掉这段。

op::Wrapper opWrapper{op::ThreadManagerMode::Asynchronous};
// Set to single-thread (for sequential processing and/or debugging and/or reducing latency)
if (FLAGS_disable_multi_thread)
    opWrapper.disableMultiThreading();

3. 多线程队列，可能导致内存问题，猜测图片加载和相关处理的数据结构。从实测看，暂时看不出内存的变化，还需要加大测试量找到最佳值，在应用进程侧控制队列和线程数。

/**
    * It sets the maximum number of elements in the queue.
    * For maximum speed, set to a very large number, but the trade-off would be:
    *  - Latency will hugely increase.
    *  - The program might go out of RAM memory (so the computer might freeze).
    * For minimum latency while keeping an optimal speed, set to -1, that will automatically
    * detect the ideal number based on how many elements are connected to that queue.
    * @param defaultMaxSizeQueues long long element with the maximum number of elements on the queue.
    */
void setDefaultMaxSizeQueues(const long long defaultMaxSizeQueues = -1);

4. 关于队列，运行时直接获取队列扔进去，等待线程队列有空闲后，进行处理。

template<typename TDatums, typename TWorker, typename TQueue>
bool ThreadManager<TDatums, TWorker, TQueue>::waitAndEmplace(TDatums& tDatums)
{
    try
    {
        if (mThreadManagerMode != ThreadManagerMode::Asynchronous
            && mThreadManagerMode != ThreadManagerMode::AsynchronousIn)
            error("Not available for this ThreadManagerMode.", __LINE__, __FUNCTION__, __FILE__);
        if (mTQueues.empty())
            error("ThreadManager already stopped or not started yet.", __LINE__, __FUNCTION__, __FILE__);
        return mTQueues[0]->waitAndEmplace(tDatums);
    }
    catch (const std::exception& e)
    {
        error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        return false;
    }
}


template<typename TDatums, typename TQueue>
bool QueueBase<TDatums, TQueue>::waitAndEmplace(TDatums& tDatums)
{
    try
    {
        std::unique_lock<std::mutex> lock{mMutex};
        mConditionVariable.wait(lock, [this]{return mTQueue.size() < getMaxSize() || mPushIsStopped; });
        return emplace(tDatums);
    }
    catch (const std::exception& e)
    {
        error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        return false;
    }
}

读完代码做一个测试

// 预先建立好100张图片，并发提交处理
func main() {
  C.initPose()
  path := "/yogadata/image/"
  suffix := ".jpeg"
  var wg sync.WaitGroup
        // 初始化大概会消耗2s的时间，排除掉这部分的影响
  time.Sleep(time.Duration(20) * time.Second)


  t1 := time.Now()
  for i := 1; i <= 100; i++ {
    num := strconv.Itoa(i)
    img := path + num + suffix
    wg.Add(1)
    go func(img string, n int) {
      defer wg.Done()
      imgc := C.CString(img)
      ret := C.getVoiceId(imgc, 0)
      fmt.Print("num:", n, "ret:", ret, "\n")
    }(img, i)
  }


  wg.Wait()
  t2 := time.Now()
  fmt.Println(t2.Sub(t1))
}