LBRYCommunity/lbrycrd
1
0
Fork 0
Code Issues 29 Pull requests 1 Projects Releases 32 Packages Wiki Activity

httpserver: replace boost threads with std

Browse source 
along with mutex/condvar/bind/etc.

httpserver handles its own interruption, so there's no reason not to use std
threading.

While we're at it, may as well kill the BOOST_FOREACH's as well.
This commit is contained in:
Cory Fields 2016-07-28 18:31:25 -04:00
parent d3773ca9ae
commit 7e87033447
2 changed files with 22 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
src/httpserver.cpp
View file

@ -35,9 +35,6 @@
#endif
#endif
#include <boost/algorithm/string/case_conv.hpp> // for to_lower()
#include <boost/foreach.hpp>
/** Maximum size of http request (request line + headers) */
static const size_t MAX_HEADERS_SIZE = 8192;
@ -69,8 +66,8 @@ class WorkQueue
{
private:
/** Mutex protects entire object */
CWaitableCriticalSection cs;
CConditionVariable cond;
std::mutex cs;
std::condition_variable cond;
std::deque<std::unique_ptr<WorkItem>> queue;
bool running;
size_t maxDepth;
@ -83,12 +80,12 @@ private:
WorkQueue &wq;
ThreadCounter(WorkQueue &w): wq(w)
{
boost::lock_guard<boost::mutex> lock(wq.cs);
std::lock_guard<std::mutex> lock(wq.cs);
wq.numThreads += 1;
}
~ThreadCounter()
{
boost::lock_guard<boost::mutex> lock(wq.cs);
std::lock_guard<std::mutex> lock(wq.cs);
wq.numThreads -= 1;
wq.cond.notify_all();
}
@ -109,7 +106,7 @@ public:
/** Enqueue a work item */
bool Enqueue(WorkItem* item)
{
boost::unique_lock<boost::mutex> lock(cs);
std::unique_lock<std::mutex> lock(cs);
if (queue.size() >= maxDepth) {
return false;
}
@ -124,7 +121,7 @@ public:
while (running) {
std::unique_ptr<WorkItem> i;
{
boost::unique_lock<boost::mutex> lock(cs);
std::unique_lock<std::mutex> lock(cs);
while (running && queue.empty())
cond.wait(lock);
if (!running)
@ -138,14 +135,14 @@ public:
/** Interrupt and exit loops */
void Interrupt()
{
boost::unique_lock<boost::mutex> lock(cs);
std::unique_lock<std::mutex> lock(cs);
running = false;
cond.notify_all();
}
/** Wait for worker threads to exit */
void WaitExit()
{
boost::unique_lock<boost::mutex> lock(cs);
std::unique_lock<std::mutex> lock(cs);
while (numThreads > 0)
cond.wait(lock);
}
@ -153,7 +150,7 @@ public:
/** Return current depth of queue */
size_t Depth()
{
boost::unique_lock<boost::mutex> lock(cs);
std::unique_lock<std::mutex> lock(cs);
return queue.size();
}
};
@ -190,7 +187,7 @@ static bool ClientAllowed(const CNetAddr& netaddr)
{
if (!netaddr.IsValid())
return false;
BOOST_FOREACH (const CSubNet& subnet, rpc_allow_subnets)
for(const CSubNet& subnet : rpc_allow_subnets)
if (subnet.Match(netaddr))
return true;
return false;
@ -204,7 +201,7 @@ static bool InitHTTPAllowList()
rpc_allow_subnets.push_back(CSubNet("::1")); // always allow IPv6 localhost
if (mapMultiArgs.count("-rpcallowip")) {
const std::vector<std::string>& vAllow = mapMultiArgs["-rpcallowip"];
BOOST_FOREACH (std::string strAllow, vAllow) {
for (std::string strAllow : vAllow) {
CSubNet subnet(strAllow);
if (!subnet.IsValid()) {
uiInterface.ThreadSafeMessageBox(
@ -216,7 +213,7 @@ static bool InitHTTPAllowList()
}
}
std::string strAllowed;
BOOST_FOREACH (const CSubNet& subnet, rpc_allow_subnets)
for (const CSubNet& subnet : rpc_allow_subnets)
strAllowed += subnet.ToString() + " ";
LogPrint("http", "Allowing HTTP connections from: %s\n", strAllowed);
return true;
@ -439,7 +436,7 @@ bool InitHTTPServer()
return true;
}
boost::thread threadHTTP;
std::thread threadHTTP;
std::future<bool> threadResult;
bool StartHTTPServer()
@ -449,10 +446,10 @@ bool StartHTTPServer()
LogPrintf("HTTP: starting %d worker threads\n", rpcThreads);
std::packaged_task<bool(event_base*, evhttp*)> task(ThreadHTTP);
threadResult = task.get_future();
threadHTTP = boost::thread(std::bind(std::move(task), eventBase, eventHTTP));
threadHTTP = std::thread(std::move(task), eventBase, eventHTTP);
for (int i = 0; i < rpcThreads; i++) {
boost::thread rpc_worker(HTTPWorkQueueRun, workQueue);
std::thread rpc_worker(HTTPWorkQueueRun, workQueue);
rpc_worker.detach();
}
return true;
@ -463,7 +460,7 @@ void InterruptHTTPServer()
LogPrint("http", "Interrupting HTTP server\n");
if (eventHTTP) {
// Unlisten sockets
BOOST_FOREACH (evhttp_bound_socket *socket, boundSockets) {
for (evhttp_bound_socket *socket : boundSockets) {
evhttp_del_accept_socket(eventHTTP, socket);
}
// Reject requests on current connections
@ -520,7 +517,7 @@ static void httpevent_callback_fn(evutil_socket_t, short, void* data)
delete self;
}
HTTPEvent::HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const boost::function<void(void)>& handler):
HTTPEvent::HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const std::function<void(void)>& handler):
deleteWhenTriggered(deleteWhenTriggered), handler(handler)
{
ev = event_new(base, -1, 0, httpevent_callback_fn, this);
@ -602,7 +599,7 @@ void HTTPRequest::WriteReply(int nStatus, const std::string& strReply)
assert(evb);
evbuffer_add(evb, strReply.data(), strReply.size());
HTTPEvent* ev = new HTTPEvent(eventBase, true,
boost::bind(evhttp_send_reply, req, nStatus, (const char*)NULL, (struct evbuffer *)NULL));
std::bind(evhttp_send_reply, req, nStatus, (const char*)NULL, (struct evbuffer *)NULL));
ev->trigger(0);
replySent = true;
req = 0; // transferred back to main thread

10
src/httpserver.h
View file

@ -7,9 +7,7 @@
#include <string>
#include <stdint.h>
#include <boost/thread.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/function.hpp>
#include <functional>
static const int DEFAULT_HTTP_THREADS=4;
static const int DEFAULT_HTTP_WORKQUEUE=16;
@ -35,7 +33,7 @@ void InterruptHTTPServer();
void StopHTTPServer();
/** Handler for requests to a certain HTTP path */
typedef boost::function<void(HTTPRequest* req, const std::string &)> HTTPRequestHandler;
typedef std::function<void(HTTPRequest* req, const std::string &)> HTTPRequestHandler;
/** Register handler for prefix.
* If multiple handlers match a prefix, the first-registered one will
* be invoked.
@ -132,7 +130,7 @@ public:
* deleteWhenTriggered deletes this event object after the event is triggered (and the handler called)
* handler is the handler to call when the event is triggered.
*/
HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const boost::function<void(void)>& handler);
HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const std::function<void(void)>& handler);
~HTTPEvent();
/** Trigger the event. If tv is 0, trigger it immediately. Otherwise trigger it after
@ -141,7 +139,7 @@ public:
void trigger(struct timeval* tv);
bool deleteWhenTriggered;
boost::function<void(void)> handler;
std::function<void(void)> handler;
private:
struct event* ev;
};