1 /* $NetBSD: qmgr.h,v 1.3 2020/03/18 19:05:17 christos Exp $ */ 2 3 /*++ 4 /* NAME 5 /* qmgr 3h 6 /* SUMMARY 7 /* queue manager data structures 8 /* SYNOPSIS 9 /* #include "qmgr.h" 10 /* DESCRIPTION 11 /* .nf 12 13 /* 14 * System library. 15 */ 16 #include <sys/time.h> 17 #include <time.h> 18 19 /* 20 * Utility library. 21 */ 22 #include <vstream.h> 23 #include <scan_dir.h> 24 25 /* 26 * Global library. 27 */ 28 #include <recipient_list.h> 29 #include <dsn.h> 30 31 /* 32 * The queue manager is built around lots of mutually-referring structures. 33 * These typedefs save some typing. 34 */ 35 typedef struct QMGR_TRANSPORT QMGR_TRANSPORT; 36 typedef struct QMGR_QUEUE QMGR_QUEUE; 37 typedef struct QMGR_ENTRY QMGR_ENTRY; 38 typedef struct QMGR_MESSAGE QMGR_MESSAGE; 39 typedef struct QMGR_TRANSPORT_LIST QMGR_TRANSPORT_LIST; 40 typedef struct QMGR_QUEUE_LIST QMGR_QUEUE_LIST; 41 typedef struct QMGR_ENTRY_LIST QMGR_ENTRY_LIST; 42 typedef struct QMGR_SCAN QMGR_SCAN; 43 typedef struct QMGR_FEEDBACK QMGR_FEEDBACK; 44 45 /* 46 * Hairy macros to update doubly-linked lists. 47 */ 48 #define QMGR_LIST_ROTATE(head, object) { \ 49 head.next->peers.prev = head.prev; \ 50 head.prev->peers.next = head.next; \ 51 head.next = object->peers.next; \ 52 if (object->peers.next) \ 53 head.next->peers.prev = 0; \ 54 head.prev = object; \ 55 object->peers.next = 0; \ 56 } 57 58 #define QMGR_LIST_UNLINK(head, type, object) { \ 59 type next = object->peers.next; \ 60 type prev = object->peers.prev; \ 61 if (prev) prev->peers.next = next; \ 62 else head.next = next; \ 63 if (next) next->peers.prev = prev; \ 64 else head.prev = prev; \ 65 object->peers.next = object->peers.prev = 0; \ 66 } 67 68 #define QMGR_LIST_APPEND(head, object) { \ 69 object->peers.next = head.next; \ 70 object->peers.prev = 0; \ 71 if (head.next) { \ 72 head.next->peers.prev = object; \ 73 } else { \ 74 head.prev = object; \ 75 } \ 76 head.next = object; \ 77 } 78 79 #define QMGR_LIST_PREPEND(head, object) { \ 80 object->peers.prev = head.prev; \ 81 object->peers.next = 0; \ 82 if (head.prev) { \ 83 head.prev->peers.next = object; \ 84 } else { \ 85 head.next = object; \ 86 } \ 87 head.prev = object; \ 88 } 89 90 #define QMGR_LIST_INIT(head) { \ 91 head.prev = 0; \ 92 head.next = 0; \ 93 } 94 95 /* 96 * Transports are looked up by name (when we have resolved a message), or 97 * round-robin wise (when we want to distribute resources fairly). 98 */ 99 struct QMGR_TRANSPORT_LIST { 100 QMGR_TRANSPORT *next; 101 QMGR_TRANSPORT *prev; 102 }; 103 104 extern struct HTABLE *qmgr_transport_byname; /* transport by name */ 105 extern QMGR_TRANSPORT_LIST qmgr_transport_list; /* transports, round robin */ 106 107 /* 108 * Delivery agents provide feedback, as hints that Postfix should expend 109 * more or fewer resources on a specific destination domain. The main.cf 110 * file specifies how feedback affects delivery concurrency: add/subtract a 111 * constant, a ratio of constants, or a constant divided by the delivery 112 * concurrency; and it specifies how much feedback must accumulate between 113 * concurrency updates. 114 */ 115 struct QMGR_FEEDBACK { 116 int hysteresis; /* to pass, need to be this tall */ 117 double base; /* pre-computed from main.cf */ 118 int index; /* none, window, sqrt(window) */ 119 }; 120 121 #define QMGR_FEEDBACK_IDX_NONE 0 /* no window dependence */ 122 #define QMGR_FEEDBACK_IDX_WIN 1 /* 1/window dependence */ 123 #if 0 124 #define QMGR_FEEDBACK_IDX_SQRT_WIN 2 /* 1/sqrt(window) dependence */ 125 #endif 126 127 #ifdef QMGR_FEEDBACK_IDX_SQRT_WIN 128 #include <math.h> 129 #endif 130 131 extern void qmgr_feedback_init(QMGR_FEEDBACK *, const char *, const char *, const char *, const char *); 132 133 #ifndef QMGR_FEEDBACK_IDX_SQRT_WIN 134 #define QMGR_FEEDBACK_VAL(fb, win) \ 135 ((fb).index == QMGR_FEEDBACK_IDX_NONE ? (fb).base : (fb).base / (win)) 136 #else 137 #define QMGR_FEEDBACK_VAL(fb, win) \ 138 ((fb).index == QMGR_FEEDBACK_IDX_NONE ? (fb).base : \ 139 (fb).index == QMGR_FEEDBACK_IDX_WIN ? (fb).base / (win) : \ 140 (fb).base / sqrt(win)) 141 #endif 142 143 /* 144 * Each transport (local, smtp-out, bounce) can have one queue per next hop 145 * name. Queues are looked up by next hop name (when we have resolved a 146 * message destination), or round-robin wise (when we want to deliver 147 * messages fairly). 148 */ 149 struct QMGR_QUEUE_LIST { 150 QMGR_QUEUE *next; 151 QMGR_QUEUE *prev; 152 }; 153 154 struct QMGR_TRANSPORT { 155 int flags; /* blocked, etc. */ 156 int pending; /* incomplete DA connections */ 157 char *name; /* transport name */ 158 int dest_concurrency_limit; /* concurrency per domain */ 159 int init_dest_concurrency; /* init. per-domain concurrency */ 160 int recipient_limit; /* recipients per transaction */ 161 struct HTABLE *queue_byname; /* queues indexed by domain */ 162 QMGR_QUEUE_LIST queue_list; /* queues, round robin order */ 163 QMGR_TRANSPORT_LIST peers; /* linkage */ 164 DSN *dsn; /* why unavailable */ 165 QMGR_FEEDBACK pos_feedback; /* positive feedback control */ 166 QMGR_FEEDBACK neg_feedback; /* negative feedback control */ 167 int fail_cohort_limit; /* flow shutdown control */ 168 int xport_rate_delay; /* suspend per delivery */ 169 int rate_delay; /* suspend per delivery */ 170 }; 171 172 #define QMGR_TRANSPORT_STAT_DEAD (1<<1) 173 #define QMGR_TRANSPORT_STAT_RATE_LOCK (1<<2) 174 175 typedef void (*QMGR_TRANSPORT_ALLOC_NOTIFY) (QMGR_TRANSPORT *, VSTREAM *); 176 extern QMGR_TRANSPORT *qmgr_transport_select(void); 177 extern void qmgr_transport_alloc(QMGR_TRANSPORT *, QMGR_TRANSPORT_ALLOC_NOTIFY); 178 extern void qmgr_transport_throttle(QMGR_TRANSPORT *, DSN *); 179 extern void qmgr_transport_unthrottle(QMGR_TRANSPORT *); 180 extern QMGR_TRANSPORT *qmgr_transport_create(const char *); 181 extern QMGR_TRANSPORT *qmgr_transport_find(const char *); 182 183 #define QMGR_TRANSPORT_THROTTLED(t) ((t)->flags & QMGR_TRANSPORT_STAT_DEAD) 184 185 /* 186 * Each next hop (e.g., a domain name) has its own queue of pending message 187 * transactions. The "todo" queue contains messages that are to be delivered 188 * to this next hop. When a message is elected for transmission, it is moved 189 * from the "todo" queue to the "busy" queue. Messages are taken from the 190 * "todo" queue in sequence. An initial destination delivery concurrency > 1 191 * ensures that one problematic message will not block all other traffic to 192 * that next hop. 193 */ 194 struct QMGR_ENTRY_LIST { 195 QMGR_ENTRY *next; 196 QMGR_ENTRY *prev; 197 }; 198 199 struct QMGR_QUEUE { 200 int dflags; /* delivery request options */ 201 time_t last_done; /* last delivery completion */ 202 char *name; /* domain name or address */ 203 char *nexthop; /* domain name */ 204 int todo_refcount; /* queue entries (todo list) */ 205 int busy_refcount; /* queue entries (busy list) */ 206 int window; /* slow open algorithm */ 207 double success; /* accumulated positive feedback */ 208 double failure; /* accumulated negative feedback */ 209 double fail_cohorts; /* pseudo-cohort failure count */ 210 QMGR_TRANSPORT *transport; /* transport linkage */ 211 QMGR_ENTRY_LIST todo; /* todo queue entries */ 212 QMGR_ENTRY_LIST busy; /* messages on the wire */ 213 QMGR_QUEUE_LIST peers; /* neighbor queues */ 214 DSN *dsn; /* why unavailable */ 215 time_t clog_time_to_warn; /* time of next warning */ 216 }; 217 218 #define QMGR_QUEUE_TODO 1 /* waiting for service */ 219 #define QMGR_QUEUE_BUSY 2 /* recipients on the wire */ 220 221 extern int qmgr_queue_count; 222 223 extern QMGR_QUEUE *qmgr_queue_create(QMGR_TRANSPORT *, const char *, const char *); 224 extern QMGR_QUEUE *qmgr_queue_select(QMGR_TRANSPORT *); 225 extern void qmgr_queue_done(QMGR_QUEUE *); 226 extern void qmgr_queue_throttle(QMGR_QUEUE *, DSN *); 227 extern void qmgr_queue_unthrottle(QMGR_QUEUE *); 228 extern QMGR_QUEUE *qmgr_queue_find(QMGR_TRANSPORT *, const char *); 229 extern void qmgr_queue_suspend(QMGR_QUEUE *, int); 230 231 /* 232 * Exclusive queue states. Originally there were only two: "throttled" and 233 * "not throttled". It was natural to encode these in the queue window size. 234 * After 10 years it's not practical to rip out all the working code and 235 * change representations, so we just clean up the names a little. 236 * 237 * Note: only the "ready" state can reach every state (including itself); 238 * non-ready states can reach only the "ready" state. Other transitions are 239 * forbidden, because they would result in dangling event handlers. 240 */ 241 #define QMGR_QUEUE_STAT_THROTTLED 0 /* back-off timer */ 242 #define QMGR_QUEUE_STAT_SUSPENDED -1 /* voluntary delay timer */ 243 #define QMGR_QUEUE_STAT_SAVED -2 /* delayed cleanup timer */ 244 #define QMGR_QUEUE_STAT_BAD -3 /* can't happen */ 245 246 #define QMGR_QUEUE_READY(q) ((q)->window > 0) 247 #define QMGR_QUEUE_THROTTLED(q) ((q)->window == QMGR_QUEUE_STAT_THROTTLED) 248 #define QMGR_QUEUE_SUSPENDED(q) ((q)->window == QMGR_QUEUE_STAT_SUSPENDED) 249 #define QMGR_QUEUE_SAVED(q) ((q)->window == QMGR_QUEUE_STAT_SAVED) 250 #define QMGR_QUEUE_BAD(q) ((q)->window <= QMGR_QUEUE_STAT_BAD) 251 252 #define QMGR_QUEUE_STATUS(q) ( \ 253 QMGR_QUEUE_READY(q) ? "ready" : \ 254 QMGR_QUEUE_THROTTLED(q) ? "throttled" : \ 255 QMGR_QUEUE_SUSPENDED(q) ? "suspended" : \ 256 QMGR_QUEUE_SAVED(q) ? "saved" : \ 257 "invalid queue status" \ 258 ) 259 260 /* 261 * Structure of one next-hop queue entry. In order to save some copying 262 * effort we allow multiple recipients per transaction. 263 */ 264 struct QMGR_ENTRY { 265 VSTREAM *stream; /* delivery process */ 266 QMGR_MESSAGE *message; /* message info */ 267 RECIPIENT_LIST rcpt_list; /* as many as it takes */ 268 QMGR_QUEUE *queue; /* parent linkage */ 269 QMGR_ENTRY_LIST peers; /* neighbor entries */ 270 }; 271 272 extern QMGR_ENTRY *qmgr_entry_select(QMGR_QUEUE *); 273 extern void qmgr_entry_unselect(QMGR_QUEUE *, QMGR_ENTRY *); 274 extern void qmgr_entry_move_todo(QMGR_QUEUE *, QMGR_ENTRY *); 275 extern void qmgr_entry_done(QMGR_ENTRY *, int); 276 extern QMGR_ENTRY *qmgr_entry_create(QMGR_QUEUE *, QMGR_MESSAGE *); 277 278 /* 279 * All common in-core information about a message is kept here. When all 280 * recipients have been tried the message file is linked to the "deferred" 281 * queue (some hosts not reachable), to the "bounce" queue (some recipients 282 * were rejected), and is then removed from the "active" queue. 283 */ 284 struct QMGR_MESSAGE { 285 int flags; /* delivery problems */ 286 int qflags; /* queuing flags */ 287 int tflags; /* tracing flags */ 288 long tflags_offset; /* offset for killing */ 289 int rflags; /* queue file read flags */ 290 VSTREAM *fp; /* open queue file or null */ 291 int refcount; /* queue entries */ 292 int single_rcpt; /* send one rcpt at a time */ 293 struct timeval arrival_time; /* start of receive transaction */ 294 time_t create_time; /* queue file create time */ 295 struct timeval active_time; /* time of entry into active queue */ 296 long warn_offset; /* warning bounce flag offset */ 297 time_t warn_time; /* time next warning to be sent */ 298 long data_offset; /* data seek offset */ 299 char *queue_name; /* queue name */ 300 char *queue_id; /* queue file */ 301 char *encoding; /* content encoding */ 302 char *sender; /* complete address */ 303 char *dsn_envid; /* DSN envelope ID */ 304 int dsn_ret; /* DSN headers/full */ 305 int smtputf8; /* requires unicode */ 306 char *verp_delims; /* VERP delimiters */ 307 char *filter_xport; /* filtering transport */ 308 char *inspect_xport; /* inspecting transport */ 309 char *redirect_addr; /* info@spammer.tld */ 310 long data_size; /* data segment size */ 311 long cont_length; /* message content length */ 312 long rcpt_offset; /* more recipients here */ 313 char *client_name; /* client hostname */ 314 char *client_addr; /* client address */ 315 char *client_port; /* client port */ 316 char *client_proto; /* client protocol */ 317 char *client_helo; /* helo parameter */ 318 char *sasl_method; /* SASL method */ 319 char *sasl_username; /* SASL user name */ 320 char *sasl_sender; /* SASL sender */ 321 char *log_ident; /* up-stream queue ID */ 322 char *rewrite_context; /* address qualification */ 323 RECIPIENT_LIST rcpt_list; /* complete addresses */ 324 }; 325 326 /* 327 * Flags 0-15 are reserved for qmgr_user.h. 328 */ 329 #define QMGR_READ_FLAG_SEEN_ALL_NON_RCPT (1<<16) 330 331 #define QMGR_MESSAGE_LOCKED ((QMGR_MESSAGE *) 1) 332 333 extern int qmgr_message_count; 334 extern int qmgr_recipient_count; 335 extern int qmgr_vrfy_pend_count; 336 337 extern void qmgr_message_free(QMGR_MESSAGE *); 338 extern void qmgr_message_update_warn(QMGR_MESSAGE *); 339 extern void qmgr_message_kill_record(QMGR_MESSAGE *, long); 340 extern QMGR_MESSAGE *qmgr_message_alloc(const char *, const char *, int, mode_t); 341 extern QMGR_MESSAGE *qmgr_message_realloc(QMGR_MESSAGE *); 342 343 #define QMGR_MSG_STATS(stats, message) \ 344 MSG_STATS_INIT2(stats, \ 345 incoming_arrival, message->arrival_time, \ 346 active_arrival, message->active_time) 347 348 /* 349 * qmgr_defer.c 350 */ 351 extern void qmgr_defer_transport(QMGR_TRANSPORT *, DSN *); 352 extern void qmgr_defer_todo(QMGR_QUEUE *, DSN *); 353 extern void qmgr_defer_recipient(QMGR_MESSAGE *, RECIPIENT *, DSN *); 354 355 /* 356 * qmgr_bounce.c 357 */ 358 extern void qmgr_bounce_recipient(QMGR_MESSAGE *, RECIPIENT *, DSN *); 359 360 /* 361 * qmgr_deliver.c 362 */ 363 extern int qmgr_deliver_concurrency; 364 extern void qmgr_deliver(QMGR_TRANSPORT *, VSTREAM *); 365 366 /* 367 * qmgr_active.c 368 */ 369 extern int qmgr_active_feed(QMGR_SCAN *, const char *); 370 extern void qmgr_active_drain(void); 371 extern void qmgr_active_done(QMGR_MESSAGE *); 372 373 /* 374 * qmgr_move.c 375 */ 376 extern void qmgr_move(const char *, const char *, time_t); 377 378 /* 379 * qmgr_enable.c 380 */ 381 extern void qmgr_enable_all(void); 382 extern void qmgr_enable_transport(QMGR_TRANSPORT *); 383 extern void qmgr_enable_queue(QMGR_QUEUE *); 384 385 /* 386 * Queue scan context. 387 */ 388 struct QMGR_SCAN { 389 char *queue; /* queue name */ 390 int flags; /* private, this run */ 391 int nflags; /* private, next run */ 392 struct SCAN_DIR *handle; /* scan */ 393 }; 394 395 /* 396 * Flags that control queue scans or destination selection. These are 397 * similar to the QMGR_REQ_XXX request codes. 398 */ 399 #define QMGR_SCAN_START (1<<0) /* start now/restart when done */ 400 #define QMGR_SCAN_ALL (1<<1) /* all queue file time stamps */ 401 #define QMGR_FLUSH_ONCE (1<<2) /* unthrottle once */ 402 #define QMGR_FLUSH_DFXP (1<<3) /* override defer_transports */ 403 #define QMGR_FLUSH_EACH (1<<4) /* unthrottle per message */ 404 #define QMGR_FORCE_EXPIRE (1<<5) /* force-defer and force-expire */ 405 406 /* 407 * qmgr_scan.c 408 */ 409 extern QMGR_SCAN *qmgr_scan_create(const char *); 410 extern void qmgr_scan_request(QMGR_SCAN *, int); 411 extern char *qmgr_scan_next(QMGR_SCAN *); 412 413 /* 414 * qmgr_error.c 415 */ 416 extern QMGR_TRANSPORT *qmgr_error_transport(const char *); 417 extern QMGR_QUEUE *qmgr_error_queue(const char *, DSN *); 418 extern char *qmgr_error_nexthop(DSN *); 419 420 /* LICENSE 421 /* .ad 422 /* .fi 423 /* The Secure Mailer license must be distributed with this software. 424 /* AUTHOR(S) 425 /* Wietse Venema 426 /* IBM T.J. Watson Research 427 /* P.O. Box 704 428 /* Yorktown Heights, NY 10598, USA 429 /* 430 /* Wietse Venema 431 /* Google, Inc. 432 /* 111 8th Avenue 433 /* New York, NY 10011, USA 434 /*--*/ 435