OpenDNSSEC-signer  1.4.3
worker.c
Go to the documentation of this file.
1 /*
2  * $Id: worker.c 7342 2013-10-09 08:54:41Z matthijs $
3  *
4  * Copyright (c) 2009 NLNet Labs. All rights reserved.
5  *
6  * Redistribution and use in source and binary forms, with or without
7  * modification, are permitted provided that the following conditions
8  * are met:
9  * 1. Redistributions of source code must retain the above copyright
10  * notice, this list of conditions and the following disclaimer.
11  * 2. Redistributions in binary form must reproduce the above copyright
12  * notice, this list of conditions and the following disclaimer in the
13  * documentation and/or other materials provided with the distribution.
14  *
15  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
16  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
17  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
18  * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
19  * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
20  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE
21  * GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
22  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
23  * IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
25  * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26  *
27  */
28 
34 #include "daemon/engine.h"
35 #include "daemon/worker.h"
36 #include "shared/allocator.h"
37 #include "shared/duration.h"
38 #include "shared/hsm.h"
39 #include "shared/locks.h"
40 #include "shared/log.h"
41 #include "shared/status.h"
42 #include "signer/tools.h"
43 #include "signer/zone.h"
44 
45 #include <time.h> /* time() */
46 
48  { WORKER_WORKER, "worker" },
49  { WORKER_DRUDGER, "drudger" },
50  { 0, NULL }
51 };
52 
53 
58 static const char*
59 worker2str(worker_id type)
60 {
61  ods_lookup_table *lt = ods_lookup_by_id(worker_str, type);
62  if (lt) {
63  return lt->name;
64  }
65  return NULL;
66 }
67 
68 
74 worker_create(allocator_type* allocator, int num, worker_id type)
75 {
76  worker_type* worker;
77  if (!allocator) {
78  return NULL;
79  }
80  worker = (worker_type*) allocator_alloc(allocator, sizeof(worker_type));
81  if (!worker) {
82  return NULL;
83  }
84  ods_log_debug("[%s[%i]] create", worker2str(type), num+1);
85  lock_basic_init(&worker->worker_lock);
86  lock_basic_set(&worker->worker_alarm);
87  lock_basic_lock(&worker->worker_lock);
88  worker->allocator = allocator;
89  worker->thread_num = num +1;
90  worker->engine = NULL;
91  worker->task = NULL;
92  worker->working_with = TASK_NONE;
93  worker->need_to_exit = 0;
94  worker->type = type;
95  worker->clock_in = 0;
96  worker->jobs_appointed = 0;
97  worker->jobs_completed = 0;
98  worker->jobs_failed = 0;
99  worker->sleeping = 0;
100  worker->waiting = 0;
101  lock_basic_unlock(&worker->worker_lock);
102  return worker;
103 }
104 
105 
110 static void
111 worker_working_with(worker_type* worker, task_id with, task_id next,
112  const char* str, const char* name, task_id* what, time_t* when)
113 {
114  worker->working_with = with;
115  ods_log_verbose("[%s[%i]] %s zone %s", worker2str(worker->type),
116  worker->thread_num, str, name);
117  *what = next;
118  *when = time_now();
119  return;
120 }
121 
122 
127 static int
128 worker_fulfilled(worker_type* worker)
129 {
130  int ret = 0;
131  ret = (worker->jobs_completed + worker->jobs_failed) ==
132  worker->jobs_appointed;
133  return ret;
134 }
135 
136 
141 static void
142 worker_clear_jobs(worker_type* worker)
143 {
144  ods_log_assert(worker);
145  lock_basic_lock(&worker->worker_lock);
146  worker->jobs_appointed = 0;
147  worker->jobs_completed = 0;
148  worker->jobs_failed = 0;
149  lock_basic_unlock(&worker->worker_lock);
150  return;
151 }
152 
153 
158 static void
159 worker_queue_rrset(worker_type* worker, fifoq_type* q, rrset_type* rrset)
160 {
162  int tries = 0;
163  ods_log_assert(worker);
164  ods_log_assert(q);
165  ods_log_assert(rrset);
166 
167  lock_basic_lock(&q->q_lock);
168  status = fifoq_push(q, (void*) rrset, worker, &tries);
169  while (status == ODS_STATUS_UNCHANGED) {
170  tries++;
171  if (worker->need_to_exit) {
173  return;
174  }
181  lock_basic_sleep(&q->q_nonfull, &q->q_lock, 5);
182  status = fifoq_push(q, (void*) rrset, worker, &tries);
183  }
185 
186  ods_log_assert(status == ODS_STATUS_OK);
187  lock_basic_lock(&worker->worker_lock);
188  worker->jobs_appointed += 1;
189  lock_basic_unlock(&worker->worker_lock);
190  return;
191 }
192 
193 
198 static void
199 worker_queue_domain(worker_type* worker, fifoq_type* q, domain_type* domain)
200 {
201  rrset_type* rrset = NULL;
202  denial_type* denial = NULL;
203  ods_log_assert(worker);
204  ods_log_assert(q);
205  ods_log_assert(domain);
206  rrset = domain->rrsets;
207  while (rrset) {
208  worker_queue_rrset(worker, q, rrset);
209  rrset = rrset->next;
210  }
211  denial = (denial_type*) domain->denial;
212  if (denial && denial->rrset) {
213  worker_queue_rrset(worker, q, denial->rrset);
214  }
215  return;
216 }
217 
218 
223 static void
224 worker_queue_zone(worker_type* worker, fifoq_type* q, zone_type* zone)
225 {
226  ldns_rbnode_t* node = LDNS_RBTREE_NULL;
227  domain_type* domain = NULL;
228  ods_log_assert(worker);
229  ods_log_assert(q);
230  ods_log_assert(zone);
231  worker_clear_jobs(worker);
232  if (!zone->db || !zone->db->domains) {
233  return;
234  }
235  if (zone->db->domains->root != LDNS_RBTREE_NULL) {
236  node = ldns_rbtree_first(zone->db->domains);
237  }
238  while (node && node != LDNS_RBTREE_NULL) {
239  domain = (domain_type*) node->data;
240  worker_queue_domain(worker, q, domain);
241  node = ldns_rbtree_next(node);
242  }
243  return;
244 }
245 
246 
251 static ods_status
252 worker_check_jobs(worker_type* worker, task_type* task)
253 {
254  ods_log_assert(worker);
255  ods_log_assert(task);
256  lock_basic_lock(&worker->worker_lock);
257  if (worker->jobs_failed) {
258  ods_log_error("[%s[%i]] sign zone %s failed: %u RRsets failed",
259  worker2str(worker->type), worker->thread_num,
260  task_who2str(task), worker->jobs_failed);
261  lock_basic_unlock(&worker->worker_lock);
262  return ODS_STATUS_ERR;
263  } else if (worker->jobs_completed != worker->jobs_appointed) {
264  ods_log_error("[%s[%i]] sign zone %s failed: processed %u of %u "
265  "RRsets", worker2str(worker->type), worker->thread_num,
266  task_who2str(task), worker->jobs_completed,
267  worker->jobs_appointed);
268  lock_basic_unlock(&worker->worker_lock);
269  return ODS_STATUS_ERR;
270  } else if (worker->need_to_exit) {
271  ods_log_debug("[%s[%i]] sign zone %s failed: worker needs to exit",
272  worker2str(worker->type), worker->thread_num, task_who2str(task));
273  lock_basic_unlock(&worker->worker_lock);
274  return ODS_STATUS_ERR;
275  } else {
276  ods_log_debug("[%s[%i]] sign zone %s ok: %u of %u RRsets "
277  "succeeded", worker2str(worker->type), worker->thread_num,
278  task_who2str(task), worker->jobs_completed,
279  worker->jobs_appointed);
280  ods_log_assert(worker->jobs_appointed == worker->jobs_completed);
281  }
282  lock_basic_unlock(&worker->worker_lock);
283  return ODS_STATUS_OK;
284 }
285 
286 
291 static void
292 worker_perform_task(worker_type* worker)
293 {
294  engine_type* engine = NULL;
295  zone_type* zone = NULL;
296  task_type* task = NULL;
297  task_id what = TASK_NONE;
298  time_t when = 0;
299  time_t never = (3600*24*365);
300  ods_status status = ODS_STATUS_OK;
301  int backup = 0;
302  time_t start = 0;
303  time_t end = 0;
304 
305  if (!worker || !worker->task || !worker->task->zone || !worker->engine) {
306  return;
307  }
308  engine = (engine_type*) worker->engine;
309  task = (task_type*) worker->task;
310  zone = (zone_type*) worker->task->zone;
311  ods_log_debug("[%s[%i]] perform task %s for zone %s at %u",
312  worker2str(worker->type), worker->thread_num, task_what2str(task->what),
313  task_who2str(task), (uint32_t) worker->clock_in);
314  /* do what you have been told to do */
315  switch (task->what) {
316  case TASK_SIGNCONF:
317  /* perform 'load signconf' task */
318  worker_working_with(worker, TASK_SIGNCONF, TASK_READ,
319  "configure", task_who2str(task), &what, &when);
320  status = tools_signconf(zone);
321  if (status == ODS_STATUS_UNCHANGED) {
322  if (!zone->signconf->last_modified) {
323  ods_log_debug("[%s[%i]] no signconf.xml for zone %s yet",
324  worker2str(worker->type), worker->thread_num,
325  task_who2str(task));
326  status = ODS_STATUS_ERR;
327  }
328  }
329  if (status == ODS_STATUS_UNCHANGED) {
330  if (task->halted != TASK_NONE && task->halted != TASK_SIGNCONF) {
331  goto task_perform_continue;
332  }
333  status = ODS_STATUS_OK;
334  } else if (status == ODS_STATUS_OK) {
335  task->interrupt = TASK_NONE;
336  task->halted = TASK_NONE;
337  } else {
338  if (task->halted == TASK_NONE) {
339  goto task_perform_fail;
340  }
341  goto task_perform_continue;
342  }
343  /* break; */
344  case TASK_READ:
345  /* perform 'read input adapter' task */
346  worker_working_with(worker, TASK_READ, TASK_SIGN,
347  "read", task_who2str(task), &what, &when);
348  task->what = TASK_READ;
349  if (!zone->signconf->last_modified) {
350  ods_log_debug("[%s[%i]] no signconf.xml for zone %s yet",
351  worker2str(worker->type), worker->thread_num,
352  task_who2str(task));
353  status = ODS_STATUS_ERR;
354  } else {
355  lhsm_check_connection((void*)engine);
356  status = tools_input(zone);
357  }
358 
359  if (status == ODS_STATUS_UNCHANGED) {
360  ods_log_verbose("[%s[%i]] zone %s unsigned data not changed, "
361  "continue", worker2str(worker->type), worker->thread_num,
362  task_who2str(task));
363  status = ODS_STATUS_OK;
364  }
365  if (status == ODS_STATUS_OK) {
366  if (task->interrupt > TASK_SIGNCONF) {
367  task->interrupt = TASK_NONE;
368  task->halted = TASK_NONE;
369  }
370  } else {
371  if (task->halted == TASK_NONE) {
372  goto task_perform_fail;
373  }
374  goto task_perform_continue;
375  }
376  /* break; */
377  case TASK_SIGN:
378  /* perform 'sign' task */
379  worker_working_with(worker, TASK_SIGN, TASK_WRITE,
380  "sign", task_who2str(task), &what, &when);
381  task->what = TASK_SIGN;
382  status = zone_update_serial(zone);
383  if (status == ODS_STATUS_OK) {
384  if (task->interrupt > TASK_SIGNCONF) {
385  task->interrupt = TASK_NONE;
386  task->halted = TASK_NONE;
387  }
388  } else {
389  ods_log_error("[%s[%i]] unable to sign zone %s: "
390  "failed to increment serial",
391  worker2str(worker->type), worker->thread_num,
392  task_who2str(task));
393  if (task->halted == TASK_NONE) {
394  goto task_perform_fail;
395  }
396  goto task_perform_continue;
397  }
398 
399  /* start timer */
400  start = time(NULL);
401  if (zone->stats) {
403  if (!zone->stats->start_time) {
404  zone->stats->start_time = start;
405  }
406  zone->stats->sig_count = 0;
407  zone->stats->sig_soa_count = 0;
408  zone->stats->sig_reuse = 0;
409  zone->stats->sig_time = 0;
411  }
412  /* check the HSM connection before queuing sign operations */
413  lhsm_check_connection((void*)engine);
414  /* prepare keys */
415  status = zone_prepare_keys(zone);
416  if (status == ODS_STATUS_OK) {
417  /* queue menial, hard signing work */
418  worker_queue_zone(worker, engine->signq, zone);
419  ods_log_deeebug("[%s[%i]] wait until drudgers are finished "
420  "signing zone %s", worker2str(worker->type),
421  worker->thread_num, task_who2str(task));
422  /* sleep until work is done */
423  worker_sleep_unless(worker, 0);
424  }
425  /* stop timer */
426  end = time(NULL);
427  /* check status and jobs */
428  if (status == ODS_STATUS_OK) {
429  status = worker_check_jobs(worker, task);
430  }
431  worker_clear_jobs(worker);
432  if (status == ODS_STATUS_OK && zone->stats) {
434  zone->stats->sig_time = (end-start);
436  }
437  if (status != ODS_STATUS_OK) {
438  if (task->halted == TASK_NONE) {
439  goto task_perform_fail;
440  }
441  goto task_perform_continue;
442  } else {
443  if (task->interrupt > TASK_SIGNCONF) {
444  task->interrupt = TASK_NONE;
445  task->halted = TASK_NONE;
446  }
447  }
448  /* break; */
449  case TASK_WRITE:
450  /* perform 'write to output adapter' task */
451  worker_working_with(worker, TASK_WRITE, TASK_SIGN,
452  "write", task_who2str(task), &what, &when);
453  task->what = TASK_WRITE;
454  status = tools_output(zone, engine);
455  if (status == ODS_STATUS_OK) {
456  if (task->interrupt > TASK_SIGNCONF) {
457  task->interrupt = TASK_NONE;
458  task->halted = TASK_NONE;
459  }
460  } else {
461  /* clear signatures? */
462  if (task->halted == TASK_NONE) {
463  goto task_perform_fail;
464  }
465  goto task_perform_continue;
466  }
467  zone->db->is_processed = 1;
468  if (zone->signconf &&
470  what = TASK_SIGN;
471  when = worker->clock_in +
473  } else {
474  ods_log_error("[%s[%i]] unable to retrieve resign interval "
475  "for zone %s: duration2time() failed",
476  worker2str(worker->type), worker->thread_num,
477  task_who2str(task));
478  ods_log_info("[%s[%i]] defaulting to 1H resign interval for "
479  "zone %s", worker2str(worker->type), worker->thread_num,
480  task_who2str(task));
481  what = TASK_SIGN;
482  when = worker->clock_in + 3600;
483  }
484  backup = 1;
485  break;
486  case TASK_NONE:
487  worker->working_with = TASK_NONE;
488  /* no task */
489  ods_log_warning("[%s[%i]] none task for zone %s",
490  worker2str(worker->type), worker->thread_num,
491  task_who2str(task));
492  when = time_now() + never;
493  break;
494  default:
495  worker->working_with = TASK_NONE;
496  /* unknown task */
497  ods_log_warning("[%s[%i]] unknown task, trying full sign zone %s",
498  worker2str(worker->type), worker->thread_num,
499  task_who2str(task));
500  what = TASK_SIGNCONF;
501  when = time_now();
502  break;
503  }
504  /* no error */
505  task->backoff = 0;
506  if (task->interrupt != TASK_NONE && task->interrupt != what) {
507  ods_log_debug("[%s[%i]] interrupt task %s for zone %s",
508  worker2str(worker->type), worker->thread_num,
509  task_what2str(what), task_who2str(task));
510  task->halted = what;
511  task->halted_when = when;
512  task->what = task->interrupt;
513  task->when = time_now();
514  } else {
515  ods_log_debug("[%s[%i]] next task %s for zone %s",
516  worker2str(worker->type), worker->thread_num,
517  task_what2str(what), task_who2str(task));
518  task->what = what;
519  task->when = when;
520  task->interrupt = TASK_NONE;
521  task->halted = TASK_NONE;
522  task->halted_when = 0;
523  }
524  /* backup the last successful run */
525  if (backup) {
526  status = zone_backup2(zone);
527  if (status != ODS_STATUS_OK) {
528  ods_log_warning("[%s[%i]] unable to backup zone %s: %s",
529  worker2str(worker->type), worker->thread_num,
530  task_who2str(task), ods_status2str(status));
531  /* just a warning */
532  status = ODS_STATUS_OK;
533  }
534  backup = 0;
535  }
536  return;
537 
538 task_perform_fail:
539  if (status != ODS_STATUS_XFR_NOT_READY) {
540  /* other statuses is critical, and we know it is not ODS_STATUS_OK */
541  ods_log_crit("[%s[%i]] CRITICAL: failed to sign zone %s: %s",
542  worker2str(worker->type), worker->thread_num,
543  task_who2str(task), ods_status2str(status));
544  }
545  /* in case of failure, also mark zone processed (for single run usage) */
546  zone->db->is_processed = 1;
547  if (task->backoff) {
548  task->backoff *= 2;
549  } else {
550  task->backoff = 60;
551  }
552  if (task->backoff > ODS_SE_MAX_BACKOFF) {
553  task->backoff = ODS_SE_MAX_BACKOFF;
554  }
555  ods_log_info("[%s[%i]] backoff task %s for zone %s with %u seconds",
556  worker2str(worker->type), worker->thread_num,
557  task_what2str(task->what), task_who2str(task), task->backoff);
558  task->when = time_now() + task->backoff;
559  return;
560 
561 task_perform_continue:
562  ods_log_info("[%s[%i]] continue task %s for zone %s",
563  worker2str(worker->type), worker->thread_num,
564  task_what2str(task->halted), task_who2str(task));
565  task->what = task->halted;
566  task->when = task->halted_when;
567  task->interrupt = TASK_NONE;
568  task->halted = TASK_NONE;
569  task->halted_when = 0;
570  return;
571 }
572 
573 
578 static void
579 worker_work(worker_type* worker)
580 {
581  time_t now = 0;
582  time_t timeout = 1;
583  engine_type* engine = NULL;
584  zone_type* zone = NULL;
585  ods_status status = ODS_STATUS_OK;
586 
587  ods_log_assert(worker);
588  ods_log_assert(worker->type == WORKER_WORKER);
589 
590  engine = (engine_type*) worker->engine;
591  while (worker->need_to_exit == 0) {
592  ods_log_debug("[%s[%i]] report for duty", worker2str(worker->type),
593  worker->thread_num);
594  now = time_now();
596  worker->task = schedule_pop_task(engine->taskq);
597  if (worker->task) {
598  worker->working_with = worker->task->what;
600  zone = (zone_type*) worker->task->zone;
601 
602  lock_basic_lock(&zone->zone_lock);
603  ods_log_debug("[%s[%i]] start working on zone %s",
604  worker2str(worker->type), worker->thread_num, zone->name);
605  worker->clock_in = time(NULL);
606  worker_perform_task(worker);
607  zone->task = worker->task;
608  ods_log_debug("[%s[%i]] finished working on zone %s",
609  worker2str(worker->type), worker->thread_num, zone->name);
610 
612  worker->task = NULL;
613  worker->working_with = TASK_NONE;
614  status = schedule_task(engine->taskq, zone->task, 1);
615  if (status != ODS_STATUS_OK) {
616  ods_log_error("[%s[%i]] unable to schedule task for zone %s: "
617  "%s", worker2str(worker->type), worker->thread_num,
618  zone->name, ods_status2str(status));
619  }
622  timeout = 1;
624  lock_basic_lock(&engine->signal_lock);
625  if (engine->need_to_reload) {
626  lock_basic_alarm(&engine->signal_cond);
627  }
628  lock_basic_unlock(&engine->signal_lock);
629 
630  } else {
631  ods_log_debug("[%s[%i]] nothing to do", worker2str(worker->type),
632  worker->thread_num);
633  worker->task = schedule_get_first_task(engine->taskq);
635  if (worker->task && !engine->taskq->loading) {
636  timeout = (worker->task->when - now);
637  } else {
638  timeout *= 2;
639  }
640  if (timeout > ODS_SE_MAX_BACKOFF) {
641  timeout = ODS_SE_MAX_BACKOFF;
642  }
643  worker->task = NULL;
644  worker_sleep(worker, timeout);
645  }
646  }
647  return;
648 }
649 
650 
655 static void
656 worker_drudge(worker_type* worker)
657 {
658  engine_type* engine = NULL;
659  zone_type* zone = NULL;
660  task_type* task = NULL;
661  rrset_type* rrset = NULL;
662  ods_status status = ODS_STATUS_OK;
663  worker_type* superior = NULL;
664  hsm_ctx_t* ctx = NULL;
665 
666  ods_log_assert(worker);
667  ods_log_assert(worker->engine);
668  ods_log_assert(worker->type == WORKER_DRUDGER);
669 
670  engine = (engine_type*) worker->engine;
671  while (worker->need_to_exit == 0) {
672  ods_log_deeebug("[%s[%i]] report for duty", worker2str(worker->type),
673  worker->thread_num);
674  /* initialize */
675  superior = NULL;
676  zone = NULL;
677  task = NULL;
678  /* get item */
679  lock_basic_lock(&engine->signq->q_lock);
680  rrset = (rrset_type*) fifoq_pop(engine->signq, &superior);
681  if (!rrset) {
682  ods_log_deeebug("[%s[%i]] nothing to do, wait",
683  worker2str(worker->type), worker->thread_num);
691  &engine->signq->q_lock, 0);
692  rrset = (rrset_type*) fifoq_pop(engine->signq, &superior);
693  }
694  lock_basic_unlock(&engine->signq->q_lock);
695  /* do some work */
696  if (rrset) {
697  ods_log_assert(superior);
698  if (!ctx) {
699  ods_log_debug("[%s[%i]] create hsm context",
700  worker2str(worker->type), worker->thread_num);
701  ctx = hsm_create_context();
702  }
703  if (!ctx) {
704  ods_log_crit("[%s[%i]] error creating libhsm context",
705  worker2str(worker->type), worker->thread_num);
706  engine->need_to_reload = 1;
707  lock_basic_lock(&superior->worker_lock);
708  superior->jobs_failed++;
709  lock_basic_unlock(&superior->worker_lock);
710  } else {
711  ods_log_assert(ctx);
712  lock_basic_lock(&superior->worker_lock);
713  task = superior->task;
714  ods_log_assert(task);
715  zone = task->zone;
716  lock_basic_unlock(&superior->worker_lock);
717  ods_log_assert(zone);
718  ods_log_assert(zone->apex);
719  ods_log_assert(zone->signconf);
720  worker->clock_in = time(NULL);
721  status = rrset_sign(ctx, rrset, superior->clock_in);
722  lock_basic_lock(&superior->worker_lock);
723  if (status == ODS_STATUS_OK) {
724  superior->jobs_completed++;
725  } else {
726  superior->jobs_failed++;
727  }
728  lock_basic_unlock(&superior->worker_lock);
729  }
730  if (worker_fulfilled(superior) && superior->sleeping) {
731  ods_log_deeebug("[%s[%i]] wake up superior[%u], work is "
732  "done", worker2str(worker->type), worker->thread_num,
733  superior->thread_num);
734  worker_wakeup(superior);
735  }
736  superior = NULL;
737  rrset = NULL;
738  }
739  /* done work */
740  }
741  /* wake up superior */
742  if (superior && superior->sleeping) {
743  ods_log_deeebug("[%s[%i]] wake up superior[%u], i am exiting",
744  worker2str(worker->type), worker->thread_num, superior->thread_num);
745  worker_wakeup(superior);
746  }
747  /* cleanup open HSM sessions */
748  if (ctx) {
749  hsm_destroy_context(ctx);
750  }
751  return;
752 }
753 
754 
759 void
761 {
762  ods_log_assert(worker);
763  switch (worker->type) {
764  case WORKER_DRUDGER:
765  worker_drudge(worker);
766  break;
767  case WORKER_WORKER:
768  worker_work(worker);
769  break;
770  default:
771  ods_log_error("[worker] illegal worker (id=%i)", worker->type);
772  break;
773  }
774  return;
775 }
776 
777 
782 void
783 worker_sleep(worker_type* worker, time_t timeout)
784 {
785  ods_log_assert(worker);
786  lock_basic_lock(&worker->worker_lock);
787  worker->sleeping = 1;
788  lock_basic_sleep(&worker->worker_alarm, &worker->worker_lock,
789  timeout);
790  lock_basic_unlock(&worker->worker_lock);
791  return;
792 }
793 
794 
799 void
800 worker_sleep_unless(worker_type* worker, time_t timeout)
801 {
802  ods_log_assert(worker);
803  lock_basic_lock(&worker->worker_lock);
804  while (!worker->need_to_exit && !worker_fulfilled(worker)) {
805  worker->sleeping = 1;
806  lock_basic_sleep(&worker->worker_alarm, &worker->worker_lock,
807  timeout);
808  ods_log_debug("[%s[%i]] somebody poked me, check completed jobs %u "
809  "appointed, %u completed, %u failed", worker2str(worker->type),
810  worker->thread_num, worker->jobs_appointed, worker->jobs_completed,
811  worker->jobs_failed);
812  }
813  lock_basic_unlock(&worker->worker_lock);
814  return;
815 }
816 
817 
822 void
824 {
825  ods_log_assert(worker);
826  if (worker && worker->sleeping && !worker->waiting) {
827  ods_log_debug("[%s[%i]] wake up", worker2str(worker->type),
828  worker->thread_num);
829  lock_basic_lock(&worker->worker_lock);
830  lock_basic_alarm(&worker->worker_alarm);
831  worker->sleeping = 0;
832  lock_basic_unlock(&worker->worker_lock);
833  }
834  return;
835 }
836 
837 
842 void
843 worker_wait_timeout(lock_basic_type* lock, cond_basic_type* condition,
844  time_t timeout)
845 {
846  lock_basic_lock(lock);
847  lock_basic_sleep(condition, lock, timeout);
848  lock_basic_unlock(lock);
849  return;
850 }
851 
852 
857 void
858 worker_wait(lock_basic_type* lock, cond_basic_type* condition)
859 {
860  worker_wait_timeout(lock, condition, 0);
861  return;
862 }
863 
864 
869 void
870 worker_notify(lock_basic_type* lock, cond_basic_type* condition)
871 {
872  lock_basic_lock(lock);
873  lock_basic_alarm(condition);
874  lock_basic_unlock(lock);
875  return;
876 }
877 
878 
883 void
884 worker_notify_all(lock_basic_type* lock, cond_basic_type* condition)
885 {
886  lock_basic_lock(lock);
887  lock_basic_broadcast(condition);
888  lock_basic_unlock(lock);
889  return;
890 }
891 
892 
897 void
899 {
900  allocator_type* allocator;
901  cond_basic_type worker_cond;
902  lock_basic_type worker_lock;
903  if (!worker) {
904  return;
905  }
906  allocator = worker->allocator;
907  worker_cond = worker->worker_alarm;
908  worker_lock = worker->worker_lock;
909  allocator_deallocate(allocator, (void*) worker);
910  lock_basic_destroy(&worker_lock);
911  lock_basic_off(&worker_cond);
912  return;
913 }
#define lock_basic_off(cond)
Definition: locks.h:100
Definition: task.h:43
rrset_type * rrset
Definition: denial.h:56
size_t jobs_completed
Definition: worker.h:61
unsigned waiting
Definition: worker.h:66
unsigned need_to_exit
Definition: worker.h:67
task_type * schedule_get_first_task(schedule_type *schedule)
Definition: schedule.c:247
void ods_log_debug(const char *format,...)
Definition: log.c:272
time_t when
Definition: task.h:61
size_t jobs_appointed
Definition: worker.h:60
lock_basic_type worker_lock
Definition: worker.h:64
#define lock_basic_destroy(lock)
Definition: locks.h:92
cond_basic_type signal_cond
Definition: engine.h:80
cond_basic_type q_threshold
Definition: fifoq.h:68
task_id interrupt
Definition: task.h:59
void * allocator_alloc(allocator_type *allocator, size_t size)
Definition: allocator.c:68
void lhsm_check_connection(void *engine)
Definition: hsm.c:114
ods_status tools_signconf(zone_type *zone)
Definition: tools.c:54
lock_basic_type q_lock
Definition: fifoq.h:67
time_t sig_time
Definition: stats.h:65
void * engine
Definition: worker.h:55
ods_status schedule_task(schedule_type *schedule, task_type *task, int log)
Definition: schedule.c:148
void ods_log_info(const char *format,...)
Definition: log.c:304
const char * task_who2str(task_type *task)
Definition: task.c:178
enum ods_enum_status ods_status
Definition: status.h:91
void worker_start(worker_type *worker)
Definition: worker.c:760
lock_basic_type zone_lock
Definition: zone.h:97
time_t backoff
Definition: task.h:63
void ods_log_error(const char *format,...)
Definition: log.c:336
lock_basic_type stats_lock
Definition: stats.h:69
const char * ods_status2str(ods_status status)
Definition: status.c:112
ldns_rbtree_t * domains
Definition: namedb.h:51
Definition: task.h:47
void * zone
Definition: task.h:65
ods_lookup_table * ods_lookup_by_id(ods_lookup_table *table, int id)
Definition: status.c:95
rrset_type * next
Definition: rrset.h:75
void worker_cleanup(worker_type *worker)
Definition: worker.c:898
#define lock_basic_set(cond)
Definition: locks.h:96
enum task_id_enum task_id
Definition: task.h:50
ods_status fifoq_push(fifoq_type *q, void *item, worker_type *worker, int *tries)
Definition: fifoq.c:121
void ods_log_crit(const char *format,...)
Definition: log.c:352
lock_basic_type signal_lock
Definition: engine.h:81
size_t jobs_failed
Definition: worker.h:62
void worker_wait(lock_basic_type *lock, cond_basic_type *condition)
Definition: worker.c:858
ods_status tools_input(zone_type *zone)
Definition: tools.c:95
task_type * task
Definition: worker.h:56
#define lock_basic_lock(lock)
Definition: locks.h:93
namedb_type * db
Definition: zone.h:88
Definition: task.h:45
void worker_sleep(worker_type *worker, time_t timeout)
Definition: worker.c:783
#define lock_basic_sleep(cond, lock, sleep)
Definition: locks.h:97
time_t halted_when
Definition: task.h:62
time_t clock_in
Definition: worker.h:59
int lock_basic_type
Definition: locks.h:90
task_type * schedule_pop_task(schedule_type *schedule)
Definition: schedule.c:287
unsigned is_processed
Definition: namedb.h:58
ods_status tools_output(zone_type *zone, engine_type *engine)
Definition: tools.c:178
void worker_notify_all(lock_basic_type *lock, cond_basic_type *condition)
Definition: worker.c:884
signconf_type * signconf
Definition: zone.h:86
ods_status zone_backup2(zone_type *zone)
Definition: zone.c:1028
time_t start_time
Definition: stats.h:67
ods_status zone_update_serial(zone_type *zone)
Definition: zone.c:471
task_id halted
Definition: task.h:60
void worker_wakeup(worker_type *worker)
Definition: worker.c:823
uint32_t sig_reuse
Definition: stats.h:64
enum worker_enum worker_id
Definition: worker.h:48
void worker_sleep_unless(worker_type *worker, time_t timeout)
Definition: worker.c:800
time_t duration2time(duration_type *duration)
Definition: duration.c:373
void ods_log_verbose(const char *format,...)
Definition: log.c:288
time_t last_modified
Definition: signconf.h:80
const char * name
Definition: status.h:96
task_id what
Definition: task.h:58
#define lock_basic_init(lock)
Definition: locks.h:91
int thread_num
Definition: worker.h:53
const char * name
Definition: zone.h:78
schedule_type * taskq
Definition: engine.h:63
uint32_t sig_soa_count
Definition: stats.h:63
ods_status zone_prepare_keys(zone_type *zone)
Definition: zone.c:430
duration_type * sig_resign_interval
Definition: signconf.h:57
cond_basic_type worker_alarm
Definition: worker.h:63
void ods_log_deeebug(const char *format,...)
Definition: log.c:256
worker_type * worker_create(allocator_type *allocator, int num, worker_id type)
Definition: worker.c:74
ods_status rrset_sign(hsm_ctx_t *ctx, rrset_type *rrset, time_t signtime)
Definition: rrset.c:664
void allocator_deallocate(allocator_type *allocator, void *data)
Definition: allocator.c:137
uint32_t sig_count
Definition: stats.h:62
unsigned sleeping
Definition: worker.h:65
task_id working_with
Definition: worker.h:57
lock_basic_type schedule_lock
Definition: schedule.h:65
worker_id type
Definition: worker.h:58
void worker_wait_timeout(lock_basic_type *lock, cond_basic_type *condition, time_t timeout)
Definition: worker.c:843
rrset_type * rrsets
Definition: domain.h:62
void * task
Definition: zone.h:94
fifoq_type * signq
Definition: engine.h:64
cond_basic_type q_nonfull
Definition: fifoq.h:69
#define ods_log_assert(x)
Definition: log.h:156
int need_to_reload
Definition: engine.h:77
void * denial
Definition: domain.h:58
#define lock_basic_alarm(cond)
Definition: locks.h:98
#define lock_basic_unlock(lock)
Definition: locks.h:94
void ods_log_warning(const char *format,...)
Definition: log.c:320
allocator_type * allocator
Definition: worker.h:52
ldns_rdf * apex
Definition: zone.h:70
const char * task_what2str(task_id what)
Definition: task.c:148
void worker_notify(lock_basic_type *lock, cond_basic_type *condition)
Definition: worker.c:870
time_t time_now(void)
Definition: duration.c:515
ods_lookup_table worker_str[]
Definition: worker.c:47
stats_type * stats
Definition: zone.h:96
#define lock_basic_broadcast(cond)
Definition: locks.h:99
void * fifoq_pop(fifoq_type *q, worker_type **worker)
Definition: fifoq.c:91