/* job.cpp * Copyright (C) 2003-2005 Tommi Maekitalo * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public License as * published by the Free Software Foundation; either version 2 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, but * is provided AS IS, WITHOUT ANY WARRANTY; without even the implied * warranty of MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, and * NON-INFRINGEMENT. See the GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * */ #include "tnt/job.h" #include #include #include #include #include log_define("tntnet.job") namespace tnt { unsigned Job::socket_read_timeout = 200; unsigned Job::socket_write_timeout = 10000; unsigned Job::keepalive_max = 1000; unsigned Job::socket_buffer_size = 16384; Job::~Job() { } void Job::clear() { parser.reset(); request.clear(); touch(); } int Job::msecToTimeout(time_t currentTime) const { return (lastAccessTime - currentTime + 1) * 1000 + getKeepAliveTimeout() - getSocketReadTimeout(); } unsigned Job::getKeepAliveTimeout() { return HttpReply::getKeepAliveTimeout(); } //////////////////////////////////////////////////////////////////////// // Tcpjob // void Tcpjob::accept(const cxxtools::net::Server& listener) { log_debug("accept"); socket.accept(listener); struct sockaddr_storage s = socket.getSockAddr(); struct sockaddr_storage sockaddr; memcpy(&sockaddr, &s, sizeof(sockaddr)); char buffer[INET6_ADDRSTRLEN]; log_debug("connection accepted from " << inet_ntop(AF_INET6, &(socket.getPeeraddr()), buffer, sizeof(buffer))); getRequest().setPeerAddr(socket.getPeeraddr()); getRequest().setServerAddr(sockaddr); getRequest().setSsl(false); } std::iostream& Tcpjob::getStream() { return socket; } int Tcpjob::getFd() const { return socket.getFd(); } void Tcpjob::setRead() { socket.setTimeout(getSocketReadTimeout()); } void Tcpjob::setWrite() { socket.setTimeout(getSocketWriteTimeout()); } #ifdef USE_SSL //////////////////////////////////////////////////////////////////////// // SslTcpjob // void SslTcpjob::accept(const SslServer& listener) { log_debug("accept (ssl)"); socket.accept(listener); log_debug("connection accepted (ssl)"); struct sockaddr_storage s = socket.getSockAddr(); struct sockaddr_storage sockaddr; memcpy(&sockaddr, &s, sizeof(sockaddr)); getRequest().setPeerAddr(socket.getPeeraddr()); getRequest().setServerAddr(sockaddr); getRequest().setSsl(true); setRead(); } std::iostream& SslTcpjob::getStream() { return socket; } int SslTcpjob::getFd() const { return socket.getFd(); } void SslTcpjob::setRead() { socket.setTimeout(getSocketReadTimeout()); } void SslTcpjob::setWrite() { socket.setTimeout(getSocketWriteTimeout()); } #endif // USE_SSL #ifdef USE_GNUTLS //////////////////////////////////////////////////////////////////////// // GnuTlsTcpjob // void GnuTlsTcpjob::accept(const GnuTlsServer& listener) { log_debug("accept (ssl)"); socket.accept(listener); log_debug("connection accepted (ssl)"); struct sockaddr_storage s = socket.getSockAddr(); struct sockaddr_storage sockaddr; memcpy(&sockaddr, &s, sizeof(sockaddr)); getRequest().setPeerAddr(socket.getPeeraddr()); getRequest().setServerAddr(sockaddr); getRequest().setSsl(true); setRead(); } std::iostream& GnuTlsTcpjob::getStream() { return socket; } int GnuTlsTcpjob::getFd() const { return socket.getFd(); } void GnuTlsTcpjob::setRead() { socket.setTimeout(getSocketReadTimeout()); } void GnuTlsTcpjob::setWrite() { socket.setTimeout(getSocketWriteTimeout()); } #endif // USE_GNUTLS ////////////////////////////////////////////////////////////////////// // Jobqueue // void Jobqueue::put(JobPtr j) { log_debug("Jobqueue::put"); j->touch(); cxxtools::MutexLock lock(mutex); if (capacity > 0) { while (jobs.size() >= capacity) { log_warn("Jobqueue full"); notFull.wait(lock); } } jobs.push_back(j); if (waitThreads == 0) { log_info("no waiting threads left"); noWaitThreads.signal(); } notEmpty.signal(); } Jobqueue::JobPtr Jobqueue::get() { // wait, until a job is available ++waitThreads; cxxtools::MutexLock lock(mutex); while (jobs.empty()) notEmpty.wait(lock); --waitThreads; log_debug("Jobqueue: fetch job " << waitThreads << " waiting threads left"); // take next job (queue is locked) JobPtr j = jobs.front(); jobs.pop_front(); // if there are more jobs, wake onther thread if (!jobs.empty()) notEmpty.signal(); notFull.signal(); return j; } }