[PATCH] libstdc++: Implementation of class strand

[PATCH] libstdc++: Implementation of class strand

[Alessio G. B.]

[-- 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;
     };

Re: [PATCH] libstdc++: Implementation of class strand

[Jonathan Wakely]

On 12/03/21 13:21 +0100, Alessio G. B. via Libstdc++ wrote:
>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.

Thanks for this patch. I'm not sure when I'll have time to review it,
and it might not be in time for the upcoming GCC 11 release. But the
patch has been received and will get reviewed, thanks.

N.B. patches being submitted for libstdc++ should be CC'd to the
gcc-patches list (where all GCC patches get sent) as well as the
libstdc++ list.

Re: [PATCH] libstdc++: Implementation of class strand

[Jonathan Wakely]

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

On Tue, 23 Mar 2021 at 16:39, Jonathan Wakely <jwakely@redhat.com> wrote:

> On 12/03/21 13:21 +0100, Alessio G. B. via Libstdc++ wrote:
> >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.
>
> Thanks for this patch. I'm not sure when I'll have time to review it,
> and it might not be in time for the upcoming GCC 11 release. But the
> patch has been received and will get reviewed, thanks.
>

Well I didn't think it would take me two years, sorry about that :-(

+ 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();
+ }

It looks like this will run a user-provided function while holding the
mutex lock. Won't that deadlock if the task added to the strand adds
another task to the same strand? Is that forbidden by some requirement in
the TS that I've forgotten?

Re: [PATCH] libstdc++: Implementation of class strand

[Thomas Rodgers]

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

IIRC, the Asio implementation checks to see if the current thread already
holds a lock, and if so, adopts the current lock and enqueues any child
tasks that are created and processes them after the user-provided function
returns, to avoid deadlock in the case that Jonathan outlines here.

On Wed, Mar 22, 2023 at 7:32 AM Jonathan Wakely via Libstdc++ <
libstdc++@gcc.gnu.org> wrote:

> On Tue, 23 Mar 2021 at 16:39, Jonathan Wakely <jwakely@redhat.com> wrote:
>
> > On 12/03/21 13:21 +0100, Alessio G. B. via Libstdc++ wrote:
> > >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.
> >
> > Thanks for this patch. I'm not sure when I'll have time to review it,
> > and it might not be in time for the upcoming GCC 11 release. But the
> > patch has been received and will get reviewed, thanks.
> >
>
> Well I didn't think it would take me two years, sorry about that :-(
>
> + 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();
> + }
>
> It looks like this will run a user-provided function while holding the
> mutex lock. Won't that deadlock if the task added to the strand adds
> another task to the same strand? Is that forbidden by some requirement in
> the TS that I've forgotten?
>
>