qubit_task/service/task_execution_service.rs
1/*******************************************************************************
2 *
3 * Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10use std::{
11 collections::HashMap,
12 panic::{
13 AssertUnwindSafe,
14 catch_unwind,
15 resume_unwind,
16 },
17 sync::{
18 Arc,
19 Condvar,
20 Mutex,
21 MutexGuard,
22 },
23};
24
25use qubit_function::{
26 Callable,
27 Runnable,
28};
29
30use qubit_executor::TaskHandle;
31use qubit_executor::service::{
32 ExecutorService,
33 StopReport,
34};
35use qubit_executor::task::spi::{
36 TaskEndpointPair,
37 TaskSlot,
38};
39use qubit_thread_pool::{
40 ExecutorServiceBuilderError,
41 PoolJob,
42 ThreadPool,
43};
44
45use super::{
46 task_execution_service_builder::TaskExecutionServiceBuilder,
47 task_execution_service_error::TaskExecutionServiceError,
48 task_execution_stats::TaskExecutionStats,
49 task_id::TaskId,
50 task_status::TaskStatus,
51};
52
53/// Managed task execution service built on [`ThreadPool`].
54///
55/// Assigns a stable business [`TaskId`] per task and tracks service-level status
56/// (submitted, running, succeeded, failed, cancelled, panicked). The typed task
57/// outcome is still retrieved through [`TaskHandle`].
58///
59/// # Responsibilities
60///
61/// - **Registry**: The same [`TaskId`] cannot be submitted again while a record
62/// for it exists; a duplicate returns [`TaskExecutionServiceError::DuplicateTask`].
63/// Use this when you need lookup by ID, optional pre-start cancellation, or
64/// long-lived task bookkeeping.
65/// - **Thread pool**: Owns a [`ThreadPool`] for queuing and worker threads; queue
66/// internals are not exposed. Configure the pool via [`TaskExecutionServiceBuilder`]
67/// or [`Self::builder`].
68/// - **Submission semantics**: [`Self::submit`] / [`Self::submit_callable`] returning
69/// `Ok(handle)` means only that the **service accepted** the task—not that it
70/// started or succeeded. Observe the final result with [`TaskHandle::get`] or by
71/// awaiting the handle’s [`Future`](std::future::Future) implementation.
72///
73/// # Suspend
74///
75/// [`Self::suspend`] rejects **new** submissions ([`TaskExecutionServiceError::Suspended`]).
76/// Tasks already queued or running are unaffected. [`Self::resume`] re-enables submission.
77///
78/// # Cancel
79///
80/// [`Self::cancel`] may succeed only **before** the task starts running; once running,
81/// cancellation behavior follows [`TaskHandle`] and the internal completion protocol.
82///
83/// # Shutdown
84///
85/// [`Self::shutdown`] and [`Self::stop`] delegate to the backing pool.
86/// [`Self::wait_termination`] blocks the current thread until all accepted work
87/// has completed, failed, panicked, or been cancelled.
88///
89/// # Example: submit, inspect status, wait for idle, shutdown
90///
91/// ```
92/// use std::error::Error;
93/// use qubit_task::service::{TaskExecutionService, TaskId, TaskStatus};
94///
95/// fn main() -> Result<(), Box<dyn Error>> {
96/// let service = TaskExecutionService::new()?;
97/// let id: TaskId = 1001;
98///
99/// let handle = service.submit(id, || Ok::<(), ()>(()))?;
100/// handle.get().unwrap();
101///
102/// assert_eq!(service.status(id), Some(TaskStatus::Succeeded));
103///
104/// service.await_idle();
105/// service.shutdown();
106/// Ok(())
107/// }
108/// ```
109///
110pub struct TaskExecutionService {
111 pool: ThreadPool,
112 state: Arc<TaskExecutionServiceState>,
113}
114
115impl TaskExecutionService {
116 /// Creates a service using the default [`super::ThreadPoolBuilder`] settings (worker
117 /// counts, queue, and other defaults match [`ThreadPool::builder`]).
118 ///
119 /// # Example
120 ///
121 /// ```
122 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
123 ///
124 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
125 /// let _service = TaskExecutionService::new()?;
126 /// Ok(())
127 /// }
128 /// ```
129 ///
130 /// # Returns
131 ///
132 /// `Ok(Self)` on success, or [`ExecutorServiceBuilderError`] if the pool cannot be built.
133 pub fn new() -> Result<Self, ExecutorServiceBuilderError> {
134 Self::builder().build()
135 }
136
137 /// Returns a [`TaskExecutionServiceBuilder`] so you can tune the backing pool
138 /// before [`TaskExecutionServiceBuilder::build`] (for example
139 /// [`super::ThreadPoolBuilder::pool_size`],
140 /// [`super::ThreadPoolBuilder::queue_capacity`]).
141 ///
142 /// # Example
143 ///
144 /// ```
145 /// use qubit_task::service::{
146 /// TaskExecutionService, ThreadPoolBuilder, ExecutorServiceBuilderError,
147 /// };
148 ///
149 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
150 /// let _service = TaskExecutionService::builder()
151 /// .thread_pool(ThreadPoolBuilder::default().pool_size(8))
152 /// .build()?;
153 /// Ok(())
154 /// }
155 /// ```
156 ///
157 /// # Returns
158 ///
159 /// A builder holding the default [`super::ThreadPoolBuilder`].
160 #[inline]
161 pub fn builder() -> TaskExecutionServiceBuilder {
162 TaskExecutionServiceBuilder::default()
163 }
164
165 /// Builds a service from an already constructed pool.
166 pub(crate) fn from_thread_pool(pool: ThreadPool) -> Self {
167 Self {
168 pool,
169 state: Arc::new(TaskExecutionServiceState::default()),
170 }
171 }
172
173 /// Submits a runnable task with a business task ID.
174 ///
175 /// # Example
176 ///
177 /// ```
178 /// use std::error::Error;
179 /// use qubit_task::service::TaskExecutionService;
180 ///
181 /// fn main() -> Result<(), Box<dyn Error>> {
182 /// let service = TaskExecutionService::new()?;
183 /// let handle = service.submit(42_u64, || Ok::<(), ()>(()))?;
184 /// handle.get().unwrap();
185 /// Ok(())
186 /// }
187 /// ```
188 ///
189 /// # Parameters
190 ///
191 /// * `task_id` - Stable business ID for registry operations.
192 /// * `task` - Runnable to execute.
193 ///
194 /// # Returns
195 ///
196 /// `Ok(handle)` if the service accepts the task. This only means
197 /// acceptance; task success is observed through the handle. Returns
198 /// [`TaskExecutionServiceError`] when the ID is duplicated, the service is
199 /// suspended, or the backing pool rejects the task.
200 #[inline]
201 pub fn submit<T, E>(&self, task_id: TaskId, mut task: T) -> Result<TaskHandle<(), E>, TaskExecutionServiceError>
202 where
203 T: Runnable<E> + Send + 'static,
204 E: Send + 'static,
205 {
206 self.submit_callable(task_id, move || task.run())
207 }
208
209 /// Submits a callable task with a business task ID.
210 ///
211 /// # Example
212 ///
213 /// ```
214 /// use std::error::Error;
215 /// use qubit_task::service::{TaskExecutionService, TaskId};
216 ///
217 /// fn main() -> Result<(), Box<dyn Error>> {
218 /// let service = TaskExecutionService::new()?;
219 /// let id: TaskId = 7;
220 /// let handle = service.submit_callable(id, || Ok::<i32, ()>(21))?;
221 /// assert_eq!(handle.get().unwrap(), 21);
222 /// Ok(())
223 /// }
224 /// ```
225 ///
226 /// # Parameters
227 ///
228 /// * `task_id` - Stable business ID for registry operations.
229 /// * `task` - Callable to execute.
230 ///
231 /// # Returns
232 ///
233 /// `Ok(handle)` if the service accepts the task. The handle reports the
234 /// typed task result while this service records only service-level status.
235 pub fn submit_callable<C, R, E>(
236 &self,
237 task_id: TaskId,
238 task: C,
239 ) -> Result<TaskHandle<R, E>, TaskExecutionServiceError>
240 where
241 C: Callable<R, E> + Send + 'static,
242 R: Send + 'static,
243 E: Send + 'static,
244 {
245 let (handle, slot) = TaskEndpointPair::new().into_parts();
246 let slot = Arc::new(Mutex::new(Some(slot)));
247 let accept_slot = Arc::clone(&slot);
248 let cancel_slot = Arc::clone(&slot);
249 let run_slot = Arc::clone(&slot);
250 let cancel_state = Arc::clone(&self.state);
251 let cancel: Arc<dyn Fn() -> bool + Send + Sync> = Arc::new(move || {
252 let slot = cancel_slot
253 .lock()
254 .expect("task slot lock should not be poisoned")
255 .take();
256 let cancelled = slot.is_some_and(TaskSlot::cancel_unstarted);
257 if cancelled {
258 cancel_state.set_status(task_id, TaskStatus::Cancelled);
259 }
260 cancelled
261 });
262
263 self.state.register(task_id, Arc::clone(&cancel))?;
264
265 let run_state = Arc::clone(&self.state);
266 let cancel_for_job = Arc::clone(&cancel);
267 let job = PoolJob::with_accept(
268 Box::new(move || {
269 if let Some(slot) = accept_slot
270 .lock()
271 .expect("task slot lock should not be poisoned")
272 .as_ref()
273 {
274 slot.accept();
275 }
276 }),
277 Box::new(move || {
278 let slot = run_slot.lock().expect("task slot lock should not be poisoned").take();
279 if let Some(slot) = slot {
280 let task = StatusReportingTask {
281 task_id,
282 task,
283 state: run_state,
284 };
285 if !slot.run(task) {
286 cancel_for_job();
287 }
288 }
289 }),
290 Box::new(move || {
291 cancel();
292 }),
293 );
294
295 if let Err(error) = self.pool.submit_job(job) {
296 self.state.remove(task_id);
297 return Err(error.into());
298 }
299 Ok(handle)
300 }
301
302 /// Attempts to cancel a submitted task by ID.
303 ///
304 /// Cancellation succeeds only before the task starts running.
305 ///
306 /// # Example
307 ///
308 /// ```
309 /// use std::error::Error;
310 /// use qubit_task::service::{TaskExecutionService, TaskId};
311 ///
312 /// fn main() -> Result<(), Box<dyn Error>> {
313 /// let service = TaskExecutionService::new()?;
314 /// let id: TaskId = 1;
315 /// let handle = service.submit(id, || Ok::<(), ()>(()))?;
316 /// // `true` only if cancelled before a worker starts the task (race with the pool).
317 /// let _cancelled = service.cancel(id);
318 /// match handle.get() {
319 /// Ok(()) => {}
320 /// Err(e) if e.is_cancelled() => {}
321 /// Err(e) => panic!("unexpected task outcome: {e:?}"),
322 /// }
323 /// Ok(())
324 /// }
325 /// ```
326 ///
327 /// # Parameters
328 ///
329 /// * `task_id` - ID of the task to cancel.
330 ///
331 /// # Returns
332 ///
333 /// `true` if the task was cancelled before start, or `false` if no active
334 /// task with this ID can be cancelled.
335 pub fn cancel(&self, task_id: TaskId) -> bool {
336 let cancel = self.state.cancel_callback(task_id);
337 cancel.is_some_and(|cancel| cancel())
338 }
339
340 /// Returns the current status of a task.
341 ///
342 /// # Example
343 ///
344 /// ```
345 /// use std::error::Error;
346 /// use qubit_task::service::{TaskExecutionService, TaskId, TaskStatus};
347 ///
348 /// fn main() -> Result<(), Box<dyn Error>> {
349 /// let service = TaskExecutionService::new()?;
350 /// let id: TaskId = 10;
351 /// let handle = service.submit(id, || Ok::<(), ()>(()))?;
352 /// handle.get().unwrap();
353 /// assert_eq!(service.status(id), Some(TaskStatus::Succeeded));
354 /// Ok(())
355 /// }
356 /// ```
357 ///
358 /// # Parameters
359 ///
360 /// * `task_id` - ID of the task to inspect.
361 ///
362 /// # Returns
363 ///
364 /// `Some(status)` if the service retains a record for this ID, or `None`
365 /// if the ID is unknown.
366 #[inline]
367 pub fn status(&self, task_id: TaskId) -> Option<TaskStatus> {
368 self.state.status(task_id)
369 }
370
371 /// Returns registry-derived task statistics.
372 ///
373 /// # Example
374 ///
375 /// ```
376 /// use std::error::Error;
377 /// use qubit_task::service::TaskExecutionService;
378 ///
379 /// fn main() -> Result<(), Box<dyn Error>> {
380 /// let service = TaskExecutionService::new()?;
381 /// let handle = service.submit(1_u64, || Ok::<(), ()>(()))?;
382 /// handle.get().unwrap();
383 /// let snapshot = service.stats();
384 /// assert!(snapshot.total >= 1);
385 /// Ok(())
386 /// }
387 /// ```
388 ///
389 /// # Returns
390 ///
391 /// A snapshot of retained task records grouped by status.
392 #[inline]
393 pub fn stats(&self) -> TaskExecutionStats {
394 self.state.stats()
395 }
396
397 /// Suspends new submissions.
398 ///
399 /// Existing submitted and running tasks continue normally.
400 ///
401 /// # Example
402 ///
403 /// ```
404 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
405 ///
406 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
407 /// let service = TaskExecutionService::new()?;
408 /// service.suspend();
409 /// assert!(service.is_suspended());
410 /// service.resume();
411 /// assert!(!service.is_suspended());
412 /// Ok(())
413 /// }
414 /// ```
415 #[inline]
416 pub fn suspend(&self) {
417 self.state.set_suspended(true);
418 }
419
420 /// Resumes accepting new submissions.
421 #[inline]
422 pub fn resume(&self) {
423 self.state.set_suspended(false);
424 }
425
426 /// Returns whether the service is suspended.
427 ///
428 /// # Returns
429 ///
430 /// `true` if new submissions are rejected before reaching the pool.
431 #[inline]
432 pub fn is_suspended(&self) -> bool {
433 self.state.is_suspended()
434 }
435
436 /// Waits for the active task snapshot observed at call time to finish.
437 ///
438 /// Tasks submitted after this method starts are not part of the waited
439 /// snapshot. This method blocks the current thread.
440 ///
441 /// # Example
442 ///
443 /// ```
444 /// use std::error::Error;
445 /// use qubit_task::service::{TaskExecutionService, TaskId};
446 ///
447 /// fn main() -> Result<(), Box<dyn Error>> {
448 /// let service = TaskExecutionService::new()?;
449 /// let a: TaskId = 1;
450 /// let b: TaskId = 2;
451 /// let h1 = service.submit(a, || Ok::<(), ()>(()))?;
452 /// let h2 = service.submit(b, || Ok::<(), ()>(()))?;
453 /// service.await_in_flight_tasks_completion();
454 /// h1.get().unwrap();
455 /// h2.get().unwrap();
456 /// Ok(())
457 /// }
458 /// ```
459 pub fn await_in_flight_tasks_completion(&self) {
460 self.state.await_in_flight_tasks_completion();
461 }
462
463 /// Waits until the service registry has no submitted or running tasks.
464 ///
465 /// This method blocks the current thread and observes real-time idleness.
466 ///
467 /// # Example
468 ///
469 /// ```
470 /// use std::error::Error;
471 /// use qubit_task::service::{TaskExecutionService, TaskId};
472 ///
473 /// fn main() -> Result<(), Box<dyn Error>> {
474 /// let service = TaskExecutionService::new()?;
475 /// let id: TaskId = 1;
476 /// let handle = service.submit(id, || Ok::<(), ()>(()))?;
477 /// handle.get().unwrap();
478 /// service.await_idle();
479 /// Ok(())
480 /// }
481 /// ```
482 pub fn await_idle(&self) {
483 self.state.await_idle();
484 }
485
486 /// Initiates graceful shutdown of the backing pool.
487 ///
488 /// # Example
489 ///
490 /// ```
491 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
492 ///
493 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
494 /// let service = TaskExecutionService::new()?;
495 /// service.shutdown();
496 /// assert!(service.is_not_running());
497 /// Ok(())
498 /// }
499 /// ```
500 #[inline]
501 pub fn shutdown(&self) {
502 self.pool.shutdown();
503 }
504
505 /// Initiates immediate stop of the backing pool.
506 ///
507 /// # Example
508 ///
509 /// ```
510 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
511 ///
512 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
513 /// let service = TaskExecutionService::new()?;
514 /// let _report = service.stop();
515 /// Ok(())
516 /// }
517 /// ```
518 ///
519 /// # Returns
520 ///
521 /// A count-based report from the backing pool.
522 #[inline]
523 pub fn stop(&self) -> StopReport {
524 self.pool.stop()
525 }
526
527 /// Returns whether the backing pool no longer accepts new work.
528 #[inline]
529 pub fn is_not_running(&self) -> bool {
530 self.pool.is_not_running()
531 }
532
533 /// Returns whether the backing pool has terminated.
534 #[inline]
535 pub fn is_terminated(&self) -> bool {
536 self.pool.is_terminated()
537 }
538
539 /// Blocks until the backing pool has terminated.
540 ///
541 /// # Example
542 ///
543 /// ```
544 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
545 ///
546 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
547 /// let service = TaskExecutionService::new()?;
548 /// service.shutdown();
549 /// service.wait_termination();
550 /// assert!(service.is_terminated());
551 /// Ok(())
552 /// }
553 /// ```
554 ///
555 /// # Returns
556 ///
557 /// Returns after shutdown and worker exit.
558 #[inline]
559 pub fn wait_termination(&self) {
560 self.pool.wait_termination();
561 }
562
563 /// Returns the backing thread pool.
564 ///
565 /// # Example
566 ///
567 /// ```
568 /// use qubit_task::service::{TaskExecutionService, ExecutorServiceBuilderError};
569 ///
570 /// fn main() -> Result<(), ExecutorServiceBuilderError> {
571 /// let service = TaskExecutionService::new()?;
572 /// let pool = service.thread_pool();
573 /// assert!(pool.maximum_pool_size() > 0);
574 /// Ok(())
575 /// }
576 /// ```
577 ///
578 /// # Returns
579 ///
580 /// A shared reference for low-level inspection such as pool statistics.
581 #[inline]
582 pub fn thread_pool(&self) -> &ThreadPool {
583 &self.pool
584 }
585}
586
587/// Shared state for [`TaskExecutionService`].
588#[derive(Default)]
589struct TaskExecutionServiceState {
590 inner: Mutex<TaskExecutionServiceInner>,
591 idle: Condvar,
592}
593
594impl TaskExecutionServiceState {
595 /// Acquires service state.
596 fn lock_inner(&self) -> MutexGuard<'_, TaskExecutionServiceInner> {
597 self.inner
598 .lock()
599 .expect("task execution service state lock should not be poisoned")
600 }
601
602 /// Registers a submitted task.
603 fn register(
604 &self,
605 task_id: TaskId,
606 cancel: Arc<dyn Fn() -> bool + Send + Sync>,
607 ) -> Result<(), TaskExecutionServiceError> {
608 let mut inner = self.lock_inner();
609 if inner.suspended {
610 return Err(TaskExecutionServiceError::Suspended);
611 }
612 if inner.tasks.contains_key(&task_id) {
613 return Err(TaskExecutionServiceError::DuplicateTask(task_id));
614 }
615 inner.tasks.insert(
616 task_id,
617 TaskRecord {
618 status: TaskStatus::Submitted,
619 cancel,
620 },
621 );
622 Ok(())
623 }
624
625 /// Removes a task record.
626 fn remove(&self, task_id: TaskId) {
627 let mut inner = self.lock_inner();
628 inner.tasks.remove(&task_id);
629 self.idle.notify_all();
630 }
631
632 /// Gets a task status.
633 fn status(&self, task_id: TaskId) -> Option<TaskStatus> {
634 self.lock_inner().tasks.get(&task_id).map(|record| record.status)
635 }
636
637 /// Gets a task cancel callback if the task is active.
638 fn cancel_callback(&self, task_id: TaskId) -> Option<Arc<dyn Fn() -> bool + Send + Sync>> {
639 let inner = self.lock_inner();
640 let record = inner.tasks.get(&task_id)?;
641 record.status.is_active().then(|| Arc::clone(&record.cancel))
642 }
643
644 /// Updates a task status.
645 fn set_status(&self, task_id: TaskId, status: TaskStatus) {
646 let mut inner = self.lock_inner();
647 let record = inner
648 .tasks
649 .get_mut(&task_id)
650 .expect("task status can only be updated for a registered task");
651 record.status = status;
652 self.idle.notify_all();
653 }
654
655 /// Updates suspended flag.
656 fn set_suspended(&self, suspended: bool) {
657 self.lock_inner().suspended = suspended;
658 }
659
660 /// Returns whether new submissions are suspended.
661 fn is_suspended(&self) -> bool {
662 self.lock_inner().suspended
663 }
664
665 /// Returns task statistics.
666 fn stats(&self) -> TaskExecutionStats {
667 let inner = self.lock_inner();
668 let mut stats = TaskExecutionStats::default();
669 for record in inner.tasks.values() {
670 stats.add_status(record.status);
671 }
672 stats
673 }
674
675 /// Waits for active task IDs observed at call time.
676 fn await_in_flight_tasks_completion(&self) {
677 let mut inner = self.lock_inner();
678 let task_ids = inner
679 .tasks
680 .iter()
681 .filter_map(|(&task_id, record)| record.status.is_active().then_some(task_id))
682 .collect::<Vec<_>>();
683 while task_ids.iter().any(|task_id| inner.task_is_active(*task_id)) {
684 inner = self.wait_for_idle_notification(inner);
685 }
686 }
687
688 /// Waits until no retained task record is active.
689 fn await_idle(&self) {
690 let mut inner = self.lock_inner();
691 while inner.has_active_tasks() {
692 inner = self.wait_for_idle_notification(inner);
693 }
694 }
695
696 /// Waits for a state transition notification.
697 fn wait_for_idle_notification<'a>(
698 &self,
699 inner: MutexGuard<'a, TaskExecutionServiceInner>,
700 ) -> MutexGuard<'a, TaskExecutionServiceInner> {
701 self.idle
702 .wait(inner)
703 .expect("task execution service state lock should not be poisoned")
704 }
705}
706
707/// Mutable service state protected by a mutex.
708#[derive(Default)]
709struct TaskExecutionServiceInner {
710 suspended: bool,
711 tasks: HashMap<TaskId, TaskRecord>,
712}
713
714impl TaskExecutionServiceInner {
715 /// Returns whether a retained task ID is still active.
716 fn task_is_active(&self, task_id: TaskId) -> bool {
717 self.tasks.get(&task_id).is_some_and(|record| record.status.is_active())
718 }
719
720 /// Returns whether any retained task is still active.
721 fn has_active_tasks(&self) -> bool {
722 self.tasks.values().any(|record| record.status.is_active())
723 }
724}
725
726/// Registry record for one managed task.
727struct TaskRecord {
728 status: TaskStatus,
729 cancel: Arc<dyn Fn() -> bool + Send + Sync>,
730}
731
732/// Callable wrapper that keeps service-level status aligned with task outcome.
733struct StatusReportingTask<C> {
734 /// Stable business task ID.
735 task_id: TaskId,
736 /// User task to execute.
737 task: C,
738 /// Shared service registry.
739 state: Arc<TaskExecutionServiceState>,
740}
741
742impl<C, R, E> Callable<R, E> for StatusReportingTask<C>
743where
744 C: Callable<R, E>,
745{
746 /// Runs the user task and records the corresponding service-level status.
747 fn call(&mut self) -> Result<R, E> {
748 self.state.set_status(self.task_id, TaskStatus::Running);
749 match catch_unwind(AssertUnwindSafe(|| self.task.call())) {
750 Ok(Ok(value)) => {
751 self.state.set_status(self.task_id, TaskStatus::Succeeded);
752 Ok(value)
753 }
754 Ok(Err(error)) => {
755 self.state.set_status(self.task_id, TaskStatus::Failed);
756 Err(error)
757 }
758 Err(payload) => {
759 self.state.set_status(self.task_id, TaskStatus::Panicked);
760 resume_unwind(payload);
761 }
762 }
763 }
764}