khipu/thread.hpp

/* Declarations for the threading interface.

   This file is part of khipu.

   khipu is free software: you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation; either version 3 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 Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <https://www.gnu.org/licenses/>.  */

#ifndef __KP_THREAD__
#define __KP_THREAD__   1

#include "interp.hpp"
#include "initop.hpp"
#include "custom.hpp"
#include "utils/lazy.hpp"
#include "sysdeps/atomic.hpp"

#ifdef KP_PLATFORM_UNIX
  #include <pthread.h>
#endif

KP_DECLS_BEGIN

// Lightweight locks for internal use.

// Initialize lightweight lock LOCKP.
inline void lwlock_init (atomic_t *lockp)
{
  *lockp = 0;
}

// Grab the lightweight lock LOCKP.
extern result<void> lwlock_grab (interpreter *interp, atomic_t *lockp);

// Grab the lightweight lock LOCKP, waiting up to TIMEOUT.
extern result<bool> lwlock_grab (interpreter *interp,
                                 atomic_t *lockp, double timeout);

// Same as above, only this doesn't check for interrupts when sleeping.
extern void lwlock_grab_nointr (interpreter *interp, atomic_t *lockp);

extern bool lwlock_grab_nointr (interpreter *interp,
                                atomic_t *lockp, double timeout);

// Drop the lightweight lock LOCKP.
extern void lwlock_drop (atomic_t *lockp);

// Try to grab the lightweight lock LOCKP.
inline bool lwlock_trygrab (atomic_t *lockp)
{
  return (atomic_cas_bool (lockp, 0, 1));
}

struct lwlock_guard
{
  atomic_t *lockp;

  lwlock_guard () : lockp (nullptr)
    {
    }

  static result<lwlock_guard>
  make (interpreter *interp, atomic_t *lp)
    {
      lwlock_guard ret;
      KP_VTRY (ret.set (interp, lp));
      return (ret);
    }

  result<int> set (interpreter *interp, atomic_t *lp)
    {
      if (this->lockp)
        lwlock_drop (this->lockp);
      if ((this->lockp = lp) != nullptr)
        KP_VTRY (lwlock_grab (interp, this->lockp));
      return (0);
    }

  ~lwlock_guard ()
    {
      if (this->lockp)
        lwlock_drop (this->lockp);
    }
};

#ifdef KP_NO_THREADS

struct lock : public custom<lock>
{
  static const uint32_t recursive_flag = 1u << 16;

  void init (bool)
    {
    }

  result<int> grap (interpreter *);
  result<int> grab (interpreter *, object, bool);
  int trygrab (interpreter *);
  int drop (interpreter *);
  void force_release (interpreter *);
};

#else

struct raw_condvar
{
  atomic_t *lockp;
  atomic_t seq;

  void init ()
    {
      this->lockp = nullptr;
      this->seq = 0;
    }

  result<int> wait (interpreter *interp, atomic_t *lk,
                    double *timeout = nullptr);
  int wait_nointr (interpreter *interp, atomic_t *lk,
                   double *timeout = nullptr);
  void wake (bool wake_all);
};

struct lock : public custom<lock>
{
  static const uint32_t recursive_flag = 1 << 16;

  atomic_t word;
  lazy<valref> ref;
  object owner;
  uint32_t lock_cnt;

  void do_visit (interpreter *interp, visitor& vs)
    {
      if (this->owner != UNBOUND)
        vs (interp, this->owner);
    }

  void init (bool recursive_p);
  result<int> grab (interpreter *interp);
  result<int> grab (interpreter *interp, object timeout, bool absolute);
  int trygrab (interpreter *interp);
  int drop (interpreter *interp);
  void force_release (interpreter *interp);
};

struct condvar : public custom<condvar>
{
  raw_condvar base;

  result<int> wait (interpreter *interp, object lk);
  result<int> wait (interpreter *interp, object lk, object timeout, bool abs);

  void wake (interpreter *, bool wake_all)
    {
      this->base.wake (wake_all);
    }
};

#ifdef KP_PLATFORM_UNIX
  typedef pthread_t system_thread_t;
#else
  typedef void* system_thread_t;
#endif

struct sleepq;

struct thread : public varobj
{
  static const int code = typecode::THREAD;

  atomic_t join_ev;
  system_thread_t handle;
  object retval;
  atomic_t ilock;
  interpreter *interp;
  dlist thr_link;
#if !defined (KP_PLATFORM_LINUX)
  sleepq *sleep_q;
#endif
};

struct sync_event
{
  atomic_t lock;
  raw_condvar cv;
  uintptr_t waiters_sig;
  uintptr_t n_waiters;
  uintptr_t limit;

  void init ()
    {
      this->lock = 0;
      this->cv.init ();
      this->reset ();
    }

  void reset ()
    {
      this->waiters_sig = this->n_waiters = this->limit = 0;
    }

  void wait (interpreter *interp);
  void wake_all (interpreter *interp);
};

#endif

struct lock_guard
{
  interpreter *interp;
  lock *lockp;

  lock_guard () : interp (nullptr), lockp (nullptr)
    {
    }

  lock_guard (interpreter *ip, lock *lp) : interp (ip), lockp (lp)
    {
    }

  static result<lock_guard>
  make (interpreter *interp, lock *lp = nullptr)
    {
      lock_guard ret (interp, lp);
      if (lp)
        KP_VTRY (lp->grab (interp));

      return (ret);
    }

  result<int> set (lock *lp)
    {
      if (this->lockp)
        this->lockp->drop (this->interp);
      if ((this->lockp = lp) != nullptr)
        KP_VTRY (this->lockp->grab (this->interp));

      return (0);
    }

  ~lock_guard ()
    {
      if (this->lockp)
        this->lockp->drop (this->interp);
    }
};

inline lock* as_lock (object obj)
{
  return ((lock *)as_custom (obj));
}

inline bool lock_p (object obj)
{
  return (lock::type_p (obj));
}

KP_EXPORT result<object> alloc_lock (interpreter *interp, bool recursive_p);

inline condvar* as_condvar (object obj)
{
  return ((condvar *)as_custom (obj));
}

inline bool condvar_p (object obj)
{
  return (condvar::type_p (obj));
}

KP_EXPORT result<object> alloc_condvar (interpreter *interp);

inline thread* as_thread (object obj)
{
  return ((thread *)unmask (obj));
}

inline bool thread_p (object obj)
{
  return (itype (obj) == typecode::THREAD);
}

KP_EXPORT dlist all_threads;
KP_EXPORT atomic_t all_threads_lock;
KP_EXPORT uint32_t num_threads;

// Return values for some thread calls.
enum
{
  THR_ERROR = -1,
  THR_TIMEOUT = -2,
  THR_DEADLK = -3,
  THR_PERM = -4,
  THR_INTR = -5
};

/* Test if only one thread is running. This test should be as cheap as
 * possible, in order to avoid locking when a thread knows it's alone. */
#ifdef KP_NO_THREADS

inline constexpr bool singlethr_p ()
{
  return (true);
}

#  define KP_MT_BEGIN(expr)   (void)0
#  define KP_MT_END(expr)     (void)0

#else

inline bool singlethr_p ()
{
  return (num_threads <= 1);
}

#  define KP_MT_BEGIN(expr)   \
     bool __tst = singlethr_p ();   \
     if (!__tst) expr

#  define KP_MT_END(expr)   \
     if (!__tst) expr

#endif

// Init OP for the threading subsystem.
KP_EXPORT init_op init_threads;

KP_DECLS_END

#endif