external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

SRS5: Improve file writer performance by fwrite with cache. v5.0.133 (#3308)

Browse source 
* SrsFileWriter leverages libc buffer to boost dvr write speed.

* Refactor SrsFileWriter to use libc file functions mockable

* Add utest and refine code.

Co-authored-by: winlin <winlin@vip.126.com>

PICK 25eb21efe8
This commit is contained in:
stone 2023-01-08 12:05:37 +08:00 • committed by winlin
parent f06a2d61f7
commit 748aa8508f
9 changed files with 358 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trunk/src/kernel/srs_kernel_error.hpp
View file

@ -105,6 +105,8 @@
XX(ERROR_BACKTRACE_PARSE_NOT_SUPPORT , 1092, "BacktraceParseNotSupport", "Backtrace parse not supported") \
XX(ERROR_BACKTRACE_PARSE_OFFSET , 1093, "BacktraceParseOffset", "Parse backtrace offset failed") \
XX(ERROR_BACKTRACE_ADDR2LINE , 1094, "BacktraceAddr2Line", "Backtrace addr2line failed") \
XX(ERROR_SYSTEM_FILE_NOT_OPEN , 1095, "FileNotOpen", "File is not opened") \
XX(ERROR_SYSTEM_FILE_SETVBUF , 1096, "FileSetVBuf", "Failed to set file vbuf") \
/**************************************************/
/* RTMP protocol error. */

106
trunk/src/kernel/srs_kernel_file.cpp
View file

@ -26,32 +26,57 @@ srs_read_t _srs_read_fn = ::read;
srs_lseek_t _srs_lseek_fn = ::lseek;
srs_close_t _srs_close_fn = ::close;
srs_fopen_t _srs_fopen_fn = ::fopen;
srs_fwrite_t _srs_fwrite_fn = ::fwrite;
srs_fread_t _srs_fread_fn = ::fread;
srs_fseek_t _srs_fseek_fn = ::fseek;
srs_fclose_t _srs_fclose_fn = ::fclose;
srs_ftell_t _srs_ftell_fn = ::ftell;
srs_setvbuf_t _srs_setvbuf_fn = ::setvbuf;
SrsFileWriter::SrsFileWriter()
{
fd = -1;
fp_ = NULL;
buf_ = NULL;
}
SrsFileWriter::~SrsFileWriter()
{
close();
srs_freepa(buf_);
}
srs_error_t SrsFileWriter::set_iobuf_size(int size)
{
srs_error_t err = srs_success;
if (fp_ == NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_NOT_OPEN, "file %s is not opened", path_.c_str());
}
srs_freepa(buf_);
buf_ = size > 0 ? new char[size] : NULL;
int r0 = _srs_setvbuf_fn(fp_, buf_, _IOFBF, size);
if (r0) {
return srs_error_new(ERROR_SYSTEM_FILE_SETVBUF, "setvbuf err, file=%s, r0=%d", path_.c_str(), r0);
}
return err;
}
srs_error_t SrsFileWriter::open(string p)
{
srs_error_t err = srs_success;
if (fd > 0) {
if (fp_ != NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_ALREADY_OPENED, "file %s already opened", p.c_str());
}
int flags = O_CREAT|O_WRONLY|O_TRUNC;
mode_t mode = S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH;
if ((fd = _srs_open_fn(p.c_str(), flags, mode)) < 0) {
if ((fp_ = _srs_fopen_fn(p.c_str(), "wb")) == NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_OPENE, "open file %s failed", p.c_str());
}
path = p;
path_ = p;
return err;
}
@ -60,70 +85,70 @@ srs_error_t SrsFileWriter::open_append(string p)
{
srs_error_t err = srs_success;
if (fd > 0) {
return srs_error_new(ERROR_SYSTEM_FILE_ALREADY_OPENED, "file %s already opened", path.c_str());
if (fp_ != NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_ALREADY_OPENED, "file %s already opened", p.c_str());
}
int flags = O_CREAT|O_APPEND|O_WRONLY;
mode_t mode = S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH;
if ((fd = _srs_open_fn(p.c_str(), flags, mode)) < 0) {
if ((fp_ = _srs_fopen_fn(p.c_str(), "ab")) == NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_OPENE, "open file %s failed", p.c_str());
}
path = p;
path_ = p;
return err;
}
void SrsFileWriter::close()
{
if (fd < 0) {
if (fp_ == NULL) {
return;
}
if (_srs_close_fn(fd) < 0) {
srs_warn("close file %s failed", path.c_str());
if (_srs_fclose_fn(fp_) < 0) {
srs_warn("close file %s failed", path_.c_str());
}
fd = -1;
fp_ = NULL;
return;
}
bool SrsFileWriter::is_open()
{
return fd > 0;
return fp_ != NULL;
}
void SrsFileWriter::seek2(int64_t offset)
{
off_t r0 = _srs_lseek_fn(fd, (off_t)offset, SEEK_SET);
srs_assert(is_open());
int r0 = _srs_fseek_fn(fp_, (long)offset, SEEK_SET);
srs_assert(r0 != -1);
}
int64_t SrsFileWriter::tellg()
{
return (int64_t)_srs_lseek_fn(fd, 0, SEEK_CUR);
srs_assert(is_open());
return _srs_ftell_fn(fp_);
}
srs_error_t SrsFileWriter::write(void* buf, size_t count, ssize_t* pnwrite)
{
srs_error_t err = srs_success;
ssize_t nwrite;
// TODO: FIXME: use st_write.
#ifdef _WIN32
if ((nwrite = ::_write(fd, buf, (unsigned int)count)) < 0) {
#else
if ((nwrite = _srs_write_fn(fd, buf, count)) < 0) {
#endif
return srs_error_new(ERROR_SYSTEM_FILE_WRITE, "write to file %s failed", path.c_str());
if (fp_ == NULL) {
return srs_error_new(ERROR_SYSTEM_FILE_NOT_OPEN, "file %s is not opened", path_.c_str());
}
size_t n = _srs_fwrite_fn(buf, 1, count, fp_);
if (n != count) {
return srs_error_new(ERROR_SYSTEM_FILE_WRITE, "write to file %s failed", path_.c_str());
}
if (pnwrite != NULL) {
*pnwrite = nwrite;
*pnwrite = (ssize_t)n;
}
return err;
}
@ -136,29 +161,30 @@ srs_error_t SrsFileWriter::writev(const iovec* iov, int iovcnt, ssize_t* pnwrite
const iovec* piov = iov + i;
ssize_t this_nwrite = 0;
if ((err = write(piov->iov_base, piov->iov_len, &this_nwrite)) != srs_success) {
return srs_error_wrap(err, "write file");
return srs_error_wrap(err, "writev");
}
nwrite += this_nwrite;
}
if (pnwrite) {
*pnwrite = nwrite;
}
return err;
}
srs_error_t SrsFileWriter::lseek(off_t offset, int whence, off_t* seeked)
{
off_t sk = _srs_lseek_fn(fd, offset, whence);
if (sk < 0) {
srs_assert(is_open());
if (_srs_fseek_fn(fp_, (long)offset, whence) == -1) {
return srs_error_new(ERROR_SYSTEM_FILE_SEEK, "seek file");
}
if (seeked) {
*seeked = sk;
*seeked = _srs_ftell_fn(fp_);
}
return srs_success;
}

21
trunk/src/kernel/srs_kernel_file.hpp
View file

@ -26,12 +26,18 @@ class SrsFileReader;
class SrsFileWriter : public ISrsWriteSeeker
{
private:
std::string path;
int fd;
std::string path_;
FILE *fp_;
char *buf_;
public:
SrsFileWriter();
virtual ~SrsFileWriter();
public:
/**
* set io buf size
*/
virtual srs_error_t set_iobuf_size(int size);
/**
* open file writer, in truncate mode.
* @param p a string indicates the path of file to open.
@ -110,5 +116,16 @@ typedef ssize_t (*srs_read_t)(int fildes, void* buf, size_t nbyte);
typedef off_t (*srs_lseek_t)(int fildes, off_t offset, int whence);
typedef int (*srs_close_t)(int fildes);
typedef FILE* (*srs_fopen_t)(const char* path, const char* mode);
typedef size_t (*srs_fwrite_t)(const void* ptr, size_t size, size_t nitems,
FILE* stream);
typedef size_t (*srs_fread_t)(void* ptr, size_t size, size_t nitems,
FILE* stream);
typedef int (*srs_fseek_t)(FILE* stream, long offset, int whence);
typedef int (*srs_fclose_t)(FILE* stream);
typedef long (*srs_ftell_t)(FILE* stream);
typedef int (*srs_setvbuf_t)(FILE* stream, char* buf, int type, size_t size);
#endif