Ai
1 Star 0 Fork 0

wangcichen/urbackup_backend

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
AcceptThread.cpp 8.21 KB
一键复制 编辑 原始数据 按行查看 历史
Martin 提交于 2020-07-06 03:18 +08:00 . Implement connection establishment via web sockets
/*************************************************************************
* UrBackup - Client/Server backup system
* Copyright (C) 2011-2016 Martin Raiber
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
**************************************************************************/
#include "vld.h"
#include "AcceptThread.h"
#include "Server.h"
#include "stringtools.h"
#include "SelectThread.h"
#include "Client.h"
#include <memory.h>
#ifndef _WIN32
#include <errno.h>
#endif
extern bool run;
namespace
{
bool prepareSocket(SOCKET s)
{
#if !defined(_WIN32) && !defined(SOCK_CLOEXEC)
fcntl(s, F_SETFD, fcntl(s, F_GETFD, 0) | FD_CLOEXEC);
#endif
int optval = 1;
int rc = setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(int));
if (rc == SOCKET_ERROR)
{
Server->Log("Failed setting SO_REUSEADDR", LL_ERROR);
return false;
}
#ifdef __APPLE__
optval = 1;
setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, (void*)&optval, sizeof(optval));
#endif
return true;
}
}
OutputCallback::OutputCallback(SOCKET fd_)
{
fd=fd_;
}
OutputCallback::~OutputCallback()
{
closesocket(fd);
}
void OutputCallback::operator() (const void* buf, size_t count)
{
size_t sent=0;
do
{
int rc = send(fd, (const char*)buf+sent, static_cast<int>(count-sent), MSG_NOSIGNAL);
if (rc < 0)
{
int ec;
#ifdef _WIN32
ec=GetLastError();
#else
ec=errno;
#endif
Server->Log("Send failed in OutputCallback ec="+convert(ec), LL_INFO);
throw std::runtime_error("Send failed in OutputCallback");
}
else
{
sent+=rc;
}
}
while(sent<count);
}
CAcceptThread::CAcceptThread( unsigned int nWorkerThreadsPerMaster, unsigned short int uPort )
: error(false), s_v6(SOCKET_ERROR), s(SOCKET_ERROR)
{
WorkerThreadsPerMaster=nWorkerThreadsPerMaster;
bool has_ipv4 = Server->getServerParameter("disable_ipv4").empty();
bool has_ipv6 = Server->getServerParameter("disable_ipv6").empty();
if (!has_ipv4 && !has_ipv6)
{
Server->Log("CAcceptThread: IPv4 and IPv6 disabled", LL_ERROR);
error = true;
return;
}
if (has_ipv4)
{
if (!init_socket_v4(uPort))
{
if (Server->getServerParameter("ipv4_optional").empty()
|| !has_ipv6)
{
error = true;
return;
}
}
}
if (has_ipv6)
{
if (!init_socket_v6(uPort))
{
if (!Server->getServerParameter("ipv6_required").empty()
|| !has_ipv4)
{
error = true;
return;
}
}
}
Server->Log("Server started up successfully!",LL_INFO);
}
CAcceptThread::~CAcceptThread()
{
closesocket(s);
if (s_v6 != SOCKET_ERROR)
closesocket(s_v6);
Server->Log("Deleting SelectThreads..");
for(size_t i=0;i<SelectThreads.size();++i)
{
delete SelectThreads[i];
}
}
#ifndef _WIN32
void printLinError(void)
{
switch(errno)
{
case EWOULDBLOCK: Server->Log("Reason: EWOULDBLOCK", LL_ERROR); break;
case EBADF: Server->Log("Reason: EBADF", LL_ERROR); break;
case ECONNABORTED: Server->Log("Reason: ECONNABORTED", LL_ERROR); break;
case EINTR: Server->Log("Reason: EINTR", LL_ERROR); break;
case EINVAL: Server->Log("Reason: EINVAL", LL_ERROR); break;
case EMFILE: Server->Log("Reason: EMFILE", LL_ERROR); break;
case ENFILE: Server->Log("Reason: ENFILE", LL_ERROR); break;
case ENOTSOCK: Server->Log("Reason: ENOTSOCK", LL_ERROR); break;
case EFAULT: Server->Log("Reason: EFAULT", LL_ERROR); break;
case ENOBUFS: Server->Log("Reason: ENOBUFS", LL_ERROR); break;
case ENOMEM: Server->Log("Reason: ENOMEM", LL_ERROR); break;
case EPROTO: Server->Log("Reason: EPROTO", LL_ERROR); break;
case EPERM: Server->Log("Reason: EPERM", LL_ERROR); break;
}
}
#endif
void CAcceptThread::operator()(bool single)
{
do
{
socklen_t addrsize=sizeof(sockaddr_in);
#ifdef _WIN32
fd_set fdset;
FD_ZERO(&fdset);
if(s!=SOCKET_ERROR)
FD_SET(s, &fdset);
if(s_v6!=SOCKET_ERROR)
FD_SET(s_v6, &fdset);
timeval lon;
lon.tv_sec=1;
lon.tv_usec=0;
_i32 rc=select((int)s+1, &fdset, 0, 0, &lon);
#else
pollfd conn[2];
conn[0].fd = s;
conn[0].events = POLLIN;
conn[0].revents = 0;
conn[1].revents = 0;
nfds_t num_fds = 1;
if (s_v6 != SOCKET_ERROR)
{
conn[1].fd = s_v6;
conn[1].events = POLLIN;
num_fds = 2;
}
int rc = poll(conn, num_fds, 1000);
#endif
if (rc<0)
return;
if (rc > 0)
{
for (size_t n = 0; n < 2; ++n)
{
#ifdef _WIN32
SOCKET accept_socket = n == 0 ? s : s_v6;
if (accept_socket == SOCKET_ERROR)
continue;
if (!FD_ISSET(accept_socket, &fdset))
continue;
#else
if (conn[n].revents == 0)
continue;
SOCKET accept_socket = conn[n].fd;
#endif
sockaddr_in naddr;
SOCKET ns = ACCEPT_CLOEXEC(accept_socket, (sockaddr*)&naddr, &addrsize);
if (ns != SOCKET_ERROR)
{
#ifdef __APPLE__
int optval = 1;
setsockopt(ns, SOL_SOCKET, SO_NOSIGPIPE, (void*)&optval, sizeof(optval));
#endif
//Server->Log("New Connection incomming", LL_INFO);
OutputCallback *output = new OutputCallback(ns);
FCGIProtocolDriver *driver = new FCGIProtocolDriver(*output);
CClient *client = new CClient();
client->set(ns, output, driver);
AddToSelectThread(client);
}
else
{
Server->Log("Accepting client failed", LL_ERROR);
#ifndef _WIN32
printLinError();
#endif
Server->wait(1000);
}
}
}
}while(!single);
}
void CAcceptThread::AddToSelectThread(CClient *client)
{
for(size_t i=0;i<SelectThreads.size();++i)
{
if( SelectThreads[i]->FreeClients()>0 )
{
SelectThreads[i]->AddClient( client );
return;
}
}
CSelectThread *nt=new CSelectThread(WorkerThreadsPerMaster);
nt->AddClient( client );
SelectThreads.push_back( nt );
Server->createThread(nt, "fastcgi: accept");
}
bool CAcceptThread::init_socket_v4(unsigned short port)
{
int type = SOCK_STREAM;
#if !defined(_WIN32) && defined(SOCK_CLOEXEC)
type |= SOCK_CLOEXEC;
#endif
s = socket(AF_INET, type, 0);
if (s < 1)
{
Server->Log("Creating SOCKET failed. Port " + convert((int)port) + " may already be in use", LL_ERROR);
return false;
}
if (!prepareSocket(s))
{
closesocket(s);
s = SOCKET_ERROR;
return false;
}
sockaddr_in addr;
memset(&addr, 0, sizeof(sockaddr_in));
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = htonl(INADDR_ANY);
if (Server->getServerParameter("fastcgi_localhost_only") == "1")
{
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
}
int rc = bind(s, (sockaddr*)&addr, sizeof(addr));
if (rc == SOCKET_ERROR)
{
Server->Log("Failed binding SOCKET to Port " + convert(port), LL_ERROR);
closesocket(s);
s = SOCKET_ERROR;
return false;
}
listen(s, 10000);
return true;
}
bool CAcceptThread::init_socket_v6(unsigned short port)
{
int type = SOCK_STREAM;
#if !defined(_WIN32) && defined(SOCK_CLOEXEC)
type |= SOCK_CLOEXEC;
#endif
s_v6 = socket(AF_INET6, type, 0);
if (s_v6<1)
{
Server->Log("Creating ipv6 SOCKET failed. Port " + convert((int)port) + " may already be in use", LL_ERROR);
return false;
}
if (!prepareSocket(s_v6))
{
closesocket(s_v6);
s_v6 = SOCKET_ERROR;
return false;
}
int optval = 1;
setsockopt(s_v6, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&optval, sizeof(optval));
sockaddr_in6 addr = {};
addr.sin6_family = AF_INET6;
addr.sin6_port = htons(port);
addr.sin6_addr = in6addr_any;
if (Server->getServerParameter("fastcgi_localhost_only") == "1")
{
addr.sin6_addr = in6addr_loopback;
}
int rc = bind(s_v6, (sockaddr*)&addr, sizeof(addr));
if (rc == SOCKET_ERROR)
{
Server->Log("Failed binding ipv6 SOCKET to Port " + convert(port), LL_ERROR);
closesocket(s_v6);
s_v6 = SOCKET_ERROR;
return false;
}
listen(s_v6, 10000);
return true;
}
bool CAcceptThread::has_error(void)
{
return error;
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
C
1
https://gitee.com/wangcichen/urbackup_backend.git
git@gitee.com:wangcichen/urbackup_backend.git
wangcichen
urbackup_backend
urbackup_backend
dev

搜索帮助