lib/thread.c

   1 /* Thread management routine
   2  * Copyright (C) 1998, 2000 Kunihiro Ishiguro <kunihiro@zebra.org>
   3  *
   4  * This file is part of GNU Zebra.
   5  *
   6  * GNU Zebra is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; either version 2, or (at your option) any
   9  * later version.
  10  *
  11  * GNU Zebra is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with GNU Zebra; see the file COPYING.  If not, write to the Free
  18  * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
  19  * 02111-1307, USA.
  20  */
  21
  22 /* #define DEBUG */
  23
  24 #include <zebra.h>
  25 #include <sys/resource.h>
  26
  27 #include "thread.h"
  28 #include "memory.h"
  29 #include "log.h"
  30 #include "hash.h"
  31 #include "pqueue.h"
  32 #include "command.h"
  33 #include "sigevent.h"
  34
  35 #if defined(__APPLE__)
  36 #include <mach/mach.h>
  37 #include <mach/mach_time.h>
  38 #endif
  39
  40 /* Recent absolute time of day */
  41 struct timeval recent_time;
  42 static struct timeval last_recent_time;
  43 /* Relative time, since startup */
  44 static struct timeval relative_time;
  45 static struct timeval relative_time_base;
  46 /* init flag */
  47 static unsigned short timers_inited;
  48
  49 static struct hash *cpu_record = NULL;
  50
  51 /* Struct timeval's tv_usec one second value.  */
  52 #define TIMER_SECOND_MICRO 1000000L
  53
  54 /* Adjust so that tv_usec is in the range [0,TIMER_SECOND_MICRO).
  55    And change negative values to 0. */
  56 static struct timeval
  57 timeval_adjust (struct timeval a)
  58 {
  59   while (a.tv_usec >= TIMER_SECOND_MICRO)
  60     {
  61       a.tv_usec -= TIMER_SECOND_MICRO;
  62       a.tv_sec++;
  63     }
  64
  65   while (a.tv_usec < 0)
  66     {
  67       a.tv_usec += TIMER_SECOND_MICRO;
  68       a.tv_sec--;
  69     }
  70
  71   if (a.tv_sec < 0)
  72       /* Change negative timeouts to 0. */
  73       a.tv_sec = a.tv_usec = 0;
  74
  75   return a;
  76 }
  77
  78 static struct timeval
  79 timeval_subtract (struct timeval a, struct timeval b)
  80 {
  81   struct timeval ret;
  82
  83   ret.tv_usec = a.tv_usec - b.tv_usec;
  84   ret.tv_sec = a.tv_sec - b.tv_sec;
  85
  86   return timeval_adjust (ret);
  87 }
  88
  89 static long
  90 timeval_cmp (struct timeval a, struct timeval b)
  91 {
  92   return (a.tv_sec == b.tv_sec
  93           ? a.tv_usec - b.tv_usec : a.tv_sec - b.tv_sec);
  94 }
  95
  96 unsigned long
  97 timeval_elapsed (struct timeval a, struct timeval b)
  98 {
  99   return (((a.tv_sec - b.tv_sec) * TIMER_SECOND_MICRO)
 100           + (a.tv_usec - b.tv_usec));
 101 }
 102
 103 #if !defined(HAVE_CLOCK_MONOTONIC) && !defined(__APPLE__)
 104 static void
 105 quagga_gettimeofday_relative_adjust (void)
 106 {
 107   struct timeval diff;
 108   if (timeval_cmp (recent_time, last_recent_time) < 0)
 109     {
 110       relative_time.tv_sec++;
 111       relative_time.tv_usec = 0;
 112     }
 113   else
 114     {
 115       diff = timeval_subtract (recent_time, last_recent_time);
 116       relative_time.tv_sec += diff.tv_sec;
 117       relative_time.tv_usec += diff.tv_usec;
 118       relative_time = timeval_adjust (relative_time);
 119     }
 120   last_recent_time = recent_time;
 121 }
 122 #endif /* !HAVE_CLOCK_MONOTONIC && !__APPLE__ */
 123
 124 /* gettimeofday wrapper, to keep recent_time updated */
 125 static int
 126 quagga_gettimeofday (struct timeval *tv)
 127 {
 128   int ret;
 129
 130   assert (tv);
 131
 132   if (!(ret = gettimeofday (&recent_time, NULL)))
 133     {
 134       /* init... */
 135       if (!timers_inited)
 136         {
 137           relative_time_base = last_recent_time = recent_time;
 138           timers_inited = 1;
 139         }
 140       /* avoid copy if user passed recent_time pointer.. */
 141       if (tv != &recent_time)
 142         *tv = recent_time;
 143       return 0;
 144     }
 145   return ret;
 146 }
 147
 148 static int
 149 quagga_get_relative (struct timeval *tv)
 150 {
 151   int ret;
 152
 153 #ifdef HAVE_CLOCK_MONOTONIC
 154   {
 155     struct timespec tp;
 156     if (!(ret = clock_gettime (CLOCK_MONOTONIC, &tp)))
 157       {
 158         relative_time.tv_sec = tp.tv_sec;
 159         relative_time.tv_usec = tp.tv_nsec / 1000;
 160       }
 161   }
 162 #elif defined(__APPLE__)
 163   {
 164     uint64_t ticks;
 165     uint64_t useconds;
 166     static mach_timebase_info_data_t timebase_info;
 167
 168     ticks = mach_absolute_time();
 169     if (timebase_info.denom == 0)
 170       mach_timebase_info(&timebase_info);
 171
 172     useconds = ticks * timebase_info.numer / timebase_info.denom / 1000;
 173     relative_time.tv_sec = useconds / 1000000;
 174     relative_time.tv_usec = useconds % 1000000;
 175
 176     return 0;
 177   }
 178 #else /* !HAVE_CLOCK_MONOTONIC && !__APPLE__ */
 179   if (!(ret = quagga_gettimeofday (&recent_time)))
 180     quagga_gettimeofday_relative_adjust();
 181 #endif /* HAVE_CLOCK_MONOTONIC */
 182
 183   if (tv)
 184     *tv = relative_time;
 185
 186   return ret;
 187 }
 188
 189 /* Get absolute time stamp, but in terms of the internal timer
 190  * Could be wrong, but at least won't go back.
 191  */
 192 static void
 193 quagga_real_stabilised (struct timeval *tv)
 194 {
 195   *tv = relative_time_base;
 196   tv->tv_sec += relative_time.tv_sec;
 197   tv->tv_usec += relative_time.tv_usec;
 198   *tv = timeval_adjust (*tv);
 199 }
 200
 201 /* Exported Quagga timestamp function.
 202  * Modelled on POSIX clock_gettime.
 203  */
 204 int
 205 quagga_gettime (enum quagga_clkid clkid, struct timeval *tv)
 206 {
 207   switch (clkid)
 208     {
 209       case QUAGGA_CLK_REALTIME:
 210         return quagga_gettimeofday (tv);
 211       case QUAGGA_CLK_MONOTONIC:
 212         return quagga_get_relative (tv);
 213       case QUAGGA_CLK_REALTIME_STABILISED:
 214         quagga_real_stabilised (tv);
 215         return 0;
 216       default:
 217         errno = EINVAL;
 218         return -1;
 219     }
 220 }
 221
 222 /* time_t value in terms of stabilised absolute time.
 223  * replacement for POSIX time()
 224  */
 225 time_t
 226 quagga_time (time_t *t)
 227 {
 228   struct timeval tv;
 229   quagga_real_stabilised (&tv);
 230   if (t)
 231     *t = tv.tv_sec;
 232   return tv.tv_sec;
 233 }
 234
 235 /* Public export of recent_relative_time by value */
 236 struct timeval
 237 recent_relative_time (void)
 238 {
 239   return relative_time;
 240 }
 241
 242 static unsigned int
 243 cpu_record_hash_key (struct cpu_thread_history *a)
 244 {
 245   return (uintptr_t) a->func;
 246 }
 247
 248 static int
 249 cpu_record_hash_cmp (const struct cpu_thread_history *a,
 250                      const struct cpu_thread_history *b)
 251 {
 252   return a->func == b->func;
 253 }
 254
 255 static void *
 256 cpu_record_hash_alloc (struct cpu_thread_history *a)
 257 {
 258   struct cpu_thread_history *new;
 259   new = XCALLOC (MTYPE_THREAD_STATS, sizeof (struct cpu_thread_history));
 260   new->func = a->func;
 261   new->funcname = a->funcname;
 262   return new;
 263 }
 264
 265 static void
 266 cpu_record_hash_free (void *a)
 267 {
 268   struct cpu_thread_history *hist = a;
 269
 270   XFREE (MTYPE_THREAD_STATS, hist);
 271 }
 272
 273 static void
 274 vty_out_cpu_thread_history(struct vty* vty,
 275                            struct cpu_thread_history *a)
 276 {
 277 #ifdef HAVE_RUSAGE
 278   vty_out(vty, "%7ld.%03ld %9d %8ld %9ld %8ld %9ld",
 279           a->cpu.total/1000, a->cpu.total%1000, a->total_calls,
 280           a->cpu.total/a->total_calls, a->cpu.max,
 281           a->real.total/a->total_calls, a->real.max);
 282 #else
 283   vty_out(vty, "%7ld.%03ld %9d %8ld %9ld",
 284           a->real.total/1000, a->real.total%1000, a->total_calls,
 285           a->real.total/a->total_calls, a->real.max);
 286 #endif
 287   vty_out(vty, " %c%c%c%c%c%c %s%s",
 288           a->types & (1 << THREAD_READ) ? 'R':' ',
 289           a->types & (1 << THREAD_WRITE) ? 'W':' ',
 290           a->types & (1 << THREAD_TIMER) ? 'T':' ',
 291           a->types & (1 << THREAD_EVENT) ? 'E':' ',
 292           a->types & (1 << THREAD_EXECUTE) ? 'X':' ',
 293           a->types & (1 << THREAD_BACKGROUND) ? 'B' : ' ',
 294           a->funcname, VTY_NEWLINE);
 295 }
 296
 297 static void
 298 cpu_record_hash_print(struct hash_backet *bucket,
 299                       void *args[])
 300 {
 301   struct cpu_thread_history *totals = args[0];
 302   struct vty *vty = args[1];
 303   thread_type *filter = args[2];
 304   struct cpu_thread_history *a = bucket->data;
 305
 306   a = bucket->data;
 307   if ( !(a->types & *filter) )
 308        return;
 309   vty_out_cpu_thread_history(vty,a);
 310   totals->total_calls += a->total_calls;
 311   totals->real.total += a->real.total;
 312   if (totals->real.max < a->real.max)
 313     totals->real.max = a->real.max;
 314 #ifdef HAVE_RUSAGE
 315   totals->cpu.total += a->cpu.total;
 316   if (totals->cpu.max < a->cpu.max)
 317     totals->cpu.max = a->cpu.max;
 318 #endif
 319 }
 320
 321 static void
 322 cpu_record_print(struct vty *vty, thread_type filter)
 323 {
 324   struct cpu_thread_history tmp;
 325   void *args[3] = {&tmp, vty, &filter};
 326
 327   memset(&tmp, 0, sizeof tmp);
 328   tmp.funcname = "TOTAL";
 329   tmp.types = filter;
 330
 331 #ifdef HAVE_RUSAGE
 332   vty_out(vty, "%21s %18s %18s%s",
 333           "", "CPU (user+system):", "Real (wall-clock):", VTY_NEWLINE);
 334 #endif
 335   vty_out(vty, "Runtime(ms)   Invoked Avg uSec Max uSecs");
 336 #ifdef HAVE_RUSAGE
 337   vty_out(vty, " Avg uSec Max uSecs");
 338 #endif
 339   vty_out(vty, "  Type  Thread%s", VTY_NEWLINE);
 340   hash_iterate(cpu_record,
 341                (void(*)(struct hash_backet*,void*))cpu_record_hash_print,
 342                args);
 343
 344   if (tmp.total_calls > 0)
 345     vty_out_cpu_thread_history(vty, &tmp);
 346 }
 347
 348 DEFUN(show_thread_cpu,
 349       show_thread_cpu_cmd,
 350       "show thread cpu [FILTER]",
 351       SHOW_STR
 352       "Thread information\n"
 353       "Thread CPU usage\n"
 354       "Display filter (rwtexb)\n")
 355 {
 356   int i = 0;
 357   thread_type filter = (thread_type) -1U;
 358
 359   if (argc > 0)
 360     {
 361       filter = 0;
 362       while (argv[0][i] != '\0')
 363         {
 364           switch ( argv[0][i] )
 365             {
 366             case 'r':
 367             case 'R':
 368               filter |= (1 << THREAD_READ);
 369               break;
 370             case 'w':
 371             case 'W':
 372               filter |= (1 << THREAD_WRITE);
 373               break;
 374             case 't':
 375             case 'T':
 376               filter |= (1 << THREAD_TIMER);
 377               break;
 378             case 'e':
 379             case 'E':
 380               filter |= (1 << THREAD_EVENT);
 381               break;
 382             case 'x':
 383             case 'X':
 384               filter |= (1 << THREAD_EXECUTE);
 385               break;
 386             case 'b':
 387             case 'B':
 388               filter |= (1 << THREAD_BACKGROUND);
 389               break;
 390             default:
 391               break;
 392             }
 393           ++i;
 394         }
 395       if (filter == 0)
 396         {
 397           vty_out(vty, "Invalid filter \"%s\" specified,"
 398                   " must contain at least one of 'RWTEXB'%s",
 399                   argv[0], VTY_NEWLINE);
 400           return CMD_WARNING;
 401         }
 402     }
 403
 404   cpu_record_print(vty, filter);
 405   return CMD_SUCCESS;
 406 }
 407
 408 static void
 409 cpu_record_hash_clear (struct hash_backet *bucket,
 410                       void *args)
 411 {
 412   thread_type *filter = args;
 413   struct cpu_thread_history *a = bucket->data;
 414
 415   a = bucket->data;
 416   if ( !(a->types & *filter) )
 417        return;
 418
 419   hash_release (cpu_record, bucket->data);
 420 }
 421
 422 static void
 423 cpu_record_clear (thread_type filter)
 424 {
 425   thread_type *tmp = &filter;
 426   hash_iterate (cpu_record,
 427                 (void (*) (struct hash_backet*,void*)) cpu_record_hash_clear,
 428                 tmp);
 429 }
 430
 431 DEFUN(clear_thread_cpu,
 432       clear_thread_cpu_cmd,
 433       "clear thread cpu [FILTER]",
 434       "Clear stored data\n"
 435       "Thread information\n"
 436       "Thread CPU usage\n"
 437       "Display filter (rwtexb)\n")
 438 {
 439   int i = 0;
 440   thread_type filter = (thread_type) -1U;
 441
 442   if (argc > 0)
 443     {
 444       filter = 0;
 445       while (argv[0][i] != '\0')
 446         {
 447           switch ( argv[0][i] )
 448             {
 449             case 'r':
 450             case 'R':
 451               filter |= (1 << THREAD_READ);
 452               break;
 453             case 'w':
 454             case 'W':
 455               filter |= (1 << THREAD_WRITE);
 456               break;
 457             case 't':
 458             case 'T':
 459               filter |= (1 << THREAD_TIMER);
 460               break;
 461             case 'e':
 462             case 'E':
 463               filter |= (1 << THREAD_EVENT);
 464               break;
 465             case 'x':
 466             case 'X':
 467               filter |= (1 << THREAD_EXECUTE);
 468               break;
 469             case 'b':
 470             case 'B':
 471               filter |= (1 << THREAD_BACKGROUND);
 472               break;
 473             default:
 474               break;
 475             }
 476           ++i;
 477         }
 478       if (filter == 0)
 479         {
 480           vty_out(vty, "Invalid filter \"%s\" specified,"
 481                   " must contain at least one of 'RWTEXB'%s",
 482                   argv[0], VTY_NEWLINE);
 483           return CMD_WARNING;
 484         }
 485     }
 486
 487   cpu_record_clear (filter);
 488   return CMD_SUCCESS;
 489 }
 490
 491 static int
 492 thread_timer_cmp(void *a, void *b)
 493 {
 494   struct thread *thread_a = a;
 495   struct thread *thread_b = b;
 496
 497   long cmp = timeval_cmp(thread_a->u.sands, thread_b->u.sands);
 498
 499   if (cmp < 0)
 500     return -1;
 501   if (cmp > 0)
 502     return 1;
 503   return 0;
 504 }
 505
 506 static void
 507 thread_timer_update(void *node, int actual_position)
 508 {
 509   struct thread *thread = node;
 510
 511   thread->index = actual_position;
 512 }
 513
 514 /* Allocate new thread master.  */
 515 struct thread_master *
 516 thread_master_create ()
 517 {
 518   struct thread_master *rv;
 519   struct rlimit limit;
 520
 521   getrlimit(RLIMIT_NOFILE, &limit);
 522
 523   if (cpu_record == NULL)
 524     cpu_record
 525       = hash_create ((unsigned int (*) (void *))cpu_record_hash_key,
 526                      (int (*) (const void *, const void *))cpu_record_hash_cmp);
 527
 528   rv = XCALLOC (MTYPE_THREAD_MASTER, sizeof (struct thread_master));
 529   if (rv == NULL)
 530     {
 531       return NULL;
 532     }
 533
 534   rv->fd_limit = (int)limit.rlim_cur;
 535   rv->read = XCALLOC (MTYPE_THREAD, sizeof (struct thread *) * rv->fd_limit);
 536   if (rv->read == NULL)
 537     {
 538       XFREE (MTYPE_THREAD_MASTER, rv);
 539       return NULL;
 540     }
 541
 542   rv->write = XCALLOC (MTYPE_THREAD, sizeof (struct thread *) * rv->fd_limit);
 543   if (rv->write == NULL)
 544     {
 545       XFREE (MTYPE_THREAD, rv->read);
 546       XFREE (MTYPE_THREAD_MASTER, rv);
 547       return NULL;
 548     }
 549
 550   /* Initialize the timer queues */
 551   rv->timer = pqueue_create();
 552   rv->background = pqueue_create();
 553   rv->timer->cmp = rv->background->cmp = thread_timer_cmp;
 554   rv->timer->update = rv->background->update = thread_timer_update;
 555
 556   return rv;
 557 }
 558
 559 /* Add a new thread to the list.  */
 560 static void
 561 thread_list_add (struct thread_list *list, struct thread *thread)
 562 {
 563   thread->next = NULL;
 564   thread->prev = list->tail;
 565   if (list->tail)
 566     list->tail->next = thread;
 567   else
 568     list->head = thread;
 569   list->tail = thread;
 570   list->count++;
 571 }
 572
 573 /* Delete a thread from the list. */
 574 static struct thread *
 575 thread_list_delete (struct thread_list *list, struct thread *thread)
 576 {
 577   if (thread->next)
 578     thread->next->prev = thread->prev;
 579   else
 580     list->tail = thread->prev;
 581   if (thread->prev)
 582     thread->prev->next = thread->next;
 583   else
 584     list->head = thread->next;
 585   thread->next = thread->prev = NULL;
 586   list->count--;
 587   return thread;
 588 }
 589
 590 static void
 591 thread_delete_fd (struct thread **thread_array, struct thread *thread)
 592 {
 593   thread_array[thread->u.fd] = NULL;
 594 }
 595
 596 static void
 597 thread_add_fd (struct thread **thread_array, struct thread *thread)
 598 {
 599   thread_array[thread->u.fd] = thread;
 600 }
 601
 602 /* Move thread to unuse list. */
 603 static void
 604 thread_add_unuse (struct thread_master *m, struct thread *thread)
 605 {
 606   assert (m != NULL && thread != NULL);
 607   assert (thread->next == NULL);
 608   assert (thread->prev == NULL);
 609   assert (thread->type == THREAD_UNUSED);
 610   thread_list_add (&m->unuse, thread);
 611 }
 612
 613 /* Free all unused thread. */
 614 static void
 615 thread_list_free (struct thread_master *m, struct thread_list *list)
 616 {
 617   struct thread *t;
 618   struct thread *next;
 619
 620   for (t = list->head; t; t = next)
 621     {
 622       next = t->next;
 623       XFREE (MTYPE_THREAD, t);
 624       list->count--;
 625       m->alloc--;
 626     }
 627 }
 628
 629 static void
 630 thread_array_free (struct thread_master *m, struct thread **thread_array)
 631 {
 632   struct thread *t;
 633   int index;
 634
 635   for (index = 0; index < m->fd_limit; ++index)
 636     {
 637       t = thread_array[index];
 638       if (t)
 639         {
 640           thread_array[index] = NULL;
 641           XFREE (MTYPE_THREAD, t);
 642           m->alloc--;
 643         }
 644     }
 645   XFREE (MTYPE_THREAD, thread_array);
 646 }
 647
 648 static void
 649 thread_queue_free (struct thread_master *m, struct pqueue *queue)
 650 {
 651   int i;
 652
 653   for (i = 0; i < queue->size; i++)
 654     XFREE(MTYPE_THREAD, queue->array[i]);
 655
 656   m->alloc -= queue->size;
 657   pqueue_delete(queue);
 658 }
 659
 660 /* Stop thread scheduler. */
 661 void
 662 thread_master_free (struct thread_master *m)
 663 {
 664   thread_array_free (m, m->read);
 665   thread_array_free (m, m->write);
 666   thread_queue_free (m, m->timer);
 667   thread_list_free (m, &m->event);
 668   thread_list_free (m, &m->ready);
 669   thread_list_free (m, &m->unuse);
 670   thread_queue_free (m, m->background);
 671
 672   XFREE (MTYPE_THREAD_MASTER, m);
 673
 674   if (cpu_record)
 675     {
 676       hash_clean (cpu_record, cpu_record_hash_free);
 677       hash_free (cpu_record);
 678       cpu_record = NULL;
 679     }
 680 }
 681
 682 /* Thread list is empty or not.  */
 683 static int
 684 thread_empty (struct thread_list *list)
 685 {
 686   return  list->head ? 0 : 1;
 687 }
 688
 689 /* Delete top of the list and return it. */
 690 static struct thread *
 691 thread_trim_head (struct thread_list *list)
 692 {
 693   if (!thread_empty (list))
 694     return thread_list_delete (list, list->head);
 695   return NULL;
 696 }
 697
 698 /* Return remain time in second. */
 699 unsigned long
 700 thread_timer_remain_second (struct thread *thread)
 701 {
 702   quagga_get_relative (NULL);
 703
 704   if (thread->u.sands.tv_sec - relative_time.tv_sec > 0)
 705     return thread->u.sands.tv_sec - relative_time.tv_sec;
 706   else
 707     return 0;
 708 }
 709
 710 struct timeval
 711 thread_timer_remain(struct thread *thread)
 712 {
 713   quagga_get_relative(NULL);
 714
 715   return timeval_subtract(thread->u.sands, relative_time);
 716 }
 717
 718 #define debugargdef  const char *funcname, const char *schedfrom, int fromln
 719 #define debugargpass funcname, schedfrom, fromln
 720
 721 /* Get new thread.  */
 722 static struct thread *
 723 thread_get (struct thread_master *m, u_char type,
 724             int (*func) (struct thread *), void *arg, debugargdef)
 725 {
 726   struct thread *thread = thread_trim_head (&m->unuse);
 727
 728   if (! thread)
 729     {
 730       thread = XCALLOC (MTYPE_THREAD, sizeof (struct thread));
 731       m->alloc++;
 732     }
 733   thread->type = type;
 734   thread->add_type = type;
 735   thread->master = m;
 736   thread->func = func;
 737   thread->arg = arg;
 738   thread->index = -1;
 739
 740   thread->funcname = funcname;
 741   thread->schedfrom = schedfrom;
 742   thread->schedfrom_line = fromln;
 743
 744   return thread;
 745 }
 746
 747 #define fd_copy_fd_set(X) (X)
 748
 749 static int
 750 fd_select (int size, thread_fd_set *read, thread_fd_set *write, thread_fd_set *except, struct timeval *t)
 751 {
 752   return(select(size, read, write, except, t));
 753 }
 754
 755 static int
 756 fd_is_set (int fd, thread_fd_set *fdset)
 757 {
 758   return FD_ISSET (fd, fdset);
 759 }
 760
 761 static int
 762 fd_clear_read_write (int fd, thread_fd_set *fdset)
 763 {
 764   if (!FD_ISSET (fd, fdset))
 765     return 0;
 766
 767   FD_CLR (fd, fdset);
 768   return 1;
 769 }
 770
 771 static struct thread *
 772 funcname_thread_add_read_write (int dir, struct thread_master *m,
 773                  int (*func) (struct thread *), void *arg, int fd,
 774                  debugargdef)
 775 {
 776   struct thread *thread = NULL;
 777   thread_fd_set *fdset = NULL;
 778
 779   if (dir == THREAD_READ)
 780     fdset = &m->readfd;
 781   else
 782     fdset = &m->writefd;
 783
 784   if (FD_ISSET (fd, fdset))
 785     {
 786       zlog (NULL, LOG_WARNING, "There is already %s fd [%d]",
 787             (dir = THREAD_READ) ? "read" : "write", fd);
 788       return NULL;
 789     }
 790
 791   FD_SET (fd, fdset);
 792
 793   thread = thread_get (m, dir, func, arg, debugargpass);
 794   thread->u.fd = fd;
 795   if (dir == THREAD_READ)
 796     thread_add_fd (m->read, thread);
 797   else
 798     thread_add_fd (m->write, thread);
 799
 800   return thread;
 801 }
 802
 803 /* Add new read thread. */
 804 struct thread *
 805 funcname_thread_add_read (struct thread_master *m,
 806                  int (*func) (struct thread *), void *arg, int fd,
 807                  debugargdef)
 808 {
 809   return funcname_thread_add_read_write (THREAD_READ, m, func,
 810                                          arg, fd, debugargpass);
 811 }
 812
 813 /* Add new write thread. */
 814 struct thread *
 815 funcname_thread_add_write (struct thread_master *m,
 816                  int (*func) (struct thread *), void *arg, int fd,
 817                  debugargdef)
 818 {
 819   return funcname_thread_add_read_write (THREAD_WRITE, m, func,
 820                                          arg, fd, debugargpass);
 821 }
 822
 823 static struct thread *
 824 funcname_thread_add_timer_timeval (struct thread_master *m,
 825                                    int (*func) (struct thread *),
 826                                   int type,
 827                                   void *arg,
 828                                   struct timeval *time_relative,
 829                                   debugargdef)
 830 {
 831   struct thread *thread;
 832   struct pqueue *queue;
 833   struct timeval alarm_time;
 834
 835   assert (m != NULL);
 836
 837   assert (type == THREAD_TIMER || type == THREAD_BACKGROUND);
 838   assert (time_relative);
 839
 840   queue = ((type == THREAD_TIMER) ? m->timer : m->background);
 841   thread = thread_get (m, type, func, arg, debugargpass);
 842
 843   /* Do we need jitter here? */
 844   quagga_get_relative (NULL);
 845   alarm_time.tv_sec = relative_time.tv_sec + time_relative->tv_sec;
 846   alarm_time.tv_usec = relative_time.tv_usec + time_relative->tv_usec;
 847   thread->u.sands = timeval_adjust(alarm_time);
 848
 849   pqueue_enqueue(thread, queue);
 850   return thread;
 851 }
 852
 853
 854 /* Add timer event thread. */
 855 struct thread *
 856 funcname_thread_add_timer (struct thread_master *m,
 857                            int (*func) (struct thread *),
 858                            void *arg, long timer,
 859                            debugargdef)
 860 {
 861   struct timeval trel;
 862
 863   assert (m != NULL);
 864
 865   trel.tv_sec = timer;
 866   trel.tv_usec = 0;
 867
 868   return funcname_thread_add_timer_timeval (m, func, THREAD_TIMER, arg,
 869                                             &trel, debugargpass);
 870 }
 871
 872 /* Add timer event thread with "millisecond" resolution */
 873 struct thread *
 874 funcname_thread_add_timer_msec (struct thread_master *m,
 875                                 int (*func) (struct thread *),
 876                                 void *arg, long timer,
 877                                 debugargdef)
 878 {
 879   struct timeval trel;
 880
 881   assert (m != NULL);
 882
 883   trel.tv_sec = timer / 1000;
 884   trel.tv_usec = 1000*(timer % 1000);
 885
 886   return funcname_thread_add_timer_timeval (m, func, THREAD_TIMER,
 887                                             arg, &trel, debugargpass);
 888 }
 889
 890 /* Add timer event thread with "millisecond" resolution */
 891 struct thread *
 892 funcname_thread_add_timer_tv (struct thread_master *m,
 893                               int (*func) (struct thread *),
 894                               void *arg, struct timeval *tv,
 895                               debugargdef)
 896 {
 897   return funcname_thread_add_timer_timeval (m, func, THREAD_TIMER,
 898                                             arg, tv, debugargpass);
 899 }
 900
 901 /* Add a background thread, with an optional millisec delay */
 902 struct thread *
 903 funcname_thread_add_background (struct thread_master *m,
 904                                 int (*func) (struct thread *),
 905                                 void *arg, long delay,
 906                                 debugargdef)
 907 {
 908   struct timeval trel;
 909
 910   assert (m != NULL);
 911
 912   if (delay)
 913     {
 914       trel.tv_sec = delay / 1000;
 915       trel.tv_usec = 1000*(delay % 1000);
 916     }
 917   else
 918     {
 919       trel.tv_sec = 0;
 920       trel.tv_usec = 0;
 921     }
 922
 923   return funcname_thread_add_timer_timeval (m, func, THREAD_BACKGROUND,
 924                                             arg, &trel, debugargpass);
 925 }
 926
 927 /* Add simple event thread. */
 928 struct thread *
 929 funcname_thread_add_event (struct thread_master *m,
 930                   int (*func) (struct thread *), void *arg, int val,
 931                   debugargdef)
 932 {
 933   struct thread *thread;
 934
 935   assert (m != NULL);
 936
 937   thread = thread_get (m, THREAD_EVENT, func, arg, debugargpass);
 938   thread->u.val = val;
 939   thread_list_add (&m->event, thread);
 940
 941   return thread;
 942 }
 943
 944 /* Cancel thread from scheduler. */
 945 void
 946 thread_cancel (struct thread *thread)
 947 {
 948   struct thread_list *list = NULL;
 949   struct pqueue *queue = NULL;
 950   struct thread **thread_array = NULL;
 951
 952   switch (thread->type)
 953     {
 954     case THREAD_READ:
 955       assert (fd_clear_read_write (thread->u.fd, &thread->master->readfd));
 956       thread_array = thread->master->read;
 957       break;
 958     case THREAD_WRITE:
 959       assert (fd_clear_read_write (thread->u.fd, &thread->master->writefd));
 960       thread_array = thread->master->write;
 961       break;
 962     case THREAD_TIMER:
 963       queue = thread->master->timer;
 964       break;
 965     case THREAD_EVENT:
 966       list = &thread->master->event;
 967       break;
 968     case THREAD_READY:
 969       list = &thread->master->ready;
 970       break;
 971     case THREAD_BACKGROUND:
 972       queue = thread->master->background;
 973       break;
 974     default:
 975       return;
 976       break;
 977     }
 978
 979   if (queue)
 980     {
 981       assert(thread->index >= 0);
 982       assert(thread == queue->array[thread->index]);
 983       pqueue_remove_at(thread->index, queue);
 984     }
 985   else if (list)
 986     {
 987       thread_list_delete (list, thread);
 988     }
 989   else if (thread_array)
 990     {
 991       thread_delete_fd (thread_array, thread);
 992     }
 993   else
 994     {
 995       assert(!"Thread should be either in queue or list or array!");
 996     }
 997
 998   thread->type = THREAD_UNUSED;
 999   thread_add_unuse (thread->master, thread);
1000 }
1001
1002 /* Delete all events which has argument value arg. */
1003 unsigned int
1004 thread_cancel_event (struct thread_master *m, void *arg)
1005 {
1006   unsigned int ret = 0;
1007   struct thread *thread;
1008
1009   thread = m->event.head;
1010   while (thread)
1011     {
1012       struct thread *t;
1013
1014       t = thread;
1015       thread = t->next;
1016
1017       if (t->arg == arg)
1018         {
1019           ret++;
1020           thread_list_delete (&m->event, t);
1021           t->type = THREAD_UNUSED;
1022           thread_add_unuse (m, t);
1023         }
1024     }
1025
1026   /* thread can be on the ready list too */
1027   thread = m->ready.head;
1028   while (thread)
1029     {
1030       struct thread *t;
1031
1032       t = thread;
1033       thread = t->next;
1034
1035       if (t->arg == arg)
1036         {
1037           ret++;
1038           thread_list_delete (&m->ready, t);
1039           t->type = THREAD_UNUSED;
1040           thread_add_unuse (m, t);
1041         }
1042     }
1043   return ret;
1044 }
1045
1046 static struct timeval *
1047 thread_timer_wait (struct pqueue *queue, struct timeval *timer_val)
1048 {
1049   if (queue->size)
1050     {
1051       struct thread *next_timer = queue->array[0];
1052       *timer_val = timeval_subtract (next_timer->u.sands, relative_time);
1053       return timer_val;
1054     }
1055   return NULL;
1056 }
1057
1058 static struct thread *
1059 thread_run (struct thread_master *m, struct thread *thread,
1060             struct thread *fetch)
1061 {
1062   *fetch = *thread;
1063   thread->type = THREAD_UNUSED;
1064   thread_add_unuse (m, thread);
1065   return fetch;
1066 }
1067
1068 static int
1069 thread_process_fds_helper (struct thread_master *m, struct thread *thread, thread_fd_set *fdset)
1070 {
1071   thread_fd_set *mfdset = NULL;
1072   struct thread **thread_array;
1073
1074   if (!thread)
1075     return 0;
1076
1077   if (thread->type == THREAD_READ)
1078     {
1079       mfdset = &m->readfd;
1080       thread_array = m->read;
1081     }
1082   else
1083     {
1084       mfdset = &m->writefd;
1085       thread_array = m->write;
1086     }
1087
1088   if (fd_is_set (THREAD_FD (thread), fdset))
1089     {
1090       fd_clear_read_write (THREAD_FD (thread), mfdset);
1091       thread_delete_fd (thread_array, thread);
1092       thread_list_add (&m->ready, thread);
1093       thread->type = THREAD_READY;
1094       return 1;
1095     }
1096   return 0;
1097 }
1098
1099 static int
1100 thread_process_fds (struct thread_master *m, thread_fd_set *rset, thread_fd_set *wset, int num)
1101 {
1102   int ready = 0, index;
1103
1104   for (index = 0; index < m->fd_limit && ready < num; ++index)
1105     {
1106       ready += thread_process_fds_helper (m, m->read[index], rset);
1107       ready += thread_process_fds_helper (m, m->write[index], wset);
1108     }
1109   return num - ready;
1110 }
1111
1112 /* Add all timers that have popped to the ready list. */
1113 static unsigned int
1114 thread_timer_process (struct pqueue *queue, struct timeval *timenow)
1115 {
1116   struct thread *thread;
1117   unsigned int ready = 0;
1118
1119   while (queue->size)
1120     {
1121       thread = queue->array[0];
1122       if (timeval_cmp (*timenow, thread->u.sands) < 0)
1123         return ready;
1124       pqueue_dequeue(queue);
1125       thread->type = THREAD_READY;
1126       thread_list_add (&thread->master->ready, thread);
1127       ready++;
1128     }
1129   return ready;
1130 }
1131
1132 /* process a list en masse, e.g. for event thread lists */
1133 static unsigned int
1134 thread_process (struct thread_list *list)
1135 {
1136   struct thread *thread;
1137   struct thread *next;
1138   unsigned int ready = 0;
1139
1140   for (thread = list->head; thread; thread = next)
1141     {
1142       next = thread->next;
1143       thread_list_delete (list, thread);
1144       thread->type = THREAD_READY;
1145       thread_list_add (&thread->master->ready, thread);
1146       ready++;
1147     }
1148   return ready;
1149 }
1150
1151
1152 /* Fetch next ready thread. */
1153 struct thread *
1154 thread_fetch (struct thread_master *m, struct thread *fetch)
1155 {
1156   struct thread *thread;
1157   thread_fd_set readfd;
1158   thread_fd_set writefd;
1159   thread_fd_set exceptfd;
1160   struct timeval timer_val = { .tv_sec = 0, .tv_usec = 0 };
1161   struct timeval timer_val_bg;
1162   struct timeval *timer_wait = &timer_val;
1163   struct timeval *timer_wait_bg;
1164
1165   while (1)
1166     {
1167       int num = 0;
1168
1169       /* Signals pre-empt everything */
1170       quagga_sigevent_process ();
1171
1172       /* Drain the ready queue of already scheduled jobs, before scheduling
1173        * more.
1174        */
1175       if ((thread = thread_trim_head (&m->ready)) != NULL)
1176         return thread_run (m, thread, fetch);
1177
1178       /* To be fair to all kinds of threads, and avoid starvation, we
1179        * need to be careful to consider all thread types for scheduling
1180        * in each quanta. I.e. we should not return early from here on.
1181        */
1182
1183       /* Normal event are the next highest priority.  */
1184       thread_process (&m->event);
1185
1186       /* Structure copy.  */
1187       readfd = fd_copy_fd_set(m->readfd);
1188       writefd = fd_copy_fd_set(m->writefd);
1189       exceptfd = fd_copy_fd_set(m->exceptfd);
1190
1191       /* Calculate select wait timer if nothing else to do */
1192       if (m->ready.count == 0)
1193         {
1194           quagga_get_relative (NULL);
1195           timer_wait = thread_timer_wait (m->timer, &timer_val);
1196           timer_wait_bg = thread_timer_wait (m->background, &timer_val_bg);
1197
1198           if (timer_wait_bg &&
1199               (!timer_wait || (timeval_cmp (*timer_wait, *timer_wait_bg) > 0)))
1200             timer_wait = timer_wait_bg;
1201         }
1202
1203       num = fd_select (FD_SETSIZE, &readfd, &writefd, &exceptfd, timer_wait);
1204
1205       /* Signals should get quick treatment */
1206       if (num < 0)
1207         {
1208           if (errno == EINTR)
1209             continue; /* signal received - process it */
1210           zlog_warn ("select() error: %s", safe_strerror (errno));
1211           return NULL;
1212         }
1213
1214       /* Check foreground timers.  Historically, they have had higher
1215          priority than I/O threads, so let's push them onto the ready
1216          list in front of the I/O threads. */
1217       quagga_get_relative (NULL);
1218       thread_timer_process (m->timer, &relative_time);
1219
1220       /* Got IO, process it */
1221       if (num > 0)
1222         thread_process_fds (m, &readfd, &writefd, num);
1223
1224 #if 0
1225       /* If any threads were made ready above (I/O or foreground timer),
1226          perhaps we should avoid adding background timers to the ready
1227          list at this time.  If this is code is uncommented, then background
1228          timer threads will not run unless there is nothing else to do. */
1229       if ((thread = thread_trim_head (&m->ready)) != NULL)
1230         return thread_run (m, thread, fetch);
1231 #endif
1232
1233       /* Background timer/events, lowest priority */
1234       thread_timer_process (m->background, &relative_time);
1235
1236       if ((thread = thread_trim_head (&m->ready)) != NULL)
1237         return thread_run (m, thread, fetch);
1238     }
1239 }
1240
1241 unsigned long
1242 thread_consumed_time (RUSAGE_T *now, RUSAGE_T *start, unsigned long *cputime)
1243 {
1244 #ifdef HAVE_RUSAGE
1245   /* This is 'user + sys' time.  */
1246   *cputime = timeval_elapsed (now->cpu.ru_utime, start->cpu.ru_utime) +
1247              timeval_elapsed (now->cpu.ru_stime, start->cpu.ru_stime);
1248 #else
1249   *cputime = 0;
1250 #endif /* HAVE_RUSAGE */
1251   return timeval_elapsed (now->real, start->real);
1252 }
1253
1254 /* We should aim to yield after THREAD_YIELD_TIME_SLOT milliseconds.
1255    Note: we are using real (wall clock) time for this calculation.
1256    It could be argued that CPU time may make more sense in certain
1257    contexts.  The things to consider are whether the thread may have
1258    blocked (in which case wall time increases, but CPU time does not),
1259    or whether the system is heavily loaded with other processes competing
1260    for CPU time.  On balance, wall clock time seems to make sense.
1261    Plus it has the added benefit that gettimeofday should be faster
1262    than calling getrusage. */
1263 int
1264 thread_should_yield (struct thread *thread)
1265 {
1266   quagga_get_relative (NULL);
1267   unsigned long t = timeval_elapsed(relative_time, thread->real);
1268   return ((t > THREAD_YIELD_TIME_SLOT) ? t : 0);
1269 }
1270
1271 void
1272 thread_getrusage (RUSAGE_T *r)
1273 {
1274   quagga_get_relative (NULL);
1275 #ifdef HAVE_RUSAGE
1276   getrusage(RUSAGE_SELF, &(r->cpu));
1277 #endif
1278   r->real = relative_time;
1279
1280 #ifdef HAVE_CLOCK_MONOTONIC
1281   /* quagga_get_relative() only updates recent_time if gettimeofday
1282    * based, not when using CLOCK_MONOTONIC. As we export recent_time
1283    * and guarantee to update it before threads are run...
1284    */
1285   quagga_gettimeofday(&recent_time);
1286 #endif /* HAVE_CLOCK_MONOTONIC */
1287 }
1288
1289 struct thread *thread_current = NULL;
1290
1291 /* We check thread consumed time. If the system has getrusage, we'll
1292    use that to get in-depth stats on the performance of the thread in addition
1293    to wall clock time stats from gettimeofday. */
1294 void
1295 thread_call (struct thread *thread)
1296 {
1297   unsigned long realtime, cputime;
1298   RUSAGE_T before, after;
1299
1300  /* Cache a pointer to the relevant cpu history thread, if the thread
1301   * does not have it yet.
1302   *
1303   * Callers submitting 'dummy threads' hence must take care that
1304   * thread->cpu is NULL
1305   */
1306   if (!thread->hist)
1307     {
1308       struct cpu_thread_history tmp;
1309
1310       tmp.func = thread->func;
1311       tmp.funcname = thread->funcname;
1312
1313       thread->hist = hash_get (cpu_record, &tmp,
1314                     (void * (*) (void *))cpu_record_hash_alloc);
1315     }
1316
1317   GETRUSAGE (&before);
1318   thread->real = before.real;
1319
1320   thread_current = thread;
1321   (*thread->func) (thread);
1322   thread_current = NULL;
1323
1324   GETRUSAGE (&after);
1325
1326   realtime = thread_consumed_time (&after, &before, &cputime);
1327   thread->hist->real.total += realtime;
1328   if (thread->hist->real.max < realtime)
1329     thread->hist->real.max = realtime;
1330 #ifdef HAVE_RUSAGE
1331   thread->hist->cpu.total += cputime;
1332   if (thread->hist->cpu.max < cputime)
1333     thread->hist->cpu.max = cputime;
1334 #endif
1335
1336   ++(thread->hist->total_calls);
1337   thread->hist->types |= (1 << thread->add_type);
1338
1339 #ifdef CONSUMED_TIME_CHECK
1340   if (realtime > CONSUMED_TIME_CHECK)
1341     {
1342       /*
1343        * We have a CPU Hog on our hands.
1344        * Whinge about it now, so we're aware this is yet another task
1345        * to fix.
1346        */
1347       zlog_warn ("SLOW THREAD: task %s (%lx) ran for %lums (cpu time %lums)",
1348                  thread->funcname,
1349                  (unsigned long) thread->func,
1350                  realtime/1000, cputime/1000);
1351     }
1352 #endif /* CONSUMED_TIME_CHECK */
1353 }
1354
1355 /* Execute thread */
1356 struct thread *
1357 funcname_thread_execute (struct thread_master *m,
1358                 int (*func)(struct thread *),
1359                 void *arg,
1360                 int val,
1361                 debugargdef)
1362 {
1363   struct thread dummy;
1364
1365   memset (&dummy, 0, sizeof (struct thread));
1366
1367   dummy.type = THREAD_EVENT;
1368   dummy.add_type = THREAD_EXECUTE;
1369   dummy.master = NULL;
1370   dummy.func = func;
1371   dummy.arg = arg;
1372   dummy.u.val = val;
1373
1374   dummy.funcname = funcname;
1375   dummy.schedfrom = schedfrom;
1376   dummy.schedfrom_line = fromln;
1377
1378   thread_call (&dummy);
1379
1380   return NULL;
1381 }