thread.cpp

Go to the documentation of this file.

// Copyright (c) 2013-2025  Made to Order Software Corp.  All Rights Reserved
//
// https://snapwebsites.org/project/cppthread
// contact@m2osw.com
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; if not, write to the Free Software Foundation, Inc.,
// 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
// self
//
#include    "cppthread/thread.h"
 
#include    "cppthread/exception.h"
#include    "cppthread/guard.h"
#include    "cppthread/log.h"
#include    "cppthread/runner.h"
 
 
// snapdev
//
#include    <snapdev/glob_to_list.h>
 
 
// C++
//
#include    <fstream>
 
 
// C
//
#include    <signal.h>
#include    <sys/auxv.h>
#include    <sys/stat.h>
#include    <sys/syscall.h>
#include    <sys/sysinfo.h>
#include    <unistd.h>
 
 
// last include
//
#include    <snapdev/poison.h>
 
 
 
 
namespace cppthread
{
 
 
 
 
 
thread::thread(std::string const & name, runner * runner)
    : f_name(name)
    , f_runner(runner)
{
    init();
}
 
 
thread::thread(std::string const & name, std::shared_ptr<runner> runner)
    : f_name(name)
    , f_runner(runner.get())
{
    init();
}
 
 
void thread::init()
{
    if(f_runner == nullptr)
    {
        throw invalid_error("runner missing in thread() constructor.");
    }
    if(f_runner->f_thread != nullptr)
    {
        throw in_use_error(
                      "this runner ("
                    + f_runner->get_name()
                    + ") is already in use.");
    }
 
    int err(pthread_attr_init(&f_thread_attr));
    if(err != 0)
    {
        // LCOV_EXCL_START
        log << log_level_t::fatal
            << "the thread attributes could not be initialized, error #"
            << err
            << '.'
            << end;
        throw invalid_error("pthread_attr_init() failed.");
        // LCOV_EXCL_STOP
    }
    err = pthread_attr_setdetachstate(&f_thread_attr, PTHREAD_CREATE_JOINABLE);
    if(err != 0)
    {
        // LCOV_EXCL_START
        log << log_level_t::fatal
            << "the thread detach state could not be initialized to joinable, error #"
            << err
            << '.'
            << end;
        pthread_attr_destroy(&f_thread_attr);
        throw invalid_error("pthread_attr_setdetachstate() failed");
        // LCOV_EXCL_STOP
    }
 
    f_runner->f_thread = this;
}
 
 
thread::~thread()
{
    try
    {
        stop();
    }
    catch(...)
    {
        // stop() may rethrow any user exception which we have to ignore in
        // a destructor...
    }
    f_runner->f_thread = nullptr;
 
    int const err(pthread_attr_destroy(&f_thread_attr));
    if(err != 0)
    {
        // LCOV_EXCL_START
        log << log_level_t::error
            << "the thread attributes could not be destroyed, error #"
            << err
            << '.'
            << end;
        // LCOV_EXCL_STOP
    }
}
 
 
std::string const & thread::get_name() const
{
    return f_name;
}
 
 
runner * thread::get_runner() const
{
    return f_runner;
}
 
 
bool thread::is_running() const
{
    guard lock(f_mutex);
    return f_running;
}
 
 
bool thread::is_stopping() const
{
    guard lock(f_mutex);
    return f_stopping;
}
 
 
void * func_internal_start(void * system_thread)
{
    thread * t(reinterpret_cast<thread *>(system_thread));
    t->internal_thread();
    return nullptr; // == pthread_exit(nullptr);
}
 
 
void thread::internal_thread()
{
    try
    {
        {
            guard lock(f_mutex);
            f_tid = gettid();
            f_started = true;
            f_mutex.signal();
        }
 
        std::string name(f_runner->get_name());
        if(!name.empty())
        {
            // make sure to limit the name to 15 characters
            //
            if(name.length() > 15)
            {
                name.resize(15);
            }
 
            // the pthread_setname_np() allows for the name to be retrieved
            // with its counter part:
            //
            //   pthread_getname_np()
            //
            pthread_setname_np(pthread_self(), name.c_str());
 
            // but to really change the name in the comm file (and therefore
            // htop, ps, etc.) we further call the set_current_thread_name()
            // function which directly writes to that file
            //
            set_current_thread_name(name);
        }
 
        // if the enter failed, do not continue
        //
        if(internal_enter())
        {
            if(internal_run())
            {
                internal_leave(leave_status_t::LEAVE_STATUS_NORMAL);
            }
            else
            {
                internal_leave(leave_status_t::LEAVE_STATUS_THREAD_FAILED);
            }
        }
        else
        {
            internal_leave(leave_status_t::LEAVE_STATUS_INITIALIZATION_FAILED);
        }
 
        // if useful (necessary) it would probably be better to call this
        // function from here; see function and read the "note" section
        // for additional info
        //
        //tcp_client_server::cleanup_on_thread_exit();
    }
    catch(std::exception const & e)
    {
        // keep a copy of the exception
        //
        f_exception = std::current_exception();
 
        if(f_log_all_exceptions)
        {
            log << log_level_t::fatal
                << "thread internal_thread() got exception: \""
                << e.what()
                << "\", exiting thread now."
                << end;
        }
 
        internal_leave(leave_status_t::LEAVE_STATUS_INSTRUMENTATION);
    }
    catch(...)
    {
        // ... any other exception terminates the whole process ...
        //
        log << log_level_t::fatal
            << "thread internal_thread() got an unknown exception (a.k.a. non-std::exception), exiting process."
            << end;
 
        // rethrow, our goal is not to ignore the exception, only to
        // have a log about it
        //
        throw;
    }
 
    // marked we are done (outside of the try/catch because if this one
    // fails, we have a big problem... (i.e. invalid mutex or more unlocks
    // than locks)
    {
        guard lock(f_mutex);
        f_running = false;
        f_tid = PID_UNDEFINED;
        f_mutex.signal();
    }
}
 
 
bool thread::internal_enter()
{
    try
    {
        f_runner->enter();
        return true;
    }
    catch(std::exception const & e)
    {
        // keep a copy of the exception
        //
        f_exception = std::current_exception();
 
        if(f_log_all_exceptions)
        {
            log << log_level_t::fatal
                << "thread internal_enter() got exception: \""
                << e.what()
                << "\", exiting thread now."
                << end;
        }
    }
 
    return false;
}
 
 
bool thread::internal_run()
{
    try
    {
        f_runner->run();
        return true;
    }
    catch(std::exception const & e)
    {
        // keep a copy of the exception
        //
        f_exception = std::current_exception();
 
        if(f_log_all_exceptions)
        {
            log << log_level_t::fatal
                << "thread internal_run() got exception: \""
                << e.what()
                << "\", exiting thread now."
                << end;
        }
    }
 
    return false;
}
 
 
void thread::internal_leave(leave_status_t status)
{
    try
    {
        f_runner->leave(status);
    }
    catch(std::exception const & e)
    {
        // keep the first exception (i.e. internal_enter() and internal_run()
        // have priority on this one)
        //
        bool force_log(true);
        if(f_exception == std::exception_ptr())
        {
            f_exception = std::current_exception();
            force_log = false;
        }
 
        if(f_log_all_exceptions || force_log)
        {
            log << log_level_t::fatal
                << "thread internal_leave() got exception: \""
                << e.what()
                << "\", exiting thread now."
                << end;
        }
    }
}
 
 
bool thread::start()
{
    guard lock(f_mutex);
 
    if(f_running || f_started)
    {
        log << log_level_t::warning
            << "the thread is already running."
            << end;
        return false;
    }
 
    if(!f_runner->is_ready())
    {
        log << log_level_t::warning
            << "the thread runner is not ready."
            << end;
        return false;
    }
 
    f_running = true;
    f_started = false;
    f_stopping = false; // make sure it is reset
    f_exception = std::exception_ptr();
 
    // by default, block all signals in threads
    //
    // Note: we check whether the gettid() is equal to the getpid() which
    //       means only the main thread tweaks the signals; this is
    //       important otherwise we could mess up the mask
    //
    libexcept::scoped_signal_mask::pointer_t block_all_signals;
    if(gettid() == getpid())
    {
        block_all_signals = std::make_shared<libexcept::scoped_signal_mask>();
    }
 
    int const err(pthread_create(&f_thread_id, &f_thread_attr, &func_internal_start, this));
    if(err != 0)
    {
        // LCOV_EXCL_START
        f_running = false;
 
        log << log_level_t::error
            << "the thread could not be created, error #"
            << err
            << '.'
            << end;
        return false;
        // LCOV_EXCL_STOP
    }
 
    while(!f_started)
    {
        f_mutex.wait();
    }
 
    return true;
}
 
 
void thread::stop(std::function<void(thread *)> callback)
{
    {
        guard lock(f_mutex);
 
        if(!f_running && !f_started)
        {
            // we return immediately in this case because
            // there is nothing to join when the thread never
            // started...
            //
            return;
        }
 
        // request the child to stop
        //
        f_stopping = true;
    }
 
    if(callback != nullptr)
    {
        callback(this);
    }
 
    // wait for the child to be stopped
    //
    // we cannot pass any results through the pthread interface so we
    // pass a nullptr for the result; instead, the user is expected to
    // add fields to his class and fill in whatever results he wants
    // there; it is going to work much better that way
    //
    pthread_join(f_thread_id, nullptr);
 
    // at this point the thread has fully exited
 
    // we are done now
    //
    // these flags are likely already the correct value except for
    // f_stopping which the stop() function manages here
    //
    f_running = false;
    f_started = false;
    f_stopping = false;
 
    // if the child died because of a standard exception, rethrow it now
    // and "lose it" at the same time
    //
    if(f_exception != std::exception_ptr())
    {
        std::exception_ptr e;
        e.swap(f_exception);
        std::rethrow_exception(e);
    }
}
 
 
pid_t thread::get_thread_tid() const
{
    guard lock(f_mutex);
    return f_tid;
}
 
 
mutex & thread::get_thread_mutex() const
{
    return f_mutex;
}
 
 
void thread::set_log_all_exceptions(bool log_all_exceptions)
{
    f_log_all_exceptions = log_all_exceptions;
}
 
 
bool thread::get_log_all_exceptions() const
{
    return f_log_all_exceptions;
}
 
 
std::exception_ptr thread::get_exception() const
{
    guard lock(f_mutex);
    return f_exception;
}
 
 
bool thread::kill(int sig)
{
    guard lock(f_mutex);
    if(f_running)
    {
        // pthread_kill() returns zero on success, otherwise it returns
        // an error code which at this point we lose
        //
        return pthread_kill(f_thread_id, sig) == 0;
    }
 
    return false;
}
 
 
void thread::mask_signals(libexcept::sig_list_t list)
{
    sigset_t set;
    sigemptyset(&set);
    for(auto const & s : list)
    {
        sigaddset(&set, s);
    }
    pthread_sigmask(SIG_BLOCK, &set, nullptr);
}
 
 
void thread::mask_all_signals()
{
    sigset_t set;
    sigfillset(&set);
    pthread_sigmask(SIG_BLOCK, &set, nullptr);
}
 
 
void thread::unmask_signals(libexcept::sig_list_t list)
{
    sigset_t set;
    sigemptyset(&set);
    for(auto const & s : list)
    {
        sigaddset(&set, s);
    }
    pthread_sigmask(SIG_UNBLOCK, &set, nullptr);
}
 
 
 
 
 
 
 
 
int get_total_number_of_processors()
{
    return get_nprocs_conf();
}
 
 
int get_number_of_available_processors()
{
    return get_nprocs();
}
 
 
pid_t gettid()
{
    return static_cast<pid_t>(syscall(SYS_gettid));
}
 
 
pid_t get_pid_max()
{
    // here we cache the result (because this value is so unlikely to
    // ever change--but really we should watch for changes?)
    //
    static pid_t pid_max = 0;
 
    if(pid_max == 0)
    {
        std::ifstream in;
        in.open("/proc/sys/kernel/pid_max", std::ios::in | std::ios::binary);
        if(in.is_open())
        {
            char buf[32];
            in.getline(buf, sizeof(buf) - 1);
            buf[sizeof(buf) - 1] = '\0';
            pid_max = std::stol(buf);
        }
    }
 
    return pid_max - 1;
}
 
 
int set_current_thread_name(std::string const & name)
{
    return set_thread_name(gettid(), name);
}
 
 
int set_thread_name(pid_t tid, std::string const & name)
{
    if(name.empty())
    {
        throw invalid_error("thread name cannot be empty.");
    }
    if(name.length() > 15)
    {
        throw out_of_range(
              "thread name is limited to 15 characters, \""
            + name
            + "\" is too long.");
    }
 
    // this is what I had in the process implementation, but it only
    // can set the current "process" (thread really) and it doesn't
    // actually work (that is, the PR_GET_NAME works, but in ps -ef
    // or htop, we do not see the change)
    //
    //if(name != nullptr
    //&& *name != '\0')
    //{
    //    prctl(PR_SET_NAME, name);
    //}
 
    std::ofstream comm("/proc/" + std::to_string(tid) + "/comm");
    comm << name;
 
    return comm ? 0 : -1;
}
 
 
std::string get_current_thread_name()
{
    return get_thread_name(gettid());
}
 
 
std::string get_thread_name(pid_t tid)
{
    std::ifstream comm("/proc/" + std::to_string(tid) + "/comm");
    std::string name;
    comm >> name;
    return name;
}
 
 
process_ids_t get_thread_ids(pid_t pid)
{
    process_ids_t results;
 
    if(pid == -1)
    {
        pid = getpid();
    }
 
    std::string pattern("/proc/");
    pattern += std::to_string(pid);
    pattern += "/task/*";
 
    snapdev::glob_to_list<std::vector<std::string>> glob;
    if(glob.read_path<
              snapdev::glob_to_list_flag_t::GLOB_FLAG_IGNORE_ERRORS
            , snapdev::glob_to_list_flag_t::GLOB_FLAG_ONLY_DIRECTORIES>(pattern))
    {
        for(auto s : glob)
        {
            pid_t id(0);
            std::string::size_type pos(s.rfind('/') + 1);
 
            bool valid(pos < s.length());
            for(; pos < s.length(); ++pos)
            {
                char const c(s[pos]);
                if(c >= '0' && c <= '9')
                {
                    id = id * 10 + c - '0';
                }
                else
                {
                    valid = false;
                    break;
                }
            }
            if(valid)
            {
                results.push_back(id);
            }
        }
    }
 
    return results;
}
 
 
bool is_process_running(pid_t pid)
{
    if(pid == getpid())
    {
        // funny guy testing whether he himself is running!?
        //
        return true;
    }
 
    std::string proc_path("/proc/");
    proc_path += std::to_string(pid);
 
    struct stat st;
    return stat(proc_path.c_str(), &st) == 0;
}
 
 
std::string get_boot_id()
{
    std::ifstream in("/proc/sys/kernel/random/boot_id");
    std::string uuid;
    if(in)
    {
        std::getline(in, uuid);
    }
    return uuid;
}
 
 
std::size_t get_thread_count()
{
    struct stat task;
    if(stat("/proc/self/task", &task) != 0)
    {
        return -1;
    }
 
    return task.st_nlink - 2;
}
 
 
bool is_using_vdso()
{
    // the call to getauxval() is only required once; the result can't change
    // within one process run (however, it has to be different on each run
    // to make it secure)
    //
    static auto g_vdso(getauxval(AT_SYSINFO_EHDR));
    return g_vdso != 0;
}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
} // namespace cppthread
// vim: ts=4 sw=4 et