thread.hh

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  You may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
 
/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */
 
#pragma once
 
#ifndef SEASTAR_MODULE
#include <seastar/core/thread_impl.hh>
#include <seastar/core/future.hh>
#include <seastar/core/do_with.hh>
#include <seastar/core/timer.hh>
#include <seastar/core/scheduling.hh>
#include <memory>
#include <type_traits>
#include <seastar/util/assert.hh>
#include <seastar/util/std-compat.hh>
#include <seastar/util/modules.hh>
#include <ucontext.h>
#include <boost/intrusive/list.hpp>
#endif
 
 
namespace seastar {
 
SEASTAR_MODULE_EXPORT_BEGIN
class thread;
class thread_attributes;
 
class thread_attributes {
public:
    std::optional<seastar::scheduling_group> sched_group;
    // For stack_size 0, a default value will be used (128KiB when writing this comment)
    size_t stack_size = 0;
};
SEASTAR_MODULE_EXPORT_END
 
extern thread_local jmp_buf_link g_unthreaded_context;
 
// Internal class holding thread state.  We can't hold this in
// \c thread itself because \c thread is movable, and we want pointers
// to this state to be captured.
class thread_context final : private task {
    struct stack_deleter {
        void operator()(char *ptr) const noexcept;
        int valgrind_id;
        stack_deleter(int valgrind_id);
    };
    using stack_holder = std::unique_ptr<char[], stack_deleter>;
 
    stack_holder _stack;
    noncopyable_function<void ()> _func;
    jmp_buf_link _context;
    promise<> _done;
    bool _joined = false;
 
    boost::intrusive::list_member_hook<> _all_link;
    using all_thread_list = boost::intrusive::list<thread_context,
        boost::intrusive::member_hook<thread_context, boost::intrusive::list_member_hook<>,
        &thread_context::_all_link>,
        boost::intrusive::constant_time_size<false>>;
 
    static thread_local all_thread_list _all_threads;
private:
    static void s_main(int lo, int hi); // all parameters MUST be 'int' for makecontext
    void setup(size_t stack_size);
    void main();
    stack_holder make_stack(size_t stack_size);
    virtual void run_and_dispose() noexcept override; // from task class
public:
    thread_context(thread_attributes attr, noncopyable_function<void ()> func);
    ~thread_context();
    void switch_in();
    void switch_out();
    bool should_yield() const;
    void reschedule();
    void yield();
    task* waiting_task() noexcept override { return _done.waiting_task(); }
    friend class thread;
    friend void thread_impl::switch_in(thread_context*);
    friend void thread_impl::switch_out(thread_context*);
    friend scheduling_group thread_impl::sched_group(const thread_context*);
};
 
 
SEASTAR_MODULE_EXPORT
class thread {
    std::unique_ptr<thread_context> _context;
    static thread_local thread* _current;
public:
    thread() = default;
    template <typename Func>
    thread(Func func);
    template <typename Func>
    thread(thread_attributes attr, Func func);
    thread(thread&& x) noexcept = default;
    thread& operator=(thread&& x) noexcept = default;
    ~thread() { SEASTAR_ASSERT(!_context || _context->_joined); }
    future<> join();
    static void yield();
    static bool should_yield() {
        return need_preempt();
    }
 
    static void maybe_yield() {
        if (should_yield()) [[unlikely]] {
            yield();
        }
    }
 
    static bool running_in_thread() {
        return thread_impl::get() != nullptr;
    }
};
 
template <typename Func>
inline
thread::thread(thread_attributes attr, Func func)
        : _context(std::make_unique<thread_context>(std::move(attr), std::move(func))) {
}
 
template <typename Func>
inline
thread::thread(Func func)
        : thread(thread_attributes(), std::move(func)) {
}
 
inline
future<>
thread::join() {
    _context->_joined = true;
    return _context->_done.get_future();
}
 
SEASTAR_MODULE_EXPORT_BEGIN
template <typename Func, typename... Args>
inline
futurize_t<std::invoke_result_t<Func, Args...>>
async(thread_attributes attr, Func&& func, Args&&... args) noexcept {
    using return_type = std::invoke_result_t<Func, Args...>;
    struct work {
        thread_attributes attr;
        Func func;
        std::tuple<Args...> args;
        promise<return_type> pr{};
        thread th{};
    };
 
    try {
        auto wp = std::make_unique<work>(work{std::move(attr), std::forward<Func>(func), std::forward_as_tuple(std::forward<Args>(args)...)});
        auto& w = *wp;
        auto ret = w.pr.get_future();
        w.th = thread(std::move(w.attr), [&w] {
            futurize<return_type>::apply(std::move(w.func), std::move(w.args)).forward_to(std::move(w.pr));
        });
        return w.th.join().then([ret = std::move(ret)] () mutable {
            return std::move(ret);
        }).finally([wp = std::move(wp)] {});
    } catch (...) {
        return futurize<return_type>::make_exception_future(std::current_exception());
    }
}
 
template <typename Func, typename... Args>
inline
futurize_t<std::invoke_result_t<Func, Args...>>
async(Func&& func, Args&&... args) noexcept {
    return async(thread_attributes{}, std::forward<Func>(func), std::forward<Args>(args)...);
}
 
SEASTAR_MODULE_EXPORT_END
}