Lines Matching full:task
74 pthread_mutex_t lock; /* part of task wake_idle */
75 struct lws_threadpool_task *task; member
125 __lws_threadpool_task_dump(struct lws_threadpool_task *task, char *buf, int len) in __lws_threadpool_task_dump() argument
131 if (!task->acquired) { in __lws_threadpool_task_dump()
133 "task: %s, QUEUED queued: %dms", in __lws_threadpool_task_dump()
134 task->name, ms_delta(now, task->created)); in __lws_threadpool_task_dump()
139 if (task->acc_running) in __lws_threadpool_task_dump()
140 runms = (int)task->acc_running; in __lws_threadpool_task_dump()
142 if (task->acc_syncing) in __lws_threadpool_task_dump()
143 syncms = (int)task->acc_syncing; in __lws_threadpool_task_dump()
145 if (!task->done) { in __lws_threadpool_task_dump()
147 "task: %s, ONGOING state %d (%dms) alive: %dms " in __lws_threadpool_task_dump()
149 "run: %d%%, sync: %d%%)", task->name, task->status, in __lws_threadpool_task_dump()
150 ms_delta(now, task->entered_state), in __lws_threadpool_task_dump()
151 ms_delta(now, task->created), in __lws_threadpool_task_dump()
152 ms_delta(task->acquired, task->created), in __lws_threadpool_task_dump()
153 ms_delta(now, task->acquired), in __lws_threadpool_task_dump()
154 pc_delta(now, task->acquired, runms), in __lws_threadpool_task_dump()
155 pc_delta(now, task->acquired, syncms)); in __lws_threadpool_task_dump()
161 "task: %s, DONE state %d lived: %dms " in __lws_threadpool_task_dump()
163 "ran: %d%%, synced: %d%%)", task->name, task->status, in __lws_threadpool_task_dump()
164 ms_delta(task->done, task->created), in __lws_threadpool_task_dump()
165 ms_delta(task->acquired, task->created), in __lws_threadpool_task_dump()
166 ms_delta(task->done, task->acquired), in __lws_threadpool_task_dump()
167 pc_delta(task->done, task->acquired, runms), in __lws_threadpool_task_dump()
168 pc_delta(task->done, task->acquired, syncms)); in __lws_threadpool_task_dump()
189 struct lws_threadpool_task *task = *c; in lws_threadpool_dump()
190 __lws_threadpool_task_dump(task, buf, sizeof(buf)); in lws_threadpool_dump()
204 struct lws_threadpool_task *task = pool->task; in lws_threadpool_dump()
206 if (task) { in lws_threadpool_dump()
207 __lws_threadpool_task_dump(task, buf, sizeof(buf)); in lws_threadpool_dump()
220 struct lws_threadpool_task *task = *c; in lws_threadpool_dump()
221 __lws_threadpool_task_dump(task, buf, sizeof(buf)); in lws_threadpool_dump()
237 state_transition(struct lws_threadpool_task *task, in state_transition() argument
240 task->entered_state = lws_now_usecs(); in state_transition()
241 task->status = status; in state_transition()
245 task_to_wsi(struct lws_threadpool_task *task) in task_to_wsi() argument
248 if (task->args.ss) in task_to_wsi()
249 return task->args.ss->wsi; in task_to_wsi()
251 return task->args.wsi; in task_to_wsi()
255 lws_threadpool_task_cleanup_destroy(struct lws_threadpool_task *task) in lws_threadpool_task_cleanup_destroy() argument
257 if (task->args.cleanup) in lws_threadpool_task_cleanup_destroy()
258 task->args.cleanup(task_to_wsi(task), task->args.user); in lws_threadpool_task_cleanup_destroy()
260 lws_dll2_remove(&task->list); in lws_threadpool_task_cleanup_destroy()
262 lwsl_thread("%s: tp %p: cleaned finished task for %s\n", in lws_threadpool_task_cleanup_destroy()
263 __func__, task->tp, lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_task_cleanup_destroy()
265 lws_free(task); in lws_threadpool_task_cleanup_destroy()
269 __lws_threadpool_reap(struct lws_threadpool_task *task) in __lws_threadpool_reap() argument
272 struct lws_threadpool *tp = task->tp; in __lws_threadpool_reap()
274 /* remove the task from the done queue */ in __lws_threadpool_reap()
280 if ((*c) == task) { in __lws_threadpool_reap()
286 lwsl_thread("%s: tp %s: reaped task %s\n", __func__, in __lws_threadpool_reap()
287 tp->name, lws_wsi_tag(task_to_wsi(task))); in __lws_threadpool_reap()
295 lwsl_err("%s: task %p not in done queue\n", __func__, task); in __lws_threadpool_reap()
298 * safe to assume there's a task to destroy in __lws_threadpool_reap()
303 lwsl_err("%s: task->tp NULL already\n", __func__); in __lws_threadpool_reap()
305 /* call the task's cleanup and delete the task itself */ in __lws_threadpool_reap()
307 lws_threadpool_task_cleanup_destroy(task); in __lws_threadpool_reap()
319 struct lws_threadpool_task **c, *task = NULL; in lws_threadpool_tsi_context() local
334 task = pool->task; in lws_threadpool_tsi_context()
335 if (!task) in lws_threadpool_tsi_context()
338 wsi = task_to_wsi(task); in lws_threadpool_tsi_context()
340 (!task->wanted_writeable_cb && in lws_threadpool_tsi_context()
341 task->status != LWS_TP_STATUS_SYNCING)) in lws_threadpool_tsi_context()
344 task->wanted_writeable_cb = 0; in lws_threadpool_tsi_context()
361 task = *c; in lws_threadpool_tsi_context()
362 wsi = task_to_wsi(task); in lws_threadpool_tsi_context()
365 (task->wanted_writeable_cb || in lws_threadpool_tsi_context()
366 task->status == LWS_TP_STATUS_SYNCING)) { in lws_threadpool_tsi_context()
368 task->wanted_writeable_cb = 0; in lws_threadpool_tsi_context()
380 c = &task->task_queue_next; in lws_threadpool_tsi_context()
393 struct lws_threadpool_task *task) in lws_threadpool_worker_sync() argument
401 lwsl_debug("%s: %p: LWS_TP_RETURN_SYNC in\n", __func__, task); in lws_threadpool_worker_sync()
404 lwsl_info("%s: %s: task %p (%s): syncing with %s\n", __func__, in lws_threadpool_worker_sync()
405 pool->tp->name, task, task->name, lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_worker_sync()
407 temp = task->status; in lws_threadpool_worker_sync()
408 state_transition(task, LWS_TP_STATUS_SYNCING); in lws_threadpool_worker_sync()
410 wsi = task_to_wsi(task); in lws_threadpool_worker_sync()
413 * if the wsi is no longer attached to this task, there is in lws_threadpool_worker_sync()
416 * the task it can't continue usefully by stopping it. in lws_threadpool_worker_sync()
420 lwsl_thread("%s: %s: task %p (%s): No longer bound to any " in lws_threadpool_worker_sync()
422 task, task->name); in lws_threadpool_worker_sync()
424 state_transition(task, LWS_TP_STATUS_STOPPING); in lws_threadpool_worker_sync()
433 * If it is exceeded, we will stop the task. in lws_threadpool_worker_sync()
438 task->wanted_writeable_cb = 1; in lws_threadpool_worker_sync()
453 * times out and we stop the task. in lws_threadpool_worker_sync()
456 if (pthread_cond_timedwait(&task->wake_idle, &pool->lock, in lws_threadpool_worker_sync()
458 task->late_sync_retries++; in lws_threadpool_worker_sync()
460 lwsl_err("%s: %s: task %p (%s): SYNC timed out " in lws_threadpool_worker_sync()
462 __func__, pool->tp->name, task, in lws_threadpool_worker_sync()
463 task->name, lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_worker_sync()
466 lws_threadpool_dequeue_task(task); in lws_threadpool_worker_sync()
467 return 1; /* destroyed task */ in lws_threadpool_worker_sync()
475 if (task->status == LWS_TP_STATUS_SYNCING) in lws_threadpool_worker_sync()
476 state_transition(task, temp); in lws_threadpool_worker_sync()
478 lwsl_debug("%s: %p: LWS_TP_RETURN_SYNC out\n", __func__, task); in lws_threadpool_worker_sync()
493 struct lws_threadpool_task **c, **c2, *task; in lws_threadpool_worker() local
500 /* we have no running task... wait and get one from the queue */ in lws_threadpool_worker()
505 * if there's no task already waiting in the queue, wait for in lws_threadpool_worker()
518 task = NULL; in lws_threadpool_worker()
519 pool->task = NULL; in lws_threadpool_worker()
527 /* is there a task at the queue tail? */ in lws_threadpool_worker()
529 pool->task = task = *c2; in lws_threadpool_worker()
530 task->acquired = pool->acquired = lws_now_usecs(); in lws_threadpool_worker()
532 *c2 = task->task_queue_next; in lws_threadpool_worker()
533 task->task_queue_next = NULL; in lws_threadpool_worker()
536 state_transition(task, LWS_TP_STATUS_RUNNING); in lws_threadpool_worker()
540 if (!task) { in lws_threadpool_worker()
545 task->wanted_writeable_cb = 0; in lws_threadpool_worker()
547 /* we have acquired a new task */ in lws_threadpool_worker()
549 __lws_threadpool_task_dump(task, buf, sizeof(buf)); in lws_threadpool_worker()
558 * 1) The task can return with LWS_TP_RETURN_CHECKING_IN to in lws_threadpool_worker()
561 * left off if the task is not being "stopped". in lws_threadpool_worker()
566 * 2) The task can return with LWS_TP_RETURN_SYNC to register in lws_threadpool_worker()
573 * 3) The task can return with LWS_TP_RETURN_FINISHED to in lws_threadpool_worker()
576 * 4) The task can return with LWS_TP_RETURN_STOPPED to indicate in lws_threadpool_worker()
584 if (tp->destroying || !task_to_wsi(task)) { in lws_threadpool_worker()
586 state_transition(task, LWS_TP_STATUS_STOPPING); in lws_threadpool_worker()
590 n = (int)task->args.task(task->args.user, task->status); in lws_threadpool_worker()
591 lwsl_debug(" %d, status %d\n", n, task->status); in lws_threadpool_worker()
592 us_accrue(&task->acc_running, then); in lws_threadpool_worker()
594 task->outlive = 1; in lws_threadpool_worker()
600 if (!task_to_wsi(task)) { in lws_threadpool_worker()
601 lwsl_debug("%s: task that wants to " in lws_threadpool_worker()
609 if (lws_threadpool_worker_sync(pool, task)) { in lws_threadpool_worker()
613 us_accrue(&task->acc_syncing, then); in lws_threadpool_worker()
616 state_transition(task, LWS_TP_STATUS_FINISHED); in lws_threadpool_worker()
619 state_transition(task, LWS_TP_STATUS_STOPPED); in lws_threadpool_worker()
622 } while (task->status == LWS_TP_STATUS_RUNNING); in lws_threadpool_worker()
628 if (pool->task->status == LWS_TP_STATUS_STOPPING) in lws_threadpool_worker()
629 state_transition(task, LWS_TP_STATUS_STOPPED); in lws_threadpool_worker()
631 /* move the task to the done queue */ in lws_threadpool_worker()
633 pool->task->task_queue_next = tp->task_done_head; in lws_threadpool_worker()
634 tp->task_done_head = task; in lws_threadpool_worker()
636 pool->task->done = lws_now_usecs(); in lws_threadpool_worker()
638 if (!pool->task->args.wsi && in lws_threadpool_worker()
639 (pool->task->status == LWS_TP_STATUS_STOPPED || in lws_threadpool_worker()
640 pool->task->status == LWS_TP_STATUS_FINISHED)) { in lws_threadpool_worker()
642 __lws_threadpool_task_dump(pool->task, buf, sizeof(buf)); in lws_threadpool_worker()
652 __lws_threadpool_reap(pool->task); in lws_threadpool_worker()
655 __lws_threadpool_task_dump(pool->task, buf, sizeof(buf)); in lws_threadpool_worker()
661 * task status */ in lws_threadpool_worker()
663 if (task_to_wsi(pool->task)) { in lws_threadpool_worker()
664 task->wanted_writeable_cb = 1; in lws_threadpool_worker()
667 lws_get_context(task_to_wsi(pool->task))); in lws_threadpool_worker()
672 pool->task = NULL; in lws_threadpool_worker()
744 struct lws_threadpool_task **c, *task; in lws_threadpool_finish() local
756 task = *c; in lws_threadpool_finish()
757 *c = task->task_queue_next; in lws_threadpool_finish()
758 task->task_queue_next = tp->task_done_head; in lws_threadpool_finish()
759 tp->task_done_head = task; in lws_threadpool_finish()
760 state_transition(task, LWS_TP_STATUS_STOPPED); in lws_threadpool_finish()
763 task->done = lws_now_usecs(); in lws_threadpool_finish()
765 c = &task->task_queue_next; in lws_threadpool_finish()
775 struct lws_threadpool_task *task, *next; in lws_threadpool_destroy() local
812 task = tp->pool_list[n].task; in lws_threadpool_destroy()
822 task = tp->task_done_head; in lws_threadpool_destroy()
823 while (task) { in lws_threadpool_destroy()
824 next = task->task_queue_next; in lws_threadpool_destroy()
825 lws_threadpool_task_cleanup_destroy(task); in lws_threadpool_destroy()
827 task = next; in lws_threadpool_destroy()
841 lws_threadpool_dequeue_task(struct lws_threadpool_task *task) in lws_threadpool_dequeue_task() argument
847 tp = task->tp; in lws_threadpool_dequeue_task()
850 if (task->outlive && !tp->destroying) { in lws_threadpool_dequeue_task()
852 /* disconnect from wsi, and wsi from task */ in lws_threadpool_dequeue_task()
854 lws_dll2_remove(&task->list); in lws_threadpool_dequeue_task()
855 task->args.wsi = NULL; in lws_threadpool_dequeue_task()
857 task->args.ss = NULL; in lws_threadpool_dequeue_task()
870 if ((*c) == task) { in lws_threadpool_dequeue_task()
871 *c = task->task_queue_next; in lws_threadpool_dequeue_task()
872 task->task_queue_next = tp->task_done_head; in lws_threadpool_dequeue_task()
873 tp->task_done_head = task; in lws_threadpool_dequeue_task()
874 state_transition(task, LWS_TP_STATUS_STOPPED); in lws_threadpool_dequeue_task()
877 task->done = lws_now_usecs(); in lws_threadpool_dequeue_task()
879 lwsl_debug("%s: tp %p: removed queued task %s\n", in lws_threadpool_dequeue_task()
880 __func__, tp, lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_dequeue_task()
891 if ((*c) == task) { in lws_threadpool_dequeue_task()
892 *c = task->task_queue_next; in lws_threadpool_dequeue_task()
893 task->task_queue_next = NULL; in lws_threadpool_dequeue_task()
894 lws_threadpool_task_cleanup_destroy(task); in lws_threadpool_dequeue_task()
904 if (!tp->pool_list[n].task || tp->pool_list[n].task != task) in lws_threadpool_dequeue_task()
918 state_transition(task, LWS_TP_STATUS_STOPPING); in lws_threadpool_dequeue_task()
920 /* disconnect from wsi, and wsi from task */ in lws_threadpool_dequeue_task()
922 lws_dll2_remove(&task->list); in lws_threadpool_dequeue_task()
923 task->args.wsi = NULL; in lws_threadpool_dequeue_task()
925 task->args.ss = NULL; in lws_threadpool_dequeue_task()
930 lwsl_debug("%s: tp %p: request stop running task " in lws_threadpool_dequeue_task()
932 lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_dequeue_task()
939 lwsl_notice("%s: tp %p: no task for %s, decoupling\n", in lws_threadpool_dequeue_task()
940 __func__, tp, lws_wsi_tag(task_to_wsi(task))); in lws_threadpool_dequeue_task()
941 lws_dll2_remove(&task->list); in lws_threadpool_dequeue_task()
942 task->args.wsi = NULL; in lws_threadpool_dequeue_task()
944 task->args.ss = NULL; in lws_threadpool_dequeue_task()
957 struct lws_threadpool_task *task; in lws_threadpool_dequeue() local
963 task = lws_container_of(wsi->tp_task_owner.head, in lws_threadpool_dequeue()
966 return lws_threadpool_dequeue_task(task); in lws_threadpool_dequeue()
974 struct lws_threadpool_task *task = NULL; in lws_threadpool_enqueue() local
999 * create the task object in lws_threadpool_enqueue()
1002 task = lws_malloc(sizeof(*task), __func__); in lws_threadpool_enqueue()
1003 if (!task) in lws_threadpool_enqueue()
1006 memset(task, 0, sizeof(*task)); in lws_threadpool_enqueue()
1007 pthread_cond_init(&task->wake_idle, NULL); in lws_threadpool_enqueue()
1008 task->args = *args; in lws_threadpool_enqueue()
1009 task->tp = tp; in lws_threadpool_enqueue()
1010 task->created = lws_now_usecs(); in lws_threadpool_enqueue()
1013 vsnprintf(task->name, sizeof(task->name) - 1, format, ap); in lws_threadpool_enqueue()
1017 * add him on the tp task queue in lws_threadpool_enqueue()
1020 task->task_queue_next = tp->task_queue_head; in lws_threadpool_enqueue()
1021 state_transition(task, LWS_TP_STATUS_QUEUED); in lws_threadpool_enqueue()
1022 tp->task_queue_head = task; in lws_threadpool_enqueue()
1032 lws_dll2_add_tail(&task->list, &args->ss->wsi->tp_task_owner); in lws_threadpool_enqueue()
1035 lws_dll2_add_tail(&task->list, &args->wsi->tp_task_owner); in lws_threadpool_enqueue()
1037 lwsl_thread("%s: tp %s: enqueued task %p (%s) for %s, depth %d\n", in lws_threadpool_enqueue()
1038 __func__, tp->name, task, task->name, in lws_threadpool_enqueue()
1039 lws_wsi_tag(task_to_wsi(task)), tp->queue_depth); in lws_threadpool_enqueue()
1041 /* alert any idle thread there's something new on the task list */ in lws_threadpool_enqueue()
1049 return task; in lws_threadpool_enqueue()
1055 lws_threadpool_task_status(struct lws_threadpool_task *task, void **user) in lws_threadpool_task_status() argument
1058 struct lws_threadpool *tp = task->tp; in lws_threadpool_task_status()
1063 *user = task->args.user; in lws_threadpool_task_status()
1064 status = task->status; in lws_threadpool_task_status()
1071 __lws_threadpool_task_dump(task, buf, sizeof(buf)); in lws_threadpool_task_status()
1074 __lws_threadpool_reap(task); in lws_threadpool_task_status()
1083 lws_threadpool_task_status_noreap(struct lws_threadpool_task *task) in lws_threadpool_task_status_noreap() argument
1085 return task->status; in lws_threadpool_task_status_noreap()
1092 struct lws_threadpool_task *task; in lws_threadpool_task_status_wsi() local
1095 lwsl_notice("%s: wsi has no task, ~=FINISHED\n", __func__); in lws_threadpool_task_status_wsi()
1101 task = lws_container_of(wsi->tp_task_owner.head, in lws_threadpool_task_status_wsi()
1104 *_task = task; in lws_threadpool_task_status_wsi()
1106 return lws_threadpool_task_status(task, user); in lws_threadpool_task_status_wsi()
1110 lws_threadpool_task_sync(struct lws_threadpool_task *task, int stop) in lws_threadpool_task_sync() argument
1113 if (!task) in lws_threadpool_task_sync()
1117 state_transition(task, LWS_TP_STATUS_STOPPING); in lws_threadpool_task_sync()
1119 pthread_mutex_lock(&task->tp->lock); in lws_threadpool_task_sync()
1120 pthread_cond_signal(&task->wake_idle); in lws_threadpool_task_sync()
1121 pthread_mutex_unlock(&task->tp->lock); in lws_threadpool_task_sync()
1126 int (*cb)(struct lws_threadpool_task *task, in lws_threadpool_foreach_task_wsi() argument
1141 struct lws_threadpool_task *task = lws_container_of(d, in lws_threadpool_foreach_task_wsi() local
1144 if (cb(task, user)) { in lws_threadpool_foreach_task_wsi()
1159 int (*cb)(struct lws_threadpool_task *task, in lws_threadpool_foreach_task_ss() argument
1170 disassociate_wsi(struct lws_threadpool_task *task, in disassociate_wsi() argument
1173 task->args.wsi = NULL; in disassociate_wsi()
1174 lws_dll2_remove(&task->list); in disassociate_wsi()