Add priority inheritance signaling

jclklib/common/Makefile

@@ -9,7 +9,8 @@ JCLKLIB_COMMON_DIR := $(shell pwd)
 JCLKLIB_TOPLEVEL_DIR := $(JCLKLIB_COMMON_DIR)/..
 LIBS = pthread rt
 LIBS_FLAGS = $(foreach lib,$(LIBS),-l$(lib))
-OBJ = jclklib_import message msgq_tport notification_msg null_msg connect_msg print sighandler subscribe_msg transport util
+OBJ = jclklib_import message msgq_tport notification_msg null_msg connect_msg print sighandler subscribe_msg transport\
+	mutex_signal
 OBJ_FILES = $(foreach f,$(OBJ),$(f).o)
 DEPENDS = $(foreach f,$(OBJ),$(f).d)
 OBJ_DIR = $(JCLKLIB_COMMON_DIR)/obj
@@ -57,5 +58,5 @@ common-obj: $(OBJ_FILES)
 
 %.o : %.cpp
 	echo "[COMPILE]" $<
-	g++ -c $< -g -I $(JCLKLIB_TOPLEVEL_DIR) -fPIC -Werror -Wall -fdiagnostics-color=always
+	g++ -c $< -g -I $(JCLKLIB_TOPLEVEL_DIR) -fPIC -Werror -Wall -Wextra -Wno-unused-parameter -fdiagnostics-color=always
 

jclklib/common/mutex_signal.cpp

@@ -0,0 +1,201 @@
+#include <common/mutex_signal.hpp>
+
+#include <cstring>
+#include <iostream>
+
+using namespace std;
+using namespace mutex_signal_test;
+
+#define NSEC_PER_SEC (1000000000)
+
+#define PTHREAD_CALL(func,err_msg,retval,...)				\
+	({								\
+		int err;						\
+		if ((err = (func)) != 0 __VA_OPT__(&& err !=) __VA_ARGS__) { \
+			cerr << string(err_msg) + " " + strerror(err) + "\n"; \
+			return retval;					\
+		}							\
+		(err == 0);						\
+        })
+
+mutex_signal::mutex_signal()
+{
+	update_available = false;
+	consumer_locked = false;
+	producer_locked = false;
+}
+
+bool mutex_signal::init()
+{
+        pthread_mutexattr_t attr;
+
+        PTHREAD_CALL(pthread_mutexattr_init(&attr), "Failed to initialize mutex attribute", false);
+        PTHREAD_CALL(pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT), "Failed to set priority inheritance", false);
+	PTHREAD_CALL(pthread_mutex_init(&p_lock, &attr), "Failed to initialize producer mutex", false);
+        PTHREAD_CALL(pthread_mutex_init(&c_lock, &attr), "Failed to initialize consumer mutex", false);
+        PTHREAD_CALL(pthread_mutex_init(&d_lock, NULL), "Failed to initialize data mutex", false);
+        PTHREAD_CALL(pthread_mutexattr_destroy(&attr), "Failed to cleanup mutex attribute", false);
+
+        return true;
+}
+
+mutex_signal::~mutex_signal()
+{
+	PTHREAD_CALL(pthread_mutex_destroy(&d_lock), "Failed to cleanup data mutex",);
+}
+
+bool mutex_signal::consumerTryLock(bool &success)
+{
+	success = PTHREAD_CALL(pthread_mutex_trylock(&c_lock), "Failed to lock consumer lock", false, EBUSY);
+	return true;
+}
+bool mutex_signal::consumerLock()
+{
+	PTHREAD_CALL(pthread_mutex_lock(&c_lock), "Failed to lock consumer lock", false);
+	return true;
+}
+bool mutex_signal::consumerUnlock()
+{
+	PTHREAD_CALL(pthread_mutex_unlock(&c_lock), "Failed to unlock consumer lock", false);
+	return true;
+}
+
+bool mutex_signal::producerTryLock(bool &success)
+{
+	success = PTHREAD_CALL(pthread_mutex_trylock(&p_lock), "Failed to lock producer lock", false, EBUSY);
+	return true;
+}
+bool mutex_signal::producerTimedLock(bool &success, chrono::time_point<chrono::system_clock> wait_until)
+{
+	struct timespec wait_until_ts;
+	wait_until_ts.tv_sec = chrono::duration_cast<chrono::nanoseconds>(wait_until.time_since_epoch()).count() / NSEC_PER_SEC;
+	wait_until_ts.tv_nsec = chrono::duration_cast<chrono::nanoseconds>(wait_until.time_since_epoch()).count() % NSEC_PER_SEC;
+
+	success = PTHREAD_CALL(pthread_mutex_timedlock(&p_lock, &wait_until_ts),"Failed to lock producer lock", false, ETIMEDOUT);
+
+	return true;
+}
+bool mutex_signal::producerLock()
+{
+	PTHREAD_CALL(pthread_mutex_lock(&p_lock), "Failed to lock producer lock", false);
+	return true;
+}
+bool mutex_signal::producerUnlock()
+{
+	PTHREAD_CALL(pthread_mutex_unlock(&p_lock), "Failed to unlock producer lock", false);
+	return true;
+}
+
+bool mutex_signal::dataTryLock(bool &success)
+{
+	success = PTHREAD_CALL(pthread_mutex_trylock(&d_lock), "Failed to lock data lock", false, EBUSY);
+	return true;
+}
+bool mutex_signal::dataTimedLock(bool &success, chrono::time_point<chrono::system_clock> wait_until)
+{
+	struct timespec wait_until_ts;
+	wait_until_ts.tv_sec = chrono::duration_cast<chrono::nanoseconds>(wait_until.time_since_epoch()).count() / NSEC_PER_SEC;
+	wait_until_ts.tv_nsec = chrono::duration_cast<chrono::nanoseconds>(wait_until.time_since_epoch()).count() % NSEC_PER_SEC;
+	success = PTHREAD_CALL(pthread_mutex_timedlock(&d_lock, &wait_until_ts),"Failed to lock data lock", false, ETIMEDOUT);
+
+	return true;
+}
+bool mutex_signal::dataLock()
+{
+	PTHREAD_CALL(pthread_mutex_lock(&d_lock), "Failed to lock data lock", false);
+	return true;
+}
+bool mutex_signal::dataUnlock()
+{
+	PTHREAD_CALL(pthread_mutex_unlock(&d_lock), "Failed to unlock data lock", false);
+	return true;
+}
+
+bool mutex_signal::preProduce()
+{
+	if (!dataLock())
+		return false;
+
+	producer_locked = true;
+	return true;
+}
+bool mutex_signal::postProduce()
+{
+	bool producer_c_lock;
+	if (!producer_locked)
+		return false;
+	update_available = true;
+	if (!dataUnlock() || !consumerTryLock(producer_c_lock))
+		return false;
+	if (producer_c_lock && (!producerUnlock() || !consumerUnlock() || !producerLock()))
+		return false;
+
+	producer_locked = false;
+	return true;
+}
+
+bool mutex_signal::preConsume(bool &success, chrono::duration<unsigned, nano> wait_for)
+{
+	auto wait_until = chrono::system_clock::now() + wait_for;
+	bool consumer_p_lock;
+
+	if (!dataTimedLock(success, wait_until))
+		return false;
+	consumer_locked = success && update_available;
+	if (!success || consumer_locked)
+		return true;
+
+	success = false;
+	if (!consumerUnlock() || !dataUnlock() || !producerTimedLock(consumer_p_lock, wait_until) || !consumerLock())
+		return false;
+	if (!consumer_p_lock)
+		return true;
+	if (!dataTimedLock(success, wait_until))
+		return false;
+	if (consumer_p_lock && !producerUnlock())
+		return false;
+
+	if (success && !update_available && !dataUnlock())
+		return false;
+	success = consumer_locked = success && update_available;
+
+	return true;
+}
+
+bool mutex_signal::postConsume()
+{
+	if (!consumer_locked)
+		return false;
+	update_available = false;
+	if(!dataUnlock())
+		return false;
+
+	consumer_locked = false;
+	return true;
+}
+
+bool mutex_signal::producerInit()
+{
+	return producerLock();
+}
+
+bool mutex_signal::producerCleanup()
+{
+	if (producer_locked && !postProduce())
+		return false;
+
+	return producerUnlock();
+}
+
+bool mutex_signal::consumerInit()
+{
+	return consumerLock();
+}
+
+bool mutex_signal::consumerCleanup()
+{
+	if (consumer_locked && !postConsume())
+		return false;
+
+	return consumerUnlock();
+}

jclklib/common/mutex_signal.hpp

@@ -0,0 +1,49 @@
+#include <chrono>
+#include <pthread.h>
+
+#ifndef MUTEX_SIGNAL_HPP
+#define MUTEX_SIGNAL_HPP
+
+namespace mutex_signal_test {
+	class mutex_signal
+	{
+	private:
+		pthread_mutex_t p_lock;
+		pthread_mutex_t c_lock;
+		pthread_mutex_t d_lock;
+		bool update_available;
+
+		bool consumer_locked;
+		bool producer_locked;
+
+		bool consumerTryLock(bool &success);
+		bool consumerLock();
+		bool consumerUnlock();
+
+		bool producerTryLock(bool &success);
+		bool producerTimedLock(bool &success, std::chrono::time_point<std::chrono::system_clock> wait_until);
+		bool producerLock();
+		bool producerUnlock();
+
+		bool dataTryLock(bool &success);
+		bool dataTimedLock(bool &success, std::chrono::time_point<std::chrono::system_clock> wait_until);
+		bool dataLock();
+		bool dataUnlock();
+	public:
+		mutex_signal();
+		~mutex_signal();
+		bool init();
+
+		bool producerInit();
+		bool producerCleanup();
+		bool preProduce();
+		bool postProduce();
+
+		bool consumerInit();
+		bool consumerCleanup();
+		bool preConsume(bool &success, std::chrono::duration<unsigned, std::nano> wait_until);
+		bool postConsume();
+	};
+}
+
+#endif/*MUTEX_SIGNAL_HPP*/

jclklib/common/util.hpp

@@ -11,6 +11,8 @@
 #ifndef UTIL_HPP
 #define UTIL_HPP
 
+#include <common/mutex_signal.hpp>
+
 #define UNIQUE_TYPEOF(x) remove_reference<decltype(*(x).get())>::type
 #define FUTURE_TYPEOF(x) decltype((x).get())
 #define DECLARE_STATIC(x,...) decltype(x) x __VA_OPT__({) __VA_ARGS__ __VA_OPT__(})
@@ -36,20 +38,5 @@ bool isFutureSet(std::future<type> &f)
 	}
 
 
-namespace JClkLibCommon {
-	class MutexSignal {
-	private:
-		pthread_mutex_t c_lock;
-		pthread_mutex_t p_lock;
-	public:
-		MutexSignal();
-		bool init();
-		bool PreProduce(bool &have);
-		bool PostProduce();
-		bool PostConsume();
-		bool PreConsume();
-	};
-}
-
 #endif/*UTIL_HPP*/