Update returns Command #282

adwhit · 2024-10-22T10:23:17Z

I have talked before that I don't like how fn update(..) does not return anything, all the effects are spawned implicitly by the capabilities. I thought I'd see whether I could make it more Elm-like and return a Command instead.

Turns out to be quite invasive but I do think it is an improvement. Especially, taking the Ev generic out of every Capability is nice. And totally removing the need for a Compose capability.

charypar

I've only read the core part of this, but have enough questions to start a conversation :)

charypar · 2024-10-22T17:38:43Z

crux_core/src/capabilities/render.rs

+    pub fn render<Event>(&self) -> Command<Event> {
+        Command::empty_effect(self.render_async())
+    }


Do I imagine this correctly that the constraint which ensures that effects coming out of an update function only ever result in Events the update function is able to process (potentially wrapped with a parent app Event) has moved from being an argument of the capability itself to being tied to the return value of the capability's API?

That hopefully means the bound is still strict and we don't allow Apps to create effects resulting in events of any type, except their own.

I assume it also means that Commands can be "lifted" to a parent app (meaning their events will be auto-wrapped with a given event)?

(future me: yes, they can)

charypar · 2024-10-22T17:47:18Z

crux_core/src/capabilities/render.rs

+    pub fn render_async(&self) -> impl Future<Output = ()> {
        let ctx = self.context.clone();
-        self.context.spawn(async move {
+        async move {
            ctx.notify_shell(RenderOperation).await;
-        });
+        }


I wonder whether there's a way to avoid this shell side-channel altogether.

Is there a way we could have a specific future type for each cardinality of the shell interaction (notify/request/subscribe), which exposes the operation it was created with, so that the operation accessible to the executor?

I can't quite see it clearly enough, but I think it would allow the executor to be aware of these "special futures" and whenever the tasks yield because they are awaiting on these special futures, the executor would know how to
a) get the operation out and send it to the shell
b) resolve the operation on that future so that it can proceed, which should result on them polling as ready

That should theoretically eliminate this channel altogether. The futures would just be a kind of "yield point carrying the value that needs processing". (That might make them more easily visible to tests as well)

I'm not totally sure this is possible to implement, but if it is, it would mean capabilities would be fully free of context.

Thinking about it a bit more - the problem with that idea is that those special futures will be inside one of the states of an anonymous future made by an async block, which is opaque, isn't it... the "innermost" future needs some way of exposing the operation that can resolve it to the outside.

But prehaps this could be done on a more local scale within the Command. The command could take a function which gets an equivalent of the context provided by the executor later. That context would just have a local channel back up to the command, allowing to bypass the anonymous future layers.... or something along those lines.

charypar · 2024-10-22T17:50:02Z

crux_core/src/capability/executor.rs

+pub(crate) struct QueuingExecutor<Event> {
    ready_queue: Receiver<TaskId>,
    ready_sender: Sender<TaskId>,
-    tasks: Mutex<Slab<Option<BoxFuture>>>,
+    tasks: Mutex<Slab<Option<BoxFuture<Event>>>>,


The Event ends up being the top-level app's Event type, presumably?

charypar · 2024-10-22T17:52:38Z

crux_core/src/capability/executor.rs

    Missing,
    Unavailable,
    Suspended,
-    Completed,
+    Completed(Command<Event>),


I'm not sure I follow this part - why does completed task hold aCommand? Is it the command that started it? or the command that is to be processed by the shell...? I'm not totally sure what's going on 😅

I would've thought the task is either just done, or if anything it's done and (optionally) resulted in an Event...?

I suppose the granularity of tasks may now be different than before, or maybe I'm just lost 🤔

When a task resolves it always returns a command.

Think of the existing model - e.g. an HTTP request that resolves to an event. When the request resolves, the resulting Event is implicitly spawned onto the executor through the spawn_event channel that is part of the capability context.

With the changes in this PR, the spawn_event channel has gone, and the event (or any other kind of Command) is returned directly from the resolved future. The command needs to be explicitly handled - either fed into app.update if it is an event, or spawned onto the executor if it is an effect (or else the command may be Command::none() in which case there is nothing left to do).

This happens in code via an interplay between Core::process and QueueingExecutor::run_all.

Wait, are you saying the task completes every time the code in the future(s) hits an .await on one of the Crux specific interactions?

No, only when it resolves. i.e. the future is polled to completion (potentially through multuple .awaits). Uh, I suppose in crux we are using 'resolve' to mean something else (Update::resolve means to resolve an async effect) so maybe not a good word to use.

So then I'm back to the question of how does it complete with a Command which is not just an Event?

I'm not sure what that would mean in principle. The parent future of the effect task has finished, there are no more effects to do, the only possible outcomes are either nothing (we're done) or an event back into the app, right? What am I missing?

It could be made to return Option<Event> instead, but you would lose some expressivity.

A Command allows one to return any combination of events and effects to run next. This is useful e.g. to do an http request and a render call concurrently. You could do them in a single future with a future::join call, but then they are coupled together. Returning both with Command::effect(render_fut).join(http_fut) decouples them.

example:

crux/examples/cat_facts/shared/src/app.rs

Lines 143 to 156 in 03a7c64

let eff1 = caps

.http

.get(FACT_API_URL)

.expect_json()

.send_and_respond(Event::SetFact);

let eff2 = caps

.http

.get(IMAGE_API_URL)

.expect_json()

.send_and_respond(Event::SetImage);

let eff3 = caps.render.render();

eff1.join(eff2).join(eff3)

Actually possibly talking cross-purposes here. The above applies to the return type for fn update(..). You want to know why any spawned future must return Command? The answer is it seems reasonable that you might want to spawn a future (or several futures) from within another future (remember we have no way to spawn futures other than to return them, as context.spawn has gone).

You might argue that futures shouldn't attempt to spawn futures directly, they should return events which subsequently kick off the futures. Maybe true? Returning a Command seems to give maximum flexibility

I seee - if you wanted to start a separate task that runs concurrently without awaiting the result. Yea, that is something we need o retain as an ability...

I suggest staring at

crux/examples/counter/shared/src/capabilities/sse.rs

Lines 34 to 65 in 03a7c64

pub fn get_json<F, Ev, T>(&self, url: impl AsRef<str>, make_event: F) -> Command<Ev>

where

F: Fn(T) -> Ev + Clone + Send + 'static,

Ev: Send + 'static,

T: DeserializeOwned,

{

let context = self.context.clone();

let url = url.as_ref().to_string();

let stream = context

.stream_from_shell(SseRequest { url })

.flat_map(move |response| match response {

SseResponse::Chunk(data) => {

// this decoder is async (even though for our purposed it doesn't need to be)

// which makes the following code a bit fiddly

let reader = decode(Cursor::new(data));

let make_event = make_event.clone();

let inner = reader.map(move |sse_event| {

if let Ok(Event::Message(msg)) = sse_event {

let t: T = serde_json::from_slice(msg.data()).unwrap();

let make_event = make_event.clone();

Command::event(make_event(t))

} else {

Command::none()

}

});

Box::pin(inner) as futures::stream::BoxStream<_>

}

SseResponse::Done => Box::pin(futures::stream::empty()),

});

Command::stream(stream)

}

until it makes sense :)

It's a stream where each output is itself a stream of Commands

It's a stream where each output is itself a stream of Commands

That seems a step too far 😅

I think the problem there starts with insisting on Event being a return, or generally the command async code working towards a final return value. I don't know that's the most straightforward approach to it.

The async code in capabilities (and now in Command) is quite intentionally pretty imperative, because it's "process" code, rather than a computation (as in - more Turing, less Church). It's much more naturally a "do this, then do that, then do that, then finish". And yes we can model that with a recursion, but as can be seen in the code above^ it's really not very straightforward that way. We could do better with some monadic style combinators like Command::event(event).and_then(...) but... I think async is just a nicer interface for this type of work.

charypar · 2024-10-22T17:54:10Z

crux_core/src/capability/executor.rs


        let future = {
            let counter = counter.clone();
            async move {
                assert_eq!(Arc::strong_count(&counter), 2);
                ShellRequest::<()>::new().await;
+                Command::none()


crux_core/src/lib.rs

charypar · 2024-10-22T17:58:30Z

crux_core/src/lib.rs

+    pub fn map<F, Event2>(self, func: F) -> Command<Event2>
+    where
+        F: Fn(Event) -> Event2 + Send + Clone + 'static,
+        Event: 'static,
+    {
+        let inner = match self.inner {
+            CommandInner::None => CommandInner::None,
+            CommandInner::Event(ev) => CommandInner::Event(func(ev)),
+            CommandInner::Effects(effects) => {
+                let effects = effects
+                    .into_iter()
+                    .map(|fut| {
+                        Box::pin(fut.map({
+                            let f = func.clone();
+                            move |cmd| cmd.map(f)
+                        })) as BoxFuture<'static, Command<Event2>>
+                    })
+                    .collect();
+                CommandInner::Effects(effects)
+            }
+        };
+        Command { inner }
+    }
+}


crux_core/src/lib.rs

StuartHarris · 2024-10-23T08:10:19Z

crux_time/src/lib.rs

@@ -105,19 +89,15 @@ where
        self.context.request_from_shell(TimeRequest::Now).await
    }

-    /// Ask to receive a notification when the specified [`Instant`] has arrived.
-    pub fn notify_at<F>(&self, instant: Instant, callback: F)


did we lose notify_at()?

It has re-appeared :)

StuartHarris · 2024-10-23T08:15:22Z

examples/cat_facts/shared/src/app.rs

+                    let _result = caps.key_value.set_async(KEY.to_string(), bytes).await;
+                    Command::none()
+                })
+                .join(async move {


presumably this could just be

.join(caps.render.render())

or is this just to show that either is valid?

I have reworked the Commands and effects to make this cleaner

adwhit · 2024-10-25T14:31:20Z

I have merged in master and updated the doc tests and example apps, so now it gives a better idea of what the change would look like from POV of a user. The documentation hasn't been updated though, that's probably a whole separate task.

adwhit added 10 commits October 21, 2024 09:07

checkpoint

753d5de

experiment

d1af942

eh hmm

fb23fd0

fix core tests

19f285b

fools errand?

9411423

progress on capabilities

892c40e

test fixes

3788bdd

remove commented-out code

f1fa0e6

render returns a Command

4927779

make Command opaque

b35127b

charypar reviewed Oct 22, 2024

View reviewed changes

StuartHarris reviewed Oct 23, 2024

View reviewed changes

adwhit added 11 commits October 24, 2024 18:57

add Command::Multiple

7e7e542

cat facts fixed

63bc435

fix core doctests

41fcf80

Merge branch 'master' into alex/command

fef27fc

post-merge fixes

5a75d96

update crux_http doctests

7db4eb7

update crux_macros doctest

0ed3ad7

tentatively introduce Command::stream

c7ac91c

update rest of the examples

e8b24d2

tidy

55bf96d

remove OneOrMany

03a7c64

adwhit marked this pull request as ready for review October 25, 2024 14:27

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Update returns Command #282

Update returns Command #282

adwhit commented Oct 22, 2024 •

edited

Loading

charypar left a comment

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

adwhit Oct 25, 2024

charypar Oct 29, 2024

adwhit Oct 29, 2024

charypar Nov 4, 2024

adwhit Nov 4, 2024 •

edited

Loading

adwhit Nov 4, 2024 •

edited

Loading

charypar Nov 4, 2024

adwhit Nov 4, 2024 •

edited

Loading

charypar Nov 5, 2024

charypar Oct 22, 2024

charypar Oct 22, 2024

StuartHarris Oct 23, 2024

adwhit Oct 25, 2024

StuartHarris Oct 23, 2024

adwhit Oct 25, 2024

adwhit commented Oct 25, 2024

	let eff1 = caps
	.http
	.get(FACT_API_URL)
	.expect_json()
	.send_and_respond(Event::SetFact);

	let eff2 = caps
	.http
	.get(IMAGE_API_URL)
	.expect_json()
	.send_and_respond(Event::SetImage);

	let eff3 = caps.render.render();
	eff1.join(eff2).join(eff3)

	pub fn get_json<F, Ev, T>(&self, url: impl AsRef<str>, make_event: F) -> Command<Ev>
	where
	F: Fn(T) -> Ev + Clone + Send + 'static,
	Ev: Send + 'static,
	T: DeserializeOwned,
	{
	let context = self.context.clone();
	let url = url.as_ref().to_string();

	let stream = context
	.stream_from_shell(SseRequest { url })
	.flat_map(move \|response\| match response {
	SseResponse::Chunk(data) => {
	// this decoder is async (even though for our purposed it doesn't need to be)
	// which makes the following code a bit fiddly
	let reader = decode(Cursor::new(data));
	let make_event = make_event.clone();
	let inner = reader.map(move \|sse_event\| {
	if let Ok(Event::Message(msg)) = sse_event {
	let t: T = serde_json::from_slice(msg.data()).unwrap();
	let make_event = make_event.clone();
	Command::event(make_event(t))
	} else {
	Command::none()
	}
	});
	Box::pin(inner) as futures::stream::BoxStream<_>
	}
	SseResponse::Done => Box::pin(futures::stream::empty()),
	});
	Command::stream(stream)
	}

Update returns Command #282

Are you sure you want to change the base?

Update returns Command #282

Conversation

adwhit commented Oct 22, 2024 • edited Loading

charypar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adwhit Nov 4, 2024 • edited Loading

Choose a reason for hiding this comment

adwhit Nov 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adwhit Nov 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adwhit commented Oct 25, 2024

adwhit commented Oct 22, 2024 •

edited

Loading

adwhit Nov 4, 2024 •

edited

Loading

adwhit Nov 4, 2024 •

edited

Loading

adwhit Nov 4, 2024 •

edited

Loading