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 <sys/resource.h>
  25 #include <sys/socket.h>
  26 #include <netinet/in.h>
  27 #include <arpa/inet.h>
  28 #include <sys/types.h>
  29 #include <unistd.h>
  30 #include <pwd.h>
  31 #include <event.h>
  32
  33 #include "net.h"
  34 #include "util.h"
  35 #include "prot.h"
  36 #include "config.h"
  37 #include "binlog.h"
  38
  39 static char *user = NULL;
  40 static int detach = 0;
  41 static int port = 11300;
  42 static struct in_addr host_addr = { INADDR_ANY };
  43
  44 static void
  45 nullfd(int fd, int flags)
  46 {
  47     int r;
  48
  49     close(fd);
  50     r = open("/dev/null", flags);
  51     if (r != fd) twarn("open(\"/dev/null\")"), exit(1);
  52 }
  53
  54 static void
  55 dfork()
  56 {
  57     pid_t p;
  58
  59     p = fork();
  60     if (p == -1) exit(1);
  61     if (p) exit(0);
  62 }
  63
  64 static void
  65 daemonize()
  66 {
  67     chdir("/");
  68     nullfd(0, O_RDONLY);
  69     nullfd(1, O_WRONLY);
  70     nullfd(2, O_WRONLY);
  71     umask(0);
  72     dfork();
  73     setsid();
  74     dfork();
  75 }
  76
  77 static void
  78 su(const char *user) {
  79     int r;
  80     struct passwd *pwent;
  81
  82     errno = 0;
  83     pwent = getpwnam(user);
  84     if (errno) twarn("getpwnam(\"%s\")", user), exit(32);
  85     if (!pwent) twarnx("getpwnam(\"%s\"): no such user", user), exit(33);
  86
  87     r = setgid(pwent->pw_gid);
  88     if (r == -1) twarn("setgid(%d \"%s\")", pwent->pw_gid, user), exit(34);
  89
  90     r = setuid(pwent->pw_uid);
  91     if (r == -1) twarn("setuid(%d \"%s\")", pwent->pw_uid, user), exit(34);
  92 }
  93
  94 void
  95 exit_cleanly(int sig)
  96 {
  97     binlog_close();
  98     exit(0);
  99 }
 100
 101
 102 static void
 103 set_sig_handlers()
 104 {
 105     int r;
 106     struct sigaction sa;
 107
 108     sa.sa_handler = SIG_IGN;
 109     sa.sa_flags = 0;
 110     r = sigemptyset(&sa.sa_mask);
 111     if (r == -1) twarn("sigemptyset()"), exit(111);
 112
 113     r = sigaction(SIGPIPE, &sa, 0);
 114     if (r == -1) twarn("sigaction(SIGPIPE)"), exit(111);
 115
 116     sa.sa_handler = enter_drain_mode;
 117     r = sigaction(SIGUSR1, &sa, 0);
 118     if (r == -1) twarn("sigaction(SIGUSR1)"), exit(111);
 119
 120     sa.sa_handler = exit_cleanly;
 121     r = sigaction(SIGINT, &sa, 0);
 122     if (r == -1) twarn("sigaction(SIGINT)"), exit(111);
 123 }
 124
 125 /* This is a workaround for a mystifying workaround in libevent's epoll
 126  * implementation. The epoll_init() function creates an epoll fd with space to
 127  * handle RLIMIT_NOFILE - 1 fds, accompanied by the following puzzling comment:
 128  * "Solaris is somewhat retarded - it's important to drop backwards
 129  * compatibility when making changes. So, don't dare to put rl.rlim_cur here."
 130  * This is presumably to work around a bug in Solaris, but it has the
 131  * unfortunate side-effect of causing epoll_ctl() (and, therefore, event_add())
 132  * to fail for a valid fd if we have hit the limit of open fds. That makes it
 133  * hard to provide reasonable behavior in that situation. So, let's reduce the
 134  * real value of RLIMIT_NOFILE by one, after epoll_init() has run. */
 135 static void
 136 nudge_fd_limit()
 137 {
 138     int r;
 139     struct rlimit rl;
 140
 141     r = getrlimit(RLIMIT_NOFILE, &rl);
 142     if (r != 0) twarn("getrlimit(RLIMIT_NOFILE)"), exit(2);
 143
 144     rl.rlim_cur--;
 145
 146     r = setrlimit(RLIMIT_NOFILE, &rl);
 147     if (r != 0) twarn("setrlimit(RLIMIT_NOFILE)"), exit(2);
 148 }
 149
 150 static void
 151 usage(char *msg, char *arg)
 152 {
 153     if (arg) warnx("%s: %s", msg, arg);
 154     fprintf(stderr, "Use: %s [OPTIONS]\n"
 155             "\n"
 156             "Options:\n"
 157             " -d       detach\n"
 158             " -b DIR   binlog directory\n"
 159             " -l ADDR  listen on address (default is 0.0.0.0)\n"
 160             " -p PORT  listen on port (default is 11300)\n"
 161             " -u USER  become user and group\n"
 162             " -z SIZE  set the maximum job size in bytes (default is %d)\n"
 163             " -v       show version information\n"
 164             " -h       show this help\n",
 165             progname, JOB_DATA_SIZE_LIMIT_DEFAULT);
 166     exit(arg ? 5 : 0);
 167 }
 168
 169 static size_t
 170 parse_size_t(char *str)
 171 {
 172     char r, x;
 173     size_t size;
 174
 175     r = sscanf(str, "%zu%c", &size, &x);
 176     if (1 != r) usage("invalid size", str);
 177     return size;
 178 }
 179
 180 static char *
 181 require_arg(char *opt, char *arg)
 182 {
 183     if (!arg) usage("option requires an argument", opt);
 184     return arg;
 185 }
 186
 187 static int
 188 parse_port(char *portstr)
 189 {
 190     int port;
 191     char *end;
 192
 193     errno = 0;
 194     port = strtol(portstr, &end, 10);
 195     if (end == portstr) usage("invalid port", portstr);
 196     if (end[0] != 0) usage("invalid port", portstr);
 197     if (errno) usage("invalid port", portstr);
 198
 199    return port;
 200 }
 201
 202 static struct in_addr
 203 parse_host(char *hoststr)
 204 {
 205     int r;
 206     struct in_addr addr;
 207
 208     r = inet_aton(hoststr, &addr);
 209     if (!r) usage("invalid address", hoststr);
 210
 211     return addr;
 212 }
 213
 214 static void
 215 opts(int argc, char **argv)
 216 {
 217     int i;
 218
 219     for (i = 1; i < argc; ++i) {
 220         if (argv[i][0] != '-') usage("unknown option", argv[i]);
 221         if (argv[i][1] == 0 || argv[i][2] != 0) usage("unknown option",argv[i]);
 222         switch (argv[i][1]) {
 223             case 'd':
 224                 detach = 1;
 225                 break;
 226             case 'p':
 227                 port = parse_port(require_arg("-p", argv[++i]));
 228                 break;
 229             case 'l':
 230                 host_addr = parse_host(require_arg("-l", argv[++i]));
 231                 break;
 232             case 'z':
 233                 job_data_size_limit = parse_size_t(require_arg("-z",
 234                                                                argv[++i]));
 235                 break;
 236             case 'u':
 237                 user = require_arg("-u", argv[++i]);
 238                 break;
 239             case 'b':
 240                 binlog_dir = require_arg("-b", argv[++i]);
 241                 break;
 242             case 'h':
 243                 usage(NULL, NULL);
 244             case 'v':
 245                 printf("beanstalkd %s\n", VERSION);
 246                 exit(0);
 247             default:
 248                 usage("unknown option", argv[i]);
 249         }
 250     }
 251 }
 252
 253 int
 254 main(int argc, char **argv)
 255 {
 256     int r;
 257     struct event_base *ev_base;
 258
 259     progname = argv[0];
 260     opts(argc, argv);
 261
 262     job_init();
 263     prot_init();
 264
 265     /* We want to make sure that only one beanstalkd tries to use the binlog
 266      * directory at a time. So acquire a lock now and never release it. */
 267     if (binlog_dir) {
 268         r = binlog_lock();
 269         if (!r) twarnx("failed to lock binlog dir %s", binlog_dir), exit(10);
 270     }
 271
 272     r = make_server_socket(host_addr, port);
 273     if (r == -1) twarnx("make_server_socket()"), exit(111);
 274
 275     if (user) su(user);
 276     ev_base = event_init();
 277     set_sig_handlers();
 278     nudge_fd_limit();
 279
 280     unbrake((evh) h_accept);
 281     prot_replay_binlog();
 282     binlog_init();
 283
 284     if (detach) {
 285         daemonize();
 286         event_reinit(ev_base);
 287     }
 288
 289     event_dispatch();
 290     binlog_close();
 291     twarnx("got here for some reason");
 292     return 0;
 293 }