Rework thread cancellation, using regular exception (#177)

BREAKS BACK COMPATIBILITY:
  - cancellation error can be caught by `pcall`
  - `canceled` thread status was renamed to `cancelled`
This commit is contained in:
mihacooper 2022-12-05 22:09:39 +01:00 committed by GitHub
parent 0aabb22587
commit ef93c6a2a8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 498 additions and 206 deletions

View File

@ -25,6 +25,7 @@ Requires C++14 compiler compliance. Tested with GCC 4.9+, clang 3.8 and Visual S
* [Important notes](#important-notes) * [Important notes](#important-notes)
* [Blocking and nonblocking operations](#blocking-and-nonblocking-operations) * [Blocking and nonblocking operations](#blocking-and-nonblocking-operations)
* [Function's upvalues](#functions-upvalues) * [Function's upvalues](#functions-upvalues)
* [Thread cancellation and pausing](#thread-cancellation-and-pausing)
* [API Reference](#api-reference) * [API Reference](#api-reference)
* [Thread](#thread) * [Thread](#thread)
* [effil.thread()](#runner--effilthreadfunc) * [effil.thread()](#runner--effilthreadfunc)
@ -45,6 +46,7 @@ Requires C++14 compiler compliance. Tested with GCC 4.9+, clang 3.8 and Visual S
* [effil.yield()](#effilyield) * [effil.yield()](#effilyield)
* [effil.sleep()](#effilsleeptime-metric) * [effil.sleep()](#effilsleeptime-metric)
* [effil.hardware_threads()](#effilhardware_threads) * [effil.hardware_threads()](#effilhardware_threads)
* [effil.pcall()](#status---effilpcallfunc)
* [Table](#table) * [Table](#table)
* [effil.table()](#table--effiltabletbl) * [effil.table()](#table--effiltabletbl)
* [__newindex: table[key] = value](#tablekey--value) * [__newindex: table[key] = value](#tablekey--value)
@ -71,6 +73,7 @@ Requires C++14 compiler compliance. Tested with GCC 4.9+, clang 3.8 and Visual S
* [effil.size()](#size--effilsizeobj) * [effil.size()](#size--effilsizeobj)
* [effil.type()](#effiltype) * [effil.type()](#effiltype)
# How to install # How to install
### Build from src on Linux and Mac ### Build from src on Linux and Mac
1. `git clone --recursive https://github.com/effil/effil effil` 1. `git clone --recursive https://github.com/effil/effil effil`
@ -256,7 +259,7 @@ local worker = effil.thread(function()
effil.sleep(999) -- worker will hang for 999 seconds effil.sleep(999) -- worker will hang for 999 seconds
end)() end)()
worker:cancel(1) -- returns true, cause blocking operation was interrupted and thread was canceled worker:cancel(1) -- returns true, cause blocking operation was interrupted and thread was cancelled
``` ```
</p> </p>
</details> </details>
@ -269,10 +272,71 @@ Working with function Effil can store function environment (`_ENV`) as well. Con
* *Lua = 5.1*: function environment is not stored at all (due to limitations of lua_setfenv we cannot use userdata) * *Lua = 5.1*: function environment is not stored at all (due to limitations of lua_setfenv we cannot use userdata)
* *Lua > 5.1*: Effil serialize and store function environment only if it's not equal to global environment (`_ENV ~= _G`). * *Lua > 5.1*: Effil serialize and store function environment only if it's not equal to global environment (`_ENV ~= _G`).
## Thread cancellation and pausing
The [`effil.thread`](#runner--effilthreadfunc) can be paused and cancelled using corresponding methods of thread object [`thread:cancel()`](#threadcanceltime-metric) and [`thread:pause()`](#threadpausetime-metric).
Thread that you try to interrupt can be interrupted in two execution points: explicit and implicit.
- Explicit points are [`effil.yield()`](#effilyield)
<details>
<summary>Example of explicit interruption point</summary>
<p>
```lua
local thread = effil.thread(function()
while true do
effil.yield()
end
-- will never reach this line
end)()
thread:cancel()
```
</p>
</details>
- Implicit points are lua debug hook invocation which is set using [lua_sethook](https://www.lua.org/manual/5.3/manual.html#lua_sethook) with LUA_MASKCOUNT.
Implicit points are optional and enabled only if [thread_runner.step](#runnerstep) > 0.
<details>
<summary>Example of implicit interruption point</summary>
<p>
```lua
local thread_runner = effil.thread(function()
while true do
end
-- will never reach this line
end)
thread_runner.step = 10
thread = thread_runner()
thread:cancel()
```
</p>
</details>
- Additionally thread can be cancelled (but not paused) in any [blocking or non-blocking waiting operation](#blocking-and-nonblocking-operations).
<details>
<summary>Example</summary>
<p>
```lua
local channel = effil.channel()
local thread = effil.thread(function()
channel:pop() -- thread hangs waiting infinitely
-- will never reach this line
end)()
thread:cancel()
```
</p>
</details>
**How does cancellation works?**
When you cancel thread it generates lua [`error`](https://lua.org.ru/manual_ru.html#pdf-error) with message `"Effil: thread is cancelled"` when it reaches any interruption point. It means that you can catch this error using [`pcall`](https://lua.org.ru/manual_ru.html#pdf-pcall) but thread will generate new error on next interruption point.
If you want to catch your own error but pass cancellation error you can use [effil.pcall()](#status---effilpcallfunc).
Status of cancelled thread will be equal to `cancelled` only if it finished with cancellation error. It means that if you catch cancellation error thread may finished with `completed` status or `failed` status if there will be some another error.
# API Reference # API Reference
## Thread ## Thread
`effil.thread` is the way to create a thread. Threads can be stopped, paused, resumed and canceled. `effil.thread` is the way to create a thread. Threads can be stopped, paused, resumed and cancelled.
All operation with threads can be synchronous (with optional timeout) or asynchronous. All operation with threads can be synchronous (with optional timeout) or asynchronous.
Each thread runs with its own lua state. Each thread runs with its own lua state.
@ -309,7 +373,7 @@ Thread handle provides API for interaction with thread.
Returns thread status. Returns thread status.
**output**: **output**:
- `status` - string values describes status of thread. Possible values are: `"running", "paused", "canceled", "completed" and "failed"`. - `status` - string values describes status of thread. Possible values are: `"running", "paused", "cancelled", "completed" and "failed"`.
- `err` - error message, if any. This value is specified only if thread status == `"failed"`. - `err` - error message, if any. This value is specified only if thread status == `"failed"`.
- `stacktrace` - stacktrace of failed thread. This value is specified only if thread status == `"failed"`. - `stacktrace` - stacktrace of failed thread. This value is specified only if thread status == `"failed"`.
@ -363,6 +427,17 @@ Returns the number of concurrent threads supported by implementation.
Basically forwards value from [std::thread::hardware_concurrency](https://en.cppreference.com/w/cpp/thread/thread/hardware_concurrency). Basically forwards value from [std::thread::hardware_concurrency](https://en.cppreference.com/w/cpp/thread/thread/hardware_concurrency).
**output**: number of concurrent hardware threads. **output**: number of concurrent hardware threads.
### `status, ... = effil.pcall(func, ...)`
Works exactly the same way as standard [pcall](https://www.lua.org/manual/5.3/manual.html#pdf-pcall) except that it will not catch thread cancellation error caused by [thread:cancel()](#threadcanceltime-metric) call.
**input:**
- func - function to call
- ... - arguments to pass to functions
**output:**
- status - `true` if no error occurred, `false` otherwise
- ... - in case of error return one additional result with message of error, otherwise return function call results
## Table ## Table
`effil.table` is a way to exchange data between effil threads. It behaves almost like standard lua tables. `effil.table` is a way to exchange data between effil threads. It behaves almost like standard lua tables.
All operations with shared table are thread safe. **Shared table stores** primitive types (number, boolean, string), function, table, light userdata and effil based userdata. **Shared table doesn't store** lua threads (coroutines) or arbitrary userdata. See examples of shared table usage [here](#examples) All operations with shared table are thread safe. **Shared table stores** primitive types (number, boolean, string), function, table, light userdata and effil based userdata. **Shared table doesn't store** lua threads (coroutines) or arbitrary userdata. See examples of shared table usage [here](#examples)

View File

@ -52,7 +52,7 @@ bool Channel::push(const sol::variadic_args& args) {
StoredArray Channel::pop(const sol::optional<int>& duration, StoredArray Channel::pop(const sol::optional<int>& duration,
const sol::optional<std::string>& period) { const sol::optional<std::string>& period) {
this_thread::interruptionPoint(); this_thread::cancellationPoint();
std::unique_lock<std::mutex> lock(ctx_->lock_); std::unique_lock<std::mutex> lock(ctx_->lock_);
{ {
this_thread::ScopedSetInterruptable interruptable(this); this_thread::ScopedSetInterruptable interruptable(this);
@ -70,7 +70,7 @@ StoredArray Channel::pop(const sol::optional<int>& duration,
else { // No time limit else { // No time limit
ctx_->cv_.wait(lock); ctx_->cv_.wait(lock);
} }
this_thread::interruptionPoint(); this_thread::cancellationPoint();
} }
} }

View File

@ -1,8 +1,9 @@
#include "threading.h" #include "thread.h"
#include "this-thread.h"
#include "thread-runner.h"
#include "shared-table.h" #include "shared-table.h"
#include "garbage-collector.h" #include "garbage-collector.h"
#include "channel.h" #include "channel.h"
#include "thread_runner.h"
#include <lua.hpp> #include <lua.hpp>
@ -100,6 +101,7 @@ int luaopen_effil(lua_State* L) {
"thread_id", this_thread::threadId, "thread_id", this_thread::threadId,
"sleep", this_thread::sleep, "sleep", this_thread::sleep,
"yield", this_thread::yield, "yield", this_thread::yield,
"pcall", this_thread::pcall,
"table", createTable, "table", createTable,
"rawset", SharedTable::luaRawSet, "rawset", SharedTable::luaRawSet,
"rawget", SharedTable::luaRawGet, "rawget", SharedTable::luaRawGet,
@ -115,7 +117,6 @@ int luaopen_effil(lua_State* L) {
"hardware_threads", std::thread::hardware_concurrency, "hardware_threads", std::thread::hardware_concurrency,
sol::meta_function::index, luaIndex sol::meta_function::index, luaIndex
); );
sol::stack::push(lua, type); sol::stack::push(lua, type);
sol::stack::pop<sol::object>(lua); sol::stack::pop<sol::object>(lua);
sol::stack::push(lua, EffilApiMarker()); sol::stack::push(lua, EffilApiMarker());

View File

@ -1,6 +1,6 @@
#pragma once #pragma once
#include <this_thread.h> #include <this-thread.h>
#include <lua-helpers.h> #include <lua-helpers.h>
#include <mutex> #include <mutex>
@ -29,20 +29,20 @@ public:
} }
void wait() { void wait() {
this_thread::interruptionPoint(); this_thread::cancellationPoint();
this_thread::ScopedSetInterruptable interruptable(this); this_thread::ScopedSetInterruptable interruptable(this);
std::unique_lock<std::mutex> lock(mutex_); std::unique_lock<std::mutex> lock(mutex_);
while (!notified_) { while (!notified_) {
cv_.wait(lock); cv_.wait(lock);
this_thread::interruptionPoint(); this_thread::cancellationPoint();
} }
} }
template <typename T> template <typename T>
bool waitFor(T period) { bool waitFor(T period) {
this_thread::interruptionPoint(); this_thread::cancellationPoint();
if (period == std::chrono::seconds(0) || notified_) if (period == std::chrono::seconds(0) || notified_)
return notified_; return notified_;
@ -54,7 +54,7 @@ public:
while (!timer.isFinished() && while (!timer.isFinished() &&
cv_.wait_for(lock, timer.left()) != std::cv_status::timeout && cv_.wait_for(lock, timer.left()) != std::cv_status::timeout &&
!notified_) { !notified_) {
this_thread::interruptionPoint(); this_thread::cancellationPoint();
} }
return notified_; return notified_;
} }

View File

@ -1,10 +1,10 @@
#include "stored-object.h" #include "stored-object.h"
#include "channel.h" #include "channel.h"
#include "threading.h" #include "thread.h"
#include "shared-table.h" #include "shared-table.h"
#include "function.h" #include "function.h"
#include "utils.h" #include "utils.h"
#include "thread_runner.h" #include "thread-runner.h"
#include <map> #include <map>
#include <vector> #include <vector>

83
src/cpp/this-thread.cpp Normal file
View File

@ -0,0 +1,83 @@
#include "this-thread.h"
#include "thread-handle.h"
#include "notifier.h"
namespace effil {
namespace this_thread {
ScopedSetInterruptable::ScopedSetInterruptable(IInterruptable* notifier) {
if (const auto thisThread = ThreadHandle::getThis()) {
thisThread->setNotifier(notifier);
}
}
ScopedSetInterruptable::~ScopedSetInterruptable() {
if (const auto thisThread = ThreadHandle::getThis()) {
thisThread->setNotifier(nullptr);
}
}
void cancellationPoint() {
const auto thisThread = ThreadHandle::getThis();
if (thisThread && thisThread->command() == ThreadHandle::Command::Cancel) {
thisThread->changeStatus(ThreadHandle::Status::Cancelled);
throw ThreadCancelException();
}
}
std::string threadId() {
std::stringstream ss;
ss << std::this_thread::get_id();
return ss.str();
}
void yield() {
if (const auto thisThread = ThreadHandle::getThis()) {
thisThread->performInterruptionPointThrow();
}
std::this_thread::yield();
}
void sleep(const sol::stack_object& duration, const sol::stack_object& metric) {
if (duration.valid()) {
REQUIRE(duration.get_type() == sol::type::number)
<< "bad argument #1 to 'effil.sleep' (number expected, got "
<< luaTypename(duration) << ")";
if (metric.valid())
{
REQUIRE(metric.get_type() == sol::type::string)
<< "bad argument #2 to 'effil.sleep' (string expected, got "
<< luaTypename(metric) << ")";
}
try {
Notifier notifier;
notifier.waitFor(fromLuaTime(duration.as<int>(),
metric.as<sol::optional<std::string>>()));
} RETHROW_WITH_PREFIX("effil.sleep");
}
else {
yield();
}
}
int pcall(lua_State* L)
{
int status;
luaL_checkany(L, 1);
status = lua_pcall(L, lua_gettop(L) - 1, LUA_MULTRET, 0);
const auto thisThread = ThreadHandle::getThis();
if (thisThread && thisThread->command() == ThreadHandle::Command::Cancel) {
lua_pushstring(L, ThreadCancelException::message);
lua_error(L);
}
lua_pushboolean(L, (status == 0));
lua_insert(L, 1);
return lua_gettop(L); /* return status + all results */
}
} // namespace this_thread
} // namespace effil

View File

@ -14,12 +14,12 @@ public:
ScopedSetInterruptable(IInterruptable* notifier); ScopedSetInterruptable(IInterruptable* notifier);
~ScopedSetInterruptable(); ~ScopedSetInterruptable();
}; };
void interruptionPoint();
// Lua API void cancellationPoint();
std::string threadId(); std::string threadId();
void yield(); void yield();
void sleep(const sol::stack_object& duration, const sol::stack_object& metric); void sleep(const sol::stack_object& duration, const sol::stack_object& metric);
int pcall(lua_State* L);
} // namespace this_thread } // namespace this_thread
} // namespace effil } // namespace effil

82
src/cpp/thread-handle.cpp Normal file
View File

@ -0,0 +1,82 @@
#include "thread-handle.h"
namespace effil {
// Thread specific pointer to current thread
static thread_local ThreadHandle* thisThreadHandle = nullptr;
static const sol::optional<std::chrono::milliseconds> NO_TIMEOUT;
ThreadHandle::ThreadHandle()
: status_(Status::Running)
, command_(Command::Run)
, currNotifier_(nullptr)
, lua_(std::make_unique<sol::state>())
{
luaL_openlibs(*lua_);
}
void ThreadHandle::putCommand(Command cmd) {
std::unique_lock<std::mutex> lock(stateLock_);
if (isFinishStatus(status_) || command() == Command::Cancel)
return;
command_ = cmd;
statusNotifier_.reset();
commandNotifier_.notify();
}
void ThreadHandle::changeStatus(Status stat) {
std::unique_lock<std::mutex> lock(stateLock_);
status_ = stat;
commandNotifier_.reset();
statusNotifier_.notify();
if (isFinishStatus(stat))
completionNotifier_.notify();
}
void ThreadHandle::performInterruptionPointImpl(const std::function<void(void)>& cancelClbk) {
switch (command()) {
case Command::Run:
break;
case Command::Cancel:
cancelClbk();
break;
case Command::Pause: {
changeStatus(Status::Paused);
Command cmd;
do {
cmd = waitForCommandChange(NO_TIMEOUT);
} while(cmd != Command::Run && cmd != Command::Cancel);
if (cmd == Command::Run) {
changeStatus(Status::Running);
} else {
cancelClbk();
}
break;
}
}
}
void ThreadHandle::performInterruptionPoint(lua_State* L) {
performInterruptionPointImpl([L](){
lua_pushstring(L, ThreadCancelException::message);
lua_error(L);
});
}
void ThreadHandle::performInterruptionPointThrow() {
performInterruptionPointImpl([](){
throw ThreadCancelException();
});
}
ThreadHandle* ThreadHandle::getThis() {
return thisThreadHandle;
}
void ThreadHandle::setThis(ThreadHandle* handle) {
assert(handle);
thisThreadHandle = handle;
}
} // namespace effil

View File

@ -1,19 +1,31 @@
#pragma once #pragma once
#include "lua-helpers.h" #include "lua-helpers.h"
#include "function.h"
#include "notifier.h" #include "notifier.h"
#include "gc-data.h"
#include <sol.hpp> #include <sol.hpp>
namespace effil { namespace effil {
class ThreadCancelException : public std::runtime_error
{
public:
static constexpr auto message = "Effil: thread is cancelled";
ThreadCancelException()
: std::runtime_error(message)
{}
};
class Thread;
class ThreadHandle : public GCData { class ThreadHandle : public GCData {
public: public:
enum class Status { enum class Status {
Running, Running,
Paused, Paused,
Canceled, Cancelled,
Completed, Completed,
Failed Failed
}; };
@ -29,6 +41,14 @@ public:
Command command() const { return command_; } Command command() const { return command_; }
void putCommand(Command cmd); void putCommand(Command cmd);
void changeStatus(Status stat); void changeStatus(Status stat);
void performInterruptionPoint(lua_State* L);
void performInterruptionPointThrow();
static ThreadHandle* getThis();
static bool isFinishStatus(Status stat) {
return stat == Status::Cancelled || stat == Status::Completed || stat == Status::Failed;
}
template <typename T> template <typename T>
Status waitForStatusChange(const sol::optional<T>& time) { Status waitForStatusChange(const sol::optional<T>& time) {
@ -90,38 +110,11 @@ private:
StoredArray result_; StoredArray result_;
IInterruptable* currNotifier_; IInterruptable* currNotifier_;
std::unique_ptr<sol::state> lua_; std::unique_ptr<sol::state> lua_;
void performInterruptionPointImpl(const std::function<void(void)>& cancelClbk);
static void setThis(ThreadHandle* handle);
friend class Thread;
}; };
class Thread : public GCObject<ThreadHandle> { } // namespace effil
public:
static void exportAPI(sol::state_view& lua);
StoredArray status(const sol::this_state& state);
StoredArray wait(const sol::this_state& state,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
StoredArray get(const sol::optional<int>& duration,
const sol::optional<std::string>& period);
bool cancel(const sol::this_state& state,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
bool pause(const sol::this_state&,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
void resume();
private:
Thread() = default;
void initialize(
const std::string& path,
const std::string& cpath,
int step,
const sol::function& function,
const sol::variadic_args& args);
friend class GC;
private:
static void runThread(Thread, Function, effil::StoredArray);
};
} // effil

View File

@ -1,4 +1,4 @@
#include "thread_runner.h" #include "thread-runner.h"
namespace effil { namespace effil {

View File

@ -1,4 +1,4 @@
#include "threading.h" #include "thread.h"
#include "gc-data.h" #include "gc-data.h"
#include "gc-object.h" #include "gc-object.h"

View File

@ -1,5 +1,6 @@
#include "threading.h" #include "thread.h"
#include "thread-handle.h"
#include "stored-object.h" #include "stored-object.h"
#include "notifier.h" #include "notifier.h"
#include "spin-mutex.h" #include "spin-mutex.h"
@ -15,27 +16,14 @@ using Command = ThreadHandle::Command;
namespace { namespace {
const sol::optional<std::chrono::milliseconds> NO_TIMEOUT;
// Thread specific pointer to current thread
static thread_local ThreadHandle* thisThreadHandle = nullptr;
// Doesn't inherit std::exception
// to prevent from catching this exception third party lua C++ libs
class LuaHookStopException {};
bool isFinishStatus(Status stat) {
return stat == Status::Canceled || stat == Status::Completed || stat == Status::Failed;
}
std::string statusToString(Status status) { std::string statusToString(Status status) {
switch (status) { switch (status) {
case Status::Running: case Status::Running:
return "running"; return "running";
case Status::Paused: case Status::Paused:
return "paused"; return "paused";
case Status::Canceled: case Status::Cancelled:
return "canceled"; return "cancelled";
case Status::Completed: case Status::Completed:
return "completed"; return "completed";
case Status::Failed: case Status::Failed:
@ -48,131 +36,26 @@ std::string statusToString(Status status) {
int luaErrorHandler(lua_State* state) { int luaErrorHandler(lua_State* state) {
luaL_traceback(state, state, nullptr, 1); luaL_traceback(state, state, nullptr, 1);
const auto stacktrace = sol::stack::pop<std::string>(state); const auto stacktrace = sol::stack::pop<std::string>(state);
thisThreadHandle->result().emplace_back(createStoredObject(stacktrace)); ThreadHandle::getThis()->result().emplace_back(createStoredObject(stacktrace));
return 1; return 1;
} }
const lua_CFunction luaErrorHandlerPtr = luaErrorHandler; const lua_CFunction luaErrorHandlerPtr = luaErrorHandler;
void luaHook(lua_State*, lua_Debug*) { void luaHook(lua_State* L, lua_Debug*) {
assert(thisThreadHandle); if (const auto thisThread = ThreadHandle::getThis()) {
switch (thisThreadHandle->command()) { thisThread->performInterruptionPoint(L);
case Command::Run:
break;
case Command::Cancel:
thisThreadHandle->changeStatus(Status::Canceled);
throw LuaHookStopException();
case Command::Pause: {
thisThreadHandle->changeStatus(Status::Paused);
Command cmd;
do {
cmd = thisThreadHandle->waitForCommandChange(NO_TIMEOUT);
} while(cmd != Command::Run && cmd != Command::Cancel);
if (cmd == Command::Run) {
thisThreadHandle->changeStatus(Status::Running);
} else {
thisThreadHandle->changeStatus(Status::Canceled);
throw LuaHookStopException();
}
break;
}
} }
} }
} // namespace } // namespace
namespace this_thread { void Thread::runThread(
Thread thread,
ScopedSetInterruptable::ScopedSetInterruptable(IInterruptable* notifier) { Function function,
if (thisThreadHandle) { effil::StoredArray arguments)
thisThreadHandle->setNotifier(notifier);
}
}
ScopedSetInterruptable::~ScopedSetInterruptable() {
if (thisThreadHandle) {
thisThreadHandle->setNotifier(nullptr);
}
}
void interruptionPoint() {
if (thisThreadHandle && thisThreadHandle->command() == Command::Cancel)
{
thisThreadHandle->changeStatus(Status::Canceled);
throw LuaHookStopException();
}
}
std::string threadId() {
std::stringstream ss;
ss << std::this_thread::get_id();
return ss.str();
}
void yield() {
luaHook(nullptr, nullptr);
std::this_thread::yield();
}
void sleep(const sol::stack_object& duration, const sol::stack_object& metric) {
if (duration.valid()) {
REQUIRE(duration.get_type() == sol::type::number)
<< "bad argument #1 to 'effil.sleep' (number expected, got "
<< luaTypename(duration) << ")";
if (metric.valid())
{
REQUIRE(metric.get_type() == sol::type::string)
<< "bad argument #2 to 'effil.sleep' (string expected, got "
<< luaTypename(metric) << ")";
}
try {
Notifier notifier;
notifier.waitFor(fromLuaTime(duration.as<int>(),
metric.as<sol::optional<std::string>>()));
} RETHROW_WITH_PREFIX("effil.sleep");
}
else {
yield();
}
}
} // namespace this_thread
ThreadHandle::ThreadHandle()
: status_(Status::Running)
, command_(Command::Run)
, currNotifier_(nullptr)
, lua_(std::make_unique<sol::state>())
{ {
luaL_openlibs(*lua_); ThreadHandle::setThis(thread.ctx_.get());
}
void ThreadHandle::putCommand(Command cmd) {
std::unique_lock<std::mutex> lock(stateLock_);
if (isFinishStatus(status_))
return;
command_ = cmd;
statusNotifier_.reset();
commandNotifier_.notify();
}
void ThreadHandle::changeStatus(Status stat) {
std::unique_lock<std::mutex> lock(stateLock_);
status_ = stat;
commandNotifier_.reset();
statusNotifier_.notify();
if (isFinishStatus(stat))
completionNotifier_.notify();
}
void Thread::runThread(Thread thread,
Function function,
effil::StoredArray arguments) {
thisThreadHandle = thread.ctx_.get();
assert(thisThreadHandle != nullptr);
try { try {
{ {
ScopeGuard reportComplete([thread, &arguments](){ ScopeGuard reportComplete([thread, &arguments](){
@ -193,7 +76,7 @@ void Thread::runThread(Thread thread,
sol::protected_function_result result = userFuncObj(std::move(arguments)); sol::protected_function_result result = userFuncObj(std::move(arguments));
if (!result.valid()) { if (!result.valid()) {
if (thread.ctx_->status() == Status::Canceled) if (thread.ctx_->status() == Status::Cancelled)
return; return;
sol::error err = result; sol::error err = result;
@ -213,15 +96,18 @@ void Thread::runThread(Thread thread,
} }
} }
thread.ctx_->changeStatus(Status::Completed); thread.ctx_->changeStatus(Status::Completed);
} catch (const LuaHookStopException&) {
thread.ctx_->changeStatus(Status::Canceled);
} catch (const std::exception& err) { } catch (const std::exception& err) {
DEBUG("thread") << "Failed with msg: " << err.what() << std::endl; if (thread.ctx_->command() == Command::Cancel && strcmp(err.what(), ThreadCancelException::message) == 0) {
auto& returns = thread.ctx_->result(); thread.ctx_->changeStatus(Status::Cancelled);
returns.insert(returns.begin(), } else {
{ createStoredObject("failed"), DEBUG("thread") << "Failed with msg: " << err.what() << std::endl;
createStoredObject(err.what()) }); auto& returns = thread.ctx_->result();
thread.ctx_->changeStatus(Status::Failed); returns.insert(returns.begin(), {
createStoredObject("failed"),
createStoredObject(err.what())
});
thread.ctx_->changeStatus(Status::Failed);
}
} }
} }
@ -319,7 +205,7 @@ bool Thread::cancel(const sol::this_state&,
ctx_->putCommand(Command::Cancel); ctx_->putCommand(Command::Cancel);
ctx_->interrupt(); ctx_->interrupt();
Status status = ctx_->waitForStatusChange(toOptionalTime(duration, period)); Status status = ctx_->waitForStatusChange(toOptionalTime(duration, period));
return isFinishStatus(status); return ThreadHandle::isFinishStatus(status);
} }
bool Thread::pause(const sol::this_state&, bool Thread::pause(const sol::this_state&,

43
src/cpp/thread.h Normal file
View File

@ -0,0 +1,43 @@
#pragma once
#include "lua-helpers.h"
#include "function.h"
#include "thread-handle.h"
#include <sol.hpp>
namespace effil {
class Thread : public GCObject<ThreadHandle> {
public:
static void exportAPI(sol::state_view& lua);
StoredArray status(const sol::this_state& state);
StoredArray wait(const sol::this_state& state,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
StoredArray get(const sol::optional<int>& duration,
const sol::optional<std::string>& period);
bool cancel(const sol::this_state& state,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
bool pause(const sol::this_state&,
const sol::optional<int>& duration,
const sol::optional<std::string>& period);
void resume();
private:
Thread() = default;
void initialize(
const std::string& path,
const std::string& cpath,
int step,
const sol::function& function,
const sol::variadic_args& args);
friend class GC;
private:
static void runThread(Thread, Function, effil::StoredArray);
};
} // effil

View File

@ -23,10 +23,10 @@ int luaopen_effil(lua_State* L);
namespace effil { namespace effil {
class Exception : public sol::error { class Exception : public std::runtime_error {
public: public:
Exception() noexcept Exception() noexcept
: sol::error("") {} : std::runtime_error("") {}
template <typename T> template <typename T>
Exception& operator<<(const T& value) { Exception& operator<<(const T& value) {

View File

@ -16,7 +16,7 @@ local function interruption_test(worker)
local start_time = os.time() local start_time = os.time()
thr:cancel(1) thr:cancel(1)
test.equal(thr:status(), "canceled") test.equal(thr:status(), "cancelled")
test.almost_equal(os.time(), start_time, 1) test.almost_equal(os.time(), start_time, 1)
state.stop = true state.stop = true
end end

View File

@ -6,7 +6,7 @@ test.thread_stress.time = function ()
local function check_time(real_time, use_time, metric) local function check_time(real_time, use_time, metric)
local start_time = os.time() local start_time = os.time()
effil.sleep(use_time, metric) effil.sleep(use_time, metric)
test.almost_equal(os.time(), start_time + real_time, 1) test.almost_equal(os.time(), start_time + real_time, 2)
end end
check_time(4, 4, nil) -- seconds by default check_time(4, 4, nil) -- seconds by default
check_time(4, 4, 's') check_time(4, 4, 's')

View File

@ -124,7 +124,7 @@ test.thread.cancel = function ()
)() )()
test.is_true(thread:cancel()) test.is_true(thread:cancel())
test.equal(thread:status(), "canceled") test.equal(thread:status(), "cancelled")
end end
test.thread.async_cancel = function () test.thread.async_cancel = function ()
@ -140,7 +140,7 @@ test.thread.async_cancel = function ()
thread:cancel(0) thread:cancel(0)
test.is_true(wait(2, function() return thread:status() ~= 'running' end)) test.is_true(wait(2, function() return thread:status() ~= 'running' end))
test.equal(thread:status(), 'canceled') test.equal(thread:status(), 'cancelled')
end end
test.thread.pause_resume_cancel = function () test.thread.pause_resume_cancel = function ()
@ -209,7 +209,7 @@ test.thread.async_pause_resume_cancel = function ()
test.is_true(wait(5, function() return (data.value - savedValue) > 100 end)) test.is_true(wait(5, function() return (data.value - savedValue) > 100 end))
thread:cancel(0) thread:cancel(0)
test.is_true(wait(5, function() return thread:status() == "canceled" end)) test.is_true(wait(5, function() return thread:status() == "cancelled" end))
thread:wait() thread:wait()
end end
@ -314,6 +314,30 @@ test.this_thread.functions = function ()
test.not_equal(share["child.id"], effil.thread_id()) test.not_equal(share["child.id"], effil.thread_id())
end end
test.this_thread.cancel_with_yield = function ()
local ctx = effil.table()
local spec = effil.thread(function()
while not ctx.stop do
-- Just waiting
end
ctx.done = true
while true do
effil.yield()
end
ctx.after_yield = true
end)
spec.step = 0
local thr = spec()
test.is_false(thr:cancel(1))
ctx.stop = true
test.is_true(thr:cancel())
test.equal(thr:status(), "cancelled")
test.is_true(ctx.done)
test.is_nil(ctx.after_yield)
end
test.this_thread.pause_with_yield = function () test.this_thread.pause_with_yield = function ()
local share = effil.table({stop = false}) local share = effil.table({stop = false})
local spec = effil.thread(function (share) local spec = effil.thread(function (share)
@ -352,12 +376,12 @@ local function call_pause(thr)
return true return true
end end
-- Regress test to check hanging when invoke pause on canceled thread -- Regress test to check hanging when invoke pause on cancelled thread
test.this_thread.pause_on_canceled_thread = function () test.this_thread.pause_on_cancelled_thread = function ()
local worker_thread = effil.thread(worker)({ need_to_stop = false}) local worker_thread = effil.thread(worker)({ need_to_stop = false})
effil.sleep(1, 's') effil.sleep(1, 's')
worker_thread:cancel() worker_thread:cancel()
test.equal(worker_thread:wait(2, "s"), "canceled") test.equal(worker_thread:wait(2, "s"), "cancelled")
test.is_true(effil.thread(call_pause)(worker_thread):get(5, "s")) test.is_true(effil.thread(call_pause)(worker_thread):get(5, "s"))
end end
@ -406,3 +430,108 @@ test.thread.traceback = function()
end end
end -- LUA_VERSION > 51 end -- LUA_VERSION > 51
test.thread.cancel_thread_with_pcall = function()
local steps = effil.table{step1 = false, step2 = false}
local pcall_results = effil.table{}
local thr = effil.thread(
function()
pcall_results.ret, pcall_results.msg = pcall(function()
while true do
effil.yield()
end
end)
steps.step1 = true
effil.yield()
steps.step2 = true -- should never reach
end
)()
test.is_true(thr:cancel())
test.equal(thr:wait(), "cancelled")
test.is_true(steps.step1)
test.is_false(steps.step2)
test.is_false(pcall_results.ret)
test.equal(pcall_results.msg, "Effil: thread is cancelled")
end
test.thread.cancel_thread_with_pcall_not_cancelled = function()
local thr = effil.thread(
function()
pcall(function()
while true do
effil.yield()
end
end)
end
)()
test.is_true(thr:cancel())
test.equal(thr:wait(), "completed")
end
test.thread.cancel_thread_with_pcall_and_another_error = function()
local msg = 'some text'
local thr = effil.thread(
function()
pcall(function()
while true do
effil.yield()
end
end)
error(msg)
end
)()
test.is_true(thr:cancel())
local status, message = thr:wait()
test.equal(status, "failed")
test.is_not_nil(string.find(message, ".+: " .. msg))
end
if not jit then
test.thread.cancel_thread_with_pcall_without_yield = function()
local thr = effil.thread(
function()
while true do
-- pass
end
end
)
thr = thr()
test.is_true(thr:cancel())
test.equal(thr:wait(), "cancelled")
end
end
test.thread.check_effil_pcall_success = function()
local inp1, inp2, inp3 = 1, "str", {}
local res, ret1, ret2, ret3 = effil.pcall(function(...) return ... end, inp1, inp2, inp3)
test.is_true(res)
test.equal(ret1, inp1)
test.equal(ret2, inp2)
test.equal(ret3, inp3)
end
test.thread.check_effil_pcall_fail = function()
local err = "some text"
local res, msg = effil.pcall(function(err) error(err) end, err)
test.is_false(res)
test.is_not_nil(string.find(msg, ".+: " .. err))
end
test.thread.check_effil_pcall_with_cancel_thread = function()
local thr = effil.thread(
function()
effil.pcall(function()
while true do
effil.yield()
end
end)
end
)()
test.is_true(thr:cancel())
test.equal(thr:wait(), "cancelled")
end