1 /* $OpenBSD: runq.c,v 1.3 2019/06/14 19:55:25 eric Exp $ */ 2 3 /* 4 * Copyright (c) 2013,2019 Eric Faurot <eric@openbsd.org> 5 * 6 * Permission to use, copy, modify, and distribute this software for any 7 * purpose with or without fee is hereby granted, provided that the above 8 * copyright notice and this permission notice appear in all copies. 9 * 10 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES 11 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF 12 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR 13 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES 14 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN 15 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF 16 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. 17 */ 18 19 #include <sys/types.h> 20 #include <sys/socket.h> 21 #include <sys/queue.h> 22 #include <sys/tree.h> 23 #include <sys/uio.h> 24 25 #include <imsg.h> 26 #include <stdio.h> 27 #include <stdlib.h> 28 #include <limits.h> 29 #include <time.h> 30 31 #include "smtpd.h" 32 33 struct job { 34 TAILQ_ENTRY(job) entry; 35 time_t when; 36 void *arg; 37 }; 38 39 struct runq { 40 TAILQ_HEAD(, job) jobs; 41 void (*cb)(struct runq *, void *); 42 struct event ev; 43 }; 44 45 static void runq_timeout(int, short, void *); 46 47 static struct runq *active; 48 49 static void 50 runq_reset(struct runq *runq) 51 { 52 struct timeval tv; 53 struct job *job; 54 time_t now; 55 56 job = TAILQ_FIRST(&runq->jobs); 57 if (job == NULL) 58 return; 59 60 now = time(NULL); 61 if (job->when <= now) 62 tv.tv_sec = 0; 63 else 64 tv.tv_sec = job->when - now; 65 tv.tv_usec = 0; 66 evtimer_add(&runq->ev, &tv); 67 } 68 69 static void 70 runq_timeout(int fd, short ev, void *arg) 71 { 72 struct runq *runq = arg; 73 struct job *job; 74 time_t now; 75 76 active = runq; 77 now = time(NULL); 78 79 while((job = TAILQ_FIRST(&runq->jobs))) { 80 if (job->when > now) 81 break; 82 TAILQ_REMOVE(&runq->jobs, job, entry); 83 runq->cb(runq, job->arg); 84 free(job); 85 } 86 87 active = NULL; 88 runq_reset(runq); 89 } 90 91 int 92 runq_init(struct runq **runqp, void (*cb)(struct runq *, void *)) 93 { 94 struct runq *runq; 95 96 runq = malloc(sizeof(*runq)); 97 if (runq == NULL) 98 return (0); 99 100 runq->cb = cb; 101 TAILQ_INIT(&runq->jobs); 102 evtimer_set(&runq->ev, runq_timeout, runq); 103 104 *runqp = runq; 105 106 return (1); 107 } 108 109 int 110 runq_schedule(struct runq *runq, time_t delay, void *arg) 111 { 112 time_t t; 113 114 time(&t); 115 return runq_schedule_at(runq, t + delay, arg); 116 } 117 118 int 119 runq_schedule_at(struct runq *runq, time_t when, void *arg) 120 { 121 struct job *job, *tmpjob; 122 123 job = malloc(sizeof(*job)); 124 if (job == NULL) 125 return (0); 126 127 job->arg = arg; 128 job->when = when; 129 130 TAILQ_FOREACH(tmpjob, &runq->jobs, entry) { 131 if (tmpjob->when > job->when) { 132 TAILQ_INSERT_BEFORE(tmpjob, job, entry); 133 goto done; 134 } 135 } 136 TAILQ_INSERT_TAIL(&runq->jobs, job, entry); 137 138 done: 139 if (runq != active && job == TAILQ_FIRST(&runq->jobs)) { 140 evtimer_del(&runq->ev); 141 runq_reset(runq); 142 } 143 return (1); 144 } 145 146 int 147 runq_cancel(struct runq *runq, void *arg) 148 { 149 struct job *job, *first; 150 151 first = TAILQ_FIRST(&runq->jobs); 152 TAILQ_FOREACH(job, &runq->jobs, entry) { 153 if (job->arg == arg) { 154 TAILQ_REMOVE(&runq->jobs, job, entry); 155 free(job); 156 if (runq != active && job == first) { 157 evtimer_del(&runq->ev); 158 runq_reset(runq); 159 } 160 return (1); 161 } 162 } 163 164 return (0); 165 } 166 167 int 168 runq_pending(struct runq *runq, void *arg, time_t *when) 169 { 170 struct job *job; 171 172 TAILQ_FOREACH(job, &runq->jobs, entry) { 173 if (job->arg == arg) { 174 if (when) 175 *when = job->when; 176 return (1); 177 } 178 } 179 180 return (0); 181 } 182