Skip to content

Z
ZLMediaKit
Commit 6023885e by xiongziliang
Options

抽象整理HttpBody类，fread模式下使用循环池提升内存性能

parent 144fb203
正在显示 包含 290 行增加 和 219 行删除
ZLToolKit @ 8d1681b5
Subproject commit 4a6029b74b4f2339e32b8c546388de51e4ec1bcb Subproject commit 8d1681b5bb247e7f47ae0f8c414f6eeb376b742b
src/Http/FileReader.h deleted 100644 → 0
/*
* MIT License
*
* Copyright (c) 2016-2019 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_FILEREADER_H
#define ZLMEDIAKIT_FILEREADER_H
#include <stdlib.h>
#include <memory>
#include "Network/Buffer.h"
using namespace std;
using namespace toolkit;
class FileReader {
public:
typedef std::shared_ptr<FileReader> Ptr;
FileReader(const std::shared_ptr<FILE> &fp,uint64_t offset,uint64_t max_size);
~FileReader();
Buffer::Ptr read(int size);
private:
std::shared_ptr<FILE> _fp;
uint64_t _max_size;
uint64_t _read_offset = 0;
std::shared_ptr<char> _map_addr;
};
#endif //ZLMEDIAKIT_FILEREADER_H
src/Http/FileReader.cpp → src/Http/HttpBody.cpp
...@@ -24,11 +24,11 @@ ...@@ -24,11 +24,11 @@
* SOFTWARE. * SOFTWARE.
*/ */
#include "FileReader.h" #include "HttpBody.h"
#include "Util/util.h" #include "Util/util.h"
#include "Util/uv_errno.h" #include "Util/uv_errno.h"
#include "Util/logger.h" #include "Util/logger.h"
#include "Common/config.h" #include "HttpClient.h"
#ifndef _WIN32 #ifndef _WIN32
#include <sys/mman.h> #include <sys/mman.h>
#endif #endif
...@@ -37,19 +37,41 @@ ...@@ -37,19 +37,41 @@
#define ENABLE_MMAP #define ENABLE_MMAP
#endif #endif
FileReader::FileReader(const std::shared_ptr<FILE> &fp, uint64_t offset, uint64_t max_size) { namespace mediakit {
HttpStringBody::HttpStringBody(const string &str){
_str = str;
}
uint64_t HttpStringBody::remainSize() {
return _str.size() - _offset;
}
Buffer::Ptr HttpStringBody::readData(uint32_t size) {
size = MIN(remainSize(),size);
if(!size){
//没有剩余字节了
return nullptr;
}
auto ret = std::make_shared<BufferString>(_str,_offset,size);
_offset += size;
return ret;
}
//////////////////////////////////////////////////////////////////
HttpFileBody::HttpFileBody(const std::shared_ptr<FILE> &fp, uint64_t offset, uint64_t max_size) {
_fp = fp; _fp = fp;
_max_size = max_size; _max_size = max_size;
#ifdef ENABLE_MMAP #ifdef ENABLE_MMAP
do { do {
int fd = fileno(fp.get()); int fd = fileno(fp.get());
if (fd < 0) { if (fd < 0) {
WarnL << "fileno失败:" << get_uv_errmsg(false); WarnL << "fileno failed:" << get_uv_errmsg(false);
break; break;
} }
auto ptr = (char *) mmap(NULL, max_size, PROT_READ, MAP_SHARED, fd, offset); auto ptr = (char *) mmap(NULL, max_size, PROT_READ, MAP_SHARED, fd, offset);
if (!ptr) { if (!ptr) {
WarnL << "mmap失败:" << get_uv_errmsg(false); WarnL << "mmap failed:" << get_uv_errmsg(false);
break; break;
} }
_map_addr.reset(ptr,[max_size](char *ptr){ _map_addr.reset(ptr,[max_size](char *ptr){
...@@ -57,15 +79,12 @@ FileReader::FileReader(const std::shared_ptr<FILE> &fp, uint64_t offset, uint64_ ...@@ -57,15 +79,12 @@ FileReader::FileReader(const std::shared_ptr<FILE> &fp, uint64_t offset, uint64_
}); });
} while (false); } while (false);
#endif #endif
if(!_map_addr){ if(!_map_addr && offset){
//未映射 //未映射,那么fseek设置偏移量
fseek(fp.get(), offset, SEEK_SET); fseek(fp.get(), offset, SEEK_SET);
} }
} }
FileReader::~FileReader() {
}
class BufferMmap : public Buffer{ class BufferMmap : public Buffer{
public: public:
...@@ -88,33 +107,131 @@ private: ...@@ -88,33 +107,131 @@ private:
char *_data; char *_data;
uint32_t _size; uint32_t _size;
}; };
Buffer::Ptr FileReader::read(int size) {
if(_read_offset >= _max_size){ uint64_t HttpFileBody::remainSize() {
//文件读完了 return _max_size - _offset;
}
Buffer::Ptr HttpFileBody::readData(uint32_t size) {
size = MIN(remainSize(),size);
if(!size){
//没有剩余字节了
return nullptr; return nullptr;
} }
int iReq = MIN(size,_max_size - _read_offset);
if(!_map_addr){ if(!_map_addr){
//映射失败，fread模式 //fread模式
int iRead; int iRead;
auto ret = std::make_shared<BufferRaw>(iReq + 1); auto ret = _pool.obtain();
ret->setCapacity(size + 1);
do{ do{
iRead = fread(ret->data(), 1, iReq, _fp.get()); iRead = fread(ret->data(), 1, size, _fp.get());
}while(-1 == iRead && UV_EINTR == get_uv_error(false)); }while(-1 == iRead && UV_EINTR == get_uv_error(false));
if(iRead > 0){ if(iRead > 0){
//读到数据了 //读到数据了
ret->setSize(iRead); ret->setSize(iRead);
_read_offset += iRead; _offset += iRead;
return std::move(ret); return std::move(ret);
} }
//没有剩余数据 //读取文件异常，文件真实长度小于声明长度
_read_offset = _max_size; _offset = _max_size;
WarnL << "read file err:" << get_uv_errmsg();
return nullptr; return nullptr;
} }
auto ret = std::make_shared<BufferMmap>(_map_addr,_read_offset,iReq); //mmap模式
_read_offset += iReq; auto ret = std::make_shared<BufferMmap>(_map_addr,_offset,size);
_offset += size;
return std::move(ret); return std::move(ret);
} }
//////////////////////////////////////////////////////////////////
HttpMultiFormBody::HttpMultiFormBody(const HttpArgs &args,const string &filePath,const string &boundary){
std::shared_ptr<FILE> fp(fopen(filePath.data(), "rb"), [](FILE *fp) {
if(fp){
fclose(fp);
}
});
if(!fp){
throw std::invalid_argument(StrPrinter << "open file failed：" << filePath << " " << get_uv_errmsg());
}
_fileBody = std::make_shared<HttpFileBody>(fp, 0, fileSize(fp.get()));
auto fileName = filePath;
auto pos = filePath.rfind('/');
if(pos != string::npos){
fileName = filePath.substr(pos + 1);
}
_bodyPrefix = multiFormBodyPrefix(args,boundary,fileName);
_bodySuffix = multiFormBodySuffix(boundary);
_totalSize = _bodyPrefix.size() + _bodySuffix.size() + _fileBody->remainSize();
}
uint64_t HttpMultiFormBody::remainSize() {
return _totalSize - _offset;
}
Buffer::Ptr HttpMultiFormBody::readData(uint32_t size){
if(_bodyPrefix.size()){
auto ret = std::make_shared<BufferString>(_bodyPrefix);
_offset += _bodyPrefix.size();
_bodyPrefix.clear();
return ret;
}
if(_fileBody->remainSize()){
auto ret = _fileBody->readData(size);
if(!ret){
//读取文件出现异常，提前中断
_offset = _totalSize;
}else{
_offset += ret->size();
}
return ret;
}
if(_bodySuffix.size()){
auto ret = std::make_shared<BufferString>(_bodySuffix);
_offset = _totalSize;
_bodySuffix.clear();
return ret;
}
return nullptr;
}
string HttpMultiFormBody::multiFormBodySuffix(const string &boundary){
string MPboundary = string("--") + boundary;
string endMPboundary = MPboundary + "--";
_StrPrinter body;
body << "\r\n" << endMPboundary;
return body;
}
uint64_t HttpMultiFormBody::fileSize(FILE *fp) {
auto current = ftell(fp);
fseek(fp,0L,SEEK_END); /* 定位到文件末尾 */
auto end = ftell(fp); /* 得到文件大小 */
fseek(fp,current,SEEK_SET);
return end - current;
}
string HttpMultiFormBody::multiFormContentType(const string &boundary){
return StrPrinter << "multipart/form-data; boundary=" << boundary;
}
string HttpMultiFormBody::multiFormBodyPrefix(const HttpArgs &args,const string &boundary,const string &fileName){
string MPboundary = string("--") + boundary;
_StrPrinter body;
for(auto &pr : args){
body << MPboundary << "\r\n";
body << "Content-Disposition: form-data; name=\"" << pr.first << "\"\r\n\r\n";
body << pr.second << "\r\n";
}
body << MPboundary << "\r\n";
body << "Content-Disposition: form-data; name=\"" << "file" << "\";filename=\"" << fileName << "\"\r\n";
body << "Content-Type: application/octet-stream\r\n\r\n" ;
return body;
}
}//namespace mediakit
src/Http/HttpBody.h 0 → 100644
/*
* MIT License
*
* Copyright (c) 2016-2019 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_FILEREADER_H
#define ZLMEDIAKIT_FILEREADER_H
#include <stdlib.h>
#include <memory>
#include "Network/Buffer.h"
#include "Util/ResourcePool.h"
#include "Util/logger.h"
using namespace std;
using namespace toolkit;
#ifndef MIN
#define MIN(a,b) ((a) < (b) ? (a) : (b) )
#endif //MIN
namespace mediakit {
/**
* http content部分基类定义
*/
class HttpBody{
public:
typedef std::shared_ptr<HttpBody> Ptr;
HttpBody(){}
virtual ~HttpBody(){}
/**
* 剩余数据大小
*/
virtual uint64_t remainSize() = 0;
/**
* 读取一定字节数，返回大小可能小于size
* @param size 请求大小
* @return 字节对象
*/
virtual Buffer::Ptr readData(uint32_t size) = 0;
};
/**
* string类型的content
*/
class HttpStringBody : public HttpBody{
public:
typedef std::shared_ptr<HttpStringBody> Ptr;
HttpStringBody(const string &str);
virtual ~HttpStringBody(){}
uint64_t remainSize() override ;
Buffer::Ptr readData(uint32_t size) override ;
private:
mutable string _str;
uint64_t _offset = 0;
};
/**
* 文件类型的content
*/
class HttpFileBody : public HttpBody{
public:
typedef std::shared_ptr<HttpFileBody> Ptr;
/**
* 构造函数
* @param fp 文件句柄，文件的偏移量必须为0
* @param offset 相对文件头的偏移量
* @param max_size 最大读取字节数，未判断是否大于文件真实大小
*/
HttpFileBody(const std::shared_ptr<FILE> &fp,uint64_t offset,uint64_t max_size);
~HttpFileBody(){};
uint64_t remainSize() override ;
Buffer::Ptr readData(uint32_t size) override;
private:
std::shared_ptr<FILE> _fp;
uint64_t _max_size;
uint64_t _offset = 0;
std::shared_ptr<char> _map_addr;
ResourcePool<BufferRaw> _pool;
};
class HttpArgs;
/**
* http MultiForm 方式提交的http content
*/
class HttpMultiFormBody : public HttpBody {
public:
typedef std::shared_ptr<HttpMultiFormBody> Ptr;
/**
* 构造函数
* @param args http提交参数列表
* @param filePath 文件路径
* @param boundary boundary字符串
*/
HttpMultiFormBody(const HttpArgs &args,const string &filePath,const string &boundary = "0xKhTmLbOuNdArY");
virtual ~HttpMultiFormBody(){}
uint64_t remainSize() override ;
Buffer::Ptr readData(uint32_t size) override;
public:
static string multiFormBodyPrefix(const HttpArgs &args,const string &boundary,const string &fileName);
static string multiFormBodySuffix(const string &boundary);
static uint64_t fileSize(FILE *fp);
static string multiFormContentType(const string &boundary);
private:
string _bodyPrefix;
string _bodySuffix;
uint64_t _offset = 0;
uint64_t _totalSize;
HttpFileBody::Ptr _fileBody;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_FILEREADER_H
src/Http/HttpClient.cpp
...@@ -242,8 +242,9 @@ void HttpClient::onRecvContent(const char *data, uint64_t len) { ...@@ -242,8 +242,9 @@ void HttpClient::onRecvContent(const char *data, uint64_t len) {
void HttpClient::onFlush() { void HttpClient::onFlush() {
_aliveTicker.resetTime(); _aliveTicker.resetTime();
GET_CONFIG(uint32_t,sendBufSize,Http::kSendBufSize);
while (_body && _body->remainSize() && !isSocketBusy()) { while (_body && _body->remainSize() && !isSocketBusy()) {
auto buffer = _body->readData(); auto buffer = _body->readData(sendBufSize);
if (!buffer) { if (!buffer) {
//数据发送结束或读取数据异常 //数据发送结束或读取数据异常
break; break;
......
src/Http/HttpClient.h
...@@ -39,7 +39,7 @@ ...@@ -39,7 +39,7 @@
#include "HttpCookie.h" #include "HttpCookie.h"
#include "HttpChunkedSplitter.h" #include "HttpChunkedSplitter.h"
#include "strCoding.h" #include "strCoding.h"
#include "HttpBody.h"
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
...@@ -64,145 +64,6 @@ public: ...@@ -64,145 +64,6 @@ public:
} }
}; };
class HttpBody{
public:
typedef std::shared_ptr<HttpBody> Ptr;
HttpBody(){}
virtual ~HttpBody(){}
//剩余数据大小
virtual uint64_t remainSize() = 0;
virtual Buffer::Ptr readData() = 0;
};
class HttpStringBody : public HttpBody{
public:
typedef std::shared_ptr<HttpStringBody> Ptr;
HttpStringBody(const string &str){
_str = str;
}
virtual ~HttpStringBody(){}
uint64_t remainSize() override {
return _str.size();
}
Buffer::Ptr readData() override {
auto ret = std::make_shared<BufferString>(_str);
_str.clear();
return ret;
}
private:
mutable string _str;
};
class HttpMultiFormBody : public HttpBody {
public:
typedef std::shared_ptr<HttpMultiFormBody> Ptr;
template<typename MapType>
HttpMultiFormBody(const MapType &args,const string &filePath,const string &boundary,uint32_t sliceSize = 4 * 1024){
_fp = fopen(filePath.data(),"rb");
if(!_fp){
throw std::invalid_argument(StrPrinter << "打开文件失败：" << filePath << " " << get_uv_errmsg());
}
auto fileName = filePath;
auto pos = filePath.rfind('/');
if(pos != string::npos){
fileName = filePath.substr(pos + 1);
}
_bodyPrefix = multiFormBodyPrefix(args,boundary,fileName);
_bodySuffix = multiFormBodySuffix(boundary);
_totalSize = _bodyPrefix.size() + _bodySuffix.size() + fileSize(_fp);
_sliceSize = sliceSize;
}
virtual ~HttpMultiFormBody(){
fclose(_fp);
}
uint64_t remainSize() override {
return _totalSize - _offset;
}
Buffer::Ptr readData() override{
if(_bodyPrefix.size()){
auto ret = std::make_shared<BufferString>(_bodyPrefix);
_offset += _bodyPrefix.size();
_bodyPrefix.clear();
return ret;
}
if(0 == feof(_fp)){
auto ret = std::make_shared<BufferRaw>(_sliceSize);
//读文件
int size;
do{
size = fread(ret->data(),1,_sliceSize,_fp);
}while(-1 == size && UV_EINTR == get_uv_error(false));
if(size == -1){
_offset = _totalSize;
WarnL << "fread failed:" << get_uv_errmsg();
return nullptr;
}
_offset += size;
ret->setSize(size);
return ret;
}
if(_bodySuffix.size()){
auto ret = std::make_shared<BufferString>(_bodySuffix);
_offset = _totalSize;
_bodySuffix.clear();
return ret;
}
return nullptr;
}
public:
template<typename MapType>
static string multiFormBodyPrefix(const MapType &args,const string &boundary,const string &fileName){
string MPboundary = string("--") + boundary;
_StrPrinter body;
for(auto &pr : args){
body << MPboundary << "\r\n";
body << "Content-Disposition: form-data; name=\"" << pr.first << "\"\r\n\r\n";
body << pr.second << "\r\n";
}
body << MPboundary << "\r\n";
body << "Content-Disposition: form-data; name=\"" << "file" << "\";filename=\"" << fileName << "\"\r\n";
body << "Content-Type: application/octet-stream\r\n\r\n" ;
return body;
}
static string multiFormBodySuffix(const string &boundary){
string MPboundary = string("--") + boundary;
string endMPboundary = MPboundary + "--";
_StrPrinter body;
body << "\r\n" << endMPboundary;
return body;
}
static uint64_t fileSize(FILE *fp) {
auto current = ftell(fp);
fseek(fp,0L,SEEK_END); /* 定位到文件末尾 */
auto end = ftell(fp); /* 得到文件大小 */
fseek(fp,current,SEEK_SET);
return end - current;
}
static string multiFormContentType(const string &boundary){
return StrPrinter << "multipart/form-data; boundary=" << boundary;
}
private:
FILE *_fp;
string _bodyPrefix;
string _bodySuffix;
uint64_t _offset = 0;
uint64_t _totalSize;
uint32_t _sliceSize;
};
class HttpClient : public TcpClient , public HttpRequestSplitter class HttpClient : public TcpClient , public HttpRequestSplitter
{ {
public: public:
......
src/Http/HttpSession.cpp
...@@ -44,7 +44,7 @@ ...@@ -44,7 +44,7 @@
#include "Util/base64.h" #include "Util/base64.h"
#include "Util/SHA1.h" #include "Util/SHA1.h"
#include "Rtmp/utils.h" #include "Rtmp/utils.h"
#include "FileReader.h" #include "HttpBody.h"
using namespace toolkit; using namespace toolkit;
namespace mediakit { namespace mediakit {
...@@ -618,10 +618,10 @@ void HttpSession::Handle_Req_GET(int64_t &content_len) { ...@@ -618,10 +618,10 @@ void HttpSession::Handle_Req_GET(int64_t &content_len) {
} }
//回复Content部分 //回复Content部分
GET_CONFIG(uint32_t,sendBufSize,Http::kSendBufSize); GET_CONFIG(uint32_t,sendBufSize,Http::kSendBufSize);
FileReader::Ptr fileReader = std::make_shared<FileReader>(pFilePtr,iRangeStart,iRangeEnd - iRangeStart + 1); HttpBody::Ptr fileBody = std::make_shared<HttpFileBody>(pFilePtr,iRangeStart,iRangeEnd - iRangeStart + 1);
weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this());
auto onFlush = [fileReader,bClose,weakSelf]() { auto onFlush = [fileBody,bClose,weakSelf]() {
auto strongSelf = weakSelf.lock(); auto strongSelf = weakSelf.lock();
if(!strongSelf){ if(!strongSelf){
//本对象已经销毁 //本对象已经销毁
...@@ -631,7 +631,7 @@ void HttpSession::Handle_Req_GET(int64_t &content_len) { ...@@ -631,7 +631,7 @@ void HttpSession::Handle_Req_GET(int64_t &content_len) {
//更新超时计时器 //更新超时计时器
strongSelf->_ticker.resetTime(); strongSelf->_ticker.resetTime();
//读取文件 //读取文件
auto sendBuf = fileReader->read(sendBufSize); auto sendBuf = fileBody->readData(sendBufSize);
if (!sendBuf) { if (!sendBuf) {
//文件读完 //文件读完
if(strongSelf->isSocketBusy()){ if(strongSelf->isSocketBusy()){
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论