d469c101052c9f4f3a3418bd4c53801a046267fd
[src/app-framework-binder.git] / src / jobs.c
1 /*
2  * Copyright (C) 2016, 2017 "IoT.bzh"
3  * Author José Bollo <jose.bollo@iot.bzh>
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *   http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17
18 #define _GNU_SOURCE
19
20 #include <stdlib.h>
21 #include <stdint.h>
22 #include <unistd.h>
23 #include <signal.h>
24 #include <time.h>
25 #include <sys/syscall.h>
26 #include <pthread.h>
27 #include <errno.h>
28 #include <assert.h>
29 #include <sys/eventfd.h>
30
31 #include <systemd/sd-event.h>
32
33 #include "jobs.h"
34 #include "sig-monitor.h"
35 #include "verbose.h"
36
37 #if 0
38 #define _alert_ "do you really want to remove monitoring?"
39 #define sig_monitor_init_timeouts()  ((void)0)
40 #define sig_monitor_clean_timeouts() ((void)0)
41 #define sig_monitor(to,cb,arg)       (cb(0,arg))
42 #endif
43
44 #define EVENT_TIMEOUT_TOP       ((uint64_t)-1)
45 #define EVENT_TIMEOUT_CHILD     ((uint64_t)10000)
46
47 /** Internal shortcut for callback */
48 typedef void (*job_cb_t)(int, void*);
49
50 /** Description of a pending job */
51 struct job
52 {
53         struct job *next;    /**< link to the next job enqueued */
54         const void *group;   /**< group of the request */
55         job_cb_t callback;   /**< processing callback */
56         void *arg;           /**< argument */
57         int timeout;         /**< timeout in second for processing the request */
58         unsigned blocked: 1; /**< is an other request blocking this one ? */
59         unsigned dropped: 1; /**< is removed ? */
60 };
61
62 /** Description of handled event loops */
63 struct evloop
64 {
65         unsigned state;        /**< encoded state */
66         int efd;               /**< event notification */
67         struct sd_event *sdev; /**< the systemd event loop */
68         pthread_cond_t  cond;  /**< condition */
69 };
70
71 #define EVLOOP_STATE_WAIT           1U
72 #define EVLOOP_STATE_RUN            2U
73 #define EVLOOP_STATE_LOCK           4U
74
75 /** Description of threads */
76 struct thread
77 {
78         struct thread *next;   /**< next thread of the list */
79         struct thread *upper;  /**< upper same thread */
80         struct job *job;       /**< currently processed job */
81         pthread_t tid;         /**< the thread id */
82         unsigned stop: 1;      /**< stop requested */
83         unsigned waits: 1;     /**< is waiting? */
84 };
85
86 /**
87  * Description of synchonous callback
88  */
89 struct sync
90 {
91         struct thread thread;   /**< thread loop data */
92         union {
93                 void (*callback)(int, void*);   /**< the synchronous callback */
94                 void (*enter)(int signum, void *closure, struct jobloop *jobloop);
95                                 /**< the entering synchronous routine */
96         };
97         void *arg;              /**< the argument of the callback */
98 };
99
100
101 /* synchronisation of threads */
102 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
103 static pthread_cond_t  cond = PTHREAD_COND_INITIALIZER;
104
105 /* count allowed, started and running threads */
106 static int allowed = 0; /** allowed count of threads */
107 static int started = 0; /** started count of threads */
108 static int running = 0; /** running count of threads */
109 static int remains = 0; /** allowed count of waiting jobs */
110
111 /* list of threads */
112 static struct thread *threads;
113 static _Thread_local struct thread *current_thread;
114 static _Thread_local struct evloop *current_evloop;
115
116 /* queue of pending jobs */
117 static struct job *first_job;
118 static struct job *free_jobs;
119
120 /* event loop */
121 static struct evloop evloop[1];
122
123 /**
124  * Create a new job with the given parameters
125  * @param group    the group of the job
126  * @param timeout  the timeout of the job (0 if none)
127  * @param callback the function that achieves the job
128  * @param arg      the argument of the callback
129  * @return the created job unblock or NULL when no more memory
130  */
131 static struct job *job_create(
132                 const void *group,
133                 int timeout,
134                 job_cb_t callback,
135                 void *arg)
136 {
137         struct job *job;
138
139         /* try recyle existing job */
140         job = free_jobs;
141         if (job)
142                 free_jobs = job->next;
143         else {
144                 /* allocation  without blocking */
145                 pthread_mutex_unlock(&mutex);
146                 job = malloc(sizeof *job);
147                 pthread_mutex_lock(&mutex);
148                 if (!job) {
149                         errno = -ENOMEM;
150                         goto end;
151                 }
152         }
153         /* initialises the job */
154         job->group = group;
155         job->timeout = timeout;
156         job->callback = callback;
157         job->arg = arg;
158         job->blocked = 0;
159         job->dropped = 0;
160 end:
161         return job;
162 }
163
164 /**
165  * Adds 'job' at the end of the list of jobs, marking it
166  * as blocked if an other job with the same group is pending.
167  * @param job the job to add
168  */
169 static void job_add(struct job *job)
170 {
171         const void *group;
172         struct job *ijob, **pjob;
173
174         /* prepare to add */
175         group = job->group;
176         job->next = NULL;
177
178         /* search end and blockers */
179         pjob = &first_job;
180         ijob = first_job;
181         while (ijob) {
182                 if (group && ijob->group == group)
183                         job->blocked = 1;
184                 pjob = &ijob->next;
185                 ijob = ijob->next;
186         }
187
188         /* queue the jobs */
189         *pjob = job;
190 }
191
192 /**
193  * Get the next job to process or NULL if none.
194  * @return the first job that isn't blocked or NULL
195  */
196 static inline struct job *job_get()
197 {
198         struct job *job = first_job;
199         while (job && job->blocked)
200                 job = job->next;
201         return job;
202 }
203
204 /**
205  * Releases the processed 'job': removes it
206  * from the list of jobs and unblock the first
207  * pending job of the same group if any.
208  * @param job the job to release
209  */
210 static inline void job_release(struct job *job)
211 {
212         struct job *ijob, **pjob;
213         const void *group;
214
215         /* first unqueue the job */
216         pjob = &first_job;
217         ijob = first_job;
218         while (ijob != job) {
219                 pjob = &ijob->next;
220                 ijob = ijob->next;
221         }
222         *pjob = job->next;
223
224         /* then unblock jobs of the same group */
225         group = job->group;
226         if (group) {
227                 ijob = job->next;
228                 while (ijob && ijob->group != group)
229                         ijob = ijob->next;
230                 if (ijob)
231                         ijob->blocked = 0;
232         }
233
234         /* recycle the job */
235         job->next = free_jobs;
236         free_jobs = job;
237 }
238
239 /**
240  * Monitored cancel callback for a job.
241  * This function is called by the monitor
242  * to cancel the job when the safe environment
243  * is set.
244  * @param signum 0 on normal flow or the number
245  *               of the signal that interrupted the normal
246  *               flow, isn't used
247  * @param arg    the job to run
248  */
249 static void job_cancel(int signum, void *arg)
250 {
251         struct job *job = arg;
252         job->callback(SIGABRT, job->arg);
253 }
254
255 /**
256  * Monitored normal callback for events.
257  * This function is called by the monitor
258  * to run the event loop when the safe environment
259  * is set.
260  * @param signum 0 on normal flow or the number
261  *               of the signal that interrupted the normal
262  *               flow
263  * @param arg     the events to run
264  */
265 static void evloop_run(int signum, void *arg)
266 {
267         int rc;
268         struct sd_event *se;
269         struct evloop *el = arg;
270
271         if (!signum) {
272                 se = el->sdev;
273                 rc = sd_event_prepare(se);
274                 if (rc < 0) {
275                         errno = -rc;
276                         ERROR("sd_event_prepare returned an error (state: %d): %m", sd_event_get_state(se));
277                 } else {
278                         if (rc == 0) {
279                                 rc = sd_event_wait(se, (uint64_t)(int64_t)-1);
280                                 if (rc < 0) {
281                                         errno = -rc;
282                                         ERROR("sd_event_wait returned an error (state: %d): %m", sd_event_get_state(se));
283                                 }
284                         }
285                         __atomic_and_fetch(&el->state, ~(EVLOOP_STATE_WAIT), __ATOMIC_RELAXED);
286
287                         if (rc > 0) {
288                                 rc = sd_event_dispatch(se);
289                                 if (rc < 0) {
290                                         errno = -rc;
291                                         ERROR("sd_event_dispatch returned an error (state: %d): %m", sd_event_get_state(se));
292                                 }
293                         }
294                 }
295         }
296         __atomic_and_fetch(&el->state, ~(EVLOOP_STATE_WAIT|EVLOOP_STATE_RUN), __ATOMIC_RELAXED);
297 }
298
299
300 /**
301  * Main processing loop of threads processing jobs.
302  * The loop must be called with the mutex locked
303  * and it returns with the mutex locked.
304  * @param me the description of the thread to use
305  * TODO: how are timeout handled when reentering?
306  */
307 static void thread_run(volatile struct thread *me)
308 {
309         struct thread **prv;
310         struct job *job;
311         struct evloop *el;
312
313         /* initialize description of itself and link it in the list */
314         me->tid = pthread_self();
315         me->stop = 0;
316         me->waits = 0;
317         me->upper = current_thread;
318         if (!current_thread) {
319                 started++;
320                 sig_monitor_init_timeouts();
321         }
322         me->next = threads;
323         threads = (struct thread*)me;
324         current_thread = (struct thread*)me;
325
326         /* loop until stopped */
327         while (!me->stop) {
328                 /* release the event loop */
329                 if (current_evloop) {
330                         __atomic_sub_fetch(&current_evloop->state, EVLOOP_STATE_LOCK, __ATOMIC_RELAXED);
331                         current_evloop = NULL;
332                 }
333
334                 /* get a job */
335                 job = job_get(first_job);
336                 if (job) {
337                         /* prepare running the job */
338                         remains++; /* increases count of job that can wait */
339                         job->blocked = 1; /* mark job as blocked */
340                         me->job = job; /* record the job (only for terminate) */
341
342                         /* run the job */
343                         pthread_mutex_unlock(&mutex);
344                         sig_monitor(job->timeout, job->callback, job->arg);
345                         pthread_mutex_lock(&mutex);
346
347                         /* release the run job */
348                         job_release(job);
349                 } else {
350                         /* no job, check events */
351                         el = &evloop[0];
352                         if (el->sdev && !__atomic_load_n(&el->state, __ATOMIC_RELAXED)) {
353                                 /* run the events */
354                                 __atomic_store_n(&el->state, EVLOOP_STATE_LOCK|EVLOOP_STATE_RUN|EVLOOP_STATE_WAIT, __ATOMIC_RELAXED);
355                                 current_evloop = el;
356                                 pthread_mutex_unlock(&mutex);
357                                 sig_monitor(0, evloop_run, el);
358                                 pthread_mutex_lock(&mutex);
359                         } else {
360                                 /* no job and not events */
361                                 running--;
362                                 if (!running)
363                                         ERROR("Entering job deep sleep! Check your bindings.");
364                                 me->waits = 1;
365                                 pthread_cond_wait(&cond, &mutex);
366                                 me->waits = 0;
367                                 running++;
368                         }
369                 }
370         }
371
372         /* release the event loop */
373         if (current_evloop) {
374                 __atomic_sub_fetch(&current_evloop->state, EVLOOP_STATE_LOCK, __ATOMIC_RELAXED);
375                 current_evloop = NULL;
376         }
377
378         /* unlink the current thread and cleanup */
379         prv = &threads;
380         while (*prv != me)
381                 prv = &(*prv)->next;
382         *prv = me->next;
383         current_thread = me->upper;
384         if (!current_thread) {
385                 sig_monitor_clean_timeouts();
386                 started--;
387         }
388 }
389
390 /**
391  * Entry point for created threads.
392  * @param data not used
393  * @return NULL
394  */
395 static void *thread_main(void *data)
396 {
397         struct thread me;
398
399         pthread_mutex_lock(&mutex);
400         running++;
401         thread_run(&me);
402         running--;
403         pthread_mutex_unlock(&mutex);
404         return NULL;
405 }
406
407 /**
408  * Starts a new thread
409  * @return 0 in case of success or -1 in case of error
410  */
411 static int start_one_thread()
412 {
413         pthread_t tid;
414         int rc;
415
416         rc = pthread_create(&tid, NULL, thread_main, NULL);
417         if (rc != 0) {
418                 /* errno = rc; */
419                 WARNING("not able to start thread: %m");
420                 rc = -1;
421         }
422         return rc;
423 }
424
425 /**
426  * Queues a new asynchronous job represented by 'callback' and 'arg'
427  * for the 'group' and the 'timeout'.
428  * Jobs are queued FIFO and are possibly executed in parallel
429  * concurrently except for job of the same group that are
430  * executed sequentially in FIFO order.
431  * @param group    The group of the job or NULL when no group.
432  * @param timeout  The maximum execution time in seconds of the job
433  *                 or 0 for unlimited time.
434  * @param callback The function to execute for achieving the job.
435  *                 Its first parameter is either 0 on normal flow
436  *                 or the signal number that broke the normal flow.
437  *                 The remaining parameter is the parameter 'arg1'
438  *                 given here.
439  * @param arg      The second argument for 'callback'
440  * @return 0 in case of success or -1 in case of error
441  */
442 int jobs_queue(
443                 const void *group,
444                 int timeout,
445                 void (*callback)(int, void*),
446                 void *arg)
447 {
448         const char *info;
449         struct job *job;
450         int rc;
451
452         pthread_mutex_lock(&mutex);
453
454         /* allocates the job */
455         job = job_create(group, timeout, callback, arg);
456         if (!job) {
457                 errno = ENOMEM;
458                 info = "out of memory";
459                 goto error;
460         }
461
462         /* check availability */
463         if (remains == 0) {
464                 errno = EBUSY;
465                 info = "too many jobs";
466                 goto error2;
467         }
468
469         /* start a thread if needed */
470         if (running == started && started < allowed) {
471                 /* all threads are busy and a new can be started */
472                 rc = start_one_thread();
473                 if (rc < 0 && started == 0) {
474                         info = "can't start first thread";
475                         goto error2;
476                 }
477         }
478
479         /* queues the job */
480         remains--;
481         job_add(job);
482
483         /* signal an existing job */
484         pthread_cond_signal(&cond);
485         pthread_mutex_unlock(&mutex);
486         return 0;
487
488 error2:
489         job->next = free_jobs;
490         free_jobs = job;
491 error:
492         ERROR("can't process job with threads: %s, %m", info);
493         pthread_mutex_unlock(&mutex);
494         return -1;
495 }
496
497 /**
498  * Internal helper function for 'jobs_enter'.
499  * @see jobs_enter, jobs_leave
500  */
501 static void enter_cb(int signum, void *closure)
502 {
503         struct sync *sync = closure;
504         sync->enter(signum, sync->arg, (void*)&sync->thread);
505 }
506
507 /**
508  * Internal helper function for 'jobs_call'.
509  * @see jobs_call
510  */
511 static void call_cb(int signum, void *closure)
512 {
513         struct sync *sync = closure;
514         sync->callback(signum, sync->arg);
515         jobs_leave((void*)&sync->thread);
516 }
517
518 /**
519  * Internal helper for synchronous jobs. It enters
520  * a new thread loop for evaluating the given job
521  * as recorded by the couple 'sync_cb' and 'sync'.
522  * @see jobs_call, jobs_enter, jobs_leave
523  */
524 static int do_sync(
525                 const void *group,
526                 int timeout,
527                 void (*sync_cb)(int signum, void *closure),
528                 struct sync *sync
529 )
530 {
531         struct job *job;
532
533         pthread_mutex_lock(&mutex);
534
535         /* allocates the job */
536         job = job_create(group, timeout, sync_cb, sync);
537         if (!job) {
538                 ERROR("out of memory");
539                 errno = ENOMEM;
540                 pthread_mutex_unlock(&mutex);
541                 return -1;
542         }
543
544         /* queues the job */
545         job_add(job);
546
547         /* run until stopped */
548         thread_run(&sync->thread);
549         pthread_mutex_unlock(&mutex);
550         return 0;
551 }
552
553 /**
554  * Enter a synchronisation point: activates the job given by 'callback'
555  * and 'closure' using 'group' and 'timeout' to control sequencing and
556  * execution time.
557  * @param group the group for sequencing jobs
558  * @param timeout the time in seconds allocated to the job
559  * @param callback the callback that will handle the job.
560  *                 it receives 3 parameters: 'signum' that will be 0
561  *                 on normal flow or the catched signal number in case
562  *                 of interrupted flow, the context 'closure' as given and
563  *                 a 'jobloop' reference that must be used when the job is
564  *                 terminated to unlock the current execution flow.
565  * @param arg the argument to the callback
566  * @return 0 on success or -1 in case of error
567  */
568 int jobs_enter(
569                 const void *group,
570                 int timeout,
571                 void (*callback)(int signum, void *closure, struct jobloop *jobloop),
572                 void *closure
573 )
574 {
575         struct sync sync;
576
577         sync.enter = callback;
578         sync.arg = closure;
579         return do_sync(group, timeout, enter_cb, &sync);
580 }
581
582 /**
583  * Unlocks the execution flow designed by 'jobloop'.
584  * @param jobloop indication of the flow to unlock
585  * @return 0 in case of success of -1 on error
586  */
587 int jobs_leave(struct jobloop *jobloop)
588 {
589         struct thread *t;
590
591         pthread_mutex_lock(&mutex);
592         t = threads;
593         while (t && t != (struct thread*)jobloop)
594                 t = t->next;
595         if (!t) {
596                 errno = EINVAL;
597         } else {
598                 t->stop = 1;
599                 if (t->waits)
600                         pthread_cond_broadcast(&cond);
601         }
602         pthread_mutex_unlock(&mutex);
603         return -!t;
604 }
605
606 /**
607  * Calls synchronously the job represented by 'callback' and 'arg1'
608  * for the 'group' and the 'timeout' and waits for its completion.
609  * @param group    The group of the job or NULL when no group.
610  * @param timeout  The maximum execution time in seconds of the job
611  *                 or 0 for unlimited time.
612  * @param callback The function to execute for achieving the job.
613  *                 Its first parameter is either 0 on normal flow
614  *                 or the signal number that broke the normal flow.
615  *                 The remaining parameter is the parameter 'arg1'
616  *                 given here.
617  * @param arg      The second argument for 'callback'
618  * @return 0 in case of success or -1 in case of error
619  */
620 int jobs_call(
621                 const void *group,
622                 int timeout,
623                 void (*callback)(int, void*),
624                 void *arg)
625 {
626         struct sync sync;
627
628         sync.callback = callback;
629         sync.arg = arg;
630
631         return do_sync(group, timeout, call_cb, &sync);
632 }
633
634 /**
635  * Internal callback for evloop management.
636  * The effect of this function is hidden: it exits
637  * the waiting poll if any. Then it wakes up a thread
638  * awaiting the evloop using signal.
639  */
640 static int on_evloop_efd(sd_event_source *s, int fd, uint32_t revents, void *userdata)
641 {
642         uint64_t x;
643         struct evloop *evloop = userdata;
644         read(evloop->efd, &x, sizeof x);
645         pthread_mutex_lock(&mutex);
646         pthread_cond_broadcast(&evloop->cond);  
647         pthread_mutex_unlock(&mutex);
648         return 1;
649 }
650
651 /**
652  * Gets a sd_event item for the current thread.
653  * @return a sd_event or NULL in case of error
654  */
655 static struct sd_event *get_sd_event_locked()
656 {
657         struct evloop *el;
658         uint64_t x;
659         int rc;
660
661         /* creates the evloop on need */
662         el = &evloop[0];
663         if (!el->sdev) {
664                 /* start the creation */
665                 el->state = 0;
666                 /* creates the eventfd for waking up polls */
667                 el->efd = eventfd(0, EFD_CLOEXEC);
668                 if (el->efd < 0) {
669                         ERROR("can't make eventfd for events");
670                         goto error1;
671                 }
672                 /* create the systemd event loop */
673                 rc = sd_event_new(&el->sdev);
674                 if (rc < 0) {
675                         ERROR("can't make new event loop");
676                         goto error2;
677                 }
678                 /* put the eventfd in the event loop */
679                 rc = sd_event_add_io(el->sdev, NULL, el->efd, EPOLLIN, on_evloop_efd, el);
680                 if (rc < 0) {
681                         ERROR("can't register eventfd");
682                         sd_event_unref(el->sdev);
683                         el->sdev = NULL;
684 error2:
685                         close(el->efd);
686 error1:
687                         return NULL;
688                 }
689         }
690
691         /* attach the event loop to the current thread */
692         if (current_evloop != el) {
693                 if (current_evloop)
694                         __atomic_sub_fetch(&current_evloop->state, EVLOOP_STATE_LOCK, __ATOMIC_RELAXED);
695                 current_evloop = el;
696                 __atomic_add_fetch(&el->state, EVLOOP_STATE_LOCK, __ATOMIC_RELAXED);
697         }
698
699         /* wait for a modifiable event loop */
700         while (__atomic_load_n(&el->state, __ATOMIC_RELAXED) & EVLOOP_STATE_WAIT) {
701                 x = 1;
702                 write(el->efd, &x, sizeof x);
703                 pthread_cond_wait(&el->cond, &mutex);
704         }
705
706         return el->sdev;
707 }
708
709 /**
710  * Gets a sd_event item for the current thread.
711  * @return a sd_event or NULL in case of error
712  */
713 struct sd_event *jobs_get_sd_event()
714 {
715         struct sd_event *result;
716
717         pthread_mutex_lock(&mutex);
718         result = get_sd_event_locked();
719         pthread_mutex_unlock(&mutex);
720
721         return result;
722 }
723
724 /**
725  * Enter the jobs processing loop.
726  * @param allowed_count Maximum count of thread for jobs including this one
727  * @param start_count   Count of thread to start now, must be lower.
728  * @param waiter_count  Maximum count of jobs that can be waiting.
729  * @param start         The start routine to activate (can't be NULL)
730  * @return 0 in case of success or -1 in case of error.
731  */
732 int jobs_start(int allowed_count, int start_count, int waiter_count, void (*start)(int signum))
733 {
734         int rc, launched;
735         struct thread me;
736         struct job *job;
737
738         assert(allowed_count >= 1);
739         assert(start_count >= 0);
740         assert(waiter_count > 0);
741         assert(start_count <= allowed_count);
742
743         rc = -1;
744         pthread_mutex_lock(&mutex);
745
746         /* check whether already running */
747         if (current_thread || allowed) {
748                 ERROR("thread already started");
749                 errno = EINVAL;
750                 goto error;
751         }
752
753         /* start */
754         if (sig_monitor_init() < 0) {
755                 ERROR("failed to initialise signal handlers");
756                 goto error;
757         }
758
759         /* records the allowed count */
760         allowed = allowed_count;
761         started = 0;
762         running = 0;
763         remains = waiter_count;
764
765         /* start at least one thread */
766         launched = 0;
767         while ((launched + 1) < start_count) {
768                 if (start_one_thread() != 0) {
769                         ERROR("Not all threads can be started");
770                         goto error;
771                 }
772                 launched++;
773         }
774
775         /* queue the start job */
776         job = job_create(NULL, 0, (job_cb_t)start, NULL);
777         if (!job) {
778                 ERROR("out of memory");
779                 errno = ENOMEM;
780                 goto error;
781         }
782         job_add(job);
783         remains--;
784
785         /* run until end */
786         thread_run(&me);
787         rc = 0;
788 error:
789         pthread_mutex_unlock(&mutex);
790         return rc;
791 }
792
793 /**
794  * Terminate all the threads and cancel all pending jobs.
795  */
796 void jobs_terminate()
797 {
798         struct job *job, *head, *tail;
799         pthread_t me, *others;
800         struct thread *t;
801         int count;
802
803         /* how am i? */
804         me = pthread_self();
805
806         /* request all threads to stop */
807         pthread_mutex_lock(&mutex);
808         allowed = 0;
809
810         /* count the number of threads */
811         count = 0;
812         t = threads;
813         while (t) {
814                 if (!t->upper && !pthread_equal(t->tid, me))
815                         count++;
816                 t = t->next;
817         }
818
819         /* fill the array of threads */
820         others = alloca(count * sizeof *others);
821         count = 0;
822         t = threads;
823         while (t) {
824                 if (!t->upper && !pthread_equal(t->tid, me))
825                         others[count++] = t->tid;
826                 t = t->next;
827         }
828
829         /* stops the threads */
830         t = threads;
831         while (t) {
832                 t->stop = 1;
833                 t = t->next;
834         }
835
836         /* wait the threads */
837         pthread_cond_broadcast(&cond);
838         pthread_mutex_unlock(&mutex);
839         while (count)
840                 pthread_join(others[--count], NULL);
841         pthread_mutex_lock(&mutex);
842
843         /* cancel pending jobs of other threads */
844         remains = 0;
845         head = first_job;
846         first_job = NULL;
847         tail = NULL;
848         while (head) {
849                 /* unlink the job */
850                 job = head;
851                 head = job->next;
852
853                 /* search if job is stacked for current */
854                 t = current_thread;
855                 while (t && t->job != job)
856                         t = t->upper;
857                 if (t) {
858                         /* yes, relink it at end */
859                         if (tail)
860                                 tail->next = job;
861                         else
862                                 first_job = job;
863                         tail = job;
864                         job->next = NULL;
865                 } else {
866                         /* no cancel the job */
867                         pthread_mutex_unlock(&mutex);
868                         sig_monitor(0, job_cancel, job);
869                         free(job);
870                         pthread_mutex_lock(&mutex);
871                 }
872         }
873         pthread_mutex_unlock(&mutex);
874 }
875