[libdispatch-changes] [149] trunk/src

source_changes at macosforge.org source_changes at macosforge.org
Thu Nov 12 16:03:21 PST 2009


Revision: 149
          http://trac.macosforge.org/projects/libdispatch/changeset/149
Author:   robert at fledge.watson.org
Date:     2009-11-12 16:03:21 -0800 (Thu, 12 Nov 2009)
Log Message:
-----------
Move most kqueue-related code from source.c to source_kevent.c, and hook
that up to the build.

Submitted by:	Paolo Bonzini <bonzini at gnu.org>

Modified Paths:
--------------
    trunk/src/Makefile.am
    trunk/src/kevent_internal.h
    trunk/src/source.c
    trunk/src/source_internal.h
    trunk/src/source_kevent.c

Modified: trunk/src/Makefile.am
===================================================================
--- trunk/src/Makefile.am	2009-11-12 23:27:00 UTC (rev 148)
+++ trunk/src/Makefile.am	2009-11-13 00:03:21 UTC (rev 149)
@@ -14,6 +14,7 @@
 	queue_kevent.c	\
 	semaphore.c	\
 	source.c	\
+	source_kevent.c	\
 	time.c
 
 libshims_la_SOURCES=	\

Modified: trunk/src/kevent_internal.h
===================================================================
--- trunk/src/kevent_internal.h	2009-11-12 23:27:00 UTC (rev 148)
+++ trunk/src/kevent_internal.h	2009-11-13 00:03:21 UTC (rev 149)
@@ -29,6 +29,19 @@
 
 #include <internal.h>
 
+struct dispatch_kevent_s {
+	TAILQ_ENTRY(dispatch_kevent_s) dk_list;
+	TAILQ_HEAD(, dispatch_source_s) dk_sources;
+	struct kevent dk_kevent;
+};
+
+#define DISPATCH_EVFILT_TIMER	(-EVFILT_SYSCOUNT - 1)
+#define DISPATCH_EVFILT_CUSTOM_ADD	(-EVFILT_SYSCOUNT - 2)
+#define DISPATCH_EVFILT_CUSTOM_OR	(-EVFILT_SYSCOUNT - 3)
+#define DISPATCH_EVFILT_SYSCOUNT	(EVFILT_SYSCOUNT + 3)
+
+extern const struct dispatch_source_vtable_s _dispatch_source_kevent_vtable;
+
 #if DISPATCH_DEBUG
 void dispatch_debug_kevents(struct kevent* kev, size_t count, const char* str);
 #else
@@ -38,4 +51,5 @@
 void _dispatch_source_drain_kevent(struct kevent *);
 void _dispatch_update_kq(const struct kevent *);
 
+
 #endif /* __DISPATCH_KEVENT_INTERNAL__ */

Modified: trunk/src/source.c
===================================================================
--- trunk/src/source.c	2009-11-12 23:27:00 UTC (rev 148)
+++ trunk/src/source.c	2009-11-13 00:03:21 UTC (rev 149)
@@ -27,65 +27,6 @@
 
 #include "kevent_internal.h"
 
-#ifdef DISPATCH_NO_LEGACY
-enum {
-	DISPATCH_TIMER_WALL_CLOCK	= 0x4,
-};
-enum {
-	DISPATCH_TIMER_INTERVAL	= 0x0,
-	DISPATCH_TIMER_ONESHOT	= 0x1,
-	DISPATCH_TIMER_ABSOLUTE	= 0x3,
-};
-enum {
-	DISPATCH_MACHPORT_DEAD = 0x1,
-	DISPATCH_MACHPORT_RECV = 0x2,
-	DISPATCH_MACHPORT_DELETED = 0x4,
-};
-#endif
-
-#define DISPATCH_EVFILT_TIMER	(-EVFILT_SYSCOUNT - 1)
-#define DISPATCH_EVFILT_CUSTOM_ADD	(-EVFILT_SYSCOUNT - 2)
-#define DISPATCH_EVFILT_CUSTOM_OR	(-EVFILT_SYSCOUNT - 3)
-#define DISPATCH_EVFILT_SYSCOUNT	(EVFILT_SYSCOUNT + 3)
-
-#define DISPATCH_TIMER_INDEX_WALL 0
-#define DISPATCH_TIMER_INDEX_MACH 1
-static struct dispatch_kevent_s _dispatch_kevent_timer[] = {
-	{
-		.dk_kevent = {
-			.ident = DISPATCH_TIMER_INDEX_WALL,
-			.filter = DISPATCH_EVFILT_TIMER,
-			.udata = &_dispatch_kevent_timer[0],
-		},
-		.dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_timer[0].dk_sources),
-	},
-	{
-		.dk_kevent = {
-			.ident = DISPATCH_TIMER_INDEX_MACH,
-			.filter = DISPATCH_EVFILT_TIMER,
-			.udata = &_dispatch_kevent_timer[1],
-		},
-		.dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_timer[1].dk_sources),
-	},
-};
-#define DISPATCH_TIMER_COUNT (sizeof _dispatch_kevent_timer / sizeof _dispatch_kevent_timer[0])
-
-static struct dispatch_kevent_s _dispatch_kevent_data_or = {
-	.dk_kevent = {
-		.filter = DISPATCH_EVFILT_CUSTOM_OR,
-		.flags = EV_CLEAR,
-		.udata = &_dispatch_kevent_data_or,
-	},
-	.dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_or.dk_sources),
-};
-static struct dispatch_kevent_s _dispatch_kevent_data_add = {
-	.dk_kevent = {
-		.filter = DISPATCH_EVFILT_CUSTOM_ADD,
-		.udata = &_dispatch_kevent_data_add,
-	},
-	.dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_add.dk_sources),
-};
-
 #ifndef DISPATCH_NO_LEGACY
 struct dispatch_source_attr_vtable_s {
 	DISPATCH_VTABLE_HEADER(dispatch_source_attr_s);
@@ -102,100 +43,8 @@
 #define _dispatch_source_call_block ((void *)-1)
 static void _dispatch_source_latch_and_call(dispatch_source_t ds);
 static void _dispatch_source_cancel_callout(dispatch_source_t ds);
-static bool _dispatch_source_probe(dispatch_source_t ds);
-static void _dispatch_source_dispose(dispatch_source_t ds);
-static void _dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent *ke);
-static size_t _dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz);
 static size_t dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz);
-static dispatch_queue_t _dispatch_source_invoke(dispatch_source_t ds);
-static void _dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags, uint32_t del_flags);
 
-static void _dispatch_kevent_merge(dispatch_source_t ds);
-static void _dispatch_kevent_release(dispatch_source_t ds);
-static void _dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags);
-#if HAVE_MACH
-static void _dispatch_kevent_machport_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags);
-static void _dispatch_kevent_machport_enable(dispatch_kevent_t dk);
-static void _dispatch_kevent_machport_disable(dispatch_kevent_t dk);
-
-static void _dispatch_drain_mach_messages(struct kevent *ke);
-#endif
-static void _dispatch_timer_list_update(dispatch_source_t ds);
-
-#if HAVE_MACH
-static void
-_dispatch_mach_notify_source_init(void *context __attribute__((unused)));
-#endif
-
-static const char *
-_evfiltstr(short filt)
-{
-	switch (filt) {
-#define _evfilt2(f) case (f): return #f
-	_evfilt2(EVFILT_READ);
-	_evfilt2(EVFILT_WRITE);
-	_evfilt2(EVFILT_AIO);
-	_evfilt2(EVFILT_VNODE);
-	_evfilt2(EVFILT_PROC);
-	_evfilt2(EVFILT_SIGNAL);
-	_evfilt2(EVFILT_TIMER);
-#if HAVE_MACH
-	_evfilt2(EVFILT_MACHPORT);
-#endif
-	_evfilt2(EVFILT_FS);
-	_evfilt2(EVFILT_USER);
-#if HAVE_DECL_EVFILT_SESSION
-	_evfilt2(EVFILT_SESSION);
-#endif
-
-	_evfilt2(DISPATCH_EVFILT_TIMER);
-	_evfilt2(DISPATCH_EVFILT_CUSTOM_ADD);
-	_evfilt2(DISPATCH_EVFILT_CUSTOM_OR);
-	default:
-		return "EVFILT_missing";
-	}
-}
-
-#define DSL_HASH_SIZE 256u	// must be a power of two
-#define DSL_HASH(x)	((x) & (DSL_HASH_SIZE - 1))
-
-static TAILQ_HEAD(, dispatch_kevent_s) _dispatch_sources[DSL_HASH_SIZE];
-
-static inline uintptr_t
-_dispatch_kevent_hash(uintptr_t ident, short filter)
-{
-	uintptr_t value;
-#if HAVE_MACH
-	value = (filter == EVFILT_MACHPORT ? MACH_PORT_INDEX(ident) : ident);
-#else
-	value = ident;
-#endif
-	return DSL_HASH(value);
-}
-
-static dispatch_kevent_t
-_dispatch_kevent_find(uintptr_t ident, short filter)
-{
-	uintptr_t hash = _dispatch_kevent_hash(ident, filter);
-	dispatch_kevent_t dki;
-
-	TAILQ_FOREACH(dki, &_dispatch_sources[hash], dk_list) {
-		if (dki->dk_kevent.ident == ident && dki->dk_kevent.filter == filter) {
-			break;
-		}
-	}
-	return dki;
-}
-
-static void
-_dispatch_kevent_insert(dispatch_kevent_t dk)
-{
-	uintptr_t hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
-					       dk->dk_kevent.filter);
-
-	TAILQ_INSERT_TAIL(&_dispatch_sources[hash], dk, dk_list);
-}
-
 void
 dispatch_source_cancel(dispatch_source_t ds)
 {
@@ -258,120 +107,6 @@
 	return ds->ds_data;
 }
 
-#if DISPATCH_DEBUG
-void
-dispatch_debug_kevents(struct kevent* kev, size_t count, const char* str)
-{
-	size_t i;
-	for (i = 0; i < count; ++i) {
-		_dispatch_log("kevent[%lu] = { ident = %p, filter = %s, flags = 0x%x, fflags = 0x%x, data = %p, udata = %p }: %s",
-			i, (void*)kev[i].ident, _evfiltstr(kev[i].filter), kev[i].flags, kev[i].fflags, (void*)kev[i].data, (void*)kev[i].udata, str);
-	}
-}
-#endif
-
-static size_t
-_dispatch_source_kevent_debug(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
-	size_t offset = _dispatch_source_debug(ds, buf, bufsiz);
-	offset += snprintf(&buf[offset], bufsiz - offset, "filter = %s }",
-			ds->ds_dkev ? _evfiltstr(ds->ds_dkev->dk_kevent.filter) : "????");
-	return offset;
-}
-
-static void
-_dispatch_source_init_tail_queue_array(void *context __attribute__((unused)))
-{
-	unsigned int i;
-	for (i = 0; i < DSL_HASH_SIZE; i++) {
-		TAILQ_INIT(&_dispatch_sources[i]);
-	}
-
-	TAILQ_INSERT_TAIL(&_dispatch_sources[DSL_HASH(DISPATCH_TIMER_INDEX_WALL)], &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_WALL], dk_list);
-	TAILQ_INSERT_TAIL(&_dispatch_sources[DSL_HASH(DISPATCH_TIMER_INDEX_MACH)], &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_MACH], dk_list);
-	TAILQ_INSERT_TAIL(&_dispatch_sources[0], &_dispatch_kevent_data_or, dk_list);
-	TAILQ_INSERT_TAIL(&_dispatch_sources[0], &_dispatch_kevent_data_add, dk_list);
-}
-
-// Find existing kevents, and merge any new flags if necessary
-void
-_dispatch_kevent_merge(dispatch_source_t ds)
-{
-	static dispatch_once_t pred;
-	dispatch_kevent_t dk;
-	typeof(dk->dk_kevent.fflags) new_flags;
-	bool do_resume = false;
-
-	if (ds->ds_is_installed) {
-		return;
-	}
-	ds->ds_is_installed = true;
-
-	dispatch_once_f(&pred, NULL, _dispatch_source_init_tail_queue_array);
-
-	dk = _dispatch_kevent_find(ds->ds_dkev->dk_kevent.ident, ds->ds_dkev->dk_kevent.filter);
-	
-	if (dk) {
-		// If an existing dispatch kevent is found, check to see if new flags
-		// need to be added to the existing kevent
-		new_flags = ~dk->dk_kevent.fflags & ds->ds_dkev->dk_kevent.fflags;
-		dk->dk_kevent.fflags |= ds->ds_dkev->dk_kevent.fflags;
-		free(ds->ds_dkev);
-		ds->ds_dkev = dk;
-		do_resume = new_flags;
-	} else {
-		dk = ds->ds_dkev;
-		_dispatch_kevent_insert(dk);
-		new_flags = dk->dk_kevent.fflags;
-		do_resume = true;
-	}
-
-	TAILQ_INSERT_TAIL(&dk->dk_sources, ds, ds_list);
-
-	// Re-register the kevent with the kernel if new flags were added
-	// by the dispatch kevent
-	if (do_resume) {
-		dk->dk_kevent.flags |= EV_ADD;
-		_dispatch_source_kevent_resume(ds, new_flags, 0);
-		ds->ds_is_armed = true;
-	}
-}
-
-
-void
-_dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags)
-{
-	switch (dk->dk_kevent.filter) {
-	case DISPATCH_EVFILT_TIMER:
-	case DISPATCH_EVFILT_CUSTOM_ADD:
-	case DISPATCH_EVFILT_CUSTOM_OR:
-		// these types not registered with kevent
-		return;
-#if HAVE_MACH
-	case EVFILT_MACHPORT:
-		_dispatch_kevent_machport_resume(dk, new_flags, del_flags);
-		break;
-#endif
-	case EVFILT_PROC:
-		if (dk->dk_kevent.flags & EV_ONESHOT) {
-			return;
-		}
-		// fall through
-	default:
-		_dispatch_update_kq(&dk->dk_kevent);
-		if (dk->dk_kevent.flags & EV_DISPATCH) {
-			dk->dk_kevent.flags &= ~EV_ADD;
-		}
-		break;
-	}
-}
-
-void
-_dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags, uint32_t del_flags)
-{
-	_dispatch_kevent_resume(ds->ds_dkev, new_flags, del_flags);
-}
-
 dispatch_queue_t
 _dispatch_source_invoke(dispatch_source_t ds)
 {
@@ -462,284 +197,7 @@
 	_dispatch_queue_dispose((dispatch_queue_t)ds);
 }
 
-#ifndef DISPATCH_NO_LEGACY
-static void
-_dispatch_kevent_debugger2(void *context, dispatch_source_t unused __attribute__((unused)))
-{
-	struct sockaddr sa;
-	socklen_t sa_len = sizeof(sa);
-	int c, fd = (int)(long)context;
-	unsigned int i;
-	dispatch_kevent_t dk;
-	dispatch_source_t ds;
-	FILE *debug_stream;
-
-	c = accept(fd, &sa, &sa_len);
-	if (c == -1) {
-		if (errno != EAGAIN) {
-			(void)dispatch_assume_zero(errno);
-		}
-		return;
-	}
-#if 0
-	int r = fcntl(c, F_SETFL, 0);	// disable non-blocking IO
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-	}
-#endif
-	debug_stream = fdopen(c, "a");
-	if (!dispatch_assume(debug_stream)) {
-		close(c);
-		return;
-	}
-
-	fprintf(debug_stream, "HTTP/1.0 200 OK\r\n");
-	fprintf(debug_stream, "Content-type: text/html\r\n");
-	fprintf(debug_stream, "Pragma: nocache\r\n");
-	fprintf(debug_stream, "\r\n");
-	fprintf(debug_stream, "<html>\n<head><title>PID %u</title></head>\n<body>\n<ul>\n", getpid());
-
-	//fprintf(debug_stream, "<tr><td>DK</td><td>DK</td><td>DK</td><td>DK</td><td>DK</td><td>DK</td><td>DK</td></tr>\n");
-
-	for (i = 0; i < DSL_HASH_SIZE; i++) {
-		if (TAILQ_EMPTY(&_dispatch_sources[i])) {
-			continue;
-		}
-		TAILQ_FOREACH(dk, &_dispatch_sources[i], dk_list) {
-			fprintf(debug_stream, "\t<br><li>DK %p ident %lu filter %s flags 0x%hx fflags 0x%x data 0x%lx udata %p\n",
-					dk, (unsigned long)dk->dk_kevent.ident, _evfiltstr(dk->dk_kevent.filter), dk->dk_kevent.flags,
-					dk->dk_kevent.fflags, (unsigned long)dk->dk_kevent.data, dk->dk_kevent.udata);
-			fprintf(debug_stream, "\t\t<ul>\n");
-			TAILQ_FOREACH(ds, &dk->dk_sources, ds_list) {
-				fprintf(debug_stream, "\t\t\t<li>DS %p refcnt 0x%x suspend 0x%x data 0x%lx mask 0x%lx flags 0x%x</li>\n",
-						ds, ds->do_ref_cnt, ds->do_suspend_cnt, ds->ds_pending_data, ds->ds_pending_data_mask,
-						ds->ds_atomic_flags);
-				if (ds->do_suspend_cnt == DISPATCH_OBJECT_SUSPEND_LOCK) {
-					dispatch_queue_t dq = ds->do_targetq;
-					fprintf(debug_stream, "\t\t<br>DQ: %p refcnt 0x%x suspend 0x%x label: %s\n", dq, dq->do_ref_cnt, dq->do_suspend_cnt, dq->dq_label);
-				}
-			}
-			fprintf(debug_stream, "\t\t</ul>\n");
-			fprintf(debug_stream, "\t</li>\n");
-		}
-	}
-	fprintf(debug_stream, "</ul>\n</body>\n</html>\n");
-	fflush(debug_stream);
-	fclose(debug_stream);
-}
-
-static void
-_dispatch_kevent_debugger(void *context __attribute__((unused)))
-{
-	union {
-		struct sockaddr_in sa_in;
-		struct sockaddr sa;
-	} sa_u = {
-		.sa_in = {
-			.sin_family = AF_INET,
-			.sin_addr = { htonl(INADDR_LOOPBACK), },
-		},
-	};
-	dispatch_source_t ds;
-	const char *valstr;
-	int val, r, fd, sock_opt = 1;
-	socklen_t slen = sizeof(sa_u);
-
-	if (issetugid()) {
-		return;
-	}
-	valstr = getenv("LIBDISPATCH_DEBUGGER");
-	if (!valstr) {
-		return;
-	}
-	val = atoi(valstr);
-	if (val == 2) {
-		sa_u.sa_in.sin_addr.s_addr = 0;
-	}
-	fd = socket(PF_INET, SOCK_STREAM, 0);
-	if (fd == -1) {
-		(void)dispatch_assume_zero(errno);
-		return;
-	}
-	r = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *)&sock_opt, (socklen_t) sizeof sock_opt);
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-		goto out_bad;
-	}
-#if 0
-	r = fcntl(fd, F_SETFL, O_NONBLOCK);
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-		goto out_bad;
-	}
-#endif
-	r = bind(fd, &sa_u.sa, sizeof(sa_u));
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-		goto out_bad;
-	}
-	r = listen(fd, SOMAXCONN);
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-		goto out_bad;
-	}
-	r = getsockname(fd, &sa_u.sa, &slen);
-	if (r == -1) {
-		(void)dispatch_assume_zero(errno);
-		goto out_bad;
-	}
-	ds = dispatch_source_read_create_f(fd, NULL, &_dispatch_mgr_q, (void *)(long)fd, _dispatch_kevent_debugger2);
-	if (dispatch_assume(ds)) {
-		_dispatch_log("LIBDISPATCH: debug port: %hu", ntohs(sa_u.sa_in.sin_port));
-		return;
-	}
-out_bad:
-	close(fd);
-}
-#endif /* DISPATCH_NO_LEGACY */
-
 void
-_dispatch_source_drain_kevent(struct kevent *ke)
-{
-#ifndef DISPATCH_NO_LEGACY
-	static dispatch_once_t pred;
-#endif
-	dispatch_kevent_t dk = ke->udata;
-	dispatch_source_t dsi;
-
-#ifndef DISPATCH_NO_LEGACY
-	dispatch_once_f(&pred, NULL, _dispatch_kevent_debugger);
-#endif
-
-	dispatch_debug_kevents(ke, 1, __func__);
-
-#if HAVE_MACH
-	if (ke->filter == EVFILT_MACHPORT) {
-		return _dispatch_drain_mach_messages(ke);
-	}
-#endif
-	dispatch_assert(dk);
-
-	if (ke->flags & EV_ONESHOT) {
-		dk->dk_kevent.flags |= EV_ONESHOT;
-	}
-
-	TAILQ_FOREACH(dsi, &dk->dk_sources, ds_list) {
-		_dispatch_source_merge_kevent(dsi, ke);
-	}
-}
-
-static void
-_dispatch_kevent_dispose(dispatch_kevent_t dk)
-{
-	uintptr_t hash;
-
-	switch (dk->dk_kevent.filter) {
-	case DISPATCH_EVFILT_TIMER:
-	case DISPATCH_EVFILT_CUSTOM_ADD:
-	case DISPATCH_EVFILT_CUSTOM_OR:
-		// these sources live on statically allocated lists
-		return;
-#if HAVE_MACH
-	case EVFILT_MACHPORT:
-		_dispatch_kevent_machport_resume(dk, 0, dk->dk_kevent.fflags);
-		break;
-#endif
-	case EVFILT_PROC:
-		if (dk->dk_kevent.flags & EV_ONESHOT) {
-			break;	// implicitly deleted
-		}
-		// fall through
-	default:
-		if (~dk->dk_kevent.flags & EV_DELETE) {
-			dk->dk_kevent.flags |= EV_DELETE;
-			_dispatch_update_kq(&dk->dk_kevent);
-		}
-		break;
-	}
-
-	hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
-				     dk->dk_kevent.filter);
-	TAILQ_REMOVE(&_dispatch_sources[hash], dk, dk_list);
-	free(dk);
-}
-
-void
-_dispatch_kevent_release(dispatch_source_t ds)
-{
-	dispatch_kevent_t dk = ds->ds_dkev;
-	dispatch_source_t dsi;
-	uint32_t del_flags, fflags = 0;
-
-	ds->ds_dkev = NULL;
-
-	TAILQ_REMOVE(&dk->dk_sources, ds, ds_list);
-
-	if (TAILQ_EMPTY(&dk->dk_sources)) {
-		_dispatch_kevent_dispose(dk);
-	} else {
-		TAILQ_FOREACH(dsi, &dk->dk_sources, ds_list) {
-			fflags |= (uint32_t)dsi->ds_pending_data_mask;
-		}
-		del_flags = (uint32_t)ds->ds_pending_data_mask & ~fflags;
-		if (del_flags) {
-			dk->dk_kevent.flags |= EV_ADD;
-			dk->dk_kevent.fflags = fflags;
-			_dispatch_kevent_resume(dk, 0, del_flags);
-		}
-	}
-
-	ds->ds_is_armed = false;
-	ds->ds_needs_rearm = false;	// re-arm is pointless and bad now
-	_dispatch_release(ds);	// the retain is done at creation time
-}
-
-void
-_dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent *ke)
-{
-	struct kevent fake;
-
-	if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == 0)) {
-		return;
-	}
-
-	// EVFILT_PROC may fail with ESRCH when the process exists but is a zombie.
-	// We simulate an exit event in this case. <rdar://problem/5067725>
-	if (ke->flags & EV_ERROR) {
-		if (ke->filter == EVFILT_PROC && ke->data == ESRCH) {
-			fake = *ke;
-			fake.flags &= ~EV_ERROR;
-			fake.fflags = NOTE_EXIT;
-			fake.data = 0;
-			ke = &fake;
-		} else {
-			// log the unexpected error
-			(void)dispatch_assume_zero(ke->data);
-			return;
-		}
-	}
-
-	if (ds->ds_is_level) {
-		// ke->data is signed and "negative available data" makes no sense
-		// zero bytes happens when EV_EOF is set
-		// 10A268 does not fail this assert with EVFILT_READ and a 10 GB file
-		dispatch_assert(ke->data >= 0l);
-		ds->ds_pending_data = ~ke->data;
-	} else if (ds->ds_is_adder) {
-		dispatch_atomic_add(&ds->ds_pending_data, ke->data);
-	} else {
-		dispatch_atomic_or(&ds->ds_pending_data, ke->fflags & ds->ds_pending_data_mask);
-	}
-
-	// EV_DISPATCH and EV_ONESHOT sources are no longer armed after delivery
-	if (ds->ds_needs_rearm) {
-		ds->ds_is_armed = false;
-	}
-
-	_dispatch_wakeup(ds);
-}
-
-void
 _dispatch_source_latch_and_call(dispatch_source_t ds)
 {
 	unsigned long prev;
@@ -801,29 +259,6 @@
 	ds->ds_cancel_handler = NULL;
 }
 
-const struct dispatch_source_vtable_s _dispatch_source_kevent_vtable = {
-	.do_type = DISPATCH_SOURCE_KEVENT_TYPE,
-	.do_kind = "kevent-source",
-	.do_invoke = _dispatch_source_invoke,
-	.do_dispose = _dispatch_source_dispose,
-	.do_probe = _dispatch_source_probe,
-	.do_debug = _dispatch_source_kevent_debug,
-};
-
-void
-dispatch_source_merge_data(dispatch_source_t ds, unsigned long val)
-{	
-	struct kevent kev = {
-		.fflags = (typeof(kev.fflags))val,
-		.data = val,
-	};
-
-	dispatch_assert(ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_ADD ||
-					ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_OR);
-
-	_dispatch_source_merge_kevent(ds, &kev);
-}
-
 size_t
 dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
 {
@@ -951,143 +386,6 @@
 #endif /* DISPATCH_NO_LEGACY */
 
 
-struct dispatch_source_type_s {
-	struct kevent ke;
-	uint64_t mask;
-	void (*init)(dispatch_source_t ds, dispatch_source_type_t type,
-	    uintptr_t handle, unsigned long mask, dispatch_queue_t q);
-};
-
-static void
-dispatch_source_type_timer_init(dispatch_source_t ds, dispatch_source_type_t type, uintptr_t handle, unsigned long mask, dispatch_queue_t q)
-{
-	ds->ds_needs_rearm = true;
-	ds->ds_timer.flags = mask;
-}
-
-const struct dispatch_source_type_s _dispatch_source_type_timer = {
-	.ke = {
-		.filter = DISPATCH_EVFILT_TIMER,
-	},
-	.mask = DISPATCH_TIMER_INTERVAL|DISPATCH_TIMER_ONESHOT|DISPATCH_TIMER_ABSOLUTE|DISPATCH_TIMER_WALL_CLOCK,
-	.init = dispatch_source_type_timer_init,
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_read = {
-	.ke = {
-		.filter = EVFILT_READ,
-		.flags = EV_DISPATCH,
-	},
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_write = {
-	.ke = {
-		.filter = EVFILT_WRITE,
-		.flags = EV_DISPATCH,
-	},
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_proc = {
-	.ke = {
-		.filter = EVFILT_PROC,
-		.flags = EV_CLEAR,
-	},
-	.mask = NOTE_EXIT|NOTE_FORK|NOTE_EXEC
-#if HAVE_DECL_NOTE_SIGNAL
-	    |NOTE_SIGNAL
-#endif
-#if HAVE_DECL_NOTE_REAP
-	    |NOTE_REAP
-#endif
-	    ,
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_signal = {
-	.ke = {
-		.filter = EVFILT_SIGNAL,
-	},
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_vnode = {
-	.ke = {
-		.filter = EVFILT_VNODE,
-		.flags = EV_CLEAR,
-	},
-	.mask = NOTE_DELETE|NOTE_WRITE|NOTE_EXTEND|NOTE_ATTRIB|NOTE_LINK|
-	    NOTE_RENAME|NOTE_REVOKE
-#if HAVE_DECL_NOTE_NONE
-	    |NOTE_NONE
-#endif
-	    ,
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_vfs = {
-	.ke = {
-		.filter = EVFILT_FS,
-		.flags = EV_CLEAR,
-	},
-	.mask = VQ_NOTRESP|VQ_NEEDAUTH|VQ_LOWDISK|VQ_MOUNT|VQ_UNMOUNT|VQ_DEAD|
-	    VQ_ASSIST|VQ_NOTRESPLOCK
-#if HAVE_DECL_VQ_UPDATE
-	    |VQ_UPDATE
-#endif
-#if HAVE_DECL_VQ_VERYLOWDISK
-	    |VQ_VERYLOWDISK
-#endif
-	    ,
-};
-
-#if HAVE_MACH
-
-static void
-dispatch_source_type_mach_send_init(dispatch_source_t ds, dispatch_source_type_t type, uintptr_t handle, unsigned long mask, dispatch_queue_t q)
-{
-	static dispatch_once_t pred;
-
-	ds->ds_is_level = false;
-	dispatch_once_f(&pred, NULL, _dispatch_mach_notify_source_init);
-}
-
-const struct dispatch_source_type_s _dispatch_source_type_mach_send = {
-	.ke = {
-		.filter = EVFILT_MACHPORT,
-		.flags = EV_DISPATCH,
-		.fflags = DISPATCH_MACHPORT_DEAD,
-	},
-	.mask = DISPATCH_MACH_SEND_DEAD,
-	.init = dispatch_source_type_mach_send_init,
-};
-
-static void
-dispatch_source_type_mach_recv_init(dispatch_source_t ds, dispatch_source_type_t type, uintptr_t handle, unsigned long mask, dispatch_queue_t q)
-{
-	ds->ds_is_level = false;
-}
-
-const struct dispatch_source_type_s _dispatch_source_type_mach_recv = {
-	.ke = {
-		.filter = EVFILT_MACHPORT,
-		.flags = EV_DISPATCH,
-		.fflags = DISPATCH_MACHPORT_RECV,
-	},
-	.init = dispatch_source_type_mach_recv_init,
-};
-#endif
-
-const struct dispatch_source_type_s _dispatch_source_type_data_add = {
-	.ke = {
-		.filter = DISPATCH_EVFILT_CUSTOM_ADD,
-	},
-};
-
-const struct dispatch_source_type_s _dispatch_source_type_data_or = {
-	.ke = {
-		.filter = DISPATCH_EVFILT_CUSTOM_OR,
-		.flags = EV_CLEAR,
-		.fflags = ~0,
-	},
-};
-
 dispatch_source_t
 dispatch_source_create(dispatch_source_type_t type,
 	uintptr_t handle,
@@ -1368,171 +666,6 @@
 }
 #endif /* DISPATCH_NO_LEGACY */
 
-// Updates the ordered list of timers based on next fire date for changes to ds.
-// Should only be called from the context of _dispatch_mgr_q.
-void
-_dispatch_timer_list_update(dispatch_source_t ds)
-{
-	dispatch_source_t dsi = NULL;
-	int idx;
-	
-	dispatch_assert(_dispatch_queue_get_current() == &_dispatch_mgr_q);
-
-	// do not reschedule timers unregistered with _dispatch_kevent_release()
-	if (!ds->ds_dkev) {
-		return;
-	}
-
-	// Ensure the source is on the global kevent lists before it is removed and
-	// readded below.
-	_dispatch_kevent_merge(ds);
-	
-	TAILQ_REMOVE(&ds->ds_dkev->dk_sources, ds, ds_list);
-
-	// change the list if the clock type has changed
-	if (ds->ds_timer.flags & DISPATCH_TIMER_WALL_CLOCK) {
-		idx = DISPATCH_TIMER_INDEX_WALL;
-	} else {
-		idx = DISPATCH_TIMER_INDEX_MACH;
-	}
-	ds->ds_dkev = &_dispatch_kevent_timer[idx];
-
-	if (ds->ds_timer.target) {
-		TAILQ_FOREACH(dsi, &ds->ds_dkev->dk_sources, ds_list) {
-			if (dsi->ds_timer.target == 0 || ds->ds_timer.target < dsi->ds_timer.target) {
-				break;
-			}
-		}
-	}
-	
-	if (dsi) {
-		TAILQ_INSERT_BEFORE(dsi, ds, ds_list);
-	} else {
-		TAILQ_INSERT_TAIL(&ds->ds_dkev->dk_sources, ds, ds_list);
-	}
-}
-
-static void
-_dispatch_run_timers2(unsigned int timer)
-{
-	dispatch_source_t ds;
-	uint64_t now, missed;
-
-	if (timer == DISPATCH_TIMER_INDEX_MACH) {
-		now = _dispatch_absolute_time();
-	} else {
-		now = _dispatch_get_nanoseconds();
-	}
-
-	while ((ds = TAILQ_FIRST(&_dispatch_kevent_timer[timer].dk_sources))) {
-		// We may find timers on the wrong list due to a pending update from
-		// dispatch_source_set_timer. Force an update of the list in that case.
-		if (timer != ds->ds_ident_hack) {
-			_dispatch_timer_list_update(ds);
-			continue;
-		}
-		if (!ds->ds_timer.target) {
-			// no configured timers on the list
-			break;
-		}
-		if (ds->ds_timer.target > now) {
-			// Done running timers for now.
-			break;
-		}
-
-		if (ds->ds_timer.flags & (DISPATCH_TIMER_ONESHOT|DISPATCH_TIMER_ABSOLUTE)) {
-			dispatch_atomic_inc(&ds->ds_pending_data);
-			ds->ds_timer.target = 0;
-		} else {
-			// Calculate number of missed intervals.
-			missed = (now - ds->ds_timer.target) / ds->ds_timer.interval;
-			dispatch_atomic_add(&ds->ds_pending_data, missed + 1);
-			ds->ds_timer.target += (missed + 1) * ds->ds_timer.interval;
-		}
-
-		_dispatch_timer_list_update(ds);
-		_dispatch_wakeup(ds);
-	}
-}
-
-void
-_dispatch_run_timers(void)
-{
-	unsigned int i;
-	for (i = 0; i < DISPATCH_TIMER_COUNT; i++) {
-		_dispatch_run_timers2(i);
-	}
-}
-
-// approx 1 year (60s * 60m * 24h * 365d)
-#define FOREVER_SEC 3153600l
-#define FOREVER_NSEC 31536000000000000ull
-
-struct timespec *
-_dispatch_get_next_timer_fire(struct timespec *howsoon)
-{
-	// <rdar://problem/6459649>
-	// kevent(2) does not allow large timeouts, so we use a long timeout
-	// instead (approximately 1 year).
-	dispatch_source_t ds = NULL;
-	unsigned int timer;
-	uint64_t now, delta_tmp, delta = UINT64_MAX;
-
-	// We are looking for the first unsuspended timer which has its target
-	// time set. Given timers are kept in order, if we hit an timer that's
-	// unset there's no point in continuing down the list.
-	for (timer = 0; timer < DISPATCH_TIMER_COUNT; timer++) {
-		TAILQ_FOREACH(ds, &_dispatch_kevent_timer[timer].dk_sources, ds_list) {
-			if (!ds->ds_timer.target) {
-				break;
-			}
-			if (DISPATCH_OBJECT_SUSPENDED(ds)) {
-				ds->ds_is_armed = false;
-			} else {
-				break;
-			}
-		}
-
-		if (!ds || !ds->ds_timer.target) {
-			continue;
-		}
-				
-		if (ds->ds_timer.flags & DISPATCH_TIMER_WALL_CLOCK) {
-			now = _dispatch_get_nanoseconds();
-		} else {
-			now = _dispatch_absolute_time();
-		}
-		if (ds->ds_timer.target <= now) {
-			howsoon->tv_sec = 0;
-			howsoon->tv_nsec = 0;
-			return howsoon;
-		}
-
-		// the subtraction cannot go negative because the previous "if"
-		// verified that the target is greater than now.
-		delta_tmp = ds->ds_timer.target - now;
-		if (!(ds->ds_timer.flags & DISPATCH_TIMER_WALL_CLOCK)) {
-			delta_tmp = _dispatch_time_mach2nano(delta_tmp);
-		}
-		if (delta_tmp < delta) {
-			delta = delta_tmp;
-		}
-	}
-	if (slowpath(delta > FOREVER_NSEC)) {
-		return NULL;
-	} else {
-		howsoon->tv_sec = (time_t)(delta / NSEC_PER_SEC);
-		howsoon->tv_nsec = (long)(delta % NSEC_PER_SEC);
-	}
-	return howsoon;
-}
-
-struct dispatch_set_timer_params {
-	dispatch_source_t ds;
-	uintptr_t ident;
-	struct dispatch_timer_source_s values;
-};
-
 // To be called from the context of the _dispatch_mgr_q
 static void
 _dispatch_source_set_timer2(void *context)
@@ -1642,430 +775,3 @@
 }
 #endif /* DISPATCH_NO_LEGACY */
 
-#if HAVE_MACH
-static dispatch_source_t _dispatch_mach_notify_source;
-static mach_port_t _dispatch_port_set;
-static mach_port_t _dispatch_event_port;
-
-#define _DISPATCH_IS_POWER_OF_TWO(v)	(!(v & (v - 1)) && v)
-#define _DISPATCH_HASH(x, y)    (_DISPATCH_IS_POWER_OF_TWO(y) ? (MACH_PORT_INDEX(x) & ((y) - 1)) : (MACH_PORT_INDEX(x) % (y)))
-
-#define _DISPATCH_MACHPORT_HASH_SIZE 32
-#define _DISPATCH_MACHPORT_HASH(x)    _DISPATCH_HASH((x), _DISPATCH_MACHPORT_HASH_SIZE)
-
-static void _dispatch_port_set_init(void *);
-static mach_port_t _dispatch_get_port_set(void);
-
-void
-_dispatch_drain_mach_messages(struct kevent *ke)
-{
-	dispatch_source_t dsi;
-	dispatch_kevent_t dk;
-	struct kevent ke2;
-
-	if (!dispatch_assume(ke->data)) {
-		return;
-	}
-	dk = _dispatch_kevent_find(ke->data, EVFILT_MACHPORT);
-	if (!dispatch_assume(dk)) {
-		return;
-	}
-	_dispatch_kevent_machport_disable(dk);	// emulate EV_DISPATCH
-
-	EV_SET(&ke2, ke->data, EVFILT_MACHPORT, EV_ADD|EV_ENABLE|EV_DISPATCH, DISPATCH_MACHPORT_RECV, 0, dk);
-
-	TAILQ_FOREACH(dsi, &dk->dk_sources, ds_list) {
-		_dispatch_source_merge_kevent(dsi, &ke2);
-	}
-}
-
-void
-_dispatch_port_set_init(void *context __attribute__((unused)))
-{
-	struct kevent kev = {
-		.filter = EVFILT_MACHPORT,
-		.flags = EV_ADD,
-	};
-	kern_return_t kr;
-
-	kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET, &_dispatch_port_set);
-	DISPATCH_VERIFY_MIG(kr);
-	(void)dispatch_assume_zero(kr);
-	kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_RECEIVE, &_dispatch_event_port);
-	DISPATCH_VERIFY_MIG(kr);
-	(void)dispatch_assume_zero(kr);
-	kr = mach_port_move_member(mach_task_self(), _dispatch_event_port, _dispatch_port_set);
-	DISPATCH_VERIFY_MIG(kr);
-	(void)dispatch_assume_zero(kr);
-
-	kev.ident = _dispatch_port_set;
-
-	_dispatch_update_kq(&kev);
-}
-
-mach_port_t
-_dispatch_get_port_set(void)
-{
-	static dispatch_once_t pred;
-
-	dispatch_once_f(&pred, NULL, _dispatch_port_set_init);
-
-	return _dispatch_port_set;
-}
-
-void
-_dispatch_kevent_machport_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags)
-{
-	mach_port_t previous, port = (mach_port_t)dk->dk_kevent.ident;
-	kern_return_t kr;
-
-	if ((new_flags & DISPATCH_MACHPORT_RECV) || (!new_flags && !del_flags && dk->dk_kevent.fflags & DISPATCH_MACHPORT_RECV)) {
-		_dispatch_kevent_machport_enable(dk);
-	}
-	if (new_flags & DISPATCH_MACHPORT_DEAD) {
-		kr = mach_port_request_notification(mach_task_self(), port, MACH_NOTIFY_DEAD_NAME, 1,
-				_dispatch_event_port, MACH_MSG_TYPE_MAKE_SEND_ONCE, &previous);
-		DISPATCH_VERIFY_MIG(kr);
-
-	
-		switch(kr) {
-			case KERN_INVALID_NAME:
-			case KERN_INVALID_RIGHT:
-				// Supress errors 
-				break;
-			default:
-			// Else, we dont expect any errors from mach. Log any errors if we do
-			if (dispatch_assume_zero(kr)) {
-				// log the error
-			} else if (dispatch_assume_zero(previous)) {
-				// Another subsystem has beat libdispatch to requesting the Mach
-				// dead-name notification on this port. We should technically cache the
-				// previous port and message it when the kernel messages our port. Or
-				// we can just say screw those subsystems and drop the previous port.
-				// They should adopt libdispatch :-P
-				kr = mach_port_deallocate(mach_task_self(), previous);
-				DISPATCH_VERIFY_MIG(kr);
-				(void)dispatch_assume_zero(kr);
-			}
-		}
-	}
-
-	if (del_flags & DISPATCH_MACHPORT_RECV) {
-		_dispatch_kevent_machport_disable(dk);
-	}
-	if (del_flags & DISPATCH_MACHPORT_DEAD) {
-		kr = mach_port_request_notification(mach_task_self(), (mach_port_t)dk->dk_kevent.ident,
-				MACH_NOTIFY_DEAD_NAME, 1, MACH_PORT_NULL, MACH_MSG_TYPE_MAKE_SEND_ONCE, &previous);
-		DISPATCH_VERIFY_MIG(kr);
-
-		switch (kr) {
-			case KERN_INVALID_NAME:
-			case KERN_INVALID_RIGHT:
-			case KERN_INVALID_ARGUMENT:
-				break;
-			default:
-			if (dispatch_assume_zero(kr)) {
-				// log the error
-			} else if (previous) {
-				// the kernel has not consumed the right yet
-				(void)dispatch_assume_zero(_dispatch_send_consume_send_once_right(previous));
-			}
-		}
-	}
-}
-
-void
-_dispatch_kevent_machport_enable(dispatch_kevent_t dk)
-{
-	mach_port_t mp = (mach_port_t)dk->dk_kevent.ident;
-	kern_return_t kr;
-
-	kr = mach_port_move_member(mach_task_self(), mp, _dispatch_get_port_set());
-	DISPATCH_VERIFY_MIG(kr);
-	switch (kr) {
-	case KERN_INVALID_NAME:
-#if DISPATCH_DEBUG
-		_dispatch_log("Corruption: Mach receive right 0x%x destroyed prematurely", mp);
-#endif
-		break;
-	default:
-	 	(void)dispatch_assume_zero(kr);
-	}
-}
-
-void
-_dispatch_kevent_machport_disable(dispatch_kevent_t dk)
-{
-	mach_port_t mp = (mach_port_t)dk->dk_kevent.ident;
-	kern_return_t kr;
-
-	kr = mach_port_move_member(mach_task_self(), mp, 0);
-	DISPATCH_VERIFY_MIG(kr);
-	switch (kr) {
-	case KERN_INVALID_RIGHT:
-	case KERN_INVALID_NAME:
-#if DISPATCH_DEBUG
-		_dispatch_log("Corruption: Mach receive right 0x%x destroyed prematurely", mp);
-#endif
-		break;
-	case 0:
-		break;
-	default:
-		(void)dispatch_assume_zero(kr);
-		break;
-	}
-}
-
-#define _DISPATCH_MIN_MSG_SZ (8ul * 1024ul - MAX_TRAILER_SIZE)
-#ifndef DISPATCH_NO_LEGACY
-dispatch_source_t
-dispatch_source_mig_create(mach_port_t mport, size_t max_msg_size, dispatch_source_attr_t attr,
-	dispatch_queue_t dq, dispatch_mig_callback_t mig_callback)
-{
-	if (max_msg_size < _DISPATCH_MIN_MSG_SZ) {
-		max_msg_size = _DISPATCH_MIN_MSG_SZ;
-	}
-	return dispatch_source_machport_create(mport, DISPATCH_MACHPORT_RECV, attr, dq,
-	^(dispatch_source_t ds) {
-		if (!dispatch_source_get_error(ds, NULL)) {
-			if (dq->dq_width != 1) {
-				dispatch_retain(ds);	// this is a shim -- use the external retain
-				dispatch_async(dq, ^{
-					dispatch_mig_server(ds, max_msg_size, mig_callback);
-					dispatch_release(ds);	// this is a shim -- use the external release
-				});
-			} else {
-				dispatch_mig_server(ds, max_msg_size, mig_callback);
-			}
-		}
-	});	
-}
-#endif /* DISPATCH_NO_LEGACY */
-
-static void
-_dispatch_mach_notify_source2(void *context)
-{
-	dispatch_source_t ds = context;
-	const size_t maxsz = MAX(
-		sizeof(union __RequestUnion___dispatch_send_libdispatch_internal_protocol_subsystem),
-		sizeof(union __ReplyUnion___dispatch_libdispatch_internal_protocol_subsystem));
-
-	dispatch_mig_server(ds, maxsz, libdispatch_internal_protocol_server);
-}
-
-static void
-_dispatch_mach_notify_source_init(void *context __attribute__((unused)))
-{
-	_dispatch_get_port_set();
-
-	_dispatch_mach_notify_source = dispatch_source_create(DISPATCH_SOURCE_TYPE_MACH_RECV, _dispatch_event_port, 0, &_dispatch_mgr_q);
-	dispatch_assert(_dispatch_mach_notify_source);
-	dispatch_set_context(_dispatch_mach_notify_source, _dispatch_mach_notify_source);
-	dispatch_source_set_event_handler_f(_dispatch_mach_notify_source, _dispatch_mach_notify_source2);
-	dispatch_resume(_dispatch_mach_notify_source);
-}
-
-kern_return_t
-_dispatch_mach_notify_port_deleted(mach_port_t notify __attribute__((unused)), mach_port_name_t name)
-{
-	dispatch_source_t dsi;
-	dispatch_kevent_t dk;
-	struct kevent kev;
-
-#if DISPATCH_DEBUG
-	_dispatch_log("Corruption: Mach send/send-once/dead-name right 0x%x deleted prematurely", name);
-#endif
-
-	dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
-	if (!dk) {
-		goto out;
-	}
-
-	EV_SET(&kev, name, EVFILT_MACHPORT, EV_ADD|EV_ENABLE|EV_DISPATCH|EV_EOF, DISPATCH_MACHPORT_DELETED, 0, dk);
-
-	TAILQ_FOREACH(dsi, &dk->dk_sources, ds_list) {
-		_dispatch_source_merge_kevent(dsi, &kev);
-		// this can never happen again
-		// this must happen after the merge
-		// this may be racy in the future, but we don't provide a 'setter' API for the mask yet
-		dsi->ds_pending_data_mask &= ~DISPATCH_MACHPORT_DELETED;
-	}
-
-	// no more sources have this flag
-	dk->dk_kevent.fflags &= ~DISPATCH_MACHPORT_DELETED;
-
-out:
-	return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_port_destroyed(mach_port_t notify __attribute__((unused)), mach_port_t name)
-{
-	kern_return_t kr;
-	// this function should never be called
-	(void)dispatch_assume_zero(name);
-	kr = mach_port_mod_refs(mach_task_self(), name, MACH_PORT_RIGHT_RECEIVE, -1);
-	DISPATCH_VERIFY_MIG(kr);
-	(void)dispatch_assume_zero(kr);
-	return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_no_senders(mach_port_t notify, mach_port_mscount_t mscnt __attribute__((unused)))
-{
-	// this function should never be called
-	(void)dispatch_assume_zero(notify);
-	return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_send_once(mach_port_t notify __attribute__((unused)))
-{
-	// we only register for dead-name notifications
-	// some code deallocated our send-once right without consuming it
-#if DISPATCH_DEBUG
-	_dispatch_log("Corruption: An app/library deleted a libdispatch dead-name notification");
-#endif
-	return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_dead_name(mach_port_t notify __attribute__((unused)), mach_port_name_t name)
-{
-	dispatch_source_t dsi;
-	dispatch_kevent_t dk;
-	struct kevent kev;
-	kern_return_t kr;
-
-	dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
-	if (!dk) {
-		goto out;
-	}
-
-	EV_SET(&kev, name, EVFILT_MACHPORT, EV_ADD|EV_ENABLE|EV_DISPATCH|EV_EOF, DISPATCH_MACHPORT_DEAD, 0, dk);
-
-	TAILQ_FOREACH(dsi, &dk->dk_sources, ds_list) {
-		_dispatch_source_merge_kevent(dsi, &kev);
-		// this can never happen again
-		// this must happen after the merge
-		// this may be racy in the future, but we don't provide a 'setter' API for the mask yet
-		dsi->ds_pending_data_mask &= ~DISPATCH_MACHPORT_DEAD;
-	}
-
-	// no more sources have this flag
-	dk->dk_kevent.fflags &= ~DISPATCH_MACHPORT_DEAD;
-
-out:
-	// the act of receiving a dead name notification allocates a dead-name right that must be deallocated
-	kr = mach_port_deallocate(mach_task_self(), name);
-	DISPATCH_VERIFY_MIG(kr);
-	//(void)dispatch_assume_zero(kr);
-
-	return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_wakeup_main_thread(mach_port_t mp __attribute__((unused)))
-{
-	// dummy function just to pop out the main thread out of mach_msg()
-	return 0;
-}
-
-kern_return_t
-_dispatch_consume_send_once_right(mach_port_t mp __attribute__((unused)))
-{
-	// dummy function to consume a send-once right
-	return 0;
-}
-
-mach_msg_return_t
-dispatch_mig_server(dispatch_source_t ds, size_t maxmsgsz, dispatch_mig_callback_t callback)
-{
-	mach_msg_options_t options = MACH_RCV_MSG | MACH_RCV_TIMEOUT
-		| MACH_RCV_TRAILER_ELEMENTS(MACH_RCV_TRAILER_CTX)
-		| MACH_RCV_TRAILER_TYPE(MACH_MSG_TRAILER_FORMAT_0);
-	mach_msg_options_t tmp_options = options;
-	mig_reply_error_t *bufTemp, *bufRequest, *bufReply;
-	mach_msg_return_t kr = 0;
-	unsigned int cnt = 1000;	// do not stall out serial queues
-	int demux_success;
-
-	maxmsgsz += MAX_TRAILER_SIZE;
-
-	// XXX FIXME -- allocate these elsewhere
-	bufRequest = alloca(maxmsgsz);
-	bufReply = alloca(maxmsgsz);
-	bufReply->Head.msgh_size = 0;	// make CLANG happy
-
-	// XXX FIXME -- change this to not starve out the target queue
-	for (;;) {
-		if (DISPATCH_OBJECT_SUSPENDED(ds) || (--cnt == 0)) {
-			options &= ~MACH_RCV_MSG;
-			tmp_options &= ~MACH_RCV_MSG;
-
-			if (!(tmp_options & MACH_SEND_MSG)) {
-				break;
-			}
-		}
-
-		kr = mach_msg(&bufReply->Head, tmp_options, bufReply->Head.msgh_size,
-				(mach_msg_size_t)maxmsgsz, (mach_port_t)ds->ds_ident_hack, 0, 0);
-
-		tmp_options = options;
-
-		if (slowpath(kr)) {
-			switch (kr) {
-			case MACH_SEND_INVALID_DEST:
-			case MACH_SEND_TIMED_OUT:
-				if (bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX) {
-					mach_msg_destroy(&bufReply->Head);
-				}
-				break;
-			case MACH_RCV_TIMED_OUT:
-			case MACH_RCV_INVALID_NAME:
-				break;
-			default:
-				(void)dispatch_assume_zero(kr);
-				break;
-			}
-			break;
-		}
-
-		if (!(tmp_options & MACH_RCV_MSG)) {
-			break;
-		}
-
-		bufTemp = bufRequest;
-		bufRequest = bufReply;
-		bufReply = bufTemp;
-
-		demux_success = callback(&bufRequest->Head, &bufReply->Head);
-
-		if (!demux_success) {
-			// destroy the request - but not the reply port
-			bufRequest->Head.msgh_remote_port = 0;
-			mach_msg_destroy(&bufRequest->Head);
-		} else if (!(bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX)) {
-			// if MACH_MSGH_BITS_COMPLEX is _not_ set, then bufReply->RetCode is present
-			if (slowpath(bufReply->RetCode)) {
-				if (bufReply->RetCode == MIG_NO_REPLY) {
-					continue;
-				}
-
-				// destroy the request - but not the reply port
-				bufRequest->Head.msgh_remote_port = 0;
-				mach_msg_destroy(&bufRequest->Head);
-			}
-		}
-
-		if (bufReply->Head.msgh_remote_port) {
-			tmp_options |= MACH_SEND_MSG;
-			if (MACH_MSGH_BITS_REMOTE(bufReply->Head.msgh_bits) != MACH_MSG_TYPE_MOVE_SEND_ONCE) {
-				tmp_options |= MACH_SEND_TIMEOUT;
-			}
-		}
-	}
-
-	return kr;
-}
-#endif /* HAVE_MACH */

Modified: trunk/src/source_internal.h
===================================================================
--- trunk/src/source_internal.h	2009-11-12 23:27:00 UTC (rev 148)
+++ trunk/src/source_internal.h	2009-11-13 00:03:21 UTC (rev 149)
@@ -38,12 +38,7 @@
 
 extern const struct dispatch_source_vtable_s _dispatch_source_kevent_vtable;
 
-struct dispatch_kevent_s {
-	TAILQ_ENTRY(dispatch_kevent_s) dk_list;
-	TAILQ_HEAD(, dispatch_source_s) dk_sources;
-	struct kevent dk_kevent;
-};
-
+struct dispatch_kevent_s;
 typedef struct dispatch_kevent_s *dispatch_kevent_t;
 
 struct dispatch_timer_source_s {
@@ -54,6 +49,12 @@
 	uint64_t flags; // dispatch_timer_flags_t
 };
 
+struct dispatch_set_timer_params {
+	dispatch_source_t ds;
+	uintptr_t ident;
+	struct dispatch_timer_source_s values;
+};
+
 #define DSF_CANCELED 1u // cancellation has been requested
 
 struct dispatch_source_s {
@@ -98,5 +99,61 @@
 
 
 void _dispatch_source_legacy_xref_release(dispatch_source_t ds);
+dispatch_queue_t _dispatch_source_invoke(dispatch_source_t ds);
+bool _dispatch_source_probe(dispatch_source_t ds);
+void _dispatch_source_dispose(dispatch_source_t ds);
+size_t _dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz);
+void _dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent *ke);
 
+void _dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags, uint32_t del_flags);
+void _dispatch_kevent_merge(dispatch_source_t ds);
+void _dispatch_kevent_release(dispatch_source_t ds);
+void _dispatch_timer_list_update(dispatch_source_t ds);
+
+struct dispatch_source_type_s {
+        struct kevent ke;
+        uint64_t mask;
+        void (*init) (dispatch_source_t ds,
+                      dispatch_source_type_t type,
+		      uintptr_t handle,
+		      unsigned long mask,
+		      dispatch_queue_t q);
+};
+
+#define DISPATCH_TIMER_INDEX_WALL 0
+#define DISPATCH_TIMER_INDEX_MACH 1
+
+#ifdef DISPATCH_NO_LEGACY
+enum {
+        DISPATCH_TIMER_WALL_CLOCK       = 0x4,
+};
+enum {
+        DISPATCH_TIMER_INTERVAL = 0x0,
+        DISPATCH_TIMER_ONESHOT  = 0x1,
+        DISPATCH_TIMER_ABSOLUTE = 0x3,
+};
+enum {
+        DISPATCH_MACHPORT_DEAD = 0x1,
+        DISPATCH_MACHPORT_RECV = 0x2,
+        DISPATCH_MACHPORT_DELETED = 0x4,
+};
+#endif
+
+
+extern const struct dispatch_source_type_s _dispatch_source_type_timer;
+extern const struct dispatch_source_type_s _dispatch_source_type_read;
+extern const struct dispatch_source_type_s _dispatch_source_type_write;
+extern const struct dispatch_source_type_s _dispatch_source_type_proc;
+extern const struct dispatch_source_type_s _dispatch_source_type_signal;
+extern const struct dispatch_source_type_s _dispatch_source_type_vnode;
+extern const struct dispatch_source_type_s _dispatch_source_type_vfs;
+
+#ifdef HAVE_MACH
+extern const struct dispatch_source_type_s _dispatch_source_type_mach_send;
+extern const struct dispatch_source_type_s _dispatch_source_type_mach_recv;
+#endif
+
+extern const struct dispatch_source_type_s _dispatch_source_type_data_add;
+extern const struct dispatch_source_type_s _dispatch_source_type_data_or;
+
 #endif /* __DISPATCH_SOURCE_INTERNAL__ */

Modified: trunk/src/source_kevent.c
===================================================================
--- trunk/src/source_kevent.c	2009-11-12 23:27:00 UTC (rev 148)
+++ trunk/src/source_kevent.c	2009-11-13 00:03:21 UTC (rev 149)
@@ -19,37 +19,18 @@
  */
 
 #include "internal.h"
+#include "kevent_internal.h"
 #if HAVE_MACH
 #include "protocol.h"
 #include "protocolServer.h"
 #endif
 #include <sys/mount.h>
 
-#include "kevent_internal.h"
-
-#ifdef DISPATCH_NO_LEGACY
-enum {
-	DISPATCH_TIMER_WALL_CLOCK	= 0x4,
-};
-enum {
-	DISPATCH_TIMER_INTERVAL	= 0x0,
-	DISPATCH_TIMER_ONESHOT	= 0x1,
-	DISPATCH_TIMER_ABSOLUTE	= 0x3,
-};
-enum {
-	DISPATCH_MACHPORT_DEAD = 0x1,
-	DISPATCH_MACHPORT_RECV = 0x2,
-	DISPATCH_MACHPORT_DELETED = 0x4,
-};
-#endif
-
 #define DISPATCH_EVFILT_TIMER	(-EVFILT_SYSCOUNT - 1)
 #define DISPATCH_EVFILT_CUSTOM_ADD	(-EVFILT_SYSCOUNT - 2)
 #define DISPATCH_EVFILT_CUSTOM_OR	(-EVFILT_SYSCOUNT - 3)
 #define DISPATCH_EVFILT_SYSCOUNT	(EVFILT_SYSCOUNT + 3)
 
-#define DISPATCH_TIMER_INDEX_WALL 0
-#define DISPATCH_TIMER_INDEX_MACH 1
 static struct dispatch_kevent_s _dispatch_kevent_timer[] = {
 	{
 		.dk_kevent = {
@@ -86,32 +67,6 @@
 	.dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_add.dk_sources),
 };
 
-#ifndef DISPATCH_NO_LEGACY
-struct dispatch_source_attr_vtable_s {
-	DISPATCH_VTABLE_HEADER(dispatch_source_attr_s);
-};
-
-struct dispatch_source_attr_s {
-	DISPATCH_STRUCT_HEADER(dispatch_source_attr_s, dispatch_source_attr_vtable_s);
-	void* finalizer_ctxt;
-	dispatch_source_finalizer_function_t finalizer_func;
-	void* context;
-};
-#endif /* DISPATCH_NO_LEGACY */
-
-#define _dispatch_source_call_block ((void *)-1)
-static void _dispatch_source_latch_and_call(dispatch_source_t ds);
-static void _dispatch_source_cancel_callout(dispatch_source_t ds);
-static bool _dispatch_source_probe(dispatch_source_t ds);
-static void _dispatch_source_dispose(dispatch_source_t ds);
-static void _dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent *ke);
-static size_t _dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz);
-static size_t dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz);
-static dispatch_queue_t _dispatch_source_invoke(dispatch_source_t ds);
-static void _dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags, uint32_t del_flags);
-
-static void _dispatch_kevent_merge(dispatch_source_t ds);
-static void _dispatch_kevent_release(dispatch_source_t ds);
 static void _dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags);
 #if HAVE_MACH
 static void _dispatch_kevent_machport_resume(dispatch_kevent_t dk, uint32_t new_flags, uint32_t del_flags);
@@ -119,11 +74,9 @@
 static void _dispatch_kevent_machport_disable(dispatch_kevent_t dk);
 
 static void _dispatch_drain_mach_messages(struct kevent *ke);
+static void
 #endif
-static void _dispatch_timer_list_update(dispatch_source_t ds);
-
 #if HAVE_MACH
-static void
 _dispatch_mach_notify_source_init(void *context __attribute__((unused)));
 #endif
 
@@ -196,70 +149,8 @@
 	TAILQ_INSERT_TAIL(&_dispatch_sources[hash], dk, dk_list);
 }
 
-void
-dispatch_source_cancel(dispatch_source_t ds)
-{
 #if DISPATCH_DEBUG
-	dispatch_debug(ds, __FUNCTION__);
-#endif
-	// Right after we set the cancel flag, someone else
-	// could potentially invoke the source, do the cancelation, 
-	// unregister the source, and deallocate it. We would
-	// need to therefore retain/release before setting the bit
-
-	_dispatch_retain(ds);
-	dispatch_atomic_or(&ds->ds_atomic_flags, DSF_CANCELED);
-	_dispatch_wakeup(ds);
-	_dispatch_release(ds);
-}
-
-#ifndef DISPATCH_NO_LEGACY
 void
-_dispatch_source_legacy_xref_release(dispatch_source_t ds)
-{
-	if (ds->ds_is_legacy) {
-		if (!(ds->ds_timer.flags & DISPATCH_TIMER_ONESHOT)) {
-			dispatch_source_cancel(ds);
-		}
-
-		// Clients often leave sources suspended at the last release
-		dispatch_atomic_and(&ds->do_suspend_cnt, DISPATCH_OBJECT_SUSPEND_LOCK);
-	} else if (slowpath(DISPATCH_OBJECT_SUSPENDED(ds))) {
-		// Arguments for and against this assert are within 6705399
-		DISPATCH_CLIENT_CRASH("Release of a suspended object");
-	}
-	_dispatch_wakeup(ds);
-	_dispatch_release(ds);
-}
-#endif /* DISPATCH_NO_LEGACY */
-
-long
-dispatch_source_testcancel(dispatch_source_t ds)
-{
-	return (bool)(ds->ds_atomic_flags & DSF_CANCELED);
-}
-
-
-unsigned long
-dispatch_source_get_mask(dispatch_source_t ds)
-{
-	return ds->ds_pending_data_mask;
-}
-
-uintptr_t
-dispatch_source_get_handle(dispatch_source_t ds)
-{
-	return (int)ds->ds_ident_hack;
-}
-
-unsigned long
-dispatch_source_get_data(dispatch_source_t ds)
-{
-	return ds->ds_data;
-}
-
-#if DISPATCH_DEBUG
-void
 dispatch_debug_kevents(struct kevent* kev, size_t count, const char* str)
 {
 	size_t i;
@@ -372,96 +263,6 @@
 	_dispatch_kevent_resume(ds->ds_dkev, new_flags, del_flags);
 }
 
-dispatch_queue_t
-_dispatch_source_invoke(dispatch_source_t ds)
-{
-	// This function performs all source actions. Each action is responsible
-	// for verifying that it takes place on the appropriate queue. If the
-	// current queue is not the correct queue for this action, the correct queue
-	// will be returned and the invoke will be re-driven on that queue.
-
-	// The order of tests here in invoke and in probe should be consistent.
-	
-	dispatch_queue_t dq = _dispatch_queue_get_current();
-
-	if (!ds->ds_is_installed) {
-		// The source needs to be installed on the manager queue.
-		if (dq != &_dispatch_mgr_q) {
-			return &_dispatch_mgr_q;
-		}
-		_dispatch_kevent_merge(ds);
-	} else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == 0)) {
-		// The source has been cancelled and needs to be uninstalled from the
-		// manager queue. After uninstallation, the cancellation handler needs
-		// to be delivered to the target queue.
-		if (ds->ds_dkev) {
-			if (dq != &_dispatch_mgr_q) {
-				return &_dispatch_mgr_q;
-			}
-			_dispatch_kevent_release(ds);
-			return ds->do_targetq;
-		} else if (ds->ds_cancel_handler) {
-			if (dq != ds->do_targetq) {
-				return ds->do_targetq;
-			}
-		}	
-		_dispatch_source_cancel_callout(ds);
-	} else if (ds->ds_pending_data) {
-		// The source has pending data to deliver via the event handler callback
-		// on the target queue. Some sources need to be rearmed on the manager
-		// queue after event delivery.
-		if (dq != ds->do_targetq) {
-			return ds->do_targetq;
-		}
-		_dispatch_source_latch_and_call(ds);
-		if (ds->ds_needs_rearm) {
-			return &_dispatch_mgr_q;
-		}
-	} else if (ds->ds_needs_rearm && !ds->ds_is_armed) {
-		// The source needs to be rearmed on the manager queue.
-		if (dq != &_dispatch_mgr_q) {
-			return &_dispatch_mgr_q;
-		}
-		_dispatch_source_kevent_resume(ds, 0, 0);
-		ds->ds_is_armed = true;
-	}
-
-	return NULL;
-}
-
-bool
-_dispatch_source_probe(dispatch_source_t ds)
-{
-	// This function determines whether the source needs to be invoked.
-	// The order of tests here in probe and in invoke should be consistent.
-
-	if (!ds->ds_is_installed) {
-		// The source needs to be installed on the manager queue.
-		return true;
-	} else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == 0)) {
-		// The source needs to be uninstalled from the manager queue, or the
-		// cancellation handler needs to be delivered to the target queue.
-		// Note: cancellation assumes installation.
-		if (ds->ds_dkev || ds->ds_cancel_handler) {
-			return true;
-		}
-	} else if (ds->ds_pending_data) {
-		// The source has pending data to deliver to the target queue.
-		return true;
-	} else if (ds->ds_needs_rearm && !ds->ds_is_armed) {
-		// The source needs to be rearmed on the manager queue.
-		return true;
-	}
-	// Nothing to do.
-	return false;
-}
-
-void
-_dispatch_source_dispose(dispatch_source_t ds)
-{
-	_dispatch_queue_dispose((dispatch_queue_t)ds);
-}
-
 #ifndef DISPATCH_NO_LEGACY
 static void
 _dispatch_kevent_debugger2(void *context, dispatch_source_t unused __attribute__((unused)))
@@ -739,68 +540,6 @@
 	_dispatch_wakeup(ds);
 }
 
-void
-_dispatch_source_latch_and_call(dispatch_source_t ds)
-{
-	unsigned long prev;
-
-	if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == 0)) {
-		return;
-	}
-	prev = dispatch_atomic_xchg(&ds->ds_pending_data, 0);
-	if (ds->ds_is_level) {
-		ds->ds_data = ~prev;
-	} else {
-		ds->ds_data = prev;
-	}
-	if (dispatch_assume(prev)) {
-		if (ds->ds_handler_func) {
-#ifndef DISPATCH_NO_LEGACY
-			((dispatch_source_handler_function_t)ds->ds_handler_func)(ds->ds_handler_ctxt, ds);
-#else
-			ds->ds_handler_func(ds->ds_handler_ctxt);
-#endif
-		}
-	}
-}
-
-void
-_dispatch_source_cancel_callout(dispatch_source_t ds)
-{
-	ds->ds_pending_data_mask = 0;
-	ds->ds_pending_data = 0;
-	ds->ds_data = 0;
-
-#ifdef __BLOCKS__
-	if (ds->ds_handler_is_block) {
-		Block_release(ds->ds_handler_ctxt);
-		ds->ds_handler_is_block = false;
-		ds->ds_handler_func = NULL;
-		ds->ds_handler_ctxt = NULL;
-	}
-#endif
-
-	if (!ds->ds_cancel_handler) {
-		return;
-	}
-	if (ds->ds_cancel_is_block) {
-#ifdef __BLOCKS__
-		dispatch_block_t b = ds->ds_cancel_handler;
-		if (ds->ds_atomic_flags & DSF_CANCELED) {
-			b();
-		}
-		Block_release(ds->ds_cancel_handler);
-		ds->ds_cancel_is_block = false;
-#endif
-	} else {
-		dispatch_function_t f = ds->ds_cancel_handler;
-		if (ds->ds_atomic_flags & DSF_CANCELED) {
-			f(ds->do_ctxt);
-		}
-	}
-	ds->ds_cancel_handler = NULL;
-}
-
 const struct dispatch_source_vtable_s _dispatch_source_kevent_vtable = {
 	.do_type = DISPATCH_SOURCE_KEVENT_TYPE,
 	.do_kind = "kevent-source",
@@ -824,141 +563,7 @@
 	_dispatch_source_merge_kevent(ds, &kev);
 }
 
-size_t
-dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
-	dispatch_queue_t target = ds->do_targetq;
-	return snprintf(buf, bufsiz,
-			"target = %s[%p], pending_data = 0x%lx, pending_data_mask = 0x%lx, ",
-			target ? target->dq_label : "", target,
-			ds->ds_pending_data, ds->ds_pending_data_mask);
-}
-
-size_t
-_dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
-        size_t offset = 0;
-        offset += snprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ", dx_kind(ds), ds);
-        offset += dispatch_object_debug_attr(ds, &buf[offset], bufsiz - offset);
-        offset += dispatch_source_debug_attr(ds, &buf[offset], bufsiz - offset);
-        return offset;
-}
-
-#ifndef DISPATCH_NO_LEGACY
 static void
-dispatch_source_attr_dispose(dispatch_source_attr_t attr)
-{
-	// release the finalizer block if necessary
-	dispatch_source_attr_set_finalizer(attr, NULL);
-	_dispatch_dispose(attr);
-}
-
-static const struct dispatch_source_attr_vtable_s dispatch_source_attr_vtable = {
-	.do_type = DISPATCH_SOURCE_ATTR_TYPE,
-	.do_kind = "source-attr",
-	.do_dispose = dispatch_source_attr_dispose,
-};
-
-dispatch_source_attr_t
-dispatch_source_attr_create(void)
-{
-	dispatch_source_attr_t rval = calloc(1, sizeof(struct dispatch_source_attr_s));
-
-	if (rval) {
-		rval->do_vtable = &dispatch_source_attr_vtable;
-		rval->do_next = DISPATCH_OBJECT_LISTLESS;
-		rval->do_targetq = dispatch_get_global_queue(0, 0);
-		rval->do_ref_cnt = 1;
-		rval->do_xref_cnt = 1;
-	}
-
-	return rval;
-}
-
-void
-dispatch_source_attr_set_finalizer_f(dispatch_source_attr_t attr,
-	void *context, dispatch_source_finalizer_function_t finalizer)
-{
-#ifdef __BLOCKS__
-	if (attr->finalizer_func == (void*)_dispatch_call_block_and_release2) {
-		Block_release(attr->finalizer_ctxt);
-	}
-#endif
-
-	attr->finalizer_ctxt = context;
-	attr->finalizer_func = finalizer;
-}
-
-#ifdef __BLOCKS__
-long
-dispatch_source_attr_set_finalizer(dispatch_source_attr_t attr,
-	dispatch_source_finalizer_t finalizer)
-{
-	void *ctxt;
-	dispatch_source_finalizer_function_t func;
-
-	if (finalizer) {
-		if (!(ctxt = Block_copy(finalizer))) {
-			return 1;
-		}
-		func = (void *)_dispatch_call_block_and_release2;
-	} else {
-		ctxt = NULL;
-		func = NULL;
-	}
-
-	dispatch_source_attr_set_finalizer_f(attr, ctxt, func);
-
-	return 0;
-}
-
-dispatch_source_finalizer_t
-dispatch_source_attr_get_finalizer(dispatch_source_attr_t attr)
-{
-	if (attr->finalizer_func == (void*)_dispatch_call_block_and_release2) {
-		return (dispatch_source_finalizer_t)attr->finalizer_ctxt;
-	} else if (attr->finalizer_func == NULL) {
-		return NULL;
-	} else {
-		abort(); // finalizer is not a block...
-	}
-}
-#endif
-
-void
-dispatch_source_attr_set_context(dispatch_source_attr_t attr, void *context)
-{
-	attr->context = context;
-}
-
-dispatch_source_attr_t
-dispatch_source_attr_copy(dispatch_source_attr_t proto)
-{
-	dispatch_source_attr_t rval = NULL;
-
-	if (proto && (rval = malloc(sizeof(struct dispatch_source_attr_s)))) {
-		memcpy(rval, proto, sizeof(struct dispatch_source_attr_s));
-#ifdef __BLOCKS__
-		if (rval->finalizer_func == (void*)_dispatch_call_block_and_release2) {
-			rval->finalizer_ctxt = Block_copy(rval->finalizer_ctxt);
-		}
-#endif
-	} else if (!proto) {
-		rval = dispatch_source_attr_create();
-	}
-	return rval;
-}
-#endif /* DISPATCH_NO_LEGACY */
-
-
-struct dispatch_source_type_s {
-	struct kevent ke;
-	uint64_t mask;
-	void (*init)(dispatch_source_t ds, dispatch_source_type_t type,
-	    uintptr_t handle, unsigned long mask, dispatch_queue_t q);
-};
-
-static void
 dispatch_source_type_timer_init(dispatch_source_t ds, dispatch_source_type_t type, uintptr_t handle, unsigned long mask, dispatch_queue_t q)
 {
 	ds->ds_needs_rearm = true;
@@ -1088,286 +693,6 @@
 	},
 };
 
-dispatch_source_t
-dispatch_source_create(dispatch_source_type_t type,
-	uintptr_t handle,
-	unsigned long mask,
-	dispatch_queue_t q)
-{
-	const struct kevent *proto_kev = &type->ke;
-	dispatch_source_t ds = NULL;
-	static char source_label[sizeof(ds->dq_label)] = "source";
-	dispatch_kevent_t dk = NULL;
-
-	// input validation
-	if (type == NULL || (mask & ~type->mask)) {
-		goto out_bad;
-	}
-
-	switch (type->ke.filter) {
-	case EVFILT_SIGNAL:
-		if (handle >= NSIG) {
-			goto out_bad;
-		}
-		break;
-	case EVFILT_FS:
-	case DISPATCH_EVFILT_CUSTOM_ADD:
-	case DISPATCH_EVFILT_CUSTOM_OR:
-	case DISPATCH_EVFILT_TIMER:
-		if (handle) {
-			goto out_bad;
-		}
-		break;
-	default:
-		break;
-	}
-	
-	ds = calloc(1ul, sizeof(struct dispatch_source_s));
-	if (slowpath(!ds)) {
-		goto out_bad;
-	}
-	dk = calloc(1ul, sizeof(struct dispatch_kevent_s));
-	if (slowpath(!dk)) {
-		goto out_bad;
-	}
-
-	dk->dk_kevent = *proto_kev;
-	dk->dk_kevent.ident = handle;
-	dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
-	dk->dk_kevent.fflags |= (uint32_t)mask;
-	dk->dk_kevent.udata = dk;
-	TAILQ_INIT(&dk->dk_sources);
-
-	// Initialize as a queue first, then override some settings below.
-	_dispatch_queue_init((dispatch_queue_t)ds);
-	memcpy(ds->dq_label, source_label, sizeof(source_label));
-
-	// Dispatch Object
-	ds->do_vtable = &_dispatch_source_kevent_vtable;
-	ds->do_ref_cnt++; // the reference the manger queue holds
-	ds->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_INTERVAL;
-	// do_targetq will be retained below, past point of no-return
-	ds->do_targetq = q;
-
-	// Dispatch Source
-	ds->ds_ident_hack = dk->dk_kevent.ident;
-	ds->ds_dkev = dk;
-	ds->ds_pending_data_mask = dk->dk_kevent.fflags;
-	if ((EV_DISPATCH|EV_ONESHOT) & proto_kev->flags) {
-		ds->ds_is_level = true;
-		ds->ds_needs_rearm = true;
-	} else if (!(EV_CLEAR & proto_kev->flags)) {
-		// we cheat and use EV_CLEAR to mean a "flag thingy"
-		ds->ds_is_adder = true;
-	}
-	
-	// Some sources require special processing
-	if (type->init != NULL) {
-		type->init(ds, type, handle, mask, q);
-	}
-	dispatch_assert(!(ds->ds_is_level && ds->ds_is_adder));
-#if DISPATCH_DEBUG
-	dispatch_debug(ds, __FUNCTION__);
-#endif
-
-	_dispatch_retain(ds->do_targetq);
-	return ds;
-	
-out_bad:
-	free(ds);
-	free(dk);
-	return NULL;
-}
-
-#ifdef __BLOCKS__
-// 6618342 Contact the team that owns the Instrument DTrace probe before renaming this symbol
-static void
-_dispatch_source_set_event_handler2(void *context)
-{
-	struct Block_layout *bl = context;
-
-	dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
-	dispatch_assert(ds->do_vtable == &_dispatch_source_kevent_vtable);
-	
-	if (ds->ds_handler_is_block && ds->ds_handler_ctxt) {
-		Block_release(ds->ds_handler_ctxt);
-	}
-	ds->ds_handler_func = bl ? (void *)bl->invoke : NULL;
-	ds->ds_handler_ctxt = bl;
-	ds->ds_handler_is_block = true;
-}
-
-void
-dispatch_source_set_event_handler(dispatch_source_t ds, dispatch_block_t handler)
-{
-	dispatch_assert(!ds->ds_is_legacy);
-	handler = _dispatch_Block_copy(handler);
-	dispatch_barrier_async_f((dispatch_queue_t)ds,
-		handler, _dispatch_source_set_event_handler2);
-}
-#endif /* __BLOCKS__ */
-
-static void
-_dispatch_source_set_event_handler_f(void *context)
-{
-	dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
-	dispatch_assert(ds->do_vtable == &_dispatch_source_kevent_vtable);
-	
-#ifdef __BLOCKS__
-	if (ds->ds_handler_is_block && ds->ds_handler_ctxt) {
-		Block_release(ds->ds_handler_ctxt);
-	}
-#endif
-	ds->ds_handler_func = context;
-	ds->ds_handler_ctxt = ds->do_ctxt;
-	ds->ds_handler_is_block = false;
-}
-
-void
-dispatch_source_set_event_handler_f(dispatch_source_t ds,
-	dispatch_function_t handler)
-{
-	dispatch_assert(!ds->ds_is_legacy);
-	dispatch_barrier_async_f((dispatch_queue_t)ds,
-		handler, _dispatch_source_set_event_handler_f);
-}
-
-#ifdef __BLOCKS__
-// 6618342 Contact the team that owns the Instrument DTrace probe before renaming this symbol
-static void
-_dispatch_source_set_cancel_handler2(void *context)
-{
-	dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
-	dispatch_assert(ds->do_vtable == &_dispatch_source_kevent_vtable);
-	
-	if (ds->ds_cancel_is_block && ds->ds_cancel_handler) {
-		Block_release(ds->ds_cancel_handler);
-	}
-	ds->ds_cancel_handler = context;
-	ds->ds_cancel_is_block = true;
-}
-
-void
-dispatch_source_set_cancel_handler(dispatch_source_t ds,
-	dispatch_block_t handler)
-{
-	dispatch_assert(!ds->ds_is_legacy);
-	handler = _dispatch_Block_copy(handler);
-	dispatch_barrier_async_f((dispatch_queue_t)ds,
-							 handler, _dispatch_source_set_cancel_handler2);
-}
-#endif /* __BLOCKS__ */
-
-static void
-_dispatch_source_set_cancel_handler_f(void *context)
-{
-	dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
-	dispatch_assert(ds->do_vtable == &_dispatch_source_kevent_vtable);
-	
-#ifdef __BLOCKS__
-	if (ds->ds_cancel_is_block && ds->ds_cancel_handler) {
-		Block_release(ds->ds_cancel_handler);
-	}
-#endif
-	ds->ds_cancel_handler = context;
-	ds->ds_cancel_is_block = false;
-}
-
-void
-dispatch_source_set_cancel_handler_f(dispatch_source_t ds,
-	dispatch_function_t handler)
-{
-	dispatch_assert(!ds->ds_is_legacy);
-	dispatch_barrier_async_f((dispatch_queue_t)ds,
-							 handler, _dispatch_source_set_cancel_handler_f);
-}
-
-#ifndef DISPATCH_NO_LEGACY
-// 6618342 Contact the team that owns the Instrument DTrace probe before renaming this symbol
-dispatch_source_t
-_dispatch_source_create2(dispatch_source_t ds,
-	dispatch_source_attr_t attr,
-	void *context,
-	dispatch_source_handler_function_t handler)
-{
-	if (ds == NULL || handler == NULL) {
-		return NULL;
-	}
-
-	ds->ds_is_legacy = true;
-
-	ds->ds_handler_func = (dispatch_function_t)handler;
-	ds->ds_handler_ctxt = context;
-		
-	if (attr && attr != DISPATCH_SOURCE_CREATE_SUSPENDED) {
-		ds->dq_finalizer_ctxt = attr->finalizer_ctxt;
-		ds->dq_finalizer_func = (typeof(ds->dq_finalizer_func))attr->finalizer_func;
-		ds->do_ctxt = attr->context;
-	}
-#ifdef __BLOCKS__
-	if (ds->dq_finalizer_func == (void*)_dispatch_call_block_and_release2) {
-		ds->dq_finalizer_ctxt = Block_copy(ds->dq_finalizer_ctxt);
-		if (!ds->dq_finalizer_ctxt) {
-			goto out_bad;
-		}
-	}
-	if (handler == _dispatch_source_call_block) {
-		struct Block_layout *bl = ds->ds_handler_ctxt = Block_copy(context);
-		if (!ds->ds_handler_ctxt) {
-			if (ds->dq_finalizer_func == (void*)_dispatch_call_block_and_release2) {
-				Block_release(ds->dq_finalizer_ctxt);
-			}
-			goto out_bad;
-		}
-		ds->ds_handler_func = (void *)bl->invoke;
-		ds->ds_handler_is_block = true;
-	}
-
-	// all legacy sources get a cancellation event on the normal event handler.
-	dispatch_function_t func = ds->ds_handler_func;
-	dispatch_source_handler_t block = ds->ds_handler_ctxt;
-	void *ctxt = ds->ds_handler_ctxt;
-	bool handler_is_block = ds->ds_handler_is_block;
-	
-	ds->ds_cancel_is_block = true;
-	if (handler_is_block) {
-		ds->ds_cancel_handler = _dispatch_Block_copy(^{
-			block(ds);
-		});
-	} else {
-		ds->ds_cancel_handler = _dispatch_Block_copy(^{
-			((dispatch_source_handler_function_t)func)(ctxt, ds);
-		});
-	}
-#endif
-	if (attr != DISPATCH_SOURCE_CREATE_SUSPENDED) {
-		dispatch_resume(ds);
-	}
-
-	return ds;
-
-#ifdef __BLOCKS__
-out_bad:
-	free(ds);
-	return NULL;
-#endif
-}
-
-long
-dispatch_source_get_error(dispatch_source_t ds, long *err_out)
-{
-	// 6863892 don't report ECANCELED until kevent is unregistered
-	if ((ds->ds_atomic_flags & DSF_CANCELED) && !ds->ds_dkev) {
-		if (err_out) {
-			*err_out = ECANCELED;
-		}
-		return DISPATCH_ERROR_DOMAIN_POSIX;
-	} else {
-		return DISPATCH_ERROR_DOMAIN_NO_ERROR;
-	}
-}
-#endif /* DISPATCH_NO_LEGACY */
-
 // Updates the ordered list of timers based on next fire date for changes to ds.
 // Should only be called from the context of _dispatch_mgr_q.
 void
@@ -1527,121 +852,6 @@
 	return howsoon;
 }
 
-struct dispatch_set_timer_params {
-	dispatch_source_t ds;
-	uintptr_t ident;
-	struct dispatch_timer_source_s values;
-};
-
-// To be called from the context of the _dispatch_mgr_q
-static void
-_dispatch_source_set_timer2(void *context)
-{
-	struct dispatch_set_timer_params *params = context;
-	dispatch_source_t ds = params->ds;
-	ds->ds_ident_hack = params->ident;
-	ds->ds_timer = params->values;
-	_dispatch_timer_list_update(ds);
-	dispatch_resume(ds);
-	dispatch_release(ds);
-	free(params);
-}
-
-void
-dispatch_source_set_timer(dispatch_source_t ds,
-	dispatch_time_t start,
-	uint64_t interval,
-	uint64_t leeway)
-{
-	struct dispatch_set_timer_params *params;
-	
-	// we use zero internally to mean disabled
-	if (interval == 0) {
-		interval = 1;
-	} else if ((int64_t)interval < 0) {
-		// 6866347 - make sure nanoseconds won't overflow
-		interval = INT64_MAX;
-	}
-
-	// Suspend the source so that it doesn't fire with pending changes
-	// The use of suspend/resume requires the external retain/release
-	dispatch_retain(ds);
-	dispatch_suspend(ds);
-	
-	if (start == DISPATCH_TIME_NOW) {
-		start = _dispatch_absolute_time();
-	} else if (start == DISPATCH_TIME_FOREVER) {
-		start = INT64_MAX;
-	}
-	if ((int64_t)leeway < 0) {
-		leeway = INT64_MAX;
-	}
-
-	while (!(params = malloc(sizeof(struct dispatch_set_timer_params)))) {
-		sleep(1);
-	}
-
-	params->ds = ds;
-	params->values.flags = ds->ds_timer.flags;
-
-	if ((int64_t)start < 0) {
-		// wall clock
-		params->ident = DISPATCH_TIMER_INDEX_WALL;
-		params->values.start = -((int64_t)start);
-		params->values.target = -((int64_t)start);
-		params->values.interval = interval;
-		params->values.leeway = leeway;
-		params->values.flags |= DISPATCH_TIMER_WALL_CLOCK;
-	} else {
-		// absolute clock
-		params->ident = DISPATCH_TIMER_INDEX_MACH;
-		params->values.start = start;
-		params->values.target = start;
-		params->values.interval = _dispatch_time_nano2mach(interval);
-		params->values.leeway = _dispatch_time_nano2mach(leeway);
-		params->values.flags &= ~DISPATCH_TIMER_WALL_CLOCK;
-	}
-
-	dispatch_barrier_async_f(&_dispatch_mgr_q, params, _dispatch_source_set_timer2);
-}
-
-#ifndef DISPATCH_NO_LEGACY
-// LEGACY
-long
-dispatch_source_timer_set_time(dispatch_source_t ds, uint64_t nanoseconds, uint64_t leeway)
-{
-	dispatch_time_t start;
-	if (nanoseconds == 0) {
-		nanoseconds = 1;
-	}
-	if (ds->ds_timer.flags == (DISPATCH_TIMER_ABSOLUTE|DISPATCH_TIMER_WALL_CLOCK)) {
-		static const struct timespec t0;
-		start = dispatch_walltime(&t0, nanoseconds);
-	} else if (ds->ds_timer.flags & DISPATCH_TIMER_WALL_CLOCK) {
-		start = dispatch_walltime(DISPATCH_TIME_NOW, nanoseconds);
-	} else {
-		start = dispatch_time(DISPATCH_TIME_NOW, nanoseconds);
-	}
-	if (ds->ds_timer.flags & (DISPATCH_TIMER_ABSOLUTE|DISPATCH_TIMER_ONESHOT)) {
-		// 6866347 - make sure nanoseconds won't overflow
-		nanoseconds = INT64_MAX; // non-repeating (~292 years)
-	}
-	dispatch_source_set_timer(ds, start, nanoseconds, leeway);
-	return 0;
-}
-
-// LEGACY
-uint64_t
-dispatch_event_get_nanoseconds(dispatch_source_t ds)
-{
-	if (ds->ds_timer.flags & DISPATCH_TIMER_WALL_CLOCK) {
-		return ds->ds_timer.interval;
-	} else {
-		return _dispatch_time_mach2nano(ds->ds_timer.interval);
-	}
-}
-#endif /* DISPATCH_NO_LEGACY */
-
 #if HAVE_MACH
 static dispatch_source_t _dispatch_mach_notify_source;
 static mach_port_t _dispatch_port_set;
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.macosforge.org/pipermail/libdispatch-changes/attachments/20091112/4efae622/attachment-0001.html>


More information about the libdispatch-changes mailing list