Index: server/fen-kernel.c
===================================================================
--- server/fen-kernel.c	(revision 0)
+++ server/fen-kernel.c	(revision 0)
@@ -0,0 +1,364 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+/* vim:set expandtab ts=4 shiftwidth=4: */
+
+#include <pthread.h>
+#include <rctl.h>
+#include <strings.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/time.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <glib.h>
+#include "fen-kernel.h"
+#include "gam_error.h"
+
+#define PE_ALLOC	128
+#define PNP_FD(pp)	(((pnode_t *)(pp))->port)
+#define MON_EVENTS	(/* FILE_ACCESS | */ FILE_MODIFIED | FILE_ATTRIB | FILE_NOFOLLOW)
+
+static ulong max_port_evnets = 512;
+static ulong max_port_events = 256;
+static GList *pn_vq;	/* the queue of ports which don't have the max objs */
+static GList *pn_fq;	/* the queue of ports which have the max objs */
+static pthread_mutex_t pn_queues_lock = PTHREAD_MUTEX_INITIALIZER;
+
+#define MUTEX_LOCK(lock) \
+do { \
+	if (pthread_mutex_lock(&lock) != 0) { \
+		abort (); \
+	} \
+} while (0)
+#define MUTEX_UNLOCK(lock) \
+do { \
+	if (pthread_mutex_unlock(&lock) != 0) { \
+		abort (); \
+	} \
+} while (0)
+
+static void pnode_list_walker_cb (gpointer data, gpointer udata);
+static void *port_fetch_event_cb (void *arg);
+static pnode_t *pnode_new ();
+static void pnode_delete (pnode_t *pn);
+
+struct pnode
+{
+	ulong ref;	/* how many fds are associated to this port */
+	pthread_mutex_t m_ref;	/* protect ref */
+	pthread_t tid;
+	int port;
+};
+
+enum {
+	CTL_CLOSE_PORT = 0,
+	CTL_JOIN_THREAD,
+	CTL_FREE_PNODE
+};
+
+void
+printevent(int event, char *pname)
+{
+        GAM_DEBUG(DEBUG_INFO, "FENKERNEL - %s :",pname);
+        if (event & FILE_ACCESS) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_ACCESS");
+        }
+        if (event & FILE_MODIFIED) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_MODIFIED");
+        }
+        if (event & FILE_ATTRIB) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_ATTRIB");
+        }
+        if (event & FILE_DELETE) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_DELETE");
+        }
+        if (event & FILE_RENAME_TO) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_RENAME_TO");
+        }
+        if (event & FILE_RENAME_FROM) {
+                GAM_DEBUG(DEBUG_INFO, " FILE_RENAME_FROM");
+        }
+        if (event & UNMOUNTED) {
+                GAM_DEBUG(DEBUG_INFO, " UNMOUNTED");
+        }
+        if (event & MOUNTEDOVER) {
+                GAM_DEBUG(DEBUG_INFO, " MOUNTEDOVER");
+        }
+        GAM_DEBUG(DEBUG_INFO, "\n");
+}
+
+static void *
+port_fetch_event_cb (void *arg)
+{
+	pnode_t *pn = (pnode_t *)arg;
+	uint_t nget;
+	port_event_t pe[PE_ALLOC];
+	int ret;
+
+	GAM_DEBUG(DEBUG_INFO, "FENKERNEL : [THREAD] started [pn] 0x%p [tid] %d\n", pn, pn->tid);
+
+	while (1) {
+#if 0
+		/* first get the number of available events */
+		if (port_getn (PNP_FD(pn), pe, 0, &nget, NULL) == 0) {
+			if (nget > PE_ALLOC)
+				nget = PE_ALLOC;
+		} else {
+			break;
+		}
+#endif
+		nget = 1;
+		if (port_getn (PNP_FD(pn), pe, PE_ALLOC, &nget, NULL) == 0) {
+			int i;
+			for (i = 0; i < nget; i++) {
+				fnode_t *f;
+				f = (fnode_t *)pe[i].portev_user;
+				/* handle event */
+				printevent(pe[i].portev_events, FN_PATH(f));
+                MUTEX_LOCK (f->ref_lock);
+                g_assert (f->ref > 0);
+                pnode_delete (f->pn);
+                f->pn = NULL;
+                f->ref --;
+                MUTEX_UNLOCK (f->ref_lock);
+                fnode_emit_events (f, pe[i].portev_events);
+			}
+		} else {
+			break;
+		}
+	}
+	GAM_DEBUG(DEBUG_INFO, "FENKERNEL : [THREAD] stopped [pn] 0x%p [tid] %d\n", pn, pn->tid);
+	return (void *)0;
+}
+
+/*
+ * ref - 1 if remove a watching file succeeded.
+ */
+static void
+pnode_delete (pnode_t *pn)
+{
+    MUTEX_LOCK (pn->m_ref);
+    MUTEX_LOCK (pn_queues_lock);
+	if (pn->ref == max_port_evnets) {
+        GAM_DEBUG(DEBUG_INFO, "FENKERNEL : move to visible queue - [pn] 0x%p [ref] %d\n", pn, pn->ref);
+		pn_fq = g_list_remove (pn_fq, pn);
+		pn_vq = g_list_prepend (pn_vq, pn);
+	}
+	if ((-- pn->ref) == 0) {
+		// corrently we can't stop a thread except we close the port fd
+		// pthread_join (pn->tid, NULL);
+		// pn->tid = 0;
+	}
+	GAM_DEBUG(DEBUG_INFO, "FENKERNEL : pnode_delete: [pn] 0x%p [ref] %d\n", pn, pn->ref);
+    MUTEX_UNLOCK (pn_queues_lock);
+    MUTEX_UNLOCK (pn->m_ref);
+}
+
+/*
+ * malloc pnode_t and port_create, start thread at pnode_ref.
+ * if pnode_new succeeded, the pnode_t will never
+ * be freed. So pnode_t can be freed only in pnode_new.
+ * Note pnode_monitor_remove_all can also free pnode_t, but currently no one
+ * invork it.
+ */
+static pnode_t *
+pnode_new ()
+{
+	pnode_t *pn = NULL;
+
+    MUTEX_LOCK (pn_queues_lock);
+	if (pn_vq) {
+		pn = (pnode_t*)pn_vq->data;
+        g_assert (pn->ref < max_port_evnets);
+	} else {
+		pn = malloc (sizeof (pnode_t));
+		if (pn != NULL) {
+			bzero ((void *)pn, sizeof (pnode_t));
+			if (pthread_mutex_init (&pn->m_ref, NULL) != 0) {
+				perror ("pthread_mutex_init");
+				free (pn);
+				pn = NULL;
+            } else if ((PNP_FD(pn) = port_create ()) == -1) {
+                perror ("port_create");
+                free (pn);
+                pn = NULL;
+            } else if (pthread_create (&pn->tid, NULL,
+                                       port_fetch_event_cb,
+                                       (void *)pn) != 0) {
+                perror ("pthread_create");
+				free (pn);
+				pn = NULL;
+            } else {
+                g_assert (g_list_find (pn_vq, pn) == NULL);
+                pn_vq = g_list_prepend (pn_vq, pn);
+			}
+		}
+	}
+	if (pn) {
+		GAM_DEBUG(DEBUG_INFO, "FENKERNEL : pnode_new: [pn] 0x%p [ref] %d\n", pn, pn->ref);
+        pn->ref++;
+        if (pn->ref == max_port_evnets) {
+            GAM_DEBUG(DEBUG_INFO, "FENKERNEL : move to full queue - [pn] 0x%p [ref] %d\n", pn, pn->ref);
+            pn_vq = g_list_remove (pn_vq, pn);
+            pn_fq = g_list_prepend (pn_fq, pn);
+            g_assert (g_list_find (pn_vq, pn) == NULL);
+        }
+	}
+    MUTEX_UNLOCK (pn_queues_lock);
+	return pn;
+}
+
+/**
+ * There is no guarantee that one fobj must be associated to one port fd.
+ * And no guarantee that the same file will be added only once.
+ * So it's up level duty to make sure only add a fobj one time.
+ *
+ * Should call fnode_ref before call this function to protect the data.
+ * And call fnode_unref if receive FILE_EXCEPTION events.
+ *
+ * Returns 0 if succeeded, -1 already added.
+ */
+extern int
+port_monitor_add (fnode_t *f, struct stat * st)
+{
+    int ret;
+    struct stat buf;
+
+    MUTEX_LOCK (f->ref_lock);
+    if (f->pn == NULL) {
+        if (st == NULL) {
+            st = &buf;
+            if ((ret = lstat (FN_PATH(f), st)) != 0) {
+                ret = errno;
+                perror ("FENKERNEL - lstat");
+                MUTEX_UNLOCK (f->ref_lock);
+                return ret;
+            }
+        }
+        fnode_set_stat (f, st);
+
+        if ((f->pn = pnode_new ()) != NULL) {
+
+            if ((ret = port_associate (PNP_FD(f->pn),
+                                       PORT_SOURCE_FILE,
+                                       (uintptr_t)&f->fobj,
+                                       MON_EVENTS,
+                                       (void *)f)) == 0) {
+                f->ref ++;
+                GAM_DEBUG(DEBUG_INFO, "FENKERNEL : associated file %s\n",
+                          FN_PATH(f));
+            } else {
+                perror ("FENKERNEL - port_associate");
+                pnode_delete (f->pn);
+                f->pn = NULL;
+                ret = errno;
+            }
+        } else {
+            ret = ENOMEM;
+        }
+    } else {
+        ret = -1;
+    }
+    MUTEX_UNLOCK (f->ref_lock);
+    return ret;
+}
+
+/**
+ * Should call fnode_unref if call this function successfully
+ *
+ * Return 0 if succeeded, -1 already removed.
+ */
+extern int
+port_monitor_remove (fnode_t *f)
+{
+	int ret;
+
+    MUTEX_LOCK (f->ref_lock);
+    if (f->pn) {
+        if ((ret = port_dissociate (PNP_FD(f->pn), PORT_SOURCE_FILE,
+                                    (uintptr_t)&f->fobj)) == 0) {
+            GAM_DEBUG(DEBUG_INFO, "FENKERNEL : dissociated file %s\n",
+                      FN_PATH(f));
+            pnode_delete (f->pn);
+            f->pn = NULL;
+            f->ref --;
+        } else {
+            ret = errno;
+            perror ("FENKERNEL - port_dissociate");
+        }
+    } else {
+        ret = -1;
+    }
+    MUTEX_UNLOCK (f->ref_lock);
+	return ret;
+}
+
+static void
+port_list_walker_cb (gpointer data, gpointer udata)
+{
+	pnode_t *pn = (pnode_t *)data;
+	switch (GPOINTER_TO_INT (udata)) {
+	case CTL_CLOSE_PORT:
+		close (PNP_FD(pn));
+		break;
+	case CTL_JOIN_THREAD:
+		pthread_join (pn->tid, NULL);
+		break;
+	case CTL_FREE_PNODE:
+	{
+		if (pthread_mutex_destroy (&pn->m_ref) != 0) {
+			perror ("pthread_mutex_destroy");
+		}
+		/* free anyway */
+		free (pn);
+		break;
+	}
+	default:
+		break;
+	}
+}
+
+static void
+pnode_monitor_remove_all ()
+{
+	pnode_t *pn;
+
+	/* close ports */
+    MUTEX_LOCK (pn_queues_lock);
+	g_list_foreach (pn_vq, port_list_walker_cb, (gpointer)CTL_CLOSE_PORT);
+	g_list_foreach (pn_fq, port_list_walker_cb, (gpointer)CTL_CLOSE_PORT);
+	/* wait for threads */
+	g_list_foreach (pn_vq, port_list_walker_cb, (gpointer)CTL_JOIN_THREAD);
+	g_list_foreach (pn_fq, port_list_walker_cb, (gpointer)CTL_JOIN_THREAD);
+	/* free pnodes */
+	g_list_foreach (pn_vq, port_list_walker_cb, (gpointer)CTL_FREE_PNODE);
+	g_list_foreach (pn_fq, port_list_walker_cb, (gpointer)CTL_FREE_PNODE);
+    MUTEX_UNLOCK (pn_queues_lock);
+}
+
+/**
+ * Get Solaris resouce values.
+ *
+ */
+
+extern int
+fen_kernel_init ()
+{
+	rctlblk_t *rblk;
+
+	if ((rblk = malloc (rctlblk_size ())) == NULL) {
+		perror ("rblk malloc");
+		exit (1);
+	}
+	if (getrctl ("process.max-port-events", NULL, rblk, RCTL_FIRST) == -1) {
+		perror ("getrctl");
+		exit (1);
+	} else {
+		if (max_port_evnets > rctlblk_get_value(rblk))
+			max_port_evnets = rctlblk_get_value(rblk);
+		GAM_DEBUG(DEBUG_INFO, "FENKERNEL : max event of a port: %u\n", max_port_evnets);
+	}
+	free (rblk);
+	return 0;
+}
