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

MXS-1489 Create mechanism for running concurrent tasks

Browse Source 
This commit introduces maxscale::future, maxscale::packaged_task
and maxscale::thread that are modeled after C++11 std::future,
std::packaged_task and std::thread as described here:
http://en.cppreference.com/w/cpp/thread

The standard classes rely upon rvalue references (and move
constructors) introduced by C++11. As the C++ compilers we must use
are pre-C++11 that feature is obviously not present. The absence of
rvalue references is circumvented by implementing regular copy
constructors and assignment operators as if the arguments were rvalue
references.

In practice the above means that when one of these objects are copied,
the state is _moved_ rendering the copied object in default initialized
state. Some care is needed to ensure that unintended copying does not
occur.
This commit is contained in:
Johan Wikman 2017-11-07 10:03:23 +02:00
parent a96c8e1326
commit d7b8e95234
5 changed files with 740 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

360
include/maxscale/future.hh Normal file
View File

@ -0,0 +1,360 @@
#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 <maxscale/debug.h>
#include <maxscale/semaphore.hh>
namespace maxscale
{
// Internal, not intended for public consumption.
class future_internal
{
public:
virtual ~future_internal() {}
};
/**
* The class template maxscale::future provides a mechanism to access the result of
* asynchronous operations. It is based upon C++11's std::future as documented here:
* http://en.cppreference.com/w/cpp/thread/future
*
* std::future uses C++11's rvalue references, which are not available on the
* environments where MaxScale is compiled. Consequently, some care is needed
* when using maxscale::future so that unintended copying does not occur.
*
* When C++11 is available, it should be straightforward to take the std::future
* into use.
*/
template<class T>
class future
{
public:
/**
* Constructs a future with no shared state.
*/
future()
{
}
/**
* Move constructor
*
* @note This looks like a regular copy-constructor, but should be treated as
* a move constructor.
*
* @param other The future to move. After the call, @c other will not refer
* to a future result.
*/
future(const future& other)
: m_sInternal(other.m_sInternal)
{
other.m_sInternal.reset();
}
/**
* Move a future.
*
* @note This looks like a regular assignment operator, but the assigned value
* is moved.
*
* @param rhs The future to move. After the call, @c rhs will not refer
* to a future result.
* @return *this
*/
future& operator = (const future& rhs)
{
future copy(rhs);
copy.swap(*this);
return *this;
}
/**
* Destructor
*/
~future()
{
}
/**
* Swap the content
*
* @param rhs The future to swap the contents with.
*/
void swap(future& rhs)
{
std::swap(m_sInternal, rhs.m_sInternal);
}
/**
* Checks if the future refers to a shared state
*
* @return True, if this future refers to shared state, false otherwise.
*/
bool valid() const
{
return m_sInternal.get() != NULL;
}
/**
* Waits until the future has a valid result and returns it.
*
* @note After the function returns, the future will no longer be valid.
*
* @return The stored value.
*/
T get()
{
ss_dassert(valid());
if (valid())
{
T rv = m_sInternal->get();
m_sInternal.reset();
return rv;
}
else
{
MXS_ERROR("Get called on non-valid future.");
return T();
}
}
/**
* Blocks until the result becomes available.
*
* @note Only a valid future can be waited for.
*/
void wait() const
{
ss_dassert(valid());
if (valid())
{
m_sInternal->wait();
}
else
{
MXS_ERROR("An attempt to wait on a non-valid future.");
}
}
public:
class internal : public future_internal
{
public:
internal()
: m_t()
, m_waited(false)
{
}
~internal()
{
}
T get()
{
wait();
return m_t;
}
void set(T t)
{
m_t = t;
m_sem.post();
}
void wait() const
{
if (!m_waited)
{
m_sem.wait();
m_waited = true;
}
}
private:
T m_t;
mutable bool m_waited;
Semaphore m_sem;
};
future(std::shared_ptr<internal> sInternal)
: m_sInternal(sInternal)
{
}
private:
mutable std::shared_ptr<internal> m_sInternal;
};
/**
* The class template std::packaged_task wraps a function so that it can be called
* asynchronously. It is based upon C++11 std::packaged_task as documented here:
* http://en.cppreference.com/w/cpp/thread/packaged_task
*
* std::packaged_task uses C++11's rvalue references, which are not available on the
* environments where MaxScale is compiled. Consequently, some care is needed
* when using maxscale::packaged_task so that unintended copying does not occur.
*
* When C++11 is available, it should be straightforward to take the std::packed_task
* into use.
*
* Contrary to std::packaged_task, also due to lack of functionality introduced by
* C++11, maxscale::packaged_task is not fully generic, but can only package a function
* returning a value and taking one argument.
*/
template<class R, class T>
class packaged_task
{
typedef typename future<R>::internal internal;
public:
typedef R (*function)(T);
/**
* Creates a packaged_task with no task and no shared state.
*/
packaged_task()
: m_f(NULL)
, m_get_future_called(false)
{
}
/**
* Creates a packaged_task referring to the provided function.
*
* @param f The function to package.
*/
packaged_task(function f)
: m_f(f)
, m_sInternal(new internal)
, m_get_future_called(false)
{
}
/**
* Move constructor
*
* @note This looks like a regular copy-constructor, but should be treated as
* a move constructor.
*
* @param other The packaged_task to move. After the call, @c other will not
* refer to a packaged task.
*/
packaged_task(const packaged_task& other)
: m_f(other.m_f)
, m_sInternal(other.m_sInternal)
, m_get_future_called(other.m_get_future_called)
{
other.m_f = NULL;
other.m_sInternal.reset();
other.m_get_future_called = false;
}
/**
* Move a packaged_task
*
* @note This looks like a regular assignment operator, but the assigned value
* is moved.
*
* @param rhs The packaged_task to move. After the call, @c rhs will not
* refer to a packaged task.
*/
packaged_task& operator = (const packaged_task& rhs)
{
packaged_task copy(rhs);
copy.swap(*this);
return *this;
}
/**
* Destructor
*/
~packaged_task()
{
if (m_sInternal.get())
{
ss_dassert(m_get_future_called);
// The ownership of m_pFuture has moved to the future
// that was obtained in the call to get_future().
if (!m_get_future_called)
{
MXS_ERROR("Packaged task destructed without future result having been asked for.");
}
}
}
/**
* Swap the content
*
* @param rhs The packaged_task to swap the contents with.
*/
void swap(packaged_task& rhs)
{
std::swap(m_f, rhs.m_f);
std::swap(m_sInternal, rhs.m_sInternal);
std::swap(m_get_future_called, rhs.m_get_future_called);
}
/**
* Checks the validity of the packaged_task.
*
* @return True, if the packaged_task contains share state, false otherwise.
*/
bool valid() const
{
return m_sInternal.get() != NULL;
}
/**
* Returns a future which shares the same shared state as this packaged_task.
*
* @note @c get_future can be called only once for each packaged_task.
*
* @return A future.
*/
future<R> get_future()
{
ss_dassert(!m_get_future_called);
if (!m_get_future_called)
{
m_get_future_called = true;
return future<R>(m_sInternal);
}
else
{
MXS_ERROR("get_future called more than once.");
return future<R>();
}
};
/**
* Calls the stored task with the provided argument.
*
* After this call, anyone waiting for the shared result will be unblocked.
*/
void operator()(T arg)
{
m_sInternal->set(m_f(arg));
}
private:
mutable function m_f;
mutable std::shared_ptr<internal> m_sInternal;
mutable bool m_get_future_called;
};
}

163
include/maxscale/thread.hh Normal file
View File

@ -0,0 +1,163 @@
#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 <tr1/memory>
#include <maxscale/thread.h>
#include <maxscale/future.hh>
namespace maxscale
{
/**
* The class maxscale::thread represents a single thread of execution. It
* is based upon C++11's std::thread as documented here:
* http://en.cppreference.com/w/cpp/thread/thread
*
* std::thread uses C++11's rvalue references, which are not available on the
* environments where MaxScale is compiled. Consequently, some care is needed
* when using maxscale::thread so that unintended copying does not occur.
*
* When C++11 is available, it should be straightforward to take the std::thread
* into use.
*/
class thread
{
public:
/**
* Creates a thread object which does not represent a running thread.
*/
thread();
/**
* Move constructor
*
* @note This looks like a regular copy-constructor, but should be treated as
* a move constructor.
*
* @param other The thread to move. After the call, @c other will not refer
* to a thread.
*/
thread(const thread& other);
/**
* Creates a new thread object and associates it with a thread of execution.
* The new thread will executed the provided task using the provided argument.
*
* @note The actual thread is started by the constructor.
*
* @param task The task to execute in the new thread.
* @param arg The argument to provide to the task when invoked.
* Must remain valid for the lifetime of the thread.
*/
template<class R, class T>
thread(packaged_task<R, T>& task, T arg)
: m_pInternal(new internal(new task_packaged_task<R,T>(task, arg)))
{
run();
}
/**
* Move a thread
*
* @note This looks like a regular assignment operator, but the assigned value
* is moved.
*
* @param rhs The thread to move. After the call, @c rhs will not
* refer to a packaged task.
*/
thread& operator = (const thread& rhs);
/**
* Destructor
*
* The thread must have been joined before the thread object is destructed.
*/
~thread();
/**
* Whether a thread is joinable
*
* @return True, if the thread can be joined, false otherwise.
*/
bool joinable() const;
/**
* Join the thread.
*/
void join();
/**
* Swap the content
*
* @param rhs The thread to swap the contents with.
*/
void swap(thread& rhs);
private:
void run();
class task
{
public:
virtual ~task() {}
virtual void run() = 0;
};
template<class R, class T>
class task_packaged_task : public task
{
public:
task_packaged_task(packaged_task<R, T>& task, T arg)
: m_task(task)
, m_arg(arg)
{
}
void run()
{
m_task(m_arg);
}
private:
packaged_task<R, T> m_task;
T m_arg;
};
class internal
{
public:
internal(task* pTask);
~internal();
bool joinable() const;
void join();
void run();
private:
void main();
static void main(void* pArg);
private:
task* m_pTask;
THREAD m_thread;
};
private:
mutable internal* m_pInternal;
};
}

3
server/core/test/CMakeLists.txt
View File

@ -15,6 +15,7 @@ add_executable(test_semaphore testsemaphore.cc)
add_executable(test_server testserver.cc)
add_executable(test_service testservice.cc)
add_executable(test_spinlock testspinlock.cc)
add_executable(test_thread testthread.cc)
add_executable(test_trxcompare testtrxcompare.cc ../../../query_classifier/test/testreader.cc)
add_executable(test_trxtracking testtrxtracking.cc)
add_executable(test_users testusers.cc)
@ -42,6 +43,7 @@ target_link_libraries(test_semaphore maxscale-common)
target_link_libraries(test_server maxscale-common)
target_link_libraries(test_service maxscale-common)
target_link_libraries(test_spinlock maxscale-common)
target_link_libraries(test_thread maxscale-common)
target_link_libraries(test_trxcompare maxscale-common)
target_link_libraries(test_trxtracking maxscale-common)
target_link_libraries(test_users maxscale-common)
@ -71,6 +73,7 @@ add_test(TestSemaphore test_semaphore)
add_test(TestServer test_server)
add_test(TestService test_service)
add_test(TestSpinlock test_spinlock)
add_test(TestThread test_thread)
add_test(TestUsers test_users)
add_test(TestUtils test_utils)
add_test(TestModulecmd testmodulecmd)

100
server/core/test/testthread.cc Normal file
View File

@ -0,0 +1,100 @@
/*
* 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 <stdlib.h>
#include <iostream>
#include <vector>
#include <maxscale/thread.hh>
// We want asserts in release mode as well.
#if !defined(SS_DEBUG)
#define SS_DEBUG
#endif
#include <maxscale/debug.h>
using std::cout;
using std::endl;
using std::vector;
int function(int i)
{
return i / 2;
}
void test_basics()
{
cout << __func__ << endl;
mxs::packaged_task<int, int> t1;
ss_dassert(!t1.valid());
mxs::packaged_task<int, int> t2(function);
ss_dassert(t2.valid());
t1 = t2; // Move task.
ss_dassert(t1.valid());
ss_dassert(!t2.valid());
mxs::future<int> f1;
ss_dassert(!f1.valid());
mxs::future<int> f2 = t1.get_future();
ss_dassert(t1.valid());
ss_dassert(f2.valid());
f1 = f2; // Move future
ss_dassert(f1.valid());
ss_dassert(!f2.valid());
}
void test_running()
{
cout << __func__ << endl;
const int N = 10;
vector<mxs::future<int> > results;
vector<mxs::thread> threads;
cout << "Starting threads" << endl;
for (int i = 0; i < N; ++i)
{
cout << i << endl;
mxs::packaged_task<int, int> task(function);
mxs::future<int> r = task.get_future();
int arg = i;
mxs::thread t(task, arg);
results.push_back(r);
threads.push_back(t);
}
cout << "All threads started." << endl;
cout << "Waiting for threads." << endl;
for (int i = 0; i < N; ++i)
{
cout << i << endl;
threads[i].join();
int got = results[i].get();
int expected = function(i);
ss_dassert(got == expected);
}
}
int main()
{
test_basics();
test_running();
return EXIT_SUCCESS;
}

114
server/core/thread.cc
View File

@ -12,6 +12,7 @@
*/
#include <maxscale/thread.h>
#include <maxscale/thread.hh>
#include <maxscale/log_manager.h>
THREAD *thread_start(THREAD *thd, void (*entry)(void *), void *arg, size_t stack_size)
@ -68,3 +69,116 @@ void thread_millisleep(int ms)
req.tv_nsec = (ms % 1000) * 1000000;
nanosleep(&req, NULL);
}
//
// maxscale::thread
//
namespace maxscale
{
thread::thread()
: m_pInternal(NULL)
{
}
thread::thread(const thread& other)
: m_pInternal(other.m_pInternal)
{
other.m_pInternal = NULL;
}
thread& thread::operator = (const thread& rhs)
{
thread copy(rhs);
copy.swap(*this);
return *this;
}
thread::~thread()
{
ss_dassert(!joinable());
if (joinable())
{
MXS_ERROR("A thread that has not been joined is destructed.");
}
else
{
delete m_pInternal;
}
}
bool thread::joinable() const
{
return m_pInternal ? m_pInternal->joinable() : false;
}
void thread::join()
{
ss_dassert(m_pInternal);
if (!m_pInternal)
{
MXS_ERROR("Attempt to join a non-joinable thread.");
}
else
{
m_pInternal->join();
}
}
void thread::swap(thread& rhs)
{
std::swap(m_pInternal, rhs.m_pInternal);
}
void thread::run()
{
ss_dassert(m_pInternal);
m_pInternal->run();
}
thread::internal::internal(thread::task* pTask)
: m_pTask(pTask)
, m_thread(0)
{
}
thread::internal::~internal()
{
ss_info_dassert(!m_pTask, "Thread not joined before destructed.");
ss_dassert(m_thread == 0);
}
bool thread::internal::joinable() const
{
return m_thread != 0;
}
void thread::internal::join()
{
ss_dassert(joinable());
thread_wait(m_thread);
delete m_pTask;
m_pTask = NULL;
m_thread = 0;
}
void thread::internal::run()
{
if (!thread_start(&m_thread, &thread::internal::main, this, 0))
{
MXS_ALERT("Could not start thread, MaxScale is likely to malfunction.");
}
}
void thread::internal::main()
{
m_pTask->run();
}
void thread::internal::main(void* pArg)
{
static_cast<internal*>(pArg)->main();
}
}