[PATCH] libstdc++: Implementation of class strand

["Alessio G. B." <alessiogiovanni.baroni@gmail.com>]

[-- Attachment #1: Type: text/plain, Size: 430 bytes --]

I expanded the implementation of the class strand of the Networking
TS. Essentially, I
implemented a token system so each thread knows when it can execute;
the system is organized
with 2 integers moving as a clock.

2021-03-12 Alessio G. Baroni <alessiogiovanni.baroni@gmail.com>

        libstdc++-v3:
            * include/experimental/executor: Implemented algorithm for
serial execution regardless of the underlying Executor.

[-- Attachment #2: strand.patch --]
[-- Type: text/x-patch, Size: 4684 bytes --]

diff --git a/libstdc++-v3/include/experimental/executor b/libstdc++-v3/include/experimental/executor
index c670f2739b6..afd5a64366f 100644
--- a/libstdc++-v3/include/experimental/executor
+++ b/libstdc++-v3/include/experimental/executor
@@ -1478,13 +1478,16 @@ inline namespace v1
 
       // construct / copy / destroy:
 
-      strand(); // TODO make state
+      strand() : _M_state(std::make_shared<_State>()) { }
 
-      explicit strand(_Executor __ex) : _M_inner_ex(__ex) { } // TODO make state
+      explicit strand(_Executor __ex)
+	  : _M_state(std::make_shared<_State>()),
+	    _M_inner_ex(__ex) { }
 
       template<typename _Alloc>
 	strand(allocator_arg_t, const _Alloc& __a, _Executor __ex)
-	: _M_inner_ex(__ex) { } // TODO make state
+	: _M_state(std::allocate_shared<_State>(__a)),
+	  _M_inner_ex(__ex) { }
 
       strand(const strand& __other) noexcept
       : _M_state(__other._M_state), _M_inner_ex(__other._M_inner_ex) { }
@@ -1508,8 +1511,10 @@ inline namespace v1
 	static_assert(is_copy_assignable<_Executor>::value,
 		      "inner executor type must be CopyAssignable");
 
-	// TODO lock __other
-	// TODO copy state
+#if defined(_GLIBCXX_HAS_GTHREADS)
+	std::lock_guard<std::mutex> __lock(__other._M_state->_M_mutex);
+#endif
+	_M_state = __other._M_state;
 	_M_inner_ex = __other._M_inner_ex;
 	return *this;
       }
@@ -1520,7 +1525,10 @@ inline namespace v1
 	static_assert(is_move_assignable<_Executor>::value,
 		      "inner executor type must be MoveAssignable");
 
-	// TODO move state
+#if defined(_GLIBCXX_HAS_GTHREADS)
+	std::lock_guard<std::mutex> __lock(__other._M_state->_M_mutex);
+#endif
+	_M_state = std::move(__other._M_state);
 	_M_inner_ex = std::move(__other._M_inner_ex);
 	return *this;
       }
@@ -1532,8 +1540,10 @@ inline namespace v1
 	  static_assert(is_convertible<_OtherExecutor, _Executor>::value,
 			"inner executor type must be compatible");
 
-	  // TODO lock __other
-	  // TODO copy state
+#if defined(_GLIBCXX_HAS_GTHREADS)
+	  std::lock_guard<std::mutex> __lock(__other._M_state->_M_mutex);
+#endif
+	  _M_state = __other._M_state;
 	  _M_inner_ex = __other._M_inner_ex;
 	  return *this;
 	}
@@ -1545,15 +1555,16 @@ inline namespace v1
 	  static_assert(is_convertible<_OtherExecutor, _Executor>::value,
 			"inner executor type must be compatible");
 
-	  // TODO move state
+#if defined(_GLIBCXX_HAS_GTHREADS)
+	  std::lock_guard<std::mutex> __lock(__other._M_state->_M_mutex);
+#endif
+	  _M_state = std::move(__other._M_state);
 	  _M_inner_ex = std::move(__other._M_inner_ex);
 	  return *this;
 	}
 
       ~strand()
       {
-	// the task queue outlives this object if non-empty
-	// TODO create circular ref in queue?
       }
 
       // strand operations:
@@ -1585,7 +1596,21 @@ inline namespace v1
 
       template<typename _Func, typename _Alloc>
 	void
-	post(_Func&& __f, const _Alloc& __a) const; // TODO
+	post(_Func&& __f, const _Alloc& __a) const
+        {
+#if defined(_GLIBCXX_HAS_GTHREADS)
+	  auto token = _M_state->new_token();
+
+	  _M_inner_ex.post(
+	      [token, __state = _M_state, &__f]()
+	      {
+		__state->invoke(token, std::forward<_Func>(__f));
+	      },
+	      __a);
+#else
+	  _M_inner_ex.post(std::forward<_Func>(__f), __a);
+#endif
+	}
 
       template<typename _Func, typename _Alloc>
 	void
@@ -1597,19 +1622,56 @@ inline namespace v1
       operator==(const strand& __a, const strand& __b)
       { return __a._M_state == __b._M_state; }
 
-      // TODO add synchronised queue
       struct _State
       {
 #if defined(_GLIBCXX_HAS_GTHREADS)
+	_State()
+	    : _M_running_on(std::this_thread::get_id()),
+	      _M_next_token(0),
+	      _M_last_token(0) { }
+
 	bool
 	running_in_this_thread() const noexcept
 	{ return std::this_thread::get_id() == _M_running_on; }
 
+	unsigned int
+	new_token()
+	{
+	  std::lock_guard<std::mutex> __lock(_M_mutex);
+
+	  return _M_last_token++;
+	}
+
+	template<typename _Func>
+	void
+	invoke(unsigned int token, _Func&& __f)
+	{
+	  std::unique_lock<std::mutex> __lock(_M_mutex);
+
+	  _M_cv.wait(__lock,
+		     [token, next_token = _M_next_token]()
+		     { return token == next_token; });
+
+	  try { decay_t<_Func>{std::forward<_Func>(__f)}(); }
+	  catch(...) { }
+
+	  _M_next_token++;
+
+	  __lock.unlock();
+
+	  _M_cv.notify_all();
+	}
+
 	std::thread::id _M_running_on;
+	std::mutex _M_mutex;
+	std::condition_variable _M_cv;
+	unsigned int _M_next_token;
+	unsigned int _M_last_token;
 #else
-	bool running_in_this_thread() const { return true; }
+	bool running_in_this_thread() const noexcept { return true; }
 #endif
       };
+
       shared_ptr<_State> _M_state;
       _Executor _M_inner_ex;
     };