C++多線程解碼音頻數據

Question

我需要使用Opus解碼器儘可能快地解碼音頻數據。

目前我的申請不夠快。解碼速度儘可能快，但我需要獲得更多的速度。

我需要解碼大約100段音頻。 T 這些部分不是連續的（它們彼此不相關）。

我正在考慮使用多線程，因此我不必等到100個解碼中的一個完成。在我的夢中，我可以同時開始一切。 我以前沒有使用過多線程。

因此，我想問問我的方法是否總體良好，或者是否存在某種思維錯誤。

謝謝。

Answer 1

這個答案很可能將來自社區需要一點點的改進，因爲它是一個很長一段時間，因爲我曾在這樣的環境中，但這裏是一個開始 -

因爲你是新的多線程在C++，從一個簡單的項目開始，創建一堆執行簡單任務的pthread。

以下是創建並行線程的快速和小例子：

#include <stdio.h> 
#include <stdlib.h> 
#include <pthread.h> 

void* ThreadStart(void* arg); 

int main(int count, char** argv) { 
     pthread_t thread1, thread2; 

     int* threadArg1 = (int*)malloc(sizeof(int)); 
     int* threadArg2 = (int*)malloc(sizeof(int)); 

     *threadArg1 = 1; 
     *threadArg2 = 2; 

     pthread_create(&thread1, NULL, &ThreadStart, (void*)threadArg1); 
     pthread_create(&thread2, NULL, &ThreadStart, (void*)threadArg2); 

     pthread_join(thread1, NULL); 
     pthread_join(thread2, NULL); 
     free(threadArg1); 
     free(threadArg2); 
} 

void* ThreadStart(void* arg) { 

     int threadNum = *((int*)arg); 
     printf("hello world from thread %d\n", threadNum); 

     return NULL; 
} 

接下來，你將要使用多個OPUS解碼器。 Opus似乎是線程安全的，只要您爲每個線程創建單獨的OpusDecoder對象。

爲了養活工作到你的線程，你需要能夠在一個線程安全的方式訪問待審批工作單位的名單。您可以使用std::vector或std::queue，但在添加到它時以及從中移除時，您必須在其周圍使用鎖，並且您需要使用計數信號量，以便線程可以阻止但保持活動狀態慢慢地將工作單元添加到隊列中（例如，從磁盤讀取的文件的緩衝區）。

以下是上述類似的一些示例代碼，演示如何使用共享隊列，以及如何使線程等待您填寫的隊列：

#include <stdio.h> 
#include <stdlib.h> 
#include <pthread.h> 
#include <queue> 
#include <semaphore.h> 
#include <unistd.h> 

void* ThreadStart(void* arg); 

static std::queue<int> workunits; 
static pthread_mutex_t workunitLock; 
static sem_t workunitCount; 

int main(int count, char** argv) { 
    pthread_t thread1, thread2; 

    pthread_mutex_init(&workunitLock, NULL); 
    sem_init(&workunitCount, 0, 0); 

    pthread_create(&thread1, NULL, &ThreadStart, NULL); 
    pthread_create(&thread2, NULL, &ThreadStart, NULL); 

    // Make a bunch of workunits while the threads are running. 
    for (int i = 0; i < 200; i++){ 
     pthread_mutex_lock(&workunitLock); 

     workunits.push(i); 
     sem_post(&workunitCount); 

     pthread_mutex_unlock(&workunitLock); 

     // Pretend that it takes some effort to create work units; 
     // this shows that the threads really do block patiently 
     // while we generate workunits. 
     usleep(5000); 
    } 

    // Sometime in the next while, the threads will be blocked on 
    // sem_wait because they're waiting for more workunits. None 
    // of them are quitting because they never saw an empty queue. 
    // Pump the semaphore once for each thread so they can wake 
    // up, see the empty queue, and return. 

    sem_post(&workunitCount); 
    sem_post(&workunitCount); 

    pthread_join(thread1, NULL); 
    pthread_join(thread2, NULL); 

    pthread_mutex_destroy(&workunitLock); 
    sem_destroy(&workunitCount); 

} 

void* ThreadStart(void* arg) { 

    int workUnit; 
    bool haveUnit = true; 

    while(haveUnit) { 
     sem_wait(&workunitCount); 

     pthread_mutex_lock(&workunitLock); 

     // Figure out if there's a unit, grab it under 
     // the lock, then release the lock as soon as we can. 
     // After we release the lock, then we can 'process' 
     // the unit without blocking everybody else. 
     haveUnit = !workunits.empty(); 

     if (haveUnit) { 
      workUnit = workunits.front(); 
      workunits.pop(); 
     } 
     pthread_mutex_unlock(&workunitLock); 

     // Now that we're not under the lock, we can spend 
     // as much time as we want processing the workunit. 
     if (haveUnit) { 
      printf("Got workunit %d\n", workUnit); 
     } 
    } 

    return NULL; 
} 

Answer 2

在加入多線程之前，作爲加快速度的解決方案，研究在訂閱下超額訂閱&的概念。

如果音頻處理涉及長時間阻塞IO調用，那麼多線程是值得的。

Answer 3

雖然你的問題含糊不清並不能真正幫助...怎麼樣：

Create a list of audio files to convert. 

While there is a free processor, 
    launch the decoder application with the next file in the queue. 
Repeat until there is nothing else in the list 

如果在測試過程中，你會發現處理器並不總是100％忙碌，推出2每處理器進行解碼。

它可以很容易地用bash/tcl/python來完成。

Answer 4

你會打亂你的工作任務。讓我們假設你的過程實際上是CPU綁定的（你指出它是，但是......通常並不那麼簡單）。

現在，你解碼100節：

我想使用多線程，這樣我就不必等到100個解碼的一個完成。在我的夢中，我可以同時開始一切。

實際上，您應該使用一個接近機器核心數量的數字。假設一個現代桌面（例如2-8個核心），一次運行100個線程只會減慢速度;內核會浪費大量時間從一個線程切換到另一個線程，並且該進程也可能使用更高的峯值資源並爭奪類似的資源。

所以，只需創建一個任務池，將活動任務的數量限制爲核心數量。每個任務將（通常）表示對一個輸入文件（節）執行的解碼工作。這樣，解碼過程實際上並不是在多個線程之間共享數據（允許您避免鎖定和其他資源爭用）。

完成後，返回並微調任務池中的進程數（例如，在多臺機器上使用完全相同的輸入和秒錶）。最快速度可能低於或高於內核數量（很可能是因爲磁盤I/O）。它也有助於配置文件。

因此，我想問一下，如果我的方法總體良好，或者在某處存在思維錯誤。

是的，如果問題是CPU綁定，那一般是好的。這也假設您的解碼器/相關軟件能夠運行多個線程。

如果這些磁盤上的文件，你會發現的問題是，你可能會需要從多核心優化你怎麼讀（寫？）的文件。因此，允許它一次運行8個作業可以使您的問題變成磁盤綁定 - 而且同時使用8個讀寫器是使用硬盤的一種不好的方式，因此您可能會發現它不如您預期的那麼快。因此，您可能需要爲您的並行解碼實現優化I/O。在這方面，使用更大的緩衝區大小，但這是在內存中的成本。

Answer 5

而不是讓你自己的線程和管理他們的，我建議你使用一個線程池，讓您的解碼任務到池中。池會將任務分配給它和系統可以處理的任意數量的線程。雖然有不同類型的線程池，所以您可以設置一些參數，例如強制它使用某些特定數量的線程，或者如果您應該允許該池繼續增加線程數。

有一點要記住的是，更多的線程並不意味着它們並行執行。我認爲，正確的說法是並行的，除非你有每個線程不同的CPU（這將使真正的並行）上運行保障

您的整個池可以停下來，如果阻塞IO。

Answer 6

您可以在一般的使用線程，但鎖定有一些的問題。我將圍繞POSIX線程和鎖定來回答這個問題，但這是相當普遍的，您可以將這個想法移植到任何平臺上。但是，如果您的工作需要任何形式的鎖定，您可能會發現以下內容有用。此外，最好還是繼續使用現有的線程，因爲線程創建代價高昂（請參閱線程池）。

對於「實時」音頻，鎖定通常是一個糟糕的主意，因爲它會增加延遲，但這對於解碼/編碼的實時作業來說是完全可以的，即使對於實時的作業，您也可以獲得更好的性能並且不會丟幀通過使用一些線程知識。

對於音頻，信號量是一個糟糕的主意。至少我的系統（POSIX信號燈）在我嘗試時速度太慢，但如果您正在考慮跨線程鎖定（而不是鎖定在一個線程中並在同一線程中解鎖的類型），則需要它們。 POSIX互斥體只允許自鎖和自解鎖（你必須在同一個線程中執行這兩個操作），否則程序可能會工作，但它是未定義的行爲，應該避免。

大多數無鎖原子操作可能會讓您有足夠的自由來使用某些功能（如鎖定），但具有更好的性能。