Skip to content

Attempt to expand listener to multi-threaded #1313

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jiankyu wants to merge 6 commits into
base: master
Choose a base branch
from
Draft

Attempt to expand listener to multi-threaded #1313

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
680148f
Attempt to expand listener to multi-threaded
jiankyu May 3, 2025
27490dd
code style: fix alignment issue
jiankyu May 6, 2025
210743d
issue-1312: bump patch version for adding new interfaces
jiankyu May 6, 2025
4f52848
issue-1312: bump minor version
jiankyu May 6, 2025
299cc5e
issue-1312: try to resolve windows build error
jiankyu May 7, 2025
10f3598
issue-1312: more robust Listener::shutdown()
jiankyu May 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions include/pistache/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,11 @@
// Allow compile-time overload
namespace Pistache::Const
{
static constexpr size_t MaxBacklog = 128;
static constexpr size_t MaxEvents = 1024;
static constexpr size_t MaxBuffer = 4096;
static constexpr size_t DefaultWorkers = 1;
static constexpr size_t MaxBacklog = 128;
static constexpr size_t MaxEvents = 1024;
static constexpr size_t MaxBuffer = 4096;
static constexpr size_t DefaultAcceptors = 1;
static constexpr size_t DefaultWorkers = 1;

static constexpr size_t DefaultTimerPoolSize = 128;

Expand Down
6 changes: 6 additions & 0 deletions include/pistache/endpoint.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ namespace Pistache::Http
{
friend class Endpoint;

Options& acceptThreads(int val);
Options& acceptThreadsName(const std::string& val);
Options& threads(int val);
Options& threadsName(const std::string& val);

Expand Down Expand Up @@ -78,6 +80,10 @@ namespace Pistache::Http
maxPayload(size_t val);

private:
// Accept thread options
int acceptThreads_;
std::string acceptThreadName_;

// Thread options
int threads_;
std::string threadsName_;
Expand Down
16 changes: 12 additions & 4 deletions include/pistache/listener.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,12 @@ namespace Pistache::Tcp

explicit Listener(const Address& address);
void init(size_t workers,
Flags<Options> options = Flags<Options>(Options::None),
const std::string& workersName = "",
int backlog = Const::MaxBacklog,
PISTACHE_STRING_LOGGER_T logger = PISTACHE_NULL_STRING_LOGGER);
Flags<Options> options = Flags<Options>(Options::None),
const std::string& workersName = "",
int backlog = Const::MaxBacklog,
size_t acceptors = Const::DefaultAcceptors,
const std::string& acceptorsName = "",
PISTACHE_STRING_LOGGER_T logger = PISTACHE_NULL_STRING_LOGGER);

void setTransportFactory(TransportFactory factory);
void setHandler(const std::shared_ptr<Handler>& handler);
Expand Down Expand Up @@ -105,6 +107,11 @@ namespace Pistache::Tcp
Flags<Options> options_;
std::thread acceptThread;

size_t acceptors_ = Const::DefaultAcceptors;
std::string acceptorsName_;
std::vector<std::thread> acceptWorkers;
std::atomic_int activeAcceptors { 0 };

size_t workers_ = Const::DefaultWorkers;
std::string workersName_;
std::shared_ptr<Handler> handler_;
Expand All @@ -117,6 +124,7 @@ namespace Pistache::Tcp
TransportFactory defaultTransportFactory() const;

bool bindListener(const struct addrinfo* addr);
void acceptWorkerFn();

void handleNewConnection();
em_socket_t acceptConnection(struct sockaddr_storage& peer_addr) const;
Expand Down
55 changes: 23 additions & 32 deletions src/common/reactor.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,22 +40,22 @@
static std::atomic_bool lLoggedSetThreadDescriptionFail = false;
#ifdef __MINGW32__

#include <windows.h>
#include <libloaderapi.h> // for GetProcAddress and GetModuleHandleA
typedef HRESULT (WINAPI *TSetThreadDescription)(HANDLE, PCWSTR);
#include <windows.h>
typedef HRESULT(WINAPI* TSetThreadDescription)(HANDLE, PCWSTR);

static std::atomic_bool lSetThreadDescriptionLoaded = false;
static std::mutex lSetThreadDescriptionLoadMutex;
static TSetThreadDescription lSetThreadDescriptionPtr = nullptr;

TSetThreadDescription getSetThreadDescriptionPtr()
static TSetThreadDescription getSetThreadDescriptionPtr()
{
if (lSetThreadDescriptionLoaded)
return(lSetThreadDescriptionPtr);
return (lSetThreadDescriptionPtr);

GUARD_AND_DBG_LOG(lSetThreadDescriptionLoadMutex);
if (lSetThreadDescriptionLoaded)
return(lSetThreadDescriptionPtr);
return (lSetThreadDescriptionPtr);

HMODULE hKernelBase = GetModuleHandleA("KernelBase.dll");

Expand All @@ -64,26 +64,23 @@ TSetThreadDescription getSetThreadDescriptionPtr()
PS_LOG_WARNING(
"Failed to get KernelBase.dll for SetThreadDescription");
lSetThreadDescriptionLoaded = true;
return(nullptr);
return (nullptr);
}

FARPROC set_thread_desc_fpptr =
GetProcAddress(hKernelBase, "SetThreadDescription");
FARPROC set_thread_desc_fpptr = GetProcAddress(hKernelBase, "SetThreadDescription");

// We do the cast in two steps, otherwise mingw-gcc complains about
// incompatible types
void * set_thread_desc_vptr =
reinterpret_cast<void *>(set_thread_desc_fpptr);
lSetThreadDescriptionPtr =
reinterpret_cast<TSetThreadDescription>(set_thread_desc_vptr);
void* set_thread_desc_vptr = reinterpret_cast<void*>(set_thread_desc_fpptr);
lSetThreadDescriptionPtr = reinterpret_cast<TSetThreadDescription>(set_thread_desc_vptr);

lSetThreadDescriptionLoaded = true;
if (!lSetThreadDescriptionPtr)
{
PS_LOG_WARNING(
"Failed to get SetThreadDescription from KernelBase.dll");
}
return(lSetThreadDescriptionPtr);
return (lSetThreadDescriptionPtr);
}
#endif // of ifdef __MINGW32__
#endif // of ifdef _IS_WINDOWS
Expand Down Expand Up @@ -299,17 +296,16 @@ namespace Pistache::Aio
break;
case 0:
break;
default:
{
if (shutdown_)
return;
default: {
if (shutdown_)
return;

GUARD_AND_DBG_LOG(shutdown_mutex_);
if (shutdown_)
return;
GUARD_AND_DBG_LOG(shutdown_mutex_);
if (shutdown_)
return;

handleFds(std::move(events));
}
handleFds(std::move(events));
}
}
}
}
Expand Down Expand Up @@ -454,11 +450,8 @@ namespace Pistache::Aio
// encode the index of the handler is that in the fast path, we
// won't need to shift the value to retrieve the fd if there is
// only one handler as all the bits will already be set to 0.
auto encodedValue =
(index << HandlerShift) |
PS_FD_CAST_TO_UNUM(uint64_t, static_cast<Fd>(value));
Polling::TagValue encodedValueTV =
static_cast<Polling::TagValue>(PS_NUM_CAST_TO_FD(encodedValue));
auto encodedValue = (index << HandlerShift) | PS_FD_CAST_TO_UNUM(uint64_t, static_cast<Fd>(value));
Polling::TagValue encodedValueTV = static_cast<Polling::TagValue>(PS_NUM_CAST_TO_FD(encodedValue));
return Polling::Tag(encodedValueTV);
}

Expand All @@ -468,8 +461,7 @@ namespace Pistache::Aio
auto value = tag.valueU64();
size_t index = value >> HandlerShift;
uint64_t maskedValue = value & DataMask;
Polling::TagValue maskedValueTV =
static_cast<Polling::TagValue>(PS_NUM_CAST_TO_FD(maskedValue));
Polling::TagValue maskedValueTV = static_cast<Polling::TagValue>(PS_NUM_CAST_TO_FD(maskedValue));

return std::make_pair(index, maskedValueTV);
}
Expand Down Expand Up @@ -720,13 +712,12 @@ namespace Pistache::Aio
#ifdef _IS_WINDOWS
const std::string threads_name(threadsName_.substr(0, 15));
const std::wstring temp(threads_name.begin(),
threads_name.end());
threads_name.end());
const LPCWSTR wide_threads_name = temp.c_str();

HRESULT hr = E_NOTIMPL;
#ifdef __MINGW32__
TSetThreadDescription set_thread_description_ptr =
getSetThreadDescriptionPtr();
TSetThreadDescription set_thread_description_ptr = getSetThreadDescriptionPtr();
if (set_thread_description_ptr)
{
hr = set_thread_description_ptr(
Expand Down
19 changes: 17 additions & 2 deletions src/server/endpoint.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,7 +251,8 @@ namespace Pistache::Http
}

Endpoint::Options::Options()
: threads_(1)
: acceptThreads_(1)
, threads_(1)
, flags_()
, backlog_(Const::MaxBacklog)
, maxRequestSize_(Const::DefaultMaxRequestSize)
Expand All @@ -264,6 +265,18 @@ namespace Pistache::Http
, sslHandshakeTimeout_(Const::DefaultSSLHandshakeTimeout)
{ }

Endpoint::Options& Endpoint::Options::acceptThreads(int val)
{
acceptThreads_ = val;
return *this;
}

Endpoint::Options& Endpoint::Options::acceptThreadsName(const std::string& val)
{
acceptThreadName_ = val;
return *this;
}

Endpoint::Options& Endpoint::Options::threads(int val)
{
threads_ = val;
Expand Down Expand Up @@ -319,7 +332,9 @@ namespace Pistache::Http

void Endpoint::init(const Endpoint::Options& options)
{
listener.init(options.threads_, options.flags_, options.threadsName_, options.backlog_);
listener.init(options.threads_, options.flags_, options.threadsName_, options.backlog_,
options.acceptThreads_, options.acceptThreadName_);

listener.setTransportFactory([this, options] {
if (!handler_)
throw std::runtime_error("Must call setHandler()");
Expand Down
Loading
Loading