chiark / gitweb /
split program compiles
[inn-innduct.git] / innduct.c
1 /*
2  *  innduct
3  *  tailing reliable realtime streaming feeder for inn
4  *  innduct.c - main program, option parsing and startup
5  *
6  *  Copyright (C) 2010 Ian Jackson <ijackson@chiark.greenend.org.uk>
7  * 
8  *  This program is free software: you can redistribute it and/or modify
9  *  it under the terms of the GNU General Public License as published by
10  *  the Free Software Foundation, either version 3 of the License, or
11  *  (at your option) any later version.
12  * 
13  *  This program is distributed in the hope that it will be useful,
14  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
15  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16  *  GNU General Public License for more details.
17  * 
18  *  You should have received a copy of the GNU General Public License
19  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
20  *
21  *  (I believe that when you compile and link this as part of the inn2
22  *  build, with the Makefile runes I have provided, all the libraries
23  *  and files which end up included in innduct are licence-compatible
24  *  with GPLv3.  If not then please let me know.  -Ian Jackson.)
25  */
26
27 #include "innduct.h"
28
29 const char *sms_names[]= {
30 #define SMS_DEF_NAME(s) #s ,
31   SMS_LIST(SMS_DEF_NAME)
32   0
33 };
34
35
36 /*----- general operational variables -----*/
37
38 /* main initialises */
39 oop_source *loop;
40 ConnList conns;
41 char *path_lock, *path_flushing, *path_defer, *path_dump;
42 char *globpat_backlog;
43 pid_t self_pid;
44 int *lowvol_perperiod;
45 int lowvol_circptr;
46 int lowvol_total; /* does not include current period */
47
48 /*---------- configuration option variables ----------*/
49 /* when changing defaults, remember to update the manpage */
50
51 const char *sitename, *remote_host;
52 const char *feedfile, *path_run, *path_cli, *path_cli_dir;
53 int quiet_multiple=0;
54 int interactive=0, try_filemon=1;
55 int try_stream=1;
56 int port=119;
57 const char *inndconffile;
58
59 int max_connections=10;
60 int max_queue_per_conn=200;
61 int target_max_feedfile_size=100000;
62 int period_seconds=30;
63 int filepoll_seconds=5;
64 int max_queue_per_ipf=-1;
65
66 int connection_setup_timeout=200;
67 int inndcomm_flush_timeout=100;
68
69 double nocheck_thresh= 95.0; /* converted from percentage by main */
70 double nocheck_decay= 100; /* conv'd from articles to lambda by main */
71
72 /* all these are initialised to seconds, and converted to periods in main */
73 int reconnect_delay_periods=1000;
74 int flushfail_retry_periods=1000;
75 int backlog_retry_minperiods=100;
76 int backlog_spontrescan_periods=300;
77 int spontaneous_flush_periods=100000;
78 int max_separated_periods=2000;
79 int need_activity_periods=1000;
80 int lowvol_thresh=3;
81 int lowvol_periods=1000;
82
83 double max_bad_data_ratio= 1; /* conv'd from percentage by main */
84 int max_bad_data_initial= 30;
85   /* in one corrupt 4096-byte block the number of newlines has
86    * mean 16 and standard deviation 3.99.  30 corresponds to z=+3.5 */
87
88 /*========== main program ==========*/
89
90 static void postfork_inputfile(InputFile *ipf) {
91   if (!ipf) return;
92   xclose(ipf->fd, "(in child) input file ", ipf->path);
93 }
94
95 static void postfork_stdio(FILE *f, const char *what, const char *what2) {
96   /* we have no stdio streams that are buffered long-term */
97   if (!f) return;
98   if (fclose(f)) syscrash("(in child) close %s%s", what, what2?what2:0);
99 }
100
101 void postfork(void) {
102   in_child= 1;
103
104   xsigsetdefault(SIGTERM);
105   xsigsetdefault(SIGINT);
106   xsigsetdefault(SIGPIPE);
107   if (terminate_sig_flag) raise(terminate_sig_flag);
108
109   postfork_inputfile(main_input_file);
110   postfork_inputfile(flushing_input_file);
111
112   Conn *conn;
113   FOR_CONN(conn)
114     conn_closefd(conn,"(in child) ");
115
116   postfork_stdio(defer, "defer file ", path_defer);
117 }
118
119 typedef struct Every Every;
120 struct Every {
121   struct timeval interval;
122   int fixed_rate;
123   void (*f)(void);
124 };
125
126 static void every_schedule(Every *e, struct timeval base);
127
128 static void *every_happens(oop_source *lp, struct timeval base, void *e_v) {
129   Every *e= e_v;
130   e->f();
131   if (!e->fixed_rate) xgettimeofday(&base);
132   every_schedule(e, base);
133   return OOP_CONTINUE;
134 }
135
136 static void every_schedule(Every *e, struct timeval base) {
137   struct timeval when;
138   timeradd(&base, &e->interval, &when);
139   loop->on_time(loop, when, every_happens, e);
140 }
141
142 static void every(int interval, int fixed_rate, void (*f)(void)) {
143   NEW_DECL(Every *,e);
144   e->interval.tv_sec= interval;
145   e->interval.tv_usec= 0;
146   e->fixed_rate= fixed_rate;
147   e->f= f;
148   struct timeval now;
149   xgettimeofday(&now);
150   every_schedule(e, now);
151 }
152
153 void period(void) {
154   char *dipf_main=     dbg_report_ipf(main_input_file);
155   char *dipf_flushing= dbg_report_ipf(flushing_input_file);
156   char *dipf_backlog=  dbg_report_ipf(backlog_input_file);
157
158   dbg("PERIOD"
159       " sms=%s[%d] conns=%d until_connect=%d"
160       " input_files main:%s flushing:%s backlog:%s[%d]"
161       " children connecting=%ld inndcomm=%ld lowvol_total=%d"
162       ,
163       sms_names[sms], until_flush, conns.count, until_connect,
164       dipf_main, dipf_flushing, dipf_backlog, until_backlog_nextscan,
165       (long)connecting_child, (long)inndcomm_child, lowvol_total
166       );
167
168   free(dipf_main);
169   free(dipf_flushing);
170   free(dipf_backlog);
171
172   if (until_connect) until_connect--;
173
174   inputfile_queue_check_expired(backlog_input_file);
175   poll_backlog_file();
176   if (!backlog_input_file) close_defer(); /* want to start on a new backlog */
177   statemc_period_poll();
178   check_assign_articles();
179   check_idle_conns();
180 }
181
182
183 /*========== option parsing ==========*/
184
185 static void vbadusage(const char *fmt, va_list al) NORET_PRINTF(1,0);
186 static void vbadusage(const char *fmt, va_list al) {
187   char *m= xvasprintf(fmt,al);
188   fprintf(stderr, "bad usage: %s\n"
189           "say --help for help, or read the manpage\n",
190           m);
191   if (interactive < 2)
192     syslog(LOG_ERR,"innduct: invoked with bad usage: %s",m);
193   exit(8);
194 }
195
196 /*---------- generic option parser ----------*/
197
198 static void badusage(const char *fmt, ...) NORET_PRINTF(1,2);
199 static void badusage(const char *fmt, ...) {
200   va_list al;
201   va_start(al,fmt);
202   vbadusage(fmt,al);
203 }
204
205 enum OptFlags {
206   of_seconds= 001000u,
207   of_boolean= 002000u,
208 };
209
210 typedef struct Option Option;
211 typedef void OptionParser(const Option*, const char *val);
212
213 struct Option {
214   int shrt;
215   const char *lng, *formarg;
216   void *store;
217   OptionParser *fn;
218   int intval;
219 };
220
221 static void parse_options(const Option *options, char ***argvp) {
222   /* on return *argvp is first non-option arg; argc is not updated */
223
224   for (;;) {
225     const char *arg= *++(*argvp);
226     if (!arg) break;
227     if (*arg != '-') break;
228     if (!strcmp(arg,"--")) { arg= *++(*argvp); break; }
229     int a;
230     while ((a= *++arg)) {
231       const Option *o;
232       if (a=='-') {
233         arg++;
234         char *equals= strchr(arg,'=');
235         unsigned len= equals ? (size_t)(equals - arg) : strlen(arg);
236         for (o=options; o->shrt || o->lng; o++)
237           if (strlen(o->lng) == len && !memcmp(o->lng,arg,len))
238             goto found_long;
239         badusage("unknown long option --%s",arg);
240       found_long:
241         if (!o->formarg) {
242           if (equals) badusage("option --%s does not take a value",o->lng);
243           arg= 0;
244         } else if (equals) {
245           arg= equals+1;
246         } else {
247           arg= *++(*argvp);
248           if (!arg) badusage("option --%s needs a value for %s",
249                              o->lng, o->formarg);
250         }
251         o->fn(o, arg);
252         break; /* eaten the whole argument now */
253       }
254       for (o=options; o->shrt || o->lng; o++)
255         if (a == o->shrt)
256           goto found_short;
257       badusage("unknown short option -%c",a);
258     found_short:
259       if (!o->formarg) {
260         o->fn(o,0);
261       } else {
262         if (!*++arg) {
263           arg= *++(*argvp);
264           if (!arg) badusage("option -%c needs a value for %s",
265                              o->shrt, o->formarg);
266         }
267         o->fn(o,arg);
268         break; /* eaten the whole argument now */
269       }
270     }
271   }
272 }
273
274 #define DELIMPERHAPS(delim,str)  (str) ? (delim) : "", (str) ? (str) : ""
275
276 static void print_options(const Option *options, FILE *f) {
277   const Option *o;
278   for (o=options; o->shrt || o->lng; o++) {
279     char shrt[2] = { o->shrt, 0 };
280     char *optspec= xasprintf("%s%s%s%s%s",
281                              o->shrt ? "-" : "", shrt,
282                              o->shrt && o->lng ? "|" : "",
283                              DELIMPERHAPS("--", o->lng));
284     fprintf(f, "  %s%s%s\n", optspec, DELIMPERHAPS(" ", o->formarg));
285     free(optspec);
286   }
287 }
288
289 /*---------- specific option types ----------*/
290
291 static void op_integer(const Option *o, const char *val) {
292   char *ep;
293   errno= 0;
294   unsigned long ul= strtoul(val,&ep,10);
295   if (*ep || ep==val || errno || ul>INT_MAX)
296     badusage("bad integer value for %s",o->lng);
297   int *store= o->store;
298   *store= ul;
299 }
300
301 static void op_double(const Option *o, const char *val) {
302   int *store= o->store;
303   char *ep;
304   errno= 0;
305   *store= strtod(val, &ep);
306   if (*ep || ep==val || errno)
307     badusage("bad floating point value for %s",o->lng);
308 }
309
310 static void op_string(const Option *o, const char *val) {
311   const char **store= o->store;
312   *store= val;
313 }
314
315 static void op_seconds(const Option *o, const char *val) {
316   int *store= o->store;
317   char *ep;
318   int unit;
319
320   double v= strtod(val,&ep);
321   if (ep==val) badusage("bad time/duration value for %s",o->lng);
322
323   if (!*ep || !strcmp(ep,"s") || !strcmp(ep,"sec")) unit= 1;
324   else if (!strcmp(ep,"m") || !strcmp(ep,"min"))    unit= 60;
325   else if (!strcmp(ep,"h") || !strcmp(ep,"hour"))   unit= 3600;
326   else if (!strcmp(ep,"d") || !strcmp(ep,"day"))    unit= 86400;
327   else if (!strcmp(ep,"das")) unit= 10;
328   else if (!strcmp(ep,"hs"))  unit= 100;
329   else if (!strcmp(ep,"ks"))  unit= 1000;
330   else if (!strcmp(ep,"Ms"))  unit= 1000000;
331   else badusage("bad units %s for time/duration value for %s",ep,o->lng);
332
333   v *= unit;
334   v= ceil(v);
335   if (v > INT_MAX) badusage("time/duration value for %s out of range",o->lng);
336   *store= v;
337 }
338
339 static void op_setint(const Option *o, const char *val) {
340   int *store= o->store;
341   *store= o->intval;
342 }
343
344 /*---------- specific options ----------*/
345
346 static void help(const Option *o, const char *val);
347
348 static const Option innduct_options[]= {
349 {'f',"feedfile",         "F",     &feedfile,                 op_string      },
350 {'q',"quiet-multiple",   0,       &quiet_multiple,           op_setint, 1   },
351 {0,"no-daemon",          0,       &interactive,              op_setint, 1   },
352 {0,"interactive",        0,       &interactive,              op_setint, 2   },
353 {0,"no-streaming",       0,       &try_stream,               op_setint, 0   },
354 {0,"no-filemon",         0,       &try_filemon,              op_setint, 0   },
355 {'C',"inndconf",         "F",     &inndconffile,             op_string      },
356 {'P',"port",             "PORT",  &port,                     op_integer     },
357 {0,"chdir",              "DIR",   &path_run,                 op_string      },
358 {0,"cli",            "DIR/|PATH", &path_cli,                 op_string      },
359 {0,"help",               0,       0,                         help           },
360
361 {0,"max-connections",    "N",     &max_connections,          op_integer     },
362 {0,"max-queue-per-conn", "N",     &max_queue_per_conn,       op_integer     },
363 {0,"max-queue-per-file", "N",     &max_queue_per_ipf,        op_integer     },
364 {0,"feedfile-flush-size","BYTES", &target_max_feedfile_size, op_integer     },
365 {0,"period-interval",    "TIME",  &period_seconds,           op_seconds     },
366
367 {0,"connection-timeout",   "TIME",  &connection_setup_timeout, op_seconds   },
368 {0,"stuck-flush-timeout",  "TIME",  &inndcomm_flush_timeout,   op_seconds   },
369 {0,"feedfile-poll",        "TIME",  &filepoll_seconds,         op_seconds   },
370
371 {0,"no-check-proportion",   "PERCENT",   &nocheck_thresh,       op_double   },
372 {0,"no-check-response-time","ARTICLES",  &nocheck_decay,        op_double   },
373
374 {0,"reconnect-interval",     "PERIOD", &reconnect_delay_periods,  op_seconds },
375 {0,"flush-retry-interval",   "PERIOD", &flushfail_retry_periods,  op_seconds },
376 {0,"earliest-deferred-retry","PERIOD", &backlog_retry_minperiods, op_seconds },
377 {0,"backlog-rescan-interval","PERIOD",&backlog_spontrescan_periods,op_seconds},
378 {0,"max-flush-interval",     "PERIOD", &spontaneous_flush_periods,op_seconds },
379 {0,"flush-finish-timeout",   "PERIOD", &max_separated_periods,    op_seconds },
380 {0,"idle-timeout",           "PERIOD", &need_activity_periods,    op_seconds },
381 {0,"low-volume-thresh",      "PERIOD", &lowvol_thresh,            op_integer },
382 {0,"low-volume-window",      "PERIOD", &lowvol_periods,           op_seconds },
383
384 {0,"max-bad-input-data-ratio","PERCENT", &max_bad_data_ratio,   op_double    },
385 {0,"max-bad-input-data-init", "PERCENT", &max_bad_data_initial, op_integer   },
386
387 {0,0}
388 };
389
390 static void printusage(FILE *f) {
391   fputs("usage: innduct [options] site [fqdn]\n"
392         "available options are:\n", f);
393   print_options(innduct_options, f);
394 }
395
396 static void help(const Option *o, const char *val) {
397   printusage(stdout);
398   if (ferror(stdout) || fflush(stdout)) {
399     perror("innduct: writing help");
400     exit(12);
401   }
402   exit(0);
403 }
404
405 static void convert_to_periods_rndup(int *store) {
406   *store += period_seconds-1;
407   *store /= period_seconds;
408 }
409
410 static int path_ends_slash(const char *specified) {
411   int l= strlen(specified);
412   assert(l);
413   return specified[l-1] == '/';
414 }
415
416 static int innduct_fatal_cleanup(void) { return 12; } /* used for libinn die */
417
418 int main(int argc, char **argv) {
419   /* set up libinn logging */
420   message_program_name= "innduct";
421   message_fatal_cleanup= innduct_fatal_cleanup;
422
423 #define INNLOGSET_CALL(fn, pfx, sysloglevel)    \
424   message_handlers_##fn(1, duct_log_##fn);
425   INNLOGSETS(INNLOGSET_CALL)
426
427   if (!argv[1]) {
428     printusage(stderr);
429     exit(8);
430   }
431
432   parse_options(innduct_options, &argv);
433
434   /* arguments */
435
436   sitename= *argv++;
437   if (!sitename) badusage("need site name argument");
438
439   if (*argv) remote_host= *argv++;
440   else remote_host= sitename;
441   
442   if (*argv) badusage("too many non-option arguments");
443
444   /* defaults */
445
446   int r= innconf_read(inndconffile);
447   if (!r) badusage("could not read inn.conf");
448
449   if (!remote_host) remote_host= sitename;
450
451   if (nocheck_thresh < 0 || nocheck_thresh > 100)
452     badusage("nocheck threshold percentage must be between 0..100");
453   nocheck_thresh *= 0.01;
454
455   if (nocheck_decay < 0.1)
456     badusage("nocheck decay articles must be at least 0.1");
457   nocheck_decay= pow(0.5, 1.0/nocheck_decay);
458
459   convert_to_periods_rndup(&reconnect_delay_periods);
460   convert_to_periods_rndup(&flushfail_retry_periods);
461   convert_to_periods_rndup(&backlog_retry_minperiods);
462   convert_to_periods_rndup(&backlog_spontrescan_periods);
463   convert_to_periods_rndup(&spontaneous_flush_periods);
464   convert_to_periods_rndup(&max_separated_periods);
465   convert_to_periods_rndup(&need_activity_periods);
466   convert_to_periods_rndup(&lowvol_periods);
467
468   if (max_bad_data_ratio < 0 || max_bad_data_ratio > 100)
469     badusage("bad input data ratio must be between 0..100");
470   max_bad_data_ratio *= 0.01;
471
472   if (!path_run)
473     path_run= innconf->pathrun;
474
475   if (!feedfile) feedfile= sitename;
476   if (!feedfile[0]) badusage("feed filename, if specified, must be nonempty");
477   if (path_ends_slash(feedfile))
478     feedfile= xasprintf("%s%s", feedfile, sitename);
479   if (feedfile[0] != '/')
480     feedfile= xasprintf("%s/%s", innconf->pathoutgoing, feedfile);
481
482   if (!path_cli) {
483     path_cli_dir= "innduct";
484   } else if (!path_cli[0] || !strcmp(path_cli,"none")) {
485     path_cli= 0; /* ok, don't then */
486   } else if (path_ends_slash(path_cli)) {
487     path_cli_dir= xasprintf("%.*s", strlen(path_cli)-1, path_cli);
488   }
489   if (path_cli_dir)
490     path_cli= xasprintf("%s/%s", path_cli_dir, sitename);
491
492   if (max_queue_per_ipf<0)
493     max_queue_per_ipf= max_queue_per_conn * 2;
494
495   const char *feedfile_forbidden= "?*[~#";
496   int c;
497   while ((c= *feedfile_forbidden++))
498     if (strchr(feedfile, c))
499       badusage("feed filename may not contain metacharacter %c",c);
500
501   int i;
502   lowvol_perperiod= xcalloc(sizeof(*lowvol_perperiod), lowvol_periods);
503   for (i=0; i<lowvol_periods; i++) {
504     lowvol_perperiod[i]= lowvol_thresh;
505     lowvol_total += lowvol_thresh;
506   }
507   lowvol_total -= lowvol_thresh;
508
509   /* set things up */
510
511   path_lock=        xasprintf("%s_lock",      feedfile);
512   path_flushing=    xasprintf("%s_flushing",  feedfile);
513   path_defer=       xasprintf("%s_defer",     feedfile);
514   path_dump=        xasprintf("%s_dump",      feedfile);
515   globpat_backlog=  xasprintf("%s_backlog*",  feedfile);
516
517   oop_source_sys *sysloop= oop_sys_new();
518   if (!sysloop) syscrash("could not create liboop event loop");
519   loop= (oop_source*)sysloop;
520
521   LIST_INIT(conns);
522
523   if (interactive < 1) {
524     for (i=3; i<255; i++)
525       /* do this now before we open syslog, etc. */
526       close(i);
527   }
528
529   logv_prefix= xasprintf("%s| ", sitename);
530   if (interactive < 2) {
531     openlog("innduct",LOG_NDELAY|LOG_PID,LOG_NEWS);
532     logv_use_syslog= 1;
533   }
534
535   if (interactive < 1) {
536     int null= open("/dev/null",O_RDWR);
537     if (null<0) sysdie("failed to open /dev/null");
538     dup2(null,0);
539     dup2(null,1);
540     dup2(null,2);
541     xclose(null, "/dev/null original fd",0);
542
543     pid_t child1= xfork("daemonise first fork");
544     if (child1) _exit(0);
545
546     pid_t sid= setsid();
547     if (sid == -1) sysdie("setsid failed");
548
549     pid_t child2= xfork("daemonise second fork");
550     if (child2) _exit(0);
551   }
552
553   self_pid= getpid();
554   if (self_pid==-1) syscrash("getpid");
555
556   r= chdir(path_run);
557   if (r) sysdie("could not chdir to pathrun %s", path_run);
558
559   statemc_lock();
560
561   init_signals();
562
563   notice("starting");
564
565   int val= 1;
566   r= SMsetup(SM_PREOPEN, &val); if (!r) warn("SMsetup SM_PREOPEN failed");
567   r= SMinit(); if (!r) die("storage manager initialisation (SMinit) failed");
568
569   if (interactive >= 2)
570     cli_stdio();
571
572   cli_init();
573
574   int filemon_ok= 0;
575   if (!try_filemon) {
576     notice("filemon: suppressed by command line option, polling");
577   } else {
578     filemon_ok= filemon_method_init();
579     if (!filemon_ok)
580       warn("filemon: no file monitoring available, polling");
581   }
582   if (!filemon_ok)
583     every(filepoll_seconds,0,filepoll);
584
585   every(period_seconds,1,period);
586
587   statemc_init();
588
589   /* let's go */
590
591   void *run= oop_sys_run(sysloop);
592   assert(run == OOP_ERROR);
593   syscrash("event loop failed");
594 }