Index: server/gam_fen.c
===================================================================
--- server/gam_fen.c	(revision 0)
+++ server/gam_fen.c	(revision 0)
@@ -0,0 +1,801 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+/* vim:set expandtab ts=4 shiftwidth=4: */
+/*
+ * Design:
+ * A Solaris port has a resource limit of events (port_max_events) which 
+ * limits the number of objects (fds) that can be actively associated objects
+ * whith the port. The default is (65536), but can be changed.
+ *
+ * project.max-port-ids identify the max number of ports
+ * process.max-port-events identify the max objs of a port
+ * process.max-file-descriptor identify the max fds of a process
+ *
+ * For a user server process, process.max-file-descriptor seems a bottleneck.
+ * I will use a port list for monitor fds to avoid process.max-file-descriptor
+ * is greater than process.max-port-events.
+ */
+#include "config.h"
+#include <strings.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/param.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <stdlib.h>
+#include <errno.h>
+#include "gam_error.h"
+#include "gam_fen.h"
+#include "gam_event.h"
+#include "gam_server.h"
+#include "gam_protocol.h"
+#include "fen-kernel.h"
+#include "fen-thread-pool.h"
+#include "fen-data.h"
+#include <glib.h>
+
+//#define FEN_DEBUG
+#define GC_INTERVAL 15000	/* once every 15 second */
+#define POLL_INTERVAL 1500	/* poll once every 1.5 second */
+#define MAX_EVENT_THREADS 25
+#define MAX_EVENT_QUEUE_LEN G_MAXINT
+
+#define MSLEEP(ms)	(usleep ((ms)*1000))
+#define MUTEX_LOCK(lock) \
+do { \
+	if (pthread_mutex_lock(&lock) != 0) { \
+		g_assert_not_reached (); \
+	} \
+} while (0)
+
+#define MUTEX_UNLOCK(lock) \
+do { \
+	if (pthread_mutex_unlock(&lock) != 0) { \
+		g_assert_not_reached (); \
+	} \
+} while (0)
+
+typedef struct event_pair
+{
+	int events;
+	fnode_t *f;
+} event_pair_t;
+
+static void
+ep_free (event_pair_t *ep)
+{
+	g_free (ep);
+}
+
+static GList *delete_pending_list = NULL;
+static thp_t *thread_pool = NULL;
+
+/* run by thread pool */
+static void events_process_task (event_pair_t *ep);
+static void poll_file_task (fnode_t *f);
+static void events_unsafe_emit (fnode_t *f,
+                                GaminEventType event,
+                                const gchar *path);
+static void find_new_file_in_dir (fnode_t *parent);
+
+/* others */
+static GaminEventType fen_to_gamin_event (int events);
+static void create_children_initially (const char *fullname,
+                                       fnode_t *parent);
+static gboolean fen_monitor_add (fnode_t *f);
+static gboolean poll_monitor_add (fnode_t *f);
+
+static int
+get_lstat (const char *path, struct stat *sbuf)
+{
+    int retval;
+    if ((retval = lstat (path, sbuf)) != 0) {
+        perror ("get_lstat");
+        retval = errno;
+    }
+    return retval;
+}
+
+static void
+publish_gam_events (fnode_t *f, int events)
+{
+	event_pair_t *ep;
+
+    fnode_ref (f);
+	ep = g_new (event_pair_t, 1);
+    ep->f = f;
+	ep->events = events;
+
+	if (thread_pool_run_full (thread_pool,
+                              (thp_run_cb) events_process_task,
+                              (gpointer) ep,
+                              (GDestroyNotify) ep_free) != 0) {
+        GAM_DEBUG(DEBUG_INFO, "FEN : LOSING EVENTS ON %s\n", FN_PATH (f));
+        fnode_unref (f);
+        ep_free (ep);
+    }
+}
+
+static void
+publish_fen_events (fnode_t *f, int events)
+{
+    publish_gam_events (f, fen_to_gamin_event (events));
+}
+
+static gboolean
+fn_hash_foreach_free_fnode (gpointer key,
+                            gpointer value,
+                            gpointer user_data)
+{
+    fnode_t *f = (fnode_t *) value;
+    if (fnode_trylock (&f->fn_lock) == 0) {
+        if (fnode_trylock (&f->ref_lock) != 0) {
+            MUTEX_UNLOCK (f->fn_lock);
+            return FALSE;
+        }
+        if (f->ref != 0 ) {
+            MUTEX_UNLOCK (f->ref_lock);
+            MUTEX_UNLOCK (f->fn_lock);
+            return FALSE;
+        }
+        /* ref_lock and fn_lock are locked now */
+        if (f->parent == NULL && f->subs == NULL) {
+            gboolean ret;
+            if (f->pn || f->is_poll) {
+                MUTEX_UNLOCK (f->ref_lock);
+                port_monitor_remove (f->pn);
+                ret = FALSE;
+                MUTEX_UNLOCK (f->fn_lock);
+            } else {
+                /* we can free this f now */
+                delete_pending_list = g_list_prepend (delete_pending_list, f);
+                ret = TRUE;
+                MUTEX_UNLOCK (f->ref_lock);
+                MUTEX_UNLOCK (f->fn_lock);
+            }
+            return ret;
+        }
+        MUTEX_UNLOCK (f->ref_lock);
+        MUTEX_UNLOCK (f->fn_lock);
+    }
+    return FALSE;
+}
+
+/* occupy a thread */
+void
+gc_thread ()
+{
+    while (1) {
+        guint size;
+        GList *keys, *idx;
+
+        if (fn_hash_trylock () == 0) {
+            fn_hash_foreach_remove (fn_hash_foreach_free_fnode,
+                                    NULL);
+            fn_hash_unlock ();
+        }
+        if (delete_pending_list) {
+            g_list_foreach (delete_pending_list,
+                            (GFunc) fnode_delete,
+                            NULL);
+            g_list_free (delete_pending_list);
+            delete_pending_list = NULL;
+        }
+        MSLEEP (GC_INTERVAL);
+    }
+}
+
+static GaminEventType
+fen_to_gamin_event (int events)
+{
+	GaminEventType et = 0;
+
+    if (events & FILE_MODIFIED ||
+	    events & FILE_ATTRIB ||
+	    events & FILE_ACCESS ||
+        events & FILE_RENAME_TO ||
+        events & MOUNTEDOVER ||
+	    events & UNMOUNTED) {
+                et |= GAMIN_EVENT_CHANGED;
+	}
+	if (events & FILE_DELETE ||
+	    events & FILE_RENAME_FROM) {
+                et |= GAMIN_EVENT_DELETED;
+	}
+	return et;
+}
+
+static void
+events_unsafe_emit (fnode_t *f, GaminEventType events, const gchar *path)
+{
+	gboolean file_is_dir;
+	if (path == NULL) {
+		path = FN_PATH(f);
+		file_is_dir = fnode_is_dir (f);
+	} else {
+		struct stat fsb;
+		memset (&fsb, 0, sizeof (struct stat));
+		lstat (path, &fsb);
+		file_is_dir = S_ISDIR (fsb.st_mode) ? TRUE : FALSE;
+	}
+
+	if (events & GAMIN_EVENT_CREATED) {
+		gam_server_emit_event(path, file_is_dir,
+				      GAMIN_EVENT_CREATED, f->subs, 1);
+	}
+	if (events & GAMIN_EVENT_CHANGED) {
+		GList *idx;
+		for (idx = f->subs; idx; idx = idx->next) {
+			GamSubscription *sub = (GamSubscription *)idx->data;
+            if (!gam_subscription_is_dir (sub)) {
+                gam_server_emit_one_event (path, file_is_dir, GAMIN_EVENT_CHANGED, sub, 1);
+			}
+		}
+        if (f->parent) {
+            MUTEX_LOCK (f->parent->fn_lock);
+            gam_server_emit_event(path, file_is_dir,
+                                  GAMIN_EVENT_CHANGED, f->parent->subs, 1);
+            MUTEX_UNLOCK (f->parent->fn_lock);
+        }
+	}
+	if (events & GAMIN_EVENT_DELETED) {
+		gam_server_emit_event(path, file_is_dir,
+				      GAMIN_EVENT_DELETED, f->subs, 1);
+        if (f->parent) {
+            MUTEX_LOCK (f->parent->fn_lock);
+            gam_server_emit_event(path, file_is_dir,
+                                  GAMIN_EVENT_DELETED, f->parent->subs, 1);
+            MUTEX_UNLOCK (f->parent->fn_lock);
+        }
+	}
+	if (events & GAMIN_EVENT_MOVED) {
+		gam_server_emit_event(path, file_is_dir,
+				      GAMIN_EVENT_MOVED, f->subs, 1);
+        if (f->parent) {
+            MUTEX_LOCK (f->parent->fn_lock);
+            gam_server_emit_event(path, file_is_dir,
+                                  GAMIN_EVENT_MOVED, f->parent->subs, 1);
+            MUTEX_UNLOCK (f->parent->fn_lock);
+        }
+	}
+}
+
+/**
+ * When got events from PORT, process them here.
+ * fnode of event_pair is refed before it's called, so should unref before
+ * it returns.
+ */
+
+static void
+events_process_task (event_pair_t *ep)
+{
+    int ret;
+	fnode_t *f = ep->f;
+	GaminEventType events = ep->events;
+
+    g_assert (f);
+    MUTEX_LOCK (f->ref_lock);
+    g_assert (f->ref > 0);
+    MUTEX_UNLOCK (f->ref_lock);
+
+	GAM_DEBUG(DEBUG_INFO, "FEN : events_process_task 0x%p %s\n", f, FN_PATH(f));
+
+    MUTEX_LOCK (f->fn_lock);
+    events_unsafe_emit (f, events, NULL);
+    if (f->parent == NULL && f->subs == NULL) {
+        fnode_unref (f);
+        MUTEX_UNLOCK (f->fn_lock);
+        return;
+    }
+    MUTEX_UNLOCK (f->fn_lock);
+
+    if (events & GAMIN_EVENT_CHANGED || events & GAMIN_EVENT_CREATED) {
+        switch (port_monitor_add (f, NULL)) {
+        case 0:
+        case -1:
+            /* scan for new created files if somebody mondir */
+            if (fnode_is_dir (f) && fnode_mdcount (f) > 0) {
+                find_new_file_in_dir (f);
+            }
+            break;
+        case ENOENT:
+            publish_gam_events (f, GAMIN_EVENT_DELETED);
+            break;
+        default:
+            if (!poll_monitor_add(f)) {
+                g_assert_not_reached ();
+            }
+        }
+    } else if (events & GAMIN_EVENT_DELETED || events == -1) {
+        /* if the parent dir is mondir,
+           unsub the parent's subs from mine. */
+        MUTEX_LOCK (f->fn_lock);
+        f->parent = NULL;
+        if (f->subs != NULL) {
+            /* turn to polling */
+            if (!poll_monitor_add(f)) {
+                g_assert_not_reached ();
+            }
+        }
+        MUTEX_UNLOCK (f->fn_lock);
+    }
+    fnode_unref (f);
+	GAM_DEBUG(DEBUG_INFO, "FEN : events_process_task 0x%p\n", f);
+}
+
+/**
+ * When a path can be monitored by PORT, we should thread to run stat(2) on
+ * and detect if the file node is created. After that we finish this thread
+ * and pass the node to PORT.
+ */
+
+static void
+poll_file_task (fnode_t *f)
+{
+    int ret;
+    struct stat buf;
+
+    //GAM_DEBUG(DEBUG_INFO, "FEN : poll_file_task %s\n", FN_PATH(f));
+
+    MUTEX_LOCK (f->fn_lock);
+    MUTEX_LOCK (f->ref_lock);
+    if (f->parent == NULL && f->subs == NULL) {
+        f->is_poll = FALSE;
+        f->ref --;
+        MUTEX_UNLOCK (f->ref_lock);
+        MUTEX_UNLOCK (f->fn_lock);
+        return;
+    }
+    MUTEX_UNLOCK (f->ref_lock);
+    MUTEX_UNLOCK (f->fn_lock);
+
+    if ((ret = get_lstat (FN_PATH(f), &buf)) == 0) {
+        switch (port_monitor_add (f, &buf)) {
+        case 0:
+        case -1:
+            MUTEX_LOCK (f->ref_lock);
+            f->is_poll = FALSE;
+            MUTEX_UNLOCK (f->ref_lock);
+            publish_gam_events (f, GAMIN_EVENT_CREATED);
+            /* scan for new created files if somebody mondir */
+            if (fnode_is_dir (f) && fnode_mdcount (f) > 0) {
+                find_new_file_in_dir (f);
+            }
+            break;
+        case ENOENT:
+            publish_gam_events (f, GAMIN_EVENT_DELETED);
+        default:
+            goto l_poll;
+        }
+    } else {
+    l_poll:
+        MSLEEP (POLL_INTERVAL);
+        MUTEX_LOCK (f->ref_lock);
+        f->is_poll = FALSE;
+        MUTEX_UNLOCK (f->ref_lock);
+        if (!poll_monitor_add(f)) {
+            g_assert_not_reached ();
+        }
+    }
+    fnode_unref (f);
+    return;
+}
+
+static gboolean
+poll_monitor_add (fnode_t *f)
+{
+    MUTEX_LOCK (f->ref_lock);
+    if (!f->is_poll) {
+        if (thread_pool_run (thread_pool,
+                             (thp_run_cb) poll_file_task, f) == 0) {
+            //GAM_DEBUG(DEBUG_INFO, "FEN : [ polling ] %s\n", FN_PATH(f));
+            f->ref ++;
+            f->is_poll = TRUE;
+        } else {
+            MUTEX_UNLOCK (f->ref_lock);
+            return FALSE;
+        }
+    }
+    MUTEX_UNLOCK (f->ref_lock);
+    return TRUE;
+}
+
+/**
+ * If the fnode_t is a directory and GAMIN_EVENT_CHANGED happenned on it,
+ * Use it to detect what happenned in the first level of the directory.
+ *
+ * In thread
+ */
+
+static void
+find_new_file_in_dir (fnode_t *parent)
+{
+	GDir *dir;
+	GError *err = NULL;
+
+	GAM_DEBUG(DEBUG_INFO, "FEN : find_new_file_in_dir 0x%p %s\n",
+              parent, FN_PATH(parent));
+	dir = g_dir_open (FN_PATH(parent), 0, &err);
+	if (dir) {
+		const char *filename;
+
+		while ((filename = g_dir_read_name (dir)))
+		{
+            fnode_t *tmpf;
+			GList *idx;
+			gchar *fullname = g_build_filename (FN_PATH(parent),
+							    filename,
+							    NULL);
+			/* If can't found the fullname from the global hash, 
+               emit the event on all mondir subs.
+               Else if foreach mondir subs, if can't find in fn subs of
+               fullname, emit event on that sub.
+            */
+            //GAM_DEBUG(DEBUG_INFO, "FEN : found %s\n", fullname);
+            if ((tmpf = fn_hash_safe_node_new (fullname,
+                                               publish_fen_events)) == NULL) {
+                GAM_DEBUG(DEBUG_INFO, "FEN : create fn failed: %s\n", fullname);
+                goto l_continue;
+            }
+
+            switch (port_monitor_add (tmpf, NULL)) {
+            case 0:
+            case -1:
+                break;
+            case ENOENT:
+            default:
+                goto l_skip;
+            }
+
+            /* set parent iff the parent has subscriptions or is mondired */
+            MUTEX_LOCK (tmpf->fn_lock);
+            MUTEX_LOCK (parent->fn_lock);
+            if ((parent->parent || parent->subs) && tmpf->parent == NULL) {
+                tmpf->parent = parent;
+                gam_server_emit_event (fullname, fnode_is_dir(tmpf), GAMIN_EVENT_CREATED, parent->subs, 1);
+            }
+            MUTEX_UNLOCK (parent->fn_lock);
+            MUTEX_UNLOCK (tmpf->fn_lock);
+
+        l_skip:
+            fnode_unref (tmpf);
+        l_continue:
+            g_free (fullname);
+        }
+		g_dir_close (dir);
+	} else {
+		GAM_DEBUG (DEBUG_INFO, "unable to open directory %s: %s\n", FN_PATH(parent), err->message);
+		g_error_free (err);
+	}
+}
+
+static void
+foreach_hash_remove_parent (gpointer key,
+                            gpointer value,
+                            gpointer user_data)
+{
+    /* hash is locked outside, note 
+       do not lock hash here or the invorked functions */
+    fnode_t *parent = (fnode_t *) user_data;
+    const gchar *path = FN_PATH(parent);
+    gchar *dirname = g_path_get_dirname (key);
+
+    /* skip root "/" */
+    if (parent != value &&
+        strcmp (path, dirname) == 0) {
+        fnode_t *f = (fnode_t *) value;
+
+        MUTEX_LOCK (f->fn_lock);
+        f->parent = NULL;
+        if (f->subs == NULL) {
+            port_monitor_remove (f);
+        }
+        MUTEX_UNLOCK (f->fn_lock);
+    }
+    g_free (dirname);
+}
+
+/*
+ * We do it because we know the parent has some mondir subscriptions.
+ * Parent is refed.
+ * Fullname is known to be existing, but skip it if it's removed when
+ * add it to port.
+ */
+
+static void
+create_children_initially (const char *fullname,
+                           fnode_t *parent)
+
+{
+	fnode_t *f;
+
+	if ((f = fn_hash_safe_node_new (fullname, publish_fen_events)) == NULL) {
+		GAM_DEBUG(DEBUG_INFO, "FEN : create_children_initially - failed to create fn\n");
+		return;
+	}
+
+	MUTEX_LOCK (f->fn_lock);
+	MUTEX_LOCK (parent->fn_lock);
+    switch (port_monitor_add (f, NULL)) {
+    case 0:
+    case -1:
+    case ENOENT:
+        break;
+    default:
+        /* skip any other unsopport error */
+        goto l_skip;
+    }
+    /* append the mondir subs */
+    f->parent = parent;
+
+ l_skip:
+    fnode_unref (f);
+	MUTEX_UNLOCK (parent->fn_lock);
+    MUTEX_UNLOCK (f->fn_lock);
+}
+
+/**
+ * First time scan the subscription file or directory. Only invoked in
+ * gam_fen_add_subscription. Here f is refed.
+ * Returns TRUE, if pathname exists.
+ */
+
+static gboolean
+gam_fen_send_initial_events (const char *pathname,
+			     fnode_t *f,
+			     GamSubscription *sub)
+{
+    int retval;
+	gboolean is_dir;
+	GaminEventType gevent;
+	int req = 1;
+    struct stat buf;
+
+	GAM_DEBUG (DEBUG_INFO, "gam_fen_send_initial_events on %s\n", pathname);
+    is_dir = gam_subscription_is_dir (sub);
+
+    switch (get_lstat (FN_PATH(f), &buf)) {
+	case 0:
+        fnode_set_stat (f, &buf);
+		gevent = GAMIN_EVENT_EXISTS;
+        retval = TRUE;
+        break;
+	case ENOENT:
+		gevent = GAMIN_EVENT_DELETED;
+        retval = FALSE;
+		is_dir = FALSE;
+        break;
+    default:
+        perror ("gam_fen_send_initial_events");
+	}
+
+	if (req != 0 || gevent == GAMIN_EVENT_DELETED)
+		gam_server_emit_one_event (pathname, is_dir, gevent, sub, 1);
+
+	if (req == 0)
+		return retval;
+
+	/* NOTE: If the sub is mondiring, we need create the new fnode
+	   for each child files/dirs and mark the f->is_md_child. */
+	if (is_dir && fnode_is_dir (f)) {
+        GDir *dir;
+        GError *err = NULL;
+
+        dir = g_dir_open (FN_PATH(f), 0, &err);
+        if (dir) {
+            const char *filename;
+            while ((filename = g_dir_read_name (dir)))
+            {
+                gboolean file_is_dir = FALSE;
+                gchar *fullname = g_build_filename (FN_PATH(f),
+                                                    filename,
+                                                    NULL);
+                if (lstat (fullname, &buf) == 0) {
+                    file_is_dir = S_ISDIR (buf.st_mode) ? TRUE : FALSE;
+                }
+                MUTEX_LOCK (f->fn_lock);
+                gam_server_emit_one_event (fullname, file_is_dir, GAMIN_EVENT_EXISTS, sub, 1);
+                MUTEX_UNLOCK (f->fn_lock);
+
+                /* if I'm a mondir */
+                if (fnode_mdcount (f) > 0) {
+                    create_children_initially (fullname, f);
+                }
+
+                g_free (fullname);
+            }
+            g_dir_close (dir);
+        } else {
+            GAM_DEBUG (DEBUG_INFO, "unable to open directory %s: %s\n", FN_PATH(f), err->message);
+            g_error_free (err);
+        }
+	}
+
+	gam_server_emit_one_event (pathname, is_dir, GAMIN_EVENT_ENDEXISTS, sub, 1);
+    return retval;
+}
+
+/**
+ * Initializes the FEN system.  This must be called before
+ * any other functions in this module.
+ *
+ * @returns TRUE if initialization succeeded, FALSE otherwise
+ */
+
+gboolean
+gam_fen_init (void)
+{
+	GError *err = NULL;
+
+	if (fen_kernel_init () != 0) {
+		GAM_DEBUG(DEBUG_INFO, "FEN : initializing failed - fen_kernel_init.\n");
+		return FALSE;
+	}
+
+    if (!g_thread_supported ()) g_thread_init (NULL);
+
+	if (!fn_hash_init ()) {
+		GAM_DEBUG(DEBUG_INFO, "FEN : initializing failed - fn_hash_init.\n");
+		return FALSE;
+	}
+
+	thread_pool = thread_pool_new (MAX_EVENT_THREADS, MAX_EVENT_QUEUE_LEN);
+	if (thread_pool == NULL) {
+		GAM_DEBUG(DEBUG_INFO, "FEN : initializing failed - thread_pool_new.\n");
+		return FALSE;
+	}
+
+#ifdef GAMIN_DEBUG
+#ifdef FEN_DEBUG
+    if (thread_pool_run (thread_pool, 
+                         (thp_run_cb) fn_hash_safe_debug, 
+                         NULL) != 0) {
+        g_assert_not_reached ();
+    }
+#endif
+#endif
+    if (thread_pool_run (thread_pool, 
+                         (thp_run_cb) gc_thread, 
+                         NULL) != 0) {
+        g_assert_not_reached ();
+    }
+	
+	gam_server_install_kernel_hooks (GAMIN_K_FEN, 
+					 gam_fen_add_subscription,
+					 gam_fen_remove_subscription,
+					 gam_fen_remove_all_for,
+					 NULL, NULL);
+
+	GAM_DEBUG(DEBUG_INFO, "FEN : initialized.\n");
+	return TRUE;
+}
+
+/**
+ * Adds a subscription to be monitored.
+ *
+ * @param sub a #GamSubscription to be polled
+ * @returns TRUE if adding the subscription succeeded, FALSE otherwise
+ */
+
+gboolean
+gam_fen_add_subscription (GamSubscription *sub)
+{
+    int ret;
+	fnode_t *f;	
+    gboolean is_existing;
+    const gchar *filename;
+	
+	GAM_DEBUG(DEBUG_INFO, "FEN : gam_fen_add_subscription\n");
+    filename = gam_subscription_get_path (sub);
+	gam_listener_add_subscription (gam_subscription_get_listener (sub), sub);
+
+	if ((f = fn_hash_safe_node_new (filename,
+                                    publish_fen_events)) == NULL) {
+		GAM_DEBUG(DEBUG_INFO, "%s: new sub failed\n", filename);
+		return FALSE;
+	}
+
+    if (gam_subscription_is_dir (sub)) {
+        fnode_mdref (f);
+    }
+
+	is_existing = gam_fen_send_initial_events (filename, f, sub);
+
+	MUTEX_LOCK (f->fn_lock);
+    switch (port_monitor_add (f, NULL)) {
+    case 0:
+    case -1:
+        ret = TRUE;
+        break;
+    case ENOENT:
+    default:
+        if (!poll_monitor_add (f)) {
+            g_assert_not_reached ();
+        }
+        ret = TRUE;
+        break;
+    }
+    g_assert (g_list_find (f->subs, sub) == NULL);
+	f->subs = g_list_prepend (f->subs, sub);
+
+    fnode_unref (f);
+	MUTEX_UNLOCK (f->fn_lock);
+
+	return ret;
+}
+
+/**
+ * Removes a subscription which was being monitored.
+ *
+ * @param sub a #GamSubscription to remove
+ * @returns TRUE if removing the subscription succeeded, FALSE otherwise
+ */
+
+gboolean
+gam_fen_remove_subscription (GamSubscription *sub)
+{
+	fnode_t *f;
+	GList *targ;
+	
+	GAM_DEBUG(DEBUG_INFO, "FEN : gam_fen_remove_subscription\n");
+
+	if ((f = fn_hash_safe_lookup (gam_subscription_get_path(sub))) == NULL) {
+        GAM_DEBUG(DEBUG_INFO, "FEN : can't find %s\n",
+                  FN_PATH(gam_subscription_get_path(sub)));
+		return TRUE;
+	}
+
+	MUTEX_LOCK (f->fn_lock);
+
+	/* If no one monitor this f, remove the sub from f->subs */
+	/* the removal of the fn will be pending in thread_pool */
+    f->subs = g_list_remove (f->subs, sub);
+
+    if (f->parent == NULL && f->subs == NULL) {
+        /* remove the childen's parent if I'm the parent */
+        fn_hash_safe_foreach (foreach_hash_remove_parent, f);
+        port_monitor_remove (f);
+    }
+	MUTEX_UNLOCK (f->fn_lock);
+
+    if (gam_subscription_is_dir (sub)) {
+        fnode_mdunref (f);
+    }
+
+	/* free subscription */
+	gam_subscription_free(sub);
+	return TRUE;
+}
+
+/**
+ * Stop monitoring all subscriptions for a given listener.
+ *
+ * @param listener a #GamListener
+ * @returns TRUE if removing the subscriptions succeeded, FALSE otherwise
+ */
+
+gboolean
+gam_fen_remove_all_for (GamListener *listener)
+{
+	GList *subs;
+	GList *idx;
+	gboolean success = TRUE;
+	
+	subs = gam_listener_get_subscriptions (listener);
+	
+	if (subs == NULL)
+		return FALSE;
+
+	for (idx = subs; idx != NULL; idx = idx->next) {
+		GamSubscription *sub = (GamSubscription *)idx->data;
+		g_assert (sub);
+		if (!gam_fen_remove_subscription (sub))
+			success = FALSE;
+	}
+	
+	if (subs) {
+		g_list_free (subs);
+		return TRUE;
+	} else {
+		return FALSE;
+	}
+}
