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