[PATCH 0/3] Use udev monitor

[PATCH 0/3] Use udev monitor

[Bastian Blank]

This patch set rips out the custom synchronisation option via sysv
semaphores and replaces it with udev monitor via netlink.

Bastian Blank (3):
  udev: Rip out semaphore synchronisation support
  udev: Save open udev state
  udev: Use udev monitor

 libdm/libdm-common.c |  368 +++++++++++++++-----------------------------------
 1 file changed, 107 insertions(+), 261 deletions(-)

-- 
Totally illogical, there was no chance.
		-- Spock, "The Galileo Seven", stardate 2822.3

[PATCH 1/3] udev: Rip out semaphore synchronisation support

[Bastian Blank]

---
 libdm/libdm-common.c |  284 +++-----------------------------------------------
 1 file changed, 13 insertions(+), 271 deletions(-)

diff --git a/libdm/libdm-common.c b/libdm/libdm-common.c
index d18de9c..bcddc3b 100644
--- a/libdm/libdm-common.c
+++ b/libdm/libdm-common.c
@@ -26,9 +26,6 @@
 #include <dirent.h>
 
 #ifdef UDEV_SYNC_SUPPORT
-#  include <sys/types.h>
-#  include <sys/ipc.h>
-#  include <sys/sem.h>
 #  include <libudev.h>
 #endif
 
@@ -47,18 +44,6 @@
 
 #define DEV_DIR "/dev/"
 
-#ifdef UDEV_SYNC_SUPPORT
-#ifdef _SEM_SEMUN_UNDEFINED
-union semun
-{
-	int val;			/* value for SETVAL */
-	struct semid_ds *buf;		/* buffer for IPC_STAT & IPC_SET */
-	unsigned short int *array;	/* array for GETALL & SETALL */
-	struct seminfo *__buf;		/* buffer for IPC_INFO */
-};
-#endif
-#endif
-
 static char _dm_dir[PATH_MAX] = DEV_DIR DM_DIR;
 static char _sysfs_dir[PATH_MAX] = "/sys/";
 static char _path0[PATH_MAX];           /* path buffer, safe 4kB on stack */
@@ -78,7 +63,6 @@ static struct selabel_handle *_selabel_handle = NULL;
 static int _udev_disabled = 0;
 
 #ifdef UDEV_SYNC_SUPPORT
-static int _semaphore_supported = -1;
 static int _udev_running = -1;
 static int _sync_with_udev = 1;
 static int _udev_checking = 1;
@@ -1993,24 +1977,6 @@ int dm_udev_wait(uint32_t cookie)
 
 #else		/* UDEV_SYNC_SUPPORT */
 
-static int _check_semaphore_is_supported(void)
-{
-	int maxid;
-	union semun arg;
-	struct seminfo seminfo;
-
-	arg.__buf = &seminfo;
-	maxid = semctl(0, 0, SEM_INFO, arg);
-
-	if (maxid < 0) {
-		log_warn("Kernel not configured for semaphores (System V IPC). "
-			 "Not using udev synchronisation code.");
-		return 0;
-	}
-
-	return 1;
-}
-
 static int _check_udev_is_running(void)
 {
 	struct udev *udev;
@@ -2041,9 +2007,6 @@ bad:
 
 static void _check_udev_sync_requirements_once(void)
 {
-	if (_semaphore_supported < 0)
-		_semaphore_supported = _check_semaphore_is_supported();
-
 	if (_udev_running < 0) {
 		_udev_running = _check_udev_is_running();
 		if (_udev_disabled && _udev_running)
@@ -2063,7 +2026,7 @@ int dm_udev_get_sync_support(void)
 {
 	_check_udev_sync_requirements_once();
 
-	return !_udev_disabled && _semaphore_supported &&
+	return !_udev_disabled &&
 		dm_cookie_supported() &&_udev_running && _sync_with_udev;
 }
 
@@ -2080,125 +2043,11 @@ int dm_udev_get_checking(void)
 	return _udev_checking;
 }
 
-static int _get_cookie_sem(uint32_t cookie, int *semid)
-{
-	if (cookie >> 16 != DM_COOKIE_MAGIC) {
-		log_error("Could not continue to access notification "
-			  "semaphore identified by cookie value %"
-			  PRIu32 " (0x%x). Incorrect cookie prefix.",
-			  cookie, cookie);
-		return 0;
-	}
-
-	if ((*semid = semget((key_t) cookie, 1, 0)) >= 0)
-		return 1;
-
-	switch (errno) {
-		case ENOENT:
-			log_error("Could not find notification "
-				  "semaphore identified by cookie "
-				  "value %" PRIu32 " (0x%x)",
-				  cookie, cookie);
-			break;
-		case EACCES:
-			log_error("No permission to access "
-				  "notificaton semaphore identified "
-				  "by cookie value %" PRIu32 " (0x%x)",
-				  cookie, cookie);
-			break;
-		default:
-			log_error("Failed to access notification "
-				   "semaphore identified by cookie "
-				   "value %" PRIu32 " (0x%x): %s",
-				  cookie, cookie, strerror(errno));
-			break;
-	}
-
-	return 0;
-}
-
-static int _udev_notify_sem_inc(uint32_t cookie, int semid)
-{
-	struct sembuf sb = {0, 1, 0};
-	int val;
-
-	if (semop(semid, &sb, 1) < 0) {
-		log_error("semid %d: semop failed for cookie 0x%" PRIx32 ": %s",
-			  semid, cookie, strerror(errno));
-		return 0;
-	}
-
- 	if ((val = semctl(semid, 0, GETVAL)) < 0) {
-		log_error("semid %d: sem_ctl GETVAL failed for "
-			  "cookie 0x%" PRIx32 ": %s",
-			  semid, cookie, strerror(errno));
-		return 0;		
-	}
-
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) incremented to %d",
-		  cookie, semid, val);
-
-	return 1;
-}
-
-static int _udev_notify_sem_dec(uint32_t cookie, int semid)
-{
-	struct sembuf sb = {0, -1, IPC_NOWAIT};
-	int val;
-
- 	if ((val = semctl(semid, 0, GETVAL)) < 0) {
-		log_error("semid %d: sem_ctl GETVAL failed for "
-			  "cookie 0x%" PRIx32 ": %s",
-			  semid, cookie, strerror(errno));
-		return 0;
-	}
-
-	if (semop(semid, &sb, 1) < 0) {
-		switch (errno) {
-			case EAGAIN:
-				log_error("semid %d: semop failed for cookie "
-					  "0x%" PRIx32 ": "
-					  "incorrect semaphore state",
-					  semid, cookie);
-				break;
-			default:
-				log_error("semid %d: semop failed for cookie "
-					  "0x%" PRIx32 ": %s",
-					  semid, cookie, strerror(errno));
-				break;
-		}
-		return 0;
-	}
-
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) decremented to %d",
-			     cookie, semid, val - 1);
-
-	return 1;
-}
-
-static int _udev_notify_sem_destroy(uint32_t cookie, int semid)
-{
-	if (semctl(semid, 0, IPC_RMID, 0) < 0) {
-		log_error("Could not cleanup notification semaphore "
-			  "identified by cookie value %" PRIu32 " (0x%x): %s",
-			  cookie, cookie, strerror(errno));
-		return 0;
-	}
-
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) destroyed", cookie,
-			     semid);
-
-	return 1;
-}
-
-static int _udev_notify_sem_create(uint32_t *cookie, int *semid)
+static int _udev_create_cookie(uint32_t *cookie)
 {
 	int fd;
-	int gen_semid;
-	int val;
 	uint16_t base_cookie;
 	uint32_t gen_cookie;
-	union semun sem_arg;
 
 	if ((fd = open("/dev/urandom", O_RDONLY)) < 0) {
 		log_error("Failed to open /dev/urandom "
@@ -2208,68 +2057,19 @@ static int _udev_notify_sem_create(uint32_t *cookie, int *semid)
 	}
 
 	/* Generate random cookie value. Be sure it is unique and non-zero. */
-	do {
-		/* FIXME Handle non-error returns from read(). Move _io() into libdm? */
-		if (read(fd, &base_cookie, sizeof(base_cookie)) != sizeof(base_cookie)) {
-			log_error("Failed to initialize notification cookie");
-			goto bad;
-		}
-
-		gen_cookie = DM_COOKIE_MAGIC << 16 | base_cookie;
-
-		if (base_cookie && (gen_semid = semget((key_t) gen_cookie,
-				    1, 0600 | IPC_CREAT | IPC_EXCL)) < 0) {
-			switch (errno) {
-				case EEXIST:
-					/* if the semaphore key exists, we
-					 * simply generate another random one */
-					base_cookie = 0;
-					break;
-				case ENOMEM:
-					log_error("Not enough memory to create "
-						  "notification semaphore");
-					goto bad;
-				case ENOSPC:
-					log_error("Limit for the maximum number "
-						  "of semaphores reached. You can "
-						  "check and set the limits in "
-						  "/proc/sys/kernel/sem.");
-					goto bad;
-				default:
-					log_error("Failed to create notification "
-						  "semaphore: %s", strerror(errno));
-					goto bad;
-			}
-		}
-	} while (!base_cookie);
-
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) created",
-			     gen_cookie, gen_semid);
-
-	sem_arg.val = 1;
-
-	if (semctl(gen_semid, 0, SETVAL, sem_arg) < 0) {
-		log_error("semid %d: semctl failed: %s", gen_semid, strerror(errno));
-		/* We have to destroy just created semaphore
-		 * so it won't stay in the system. */
-		(void) _udev_notify_sem_destroy(gen_cookie, gen_semid);
+	/* FIXME Handle non-error returns from read(). Move _io() into libdm? */
+	if (read(fd, &base_cookie, sizeof(base_cookie)) != sizeof(base_cookie)) {
+		log_error("Failed to initialize notification cookie");
 		goto bad;
 	}
 
- 	if ((val = semctl(gen_semid, 0, GETVAL)) < 0) {
-		log_error("semid %d: sem_ctl GETVAL failed for "
-			  "cookie 0x%" PRIx32 ": %s",
-			  gen_semid, gen_cookie, strerror(errno));
-		goto bad;
-	}
+	gen_cookie = DM_COOKIE_MAGIC << 16 | base_cookie;
 
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) incremented to %d",
-			     gen_cookie, gen_semid, val);
+	log_debug_activation("Udev cookie 0x%" PRIx32 " created", gen_cookie);
 
 	if (close(fd))
 		stack;
 
-	*semid = gen_semid;
 	*cookie = gen_cookie;
 
 	return 1;
@@ -2285,14 +2085,12 @@ bad:
 
 int dm_udev_create_cookie(uint32_t *cookie)
 {
-	int semid;
-
 	if (!dm_udev_get_sync_support()) {
 		*cookie = 0;
 		return 1;
 	}
 
-	return _udev_notify_sem_create(cookie, &semid);
+	return _udev_create_cookie(cookie);
 }
 
 static const char *_task_type_disp(int type)
@@ -2342,8 +2140,6 @@ static const char *_task_type_disp(int type)
 
 int dm_task_set_cookie(struct dm_task *dmt, uint32_t *cookie, uint16_t flags)
 {
-	int semid;
-
 	_set_cookie_flags(dmt, flags);
 
 	if (!dm_udev_get_sync_support()) {
@@ -2352,24 +2148,14 @@ int dm_task_set_cookie(struct dm_task *dmt, uint32_t *cookie, uint16_t flags)
 		return 1;
 	}
 
-	if (*cookie) {
-		if (!_get_cookie_sem(*cookie, &semid))
-			goto_bad;
-	} else if (!_udev_notify_sem_create(cookie, &semid))
+	if (!*cookie && !_udev_create_cookie(cookie))
 		goto_bad;
 
-	if (!_udev_notify_sem_inc(*cookie, semid)) {
-		log_error("Could not set notification semaphore "
-			  "identified by cookie value %" PRIu32 " (0x%x)",
-			  *cookie, *cookie);
-		goto bad;
-	}
-
 	dmt->event_nr |= ~DM_UDEV_FLAGS_MASK & *cookie;
 	dmt->cookie_set = 1;
 
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) assigned to "
-			     "%s task(%d) with flags%s%s%s%s%s%s%s (0x%" PRIx16 ")", *cookie, semid, _task_type_disp(dmt->type), dmt->type, 
+	log_debug_activation("Udev cookie 0x%" PRIx32 " assigned to "
+			     "%s task(%d) with flags%s%s%s%s%s%s%s (0x%" PRIx16 ")", *cookie, _task_type_disp(dmt->type), dmt->type, 
 			     (flags & DM_UDEV_DISABLE_DM_RULES_FLAG) ? " DISABLE_DM_RULES" : "",
 			     (flags & DM_UDEV_DISABLE_SUBSYSTEM_RULES_FLAG) ? " DISABLE_SUBSYSTEM_RULES" : "",
 			     (flags & DM_UDEV_DISABLE_DISK_RULES_FLAG) ? " DISABLE_DISK_RULES" : "",
@@ -2388,62 +2174,18 @@ bad:
 
 int dm_udev_complete(uint32_t cookie)
 {
-	int semid;
-
 	if (!cookie || !dm_udev_get_sync_support())
 		return 1;
 
-	if (!_get_cookie_sem(cookie, &semid))
-		return_0;
-
-	if (!_udev_notify_sem_dec(cookie, semid)) {
-		log_error("Could not signal waiting process using notification "
-			  "semaphore identified by cookie value %" PRIu32 " (0x%x)",
-			  cookie, cookie);
-		return 0;
-	}
-
-	return 1;
+	return_0;
 }
 
 static int _udev_wait(uint32_t cookie)
 {
-	int semid;
-	struct sembuf sb = {0, 0, 0};
-
 	if (!cookie || !dm_udev_get_sync_support())
 		return 1;
 
-	if (!_get_cookie_sem(cookie, &semid))
-		return_0;
-
-	if (!_udev_notify_sem_dec(cookie, semid)) {
-		log_error("Failed to set a proper state for notification "
-			  "semaphore identified by cookie value %" PRIu32 " (0x%x) "
-			  "to initialize waiting for incoming notifications.",
-			  cookie, cookie);
-		(void) _udev_notify_sem_destroy(cookie, semid);
-		return 0;
-	}
-
-	log_debug_activation("Udev cookie 0x%" PRIx32 " (semid %d) waiting for zero",
-			     cookie, semid);
-
-repeat_wait:
-	if (semop(semid, &sb, 1) < 0) {
-		if (errno == EINTR)
-			goto repeat_wait;
-		else if (errno == EIDRM)
-			return 1;
-
-		log_error("Could not set wait state for notification semaphore "
-			  "identified by cookie value %" PRIu32 " (0x%x): %s",
-			  cookie, cookie, strerror(errno));
-		(void) _udev_notify_sem_destroy(cookie, semid);
-		return 0;
-	}
-
-	return _udev_notify_sem_destroy(cookie, semid);
+	return_0;
 }
 
 int dm_udev_wait(uint32_t cookie)
-- 
You're dead, Jim.
		-- McCoy, "The Tholian Web", stardate unknown

[PATCH 2/3] udev: Save open udev state

[Bastian Blank]

---
 libdm/libdm-common.c |   23 ++++++++++++++++++-----
 1 file changed, 18 insertions(+), 5 deletions(-)

diff --git a/libdm/libdm-common.c b/libdm/libdm-common.c
index bcddc3b..d028029 100644
--- a/libdm/libdm-common.c
+++ b/libdm/libdm-common.c
@@ -63,6 +63,7 @@ static struct selabel_handle *_selabel_handle = NULL;
 static int _udev_disabled = 0;
 
 #ifdef UDEV_SYNC_SUPPORT
+static struct udev *_udev;
 static int _udev_running = -1;
 static int _sync_with_udev = 1;
 static int _udev_checking = 1;
@@ -1977,17 +1978,30 @@ int dm_udev_wait(uint32_t cookie)
 
 #else		/* UDEV_SYNC_SUPPORT */
 
+static struct udev *_get_udev(void)
+{
+	if (!_udev)
+		_udev = udev_new();
+	if (!_udev)
+		goto_bad;
+
+	return _udev;
+
+bad:
+	log_error("Could not get udev state. Assuming udev is not running.");
+	return 0;
+}
+
 static int _check_udev_is_running(void)
 {
 	struct udev *udev;
 	struct udev_queue *udev_queue;
 	int r;
 
-	if (!(udev = udev_new()))
-		goto_bad;
+	if (!(udev = _get_udev()))
+		return 0;
 
 	if (!(udev_queue = udev_queue_new(udev))) {
-		udev_unref(udev);
 		goto_bad;
 	}
 
@@ -1996,12 +2010,11 @@ static int _check_udev_is_running(void)
 				     "Not using udev synchronisation code.");
 
 	udev_queue_unref(udev_queue);
-	udev_unref(udev);
 
 	return r;
 
 bad:
-	log_error("Could not get udev state. Assuming udev is not running.");
+	log_error("Could not get udev queue state. Assuming udev is not running.");
 	return 0;
 }
 
-- 
1.7.10.4

-- 
You can't evaluate a man by logic alone.
		-- McCoy, "I, Mudd", stardate 4513.3

[PATCH 3/3] udev: Use udev monitor

[Bastian Blank]

---
 libdm/libdm-common.c |   93 +++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 92 insertions(+), 1 deletion(-)

diff --git a/libdm/libdm-common.c b/libdm/libdm-common.c
index d028029..fdabf80 100644
--- a/libdm/libdm-common.c
+++ b/libdm/libdm-common.c
@@ -26,6 +26,7 @@
 #include <dirent.h>
 
 #ifdef UDEV_SYNC_SUPPORT
+#  include <poll.h>
 #  include <libudev.h>
 #endif
 
@@ -64,7 +65,9 @@ static int _udev_disabled = 0;
 
 #ifdef UDEV_SYNC_SUPPORT
 static struct udev *_udev;
+static struct udev_monitor *_udev_monitor;
 static int _udev_running = -1;
+static unsigned int _udev_monitor_inflight;
 static int _sync_with_udev = 1;
 static int _udev_checking = 1;
 #endif
@@ -1992,6 +1995,55 @@ bad:
 	return 0;
 }
 
+static struct udev_monitor *_get_udev_monitor(void)
+{
+	struct udev *udev = _get_udev();
+
+	if (!udev)
+		return 0;
+
+	if (!_udev_monitor) {
+		log_debug("Opening udev monitor");
+
+		_udev_monitor = udev_monitor_new_from_netlink(udev, "udev");
+		if (!_udev_monitor)
+			goto_bad;
+
+		udev_monitor_filter_add_match_subsystem_devtype(_udev_monitor, "block", "disk");
+		udev_monitor_enable_receiving(_udev_monitor);
+	}
+
+	return _udev_monitor;
+
+bad:
+	log_error("Could not get udev monitor. Assuming udev is not running.");
+	return 0;
+}
+
+static void _close_udev_monitor(void)
+{
+	log_debug("Closing udev monitor");
+	udev_monitor_unref(_udev_monitor);
+	_udev_monitor = NULL;
+}
+
+static void _push_udev_monitor_inflight(void)
+{
+	_udev_monitor_inflight++;
+	if (_udev_monitor_inflight == 0)
+		abort();
+	_get_udev_monitor();
+}
+
+static void _pop_udev_monitor_inflight(void)
+{
+	if (_udev_monitor_inflight == 0)
+		abort();
+	_udev_monitor_inflight--;
+	if (_udev_monitor_inflight == 0)
+		_close_udev_monitor();
+}
+
 static int _check_udev_is_running(void)
 {
 	struct udev *udev;
@@ -2166,6 +2218,7 @@ int dm_task_set_cookie(struct dm_task *dmt, uint32_t *cookie, uint16_t flags)
 
 	dmt->event_nr |= ~DM_UDEV_FLAGS_MASK & *cookie;
 	dmt->cookie_set = 1;
+	_push_udev_monitor_inflight();
 
 	log_debug_activation("Udev cookie 0x%" PRIx32 " assigned to "
 			     "%s task(%d) with flags%s%s%s%s%s%s%s (0x%" PRIx16 ")", *cookie, _task_type_disp(dmt->type), dmt->type, 
@@ -2195,10 +2248,48 @@ int dm_udev_complete(uint32_t cookie)
 
 static int _udev_wait(uint32_t cookie)
 {
+	struct udev_monitor *monitor;
+	struct pollfd p = { -1, POLLIN, 0 };
+	int r = 0;
+
 	if (!cookie || !dm_udev_get_sync_support())
 		return 1;
 
-	return_0;
+	cookie &= ~DM_UDEV_FLAGS_MASK;
+	log_debug("Udev cookie 0x%" PRIx32 " expected", cookie);
+
+	monitor = _get_udev_monitor();
+
+	p.fd = udev_monitor_get_fd(monitor);
+
+	while (!r)
+	{
+		// TODO: timeout
+		poll(&p, 1, -1);
+
+		struct udev_device *device = udev_monitor_receive_device(monitor);
+		struct udev_list_entry *l;
+
+		udev_list_entry_foreach(l, udev_device_get_properties_list_entry(device))
+		{
+			if (strcmp("DM_COOKIE", udev_list_entry_get_name(l)) == 0)
+			{
+				const char *value = udev_list_entry_get_value(l);
+				uint32_t cookie_current = strtol(value, NULL, 10) & ~DM_UDEV_FLAGS_MASK;
+				log_debug("Udev cookie 0x%" PRIx32 " found", cookie_current);
+
+				if (cookie == cookie_current)
+					r = 1;
+				break;
+			}
+		}
+
+		udev_device_unref(device);
+	}
+
+	_pop_udev_monitor_inflight();
+
+	return r;
 }
 
 int dm_udev_wait(uint32_t cookie)
-- 
I have never understood the female capacity to avoid a direct answer to
any question.
		-- Spock, "This Side of Paradise", stardate 3417.3

[PATCH 3/3] udev: Use udev monitor

[Peter Rajnoha]

Hi Bastian,

On 08.06.2013 11:56, Bastian Blank wrote:
> ---
>  libdm/libdm-common.c |   93 +++++++++++++++++++++++++++++++++++++++++++++++++-
>  1 file changed, 92 insertions(+), 1 deletion(-)

> +static unsigned int _udev_monitor_inflight;

The counter is a global variable - what if we have several cookies
being processed? This needs to be wrapped and referenced by cookie id then.

>  static int _udev_wait(uint32_t cookie)
>  {
> +	struct udev_monitor *monitor;
> +	struct pollfd p = { -1, POLLIN, 0 };
> +	int r = 0;
> +
>  	if (!cookie || !dm_udev_get_sync_support())
>  		return 1;
>  
> -	return_0;
> +	cookie &= ~DM_UDEV_FLAGS_MASK;
> +	log_debug("Udev cookie 0x%" PRIx32 " expected", cookie);
> +
> +	monitor = _get_udev_monitor();
> +
> +	p.fd = udev_monitor_get_fd(monitor);
> +
> +	while (!r)
> +	{
> +		// TODO: timeout

Yes, that would be nice in my opinion as well (once I wanted to add
timeouts for semaphores even, but...). If providing a timeout, we need
to take great care about numerous devices being processed under one
cookie - so it needs to be a function of that number of devices (e.g.
activating 1000 and more LVs under one cookie could really take a long
time with udev - so we need to take care to not provide premature timeouts).

Thing is, that if udev works correctly (with kernel, of course), we
don't need timeouts. But udev kills its forked processes on timeout
(which is normally 3 min I think). I've reported this to udev people and
I asked them for a new rule that could provide a "cleanup" on timeout
directly in udev rules, but they don't seem to change anything here. The
problem here is that if this killing happens in udev on timeout, the
udev db is incomplete and we can't detect this state easily.

So the timeout on LVM side is just a workaround we could possibly
provide until udev gets fixed in this area. But a complete solution here
is on udev side imho.

> +		poll(&p, 1, -1);
> +
> +		struct udev_device *device = udev_monitor_receive_device(monitor);
> +		struct udev_list_entry *l;
> +
> +		udev_list_entry_foreach(l, udev_device_get_properties_list_entry(device))
> +		{
> +			if (strcmp("DM_COOKIE", udev_list_entry_get_name(l)) == 0)
> +			{
> +				const char *value = udev_list_entry_get_value(l);
> +				uint32_t cookie_current = strtol(value, NULL, 10) & ~DM_UDEV_FLAGS_MASK;
> +				log_debug("Udev cookie 0x%" PRIx32 " found", cookie_current);
> +
> +				if (cookie == cookie_current)
> +					r = 1;

This breaks with "success"...

> +				break;
> +			}
> +		}
> +
> +		udev_device_unref(device);
> +	}
> +
> +	_pop_udev_monitor_inflight();
> +

...pops just one value for one event...

> +	return r;

...but we could still have several devices under one cookie - e.g.
processing several LVs at once (which is common in LVM code, for example
activating the whole VG with several LVs). So this part needs fixing.

Peter