Extension_ThreadCondition.cpp

/* -----------------------------------------------------------------------------
Copyright 2021 Kevin P. Barry

Licensed under the Apache License, Version 2.0 (the "License");
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.
----------------------------------------------------------------------------- */

// Author: Kevin P. Barry [ta0kira@gmail.com]

#include <math.h>
#include <pthread.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>

#include "category-source.hpp"
#include "Streamlined_ThreadCondition.hpp"
#include "Category_ThreadCondition.hpp"
#include "Category_ConditionResume.hpp"
#include "Category_ConditionWait.hpp"

#ifdef ZEOLITE_PUBLIC_NAMESPACE
namespace ZEOLITE_PUBLIC_NAMESPACE {
#endif  // ZEOLITE_PUBLIC_NAMESPACE

BoxedValue CreateValue_ThreadCondition(S<const Type_ThreadCondition> parent);

struct ExtCategory_ThreadCondition : public Category_ThreadCondition {
};

struct ExtType_ThreadCondition : public Type_ThreadCondition {
  inline ExtType_ThreadCondition(Category_ThreadCondition& p, Params<0>::Type params) : Type_ThreadCondition(p, params) {}

  ReturnTuple Call_new(const ParamTuple& params, const ValueTuple& args) const final {
    TRACE_FUNCTION("ThreadCondition.new")
    return ReturnTuple(CreateValue_ThreadCondition(PARAM_SELF));
  }
};

struct ExtValue_ThreadCondition : public Value_ThreadCondition {
  inline ExtValue_ThreadCondition(S<const Type_ThreadCondition> p)
    : Value_ThreadCondition(std::move(p)) {
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    // NOTE: Error checking is required to catch attempted waits without first
    // locking the mutex.
    pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_ERRORCHECK);
    pthread_mutex_init(&mutex, &mutex_attr);
  }

  ReturnTuple Call_lock(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.lock")
    int error = 0;
    if ((error = pthread_mutex_lock(&mutex)) != 0) {
      FailError("Error locking mutex", error);
    }
    return ReturnTuple(VAR_SELF);
  }

  ReturnTuple Call_resumeAll(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.resumeAll")
    int error = 0;
    if ((error = pthread_cond_broadcast(&cond)) != 0) {
      FailError("Error resuming threads", error);
    }
    return ReturnTuple(VAR_SELF);
  }

  ReturnTuple Call_resumeOne(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.resumeOne")
    int error = 0;
    if ((error = pthread_cond_signal(&cond)) != 0) {
      FailError("Error resuming thread", error);
    }
    return ReturnTuple(VAR_SELF);
  }

  ReturnTuple Call_timedWait(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.timedWait")
    int error = 0;
    const PrimFloat Var_arg1 = (args.At(0)).AsFloat();
    if (Var_arg1 < 0) {
      FAIL() << "Bad wait time " << Var_arg1;
    }
    struct timeval now;
    if (gettimeofday(&now, NULL) != 0) {
      FailError("Error getting current time", errno);
    }
    const PrimFloat abs_time = Var_arg1 + (PrimFloat) now.tv_sec + ((PrimFloat) now.tv_usec / 1000000.0);
    struct timespec timeout{ (int) trunc(abs_time), (int) (1000000000.0 * (abs_time-trunc(abs_time))) };
    error = pthread_cond_timedwait(&cond, &mutex, &timeout);
    if (error == ETIMEDOUT) {
      return ReturnTuple(Box_Bool(false));
    }
    if (error != 0) {
      FailError("Error waiting for condition", error);
    }
    return ReturnTuple(Box_Bool(true));
  }

  ReturnTuple Call_unlock(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.unlock")
    int error = 0;
    if ((error = pthread_mutex_unlock(&mutex)) != 0) {
      FailError("Error unlocking mutex", error);
    }
    return ReturnTuple(VAR_SELF);
  }

  ReturnTuple Call_wait(const ParamTuple& params, const ValueTuple& args) final {
    TRACE_FUNCTION("ThreadCondition.wait")
    int error = 0;
    if ((error = pthread_cond_wait(&cond, &mutex)) != 0) {
      FailError("Error waiting for condition", error);
    }
    return ReturnTuple(VAR_SELF);
  }

  void FailError(const std::string& context, int error) const {
    TRACE_CREATION
    FAIL() << context << ": " << strerror(error) << " (error " << error << ")";
  }

  ~ExtValue_ThreadCondition() {
    int error = 0;
    // Nothing should be waiting on the condition, because then there would
    // still be a reference to this object.
    if ((error = pthread_cond_destroy(&cond)) != 0) {
      FailError("Error cleaning up condition", error);
    }
    if ((error = pthread_mutex_destroy(&mutex)) != 0) {
      FailError("Error cleaning up mutex", error);
    }
  }

  pthread_cond_t  cond  = PTHREAD_COND_INITIALIZER;
  pthread_mutex_t mutex;
  CAPTURE_CREATION("ThreadCondition")
};

Category_ThreadCondition& CreateCategory_ThreadCondition() {
  static auto& category = *new ExtCategory_ThreadCondition();
  return category;
}

S<const Type_ThreadCondition> CreateType_ThreadCondition(const Params<0>::Type& params) {
  static const auto cached = S_get(new ExtType_ThreadCondition(CreateCategory_ThreadCondition(), Params<0>::Type()));
  return cached;
}

void RemoveType_ThreadCondition(const Params<0>::Type& params) {}

BoxedValue CreateValue_ThreadCondition(S<const Type_ThreadCondition> parent) {
  return BoxedValue::New<ExtValue_ThreadCondition>(std::move(parent));
}

#ifdef ZEOLITE_PUBLIC_NAMESPACE
}  // namespace ZEOLITE_PUBLIC_NAMESPACE
using namespace ZEOLITE_PUBLIC_NAMESPACE;
#endif  // ZEOLITE_PUBLIC_NAMESPACE