lsyncd.c

@@ -8,9 +8,9 @@
 #endif
 
 #include <sys/stat.h>
+#include <sys/times.h>
 #include <sys/types.h>
 #include <sys/wait.h>
-
 #include <dirent.h>
 #include <errno.h>
 #include <limits.h>
@@ -21,11 +21,26 @@
 #include <string.h>
 #include <unistd.h>
 
-
 #include <lua.h>
 #include <lualib.h>
 #include <lauxlib.h>
 
+/**
+ * Macros to compare times() values
+ * (borrowed from linux/jiffies.h)
+ *
+ * time_after(a,b) returns true if the time a is after time b.
+ */
+#define time_after(a,b)         ((long)(b) - (long)(a) < 0)
+#define time_before(a,b)        time_after(b,a)
+#define time_after_eq(a,b)      ((long)(a) - (long)(b) >= 0)
+#define time_before_eq(a,b)     time_after_eq(b,a)
+
+/**
+ * Number of inotifies to read max. at once from the kernel.
+ */
+#define INOTIFY_BUF_LEN     (64 * (sizeof(struct inotify_event) + 16))
+
 /**
  * The Lua part of lsyncd.
  */
@@ -57,6 +72,11 @@ struct settings settings = {0,};
  */
 volatile sig_atomic_t reset = 0;
 
+/**
+ * The kernels clock ticks per second.
+ */
+long clocks_per_sec; 
+
 /**
  * "secured" calloc.
  */
@@ -125,6 +145,17 @@ l_add_watch(lua_State *L)
 	return 1;
 }
 
+/**
+ * Returns (on Lua stack) the current kernels clock state (jiffies, times() call)
+ */
+static int
+l_now(lua_State *L) 
+{
+	clock_t c = times(NULL);
+	lua_pushinteger(L, c);
+	return 1;
+}
+
 /**
  * Executes a subprocess. Does not wait for it to return.
  * 
@@ -310,12 +341,13 @@ l_terminate(lua_State *L)
 }
 
 static const luaL_reg lsyncdlib[] = {
-		{"add_watch", l_add_watch},
+		{"add_watch", l_add_watch },
-		{"exec",      l_exec},
+		{"now",       l_now       },
-		{"real_dir",  l_real_dir},
+		{"exec",      l_exec      },
-		{"stackdump", l_stackdump},
+		{"real_dir",  l_real_dir  },
-		{"sub_dirs",  l_sub_dirs},
+		{"stackdump", l_stackdump },
-		{"terminate", l_terminate},
+		{"sub_dirs",  l_sub_dirs  },
+		{"terminate", l_terminate },
 		{NULL, NULL}
 };
 
@@ -344,7 +376,7 @@ get_settings(lua_State *L)
 		return;
 	}
 	
-	/* logfile */
+	/* get logfile */
 	lua_pushstring(L, "logfile");
 	lua_gettable(L, -2);
 	if (settings.logfile) {
@@ -355,6 +387,9 @@ get_settings(lua_State *L)
 		settings.logfile = s_strdup(luaL_checkstring(L, -1));
 	}
 	lua_pop(L, 1);
+	
+	/* pop the settings table */
+	lua_pop(L, 1);
 }
 
 /**
@@ -394,8 +429,10 @@ wait_startup(lua_State *L)
 			remaining++;
 		}
 	}
+	/* since contents are copied into pids[] pop the lua table */
+	lua_pop(L, 1);
 	
-	/* waits for the children */
+	/* starts waiting for the children */
 	while(remaining) {
 		/* argument for waitpid, and exitcode of child */
 		int status, exitcode;
@@ -436,18 +473,91 @@ wait_startup(lua_State *L)
 				if (newp == 0) {
 					remaining--;
 				}
+				/* does not break, in case there are duplicate pids (whyever) */
 			}
 		}
 	}
 	free(pids);
 }
 
+/**
+ * Normal operation happens in here.
+ */
+void
+masterloop(lua_State *L)
+{
+	while(!reset) {
+		char readbuf[INOTIFY_BUF_LEN];
+		int alarm_state;
+		clock_t now = times(NULL);
+		clock_t alarm_time;
+		bool do_read = false;
+		ssize_t len; 
+
+		/* query runner about soonest alarm  */
+		lua_getglobal(L, "lsyncd_get_alarm");
+		lua_pushnumber(L, now);
+		lua_call(L, 1, 2);
+		alarm_state = luaL_checkinteger(L, -2);
+		alarm_time = (clock_t) luaL_checknumber(L, -1);
+		lua_pop(L, 2);
+
+	
+		if (alarm_state < 0) {
+			/* there is a delay that wants to be handled already                  */
+			/* thus do not read from inotify_fd and jump directly to its handling */
+			printf("core: immediately handling delayed entries\n");
+			do_read = 0;
+		} else if (alarm_state > 0) {
+			/* use select() to determine what happens next      */
+			/* + a new event on inotify                         */
+			/* + an alarm on timeout                            */
+			/* + the return of a child process                  */
+			fd_set readfds;
+			struct timeval tv;
+
+			if (time_after(now, alarm_time)) {
+				/* should never happen */
+				printf("Internal failure, alarm_time is in past!\n");
+				exit(-1); //ERRNO
+			}
+
+			tv.tv_sec  = (alarm_time - now) / clocks_per_sec;
+			tv.tv_usec = (alarm_time - now) * 1000000 / clocks_per_sec % 1000000;
+			/* if select returns a positive number there is data on inotify */
+			/* on zero the timemout occured.                                */
+			FD_ZERO(&readfds);
+			FD_SET(inotify_fd, &readfds);
+			do_read = select(inotify_fd + 1, &readfds, NULL, NULL, &tv);
+
+			if (do_read) {
+				printf("core: theres data on inotify.\n");
+			} else {
+				printf("core: select() timeout, doing delays.\n");
+			}
+		} else {
+			// if nothing to wait for, enter a blocking read
+			printf("core: gone blocking\n");
+			do_read = 1;
+		}
+
+		if (do_read) {
+			len = read (inotify_fd, readbuf, INOTIFY_BUF_LEN);
+		} else {
+			len = 0;
+		}
+	}
+}
+
 /**
  * Main
  */
 int
 main(int argc, char *argv[])
 {
+	/* kernel parameters */
+	clocks_per_sec = sysconf(_SC_CLK_TCK);
+
 	/* the Lua interpreter */
 	lua_State* L;
 
@@ -516,7 +626,9 @@ main(int argc, char *argv[])
 
 	/* enter normal operation */
 	lua_getglobal(L, "normalop");
-	lua_call(L, 0, 1);
+	lua_call(L, 0, 0);
+
+	masterloop(L);
 
 	/* cleanup */
 	close(inotify_fd);

lsyncd.lua

@@ -77,6 +77,19 @@ function lsyncd_initialize()
 	end
 end
 
+----
+-- Calle by core to determine soonest alarm.
+--
+-- @param  now   ... the current time representation.
+--
+-- @return two variables.
+--         number -1 means ... alarm is in the past.
+--                 0 means ... no alarm, core can in untimed sleep
+--                 1 means ... alarm time specified.
+--         times           ... the alarm time (only read if number is 1)
+function lsyncd_get_alarm()
+	return 0, 0
+end
 
 ------------------------------------------------------------------------------
 -- lsyncd user interface
@@ -104,7 +117,8 @@ function default_startup()
 	print("--- startup ---")
 	local pids = { }
 	for i, o in ipairs(origins) do
-		print("initialize recursive rsync: " .. o.source .. " -> " .. o.targetpath)
+		print("startup recursive rsync: " .. o.source .. " -> " .. o.targetpath)
+		-- TODO userchangeablefunction
 		pid = lsyncd.exec("/usr/bin/rsync", "-ltrs", o.source, o.targetpath)
 		table.insert(pids, pid)
 	end