hcq/MaxScale
hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MXS-1754 Reintroduce maxscale::Worker

Browse Source 
Worker is now the base class of all workers. It has a message
queue and can be run in a thread of its own, or in the calling
thread. Worker can not be used as such, but a concrete worker
class must be derived from it. Currently there is only one
concrete class RoutingWorker.

There is some overlapping in functionality between Worker and
RoutingWorker, as there is e.g. a need for broadcasting a
message to all routing workers, but not to other workers.

Currently other workers can not be created as the array for
holding the pointers to the workers is exactly as large as
there will be RoutingWorkers. That will be changed so that
the maximum number of threads is hardwired to some ridiculous
value such as 128. That's the first step in the path towards
a situation where the number of worker threads can be changed
at runtime.
This commit is contained in:
Johan Wikman 2018-04-13 09:51:27 +03:00
parent 230876cd69
commit b36f6faa7e
19 changed files with 2338 additions and 1892 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
include/maxscale/routingworker.h Normal file
View File

@ -0,0 +1,97 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2020-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include <maxscale/cdefs.h>
#include <maxscale/session.h>
#include <maxscale/worker.h>
MXS_BEGIN_DECLS
#define MXS_RWORKER_MAIN -1
/**
* Return the routing worker associated with the provided worker id.
*
* @param worker_id A worker id. If MXS_RWORKER_MAIN is used, the
* routing worker running in the main thread will
* be returned.
*
* @return The corresponding routing worker instance, or NULL if the
* id does not correspond to a routing worker.
*/
MXS_WORKER* mxs_rworker_get(int worker_id);
/**
* Return the current routing worker.
*
* @return A routing worker, or NULL if there is no current routing worker.
*/
MXS_WORKER* mxs_rworker_get_current();
/**
* Return the id of the current routing worker.
*
* @return The id of the routing worker, or -1 if there is no current
* routing worker.
*/
int mxs_rworker_get_current_id();
/**
* Broadcast a message to all routing workers.
*
* @param msg_id The message id.
* @param arg1 Message specific first argument.
* @param arg2 Message specific second argument.
*
* @return The number of messages posted; if less that ne number of workers
* then some postings failed.
*
* @attention The return value tells *only* whether message could be posted,
* *not* that it has reached the worker.
*
* @attentsion Exactly the same arguments are passed to all workers. Take that
* into account if the passed data must be freed.
*
* @attention This function is signal safe.
*/
size_t mxs_rworker_broadcast_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Add a session to the current routing worker's session container. Currently
* only required for some special commands e.g. "KILL <process_id>" to work.
*
* @param session Session to add.
* @return true if successful, false if id already existed in map.
*/
bool mxs_rworker_register_session(MXS_SESSION* session);
/**
* Remove a session from the current routing worker's session container. Does
* not actually remove anything from an epoll-set or affect the session in any
* way.
*
* @param id Which id to remove.
* @return The removed session or NULL if not found.
*/
bool mxs_rworker_deregister_session(uint64_t id);
/**
* Find a session in the current routing worker's session container.
*
* @param id Which id to find.
* @return The found session or NULL if not found.
*/
MXS_SESSION* mxs_rworker_find_session(uint64_t id);
MXS_END_DECLS

37
include/maxscale/worker.h
View File

@ -70,12 +70,17 @@ MXS_WORKER* mxs_worker_get(int worker_id);
*/
int mxs_worker_id(MXS_WORKER* pWorker);
/**
* Return the current worker.
*
* @return A worker, or NULL if there is no current worker.
*/
MXS_WORKER* mxs_worker_get_current();
/**
* Return the id of the worker.
*
* @return The id of the worker.
*
* @attention If there is no current worker, then -1 will be returned.
* @return The id of the worker, or -1 if there is no current worker.
*/
int mxs_worker_get_current_id();
@ -117,32 +122,6 @@ bool mxs_worker_post_message(MXS_WORKER* worker, uint32_t msg_id, intptr_t arg1,
*/
size_t mxs_worker_broadcast_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Add a session to the current worker's session container. Currently only
* required for some special commands e.g. "KILL <process_id>" to work.
*
* @param session Session to add.
* @return true if successful, false if id already existed in map.
*/
bool mxs_worker_register_session(MXS_SESSION* session);
/**
* Remove a session from the current worker's session container. Does not actually
* remove anything from an epoll-set or affect the session in any way.
*
* @param id Which id to remove.
* @return The removed session or NULL if not found.
*/
bool mxs_worker_deregister_session(uint64_t id);
/**
* Find a session in the current worker's session container.
*
* @param id Which id to find.
* @return The found session or NULL if not found.
*/
MXS_SESSION* mxs_worker_find_session(uint64_t id);
/**
* @brief Convert a worker to JSON format
*

1
server/core/CMakeLists.txt
View File

@ -52,6 +52,7 @@ add_library(maxscale-common SHARED
thread.cc
users.cc
utils.cc
worker.cc
workertask.cc
)

9
server/core/dcb.cc
View File

@ -61,6 +61,7 @@
#include "internal/session.h"
#include "internal/workertask.hh"
using maxscale::RoutingWorker;
using maxscale::Worker;
using maxscale::WorkerTask;
using maxscale::Semaphore;
@ -208,8 +209,8 @@ dcb_alloc(dcb_role_t role, SERV_LISTENER *listener)
else
{
/** Otherwise the DCB is owned by the thread that allocates it */
ss_dassert(Worker::get_current_id() != -1);
newdcb->poll.thread.id = Worker::get_current_id();
ss_dassert(RoutingWorker::get_current_id() != -1);
newdcb->poll.thread.id = RoutingWorker::get_current_id();
}
return newdcb;
@ -1148,7 +1149,7 @@ void dcb_close(DCB *dcb)
}
else
{
Worker* worker = Worker::get(dcb->poll.thread.id);
RoutingWorker* worker = RoutingWorker::get(dcb->poll.thread.id);
ss_dassert(worker);
worker->register_zombie(dcb);
@ -1179,7 +1180,7 @@ void dcb_close_in_owning_thread(DCB* dcb)
// TODO: reference counted, so that we could addref before posting, thus
// TODO: preventing too early a deletion.
MXS_WORKER* worker = mxs_worker_get(dcb->poll.thread.id); // The owning worker
MXS_WORKER* worker = mxs_rworker_get(dcb->poll.thread.id); // The owning worker
ss_dassert(worker);
intptr_t arg1 = (intptr_t)cb_dcb_close_in_owning_thread;

47
server/core/gateway.cc
View File

@ -1399,8 +1399,6 @@ bool set_dirs(const char *basedir)
int main(int argc, char **argv)
{
int rc = MAXSCALE_SHUTDOWN;
int n_threads; /*< number of epoll listener threads */
size_t thread_stack_size;
int n_services;
int eno = 0; /*< local variable for errno */
int opt;
@ -2129,6 +2127,13 @@ int main(int argc, char **argv)
goto return_main;
}
if (!RoutingWorker::init())
{
MXS_ERROR("Failed to initialize routing workers.");
rc = MAXSCALE_INTERNALERROR;
goto return_main;
}
/* Init MaxScale modules */
if (!modules_process_init())
{
@ -2183,27 +2188,14 @@ int main(int argc, char **argv)
}
/*<
* Start the polling threads, note this is one less than is
* configured as the main thread will also poll.
* Start the routing workers running in their own thread.
*/
n_threads = config_threadcount();
/*<
* Start workers. We start from 1, worker 0 will be running in the main thread.
*/
thread_stack_size = config_thread_stack_size();
for (int i = 1; i < n_threads; i++)
if (!RoutingWorker::start_threaded_workers())
{
worker = Worker::get(i);
ss_dassert(worker);
if (!worker->start(thread_stack_size))
{
const char* logerr = "Failed to start worker thread.";
print_log_n_stderr(true, true, logerr, logerr, 0);
rc = MAXSCALE_INTERNALERROR;
goto return_main;
}
const char* logerr = "Failed to start routing workers.";
print_log_n_stderr(true, true, logerr, logerr, 0);
rc = MAXSCALE_INTERNALERROR;
goto return_main;
}
if (cnf->admin_enabled)
@ -2232,7 +2224,7 @@ int main(int argc, char **argv)
}
MXS_NOTICE("MaxScale started with %d worker threads, each with a stack size of %lu bytes.",
n_threads, thread_stack_size);
config_threadcount(), config_thread_stack_size());
/**
* Successful start, notify the parent process that it can exit.
@ -2246,7 +2238,7 @@ int main(int argc, char **argv)
/*<
* Run worker 0 in the main thread.
*/
worker = Worker::get(0);
worker = RoutingWorker::get(RoutingWorker::MAIN);
ss_dassert(worker);
worker->run();
@ -2267,13 +2259,7 @@ int main(int argc, char **argv)
/*<
* Wait for worker threads to exit.
*/
for (int i = 1; i < n_threads; i++)
{
worker = Worker::get(i);
ss_dassert(worker);
worker->join();
}
RoutingWorker::join_threaded_workers();
MXS_NOTICE("All workers have shut down.");
@ -2282,6 +2268,7 @@ int main(int argc, char **argv)
*/
service_destroy_instances();
RoutingWorker::finish();
Worker::finish();
MessageQueue::finish();

14
server/core/internal/messagequeue.hh
View File

@ -19,7 +19,7 @@ namespace maxscale
{
class MessageQueue;
class RoutingWorker;
class Worker;
/**
* An instance of @c MessageQueueMessage can be sent over a @c MessageQueue from
@ -180,7 +180,7 @@ public:
* @attention If the message queue is currently added to a worker, it
* will first be removed from that worker.
*/
bool add_to_worker(RoutingWorker* pWorker);
bool add_to_worker(Worker* pWorker);
/**
* Removes the message queue from the worker it is currently added to.
@ -188,7 +188,7 @@ public:
* @return The worker the message queue was associated with, or NULL
* if it was not associated with any.
*/
RoutingWorker* remove_from_worker();
Worker* remove_from_worker();
private:
MessageQueue(Handler* pHandler, int read_fd, int write_fd);
@ -198,10 +198,10 @@ private:
static uint32_t poll_handler(MXS_POLL_DATA* pData, int thread_id, uint32_t events);
private:
Handler& m_handler;
int m_read_fd;
int m_write_fd;
RoutingWorker* m_pWorker;
Handler& m_handler;
int m_read_fd;
int m_write_fd;
Worker* m_pWorker;
};
}

809
server/core/internal/routingworker.hh
View File

@ -13,450 +13,30 @@
*/
#include <maxscale/cppdefs.hh>
#include <memory>
#include <vector>
#include <maxscale/platform.h>
#include <maxscale/session.h>
#include <maxscale/utils.hh>
#include "messagequeue.hh"
#include "poll.h"
#include "worker.h"
#include "workertask.hh"
#include <maxscale/routingworker.h>
#include "worker.hh"
#include "session.hh"
namespace maxscale
{
class Semaphore;
struct WORKER_STATISTICS
class RoutingWorker : public Worker
, private MXS_POLL_DATA
{
WORKER_STATISTICS()
{
memset(this, 0, sizeof(WORKER_STATISTICS));
}
enum
{
MAXNFDS = 10,
N_QUEUE_TIMES = 30
};
int64_t n_read; /*< Number of read events */
int64_t n_write; /*< Number of write events */
int64_t n_error; /*< Number of error events */
int64_t n_hup; /*< Number of hangup events */
int64_t n_accept; /*< Number of accept events */
int64_t n_polls; /*< Number of poll cycles */
int64_t n_pollev; /*< Number of polls returning events */
int64_t n_nbpollev; /*< Number of polls returning events */
int64_t n_fds[MAXNFDS]; /*< Number of wakeups with particular n_fds value */
int64_t evq_length; /*< Event queue length */
int64_t evq_max; /*< Maximum event queue length */
int64_t blockingpolls; /*< Number of epoll_waits with a timeout specified */
uint32_t qtimes[N_QUEUE_TIMES + 1];
uint32_t exectimes[N_QUEUE_TIMES + 1];
int64_t maxqtime;
int64_t maxexectime;
};
/**
* WorkerLoad is a class that calculates the load percentage of a worker
* thread, based upon the relative amount of time the worker spends in
* epoll_wait().
*
* If during a time period of length T milliseconds, the worker thread
* spends t milliseconds in epoll_wait(), then the load of the worker is
* calculated as 100 * ((T - t) / T). That is, if the worker spends all
* the time in epoll_wait(), then the load is 0 and if the worker spends
* no time waiting in epoll_wait(), then the load is 100.
*/
class WorkerLoad
{
WorkerLoad(const WorkerLoad&);
WorkerLoad& operator = (const WorkerLoad&);
RoutingWorker(const RoutingWorker&) = delete;
RoutingWorker& operator = (const RoutingWorker&) = delete;
public:
enum counter_t
{
ONE_SECOND = 1000,
ONE_MINUTE = 60 * ONE_SECOND,
ONE_HOUR = 60 * ONE_MINUTE,
};
enum
{
GRANULARITY = ONE_SECOND
MAIN = -1
};
/**
* Constructor
*/
WorkerLoad();
/**
* Reset the load calculation. Should be called immediately before the
* worker enters its eternal epoll_wait()-loop.
*/
void reset()
{
uint64_t now = get_time();
m_start_time = now;
m_wait_start = 0;
m_wait_time = 0;
}
/**
* To be used for signaling that the worker is about to call epoll_wait().
*
* @param now The current time.
*/
void about_to_wait(uint64_t now)
{
m_wait_start = now;
}
void about_to_wait()
{
about_to_wait(get_time());
}
/**
* To be used for signaling that the worker has returned from epoll_wait().
*
* @param now The current time.
*/
void about_to_work(uint64_t now);
void about_to_work()
{
about_to_work(get_time());
}
/**
* Returns the last calculated load,
*
* @return A value between 0 and 100.
*/
uint8_t percentage(counter_t counter) const
{
switch (counter)
{
case ONE_SECOND:
return m_load_1_second.value();
case ONE_MINUTE:
return m_load_1_minute.value();
case ONE_HOUR:
return m_load_1_hour.value();
default:
ss_dassert(!true);
return 0;
};
}
/**
* When was the last 1 second period started.
*
* @return The start time.
*/
uint64_t start_time() const
{
return m_start_time;
}
/**
* Returns the current time using CLOCK_MONOTONIC.
*
* @return Current time in milliseconds.
*/
static uint64_t get_time();
private:
/**
* Average is a base class for classes intended to be used for calculating
* averages. An Average may have a dependant Average whose value depends
* upon the value of the first. At certain moments, an Average may trigger
* its dependant Average to update itself.
*/
class Average
{
Average(const Average&);
Average& operator = (const Average&);
public:
/**
* Constructor
*
* @param pDependant An optional dependant average.
*/
Average(Average* pDependant = NULL)
: m_pDependant(pDependant)
, m_value(0)
{}
virtual ~Average();
/**
* Add a value to the Average. The exact meaning depends upon the
* concrete Average class.
*
* If the addition of the value in some sense represents a full cycle
* in the average calculation, then the instance will call add_value()
* on its dependant, otherwise it will call update_value(). In both cases
* with its own value as argument.
*
* @param value The value to be added.
*
* @return True if the addition of the value caused a full cycle
* in the average calculation, false otherwise.
*/
virtual bool add_value(uint8_t value) = 0;
/**
* Update the value of the Average. The exact meaning depends upon the
* concrete Average class. Will also call update_value() of its dependant
* with its own value as argument.
*
* @param value The value to be updated.
*/
virtual void update_value(uint8_t value) = 0;
/**
* Return the average value.
*
* @return The value represented by the Average.
*/
uint8_t value() const
{
return atomic_load_uint32(&m_value);
}
protected:
Average* m_pDependant; /*< The optional dependant Average. */
uint32_t m_value; /*< The current average value. */
protected:
void set_value(uint32_t value)
{
atomic_store_uint32(&m_value, value);
}
};
/**
* An Average consisting of a single value.
*/
class Average1 : public Average
{
public:
Average1(Average* pDependant = NULL)
: Average(pDependant)
{
}
bool add_value(uint8_t value)
{
set_value(value);
// Every addition of a value represents a full cycle.
if (m_pDependant)
{
m_pDependant->add_value(value);
}
return true;
}
void update_value(uint8_t value)
{
set_value(value);
if (m_pDependant)
{
m_pDependant->update_value(value);
}
}
};
/**
* An Average calculated from N values.
*/
template<size_t N>
class AverageN : public Average
{
public:
AverageN(Average* pDependant = NULL)
: Average(pDependant)
, m_end(m_begin + N)
, m_i(m_begin)
, m_sum(0)
, m_nValues(0)
{
}
bool add_value(uint8_t value)
{
if (m_nValues == N)
{
// If as many values that fit has been added, then remove the
// least recent value from the sum.
m_sum -= *m_i;
}
else
{
// Otherwise make a note that a new value is added.
++m_nValues;
}
*m_i = value;
m_sum += *m_i; // Update the sum of all values.
m_i = next(m_i);
uint32_t average = m_sum / m_nValues;
set_value(average);
if (m_pDependant)
{
if (m_i == m_begin)
{
// If we have looped around we have performed a full cycle and will
// add a new value to the dependant average.
m_pDependant->add_value(average);
}
else
{
// Otherwise we just update the most recent value.
m_pDependant->update_value(average);
}
}
return m_i == m_begin;
}
void update_value(uint8_t value)
{
if (m_nValues == 0)
{
// If no values have been added yet, there's nothing to update but we
// need to add the value.
add_value(value);
}
else
{
// Otherwise we update the most recent value.
uint8_t* p = prev(m_i);
m_sum -= *p;
*p = value;
m_sum += *p;
uint32_t average = m_sum / m_nValues;
set_value(average);
if (m_pDependant)
{
m_pDependant->update_value(average);
}
}
}
private:
uint8_t* prev(uint8_t* p)
{
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
if (p > m_begin)
{
--p;
}
else
{
ss_dassert(p == m_begin);
p = m_end - 1;
}
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
return p;
}
uint8_t* next(uint8_t* p)
{
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
++p;
if (p == m_end)
{
p = m_begin;
}
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
return p;
}
private:
uint8_t m_begin[N]; /*< Buffer containing values from which the average is calculated. */
uint8_t* m_end; /*< Points to one past the end of the buffer. */
uint8_t* m_i; /*< Current position in the buffer. */
uint32_t m_sum; /*< Sum of all values in the buffer. */
uint32_t m_nValues; /*< How many values the buffer contains. */
};
uint64_t m_start_time; /*< When was the current 1-second period started. */
uint64_t m_wait_start; /*< The time when the worker entered epoll_wait(). */
uint64_t m_wait_time; /*< How much time the worker has spent in epoll_wait(). */
AverageN<60> m_load_1_hour; /*< The average load during the last hour. */
AverageN<60> m_load_1_minute; /*< The average load during the last minute. */
Average1 m_load_1_second; /*< The load during the last 1-second period. */
};
class RoutingWorker;
typedef RoutingWorker Worker;
class RoutingWorker : public MXS_WORKER
, private MessageQueue::Handler
, private MXS_POLL_DATA
{
RoutingWorker(const RoutingWorker&);
RoutingWorker& operator = (const RoutingWorker&);
public:
typedef WORKER_STATISTICS STATISTICS;
typedef WorkerTask Task;
typedef WorkerDisposableTask DisposableTask;
typedef Registry<MXS_SESSION> SessionsById;
typedef std::vector<DCB*> Zombies;
typedef WorkerLoad Load;
enum state_t
{
STOPPED,
IDLE,
POLLING,
PROCESSING,
ZPROCESSING
};
enum execute_mode_t
{
EXECUTE_AUTO, /**< Execute tasks immediately */
EXECUTE_QUEUED /**< Only queue tasks for execution */
};
/**
* Initialize the worker mechanism.
* Initialize the routing worker mechanism.
*
* To be called once at process startup. This will cause as many workers
* to be created as the number of threads defined.
@ -474,102 +54,6 @@ public:
*/
static void finish();
/**
* Returns the id of the worker
*
* @return The id of the worker.
*/
int id() const
{
return m_id;
}
int load(Load::counter_t counter)
{
return m_load.percentage(counter);
}
/**
* Returns the state of the worker.
*
* @return The current state.
*
* @attentions The state might have changed the moment after the function returns.
*/
state_t state() const
{
return m_state;
}
/**
* Returns statistics for this worker.
*
* @return The worker specific statistics.
*
* @attentions The statistics may change at any time.
*/
const STATISTICS& statistics() const
{
return m_statistics;
}
/**
* Returns statistics for all workers.
*
* @return Combined statistics.
*
* @attentions The statistics may no longer be accurate by the time it has
* been returned. The returned values may also not represent a
* 100% consistent set.
*/
static STATISTICS get_statistics();
/**
* Return a specific combined statistic value.
*
* @param what What to return.
*
* @return The corresponding value.
*/
static int64_t get_one_statistic(POLL_STAT what);
/**
* Return this worker's statistics.
*
* @return Local statistics for this worker.
*/
const STATISTICS& get_local_statistics() const
{
return m_statistics;
}
/**
* Return the count of descriptors.
*
* @param pnCurrent On output the current number of descriptors.
* @param pnTotal On output the total number of descriptors.
*/
void get_descriptor_counts(uint32_t* pnCurrent, uint64_t* pnTotal);
/**
* Add a file descriptor to the epoll instance of the worker.
*
* @param fd The file descriptor to be added.
* @param events Mask of epoll event types.
* @param pData The poll data associated with the descriptor:
*
* data->handler : Handler that knows how to deal with events
* for this particular type of 'struct mxs_poll_data'.
* data->thread.id: Will be updated by the worker.
*
* @attention The provided file descriptor must be non-blocking.
* @attention @c pData must remain valid until the file descriptor is
* removed from the worker.
*
* @return True, if the descriptor could be added, false otherwise.
*/
bool add_fd(int fd, uint32_t events, MXS_POLL_DATA* pData);
/**
* Add a file descriptor to the epoll instance shared between all workers.
* Events occuring on the provided file descriptor will be handled by all
@ -589,15 +73,6 @@ public:
*/
static bool add_shared_fd(int fd, uint32_t events, MXS_POLL_DATA* pData);
/**
* Remove a file descriptor from the worker's epoll instance.
*
* @param fd The file descriptor to be removed.
*
* @return True on success, false on failure.
*/
bool remove_fd(int fd);
/**
* Remove a file descriptor from the epoll instance shared between all workers.
*
@ -616,180 +91,6 @@ public:
*/
void register_zombie(DCB* pZombie);
/**
* Main function of worker.
*
* The worker will run the poll loop, until it is told to shut down.
*
* @attention This function will run in the calling thread.
*/
void run();
/**
* Run worker in separate thread.
*
* This function will start a new thread, in which the `run`
* function will be executed.
*
* @param stack_size The stack size of the new thread. A value of 0 means
* that the pthread default should be used.
*
* @return True if the thread could be started, false otherwise.
*/
bool start(size_t stack_size = 0);
/**
* Waits for the worker to finish.
*/
void join();
/**
* Initate shutdown of worker.
*
* @attention A call to this function will only initiate the shutdowm,
* the worker will not have shut down when the function returns.
*
* @attention This function is signal safe.
*/
void shutdown();
/**
* Query whether worker should shutdown.
*
* @return True, if the worker should shut down, false otherwise.
*/
bool should_shutdown() const
{
return m_should_shutdown;
}
/**
* Posts a task to a worker for execution.
*
* @param pTask The task to be executed.
* @param pSem If non-NULL, will be posted once the task's `execute` return.
* @param mode Execution mode
*
* @return True if the task could be posted (i.e. not executed), false otherwise.
*
* @attention The instance must remain valid for as long as it takes for the
* task to be transferred to the worker and its `execute` function
* to be called.
*
* The semaphore can be used for waiting for the task to be finished.
*
* @code
* Semaphore sem;
* MyTask task;
*
* pWorker->execute(&task, &sem);
* sem.wait();
*
* MyResult& result = task.result();
* @endcode
*/
bool post(Task* pTask, Semaphore* pSem = NULL, enum execute_mode_t mode = EXECUTE_AUTO);
/**
* Posts a task to a worker for execution.
*
* @param pTask The task to be executed.
* @param mode Execution mode
*
* @return True if the task could be posted (i.e. not executed), false otherwise.
*
* @attention Once the task has been executed, it will be deleted.
*/
bool post(std::auto_ptr<DisposableTask> sTask, enum execute_mode_t mode = EXECUTE_AUTO);
template<class T>
bool post(std::auto_ptr<T> sTask, enum execute_mode_t mode = EXECUTE_AUTO)
{
return post(std::auto_ptr<DisposableTask>(sTask.release()), mode);
}
/**
* Posts a task to all workers for execution.
*
* @param pTask The task to be executed.
* @param pSem If non-NULL, will be posted once per worker when the task's
* `execute` return.
*
* @return How many workers the task was posted to.
*
* @attention The very same task will be posted to all workers. The task
* should either not have any sharable data or then it should
* have data specific to each worker that can be accessed
* without locks.
*/
static size_t broadcast(Task* pTask, Semaphore* pSem = NULL);
/**
* Posts a task to all workers for execution.
*
* @param pTask The task to be executed.
*
* @return How many workers the task was posted to.
*
* @attention The very same task will be posted to all workers. The task
* should either not have any sharable data or then it should
* have data specific to each worker that can be accessed
* without locks.
*
* @attention Once the task has been executed by all workers, it will
* be deleted.
*/
static size_t broadcast(std::auto_ptr<DisposableTask> sTask);
template<class T>
static size_t broadcast(std::auto_ptr<T> sTask)
{
return broadcast(std::auto_ptr<DisposableTask>(sTask.release()));
}
/**
* Executes a task on all workers in serial mode (the task is executed
* on at most one worker thread at a time). When the function returns
* the task has been executed on all workers.
*
* @param task The task to be executed.
*
* @return How many workers the task was posted to.
*
* @warning This function is extremely inefficient and will be slow compared
* to the other functions. Only use this function when printing thread-specific
* data to stdout.
*/
static size_t execute_serially(Task& task);
/**
* Executes a task on all workers concurrently and waits until all workers
* are done. That is, when the function returns the task has been executed
* by all workers.
*
* @param task The task to be executed.
*
* @return How many workers the task was posted to.
*/
static size_t execute_concurrently(Task& task);
/**
* Post a message to a worker.
*
* @param msg_id The message id.
* @param arg1 Message specific first argument.
* @param arg2 Message specific second argument.
*
* @return True if the message could be sent, false otherwise. If the message
* posting fails, errno is set appropriately.
*
* @attention The return value tells *only* whether the message could be sent,
* *not* that it has reached the worker.
*
* @attention This function is signal safe.
*/
bool post_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Return a reference to the session registry of this worker.
*
@ -797,40 +98,11 @@ public:
*/
SessionsById& session_registry();
/**
* Broadcast a message to all worker.
*
* @param msg_id The message id.
* @param arg1 Message specific first argument.
* @param arg2 Message specific second argument.
*
* @return The number of messages posted; if less that ne number of workers
* then some postings failed.
*
* @attention The return value tells *only* whether message could be posted,
* *not* that it has reached the worker.
*
* @attentsion Exactly the same arguments are passed to all workers. Take that
* into account if the passed data must be freed.
*
* @attention This function is signal safe.
*/
static size_t broadcast_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Initate shutdown of all workers.
*
* @attention A call to this function will only initiate the shutdowm,
* the workers will not have shut down when the function returns.
*
* @attention This function is signal safe.
*/
static void shutdown_all();
/**
* Return the worker associated with the provided worker id.
*
* @param worker_id A worker id.
* @param worker_id A worker id. By specifying MAIN, the routing worker
* running in the main thread will be returned.
*
* @return The corresponding worker instance, or NULL if the id does
* not correspond to a worker.
@ -852,59 +124,50 @@ public:
static int get_current_id();
/**
* Set the number of non-blocking poll cycles that will be done before
* a blocking poll will take place.
* Starts all routing workers but the main worker (the one running in
* the main thread).
*
* @param nbpolls Number of non-blocking polls to perform before blocking.
* @return True, if all secondary workers could be started.
*/
static bool start_threaded_workers();
/**
* Waits for all threaded workers.
*/
static void join_threaded_workers();
/**
* Deprecated
*/
static void set_nonblocking_polls(unsigned int nbpolls);
/**
* Maximum time to block in epoll_wait.
*
* @param maxwait Maximum wait time in millliseconds.
* Deprecated
*/
static void set_maxwait(unsigned int maxwait);
private:
RoutingWorker(int id,
int epoll_fd);
RoutingWorker();
virtual ~RoutingWorker();
static RoutingWorker* create(int id, int epoll_listener_fd);
static RoutingWorker* create(int epoll_listener_fd);
bool pre_run(); // override
void post_run(); // override
void epoll_tick(); // override
void delete_zombies();
bool post_disposable(DisposableTask* pTask, enum execute_mode_t mode = EXECUTE_AUTO);
void handle_message(MessageQueue& queue, const MessageQueue::Message& msg); // override
static void thread_main(void* arg);
void poll_waitevents();
static uint32_t epoll_instance_handler(struct mxs_poll_data* data, int wid, uint32_t events);
uint32_t handle_epoll_events(uint32_t events);
private:
int m_id; /*< The id of the worker. */
state_t m_state; /*< The state of the worker */
int m_epoll_fd; /*< The epoll file descriptor. */
STATISTICS m_statistics; /*< Worker statistics. */
MessageQueue* m_pQueue; /*< The message queue of the worker. */
THREAD m_thread; /*< The thread handle of the worker. */
bool m_started; /*< Whether the thread has been started or not. */
bool m_should_shutdown; /*< Whether shutdown should be performed. */
bool m_shutdown_initiated; /*< Whether shutdown has been initated. */
SessionsById m_sessions; /*< A mapping of session_id->MXS_SESSION. The map
* should contain sessions exclusive to this
* worker and not e.g. listener sessions. For now,
* it's up to the protocol to decide whether a new
* session is added to the map. */
Zombies m_zombies; /*< DCBs to be deleted. */
uint32_t m_nCurrent_descriptors; /*< Current number of descriptors. */
uint64_t m_nTotal_descriptors; /*< Total number of descriptors. */
Load m_load;
SessionsById m_sessions; /*< A mapping of session_id->MXS_SESSION. The map
* should contain sessions exclusive to this
* worker and not e.g. listener sessions. For now,
* it's up to the protocol to decide whether a new
* session is added to the map. */
Zombies m_zombies; /*< DCBs to be deleted. */
};
}

862
server/core/internal/worker.hh Normal file
View File

@ -0,0 +1,862 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2020-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include <maxscale/cppdefs.hh>
#include <memory>
#include <vector>
#include <maxscale/platform.h>
#include <maxscale/session.h>
#include <maxscale/utils.hh>
#include "messagequeue.hh"
#include "poll.h"
#include "worker.h"
#include "workertask.hh"
#include "session.hh"
namespace maxscale
{
class Semaphore;
struct WORKER_STATISTICS
{
WORKER_STATISTICS()
{
memset(this, 0, sizeof(WORKER_STATISTICS));
}
enum
{
MAXNFDS = 10,
N_QUEUE_TIMES = 30
};
int64_t n_read; /*< Number of read events */
int64_t n_write; /*< Number of write events */
int64_t n_error; /*< Number of error events */
int64_t n_hup; /*< Number of hangup events */
int64_t n_accept; /*< Number of accept events */
int64_t n_polls; /*< Number of poll cycles */
int64_t n_pollev; /*< Number of polls returning events */
int64_t n_nbpollev; /*< Number of polls returning events */
int64_t n_fds[MAXNFDS]; /*< Number of wakeups with particular n_fds value */
int64_t evq_length; /*< Event queue length */
int64_t evq_max; /*< Maximum event queue length */
int64_t blockingpolls; /*< Number of epoll_waits with a timeout specified */
uint32_t qtimes[N_QUEUE_TIMES + 1];
uint32_t exectimes[N_QUEUE_TIMES + 1];
int64_t maxqtime;
int64_t maxexectime;
};
/**
* WorkerLoad is a class that calculates the load percentage of a worker
* thread, based upon the relative amount of time the worker spends in
* epoll_wait().
*
* If during a time period of length T milliseconds, the worker thread
* spends t milliseconds in epoll_wait(), then the load of the worker is
* calculated as 100 * ((T - t) / T). That is, if the worker spends all
* the time in epoll_wait(), then the load is 0 and if the worker spends
* no time waiting in epoll_wait(), then the load is 100.
*/
class WorkerLoad
{
WorkerLoad(const WorkerLoad&) = delete;
WorkerLoad& operator = (const WorkerLoad&) = delete;
public:
enum counter_t
{
ONE_SECOND = 1000,
ONE_MINUTE = 60 * ONE_SECOND,
ONE_HOUR = 60 * ONE_MINUTE,
};
enum
{
GRANULARITY = ONE_SECOND
};
/**
* Constructor
*/
WorkerLoad();
/**
* Reset the load calculation. Should be called immediately before the
* worker enters its eternal epoll_wait()-loop.
*/
void reset()
{
uint64_t now = get_time();
m_start_time = now;
m_wait_start = 0;
m_wait_time = 0;
}
/**
* To be used for signaling that the worker is about to call epoll_wait().
*
* @param now The current time.
*/
void about_to_wait(uint64_t now)
{
m_wait_start = now;
}
void about_to_wait()
{
about_to_wait(get_time());
}
/**
* To be used for signaling that the worker has returned from epoll_wait().
*
* @param now The current time.
*/
void about_to_work(uint64_t now);
void about_to_work()
{
about_to_work(get_time());
}
/**
* Returns the last calculated load,
*
* @return A value between 0 and 100.
*/
uint8_t percentage(counter_t counter) const
{
switch (counter)
{
case ONE_SECOND:
return m_load_1_second.value();
case ONE_MINUTE:
return m_load_1_minute.value();
case ONE_HOUR:
return m_load_1_hour.value();
default:
ss_dassert(!true);
return 0;
};
}
/**
* When was the last 1 second period started.
*
* @return The start time.
*/
uint64_t start_time() const
{
return m_start_time;
}
/**
* Returns the current time using CLOCK_MONOTONIC.
*
* @return Current time in milliseconds.
*/
static uint64_t get_time();
private:
/**
* Average is a base class for classes intended to be used for calculating
* averages. An Average may have a dependant Average whose value depends
* upon the value of the first. At certain moments, an Average may trigger
* its dependant Average to update itself.
*/
class Average
{
Average(const Average&) = delete;
Average& operator = (const Average&) = delete;
public:
/**
* Constructor
*
* @param pDependant An optional dependant average.
*/
Average(Average* pDependant = NULL)
: m_pDependant(pDependant)
, m_value(0)
{}
virtual ~Average();
/**
* Add a value to the Average. The exact meaning depends upon the
* concrete Average class.
*
* If the addition of the value in some sense represents a full cycle
* in the average calculation, then the instance will call add_value()
* on its dependant, otherwise it will call update_value(). In both cases
* with its own value as argument.
*
* @param value The value to be added.
*
* @return True if the addition of the value caused a full cycle
* in the average calculation, false otherwise.
*/
virtual bool add_value(uint8_t value) = 0;
/**
* Update the value of the Average. The exact meaning depends upon the
* concrete Average class. Will also call update_value() of its dependant
* with its own value as argument.
*
* @param value The value to be updated.
*/
virtual void update_value(uint8_t value) = 0;
/**
* Return the average value.
*
* @return The value represented by the Average.
*/
uint8_t value() const
{
return atomic_load_uint32(&m_value);
}
protected:
Average* m_pDependant; /*< The optional dependant Average. */
uint32_t m_value; /*< The current average value. */
protected:
void set_value(uint32_t value)
{
atomic_store_uint32(&m_value, value);
}
};
/**
* An Average consisting of a single value.
*/
class Average1 : public Average
{
public:
Average1(Average* pDependant = NULL)
: Average(pDependant)
{
}
bool add_value(uint8_t value)
{
set_value(value);
// Every addition of a value represents a full cycle.
if (m_pDependant)
{
m_pDependant->add_value(value);
}
return true;
}
void update_value(uint8_t value)
{
set_value(value);
if (m_pDependant)
{
m_pDependant->update_value(value);
}
}
};
/**
* An Average calculated from N values.
*/
template<size_t N>
class AverageN : public Average
{
public:
AverageN(Average* pDependant = NULL)
: Average(pDependant)
, m_end(m_begin + N)
, m_i(m_begin)
, m_sum(0)
, m_nValues(0)
{
}
bool add_value(uint8_t value)
{
if (m_nValues == N)
{
// If as many values that fit has been added, then remove the
// least recent value from the sum.
m_sum -= *m_i;
}
else
{
// Otherwise make a note that a new value is added.
++m_nValues;
}
*m_i = value;
m_sum += *m_i; // Update the sum of all values.
m_i = next(m_i);
uint32_t average = m_sum / m_nValues;
set_value(average);
if (m_pDependant)
{
if (m_i == m_begin)
{
// If we have looped around we have performed a full cycle and will
// add a new value to the dependant average.
m_pDependant->add_value(average);
}
else
{
// Otherwise we just update the most recent value.
m_pDependant->update_value(average);
}
}
return m_i == m_begin;
}
void update_value(uint8_t value)
{
if (m_nValues == 0)
{
// If no values have been added yet, there's nothing to update but we
// need to add the value.
add_value(value);
}
else
{
// Otherwise we update the most recent value.
uint8_t* p = prev(m_i);
m_sum -= *p;
*p = value;
m_sum += *p;
uint32_t average = m_sum / m_nValues;
set_value(average);
if (m_pDependant)
{
m_pDependant->update_value(average);
}
}
}
private:
uint8_t* prev(uint8_t* p)
{
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
if (p > m_begin)
{
--p;
}
else
{
ss_dassert(p == m_begin);
p = m_end - 1;
}
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
return p;
}
uint8_t* next(uint8_t* p)
{
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
++p;
if (p == m_end)
{
p = m_begin;
}
ss_dassert(p >= m_begin);
ss_dassert(p < m_end);
return p;
}
private:
uint8_t m_begin[N]; /*< Buffer containing values from which the average is calculated. */
uint8_t* m_end; /*< Points to one past the end of the buffer. */
uint8_t* m_i; /*< Current position in the buffer. */
uint32_t m_sum; /*< Sum of all values in the buffer. */
uint32_t m_nValues; /*< How many values the buffer contains. */
};
uint64_t m_start_time; /*< When was the current 1-second period started. */
uint64_t m_wait_start; /*< The time when the worker entered epoll_wait(). */
uint64_t m_wait_time; /*< How much time the worker has spent in epoll_wait(). */
AverageN<60> m_load_1_hour; /*< The average load during the last hour. */
AverageN<60> m_load_1_minute; /*< The average load during the last minute. */
Average1 m_load_1_second; /*< The load during the last 1-second period. */
};
class Worker : public MXS_WORKER
, private MessageQueue::Handler
{
Worker(const Worker&) = delete;
Worker& operator = (const Worker&) = delete;
public:
typedef WORKER_STATISTICS STATISTICS;
typedef WorkerTask Task;
typedef WorkerDisposableTask DisposableTask;
typedef WorkerLoad Load;
enum state_t
{
STOPPED,
IDLE,
POLLING,
PROCESSING,
ZPROCESSING
};
enum execute_mode_t
{
EXECUTE_AUTO, /**< Execute tasks immediately */
EXECUTE_QUEUED /**< Only queue tasks for execution */
};
/**
* Initialize the worker mechanism.
*
* To be called once at process startup. This will cause as many workers
* to be created as the number of threads defined.
*
* @return True if the initialization succeeded, false otherwise.
*/
static bool init();
/**
* Finalize the worker mechanism.
*
* To be called once at process shutdown. This will cause all workers
* to be destroyed. When the function is called, no worker should be
* running anymore.
*/
static void finish();
/**
* Returns the id of the worker
*
* @return The id of the worker.
*/
int id() const
{
return m_id;
}
int load(Load::counter_t counter)
{
return m_load.percentage(counter);
}
/**
* Returns the state of the worker.
*
* @return The current state.
*
* @attentions The state might have changed the moment after the function returns.
*/
state_t state() const
{
return m_state;
}
/**
* Returns statistics for this worker.
*
* @return The worker specific statistics.
*
* @attentions The statistics may change at any time.
*/
const STATISTICS& statistics() const
{
return m_statistics;
}
/**
* Returns statistics for all workers.
*
* @return Combined statistics.
*
* @attentions The statistics may no longer be accurate by the time it has
* been returned. The returned values may also not represent a
* 100% consistent set.
*/
static STATISTICS get_statistics();
/**
* Return a specific combined statistic value.
*
* @param what What to return.
*
* @return The corresponding value.
*/
static int64_t get_one_statistic(POLL_STAT what);
/**
* Return this worker's statistics.
*
* @return Local statistics for this worker.
*/
const STATISTICS& get_local_statistics() const
{
return m_statistics;
}
/**
* Return the count of descriptors.
*
* @param pnCurrent On output the current number of descriptors.
* @param pnTotal On output the total number of descriptors.
*/
void get_descriptor_counts(uint32_t* pnCurrent, uint64_t* pnTotal);
/**
* Add a file descriptor to the epoll instance of the worker.
*
* @param fd The file descriptor to be added.
* @param events Mask of epoll event types.
* @param pData The poll data associated with the descriptor:
*
* data->handler : Handler that knows how to deal with events
* for this particular type of 'struct mxs_poll_data'.
* data->thread.id: Will be updated by the worker.
*
* @attention The provided file descriptor must be non-blocking.
* @attention @c pData must remain valid until the file descriptor is
* removed from the worker.
*
* @return True, if the descriptor could be added, false otherwise.
*/
bool add_fd(int fd, uint32_t events, MXS_POLL_DATA* pData);
/**
* Remove a file descriptor from the worker's epoll instance.
*
* @param fd The file descriptor to be removed.
*
* @return True on success, false on failure.
*/
bool remove_fd(int fd);
/**
* Main function of worker.
*
* The worker will run the poll loop, until it is told to shut down.
*
* @attention This function will run in the calling thread.
*/
void run();
/**
* Run worker in separate thread.
*
* This function will start a new thread, in which the `run`
* function will be executed.
*
* @param stack_size The stack size of the new thread. A value of 0 means
* that the pthread default should be used.
*
* @return True if the thread could be started, false otherwise.
*/
bool start(size_t stack_size = 0);
/**
* Waits for the worker to finish.
*/
void join();
/**
* Initate shutdown of worker.
*
* @attention A call to this function will only initiate the shutdowm,
* the worker will not have shut down when the function returns.
*
* @attention This function is signal safe.
*/
void shutdown();
/**
* Query whether worker should shutdown.
*
* @return True, if the worker should shut down, false otherwise.
*/
bool should_shutdown() const
{
return m_should_shutdown;
}
/**
* Posts a task to a worker for execution.
*
* @param pTask The task to be executed.
* @param pSem If non-NULL, will be posted once the task's `execute` return.
* @param mode Execution mode
*
* @return True if the task could be posted (i.e. not executed), false otherwise.
*
* @attention The instance must remain valid for as long as it takes for the
* task to be transferred to the worker and its `execute` function
* to be called.
*
* The semaphore can be used for waiting for the task to be finished.
*
* @code
* Semaphore sem;
* MyTask task;
*
* pWorker->execute(&task, &sem);
* sem.wait();
*
* MyResult& result = task.result();
* @endcode
*/
bool post(Task* pTask, Semaphore* pSem = NULL, enum execute_mode_t mode = EXECUTE_AUTO);
/**
* Posts a task to a worker for execution.
*
* @param pTask The task to be executed.
* @param mode Execution mode
*
* @return True if the task could be posted (i.e. not executed), false otherwise.
*
* @attention Once the task has been executed, it will be deleted.
*/
bool post(std::auto_ptr<DisposableTask> sTask, enum execute_mode_t mode = EXECUTE_AUTO);
template<class T>
bool post(std::auto_ptr<T> sTask, enum execute_mode_t mode = EXECUTE_AUTO)
{
return post(std::auto_ptr<DisposableTask>(sTask.release()), mode);
}
/**
* Posts a task to all workers for execution.
*
* @param pTask The task to be executed.
* @param pSem If non-NULL, will be posted once per worker when the task's
* `execute` return.
*
* @return How many workers the task was posted to.
*
* @attention The very same task will be posted to all workers. The task
* should either not have any sharable data or then it should
* have data specific to each worker that can be accessed
* without locks.
*/
static size_t broadcast(Task* pTask, Semaphore* pSem = NULL);
/**
* Posts a task to all workers for execution.
*
* @param pTask The task to be executed.
*
* @return How many workers the task was posted to.
*
* @attention The very same task will be posted to all workers. The task
* should either not have any sharable data or then it should
* have data specific to each worker that can be accessed
* without locks.
*
* @attention Once the task has been executed by all workers, it will
* be deleted.
*/
static size_t broadcast(std::auto_ptr<DisposableTask> sTask);
template<class T>
static size_t broadcast(std::auto_ptr<T> sTask)
{
return broadcast(std::auto_ptr<DisposableTask>(sTask.release()));
}
/**
* Executes a task on all workers in serial mode (the task is executed
* on at most one worker thread at a time). When the function returns
* the task has been executed on all workers.
*
* @param task The task to be executed.
*
* @return How many workers the task was posted to.
*
* @warning This function is extremely inefficient and will be slow compared
* to the other functions. Only use this function when printing thread-specific
* data to stdout.
*/
static size_t execute_serially(Task& task);
/**
* Executes a task on all workers concurrently and waits until all workers
* are done. That is, when the function returns the task has been executed
* by all workers.
*
* @param task The task to be executed.
*
* @return How many workers the task was posted to.
*/
static size_t execute_concurrently(Task& task);
/**
* Post a message to a worker.
*
* @param msg_id The message id.
* @param arg1 Message specific first argument.
* @param arg2 Message specific second argument.
*
* @return True if the message could be sent, false otherwise. If the message
* posting fails, errno is set appropriately.
*
* @attention The return value tells *only* whether the message could be sent,
* *not* that it has reached the worker.
*
* @attention This function is signal safe.
*/
bool post_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Broadcast a message to all worker.
*
* @param msg_id The message id.
* @param arg1 Message specific first argument.
* @param arg2 Message specific second argument.
*
* @return The number of messages posted; if less that ne number of workers
* then some postings failed.
*
* @attention The return value tells *only* whether message could be posted,
* *not* that it has reached the worker.
*
* @attentsion Exactly the same arguments are passed to all workers. Take that
* into account if the passed data must be freed.
*
* @attention This function is signal safe.
*/
static size_t broadcast_message(uint32_t msg_id, intptr_t arg1, intptr_t arg2);
/**
* Initate shutdown of all workers.
*
* @attention A call to this function will only initiate the shutdowm,
* the workers will not have shut down when the function returns.
*
* @attention This function is signal safe.
*/
static void shutdown_all();
/**
* Return the worker associated with the provided worker id.
*
* @param worker_id A worker id.
*
* @return The corresponding worker instance, or NULL if the id does
* not correspond to a worker.
*/
static Worker* get(int worker_id);
/**
* Return the worker associated with the current thread.
*
* @return The worker instance, or NULL if the current thread does not have a worker.
*/
static Worker* get_current();
/**
* Return the worker id associated with the current thread.
*
* @return A worker instance, or -1 if the current thread does not have a worker.
*/
static int get_current_id();
protected:
Worker();
virtual ~Worker();
/**
* Called by Worker::run() before starting the epoll loop.
*
* @return True, if the epoll loop should be started, false otherwise.
*/
virtual bool pre_run() = 0;
/**
* Called by Worker::run() after the epoll loop has finished.
*/
virtual void post_run() = 0;
/**
* Called by Worker::run() once per epoll loop.
*/
virtual void epoll_tick() = 0;
/**
* Helper for resolving epoll-errors. In case of fatal ones, SIGABRT
* will be raised.
*
* @param fd The epoll file descriptor.
* @param errnum The errno of the operation.
* @param op Either EPOLL_CTL_ADD or EPOLL_CTL_DEL.
*/
static void resolve_poll_error(int fd, int err, int op);
protected:
const int m_id; /*< The id of the worker. */
const int m_epoll_fd; /*< The epoll file descriptor. */
state_t m_state; /*< The state of the worker */
private:
bool post_disposable(DisposableTask* pTask, enum execute_mode_t mode = EXECUTE_AUTO);
void handle_message(MessageQueue& queue, const MessageQueue::Message& msg); // override
static void thread_main(void* arg);
void poll_waitevents();
private:
STATISTICS m_statistics; /*< Worker statistics. */
MessageQueue* m_pQueue; /*< The message queue of the worker. */
THREAD m_thread; /*< The thread handle of the worker. */
bool m_started; /*< Whether the thread has been started or not. */
bool m_should_shutdown; /*< Whether shutdown should be performed. */
bool m_shutdown_initiated; /*< Whether shutdown has been initated. */
uint32_t m_nCurrent_descriptors; /*< Current number of descriptors. */
uint64_t m_nTotal_descriptors; /*< Total number of descriptors. */
Load m_load;
};
}

8
server/core/internal/workertask.hh
View File

@ -17,7 +17,7 @@
namespace maxscale
{
class RoutingWorker;
class Worker;
/**
* A WorkerTask represents a task to be performed by a Worker.
@ -38,7 +38,7 @@ public:
* @attention As the function is called by a worker, the body of `execute`
* should execute quickly and not perform any blocking operations.
*/
virtual void execute(RoutingWorker& worker) = 0;
virtual void execute(Worker& worker) = 0;
};
/**
@ -69,10 +69,10 @@ protected:
* @attention As the function is called by a worker, the body of `execute`
* should execute quickly and not perform any blocking operations.
*/
virtual void execute(RoutingWorker& worker) = 0;
virtual void execute(Worker& worker) = 0;
private:
friend class RoutingWorker;
friend class Worker;
void inc_ref();
void dec_ref();

6
server/core/messagequeue.cc
View File

@ -165,7 +165,7 @@ bool MessageQueue::post(const Message& message) const
return rv;
}
bool MessageQueue::add_to_worker(RoutingWorker* pWorker)
bool MessageQueue::add_to_worker(Worker* pWorker)
{
if (m_pWorker)
{
@ -181,9 +181,9 @@ bool MessageQueue::add_to_worker(RoutingWorker* pWorker)
return m_pWorker != NULL;
}
RoutingWorker* MessageQueue::remove_from_worker()
Worker* MessageQueue::remove_from_worker()
{
RoutingWorker* pWorker = m_pWorker;
Worker* pWorker = m_pWorker;
if (m_pWorker)
{

19
server/core/poll.cc
View File

@ -39,6 +39,7 @@
#include "internal/routingworker.hh"
using maxscale::Worker;
using maxscale::RoutingWorker;
static int n_threads; /*< Number of threads */
@ -63,17 +64,17 @@ static bool add_fd_to_worker(int wid, int fd, uint32_t events, MXS_POLL_DATA* da
return worker->add_fd(fd, events, data);
}
static bool add_fd_to_workers(int fd, uint32_t events, MXS_POLL_DATA* data)
static bool add_fd_to_routing_workers(int fd, uint32_t events, MXS_POLL_DATA* data)
{
bool rv = true;
int thread_id = data->thread.id;
rv = Worker::add_shared_fd(fd, events, data);
rv = RoutingWorker::add_shared_fd(fd, events, data);
if (rv)
{
// The DCB will appear on the list of the calling thread.
int wid = Worker::get_current_id();
int wid = RoutingWorker::get_current_id();
if (wid == -1)
{
@ -101,7 +102,7 @@ bool poll_add_fd_to_worker(int wid, int fd, uint32_t events, MXS_POLL_DATA* data
if (wid == MXS_WORKER_ALL)
{
rv = add_fd_to_workers(fd, events, data);
rv = add_fd_to_routing_workers(fd, events, data);
}
else
{
@ -123,9 +124,9 @@ static bool remove_fd_from_worker(int wid, int fd)
return worker->remove_fd(fd);
}
static bool remove_fd_from_workers(int fd)
static bool remove_fd_from_routing_workers(int fd)
{
return Worker::remove_shared_fd(fd);
return RoutingWorker::remove_shared_fd(fd);
}
bool poll_remove_fd_from_worker(int wid, int fd)
@ -134,7 +135,7 @@ bool poll_remove_fd_from_worker(int wid, int fd)
if (wid == MXS_WORKER_ALL)
{
rv = remove_fd_from_workers(fd);
rv = remove_fd_from_routing_workers(fd);
}
else
{
@ -156,7 +157,7 @@ bool poll_remove_fd_from_worker(int wid, int fd)
void
poll_set_nonblocking_polls(unsigned int nbpolls)
{
Worker::set_nonblocking_polls(nbpolls);
RoutingWorker::set_nonblocking_polls(nbpolls);
}
/**
@ -169,7 +170,7 @@ poll_set_nonblocking_polls(unsigned int nbpolls)
void
poll_set_maxwait(unsigned int maxwait)
{
Worker::set_maxwait(maxwait);
RoutingWorker::set_maxwait(maxwait);
}
/**

3
server/core/resource.cc
View File

@ -34,7 +34,6 @@
#include "internal/service.h"
#include "internal/config_runtime.h"
#include "internal/modules.h"
#include "internal/worker.h"
#include "internal/routingworker.hh"
using std::list;
@ -128,7 +127,7 @@ bool Resource::matching_variable_path(const string& path, const string& target)
char* end;
int id = strtol(target.c_str(), &end, 10);
if (*end == '\0' && mxs_worker_get(id))
if (*end == '\0' && mxs_rworker_get(id))
{
rval = true;
}

1205
server/core/routingworker.cc
View File

File diff suppressed because it is too large Load Diff

4
server/core/service.cc
View File

@ -50,7 +50,7 @@
#include <maxscale/version.h>
#include <maxscale/jansson.h>
#include <maxscale/json_api.h>
#include <maxscale/worker.h>
#include <maxscale/routingworker.h>
#include "internal/config.h"
#include "internal/filter.h"
@ -1695,7 +1695,7 @@ int service_refresh_users(SERVICE *service)
{
ss_dassert(service);
int ret = 1;
int self = mxs_worker_get_current_id();
int self = mxs_rworker_get_current_id();
ss_dassert(self >= 0);
time_t now = time(NULL);

1071
server/core/worker.cc Normal file
View File

File diff suppressed because it is too large Load Diff

4
server/modules/authenticator/MySQLAuth/mysql_auth.c
View File

@ -34,7 +34,7 @@
#include <maxscale/paths.h>
#include <maxscale/secrets.h>
#include <maxscale/utils.h>
#include <maxscale/worker.h>
#include <maxscale/routingworker.h>
static void* mysql_auth_init(char **options);
static bool mysql_auth_set_protocol_data(DCB *dcb, GWBUF *buf);
@ -137,7 +137,7 @@ static bool open_instance_database(const char *path, sqlite3 **handle)
sqlite3* get_handle(MYSQL_AUTH* instance)
{
int i = mxs_worker_get_current_id();
int i = mxs_rworker_get_current_id();
ss_dassert(i >= 0);
if (instance->handles[i] == NULL)

6
server/modules/protocol/MySQL/mariadbclient/mysql_client.cc
View File

@ -33,10 +33,10 @@
#include <maxscale/protocol/mysql.h>
#include <maxscale/query_classifier.h>
#include <maxscale/router.h>
#include <maxscale/routingworker.h>
#include <maxscale/session.h>
#include <maxscale/ssl.h>
#include <maxscale/utils.h>
#include <maxscale/worker.h>
#include "setparser.hh"
#include "sqlmodeparser.hh"
@ -711,7 +711,7 @@ gw_read_do_authentication(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
// For the time being only the sql_mode is stored in MXS_SESSION::client_protocol_data.
session->client_protocol_data = QC_SQL_MODE_DEFAULT;
protocol->protocol_auth_state = MXS_AUTH_STATE_COMPLETE;
ss_debug(bool check = ) mxs_worker_register_session(session);
ss_debug(bool check = ) mxs_rworker_register_session(session);
ss_dassert(check);
mxs_mysql_send_ok(dcb, next_sequence, 0, NULL);
@ -1439,7 +1439,7 @@ static int gw_client_close(DCB *dcb)
{
ss_dassert(target->state == SESSION_STATE_ROUTER_READY ||
target->state == SESSION_STATE_STOPPING);
ss_debug(bool removed = ) mxs_worker_deregister_session(target->ses_id);
ss_debug(bool removed = ) mxs_rworker_deregister_session(target->ses_id);
ss_dassert(removed);
session_close(target);
}

10
server/modules/routing/binlogrouter/blr_master.c
View File

@ -44,13 +44,13 @@
#include <maxscale/log_manager.h>
#include <maxscale/protocol/mysql.h>
#include <maxscale/router.h>
#include <maxscale/routingworker.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/session.h>
#include <maxscale/spinlock.h>
#include <maxscale/thread.h>
#include <maxscale/utils.h>
#include <maxscale/worker.h>
static GWBUF *blr_make_query(DCB *dcb, char *query);
static GWBUF *blr_make_registration(ROUTER_INSTANCE *router);
@ -149,7 +149,7 @@ typedef enum
static void blr_start_master(void* data)
{
ROUTER_INSTANCE *router = (ROUTER_INSTANCE*)data;
ss_dassert(mxs_worker_get_current_id() == 0);
ss_dassert(mxs_rworker_get_current() == mxs_rworker_get(MXS_RWORKER_MAIN));
if (router->client)
{
@ -235,7 +235,7 @@ static void blr_start_master(void* data)
* 'client' is the fake DCB that emulates a client session:
* we need to set the poll.thread.id for the "dummy client"
*/
client->session->client_dcb->poll.thread.id = mxs_worker_get_current_id();
client->session->client_dcb->poll.thread.id = mxs_rworker_get_current_id();
/* Connect to configured master server */
if ((router->master = dcb_connect(router->service->dbref->server,
@ -329,7 +329,7 @@ bool blr_start_master_in_main(void* data)
// The master should be connected to in the main worker, so we post it a
// message and call `blr_start_master` there.
MXS_WORKER* worker = mxs_worker_get(0); // The worker running in the main thread.
MXS_WORKER* worker = mxs_rworker_get(MXS_RWORKER_MAIN); // The worker running in the main thread.
ss_dassert(worker);
intptr_t arg1 = (intptr_t)worker_cb_start_master;
@ -367,7 +367,7 @@ void blr_close_master_in_main(void* data)
// The master should be connected to in the main worker, so we post it a
// message and call `blr_master_close` there.
MXS_WORKER* worker = mxs_worker_get(0); // The worker running in the main thread.
MXS_WORKER* worker = mxs_rworker_get(MXS_RWORKER_MAIN); // The worker running in the main thread.
ss_dassert(worker);
intptr_t arg1 = (intptr_t)worker_cb_close_master;

18
server/modules/routing/debugcli/debugcmd.c
View File

@ -52,7 +52,7 @@
#include <maxscale/users.h>
#include <maxscale/utils.h>
#include <maxscale/version.h>
#include <maxscale/worker.h>
#include <maxscale/routingworker.h>
#include <debugcli.h>
@ -876,21 +876,9 @@ static void cmd_AddServer(DCB *dcb, SERVER *server, char *v1, char *v2, char *v3
*/
void ping_workers(DCB* dcb)
{
int n_workers = config_threadcount();
int n = mxs_rworker_broadcast_message(MXS_WORKER_MSG_PING, 0, 0);
for (int i = 0; i < n_workers; ++i)
{
MXS_WORKER *worker = mxs_worker_get(i);
if (mxs_worker_post_message(worker, MXS_WORKER_MSG_PING, 0, 0))
{
dcb_printf(dcb, "Posted message to worker %d.\n", i);
}
else
{
dcb_printf(dcb, "Could not post message to worker %d: %s\n", i, mxs_strerror(errno));
}
}
dcb_printf(dcb, "Broadcasted ping message to %d workers.\n", n);
}
struct subcommand pingoptions[] =