在linux / bash中使用非阻塞FIFO流式传输视频
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了在linux / bash中使用非阻塞FIFO流式传输视频相关的知识,希望对你有一定的参考价值。
我正在努力实现以下目标:
- 将我的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:
Linux non-blocking fifo (on demand logging)
它现在阻止我的$ 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 <iostream>
#include <chrono>
#include <thread>
#include <vector>
#include <unistd.h>
#include <atomic>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#define BUFSZ 65536
#define NBUFS 64
class Buffer{
public:
int bytes=0;
std::atomic<int> NeedsWriteToDisk{0};
std::atomic<int> NeedsWriteToFifo{0};
unsigned char data[BUFSZ];
};
std::vector<Buffer> 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<unsigned char*>(&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<unsigned char*>(&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<unsigned char*>(&buffers[bufIndex].data);
while(totalBytes<(BUFSZ*.75)){
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 / bash中使用非阻塞FIFO流式传输视频的主要内容,如果未能解决你的问题,请参考以下文章