1 /* beanstalk - fast, general-purpose work queue */
3 /* Copyright (C) 2007 Keith Rarick and Philotic Inc.
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.
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.
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/>.
26 #include <sys/resource.h>
27 #include <sys/socket.h>
28 #include <netinet/in.h>
29 #include <arpa/inet.h>
30 #include <sys/types.h>
41 static char *user
= NULL
;
42 static int detach
= 0;
43 static int port
= 11300;
44 static struct in_addr host_addr
;
47 nullfd(int fd
, int flags
)
52 r
= open("/dev/null", flags
);
53 if (r
!= fd
) twarn("open(\"/dev/null\")"), exit(1);
72 if (r
) return twarn("chdir");
84 su(const char *user
) {
89 pwent
= getpwnam(user
);
90 if (errno
) twarn("getpwnam(\"%s\")", user
), exit(32);
91 if (!pwent
) twarnx("getpwnam(\"%s\"): no such user", user
), exit(33);
93 r
= setgid(pwent
->pw_gid
);
94 if (r
== -1) twarn("setgid(%d \"%s\")", pwent
->pw_gid
, user
), exit(34);
96 r
= setuid(pwent
->pw_uid
);
97 if (r
== -1) twarn("setuid(%d \"%s\")", pwent
->pw_uid
, user
), exit(34);
101 exit_cleanly(int sig
)
114 sa
.sa_handler
= SIG_IGN
;
116 r
= sigemptyset(&sa
.sa_mask
);
117 if (r
== -1) twarn("sigemptyset()"), exit(111);
119 r
= sigaction(SIGPIPE
, &sa
, 0);
120 if (r
== -1) twarn("sigaction(SIGPIPE)"), exit(111);
122 sa
.sa_handler
= enter_drain_mode
;
123 r
= sigaction(SIGUSR1
, &sa
, 0);
124 if (r
== -1) twarn("sigaction(SIGUSR1)"), exit(111);
126 sa
.sa_handler
= exit_cleanly
;
127 r
= sigaction(SIGINT
, &sa
, 0);
128 if (r
== -1) twarn("sigaction(SIGINT)"), exit(111);
130 sa
.sa_handler
= exit_cleanly
;
131 r
= sigaction(SIGTERM
, &sa
, 0);
132 if (r
== -1) twarn("sigaction(SIGTERM)"), exit(111);
135 /* This is a workaround for a mystifying workaround in libevent's epoll
136 * implementation. The epoll_init() function creates an epoll fd with space to
137 * handle RLIMIT_NOFILE - 1 fds, accompanied by the following puzzling comment:
138 * "Solaris is somewhat retarded - it's important to drop backwards
139 * compatibility when making changes. So, don't dare to put rl.rlim_cur here."
140 * This is presumably to work around a bug in Solaris, but it has the
141 * unfortunate side-effect of causing epoll_ctl() (and, therefore, event_add())
142 * to fail for a valid fd if we have hit the limit of open fds. That makes it
143 * hard to provide reasonable behavior in that situation. So, let's reduce the
144 * real value of RLIMIT_NOFILE by one, after epoll_init() has run. */
151 r
= getrlimit(RLIMIT_NOFILE
, &rl
);
152 if (r
!= 0) twarn("getrlimit(RLIMIT_NOFILE)"), exit(2);
156 r
= setrlimit(RLIMIT_NOFILE
, &rl
);
157 if (r
!= 0) twarn("setrlimit(RLIMIT_NOFILE)"), exit(2);
161 usage(char *msg
, char *arg
)
163 if (arg
) warnx("%s: %s", msg
, arg
);
164 fprintf(stderr
, "Use: %s [OPTIONS]\n"
168 " -b DIR binlog directory\n"
169 " -f MS fsync at most once every MS milliseconds"
170 " (use -f 0 for \"always fsync\")\n"
171 " -F never fsync (default)\n"
172 " -l ADDR listen on address (default is 0.0.0.0)\n"
173 " -p PORT listen on port (default is 11300)\n"
174 " -u USER become user and group\n"
175 " -z SIZE set the maximum job size in bytes (default is %d)\n"
176 " -s SIZE set the size of each binlog file (default is %d)\n"
177 " -v show version information\n"
178 " -h show this help\n",
179 progname
, JOB_DATA_SIZE_LIMIT_DEFAULT
, BINLOG_SIZE_LIMIT_DEFAULT
);
184 parse_size_t(char *str
)
189 r
= sscanf(str
, "%zu%c", &size
, &x
);
190 if (1 != r
) usage("invalid size", str
);
195 require_arg(char *opt
, char *arg
)
197 if (!arg
) usage("option requires an argument", opt
);
202 parse_port(char *portstr
)
208 port
= strtol(portstr
, &end
, 10);
209 if (end
== portstr
) usage("invalid port", portstr
);
210 if (end
[0] != 0) usage("invalid port", portstr
);
211 if (errno
) usage("invalid port", portstr
);
216 static struct in_addr
217 parse_host(char *hoststr
)
222 r
= inet_aton(hoststr
, &addr
);
223 if (!r
) usage("invalid address", hoststr
);
229 opts(int argc
, char **argv
)
233 for (i
= 1; i
< argc
; ++i
) {
234 if (argv
[i
][0] != '-') usage("unknown option", argv
[i
]);
235 if (argv
[i
][1] == 0 || argv
[i
][2] != 0) usage("unknown option",argv
[i
]);
236 switch (argv
[i
][1]) {
241 port
= parse_port(require_arg("-p", argv
[++i
]));
244 host_addr
= parse_host(require_arg("-l", argv
[++i
]));
247 job_data_size_limit
= parse_size_t(require_arg("-z",
251 binlog_size_limit
= parse_size_t(require_arg("-s", argv
[++i
]));
254 fsync_throttle_ms
= parse_size_t(require_arg("-f", argv
[++i
]));
261 user
= require_arg("-u", argv
[++i
]);
264 binlog_dir
= require_arg("-b", argv
[++i
]);
269 printf("beanstalkd %s\n", VERSION
);
272 usage("unknown option", argv
[i
]);
278 main(int argc
, char **argv
)
281 struct event_base
*ev_base
;
282 struct job binlog_jobs
= {};
284 host_addr
.s_addr
= INADDR_ANY
;
292 /* We want to make sure that only one beanstalkd tries to use the binlog
293 * directory at a time. So acquire a lock now and never release it. */
296 if (!r
) twarnx("failed to lock binlog dir %s", binlog_dir
), exit(10);
299 r
= make_server_socket(host_addr
, port
);
300 if (r
== -1) twarnx("make_server_socket()"), exit(111);
303 ev_base
= event_init();
307 unbrake((evh
) h_accept
);
309 binlog_jobs
.prev
= binlog_jobs
.next
= &binlog_jobs
;
310 binlog_init(&binlog_jobs
);
311 prot_replay_binlog(&binlog_jobs
);
315 event_reinit(ev_base
);
320 twarnx("got here for some reason");