我正在努力實現以下目標：

將我的Raspberry Pi相機中的視頻寫入磁盤，不受任何流式干擾

通過網絡流式傳輸相同的視頻優化延遲

重要的是流不會干擾正在寫入磁盤的視頻，因為網絡連接可能不穩定，例如WiFi路由器可能超出范圍等。

要做到這一點，我嘗試的第一件事是：

#Receiver side

FPS="30"

netcat -l -p 5000 | mplayer -vf scale -zoom -xy 1280 -fps $FPS -cache-min 50 -cache 1024 - &

#RPi side

FPS="30"

mkfifo netcat_fifo

raspivid -t 0 -md 5 -fps $FPS -o - | tee --output-error=warn netcat_fifo > $video_out &

cat netcat_fifo | netcat -v 192.168.0.101 5000 &> $netcat_log &

流媒體工作得非常好。但是，當我關閉路由器，模擬網絡問題時，我的$ video_out被切斷了。我認為這是由于netcat_fifo的反壓。

我在stackexchange找到了一個關于非阻塞FIFO的解決方案，通過用ftee替換tee：

它現在阻止我的$ video_out受到流媒體的影響，但是流媒體本身非常不穩定。最好的結果是使用以下腳本：

#RPi side

FPS="30"

MULTIPIPE="ftee"

mkfifo netcat_fifo

raspivid -t 0 -md 5 -fps $FPS -o - | ./${MULTIPIPE} netcat_fifo > $video_out &

cat netcat_fifo | mbuffer --direct -t -s 2k 2> $mbuffer_log | netcat -v 192.168.0.101 5000 &> $netcat_log &

當我檢查mbuffer日志時，我診斷出一個大部分時間仍為空的FIFO，但具有99-100％利用率的峰值。在這些峰值期間，我的mplayer接收方在解碼視頻時有很多錯誤，需要大約5秒才能恢復。在此間隔之后，mbuffer日志再次顯示空FIFO。 empty-> full-> empty繼續打開。

我有兩個問題：

我使用正確的方法來解決我的問題嗎？

如果是這樣，我如何在保持$ video_out文件完整的同時渲染我的流媒體更強大？

答案

我有一點嘗試，它似乎在我的Raspberry Pi 3上非常穩定地工作。它評論很好，所以它應該很容易理解，但你可以隨時詢問是否有任何問題。

基本上有3個線程：

主程序 - 它不斷從stdin讀取它的raspivid并循環地將數據放入一堆緩沖區

磁盤寫入程序線程 - 它不斷循環遍歷緩沖區列表，等待下一個緩沖區變滿。當緩沖區已滿時，它會將內容寫入磁盤，將緩沖區標記為已寫入并移至下一個緩沖區

fifo writer線程 - 它不斷循環遍歷緩沖區列表，等待下一個緩沖區變滿。當緩沖區已滿時，它會將內容寫入fifo，刷新fifo以減少滯后并將緩沖區標記為已寫入并移至下一個緩沖區。錯誤被忽略。

所以，這是代碼：

// main.cpp

// Mark Setchell

//

// Read video stream from "raspivid" and write (independently) to both disk file

// and stdout - for onward netcatting to another host.

//

// Compiles with:

// g++ main.cpp -o main -lpthread

//

// Run on Raspberry Pi with:

// raspivid -t 0 -md 5 -fps 30 -o - | ./main video.h264 | netcat -v 192.168.0.8 5000

//

// Receive on other host with:

// netcat -l -p 5000 | mplayer -vf scale -zoom -xy 1280 -fps 30 -cache-min 50 -cache 1024 -

#include

#include

#include

#include

#include

#include

#include

#include

#include

#define BUFSZ 65536

#define NBUFS 64

class Buffer{

public:

int bytes=0;

std::atomic NeedsWriteToDisk{0};

std::atomic NeedsWriteToFifo{0};

unsigned char data[BUFSZ];

};

std::vector buffers(NBUFS);

// This is the DiskWriter thread.

// It loops through all the buffers waiting in turn for each one to become ready

// then writes it to disk and marks the buffer as written before moving to next

// buffer.

void DiskWriter(char* filename){

int bufIndex=0;

// Open output file

int fd=open(filename,O_CREAT|O_WRONLY|O_TRUNC,S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP);

if(fd==-1)

{

std::cerr << "ERROR: Unable to open output file" << std::endl;

exit(EXIT_FAILURE);

}

bool Error=false;

while(!Error){

// Wait for buffer to be filled by main thread

while(buffers[bufIndex].NeedsWriteToDisk!=1){

// std::this_thread::sleep_for(std::chrono::milliseconds(1));

}

// Write to disk

int bytesToWrite=buffers[bufIndex].bytes;

int bytesWritten=write(fd,reinterpret_cast(&buffers[bufIndex].data),bytesToWrite);

if(bytesWritten!=bytesToWrite){

std::cerr << "ERROR: Unable to write to disk" << std::endl;

exit(EXIT_FAILURE);

}

// Mark buffer as written

buffers[bufIndex].NeedsWriteToDisk=0;

// Move to next buffer

bufIndex=(bufIndex+1)%NBUFS;

}

}

// This is the FifoWriter thread.

// It loops through all the buffers waiting in turn for each one to become ready

// then writes it to the Fifo, flushes it for reduced lag, and marks the buffer

// as written before moving to next one. Errors are ignored.

void FifoWriter(){

int bufIndex=0;

bool Error=false;

while(!Error){

// Wait for buffer to be filled by main thread

while(buffers[bufIndex].NeedsWriteToFifo!=1){

// std::this_thread::sleep_for(std::chrono::milliseconds(1));

}

// Write to fifo

int bytesToWrite=buffers[bufIndex].bytes;

int bytesWritten=write(STDOUT_FILENO,reinterpret_cast(&buffers[bufIndex].data),bytesToWrite);

if(bytesWritten!=bytesToWrite){

std::cerr << "ERROR: Unable to write to fifo" << std::endl;

}

// Try to reduce lag

fflush(stdout);

// Mark buffer as written

buffers[bufIndex].NeedsWriteToFifo=0;

// Move to next buffer

bufIndex=(bufIndex+1)%NBUFS;

}

}

int main(int argc, char *argv[])

{

int bufIndex=0;

if(argc!=2){

std::cerr << "ERROR: Usage " << argv[0] << " filename" << std::endl;

exit(EXIT_FAILURE);

}

char * filename = argv[1];

// Start disk and fifo writing threads in parallel

std::thread tDiskWriter(DiskWriter,filename);

std::thread tFifoWriter(FifoWriter);

bool Error=false;

// Continuously fill buffers from "raspivid" on stdin. Mark as full and

// needing output to disk and fifo before moving to next buffer.

while(!Error)

{

// Check disk writer is not behind before re-using buffer

if(buffers[bufIndex].NeedsWriteToDisk==1){

std::cerr << "ERROR: Disk writer is behind by " << NBUFS << " buffers" << std::endl;

}

// Check fifo writer is not behind before re-using buffer

if(buffers[bufIndex].NeedsWriteToFifo==1){

std::cerr << "ERROR: Fifo writer is behind by " << NBUFS << " buffers" << std::endl;

}

// Read from STDIN till buffer is pretty full

int bytes;

int totalBytes=0;

int bytesToRead=BUFSZ;

unsigned char* ptr=reinterpret_cast(&buffers[bufIndex].data);

while(totalBytes

bytes = read(STDIN_FILENO,ptr,bytesToRead);

if(bytes<=0){

Error=true;

break;

}

ptr+=bytes;

totalBytes+=bytes;

bytesToRead-=bytes;

}

// Signal buffer ready for writing

buffers[bufIndex].bytes=totalBytes;

buffers[bufIndex].NeedsWriteToDisk=1;

buffers[bufIndex].NeedsWriteToFifo=1;

// Move to next buffer

bufIndex=(bufIndex+1)%NBUFS;

}

}

總結

以上是生活随笔為你收集整理的linux fifo数据流,在linux / bash中使用非阻塞FIFO流式传输视频(示例代码)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。