octoleo
/
lsyncd
mirror of https://github.com/octoleo/lsyncd.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

cleanup of filedescriptors, added dryrun option

This commit is contained in:
Axel Kittenberger 2010-11-25 22:10:24 +00:00
parent 134f570ac4
commit 5519e2be8e
5 changed files with 210 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile.am
View File

@ -1,7 +1,7 @@
AUTOMAKE_OPTIONS = foreign AUTOMAKE_OPTIONS = foreign
CFLAGS = -Wall $(LIBLUA_CFLAGS) CFLAGS = -Wall $(LIBLUA_CFLAGS)
bin_PROGRAMS = lsyncd bin_PROGRAMS = lsyncd
lsyncd_SOURCES = lsyncd.c lsyncd.lua inotify.c lsyncd_SOURCES = lsyncd.h lsyncd.c lsyncd.lua inotify.c
lsyncd_LDADD = $(LIBLUA_LIBS) lsyncd_LDADD = $(LIBLUA_LIBS)
exampledir = $(datarootdir)/doc/@PACKAGE@ exampledir = $(datarootdir)/doc/@PACKAGE@
dist_example_DATA = \ dist_example_DATA = \

31
inotify.c
View File

@ -252,8 +252,13 @@ static char * readbuf = NULL;
* to the runner. * to the runner.
*/ */
static void static void
inotify_ready(lua_State *L, int fd, void *extra) inotify_ready(lua_State *L, struct observance *observance)
{ {
if (observance->fd != inotify_fd) {
logstring("Error",
"internal fail, inotify_ready on non-inotify file descriptor.");
exit(-1); // ERRNO
}
while(true) { while(true) {
ptrdiff_t len; ptrdiff_t len;
int err; int err;
@ -307,6 +312,18 @@ inotify_ready(lua_State *L, int fd, void *extra)
} }
} }
/**
* Called by function pointer when the core doesnt want the
* inotify fd anymore (term, hup or overflow)
*/
static void
inotify_tidy(struct observance *observance)
{
close(observance->fd);
free(readbuf);
readbuf = NULL;
}
/** /**
* registers inotify functions. * registers inotify functions.
*/ */
@ -338,16 +355,6 @@ open_inotify(lua_State *L) {
close_exec_fd(inotify_fd); close_exec_fd(inotify_fd);
non_block_fd(inotify_fd); non_block_fd(inotify_fd);
observe_fd(inotify_fd, inotify_ready, NULL, NULL); observe_fd(inotify_fd, inotify_ready, NULL, inotify_tidy, NULL);
}
/**
* closes inotify
*/
extern void
close_inotify() {
close(inotify_fd);
free(readbuf);
readbuf = NULL;
} }

209
lsyncd.c
View File

@ -158,7 +158,7 @@ add_logcat(const char *name, int priority)
return true; return true;
} }
if (!strcmp("scarce", name)) { if (!strcmp("scarce", name)) {
settings.log_level = LOG_ERR; settings.log_level = LOG_WARNING;
return true; return true;
} }
@ -364,26 +364,31 @@ struct pipemsg {
* writeable again * writeable again
*/ */
static void static void
pipe_writey(lua_State *L, int fd, void *extra) { pipe_writey(lua_State *L, struct observance *observance) {
struct pipemsg *pm = (struct pipemsg *) extra; struct pipemsg *pm = (struct pipemsg *) observance->extra;
int len = write(fd, pm->text + pm->pos, pm->tlen - pm->pos); int len = write(observance->fd, pm->text + pm->pos, pm->tlen - pm->pos);
bool do_close = false;
pm->pos += len; pm->pos += len;
if (len < 0) { if (len < 0) {
logstring("Normal", "broken pipe."); logstring("Normal", "broken pipe.");
do_close = true; nonobserve_fd(observance->fd);
} else if (pm->pos >= pm->tlen) { } else if (pm->pos >= pm->tlen) {
logstring("Debug", "finished pipe."); logstring("Debug", "finished pipe.");
do_close = true; nonobserve_fd(observance->fd);
}
if (do_close) {
close(fd);
free(pm->text);
free(extra);
unobserve_fd(fd);
} }
} }
/**
* TODO
*/
static void
pipe_tidy(struct observance *observance)
{
struct pipemsg *pm = (struct pipemsg *) observance->extra;
close(observance->fd);
free(pm->text);
free(pm);
}
/***************************************************************************** /*****************************************************************************
* helper routines. * helper routines.
@ -691,7 +696,6 @@ l_exec(lua_State *L)
len = write(pipefd[1], pipe_text, tlen); len = write(pipefd[1], pipe_text, tlen);
if (len < 0) { if (len < 0) {
logstring("Normal", "immediatly broken pipe."); logstring("Normal", "immediatly broken pipe.");
close(pipefd[0]);
} }
if (len == tlen) { if (len == tlen) {
/* usual and best case, the pipe accepted all input -> close */ /* usual and best case, the pipe accepted all input -> close */
@ -699,14 +703,13 @@ l_exec(lua_State *L)
logstring("Exec", "one-sweeped pipe"); logstring("Exec", "one-sweeped pipe");
} else { } else {
struct pipemsg *pm; struct pipemsg *pm;
logstring("Exec", "adding pipe observer"); logstring("Exec", "adding pipe observance");
pm = s_calloc(1, sizeof(struct pipemsg)); pm = s_calloc(1, sizeof(struct pipemsg));
pm->text = s_strdup(pipe_text); pm->text = s_strdup(pipe_text);
pm->tlen = tlen; pm->tlen = tlen;
pm->pos = len; pm->pos = len;
observe_fd(pipefd[1], NULL, pipe_writey, pm); observe_fd(pipefd[1], NULL, pipe_writey, pipe_tidy, pm);
} }
close(pipefd[0]);
} }
free(argv); free(argv);
lua_pushnumber(L, pid); lua_pushnumber(L, pid);
@ -887,6 +890,7 @@ l_configure(lua_State *L)
if (!settings.log_file) { if (!settings.log_file) {
settings.log_syslog = true; settings.log_syslog = true;
} }
// XXX
logstring("Debug", "daemonizing now."); logstring("Debug", "daemonizing now.");
if (daemon(0, 0)) { if (daemon(0, 0)) {
logstring("Error", "Failed to daemonize"); logstring("Error", "Failed to daemonize");
@ -973,38 +977,12 @@ load_runner_func(lua_State *L,
} }
/**
* An observer to be called when a file descritor becomes
* read-ready or write-ready.
*/
struct observer {
/**
* The file descriptor to observe.
*/
int fd;
/**
* Function to call when read becomes ready.
*/
void (*ready)(lua_State *L, int fd, void *extra);
/**
* Function to call when write becomes ready.
*/
void (*writey)(lua_State *L, int fd, void *extra);
/**
* Extra tokens to pass to the functions-
*/
void *extra;
};
/** /**
* List of file descriptor watches. * List of file descriptor watches.
*/ */
static struct observer * observers = NULL; static struct observance * observances = NULL;
static int observers_len = 0; static int observances_len = 0;
static int observers_size = 0; static int observances_size = 0;
/** /**
* List of file descriptors to unobserve. * List of file descriptors to unobserve.
@ -1012,14 +990,14 @@ static int observers_size = 0;
* not be altered, thus unobserve stores here the * not be altered, thus unobserve stores here the
* actions that will be delayed. * actions that will be delayed.
*/ */
static int *unobservers = NULL; static int *nonobservances = NULL;
static int unobservers_len = 0; static int nonobservances_len = 0;
static int unobservers_size = 0; static int nonobservances_size = 0;
/** /**
* true while the observers list is being handled. * true while the observances list is being handled.
*/ */
static bool observer_action = false; static bool observance_action = false;
/** /**
* Core watches a filedescriptor to become ready, * Core watches a filedescriptor to become ready,
@ -1027,83 +1005,88 @@ static bool observer_action = false;
*/ */
extern void extern void
observe_fd(int fd, observe_fd(int fd,
void (*ready)(lua_State *L, int fd, void *extra), void (*ready) (lua_State *L, struct observance *observance),
void (*writey)(lua_State *L, int fd, void *extra), void (*writey)(lua_State *L, struct observance *observance),
void *extra) void (*tidy) (struct observance *observance),
void *extra)
{ {
int pos; int pos;
if (observer_action) { if (observance_action) {
// TODO // TODO
logstring("Error", logstring("Error",
"Adding observers in ready/writey handlers not yet supported"); "Adding observance in ready/writey handlers not yet supported");
exit(-1); // ERRNO exit(-1); // ERRNO
} }
if (observers_len + 1 > observers_size) { if (observances_len + 1 > observances_size) {
observers_size = observers_len + 1; observances_size = observances_len + 1;
observers = s_realloc(observers, observances = s_realloc(observances,
observers_size * sizeof(struct observer)); observances_size * sizeof(struct observance));
} }
for(pos = 0; pos < observers_len; pos++) { for(pos = 0; pos < observances_len; pos++) {
if (observers[pos].fd <= fd) { if (observances[pos].fd <= fd) {
break; break;
} }
} }
if (observers[pos].fd == fd) { if (observances[pos].fd == fd) {
logstring("Error", logstring("Error",
"Observing already an observed file descriptor."); "Observing already an observed file descriptor.");
exit(-1); // ERRNO exit(-1); // ERRNO
} }
memmove(observers + pos + 1, observers + pos, memmove(observances + pos + 1, observances + pos,
(observers_len - pos) * (sizeof(struct observer))); (observances_len - pos) * (sizeof(struct observance)));
observers_len++; observances_len++;
observers[pos].fd = fd; observances[pos].fd = fd;
observers[pos].ready = ready; observances[pos].ready = ready;
observers[pos].writey = writey; observances[pos].writey = writey;
observers[pos].extra = extra; observances[pos].tidy = tidy;
observances[pos].extra = extra;
} }
/** /**
* Makes core no longer watch fd. * Makes core no longer watch fd.
*/ */
extern void extern void
unobserve_fd(int fd) nonobserve_fd(int fd)
{ {
int pos; int pos;
if (observer_action) { if (observance_action) {
/* this function is called through a ready/writey handler /* this function is called through a ready/writey handler
* while the core works through the observer list, thus * while the core works through the observances list, thus
* it does not alter the list, but stores this actions * it does not alter the list, but stores this actions
* on a stack * on a stack
*/ */
unobservers_len++; nonobservances_len++;
if (unobservers_len > unobservers_size) { if (nonobservances_len > nonobservances_size) {
unobservers_size = unobservers_len; nonobservances_size = nonobservances_len;
unobservers = s_realloc(unobservers, nonobservances = s_realloc(nonobservances,
unobservers_size * sizeof(int)); nonobservances_size * sizeof(int));
} }
unobservers[unobservers_len - 1] = fd; nonobservances[nonobservances_len - 1] = fd;
return; return;
} }
/* looks for the fd */ /* looks for the fd */
for(pos = 0; pos < observers_len; pos++) { for(pos = 0; pos < observances_len; pos++) {
if (observers[pos].fd == fd) { if (observances[pos].fd == fd) {
break; break;
} }
} }
if (pos >= observers_len) { if (pos >= observances_len) {
logstring("Error", logstring("Error",
"internal fail, not observer file descriptor in unobserve"); "internal fail, not observed file descriptor in nonobserve_fd()");
exit(-1); //ERRNO exit(-1); //ERRNO
} }
if (observances[pos].tidy) {
observances[pos].tidy(observances + pos);
}
/* and moves the list down */ /* and moves the list down */
memmove(observers + pos, observers + pos + 1, memmove(observances + pos, observances + pos + 1,
(observers_len - pos) * (sizeof(struct observer))); (observances_len - pos) * (sizeof(struct observance)));
observers_len--; observances_len--;
} }
/** /**
@ -1164,52 +1147,52 @@ masterloop(lua_State *L)
FD_ZERO(&rfds); FD_ZERO(&rfds);
FD_ZERO(&wfds); FD_ZERO(&wfds);
for(pi = 0; pi < observers_len; pi++) { for(pi = 0; pi < observances_len; pi++) {
int fd = observers[pi].fd; int fd = observances[pi].fd;
if (observers[pi].ready) { if (observances[pi].ready) {
FD_SET(fd, &rfds); FD_SET(fd, &rfds);
} }
if (observers[pi].writey) { if (observances[pi].writey) {
FD_SET(fd, &wfds); FD_SET(fd, &wfds);
} }
} }
/* the great select */ /* the great select */
pr = pselect( pr = pselect(
observers[observers_len - 1].fd + 1, observances[observances_len - 1].fd + 1,
&rfds, &wfds, NULL, &rfds, &wfds, NULL,
have_alarm ? &tv : NULL, &sigset); have_alarm ? &tv : NULL, &sigset);
if (pr >= 0) { if (pr >= 0) {
/* walks through the observers calling ready/writey fds */ /* walks through the observances calling ready/writey fds */
observer_action = true; observance_action = true;
for(pi = 0; pi < observers_len; pi++) { for(pi = 0; pi < observances_len; pi++) {
int fd = observers[pi].fd; struct observance *obs = observances + pi;
void *extra = observers[pi].extra; int fd = obs->fd;
if (hup || term) { if (hup || term) {
break; break;
} }
if (observers[pi].ready && FD_ISSET(fd, &rfds)) { if (obs->ready && FD_ISSET(fd, &rfds)) {
observers[pi].ready(L, fd, extra); obs->ready(L, obs);
} }
if (hup || term) { if (hup || term) {
break; break;
} }
if (unobservers_len > 0 && if (nonobservances_len > 0 &&
unobservers[unobservers_len - 1] == fd) { nonobservances[nonobservances_len - 1] == fd) {
/* ready() unobserved itself */ /* ready() nonobserved itself */
continue; continue;
} }
if (observers[pi].writey && FD_ISSET(fd, &wfds)) { if (obs->writey && FD_ISSET(fd, &wfds)) {
observers[pi].writey(L, fd, extra); obs->writey(L, obs);
} }
} }
observer_action = false; observance_action = false;
/* work through delayed unobserve_fd() calls */ /* work through delayed unobserve_fd() calls */
for (pi = 0; pi < unobservers_len; pi++) { for (pi = 0; pi < nonobservances_len; pi++) {
unobserve_fd(unobservers[pi]); nonobserve_fd(nonobservances[pi]);
} }
unobservers_len = 0; nonobservances_len = 0;
} }
} }
} }
@ -1305,6 +1288,7 @@ main1(int argc, char *argv[])
/* prepares logging early */ /* prepares logging early */
int i = 1; int i = 1;
add_logcat("Normal", LOG_NOTICE); add_logcat("Normal", LOG_NOTICE);
add_logcat("Warn", LOG_WARNING);
add_logcat("Error", LOG_ERR); add_logcat("Error", LOG_ERR);
while (i < argc) { while (i < argc) {
if (strcmp(argv[i], "-log") && strcmp(argv[i], "--log")) { if (strcmp(argv[i], "-log") && strcmp(argv[i], "--log")) {
@ -1528,6 +1512,16 @@ main1(int argc, char *argv[])
masterloop(L); masterloop(L);
/* cleanup */ /* cleanup */
{
/* close/frees observances */
int i;
for(i = 0; i < observances_len; i++) {
struct observance *obs = observances + i;
if (obs->tidy) {
obs->tidy(obs);
}
}
}
{ {
/* frees logging categories */ /* frees logging categories */
int ci; int ci;
@ -1556,7 +1550,6 @@ main1(int argc, char *argv[])
settings.log_level = 0, settings.log_level = 0,
settings.nodaemon = false, settings.nodaemon = false,
close_inotify();
lua_close(L); lua_close(L);
return 0; return 0;
} }

38
lsyncd.h
View File

@ -103,23 +103,43 @@ extern void non_block_fd(int fd);
/* Sets the close-on-exit flag for a file descriptor. */ /* Sets the close-on-exit flag for a file descriptor. */
extern void close_exec_fd(int fd); extern void close_exec_fd(int fd);
/**
* An observance calls functions when a file descritor becomes
* read-ready or write-ready.
*/
struct observance {
/* The file descriptor observed. */
int fd;
/* Function to call when read becomes ready. */
void (*ready)(lua_State *L, struct observance *observance);
/* Function to call when write becomes ready. */
void (*writey)(lua_State *L, struct observance *observance);
/* Function that tidies up, closes fd, frees extra data. etc. */
void (*tidy)(struct observance *observance);
/* Extra tokens to pass to the functions */
void *extra;
};
/* makes the core to observe a file descriptor */ /* makes the core to observe a file descriptor */
extern void observe_fd( extern void
int fd, observe_fd(int fd,
void (*ready)(lua_State *L, int fd, void *extra), void (*ready) (lua_State *L, struct observance *observance),
void (*writey)(lua_State *L, int fd, void *extra), void (*writey)(lua_State *L, struct observance *observance),
void *extra); void (*tidy) (struct observance *observance),
void *extra);
/* stops the core to observe a file descriptor */ /* stops the core to observe a file descriptor */
extern void unobserve_fd(int fd); extern void nonobserve_fd(int fd);
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* inotify * inotify
*/ */
extern void register_inotify(lua_State *L); extern void register_inotify(lua_State *L);
extern void open_inotify(lua_State *L); extern void open_inotify(lua_State *L);
extern void close_inotify();
#endif #endif

70
lsyncd.lua
View File

@ -1402,9 +1402,6 @@ local Syncs = (function()
end end
-- loads a default value for an option if not existent -- loads a default value for an option if not existent
if not settings then
settings = {}
end
local defaultValues = { local defaultValues = {
'action', 'action',
'collapse', 'collapse',
@ -2075,6 +2072,7 @@ USAGE:
lsyncd [OPTIONS] -rsyncssh [SOURCE] [HOST] [TARGETDIR] lsyncd [OPTIONS] -rsyncssh [SOURCE] [HOST] [TARGETDIR]
OPTIONS: OPTIONS:
-dryrun Subprocesses are not actually invoked.
-help Shows this -help Shows this
-log all Logs everything (debug) -log all Logs everything (debug)
-log scarce Logs errors only -log scarce Logs errors only
@ -2111,6 +2109,10 @@ function runner.configure(args)
-- a list of all valid --options -- a list of all valid --options
local options = { local options = {
-- log is handled by core already. -- log is handled by core already.
dryrun =
{0, function()
clSettings.dryrun=true
end},
log = log =
{1, nil}, {1, nil},
logfile = logfile =
@ -2407,13 +2409,42 @@ function spawn(agent, binary, ...)
if lsyncdStatus == "fade" then if lsyncdStatus == "fade" then
log("Normal", "ignored spawn processs since status fading") log("Normal", "ignored spawn processs since status fading")
end end
local pid = lsyncd.exec(binary, ...) if not settings.dryrun then
if pid and pid > 0 then local pid = lsyncd.exec(binary, ...)
if pid and pid > 0 then
local sync = InletControl.getSync()
local delay = InletControl.getDelay(agent)
if delay then
delay.status = "active"
sync.processes[pid] = delay
else
local dlist = InletControl.getDelayList(agent)
if not dlist then
error("spawning with an unknown agent", 2)
end
for k, d in pairs(dlist) do
if type(k) == "number" then
d.status = "active"
end
end
sync.processes[pid] = dlist
end
end
else
local a1, a2 = ...
if a1 ~= "<" then
log("Normal", "would call ", binary, " ", table.concat({...}, " "))
else
local aa = {...}
table.remove(aa, 1)
table.remove(aa, 1)
log("Normal", "would call ", binary, " ", table.concat(aa, " "),
"\ninput pipe <\n", a2)
end
local sync = InletControl.getSync() local sync = InletControl.getSync()
local delay = InletControl.getDelay(agent) local delay = InletControl.getDelay(agent)
if delay then if delay then
delay.status = "active" sync:removeDelay(delay)
sync.processes[pid] = delay
else else
local dlist = InletControl.getDelayList(agent) local dlist = InletControl.getDelayList(agent)
if not dlist then if not dlist then
@ -2421,10 +2452,9 @@ function spawn(agent, binary, ...)
end end
for k, d in pairs(dlist) do for k, d in pairs(dlist) do
if type(k) == "number" then if type(k) == "number" then
d.status = "active" sync:removeDelay(d)
end end
end end
sync.processes[pid] = dlist
end end
end end
end end
@ -2436,7 +2466,6 @@ function spawnShell(agent, command, ...)
return spawn(agent, "/bin/sh", "-c", command, "/bin/sh", ...) return spawn(agent, "/bin/sh", "-c", command, "/bin/sh", ...)
end end
----- -----
-- Comfort routine also for user. -- Comfort routine also for user.
-- Returns true if 'String' starts with 'Start' -- Returns true if 'String' starts with 'Start'
@ -2454,6 +2483,27 @@ function string.ends(String,End)
end end
------
-- Replaces default os.execute with a warning message,
-- doing the execute nevertheless, if not in dryrun.
--
local os_execute = os.execute
os.execute = function(...)
log("Warn", "using os.execute makes Lsyncd splutter, use spawn() instead.")
if not settings.dryrun then
os_execute(...)
else
log("Normal", "would os.execute ", table.concat({...}, " "))
end
end
-----
-- An empty settings table optionally for the config to
-- expand instead of replace.
--
settings = {}
--============================================================================ --============================================================================
-- Lsyncd default settings -- Lsyncd default settings
--============================================================================ --============================================================================