any_sender_of<> did not support stop_when sender

[MoFHeka]

set_value cause template argument deduction/substitution failing

#include <unifex/any_sender_of.hpp>
#include <unifex/just.hpp>
#include <unifex/let_done.hpp>
#include <unifex/stop_when.hpp>
#include <chrono>
#include <memory>

int main() {
  using AnySender = unifex::any_sender_of<>;
  
  // Simulate the handler returning AnySender
  auto handler_sender = AnySender{unifex::just()};
  
  // Timeout duration
  std::chrono::seconds timeout(5);
  
  // This line causes compilation failure:
  // The issue is that schedule_after requires a scheduler context,
  // but we're trying to use it directly without a scheduler instance
  auto timeout_sender = std::move(handler_sender)
                        | unifex::stop_when(unifex::schedule_after(timeout));
  /*
  auto timeout_sender = std::move(handler_sender)  // Success
  */
  
  // Further processing with let_done
  auto result_sender = std::move(timeout_sender)
                       | unifex::let_done([]() {
                           return AnySender{unifex::just()};
                         });
  
  // Convert back to AnySender
  AnySender final_result{std::move(result_sender)};
  
  return 0;
}

[MaxMotovilov]

Could you link to a godbolt or gist with your failing example?

[MoFHeka]

@MaxMotovilov Thank you for response. Here is the code: https://godbolt.org/z/d7a4shbch

[MaxMotovilov]

It is a known issue, alas - it is possible to satisfy the get_scheduler() receiver query via any_sender_of<>, but the type-erased any_scheduler<> relied upon to do so does not provide time scheduling functionality. #296 is a very similar problem with task<>.

This is the example of passing any_scheduler<> through a receiver query on a type-erased sender:

template <typename... T>
using any_scheduled_sender_of = unifex::with_receiver_queries<
    unifex::overload<unifex::any_scheduler(const unifex::this_&) noexcept>(
        unifex::get_scheduler)>::any_sender_of<T...>;

as I mentioned above, it does not solve your immediate problem :-(

Type erasing facilities provided by libunifex are, unfortunately, incomplete. Perhaps it is possible to achieve your ends in some other way? Admittedly, I quite often pass around references to unifex::timed_single_thread_context in my code for this very reason...

[MoFHeka]

Thank. It's a pity that any_scheduled_sender_of doesn't work.

The reason for using the sender type erase is that in my code logic, different sender pipes will be returned for success or failure or other situations. In this case, what is the best way to organize the code?

[MaxMotovilov]

Well, one observation is that typically the scheduler used for timeouts is separate from the scheduler used for actual work, especially when the latter is single threaded. One reason to do it that way is to ensure that timeouts aren't affected by blocking calls, deadlocks or CPU saturation on the actual worker thread. Although it is by no means an absolute remedy, it at least provides ability to log errors in a timely manner or, if worse comes to worst, kill the process. In this situation the scheduler for timeouts cannot be obtained from the receiver anyways and has to be carried "on the side" (i.e. reference to context passed into sender's constructor).

Maybe I need to mention for clarity that in stop_when(sender, schedule_after(timeout)) the sender of the schedule_after() algorithm obtains the scheduler by performing get_scheduler(receiver) query on the receiver passed by the stop_when() algorithm into the former sender's connect() method. The two-argument overload of schedule_after() has an explicit scheduler to work with and does not need the receiver query.

Coming to the issue from a totally different direction, in sender pipelines you can often express conditionals related to success or failure of the previous step without type erasure. One good tool for this is just_void_or_done() in combination with side effects through context references held by let_value(). As an example I can point to the PR I merged yesterday (https://github.com/facebookexperimental/libunifex/pull/661/files). Such logic - while really satisfying for the original programmer to write :-D - may present a real problem from maintenance standpoint, thus co-routines are the usually preferred approach to express the conditional asynchronous flow.

[MoFHeka]

Coming to the issue from a totally different direction, in sender pipelines you can often express conditionals related to success or failure of the previous step without type erasure. One good tool for this is just_void_or_done() in combination with side effects through context references held by let_value(). As an example I can point to the PR I merged yesterday (https://github.com/facebookexperimental/libunifex/pull/661/files). Such logic - while really satisfying for the original programmer to write :-D - may present a real problem from maintenance standpoint, thus co-routines are the usually preferred approach to express the conditional asynchronous flow.

https://godbolt.org/z/EzaK3xqvx
when_any works. But its returning also could not be wrapped by any_sender_of.

Also I worry about if there is performance disadvantage with co-routines?

[MaxMotovilov]

when_any works. But its returning also could not be wrapped by any_sender_of.

Oh, I mostly meant it as an example of how to get around type erasure when you need to branch on done (or failure) - inside the implementation of when_any().

Also I worry about if there is performance disadvantage with co-routines?

There are costs associated with co-routines, yes. I never really had to conduct a real benchmarking to see just how considerable they were - my projects never involved short async loops or massive parallelism. The performance concern we usually have to deal with in Meta is binary size... not so much raw CPU or even heap use. What specific aspects of performance are you worried about?

[MoFHeka]

I‘m working on a RDMA pipe, so a sending/receiving routine may be very fast.
Besides, is it a good choice for variant_sender? what variant_sender exactly designed for?
@MaxMotovilov

[MaxMotovilov]

what variant_sender exactly designed for?

It's from well before my time and I never had a chance to use it; from looking at the code I can tell that it serves to select one of the specified senders based on receiver type - i.e. picks the sender with value signature compatible with what receiver expects. I do not know what the use case is - @ispeters?

[MoFHeka]

@MaxMotovilov The performance of unifex::when_any was not as expected. After the task sender was completed and did not reach the timeout sender called, it did not immediately move on to the next sender but got dead lock.
Any good idea for debugging?

using ServerProcessValidMessageTimeoutSenderType =                            //
    decltype(unifex::when_any(                                                //
      unifex::on(                                                             //
        std::declval<unifex::static_thread_pool>().get_scheduler(),           //
        std::declval<RequestHandlerReturnType>())                             //
        | unifex::let_error(TransformErrorToRpcException_{}),                 //
      unifex::on(                                                             //
        std::declval<unifex::timed_single_thread_context>().get_scheduler(),  //
        unifex::schedule_after(std::declval<std::chrono::seconds>()))         //
        | unifex::let_error(TransformErrorToRpcException_{})));

[MaxMotovilov]

I am not sure I understand - the quoted code is just a decltype, not an executed pipeline. As with any cancellation, the cancelled sender still must complete (with set_done() or otherwise) before the entire algorithm completes.

As an example - https://godbolt.org/z/x7WKr6Kz5 - works