src/sys/component_manager/src/builtin/system_controller.rs - fuchsia - Git at Google

 // Copyright 2019 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 use {
     crate::{
         capability::{CapabilityProvider, CapabilitySource, InternalCapability},
         channel,
         model::{
             actions::{ActionSet, ShutdownAction},
             error::ModelError,
             hooks::{Event, EventPayload, EventType, Hook, HooksRegistration},
             model::Model,
         },
     },
     anyhow::{format_err, Context as _, Error},
     async_trait::async_trait,
     cm_rust::CapabilityName,
     fidl::endpoints::ServerEnd,
     fidl_fuchsia_sys2::*,
     fuchsia_async::{self as fasync},
     fuchsia_zircon as zx,
     futures::prelude::*,
     lazy_static::lazy_static,
     log::warn,
     std::{
         path::PathBuf,
         sync::{Arc, Weak},
         time::Duration,
     },
 };

 lazy_static! {
     pub static ref SYSTEM_CONTROLLER_CAPABILITY_NAME: CapabilityName =
         "fuchsia.sys2.SystemController".into();
 }

 #[derive(Clone)]
 pub struct SystemController {
     model: Weak<Model>,
     shutdown_timeout: Duration,
 }

 impl SystemController {
     pub fn new(model: Weak<Model>, shutdown_timeout: Duration) -> Self {
         Self { model, shutdown_timeout }
     }

     pub fn hooks(self: &Arc<Self>) -> Vec<HooksRegistration> {
         vec![HooksRegistration::new(
             "SystemController",
             vec![EventType::CapabilityRouted],
             Arc::downgrade(self) as Weak<dyn Hook>,
         )]
     }

     async fn on_framework_capability_routed_async<'a>(
         self: Arc<Self>,
         capability: &'a InternalCapability,
         capability_provider: Option<Box<dyn CapabilityProvider>>,
     ) -> Result<Option<Box<dyn CapabilityProvider>>, ModelError> {
         if capability.matches_protocol(&SYSTEM_CONTROLLER_CAPABILITY_NAME) {
             Ok(Some(Box::new(SystemControllerCapabilityProvider::new(
                 self.model.clone(),
                 self.shutdown_timeout.clone(),
             )) as Box<dyn CapabilityProvider>))
         } else {
             Ok(capability_provider)
         }
     }
 }

 #[async_trait]
 impl Hook for SystemController {
     async fn on(self: Arc<Self>, event: &Event) -> Result<(), ModelError> {
         if let Ok(EventPayload::CapabilityRouted {
             source: CapabilitySource::Builtin { capability },
             capability_provider,
         }) = &event.result
         {
             let mut capability_provider = capability_provider.lock().await;
             *capability_provider = self
                 .on_framework_capability_routed_async(&capability, capability_provider.take())
                 .await?;
         };
         Ok(())
     }
 }

 pub struct SystemControllerCapabilityProvider {
     model: Weak<Model>,
     request_timeout: Duration,
 }

 impl SystemControllerCapabilityProvider {
     // TODO (jmatt) allow timeout to be supplied in the constructor
     pub fn new(model: Weak<Model>, request_timeout: Duration) -> Self {
         Self { model, request_timeout }
     }

     async fn open_async(self, mut stream: SystemControllerRequestStream) -> Result<(), Error> {
         while let Some(request) = stream.try_next().await? {
             // TODO(jmatt) There is the potential for a race here. If
             // the thing that called SystemController.Shutdown is a
             // component that component_manager controls, it should
             // be gone by now. Sending a response doesn't make a lot
             // of sense in this case. However, the caller might live
             // outside component_manager, in which case a response
             // does make sense. Figure out if our behavior should be
             // different and/or whether we should drop the response
             // from this API.
             match request {
                 // Shutting down the root component causes component_manager to
                 // exit. main.rs waits on the model to observe the root realm
                 // disappear.
                 SystemControllerRequest::Shutdown { responder } => {
                     let timeout = zx::Duration::from(self.request_timeout);
                     fasync::Task::spawn(async move {
                         fasync::Timer::new(fasync::Time::after(timeout)).await;
                         panic!("Component manager did not complete shutdown in allowed time.");
                     })
                     .detach();
                     let root_realm = self
                         .model
                         .upgrade()
                         .ok_or(format_err!("model is dropped"))?
                         .root_realm
                         .clone();
                     ActionSet::register(root_realm, ShutdownAction::new())
                         .await
                         .context("got error waiting for shutdown action to complete")?;
                     match responder.send() {
                         Ok(()) => {}
                         Err(e) => {
                             println!(
                                 "error sending response to shutdown requester:\
                                  {}\n shut down proceeding",
                                 e
                             );
                         }
                     }
                 }
             }
         }
         Ok(())
     }
 }

 #[async_trait]
 impl CapabilityProvider for SystemControllerCapabilityProvider {
     async fn open(
         self: Box<Self>,
         _flags: u32,
         _open_mode: u32,
         _relative_path: PathBuf,
         server_end: &mut zx::Channel,
     ) -> Result<(), ModelError> {
         let server_end = channel::take_channel(server_end);
         let server_end = ServerEnd::<SystemControllerMarker>::new(server_end);
         let stream: SystemControllerRequestStream =
             server_end.into_stream().map_err(ModelError::stream_creation_error)?;
         fasync::Task::spawn(async move {
             let result = self.open_async(stream).await;
             if let Err(e) = result {
                 warn!("SystemController.open failed: {}", e);
             }
         })
         .detach();

         Ok(())
     }
 }

 #[cfg(test)]
 mod tests {
     use {
         super::*,
         crate::model::{
             binding::Binder,
             hooks::{EventType, Hook, HooksRegistration},
             realm::BindReason,
             testing::test_helpers::{
                 component_decl_with_test_runner, ActionsTest, ComponentDeclBuilder, ComponentInfo,
             },
         },
         async_trait::async_trait,
         fidl::endpoints,
         fidl_fuchsia_sys2 as fsys,
         moniker::AbsoluteMoniker,
         std::{boxed::Box, convert::TryFrom, sync::Arc, time::Duration},
     };

     /// Use SystemController to shut down a system whose root has the child `a`
     /// and `a` has descendents as shown in the diagram below.
     ///  a
     ///   \
     ///    b
     ///   / \
     ///  c   d
     #[fuchsia_async::run_singlethreaded(test)]
     async fn test_system_controller() {
         // Configure and start realm
         let components = vec![
             ("root", ComponentDeclBuilder::new().add_lazy_child("a").build()),
             ("a", ComponentDeclBuilder::new().add_eager_child("b").build()),
             ("b", ComponentDeclBuilder::new().add_eager_child("c").add_eager_child("d").build()),
             ("c", component_decl_with_test_runner()),
             ("d", component_decl_with_test_runner()),
         ];
         let test = ActionsTest::new("root", components, None).await;
         let realm_a = test.look_up(vec!["a:0"].into()).await;
         let realm_b = test.look_up(vec!["a:0", "b:0"].into()).await;
         let realm_c = test.look_up(vec!["a:0", "b:0", "c:0"].into()).await;
         let realm_d = test.look_up(vec!["a:0", "b:0", "d:0"].into()).await;
         test.model
             .bind(&realm_a.abs_moniker, &BindReason::BindChild { parent: AbsoluteMoniker::root() })
             .await
             .expect("could not bind to a");

         // Wire up connections to SystemController
         let sys_controller = Box::new(SystemControllerCapabilityProvider::new(
             Arc::downgrade(&test.model),
             // allow simulated shutdown to take up to 30 days
             Duration::from_secs(60 * 60 * 24 * 30),
         ));
         let (client_channel, server_channel) =
             endpoints::create_endpoints::<fsys::SystemControllerMarker>()
                 .expect("failed creating channel endpoints");
         let mut server_channel = server_channel.into_channel();
         sys_controller
             .open(0, 0, PathBuf::new(), &mut server_channel)
             .await
             .expect("failed to open capability");
         let controller_proxy =
             client_channel.into_proxy().expect("failed converting endpoint into proxy");

         let root_realm_info = ComponentInfo::new(test.model.root_realm.clone()).await;
         let realm_a_info = ComponentInfo::new(realm_a.clone()).await;
         let realm_b_info = ComponentInfo::new(realm_b.clone()).await;
         let realm_c_info = ComponentInfo::new(realm_c.clone()).await;
         let realm_d_info = ComponentInfo::new(realm_d.clone()).await;

         // Check that the root realm is still here
         root_realm_info.check_not_shut_down(&test.runner).await;
         realm_a_info.check_not_shut_down(&test.runner).await;
         realm_b_info.check_not_shut_down(&test.runner).await;
         realm_c_info.check_not_shut_down(&test.runner).await;
         realm_d_info.check_not_shut_down(&test.runner).await;

         // Ask the SystemController to shut down the system and wait to be
         // notified that the room realm stopped.
         let completion = test.builtin_environment.wait_for_root_realm_stop();
         controller_proxy.shutdown().await.expect("shutdown request failed");
         completion.await;

         // Check state bits to confirm root realm looks shut down
         root_realm_info.check_is_shut_down(&test.runner).await;
         realm_a_info.check_is_shut_down(&test.runner).await;
         realm_b_info.check_is_shut_down(&test.runner).await;
         realm_c_info.check_is_shut_down(&test.runner).await;
         realm_d_info.check_is_shut_down(&test.runner).await;
     }

     #[test]
     #[should_panic(expected = "Component manager did not complete shutdown in allowed time.")]
     fn test_timeout() {
         const TIMEOUT_SECONDS: i64 = 6;
         const EVENT_PAUSE_SECONDS: i64 = TIMEOUT_SECONDS + 1;
         struct StopHook;
         #[async_trait]
         impl Hook for StopHook {
             async fn on(self: Arc<Self>, _event: &Event) -> Result<(), ModelError> {
                 fasync::Timer::new(fasync::Time::after(zx::Duration::from_seconds(
                     EVENT_PAUSE_SECONDS.into(),
                 )))
                 .await;
                 Ok(())
             }
         }

         let mut exec = fasync::Executor::new_with_fake_time().unwrap();
         let mut test_logic = Box::pin(async {
             // Configure and start realm
             let components = vec![
                 ("root", ComponentDeclBuilder::new().add_eager_child("a").build()),
                 ("a", ComponentDeclBuilder::new().build()),
             ];

             let s = StopHook {};
             let s_hook: Arc<dyn Hook> = Arc::new(s);
             let hooks_reg = HooksRegistration::new(
                 "stop hook",
                 vec![EventType::Stopped],
                 Arc::downgrade(&s_hook),
             );

             let test = ActionsTest::new_with_hooks("root", components, None, vec![hooks_reg]).await;
             let realm_a = test.look_up(vec!["a:0"].into()).await;
             test.model
                 .bind(
                     &realm_a.abs_moniker,
                     &BindReason::BindChild { parent: AbsoluteMoniker::root() },
                 )
                 .await
                 .expect("could not bind to a");

             // Wire up connections to SystemController
             let sys_controller = Box::new(SystemControllerCapabilityProvider::new(
                 Arc::downgrade(&test.model),
                 // require shutdown in a second
                 Duration::from_secs(u64::try_from(TIMEOUT_SECONDS).unwrap()),
             ));
             let (client_channel, server_channel) =
                 endpoints::create_endpoints::<fsys::SystemControllerMarker>()
                     .expect("failed creating channel endpoints");
             let mut server_channel = server_channel.into_channel();
             sys_controller
                 .open(0, 0, PathBuf::new(), &mut server_channel)
                 .await
                 .expect("failed to open capability");
             let controller_proxy =
                 client_channel.into_proxy().expect("failed converting endpoint into proxy");

             let root_realm_info = ComponentInfo::new(test.model.root_realm.clone()).await;
             let realm_a_info = ComponentInfo::new(realm_a.clone()).await;

             // Check that the root realm is still here
             root_realm_info.check_not_shut_down(&test.runner).await;
             realm_a_info.check_not_shut_down(&test.runner).await;

             // Ask the SystemController to shut down the system and wait to be
             // notified that the room realm stopped.
             let _completion = test.builtin_environment.wait_for_root_realm_stop();
             controller_proxy.shutdown().await.expect("shutdown request failed");
         });

         assert_eq!(std::task::Poll::Pending, exec.run_until_stalled(&mut test_logic));

         let new_time = fasync::Time::from_nanos(
             exec.now().into_nanos() + zx::Duration::from_seconds(TIMEOUT_SECONDS).into_nanos(),
         );

         exec.set_fake_time(new_time);
         exec.wake_expired_timers();

         assert_eq!(std::task::Poll::Pending, exec.run_until_stalled(&mut test_logic));
     }
 }
	// Copyright 2019 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	use {
	crate::{
	capability::{CapabilityProvider, CapabilitySource, InternalCapability},
	channel,
	model::{
	actions::{ActionSet, ShutdownAction},
	error::ModelError,
	hooks::{Event, EventPayload, EventType, Hook, HooksRegistration},
	model::Model,
	},
	},
	anyhow::{format_err, Context as _, Error},
	async_trait::async_trait,
	cm_rust::CapabilityName,
	fidl::endpoints::ServerEnd,
	fidl_fuchsia_sys2::*,
	fuchsia_async::{self as fasync},
	fuchsia_zircon as zx,
	futures::prelude::*,
	lazy_static::lazy_static,
	log::warn,
	std::{
	path::PathBuf,
	sync::{Arc, Weak},
	time::Duration,
	},
	};

	lazy_static! {
	pub static ref SYSTEM_CONTROLLER_CAPABILITY_NAME: CapabilityName =
	"fuchsia.sys2.SystemController".into();
	}

	#[derive(Clone)]
	pub struct SystemController {
	model: Weak<Model>,
	shutdown_timeout: Duration,
	}

	impl SystemController {
	pub fn new(model: Weak<Model>, shutdown_timeout: Duration) -> Self {
	Self { model, shutdown_timeout }
	}

	pub fn hooks(self: &Arc<Self>) -> Vec<HooksRegistration> {
	vec![HooksRegistration::new(
	"SystemController",
	vec![EventType::CapabilityRouted],
	Arc::downgrade(self) as Weak<dyn Hook>,
	)]
	}

	async fn on_framework_capability_routed_async<'a>(
	self: Arc<Self>,
	capability: &'a InternalCapability,
	capability_provider: Option<Box<dyn CapabilityProvider>>,
	) -> Result<Option<Box<dyn CapabilityProvider>>, ModelError> {
	if capability.matches_protocol(&SYSTEM_CONTROLLER_CAPABILITY_NAME) {
	Ok(Some(Box::new(SystemControllerCapabilityProvider::new(
	self.model.clone(),
	self.shutdown_timeout.clone(),
	)) as Box<dyn CapabilityProvider>))
	} else {
	Ok(capability_provider)
	}
	}
	}

	#[async_trait]
	impl Hook for SystemController {
	async fn on(self: Arc<Self>, event: &Event) -> Result<(), ModelError> {
	if let Ok(EventPayload::CapabilityRouted {
	source: CapabilitySource::Builtin { capability },
	capability_provider,
	}) = &event.result
	{
	let mut capability_provider = capability_provider.lock().await;
	*capability_provider = self
	.on_framework_capability_routed_async(&capability, capability_provider.take())
	.await?;
	};
	Ok(())
	}
	}

	pub struct SystemControllerCapabilityProvider {
	model: Weak<Model>,
	request_timeout: Duration,
	}

	impl SystemControllerCapabilityProvider {
	// TODO (jmatt) allow timeout to be supplied in the constructor
	pub fn new(model: Weak<Model>, request_timeout: Duration) -> Self {
	Self { model, request_timeout }
	}

	async fn open_async(self, mut stream: SystemControllerRequestStream) -> Result<(), Error> {
	while let Some(request) = stream.try_next().await? {
	// TODO(jmatt) There is the potential for a race here. If
	// the thing that called SystemController.Shutdown is a
	// component that component_manager controls, it should
	// be gone by now. Sending a response doesn't make a lot
	// of sense in this case. However, the caller might live
	// outside component_manager, in which case a response
	// does make sense. Figure out if our behavior should be
	// different and/or whether we should drop the response
	// from this API.
	match request {
	// Shutting down the root component causes component_manager to
	// exit. main.rs waits on the model to observe the root realm
	// disappear.
	SystemControllerRequest::Shutdown { responder } => {
	let timeout = zx::Duration::from(self.request_timeout);
	fasync::Task::spawn(async move {
	fasync::Timer::new(fasync::Time::after(timeout)).await;
	panic!("Component manager did not complete shutdown in allowed time.");
	})
	.detach();
	let root_realm = self
	.model
	.upgrade()
	.ok_or(format_err!("model is dropped"))?
	.root_realm
	.clone();
	ActionSet::register(root_realm, ShutdownAction::new())
	.await
	.context("got error waiting for shutdown action to complete")?;
	match responder.send() {
	Ok(()) => {}
	Err(e) => {
	println!(
	"error sending response to shutdown requester:\
	{}\n shut down proceeding",
	e
	);
	}
	}
	}
	}
	}
	Ok(())
	}
	}

	#[async_trait]
	impl CapabilityProvider for SystemControllerCapabilityProvider {
	async fn open(
	self: Box<Self>,
	_flags: u32,
	_open_mode: u32,
	_relative_path: PathBuf,
	server_end: &mut zx::Channel,
	) -> Result<(), ModelError> {
	let server_end = channel::take_channel(server_end);
	let server_end = ServerEnd::<SystemControllerMarker>::new(server_end);
	let stream: SystemControllerRequestStream =
	server_end.into_stream().map_err(ModelError::stream_creation_error)?;
	fasync::Task::spawn(async move {
	let result = self.open_async(stream).await;
	if let Err(e) = result {
	warn!("SystemController.open failed: {}", e);
	}
	})
	.detach();

	Ok(())
	}
	}

	#[cfg(test)]
	mod tests {
	use {
	super::*,
	crate::model::{
	binding::Binder,
	hooks::{EventType, Hook, HooksRegistration},
	realm::BindReason,
	testing::test_helpers::{
	component_decl_with_test_runner, ActionsTest, ComponentDeclBuilder, ComponentInfo,
	},
	},
	async_trait::async_trait,
	fidl::endpoints,
	fidl_fuchsia_sys2 as fsys,
	moniker::AbsoluteMoniker,
	std::{boxed::Box, convert::TryFrom, sync::Arc, time::Duration},
	};

	/// Use SystemController to shut down a system whose root has the child `a`
	/// and `a` has descendents as shown in the diagram below.
	/// a
	/// \
	/// b
	/// / \
	/// c d
	#[fuchsia_async::run_singlethreaded(test)]
	async fn test_system_controller() {
	// Configure and start realm
	let components = vec![
	("root", ComponentDeclBuilder::new().add_lazy_child("a").build()),
	("a", ComponentDeclBuilder::new().add_eager_child("b").build()),
	("b", ComponentDeclBuilder::new().add_eager_child("c").add_eager_child("d").build()),
	("c", component_decl_with_test_runner()),
	("d", component_decl_with_test_runner()),
	];
	let test = ActionsTest::new("root", components, None).await;
	let realm_a = test.look_up(vec!["a:0"].into()).await;
	let realm_b = test.look_up(vec!["a:0", "b:0"].into()).await;
	let realm_c = test.look_up(vec!["a:0", "b:0", "c:0"].into()).await;
	let realm_d = test.look_up(vec!["a:0", "b:0", "d:0"].into()).await;
	test.model
	.bind(&realm_a.abs_moniker, &BindReason::BindChild { parent: AbsoluteMoniker::root() })
	.await
	.expect("could not bind to a");

	// Wire up connections to SystemController
	let sys_controller = Box::new(SystemControllerCapabilityProvider::new(
	Arc::downgrade(&test.model),
	// allow simulated shutdown to take up to 30 days
	Duration::from_secs(60 * 60 * 24 * 30),
	));
	let (client_channel, server_channel) =
	endpoints::create_endpoints::<fsys::SystemControllerMarker>()
	.expect("failed creating channel endpoints");
	let mut server_channel = server_channel.into_channel();
	sys_controller
	.open(0, 0, PathBuf::new(), &mut server_channel)
	.await
	.expect("failed to open capability");
	let controller_proxy =
	client_channel.into_proxy().expect("failed converting endpoint into proxy");

	let root_realm_info = ComponentInfo::new(test.model.root_realm.clone()).await;
	let realm_a_info = ComponentInfo::new(realm_a.clone()).await;
	let realm_b_info = ComponentInfo::new(realm_b.clone()).await;
	let realm_c_info = ComponentInfo::new(realm_c.clone()).await;
	let realm_d_info = ComponentInfo::new(realm_d.clone()).await;

	// Check that the root realm is still here
	root_realm_info.check_not_shut_down(&test.runner).await;
	realm_a_info.check_not_shut_down(&test.runner).await;
	realm_b_info.check_not_shut_down(&test.runner).await;
	realm_c_info.check_not_shut_down(&test.runner).await;
	realm_d_info.check_not_shut_down(&test.runner).await;

	// Ask the SystemController to shut down the system and wait to be
	// notified that the room realm stopped.
	let completion = test.builtin_environment.wait_for_root_realm_stop();
	controller_proxy.shutdown().await.expect("shutdown request failed");
	completion.await;

	// Check state bits to confirm root realm looks shut down
	root_realm_info.check_is_shut_down(&test.runner).await;
	realm_a_info.check_is_shut_down(&test.runner).await;
	realm_b_info.check_is_shut_down(&test.runner).await;
	realm_c_info.check_is_shut_down(&test.runner).await;
	realm_d_info.check_is_shut_down(&test.runner).await;
	}

	#[test]
	#[should_panic(expected = "Component manager did not complete shutdown in allowed time.")]
	fn test_timeout() {
	const TIMEOUT_SECONDS: i64 = 6;
	const EVENT_PAUSE_SECONDS: i64 = TIMEOUT_SECONDS + 1;
	struct StopHook;
	#[async_trait]
	impl Hook for StopHook {
	async fn on(self: Arc<Self>, _event: &Event) -> Result<(), ModelError> {
	fasync::Timer::new(fasync::Time::after(zx::Duration::from_seconds(
	EVENT_PAUSE_SECONDS.into(),
	)))
	.await;
	Ok(())
	}
	}

	let mut exec = fasync::Executor::new_with_fake_time().unwrap();
	let mut test_logic = Box::pin(async {
	// Configure and start realm
	let components = vec![
	("root", ComponentDeclBuilder::new().add_eager_child("a").build()),
	("a", ComponentDeclBuilder::new().build()),
	];

	let s = StopHook {};
	let s_hook: Arc<dyn Hook> = Arc::new(s);
	let hooks_reg = HooksRegistration::new(
	"stop hook",
	vec![EventType::Stopped],
	Arc::downgrade(&s_hook),
	);

	let test = ActionsTest::new_with_hooks("root", components, None, vec![hooks_reg]).await;
	let realm_a = test.look_up(vec!["a:0"].into()).await;
	test.model
	.bind(
	&realm_a.abs_moniker,
	&BindReason::BindChild { parent: AbsoluteMoniker::root() },
	)
	.await
	.expect("could not bind to a");

	// Wire up connections to SystemController
	let sys_controller = Box::new(SystemControllerCapabilityProvider::new(
	Arc::downgrade(&test.model),
	// require shutdown in a second
	Duration::from_secs(u64::try_from(TIMEOUT_SECONDS).unwrap()),
	));
	let (client_channel, server_channel) =
	endpoints::create_endpoints::<fsys::SystemControllerMarker>()
	.expect("failed creating channel endpoints");
	let mut server_channel = server_channel.into_channel();
	sys_controller
	.open(0, 0, PathBuf::new(), &mut server_channel)
	.await
	.expect("failed to open capability");
	let controller_proxy =
	client_channel.into_proxy().expect("failed converting endpoint into proxy");

	let root_realm_info = ComponentInfo::new(test.model.root_realm.clone()).await;
	let realm_a_info = ComponentInfo::new(realm_a.clone()).await;

	// Check that the root realm is still here
	root_realm_info.check_not_shut_down(&test.runner).await;
	realm_a_info.check_not_shut_down(&test.runner).await;

	// Ask the SystemController to shut down the system and wait to be
	// notified that the room realm stopped.
	let _completion = test.builtin_environment.wait_for_root_realm_stop();
	controller_proxy.shutdown().await.expect("shutdown request failed");
	});

	assert_eq!(std::task::Poll::Pending, exec.run_until_stalled(&mut test_logic));

	let new_time = fasync::Time::from_nanos(
	exec.now().into_nanos() + zx::Duration::from_seconds(TIMEOUT_SECONDS).into_nanos(),
	);

	exec.set_fake_time(new_time);
	exec.wake_expired_timers();

	assert_eq!(std::task::Poll::Pending, exec.run_until_stalled(&mut test_logic));
	}
	}