beanstalkd.c

   1 /* beanstalk - fast, general-purpose work queue */
   2
   3 /* Copyright (C) 2007 Keith Rarick and Philotic Inc.
   4
   5  * This program is free software: you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation, either version 3 of the License, or
   8  * (at your option) any later version.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include <signal.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <errno.h>
  23 #include <sys/stat.h>
  24 #include <unistd.h>
  25 #include <pwd.h>
  26 #include <fcntl.h>
  27 #include "sd-daemon.h"
  28 #include "version.h"
  29 #include "dat.h"
  30
  31 static char *user = NULL;
  32 static int detach = 0;
  33 static char *port = "11300";
  34 static char *host_addr;
  35
  36 static void
  37 nullfd(int fd, int flags)
  38 {
  39     int r;
  40
  41     close(fd);
  42     r = open("/dev/null", flags);
  43     if (r != fd) twarn("open(\"/dev/null\")"), exit(1);
  44 }
  45
  46 static void
  47 dfork()
  48 {
  49     pid_t p;
  50
  51     p = fork();
  52     if (p == -1) exit(1);
  53     if (p) exit(0);
  54 }
  55
  56 static void
  57 daemonize()
  58 {
  59     int r;
  60
  61     r = chdir("/");
  62     if (r) return twarn("chdir");
  63
  64     nullfd(0, O_RDONLY);
  65     nullfd(1, O_WRONLY);
  66     nullfd(2, O_WRONLY);
  67     umask(0);
  68     dfork();
  69     setsid();
  70     dfork();
  71 }
  72
  73 static void
  74 su(const char *user) {
  75     int r;
  76     struct passwd *pwent;
  77
  78     errno = 0;
  79     pwent = getpwnam(user);
  80     if (errno) twarn("getpwnam(\"%s\")", user), exit(32);
  81     if (!pwent) twarnx("getpwnam(\"%s\"): no such user", user), exit(33);
  82
  83     r = setgid(pwent->pw_gid);
  84     if (r == -1) twarn("setgid(%d \"%s\")", pwent->pw_gid, user), exit(34);
  85
  86     r = setuid(pwent->pw_uid);
  87     if (r == -1) twarn("setuid(%d \"%s\")", pwent->pw_uid, user), exit(34);
  88 }
  89
  90 void
  91 exit_cleanly(int sig)
  92 {
  93     binlog_shutdown();
  94     exit(0);
  95 }
  96
  97
  98 static void
  99 set_sig_handlers()
 100 {
 101     int r;
 102     struct sigaction sa;
 103
 104     sa.sa_handler = SIG_IGN;
 105     sa.sa_flags = 0;
 106     r = sigemptyset(&sa.sa_mask);
 107     if (r == -1) twarn("sigemptyset()"), exit(111);
 108
 109     r = sigaction(SIGPIPE, &sa, 0);
 110     if (r == -1) twarn("sigaction(SIGPIPE)"), exit(111);
 111
 112     sa.sa_handler = enter_drain_mode;
 113     r = sigaction(SIGUSR1, &sa, 0);
 114     if (r == -1) twarn("sigaction(SIGUSR1)"), exit(111);
 115
 116     sa.sa_handler = exit_cleanly;
 117     r = sigaction(SIGINT, &sa, 0);
 118     if (r == -1) twarn("sigaction(SIGINT)"), exit(111);
 119
 120     sa.sa_handler = exit_cleanly;
 121     r = sigaction(SIGTERM, &sa, 0);
 122     if (r == -1) twarn("sigaction(SIGTERM)"), exit(111);
 123 }
 124
 125 static void
 126 usage(char *msg, char *arg)
 127 {
 128     if (arg) warnx("%s: %s", msg, arg);
 129     fprintf(stderr, "Use: %s [OPTIONS]\n"
 130             "\n"
 131             "Options:\n"
 132             " -d       detach\n"
 133             " -b DIR   binlog directory (must be absolute path if used with -d)\n"
 134             " -f MS    fsync at most once every MS milliseconds"
 135                        " (use -f 0 for \"always fsync\")\n"
 136             " -F       never fsync (default)\n"
 137             " -l ADDR  listen on address (default is 0.0.0.0)\n"
 138             " -p PORT  listen on port (default is 11300)\n"
 139             " -u USER  become user and group\n"
 140             " -z BYTES set the maximum job size in bytes (default is %d)\n"
 141             " -s BYTES set the size of each binlog file (default is %d)\n"
 142             "            (will be rounded up to a multiple of 512 bytes)\n"
 143             " -v       show version information\n"
 144             " -h       show this help\n",
 145             progname, JOB_DATA_SIZE_LIMIT_DEFAULT, BINLOG_SIZE_LIMIT_DEFAULT);
 146     exit(arg ? 5 : 0);
 147 }
 148
 149 static size_t
 150 parse_size_t(char *str)
 151 {
 152     char r, x;
 153     size_t size;
 154
 155     r = sscanf(str, "%zu%c", &size, &x);
 156     if (1 != r) usage("invalid size", str);
 157     return size;
 158 }
 159
 160 static char *
 161 require_arg(char *opt, char *arg)
 162 {
 163     if (!arg) usage("option requires an argument", opt);
 164     return arg;
 165 }
 166
 167 static void
 168 warn_systemd_ignored_option(char *opt, char *arg)
 169 {
 170     if (sd_listen_fds(0) > 0) {
 171         warnx("inherited listen fd; ignoring option: %s %s", opt, arg);
 172     }
 173 }
 174
 175 static void
 176 opts(int argc, char **argv)
 177 {
 178     int i;
 179
 180     for (i = 1; i < argc; ++i) {
 181         if (argv[i][0] != '-') usage("unknown option", argv[i]);
 182         if (argv[i][1] == 0 || argv[i][2] != 0) usage("unknown option",argv[i]);
 183         switch (argv[i][1]) {
 184             case 'd':
 185                 detach = 1;
 186                 break;
 187             case 'p':
 188                 port = require_arg("-p", argv[++i]);
 189                 warn_systemd_ignored_option("-p", argv[i]);
 190                 break;
 191             case 'l':
 192                 host_addr = require_arg("-l", argv[++i]);
 193                 warn_systemd_ignored_option("-l", argv[i]);
 194                 break;
 195             case 'z':
 196                 job_data_size_limit = parse_size_t(require_arg("-z",
 197                                                                argv[++i]));
 198                 break;
 199             case 's':
 200                 binlog_size_limit = parse_size_t(require_arg("-s", argv[++i]));
 201                 break;
 202             case 'f':
 203                 fsync_throttle_ms = parse_size_t(require_arg("-f", argv[++i]));
 204                 enable_fsync = 1;
 205                 break;
 206             case 'F':
 207                 enable_fsync = 0;
 208                 break;
 209             case 'u':
 210                 user = require_arg("-u", argv[++i]);
 211                 break;
 212             case 'b':
 213                 binlog_dir = require_arg("-b", argv[++i]);
 214                 break;
 215             case 'h':
 216                 usage(NULL, NULL);
 217             case 'v':
 218                 printf("beanstalkd %s\n", VERSION);
 219                 exit(0);
 220             default:
 221                 usage("unknown option", argv[i]);
 222         }
 223     }
 224 }
 225
 226 int
 227 main(int argc, char **argv)
 228 {
 229     int r;
 230     Srv s = {};
 231     struct job binlog_jobs = {};
 232
 233     progname = argv[0];
 234     opts(argc, argv);
 235
 236     if (detach && binlog_dir) {
 237         if (binlog_dir[0] != '/') {
 238             warnx("The -b option requires an absolute path when used with -d.");
 239             usage("Path is not absolute", binlog_dir);
 240         }
 241     }
 242
 243     r = make_server_socket(host_addr, port);
 244     if (r == -1) twarnx("make_server_socket()"), exit(111);
 245     s.sock.fd = r;
 246
 247     prot_init();
 248
 249     /* We want to make sure that only one beanstalkd tries to use the binlog
 250      * directory at a time. So acquire a lock now and never release it. */
 251     if (binlog_dir) {
 252         r = binlog_lock();
 253         if (!r) twarnx("failed to lock binlog dir %s", binlog_dir), exit(10);
 254     }
 255
 256     if (user) su(user);
 257     set_sig_handlers();
 258
 259     if (binlog_dir) {
 260         binlog_jobs.prev = binlog_jobs.next = &binlog_jobs;
 261         binlog_init(&binlog_jobs);
 262         prot_replay_binlog(&binlog_jobs);
 263     }
 264
 265     if (detach) {
 266         daemonize();
 267     }
 268
 269     srv(&s);
 270     binlog_shutdown();
 271     return 0;
 272 }