2012-09-16 [mones] 3.8.1cvs60
[claws.git] / src / etpan / etpan-thread-manager.c
1 /*
2  * Claws Mail -- a GTK+ based, lightweight, and fast e-mail client
3  * Copyright (C) 2005-2012 DINH Viet Hoa and the Claws Mail team
4  *
5  * This program is free software; you can redistribute it and/or modify
6  * it under the terms of the GNU General Public License as published by
7  * the Free Software Foundation; either version 3 of the License, or
8  * (at your option) any later version.
9  *
10  * This program is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License for more details.
14  *
15  * You should have received a copy of the GNU General Public License
16  * along with this program. If not, see <http://www.gnu.org/licenses/>.
17  * 
18  */
19
20 #ifdef HAVE_CONFIG_H
21 #  include "config.h"
22 #include "claws-features.h"
23 #endif
24
25 #ifdef HAVE_LIBETPAN
26
27 #include "etpan-thread-manager.h"
28
29 #include <glib.h>
30 #include <stdlib.h>
31 #include <pthread.h>
32 #include <libetpan/mailsem.h>
33 #include <semaphore.h>
34 #include <unistd.h>
35 #include <fcntl.h>
36
37 #include "etpan-errors.h"
38 #include "utils.h"
39
40 #define POOL_UNBOUND_MAX 4
41
42 #define POOL_INIT_SIZE 8
43 #define OP_INIT_SIZE 8
44
45 static int etpan_thread_start(struct etpan_thread * thread);
46 static void etpan_thread_free(struct etpan_thread * thread);
47 static unsigned int etpan_thread_get_load(struct etpan_thread * thread);
48 static int etpan_thread_is_bound(struct etpan_thread * thread);
49 static int etpan_thread_manager_is_stopped(struct etpan_thread_manager * manager);
50 static void etpan_thread_join(struct etpan_thread * thread);
51 static struct etpan_thread * etpan_thread_new(void);
52 static int etpan_thread_op_cancelled(struct etpan_thread_op * op);
53 static void etpan_thread_op_lock(struct etpan_thread_op * op);
54 static void etpan_thread_op_unlock(struct etpan_thread_op * op);
55 static void etpan_thread_stop(struct etpan_thread * thread);
56
57 #if 0
58 static void etpan_thread_bind(struct etpan_thread * thread);
59 static int etpan_thread_manager_op_schedule(struct etpan_thread_manager * manager,
60              struct etpan_thread_op * op);
61 static void etpan_thread_manager_start(struct etpan_thread_manager * manager);
62 static void etpan_thread_op_cancel(struct etpan_thread_op * op);
63 #endif
64
65 enum {
66   TERMINATE_STATE_NONE,
67   TERMINATE_STATE_REQUESTED,
68   TERMINATE_STATE_DONE,
69 };
70
71 struct etpan_thread_manager * etpan_thread_manager_new(void)
72 {
73   struct etpan_thread_manager * manager;
74   int r;
75   
76   manager = malloc(sizeof(* manager));
77   if (manager == NULL)
78     goto err;
79   
80   manager->thread_pool = carray_new(POOL_INIT_SIZE);
81   if (manager->thread_pool == NULL)
82     goto free;
83
84   manager->thread_pending = carray_new(POOL_INIT_SIZE);
85   if (manager->thread_pending == NULL)
86     goto free_pool;
87   
88   manager->can_create_thread = 1;
89   manager->unbound_count = 0;
90   
91   r = pipe(manager->notify_fds);
92   if (r < 0)
93     goto free_pending;
94   
95   return manager;
96   
97  free_pending:
98   carray_free(manager->thread_pending);
99  free_pool:
100   carray_free(manager->thread_pool);
101  free:
102   free(manager);
103  err:
104   return NULL;
105 }
106
107 void etpan_thread_manager_free(struct etpan_thread_manager * manager)
108 {
109   close(manager->notify_fds[1]);
110   close(manager->notify_fds[0]);
111   carray_free(manager->thread_pending);
112   carray_free(manager->thread_pool);
113   free(manager);
114 }
115
116 static struct etpan_thread * etpan_thread_new(void)
117 {
118   struct etpan_thread * thread;
119   int r;
120   
121   thread = malloc(sizeof(* thread));
122   if (thread == NULL)
123     goto err;
124   
125   r = pthread_mutex_init(&thread->lock, NULL);
126   if (r != 0)
127     goto free;
128
129   thread->op_list = carray_new(OP_INIT_SIZE);
130   if (thread->op_list == NULL)
131     goto destroy_lock;
132   
133   thread->op_done_list = carray_new(OP_INIT_SIZE);
134   if (thread->op_done_list == NULL)
135     goto free_op_list;
136   
137   thread->start_sem = mailsem_new();
138   if (thread->start_sem == NULL)
139     goto free_op_done_list;
140   
141   thread->stop_sem = mailsem_new();
142   if (thread->stop_sem == NULL)
143     goto free_startsem;
144   
145   thread->op_sem = mailsem_new();
146   if (thread->op_sem == NULL)
147     goto free_stopsem;
148   
149   thread->manager = NULL;
150   thread->bound_count = 0;
151   thread->terminate_state = TERMINATE_STATE_NONE;
152   
153   return thread;
154   
155  free_stopsem:
156   mailsem_free(thread->stop_sem);
157  free_startsem:
158   mailsem_free(thread->start_sem);
159  free_op_done_list:
160   carray_free(thread->op_done_list);
161  free_op_list:
162   carray_free(thread->op_list);
163  destroy_lock:
164   pthread_mutex_destroy(&thread->lock);
165  free:
166   free(thread);
167  err:
168   return NULL;
169 }
170
171 static void etpan_thread_free(struct etpan_thread * thread)
172 {
173   mailsem_free(thread->op_sem);
174   mailsem_free(thread->stop_sem);
175   mailsem_free(thread->start_sem);
176   carray_free(thread->op_done_list);
177   carray_free(thread->op_list);
178   pthread_mutex_destroy(&thread->lock);
179   free(thread);
180 }
181
182 struct etpan_thread_op * etpan_thread_op_new(void)
183 {
184   struct etpan_thread_op * op;
185   int r;
186   
187   op = malloc(sizeof(* op));
188   if (op == NULL)
189     goto err;
190   
191   op->thread = NULL;
192   op->run = NULL;
193   op->callback = NULL;
194   op->callback_data = NULL;
195   op->callback_called = 0;
196   op->cancellable = 0;
197   op->cancelled = 0;
198   op->param = NULL;
199   op->result = NULL;
200   op->finished = 0;
201   op->imap = NULL;
202   op->nntp = NULL;
203
204   r = pthread_mutex_init(&op->lock, NULL);
205   if (r != 0)
206     goto free;
207   
208   return op;
209   
210  free:
211   free(op);
212  err:
213   return NULL;
214 }
215
216 void etpan_thread_op_free(struct etpan_thread_op * op)
217 {
218   pthread_mutex_destroy(&op->lock);
219   free(op);
220 }
221
222 static struct etpan_thread *
223 etpan_thread_manager_create_thread(struct etpan_thread_manager * manager)
224 {
225   struct etpan_thread * thread;
226   int r;
227   
228   thread = etpan_thread_new();
229   if (thread == NULL)
230     goto err;
231   
232   thread->manager = manager;
233   
234   r = etpan_thread_start(thread);
235   if (r != NO_ERROR)
236     goto free_thread;
237   
238   r = carray_add(manager->thread_pool, thread, NULL);
239   if (r < 0) {
240     etpan_thread_stop(thread);
241     goto free_thread;
242   }
243   
244   return thread;
245   
246  free_thread:
247   etpan_thread_free(thread);
248  err:
249   return NULL;
250 }
251
252 static void
253 etpan_thread_manager_terminate_thread(struct etpan_thread_manager * manager,
254     struct etpan_thread * thread)
255 {
256   unsigned int i;
257   int r;
258   
259   for(i = 0 ; i < carray_count(manager->thread_pool) ; i ++) {
260     if (carray_get(manager->thread_pool, i) == thread) {
261       carray_delete(manager->thread_pool, i);
262       break;
263     }
264   }
265   
266   if (!etpan_thread_is_bound(thread))
267     manager->unbound_count --;
268   
269   r = carray_add(manager->thread_pending, thread, NULL);
270   if (r < 0) {
271     g_warning("complete failure of thread due to lack of memory (thread stop)");
272   }
273   
274   etpan_thread_stop(thread);
275 }
276
277 static void manager_notify(struct etpan_thread_manager * manager)
278 {
279   char ch;
280   ssize_t r;
281   
282   ch = 1;
283   r = write(manager->notify_fds[1], &ch, 1);
284   if (r < 0) {
285     g_warning("error writing notification to etpan thread manager");
286   }
287 }
288
289 static void manager_ack(struct etpan_thread_manager * manager)
290 {
291 #ifndef G_OS_WIN32
292   char ch;
293   ssize_t r;
294   r = read(manager->notify_fds[0], &ch, 1);
295   if (r != 1) {
296     g_warning("error reading notification from etpan thread manager");
297   }
298 #else
299   /* done in the GIOChannel handler in imap-thread.c and nntp-thread.c */
300 #endif
301 }
302
303 static void thread_lock(struct etpan_thread * thread)
304 {
305   pthread_mutex_lock(&thread->lock);
306 }
307
308 static void thread_unlock(struct etpan_thread * thread)
309 {
310   pthread_mutex_unlock(&thread->lock);
311 }
312
313 static void thread_notify(struct etpan_thread * thread)
314 {
315   manager_notify(thread->manager);
316 }
317
318 static void * thread_run(void * data)
319 {
320   struct etpan_thread * thread;
321   int r;
322   
323   thread = data;
324   
325   mailsem_up(thread->start_sem);
326   
327   while (1) {
328     int do_quit;
329     struct etpan_thread_op * op;
330     
331     mailsem_down(thread->op_sem);
332     
333     do_quit = 0;
334     op = NULL;
335     thread_lock(thread);
336     if (carray_count(thread->op_list) > 0) {
337       op = carray_get(thread->op_list, 0);
338       carray_delete_slow(thread->op_list, 0);
339     }
340     else {
341       do_quit = 1;
342     }
343     thread_unlock(thread);
344     
345     if (do_quit) {
346       break;
347     }
348     
349     if (!etpan_thread_op_cancelled(op)) {
350       if (op->run != NULL)
351         op->run(op);
352     }
353     
354     thread_lock(thread);
355     r = carray_add(thread->op_done_list, op, NULL);
356     if (r < 0) {
357       g_warning("complete failure of thread due to lack of memory (op done)");
358     }
359     thread_unlock(thread);
360     
361     thread_notify(thread);
362   }
363   
364   thread_lock(thread);
365   thread->terminate_state = TERMINATE_STATE_DONE;
366   thread_unlock(thread);
367   
368   thread_notify(thread);
369   
370   mailsem_up(thread->stop_sem);
371   
372   return NULL;
373 }
374
375 static int etpan_thread_start(struct etpan_thread * thread)
376 {
377   int r;
378   
379   r = pthread_create(&thread->th_id, NULL, thread_run, thread);
380   if (r != 0)
381     return ERROR_MEMORY;
382   
383   mailsem_down(thread->start_sem);
384   
385   return NO_ERROR;
386 }
387
388 static void etpan_thread_stop(struct etpan_thread * thread)
389 {
390   thread_lock(thread);
391   thread->terminate_state = TERMINATE_STATE_REQUESTED;
392   thread_unlock(thread);
393   
394   mailsem_up(thread->op_sem);
395   
396   /* this thread will be joined in the manager loop */
397 }
398
399 static int etpan_thread_is_stopped(struct etpan_thread * thread)
400 {
401   int stopped;
402   
403   thread_lock(thread);
404   stopped = (thread->terminate_state == TERMINATE_STATE_DONE);
405   thread_unlock(thread);
406   
407   return stopped;
408 }
409
410 static void etpan_thread_join(struct etpan_thread * thread)
411 {
412   mailsem_down(thread->stop_sem);
413   pthread_join(thread->th_id, NULL);
414 }
415
416 struct etpan_thread *
417 etpan_thread_manager_get_thread(struct etpan_thread_manager * manager)
418 {
419   struct etpan_thread * chosen_thread;
420   unsigned int chosen_thread_load;
421   unsigned int i;
422   struct etpan_thread * thread;
423   
424   /* chose a thread */
425   
426   chosen_thread = NULL;
427   chosen_thread_load = 0;
428   
429   for(i = 0 ; i < carray_count(manager->thread_pool) ; i ++) {
430     thread = carray_get(manager->thread_pool, i);
431     if (etpan_thread_is_bound(thread))
432       continue;
433     
434     if (chosen_thread == NULL) {
435       chosen_thread = thread;
436       chosen_thread_load = etpan_thread_get_load(thread);
437       
438       if (chosen_thread_load == 0)
439         break;
440     }
441     else {
442       unsigned int load;
443       
444       load = etpan_thread_get_load(thread);
445       
446       if (load < chosen_thread_load) {
447         chosen_thread = thread;
448         chosen_thread_load = load;
449       }
450     }
451   }
452   
453   if (chosen_thread != NULL) {
454     if (manager->can_create_thread && (chosen_thread_load != 0)) {
455       chosen_thread = NULL;
456     }
457   }
458   
459   /* choice done */
460   
461   if (chosen_thread != NULL)
462     return chosen_thread;
463   
464   thread = etpan_thread_manager_create_thread(manager);
465   if (thread == NULL)
466     goto err;
467   
468   manager->unbound_count ++;
469   if (manager->unbound_count >= POOL_UNBOUND_MAX)
470     manager->can_create_thread = 0;
471   
472   return thread;
473   
474  err:
475   return NULL;
476 }
477
478 static unsigned int etpan_thread_get_load(struct etpan_thread * thread)
479 {
480   unsigned int load;
481   
482   thread_lock(thread);
483   load = carray_count(thread->op_list);
484   thread_unlock(thread);
485   
486   return load;
487 }
488
489 #if 0
490 static void etpan_thread_bind(struct etpan_thread * thread)
491 {
492   thread->bound_count ++;
493 }
494 #endif
495
496 void etpan_thread_unbind(struct etpan_thread * thread)
497 {
498   thread->bound_count --;
499 }
500
501 static int etpan_thread_is_bound(struct etpan_thread * thread)
502 {
503   return (thread->bound_count != 0);
504 }
505
506 int etpan_thread_op_schedule(struct etpan_thread * thread,
507                              struct etpan_thread_op * op)
508 {
509   int r;
510   
511   if (thread->terminate_state != TERMINATE_STATE_NONE)
512     return ERROR_INVAL;
513   
514   thread_lock(thread);
515   r = carray_add(thread->op_list, op, NULL);
516   thread_unlock(thread);
517   
518   if (r < 0)
519     return ERROR_MEMORY;
520   
521   op->thread = thread;
522   mailsem_up(thread->op_sem);
523   
524   return NO_ERROR;
525 }
526
527 static void etpan_thread_op_lock(struct etpan_thread_op * op)
528 {
529   pthread_mutex_lock(&op->lock);
530 }
531
532 static void etpan_thread_op_unlock(struct etpan_thread_op * op)
533 {
534   pthread_mutex_unlock(&op->lock);
535 }
536
537 static int etpan_thread_op_cancelled(struct etpan_thread_op * op)
538 {
539   int cancelled;
540   
541   cancelled = 0;
542   etpan_thread_op_lock(op);
543   if (op->cancellable)
544     cancelled = op->cancelled;
545   etpan_thread_op_unlock(op);
546   
547   return cancelled;
548 }
549
550 int etpan_thread_manager_get_fd(struct etpan_thread_manager * manager)
551 {
552   return manager->notify_fds[0];
553 }
554
555 static void loop_thread_list(carray * op_to_notify,
556     carray * thread_list)
557 {
558   unsigned int i;
559   int r;
560   
561   for(i = 0 ; i < carray_count(thread_list) ; i ++) {
562     struct etpan_thread * thread;
563     unsigned int j;
564     
565     thread = carray_get(thread_list, i);
566     
567     thread_lock(thread);
568     
569     for(j = 0 ; j < carray_count(thread->op_done_list) ; j ++) {
570       struct etpan_thread_op * op;
571       
572       op = carray_get(thread->op_done_list, j);
573       r = carray_add(op_to_notify, op, NULL);
574       if (r < 0) {
575         g_warning("complete failure of thread due to lack of memory (callback)");
576         break;
577       }
578     }
579     carray_set_size(thread->op_done_list, 0);
580     
581     thread_unlock(thread);
582   }
583 }
584
585 void etpan_thread_manager_loop(struct etpan_thread_manager * manager)
586 {
587   carray * op_to_notify;
588   unsigned int i;
589   
590   manager_ack(manager);
591   
592   op_to_notify = carray_new(OP_INIT_SIZE);
593   
594   loop_thread_list(op_to_notify, manager->thread_pool);
595   loop_thread_list(op_to_notify, manager->thread_pending);
596   
597   for(i = 0 ; i < carray_count(op_to_notify) ; i ++) {
598     struct etpan_thread_op * op;
599     
600     op = carray_get(op_to_notify, i);
601     
602     etpan_thread_op_lock(op);
603     
604     if (!op->callback_called) {
605       if (op->callback != NULL)
606         op->callback(op->cancelled, op->result, op->callback_data);
607     }
608     
609     etpan_thread_op_unlock(op);
610     
611     if (op->cleanup != NULL)
612       op->cleanup(op);
613   }
614   
615   carray_free(op_to_notify);
616   
617   i = 0;
618   while (i < carray_count(manager->thread_pending)) {
619     struct etpan_thread * thread;
620     
621     thread = carray_get(manager->thread_pending, i);
622     
623     if (etpan_thread_is_stopped(thread)) {
624       etpan_thread_join(thread);
625       
626       etpan_thread_free(thread);
627       
628       carray_delete_slow(manager->thread_pending, i);
629     }
630     else {
631       i ++;
632     }
633   }
634 }
635
636 #if 0
637 static void etpan_thread_manager_start(struct etpan_thread_manager * manager)
638 {
639   /* do nothing */
640 }
641 #endif
642
643 void etpan_thread_manager_stop(struct etpan_thread_manager * manager)
644 {
645   while (carray_count(manager->thread_pool) > 0) {
646     struct etpan_thread * thread;
647     
648     thread = carray_get(manager->thread_pool, 0);
649     etpan_thread_manager_terminate_thread(manager, thread);
650   }
651 }
652
653 static int etpan_thread_manager_is_stopped(struct etpan_thread_manager * manager)
654 {
655   return ((carray_count(manager->thread_pending) == 0) && 
656       (carray_count(manager->thread_pool) == 0));
657 }
658
659 void etpan_thread_manager_join(struct etpan_thread_manager * manager)
660 {
661   while (!etpan_thread_manager_is_stopped(manager)) {
662     etpan_thread_manager_loop(manager);
663   }
664 }
665 #endif