Porting to FreeBSD, chapter 4

Started kqueue/kevent native support

@@ -2,9 +2,25 @@ ACLOCAL_AMFLAGS = -I m4
 
 if CLSYNC
 bin_PROGRAMS = clsync
-clsync_SOURCES = sync.c cluster.c main.c error.c fileutils.c malloc.c glibex.c
+clsync_SOURCES = sync.c cluster.c main.c error.c fileutils.c malloc.c glibex.c indexes.c
 clsync_CFLAGS = $(AM_CFLAGS)
 
+if HAVE_KQUEUE
+clsync_CFLAGS += -DKQUEUE_SUPPORT
+clsync_SOURCES += kqueue.c
+endif
+if HAVE_INOTIFY
+clsync_CFLAGS += -DINOTIFY_SUPPORT
+clsync_SOURCES += inotify.c
+if INOTIFY_OLD
+clsync_CFLAGS += -DINOTIFY_OLD
+endif
+endif
+if HAVE_FANOTIFY
+clsync_CFLAGS += -DFANOTIFY_SUPPORT
+clsync_SOURCES += fanotify.c
+endif
+
 if SOCKET
 clsync_SOURCES += socket.c control.c
 endif

@@ -18,6 +18,9 @@
  */
 
 
+#ifndef __CLSYNC_COMMON_H
+#define __CLSYNC_COMMON_H
+
 #define _GNU_SOURCE
 //#define _XOPEN_SOURCE 700
 #define _LARGEFILE64_SOURCE
@@ -38,11 +41,14 @@
 #include <errno.h>
 #include <ctype.h>
 #include <signal.h>
-#include <sys/inotify.h>
-#ifndef __FreeBSD__
-#	ifdef FANOTIFY_SUPPORT
-#		include <sys/fanotify.h>
-#	endif
+#ifdef KQUEUE_SUPPORT
+#	include <sys/event.h>
+#endif
+#ifdef INOTIFY_SUPPORT
+#	include <sys/inotify.h>
+#endif
+#ifdef FANOTIFY_SUPPORT
+#	include <sys/fanotify.h>
 #endif
 #include <sys/wait.h>
 #include <fts.h>
@@ -54,6 +60,7 @@
 #include <netinet/in.h>
 #include <libgen.h>
 #include <pthread.h>
+#include <glib.h>
 
 #ifdef HAVE_CAPABILITIES
 #	include <sys/capability.h>	// for capset()/capget() for --preserve-file-access
@@ -67,7 +74,6 @@
 
 #include "clsync.h"
 #include "ctx.h"
-#include "indexes.h"
 #include "program.h"
 
 #include <sys/param.h>
@@ -127,7 +133,12 @@ enum notifyengine_enum {
 #ifdef FANOTIFY_SUPPORT
 	NE_FANOTIFY,
 #endif
-	NE_INOTIFY
+#ifdef INOTIFY_SUPPORT
+	NE_INOTIFY,
+#endif
+#ifdef KQUEUE_SUPPORT
+	NE_KQUEUE,
+#endif
 };
 typedef enum notifyengine_enum notifyenfine_t;
 
@@ -225,7 +236,7 @@ struct dosync_arg {
 	char outf_path[PATH_MAX+1];
 	FILE *outf;
 	ctx_t *ctx_p;
-	indexes_t *indexes_p;
+	struct indexes *indexes_p;
 	void *data;
 	int linescount;
 	api_eventinfo_t *api_ei;
@@ -284,3 +295,5 @@ enum unified_evetnmask {
 	UEM_DELETED	= 0x04,
 };
 
+#endif
+

@@ -29,8 +29,16 @@
 // children count limit
 #define MAXCHILDREN			(1<<8)
 
+#if __linux__
+#	define DEFAULT_NOTIFYENGINE	NE_INOTIFY
+#else
+#	if __FreeBSD__
+#		define	DEFAULT_NOTIFYENGINE	NE_KQUEUE
+#	else
+#		define	DEFAULT_NOTIFYENGINE	NE_UNDEFINED
+#	endif
+#endif
 #define DEFAULT_RULES_PERM		RA_ALL
-#define DEFAULT_NOTIFYENGINE		NE_INOTIFY
 #define DEFAULT_COLLECTDELAY		30
 #define DEFAULT_SYNCDELAY		(DEFAULT_COLLECTDELAY)
 #define DEFAULT_BFILETHRESHOLD		(128 * 1024 * 1024)

@@ -15,22 +15,19 @@ LT_INIT
 AC_CANONICAL_HOST
 case $host_os in
 	*bsd*)
-		AC_SEARCH_LIBS([inotify_init], [inotify], [],
-			[AC_MSG_ERROR("Unable to find libinotify")])
 		AC_SEARCH_LIBS([backtrace], [execinfo], [],
 			[AC_MSG_ERROR("Unable to find libexecinfo")])
 
-		dnl pthread
 		AC_CHECK_LIB(pthread, pthread_create)
 		PTHREAD_LIBS=-pthread
 		AC_SUBST(PTHREAD_LIBS)
 		;;
 
 	linux*)
-		dnl pthread
 		AC_CHECK_LIB([pthread], [pthread_create],
 			[CPPFLAGS+=" -pthread" LDFLAGS+=" -pthread"],
 			[AC_MSG_ERROR("Pthread support is mandatory")])
+
 		;;
 esac
 
@@ -128,14 +125,115 @@ AS_HELP_STRING(--with-capabilities,
 [enable linux capabilities support, default: disabled]))
 
 AS_IF(
-    [test "x$with_capabilities" = "xyes"], [
-        AC_CHECK_HEADER(
-            [sys/capability.h],
-            [CPPFLAGS="${CPPFLAGS} -DHAVE_CAPABILITIES"],
-            [AC_MSG_ERROR("Unable to find sys/capability.h")])
-])
+	[test "x$with_capabilities" = "xyes"],
+		[AC_CHECK_HEADER(
+			[sys/capability.h],
+			[CPPFLAGS="${CPPFLAGS} -DHAVE_CAPABILITIES"],
+			[AC_MSG_ERROR("Unable to find sys/capability.h")]
+		)]
+)
+
+AC_ARG_WITH(kqueue,
+	AS_HELP_STRING(--with-kqueue,
+		[Enable kqueue support; values: no, native, lib, check; default: check]),
+	[],
+	[with_kqueue=check]
+)
+
+AC_ARG_WITH(inotify,
+	AS_HELP_STRING(--with-inotify,
+		[Enable inotify support; values: no, native, lib, check; default: check]),
+	[],
+	[with_inotify=check]
+)
+
+case "$with_kqueue" in
+	check)
+		AC_CHECK_FUNC([kqueue],
+			[
+				HAVE_KQUEUE=1
+			],
+			AC_CHECK_LIB([kqueue], [kqueue],
+				[
+					LDFLAGS="${LDFLAGS} -lkqueue"
+					HAVE_KQUEUE=1
+				],
+			)
+		)
+		;;
+	native)
+		AC_CHECK_FUNC([kqueue],
+			[
+				HAVE_KQUEUE=1
+			],
+			[
+				AC_MSG_FAILURE(
+					[There is no kqueue native support on this system])
+			]
+		)
+		;;
+	lib)
+		AC_CHECK_LIB([kqueue], [kqueue],
+			[
+				LDFLAGS="${LDFLAGS} -lkqueue"
+				HAVE_KQUEUE=1
+			],
+			[
+				AC_MSG_FAILURE(
+					[Cannot find libkqueue])
+			]
+		)
+		;;
+esac
+
+INOTIFY_LIBS=-linotify
+case "$with_inotify" in
+	check)
+		AC_CHECK_FUNC([inotify_init],
+			[
+				HAVE_INOTIFY=1
+			],
+			AC_CHECK_LIB([inotify], [inotify_init],
+				[
+					LDFLAGS="${LDFLAGS} -linotify"
+					HAVE_INOTIFY=1
+				],
+			)
+		)
+		;;
+	native)
+		AC_CHECK_FUNC([inotify_init],
+			[
+				HAVE_INOTIFY=1
+			],
+			[
+				AC_MSG_FAILURE(
+					[There is no inotify native support on this system])
+			]
+		)
+		;;
+	lib)
+		AC_CHECK_LIB([inotify], [inotify_init],
+			[
+				LDFLAGS="${LDFLAGS} -linotify"
+				HAVE_INOTIFY=1
+			],
+			[
+				AC_MSG_FAILURE(
+					[Cannot find libinotify])
+			]
+		)
+		;;
+esac
+
+AS_IF([test "$HAVE_INOTIFY" != ""], [AC_CHECK_FUNC([inotify_init1], [], [INOTIFY_OLD=1])])
 
+AM_CONDITIONAL([HAVE_KQUEUE],   [test "$HAVE_KQUEUE"   != ""])
+AM_CONDITIONAL([HAVE_INOTIFY],  [test "$HAVE_INOTIFY"  != ""])
+AM_CONDITIONAL([INOTIFY_OLD],   [test "$INOTIFY_OLD"   != ""])
+AM_CONDITIONAL([HAVE_FANOTIFY], [test "$HAVE_FANOTIFY" != ""])
 
+AS_IF([test "$HAVE_KQUEUE" = '' -a "$HAVE_INOTIFY" = '' -a "$HAVE_FANOTIFY" = ''], [AC_MSG_FAILURE([kqueue and inotify are not supported on this system])])
 
 LIBS="${GLIB_LIBS} ${LIBS}"
 AM_CPPFLAGS="${GLIB_CFLAGS}"

@@ -70,10 +70,6 @@ enum flags_enum {
 	DONTUNLINK	= 'U',
 	INITFULL	= 'F',
 	SYNCTIMEOUT	= 'k',
-#ifdef FANOTIFY_SUPPORT
-	FANOTIFY	= 'f',
-#endif
-	INOTIFY		= 'i',
 	LABEL		= 'l',
 	SHOW_VERSION	= 'V',
 
@@ -100,6 +96,16 @@ enum flags_enum {
 	DUMPDIR			= 17|OPTION_LONGOPTONLY,
 
 	CONFIGBLOCKINHERITS	= 18|OPTION_LONGOPTONLY,
+
+#ifdef FANOTIFY_SUPPORT
+	FANOTIFY		= 19|OPTION_LONGOPTONLY,
+#endif
+#ifdef INOTIFY_SUPPORT
+	INOTIFY			= 20|OPTION_LONGOPTONLY,
+#endif
+#ifdef KQUEUE_SUPPORT
+	KQUEUE			= 21|OPTION_LONGOPTONLY,
+#endif
 };
 typedef enum flags_enum flags_t;
 
@@ -170,6 +176,11 @@ struct api_functs {
 };
 typedef struct api_functs api_functs_t;
 
+struct notifyenginefuncts {
+	int (*wait)(int notify_d, struct ctx *ctx_p, struct indexes *indexes_p);
+	int (*handle)(int notify_d, struct ctx *ctx_p, struct indexes *indexes_p);
+};
+
 struct ctx {
 #ifndef LIBCLSYNC
 	uid_t uid;
@@ -225,6 +236,7 @@ struct ctx {
 	char *rulfpath;
 	char *listoutdir;
 	int notifyengine;
+	struct notifyenginefuncts notifyenginefunct;
 	int retries;
 	size_t bfilethreshold;
 	unsigned int syncdelay;

@@ -17,6 +17,9 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+#ifndef __CLSYNC_ERROR_H
+#define __CLSYNC_ERROR_H
+
 #define BACKTRACE_LENGTH	256
 
 extern void _critical( const char *const function_name, const char *fmt, ...);
@@ -47,3 +50,5 @@ enum outputmethod {
 };
 typedef enum outputmethod outputmethod_t;
 
+#endif
+

+/*
+    clsync - file tree sync utility based on inotify/kqueue
+    
+    Copyright (C) 2013-2014 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
+    
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+    
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#if 0
+int fanotify_loop(int fanotify_d, ctx_t *ctx_p, indexes_t *indexes_p) {
+	struct fanotify_event_metadata buf[BUFSIZ/sizeof(struct fanotify_event_metadata) + 1];
+	int state = STATE_RUNNING;
+	state_p = &state;
+
+	while(state != STATE_EXIT) {
+		struct fanotify_event_metadata *metadata;
+		size_t len = read(fanotify_d, (void *)buf, sizeof(buf)-sizeof(*buf));
+		metadata=buf;
+		if(len == -1) {
+			error("cannot read(%i, &metadata, sizeof(metadata)).", fanotify_d);
+			return errno;
+		}
+		while(FAN_EVENT_OK(metadata, len)) {
+			debug(2, "metadata->pid: %i; metadata->fd: %i", metadata->pid, metadata->fd);
+			if (metadata->fd != FAN_NOFD) {
+				if (metadata->fd >= 0) {
+					char *fpath = fd2fpath_malloc(metadata->fd);
+					sync_queuesync(fpath_rel, 0, ctx_p, indexes_p, QUEUE_AUTO);
+					debug(2, "Event %i on \"%s\".", metadata->mask, fpath);
+					free(fpath);
+				}
+			}
+			close(metadata->fd);
+			metadata = FAN_EVENT_NEXT(metadata, len);
+		}
+		int ret;
+		if((ret=sync_idle(fanotify_d, ctx_p, indexes_p))) {
+			error("got error while sync_idle().");
+			return ret;
+		}
+	}
+	return 0;
+}
+#endif
+

+/*
+    clsync - file tree sync utility based on inotify/kqueue
+    
+    Copyright (C) 2013-2014 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
+    
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+    
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+

+/*
+    clsync - file tree sync utility based on inotify/kqueue
+    
+    Copyright (C) 2013-2014 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
+    
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+    
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "common.h"
+#include "error.h"
+

@@ -22,6 +22,10 @@
 
 #include <glib.h>
 
+#include "common.h"
+#include "error.h"
+#include "malloc.h"
+
 struct indexes {
 	GHashTable *wd2fpath_ht;			// watching descriptor -> file path
 	GHashTable *fpath2wd_ht;			// file path -> watching descriptor
@@ -34,5 +38,129 @@ struct indexes {
 };
 typedef struct indexes indexes_t;
 
+// Removes necessary rows from hash_tables if some watching descriptor closed
+// Return: 0 on success, non-zero on fail
+
+static inline int indexes_remove_bywd(indexes_t *indexes_p, int wd) {
+	int ret=0;
+
+	char *fpath = g_hash_table_lookup(indexes_p->wd2fpath_ht, GINT_TO_POINTER(wd));
+
+	ret |= g_hash_table_remove(indexes_p->wd2fpath_ht, GINT_TO_POINTER(wd));
+	if(fpath == NULL) {
+		error("Cannot remove from index \"fpath2wd\" by wd %i.", wd);
+		return -1;
+	}
+	ret |= g_hash_table_remove(indexes_p->fpath2wd_ht, fpath);
+
+	return ret;
+}
+
+// Adds necessary rows to hash_tables if some watching descriptor opened
+// Return: 0 on success, non-zero on fail
+
+static inline int indexes_add_wd(indexes_t *indexes_p, int wd, const char *fpath_const, size_t fpathlen) {
+	debug(3, "indexes_add_wd(indexes_p, %i, \"%s\", %i)", wd, fpath_const, fpathlen);
+
+	char *fpath = xmalloc(fpathlen+1);
+	memcpy(fpath, fpath_const, fpathlen+1);
+	g_hash_table_insert(indexes_p->wd2fpath_ht, GINT_TO_POINTER(wd), fpath);
+	g_hash_table_insert(indexes_p->fpath2wd_ht, fpath, GINT_TO_POINTER(wd));
+
+	return 0;
+}
+
+// Lookups file path by watching descriptor from hash_tables
+// Return: file path on success, NULL on fail
+
+static inline char *indexes_wd2fpath(indexes_t *indexes_p, int wd) {
+	return g_hash_table_lookup(indexes_p->wd2fpath_ht, GINT_TO_POINTER(wd));
+}
+
+//
+
+static inline int indexes_fpath2wd(indexes_t *indexes_p, const char *fpath) {
+	gpointer gint_p = g_hash_table_lookup(indexes_p->fpath2wd_ht, fpath);
+	if(gint_p == NULL)
+		return -1;
+
+	return GPOINTER_TO_INT(gint_p);
+}
+
+static inline eventinfo_t *indexes_fpath2ei(indexes_t *indexes_p, const char *fpath) {
+	return (eventinfo_t *)g_hash_table_lookup(indexes_p->fpath2ei_ht, fpath);
+}
+
+static inline int indexes_fpath2ei_add(indexes_t *indexes_p, char *fpath, eventinfo_t *evinfo) {
+	debug(5, "\"%s\"", fpath);
+	g_hash_table_replace(indexes_p->fpath2ei_ht, fpath, evinfo);
+
+	return 0;
+}
+
+static inline int indexes_queueevent(indexes_t *indexes_p, char *fpath, eventinfo_t *evinfo, queue_id_t queue_id) {
+
+	g_hash_table_replace(indexes_p->fpath2ei_coll_ht[queue_id], fpath, evinfo);
+
+	debug(3, "indexes_queueevent(indexes_p, \"%s\", evinfo, %i). It's now %i events collected in queue %i.", fpath, queue_id, g_hash_table_size(indexes_p->fpath2ei_coll_ht[queue_id]), queue_id);
+	return 0;
+}
+
+static inline eventinfo_t *indexes_lookupinqueue(indexes_t *indexes_p, const char *fpath, queue_id_t queue_id) {
+	return (eventinfo_t *)g_hash_table_lookup(indexes_p->fpath2ei_coll_ht[queue_id], fpath);
+}
+
+static inline int indexes_queuelen(indexes_t *indexes_p, queue_id_t queue_id) {
+	return g_hash_table_size(indexes_p->fpath2ei_coll_ht[queue_id]);
+}
+
+static inline int indexes_removefromqueue(indexes_t *indexes_p, char *fpath, queue_id_t queue_id) {
+//	debug(3, "indexes_removefromqueue(indexes_p, \"%s\", %i).", fpath, queue_id);
+
+	g_hash_table_remove(indexes_p->fpath2ei_coll_ht[queue_id], fpath);
+
+	debug(3, "indexes_removefromqueue(indexes_p, \"%s\", %i). It's now %i events collected in queue %i.", fpath, queue_id, g_hash_table_size(indexes_p->fpath2ei_coll_ht[queue_id]), queue_id);
+	return 0;
+}
+
+static inline int indexes_addexclude(indexes_t *indexes_p, char *fpath, eventinfo_flags_t flags, queue_id_t queue_id) {
+	g_hash_table_replace(indexes_p->exc_fpath_coll_ht[queue_id], fpath, GINT_TO_POINTER(flags));
+
+	debug(3, "indexes_addexclude(indexes_p, \"%s\", %i). It's now %i events collected in queue %i.", fpath, queue_id, g_hash_table_size(indexes_p->exc_fpath_coll_ht[queue_id]), queue_id);
+	return 0;
+}
+
+static inline int indexes_addexclude_aggr(indexes_t *indexes_p, char *fpath, eventinfo_flags_t flags) {
+	debug(3, "indexes_addexclude_aggr(indexes_p, \"%s\", %u).", fpath, flags);
+
+	gpointer flags_gp = g_hash_table_lookup(indexes_p->exc_fpath_ht, fpath);
+	if(flags_gp != NULL)
+		flags |= GPOINTER_TO_INT(flags_gp);
+
+	// Removing extra flags
+	if((flags&(EVIF_RECURSIVELY | EVIF_CONTENTRECURSIVELY)) == (EVIF_RECURSIVELY | EVIF_CONTENTRECURSIVELY))
+		flags &= ~EVIF_CONTENTRECURSIVELY;
+
+	g_hash_table_replace(indexes_p->exc_fpath_ht, fpath, GINT_TO_POINTER(flags));
+
+	debug(3, "indexes_addexclude_aggr(indexes_p, \"%s\", flags): %u.", fpath, flags);
+	return 0;
+}
+
+static inline int indexes_outaggr_add(indexes_t *indexes_p, char *outline, eventinfo_flags_t flags) {
+	gpointer flags_gp = g_hash_table_lookup(indexes_p->out_lines_aggr_ht, outline);
+	if(flags_gp != NULL)
+		flags |= GPOINTER_TO_INT(flags_gp);
+
+	// Removing extra flags
+	if((flags&(EVIF_RECURSIVELY | EVIF_CONTENTRECURSIVELY)) == (EVIF_RECURSIVELY | EVIF_CONTENTRECURSIVELY))
+		flags &= ~EVIF_CONTENTRECURSIVELY;
+
+	g_hash_table_replace(indexes_p->out_lines_aggr_ht, outline, GINT_TO_POINTER(flags));
+
+	debug(3, "indexes_outaggr_aggr(indexes_p, \"%s\").", outline);
+	return 0;
+}
+
 #endif
 

+/*
+    clsync - file tree sync utility based on inotify/kqueue
+    
+    Copyright (C) 2013-2014 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
+    
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+    
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "common.h"
+#include "port-hacks.h"
+#include "error.h"
+#include "sync.h"
+#include "indexes.h"
+#include "inotify.h"
+
+int inotify_wait(int inotify_d, ctx_t *ctx_p, indexes_t *indexes_p) {
+	static struct timeval tv;
+	time_t tm = time(NULL);
+	long delay = ((unsigned long)~0 >> 1);
+
+	threadsinfo_t *threadsinfo_p = thread_info();
+
+	debug(4, "pthread_mutex_unlock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE])");
+	pthread_cond_broadcast(&threadsinfo_p->cond[PTHREAD_MUTEX_STATE]);
+	pthread_mutex_unlock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE]);
+
+
+	fd_set rfds;
+	FD_ZERO(&rfds);
+	FD_SET(inotify_d, &rfds);
+
+	long queue_id = 0;
+	while(queue_id < QUEUE_MAX) {
+		queueinfo_t *queueinfo = &ctx_p->_queues[queue_id++];
+
+		if(!queueinfo->stime)
+			continue;
+
+		if(queueinfo->collectdelay == COLLECTDELAY_INSTANT) {
+			debug(3, "There're events in instant queue (#%i), don't waiting.", queue_id-1);
+			return 0;
+		}
+
+		int qdelay = queueinfo->stime + queueinfo->collectdelay - tm;
+		debug(3, "queue #%i: %i %i %i -> %i", queue_id-1, queueinfo->stime, queueinfo->collectdelay, tm, qdelay);
+		if(qdelay < -(long)ctx_p->syncdelay)
+			qdelay = -(long)ctx_p->syncdelay;
+
+		delay = MIN(delay, qdelay);
+	}
+
+	long synctime_delay = ((long)ctx_p->synctime) - ((long)tm);
+	synctime_delay = synctime_delay > 0 ? synctime_delay : 0;
+
+	debug(3, "delay = MAX(%li, %li)", delay, synctime_delay);
+	delay = MAX(delay, synctime_delay);
+	delay = delay > 0 ? delay : 0;
+
+	if(ctx_p->flags[THREADING]) {
+		time_t _thread_nextexpiretime = thread_nextexpiretime();
+		debug(3, "thread_nextexpiretime == %i", _thread_nextexpiretime);
+		if(_thread_nextexpiretime) {
+			long thread_expiredelay = (long)thread_nextexpiretime() - (long)tm + 1; // +1 is to make "tm>threadinfo_p->expiretime" after select() definitely TRUE
+			debug(3, "thread_expiredelay == %i", thread_expiredelay);
+			thread_expiredelay = thread_expiredelay > 0 ? thread_expiredelay : 0;
+			debug(3, "delay = MIN(%li, %li)", delay, thread_expiredelay);
+			delay = MIN(delay, thread_expiredelay);
+		}
+	}
+
+	if((!delay) || (*state_p != STATE_RUNNING))
+		return 0;
+
+	if(ctx_p->flags[EXITONNOEVENTS]) { // zero delay if "--exit-on-no-events" is set
+		tv.tv_sec  = 0;
+		tv.tv_usec = 0;
+	} else {
+		debug(3, "sleeping for %li second(s).", SLEEP_SECONDS);
+		sleep(SLEEP_SECONDS);
+		delay = ((long)delay)>SLEEP_SECONDS ? delay-SLEEP_SECONDS : 0;
+
+		tv.tv_sec  = delay;
+		tv.tv_usec = 0;
+	}
+
+	debug(4, "pthread_mutex_lock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE])");
+	pthread_mutex_lock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE]);
+
+	if(*state_p != STATE_RUNNING)
+		return 0;
+
+	debug(4, "pthread_mutex_unlock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE])");
+	pthread_cond_broadcast(&threadsinfo_p->cond[PTHREAD_MUTEX_STATE]);
+	pthread_mutex_lock(&threadsinfo_p->mutex[PTHREAD_MUTEX_SELECT]);
+	pthread_mutex_unlock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE]);
+
+	debug(3, "select with timeout %li secs.", tv.tv_sec);
+	int ret = select(inotify_d+1, &rfds, NULL, NULL, &tv);
+
+	pthread_mutex_unlock(&threadsinfo_p->mutex[PTHREAD_MUTEX_SELECT]);
+
+	if((ret == -1) && (errno == EINTR)) {
+		errno = 0;
+		ret   = 0;
+	}
+
+	debug(4, "pthread_mutex_lock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE])");
+	pthread_mutex_lock(&threadsinfo_p->mutex[PTHREAD_MUTEX_STATE]);
+
+	if((ctx_p->flags[EXITONNOEVENTS]) && (ret == 0)) // if not events and "--exit-on-no-events" is set
+		*state_p = STATE_EXIT;
+
+	return ret;
+}
+
+#define INOTIFY_HANDLE_CONTINUE {\
+	ptr += sizeof(struct inotify_event) + event->len;\
+	count++;\
+	continue;\
+}
+
+int inotify_handle(int inotify_d, ctx_t *ctx_p, indexes_t *indexes_p) {
+	static struct timeval tv={0};
+
+	int count = 0;
+
+	fd_set rfds;
+	FD_ZERO(&rfds);
+	FD_SET(inotify_d, &rfds);
+
+	char   *path_rel	= NULL;
+	size_t  path_rel_len	= 0;
+	char   *path_full	= NULL;
+	size_t  path_full_size	= 0;
+	while (select(FD_SETSIZE, &rfds, NULL, NULL, &tv)) {
+
+		char buf[BUFSIZ + 1];
+		size_t r = read(inotify_d, buf, BUFSIZ);
+		if (r <= 0) {
+			error("Got error while reading events from inotify with read().");
+			count = -1;
+			goto l_inotify_handle_end;
+		}
+
+#ifdef PARANOID
+		g_hash_table_remove_all(indexes_p->fpath2ei_ht);
+#endif
+
+		char *ptr =  buf;
+		char *end = &buf[r];
+		while (ptr < end) {
+			struct inotify_event *event = (struct inotify_event *)ptr;
+
+			// Removing stale wd-s
+
+			if(event->mask & IN_IGNORED) {
+				debug(2, "Cleaning up info about watch descriptor %i.", event->wd);
+				indexes_remove_bywd(indexes_p, event->wd);
+				INOTIFY_HANDLE_CONTINUE;
+			}
+
+			// Getting path
+
+			char *fpath = indexes_wd2fpath(indexes_p, event->wd);
+
+			if(fpath == NULL) {
+				debug(2, "Event %p on stale watch (wd: %i).", (void *)(long)event->mask, event->wd);
+				INOTIFY_HANDLE_CONTINUE;
+			}
+			debug(2, "Event %p on \"%s\" (wd: %i; fpath: \"%s\").", (void *)(long)event->mask, event->len>0?event->name:"", event->wd, fpath);
+
+			// Getting full path
+
+			size_t path_full_memreq = strlen(fpath) + event->len + 2;
+			if (path_full_size < path_full_memreq) {
+				path_full      = xrealloc(path_full, path_full_memreq);
+				path_full_size = path_full_memreq;
+			}
+
+			if (event->len>0)
+				sprintf(path_full, "%s/%s", fpath, event->name);
+			else
+				sprintf(path_full, "%s", fpath);
+
+			// Getting infomation about file/dir/etc
+
+			stat64_t lstat;
+			mode_t st_mode;
+			size_t st_size;
+			if (lstat64(path_full, &lstat)) {
+				debug(2, "Cannot lstat(\"%s\", lstat). Seems, that the object disappeared.", path_full);
+				if(event->mask & IN_ISDIR)