2004-04-08 Marcus Brinkmann <marcus@gnu.org>

	* bucket-inject.c, obj-copy-out.c: New files.
	* Makefile.am (libhurd_cap_server_a_SOURCES): Add bucket-inject.c
	and obj-copy-out.c
	* bucket-create.c (hurd_cap_bucket_create): Initialize members
	is_managed, nr_caps, waiting_rpcs of BUCKET.  Set R_BUCKET.
	* cap-server-intern.h (_hurd_cap_client_dealloc): Add new argument
	BUCKET to prototype.
	(struct hurd_cap_client): Remove declaration.
	(struct _hurd_cap_list_item): Add new member tid.  Change type for
	member client to _hurd_cap_client_t.
	(_hurd_cap_list_item_add, _hurd_cap_list_item_remove,
	_hurd_cap_list_item_dequeued): New inline functions.
	(struct _hurd_cap_obj_entry): Rename member IDX to ID.
	(_hurd_cap_obj_copy_out): New prototype.
	(_hurd_cap_client_create): Remove argument R_IDX from prototype.
	(struct _hurd_cap_bucket): Add new members MANAGER, IS_MANAGED,
	IS_MANAGER_WAITING, NR_CAPS, WAITING_RPCS, and FREE_WORKER.
	(_hurd_cap_client_t): Type definition moved to ...
	* cap-server.h (_hurd_cap_client_t): Here.
	(struct _hurd_cap_client_t): New declaration.
	(struct hurd_cap_rpc_context): Define it.
	(hurd_cap_class_demux_t): Renamed to ...
	(hurd_cap_class_demuxer_t): ... this.
	(hurd_cap_class_create, hurd_cap_class_init): Use new type for demuxer
	argument in prototype.
	(hurd_cap_bucket_inject): New prototype.
	* cap-server.h: Include <hurd/types.h>
	* class-create (hurd_cap_class_create): Use new type for demuxer
	argument.  Set R_CLASS.
	* class-init (hurd_cap_class_init): Use new type for demuxer argument.
	* client-release.c (_hurd_cap_client_dealloc): Take new argument
	BUCKET.  New local variable NR_CAPS.  Keep track of number of
	capabilities removed.  Update BUCKET->nr_caps before return.
	(_hurd_cap_client_release): Pass new argument BUCKET to
	_hurd_cap_client_release.
	* client-create.c (_hurd_cap_client_create): Remove argument
	R_IDX.  Consequently, do not set R_IDX anymore.  Set R_CLIENT.
	Pass new argument BUCKET to _hurd_cap_client_dealloc.
	* bucket-inhibit.c (hurd_cap_bucket_end): Check BUCKET->nr_caps if
	FORCE flag is not set.  Cancel the manager thread if needed.
	(_hurd_cap_bucket_cond_busy): Move to ...
	* cap-server-intern.h (_hurd_cap_bucket_cond_busy): ... here.
	Add attribute always-inline.
	(_hurd_cap_bucket_cond_check): New inline function.
	* client-inhibit.c (_hurd_cap_client_cond_busy): Move to ...
	* cap-server-intern.h (_hurd_cap_client_cond_busy): ... here.
	Add attribute always-inline.
	(_hurd_cap_client_cond_check): New inline function.
	* class-inhibit.c (_hurd_cap_class_cond_busy): Move to ...
	* cap-server-intern.h (_hurd_cap_class_cond_busy): ... here.
	Add attribute always-inline.
	(_hurd_cap_class_cond_check): New inline function.
	* obj-inhibit.c (_hurd_cap_obj_cond_busy): Move to ...
	* cap-server-intern.h (_hurd_cap_obj_cond_busy): ... here.
	Add attribute always-inline.
	(_hurd_cap_obj_cond_check): New inline function.

18 files changed, 1407 insertions, 154 deletions

diff --git a/libhurd-cap-server/ChangeLog b/libhurd-cap-server/ChangeLog
index f70d923..206567e 100644
--- a/libhurd-cap-server/ChangeLog
+++ b/libhurd-cap-server/ChangeLog
@@ -1,3 +1,62 @@
+2004-04-08  Marcus Brinkmann  <marcus@gnu.org>
+
+	* bucket-inject.c, obj-copy-out.c: New files.
+	* Makefile.am (libhurd_cap_server_a_SOURCES): Add bucket-inject.c
+	and obj-copy-out.c
+	* bucket-create.c (hurd_cap_bucket_create): Initialize members
+	is_managed, nr_caps, waiting_rpcs of BUCKET.  Set R_BUCKET.
+	* cap-server-intern.h (_hurd_cap_client_dealloc): Add new argument
+	BUCKET to prototype.
+	(struct hurd_cap_client): Remove declaration.
+	(struct _hurd_cap_list_item): Add new member tid.  Change type for
+	member client to _hurd_cap_client_t.
+	(_hurd_cap_list_item_add, _hurd_cap_list_item_remove,
+	_hurd_cap_list_item_dequeued): New inline functions.
+	(struct _hurd_cap_obj_entry): Rename member IDX to ID.
+	(_hurd_cap_obj_copy_out): New prototype.
+	(_hurd_cap_client_create): Remove argument R_IDX from prototype.
+	(struct _hurd_cap_bucket): Add new members MANAGER, IS_MANAGED,
+	IS_MANAGER_WAITING, NR_CAPS, WAITING_RPCS, and FREE_WORKER.
+	(_hurd_cap_client_t): Type definition moved to ...
+	* cap-server.h (_hurd_cap_client_t): Here.
+	(struct _hurd_cap_client_t): New declaration.
+	(struct hurd_cap_rpc_context): Define it.
+	(hurd_cap_class_demux_t): Renamed to ...
+	(hurd_cap_class_demuxer_t): ... this.
+	(hurd_cap_class_create, hurd_cap_class_init): Use new type for demuxer
+	argument in prototype.
+	(hurd_cap_bucket_inject): New prototype.
+	* cap-server.h: Include <hurd/types.h>
+	* class-create (hurd_cap_class_create): Use new type for demuxer
+	argument.  Set R_CLASS.
+	* class-init (hurd_cap_class_init): Use new type for demuxer argument.
+	* client-release.c (_hurd_cap_client_dealloc): Take new argument
+	BUCKET.  New local variable NR_CAPS.  Keep track of number of
+	capabilities removed.  Update BUCKET->nr_caps before return.
+	(_hurd_cap_client_release): Pass new argument BUCKET to
+	_hurd_cap_client_release.
+	* client-create.c (_hurd_cap_client_create): Remove argument
+	R_IDX.  Consequently, do not set R_IDX anymore.  Set R_CLIENT.
+	Pass new argument BUCKET to _hurd_cap_client_dealloc.
+	* bucket-inhibit.c (hurd_cap_bucket_end): Check BUCKET->nr_caps if
+	FORCE flag is not set.  Cancel the manager thread if needed.
+	(_hurd_cap_bucket_cond_busy): Move to ...
+	* cap-server-intern.h (_hurd_cap_bucket_cond_busy): ... here.
+	Add attribute always-inline.
+	(_hurd_cap_bucket_cond_check): New inline function.
+	* client-inhibit.c (_hurd_cap_client_cond_busy): Move to ...
+	* cap-server-intern.h (_hurd_cap_client_cond_busy): ... here.
+	Add attribute always-inline.
+	(_hurd_cap_client_cond_check): New inline function.
+	* class-inhibit.c (_hurd_cap_class_cond_busy): Move to ...
+	* cap-server-intern.h (_hurd_cap_class_cond_busy): ... here.
+	Add attribute always-inline.
+	(_hurd_cap_class_cond_check): New inline function.
+	* obj-inhibit.c (_hurd_cap_obj_cond_busy): Move to ...
+	* cap-server-intern.h (_hurd_cap_obj_cond_busy): ... here.
+	Add attribute always-inline.
+	(_hurd_cap_obj_cond_check): New inline function.
+
 2004-04-01  Marcus Brinkmann  <marcus@gnu.org>
 
 	* Makefile.am (libhurd_cap_server_a_SOURCES): Remove bucket-add.c,
diff --git a/libhurd-cap-server/Makefile.am b/libhurd-cap-server/Makefile.am
index 301a4a8..bbeb779 100644
--- a/libhurd-cap-server/Makefile.am
+++ b/libhurd-cap-server/Makefile.am
@@ -34,7 +34,7 @@ libhurd_cap_server_a_SOURCES = table.h table.c				\
 	class-destroy.c class-free.c					\
 	class-alloc.c class-inhibit.c					\
 	obj-dealloc.c obj-drop.c obj-inhibit.c				\
-	obj-entry-space.c						\
+	obj-entry-space.c obj-copy-out.c				\
 	client-create.c client-release.c client-inhibit.c		\
 	bucket-create.c bucket-free.c bucket-inhibit.c			\
-	bucket-manage-mt.c
+	bucket-manage-mt.c bucket-inject.c
diff --git a/libhurd-cap-server/README b/libhurd-cap-server/README
index eb1612a..d77d09a 100644
--- a/libhurd-cap-server/README
+++ b/libhurd-cap-server/README
@@ -1,7 +1,10 @@
 TODO:
 
+* Keep track of number of allocated capabilities or otherwise enable
+  to check if a bucket is still in use!!!  (so at least a per-client
+  count is required).
 * Implement bucket-inhibit correctly. !!!  Also see:
-* Decide if it is needed to prevent capability allocation
+* Decide if it is needed to allow capability allocation
   outside of RPCs (as return arguments).  In that case, allocation
   must be blocked while it is checked if no extinct users exists -
   otherwise inhibiting RPCs is good enough.
diff --git a/libhurd-cap-server/bucket-create.c b/libhurd-cap-server/bucket-create.c
index bfa99f0..65a61da 100644
--- a/libhurd-cap-server/bucket-create.c
+++ b/libhurd-cap-server/bucket-create.c
@@ -125,6 +125,7 @@ hurd_cap_bucket_create (hurd_cap_bucket_t *r_bucket)
   if (err)
     goto err_lock;
 
+  bucket->is_managed = false;
   bucket->state = _HURD_CAP_STATE_GREEN;
 
   err = pthread_cond_init (&bucket->cond, NULL);
@@ -134,7 +135,9 @@ hurd_cap_bucket_create (hurd_cap_bucket_t *r_bucket)
   /* The member cond_waiter will be initialized when the state changes
      to _HURD_CAP_STATE_YELLOW.  */
 
+  bucket->nr_caps = 0;
   bucket->pending_rpcs = NULL;
+  bucket->waiting_rpcs = NULL;
 
   hurd_ihash_init (&bucket->senders,
 		   offsetof (struct _hurd_cap_list_item, locp));
@@ -152,6 +155,7 @@ hurd_cap_bucket_create (hurd_cap_bucket_t *r_bucket)
   bucket->client_death_notify.hook = bucket;
   hurd_task_death_notify_add (&bucket->client_death_notify);
 
+  *r_bucket = bucket;
   return 0;
 
 #if 0
diff --git a/libhurd-cap-server/bucket-inhibit.c b/libhurd-cap-server/bucket-inhibit.c
index 1e0d400..09e15f9 100644
--- a/libhurd-cap-server/bucket-inhibit.c
+++ b/libhurd-cap-server/bucket-inhibit.c
@@ -30,23 +30,6 @@
 #include "cap-server-intern.h"
 
 
-/* Return true if there are still outstanding RPCs in this bucket, and
-   fails if not.  This is only valid if hurd_cap_bucket_inhibit is in
-   progress (ie, if bucket->state is _HURD_CAP_STATE_YELLOW).
-   FIXME: We will need this in the RPC worker thread code, where the
-   last worker will get false as return value and then has to change
-   the state to RED and signal (broadcast?) the condition.  */
-static inline int
-_hurd_cap_bucket_cond_busy (hurd_cap_bucket_t bucket)
-{
-  /* We have to remain in the state yellow until there are no pending
-     RPC threads except maybe the waiter.  */
-  return bucket->pending_rpcs
-    && (bucket->pending_rpcs->thread != bucket->cond_waiter
-	|| bucket->pending_rpcs->next);
-}
-
-
 /* Inhibit all RPCs on the capability bucket BUCKET (which must not
    be locked).  You _must_ follow up with a hurd_cap_bucket_resume
    operation, and hold at least one reference to the object
@@ -62,6 +45,8 @@ hurd_cap_bucket_inhibit (hurd_cap_bucket_t bucket)
      this function is called within an RPC, we are going to be
      canceled anyway.  Otherwise, it ensures that bucket inhibitions
      are fully serialized (per bucket).  */
+  /* FIXME: Do something if the state is _HURD_CAP_STATE_BLACK?  Can
+     only happen if we are called from outside any RPCs.  */
   while (bucket->state != _HURD_CAP_STATE_GREEN)
     {
       err = hurd_cond_wait (&bucket->cond, &bucket->lock);
@@ -137,20 +122,19 @@ hurd_cap_bucket_end (hurd_cap_bucket_t bucket, bool force)
 {
   pthread_mutex_lock (&bucket->lock);
 
-  /* FIXME: Check if we can go away while honoring the FORCE flag.  */
+  if (!force && bucket->nr_caps)
+    {
+      pthread_mutex_unlock (&bucket->lock);
+      return EBUSY;
+    }
 
   bucket->state = _HURD_CAP_STATE_BLACK;
 
   /* Broadcast the change to all potential waiters.  */
   pthread_cond_broadcast (&bucket->cond);
 
-  /* FIXME: This is not enough.  We need to cancel or otherwise notify
-     the manager thread that it is supposed to go away, if it is
-     currently blocking in an open receive.  For example, we could
-     send it a special message.  However, the manager thread must
-     ensure that our own thread is correctly finished (ie, if we are
-     called from within an RPC, the RPC must finish before the manager
-     thread is allowed to go away.  */
+  if (bucket->is_managed && bucket->cond_waiter != bucket->manager)
+    pthread_cancel (bucket->manager);
 
   pthread_mutex_unlock (&bucket->lock);
 }
diff --git a/libhurd-cap-server/bucket-inject.c b/libhurd-cap-server/bucket-inject.c
new file mode 100644
index 0000000..4363333
--- /dev/null
+++ b/libhurd-cap-server/bucket-inject.c
@@ -0,0 +1,61 @@
+/* bucket-inject.c - Copy out a capability to a client.
+   Copyright (C) 2004 Free Software Foundation, Inc.
+   Written by Marcus Brinkmann <marcus@gnu.org>
+
+   This file is part of the GNU Hurd.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2.1 of the License, or (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this program; if not, write to the Free
+   Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
+   02111-1307 USA.  */
+
+#if HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <errno.h>
+#include <pthread.h>
+#include <stdlib.h>
+
+#include "cap-server-intern.h"
+
+
+/* Copy out a capability for the capability OBJ to the client with the
+   task ID TASK_ID.  Returns the capability (valid only for this user)
+   in *R_CAP, or an error.  It is not safe to call this from outside
+   an RPC on OBJ while the manager is running.  */
+error_t
+hurd_cap_bucket_inject (hurd_cap_bucket_t bucket, hurd_cap_obj_t obj,
+			hurd_task_id_t task_id, hurd_cap_t *r_cap)
+{
+  error_t err;
+  _hurd_cap_client_t client;
+  hurd_cap_id_t cap_id;
+
+  /* FIXME: Of course, this function needs to take a task info
+     capability and pass it to client create.  */
+
+  err = _hurd_cap_client_create (bucket, task_id, &client);
+  if (err)
+    return err;
+
+  pthread_mutex_lock (&obj->lock);
+  err = _hurd_cap_obj_copy_out (obj, bucket, client, &cap_id);
+  pthread_mutex_unlock (&obj->lock);
+  _hurd_cap_client_release (bucket, client->id);
+  if (err)
+    return err;
+
+  return hurd_cap_make (client->id, cap_id);
+}
+
diff --git a/libhurd-cap-server/bucket-manage-mt.c b/libhurd-cap-server/bucket-manage-mt.c
index 6369fcb..790f4d8 100644
--- a/libhurd-cap-server/bucket-manage-mt.c
+++ b/libhurd-cap-server/bucket-manage-mt.c
@@ -26,34 +26,879 @@
 #include <errno.h>
 #include <stdlib.h>
 #include <assert.h>
+#include <stdint.h>
 #include <pthread.h>
 
 #include "cap-server-intern.h"
 
 
-/* Start managing RPCs on the bucket BUCKET.  The BOOTSTRAP capability
+/* FIXME: Throughout this file, for debugging the behaviour could be
+   relaxed to return errors to callers which would otherwise be
+   ignored (due to malformed requests etc).  */
+
+
+/* FIXME: This section contains a lot of random junk that maybe should
+   be somewhere else (or not).  */
+
+/* Cancel the pending RPC of the specified thread.  */
+#define HURD_CAP_MSG_LABEL_CANCEL	0x100
+#define HURD_CAP_MSG_LABEL_GET_ROOT	0x101
+
+/* Some error for this.  */
+#define ECAP_NOREPLY 0x10001
+#define ECAP_DIED    0x10002
+
+
+/* The msg labels of the reply from the worker to the manager.  */
+#define _HURD_CAP_MSG_WORKER_ACCEPTED 0
+#define _HURD_CAP_MSG_WORKER_REJECTED 1
+
+struct worker_info
+{
+  /* The bucket.  */
+  hurd_cap_bucket_t bucket;
+
+  /* The manager thread.  */
+  l4_thread_id_t manager_tid;
+
+  /* The timeout for the worker thread as an L4 time period.  */
+  l4_time_t timeout;
+};
+
+
+static void
+__attribute__((always_inline))
+reply_err (l4_thread_id_t to, error_t err)
+{
+#define HURD_L4_ERROR_LABEL ((uint16_t) INT16_MIN)
+#define HURD_L4_ERROR_TAG ((HURD_L4_ERROR_LABEL << 16) & 1)
+  l4_set_msg_tag (HURD_L4_ERROR_TAG);
+  l4_load_mr (1, err);
+  l4_reply (to);
+}
+
+
+/* Lookup the client with client ID CLIENT_ID and return it in
+   *R_CLIENT with one reference for the entry in the bucket.  It is
+   *verified that the client is in fact the one with the task ID
+   *TASK_ID.  */
+static error_t
+__attribute__((always_inline))
+lookup_client (hurd_cap_bucket_t bucket, hurd_cap_client_id_t client_id,
+	       hurd_task_id_t task_id, _hurd_cap_client_t *r_client)
+{
+  error_t err = 0;
+  _hurd_cap_client_entry_t entry;
+
+  pthread_mutex_lock (&bucket->lock);
+  /* Look up the client by its ID.  */
+  entry = hurd_table_lookup (&bucket->clients, client_id);
+  if (!entry || entry->dead || entry->client->task_id != task_id)
+    err = ECAP_NOREPLY;
+  else
+    {
+      entry->refs++;
+      *r_client = entry->client;
+    }
+  pthread_mutex_unlock (&bucket->lock);
+
+  return err;
+}
+
+
+/* Process the message MSG from thread FROM in worker thread WORKER of
+   bucket BUCKET.  Return ECAP_NOREPLY if no reply should be sent.  Any
+   other error will be replied to the user.  If 0 is returned, MSG
+   must contain the reply message.  */
+static error_t
+__attribute__((always_inline))
+manage_demuxer (hurd_cap_rpc_context_t ctx, _hurd_cap_list_item_t worker)
+{
+  error_t err = 0;
+  hurd_cap_bucket_t bucket = ctx->bucket;
+  l4_thread_id_t from = ctx->from;
+  _hurd_cap_client_t client;
+  hurd_cap_t cap;
+  hurd_cap_class_t cap_class;
+  hurd_cap_obj_t obj;
+  _hurd_cap_obj_entry_t obj_entry;
+  struct _hurd_cap_list_item worker_client;
+  struct _hurd_cap_list_item worker_class;
+  struct _hurd_cap_list_item worker_obj;
+
+  worker_client.thread = worker->thread;
+  worker_client.tid = worker->tid;
+  worker_client.next = NULL;
+  worker_client.prevp = NULL;
+
+  worker_class = worker_client;
+  worker_obj = worker_client;
+
+  if (l4_msg_label (ctx->msg) == HURD_CAP_MSG_LABEL_GET_ROOT)
+    {
+      /* This is the "get the root capability" RPC.  FIXME: Needs to
+	 be implemented.  */
+      return ENOSYS;
+    }
+
+  /* Every normal RPC must have at least one untyped word, which
+     contains the client and capability ID.  Otherwise the message is
+     malformed, and thus ignored.  */
+  if (l4_untyped_words (l4_msg_msg_tag (ctx->msg)) < 1)
+    return ECAP_NOREPLY;
+  cap = l4_msg_word (ctx->msg, 0);
+
+  err = lookup_client (bucket, hurd_cap_client_id (cap), ctx->sender, &client);
+  if (err)
+    return err;
+
+  /* At this point, CLIENT_ID and CLIENT are valid, and we have one
+     reference for the client entry.  */
+
+  pthread_mutex_lock (&client->lock);
+  while (!err && client->state != _HURD_CAP_STATE_GREEN)
+    {
+      if (client->state == _HURD_CAP_STATE_BLACK)
+	err = ECAP_NOREPLY;
+      else
+	err = hurd_cond_wait (&bucket->cond, &bucket->lock);
+    }
+  if (err)
+    {
+      pthread_mutex_unlock (&client->lock);
+      /* Either the client died, or we have been canceled.  */
+      _hurd_cap_client_release (bucket, client->id);
+      return err;
+    }
+
+  {
+    _hurd_cap_obj_entry_t *entry;
+
+    entry = (_hurd_cap_obj_entry_t *) hurd_table_lookup (&client->caps,
+							 hurd_cap_id (cap));
+    if (!entry)
+      err = ECAP_NOREPLY;
+    else
+      {
+	obj_entry = *entry;
+
+	if (!obj_entry->external_refs)
+	  err = ECAP_NOREPLY;
+	else if (obj_entry->dead)
+	  err = ECAP_DIED;
+	else
+	  {
+	    obj_entry->internal_refs++;
+	    obj = obj_entry->cap_obj;
+	  }
+      }
+  }
+  if (err)
+    {
+      /* Either the capability ID is invalid, or it was revoked.  */
+      pthread_mutex_unlock (&client->lock);
+      _hurd_cap_client_release (bucket, client->id);
+      return err;
+    }
+
+  /* At this point, CAP_ID, OBJ_ENTRY and OBJ are valid.  We have one
+     internal reference for the capability entry.  */
+
+  /* Add ourself to the pending_rpcs list of the client.  */
+  _hurd_cap_list_item_add (&client->pending_rpcs, &worker_client);
+  pthread_mutex_unlock (&client->lock);
+
+  cap_class = obj->cap_class;
+
+  pthread_mutex_lock (&cap_class->lock);
+  /* First, we have to check if the class is inhibited, and if it is,
+     we have to wait until it is uninhibited.  */
+  while (!err && cap_class->state != _HURD_CAP_STATE_GREEN)
+    err = hurd_cond_wait (&cap_class->cond, &cap_class->lock);
+  if (err)
+    {
+      /* Canceled.  */
+      pthread_mutex_unlock (&cap_class->lock);
+      goto client_cleanup;
+    }
+
+  _hurd_cap_list_item_add (&cap_class->pending_rpcs, &worker_class);
+  pthread_mutex_unlock (&cap_class->lock);
+
+
+  pthread_mutex_lock (&obj->lock);
+  /* First, we have to check if the object is inhibited, and if it is,
+     we have to wait until it is uninhibited.  */
+  if (obj->state != _HURD_CAP_STATE_GREEN)
+    {
+      pthread_mutex_unlock (&obj->lock);
+      pthread_mutex_lock (&cap_class->obj_cond_lock);
+      pthread_mutex_lock (&obj->lock);
+      while (!err && obj->state != _HURD_CAP_STATE_GREEN)
+	{
+	  pthread_mutex_unlock (&obj->lock);
+	  err = hurd_cond_wait (&cap_class->obj_cond,
+				&cap_class->obj_cond_lock);
+	  pthread_mutex_lock (&obj->lock);
+	}
+      pthread_mutex_unlock (&cap_class->obj_cond_lock);
+    }
+  if (err)
+    {
+      /* Canceled.  */
+      pthread_mutex_unlock (&obj->lock);
+      goto class_cleanup;
+    }
+
+  /* Now check if the client still has the capability, or if it was
+     revoked.  */
+  pthread_mutex_lock (&client->lock);
+  if (obj_entry->dead)
+    err = ECAP_DIED;
+  pthread_mutex_unlock (&client->lock);
+  if (err)
+    {
+      /* The capability was revoked in the meantime.  */
+      pthread_mutex_unlock (&obj->lock);
+      goto class_cleanup;
+    }
+
+  _hurd_cap_list_item_add (&obj->pending_rpcs, &worker_obj);
+  pthread_mutex_unlock (&obj->lock);
+
+
+  /* At this point, we have looked up the capability, acquired an
+     internal reference for its entry in the client table (which
+     implicitely keeps a reference acquired for the object itself),
+     acquired a reference for the capability client entry in the
+     bucket, and have added an item to the pending_rpcs lists in the
+     client, class and object.  With all this, we can finally start to
+     process the message for real.  */
+
+  /* FIXME: Call the internal demuxer here, for things like reference
+     counter modification, cap passing etc.  */
+
+  /* Invoke the class-specific demuxer.  */
+  ctx->client = client;
+  ctx->obj = obj;
+  (*cap_class->demuxer) (ctx);
+
+  /* Clean up.  */
+  pthread_mutex_lock (&obj->lock);
+  _hurd_cap_list_item_remove (&worker_obj);
+  _hurd_cap_obj_cond_check (obj);
+  pthread_mutex_unlock (&obj->lock);
+
+ class_cleanup:
+  pthread_mutex_lock (&cap_class->lock);
+  _hurd_cap_list_item_remove (&worker_class);
+  _hurd_cap_class_cond_check (cap_class);
+  pthread_mutex_unlock (&cap_class->lock);
+
+ client_cleanup:
+  pthread_mutex_lock (&client->lock);
+  _hurd_cap_list_item_remove (&worker_client);
+  _hurd_cap_client_cond_check (bucket, client);
+
+  /* You are not allowed to revoke a capability while there are
+     pending RPCs on it.  This is the reason why we know that there
+     must be at least one extra internal reference.  FIXME: For
+     cleanliness, this could still call some inline function that does
+     the decrement.  The assert can be a hint to the compiler to
+     optimize the inline function expansion anyway.  */
+  assert (!obj_entry->dead);
+  assert (obj_entry->internal_refs > 1);
+  obj_entry->internal_refs--;
+  pthread_mutex_unlock (&client->lock);
+
+  _hurd_cap_client_release (bucket, client->id);
+
+  return err;
+}
+
+
+/* A worker thread for RPC processing.  The behaviour of this function
+   is tightly integrated with the behaviour of the manager thread.  */
+static void *
+manage_mt_worker (void *arg)
+{
+  struct worker_info *info = (struct worker_info *) arg;
+  hurd_cap_bucket_t bucket = info->bucket;
+  struct _hurd_cap_list_item worker_item;
+  _hurd_cap_list_item_t worker = &worker_item;
+  l4_thread_id_t manager = info->manager_tid;
+  l4_time_t timeout = info->timeout;
+  l4_thread_id_t from;
+  l4_msg_tag_t msg_tag;
+
+  /* Prepare the worker queue item.  As we are always the current
+     worker thread when we are started up, we do not add ourselves to
+     the free list.  */
+  worker->thread = pthread_self ();
+  worker->tid = l4_myself ();
+  worker->next = NULL;
+  worker->prevp = NULL;
+
+  /* When we are started up, we are supposed to listen as soon as
+     possible to the next incoming message.  The first time, we do
+     this without a timeout.  */
+  from = manager;
+  msg_tag = l4_wait (&from);
+
+  while (1)
+    {
+      if (__builtin_expect (l4_ipc_failed (msg_tag), 0))
+	{
+	  /* Slow path.  */
+
+	  l4_word_t err_code = l4_error_code ();
+	  l4_word_t ipc_err = (err_code >> 1) & 0x7;
+	  unsigned int current_worker_is_us;
+
+	  if (ipc_err == L4_IPC_CANCELED || ipc_err == L4_IPC_ABORTED)
+	    /* We have been canceled for shutdown.  */
+	    break;
+
+	  /* The only other error that can happen is a timeout waiting
+	     for the message.  */
+	  assert (ipc_err == L4_IPC_TIMEOUT);
+
+	  pthread_mutex_lock (&bucket->lock);
+	  /* If we are not on the free queue, we are the current worker.  */
+	  if (worker->prevp == NULL)
+	    current_worker_is_us = 1;
+	  else
+	    current_worker_is_us = 0;
+	  pthread_mutex_unlock (&bucket->lock);
+
+	  /* If we are not the current worker, then we can just exit
+	     now because of our timeout.  */
+	  if (!current_worker_is_us)
+	    break;
+
+	  /* If we are the current worker, we should wait here for the
+	     next message without a timeout.  */
+	  from = manager;
+	  msg_tag = l4_wait (&from);
+	  /* From here, we will loop all over to the beginning of the
+	     while(1) block.  */
+	}
+      else
+	{
+	  /* Fast path.  Process the RPC.  */
+	  error_t err = 0;
+	  struct hurd_cap_rpc_context ctx;
+	  bool inhibited = false;
+
+	  /* IMPORTANT NOTE: The manager thread is blocked until we
+	     reply a message with a label MSG_ACCEPTED or
+	     MSG_REJECTED.  We are supposed to return such a message
+	     as quickly as possible.  In the accepted case, we should
+	     then process the message, while in the rejected case we
+	     should rapidly go into the next receive.  */
+
+	  /* Before we can work on the message, we need to copy it.
+	     This is because the MRs holding the message might be
+	     overridden by the pthread implementation or other
+	     function calls we make.  In particular,
+	     pthread_mutex_lock is can mangle the message buffer.  */
+	  l4_msg_store (msg_tag, ctx.msg);
+
+	  assert (l4_ipc_propagated ());
+	  assert (l4_thread_is_equal (l4_actual_sender (), manager));
+
+	  pthread_mutex_lock (&bucket->lock);
+	  /* We process cancellation messages irregardless of the
+	     bucket state.  */
+	  if (l4_msg_label (ctx.msg) == HURD_CAP_MSG_LABEL_CANCEL)
+	    {
+	      if (l4_untyped_words (l4_msg_msg_tag (ctx.msg)) == 1)
+		{
+		  l4_thread_id_t tid = l4_msg_word (ctx.msg, 0);
+		  
+		  /* First verify access.  Threads are only allowed to
+		     cancel RPCs from other threads in the task.  */
+		  if (hurd_task_id_from_thread_id (tid)
+		      == hurd_task_id_from_thread_id (from))
+		    {
+		      /* We allow cancel requests even if normal RPCs
+			 are inhibited.  */
+		      _hurd_cap_list_item_t pending_worker;
+		      
+		      pending_worker = hurd_ihash_find (&bucket->senders,
+							tid);
+		      if (!pending_worker)
+			reply_err (from, ESRCH);
+		      else
+			{
+			  /* Found it.  Cancel it.  */
+			  pthread_cancel (pending_worker->thread);
+			  /* Reply success.  */
+			  reply_err (from, 0);
+			}
+		    }
+		}
+	      /* Set the error variable so that we return to the
+		 manager immediately.  */
+	      err = ECAP_NOREPLY;
+	    }
+	  else
+	    {
+	      /* Normal RPCs.  */
+	      if (bucket->state == _HURD_CAP_STATE_BLACK)
+		{
+		  /* The bucket operations have been ended, and the
+		     manager has already been canceled.  We know that
+		     the BUCKET->senders hash is empty, so we can
+		     quickly process the message.  */
+
+		  /* This is a normal RPC.  We cancel it immediately.  */
+		  reply_err (from, ECANCELED);
+
+		  /* Now set ERR to any error, so we return to the
+		     manager.  */
+		  err = ECAP_NOREPLY;	/* Doesn't matter which error.  */
+		}
+	      else
+		{
+		  if (bucket->state != _HURD_CAP_STATE_GREEN)
+		    {
+		      /* If we are inhibited, we will have to wait
+			 until we are uninhibited.  */
+		      inhibited = true;
+		    }
+
+		  /* FIXME: This is inefficient.  ihash should support
+		     an "add if not there" function.  */
+		  if (hurd_ihash_find (&bucket->senders, from))
+		    err = EBUSY;
+		  else
+		    {
+		      /* FIXME: We know intimately that pthread_self is not
+			 _HURD_IHASH_EMPTY or _HURD_IHASH_DELETED.  */
+		      err = hurd_ihash_add (&bucket->senders, from, worker);
+		    }
+		}
+	    }
+	    
+	  if (err)
+	    {
+	      pthread_mutex_unlock (&bucket->lock);
+
+	      /* Either we already processed the message above, or
+		 this user thread is currently in an RPC.  We don't
+		 allow asynchronous operation for security reason
+		 (preventing DoS attacks).  Silently drop the
+		 message.  */
+	      msg_tag = l4_niltag;
+	      l4_set_msg_label (&msg_tag, _HURD_CAP_MSG_WORKER_REJECTED);
+	      l4_load_mr (0, msg_tag);
+
+	      /* Reply to the manager that we don't accept the message
+		 and wait for the next message without a timeout
+		 (because now we know we are the current worker).  */
+	      from = manager;
+	      msg_tag = l4_lreply_wait (manager, &from);
+
+	      /* From here, we will loop all over to the beginning of
+		 the while(1) block.  */
+	    }
+	  else
+	    {
+	      _hurd_cap_list_item_add (inhibited ? &bucket->waiting_rpcs
+				       : &bucket->pending_rpcs, worker);
+	      pthread_mutex_unlock (&bucket->lock);
+
+	      msg_tag = l4_niltag;
+	      l4_set_msg_label (&msg_tag, _HURD_CAP_MSG_WORKER_ACCEPTED);
+	      l4_load_mr (0, msg_tag);
+	      msg_tag = l4_reply (manager);
+	      assert (l4_ipc_succeeded (msg_tag));
+
+	      /* Now we are "detached" from the manager in the sense
+		 that we are not the current worker thread
+		 anymore.  */
+
+	      if (inhibited)
+		{
+		  pthread_mutex_lock (&bucket->lock);
+		  while (!err && bucket->state != _HURD_CAP_STATE_GREEN
+			 && bucket->state != _HURD_CAP_STATE_BLACK)
+		    err = hurd_cond_wait (&bucket->cond, &bucket->lock);
+		  if (!err)
+		    {
+		      if (bucket->state == _HURD_CAP_STATE_BLACK)
+			err = ECANCELED;
+		      else 
+			{
+			  /* State is _HURD_CAP_STATE_GREEN.  Move
+			     ourselves to the pending RPC list.  */
+			  _hurd_cap_list_item_remove (worker);
+			  _hurd_cap_list_item_add (&bucket->pending_rpcs,
+						   worker);
+			}
+		    }
+		  pthread_mutex_unlock (&bucket->lock);
+		}
+
+	      if (!err)
+		{
+		  /* Process the message.  */
+		  ctx.sender = hurd_task_id_from_thread_id (from);
+		  ctx.bucket = bucket;
+		  ctx.from = from;
+		  err = manage_demuxer (&ctx, worker);
+		}
+
+	      /* Post-processing.  */
+
+	      pthread_mutex_lock (&bucket->lock);
+	      /* We have to add ourselves to the free list before (or
+		 at the same time) as removing the client from the
+		 pending hash, and before replying to the RPC (if we
+		 reply in the worker thread at all).  The former is
+		 necessary to make sure that no new thread is created
+		 in the race that would otherwise exist, namely after
+		 replying and before adding ourself to the free list.
+		 The latter is required because a client that
+		 immediately follows up with a new message of course
+		 can expect that to work properly.  */
+
+	      if (bucket->is_manager_waiting && !bucket->free_worker)
+		{
+		  /* The manager is starving for worker threads.  */
+		  pthread_cond_broadcast (&bucket->cond);
+		}
+
+	      /* Remove from pending_rpcs (or waiting_rpcs) list.  */
+	      _hurd_cap_list_item_remove (worker);
+	      /* The last waiting RPC may have to signal the manager.  */
+	      if (inhibited && bucket->state == _HURD_CAP_STATE_BLACK
+		  && !bucket->waiting_rpcs)
+		pthread_cond_broadcast (&bucket->cond);
+	      _hurd_cap_list_item_add (&bucket->free_worker, worker);
+
+	      _hurd_cap_bucket_cond_check (bucket);
+
+	      /* Now that we are back on the free list it is safe to
+		 let in the next RPC by this thread.  */
+	      hurd_ihash_locp_remove (&bucket->senders, worker->locp);
+
+	      /* FIXME: Reap the cancellation flag here.  If it was
+		 set, we have been effectively unblocked now.  From
+		 now on, canceling us means something different than
+		 cancelling a pending RPC (it means terminating the
+		 worker thread).  */
+
+	      pthread_mutex_unlock (&bucket->lock);
+
+	      /* Finally, return the reply message, if appropriate.  */
+	      if (__builtin_expect (err != ECAP_NOREPLY, 1))
+		{
+		  if (__builtin_expect (err, 0))
+		    reply_err (from, err);
+		  else
+		    {
+		      l4_msg_load (ctx.msg);
+		      l4_reply (from);
+		    }
+		}
+
+	      /* Now listen for the next message, with a timeout.  */
+	      from = manager;
+	      msg_tag = l4_wait_timeout (timeout, &from);
+
+	      /* From here, we will loop all over to the beginning of
+		 the while(1) block.  */
+	    }
+	}
+    }
+
+  /* At this point, we have been canceled while being on the free
+     list, so we should go away.  */
+
+  pthread_mutex_lock (&bucket->lock);
+  if (_hurd_cap_list_item_dequeued (worker))
+    {
+      /* We are the current worker thread.  We are the last worker
+	 thread the manager thread will cancel.  */
+      pthread_cond_broadcast (&bucket->cond);
+    }
+  else
+    {
+      _hurd_cap_list_item_remove (worker);
+      if (bucket->is_manager_waiting && !bucket->free_worker)
+	{
+	  /* The manager is shutting down.  We are the last free
+	     worker (except for the current worker thread) to be
+	     canceled.  */
+	  pthread_cond_broadcast (&bucket->cond);
+	}
+    }
+  pthread_mutex_unlock (&bucket->lock);
+
+  return NULL;
+}
+
+
+/* Return the next free worker thread.  If no free worker thread is
+   available, create a new one.  If that fails, block until one
+   becomes free.  If we are interrupted while blocking, return
+   l4_nilthread.  */
+static l4_thread_id_t
+manage_mt_get_next_worker (struct worker_info *info, pthread_t *worker_thread)
+{
+  hurd_cap_bucket_t bucket = info->bucket;
+  l4_thread_id_t worker = l4_nilthread;
+  _hurd_cap_list_item_t worker_item;
+
+  pthread_mutex_lock (&bucket->lock);
+
+  if (__builtin_expect (bucket->free_worker == NULL, 0))
+    {
+      /* Slow path.  Create a new thread and use that.  */
+      error_t err;
+
+      pthread_mutex_unlock (&bucket->lock);
+      worker = pthread_pool_get_np ();
+      if (worker == l4_nilthread)
+	err = EAGAIN;
+      else
+	err = pthread_create_from_l4_tid_np (worker_thread, NULL,
+					     worker, manage_mt_worker, info);
+      
+      if (!err)
+	{
+	  pthread_detach (*worker_thread);
+	  return worker;
+	}
+      else
+	{
+	  pthread_mutex_lock (&bucket->lock);
+	  if (!bucket->free_worker)
+	    {
+	      /* Creating a new thread failed.  As a last resort, put
+		 ourself to sleep until we are woken up by the next
+		 free worker.  Hopefully not all workers are blocking
+		 forever.  */
+
+	      /* FIXME: To fix the case where all workers are blocking
+		 forever, cancel one (or more? all?) (random? oldest?)
+		 worker threads.  Usually, that user will restart, but
+		 it will nevertheless allow us to make some (although
+		 slow) process.  */
+
+	      /* The next worker thread that adds itself to the free
+		 list will broadcast the condition.  */
+	      bucket->is_manager_waiting = true;
+	      do
+		err = hurd_cond_wait (&bucket->cond, &bucket->lock);
+	      while (!err && !bucket->free_worker);
+
+	      if (err)
+		{
+		  pthread_mutex_unlock (&bucket->lock);
+		  return l4_nilthread;
+		}
+	    }
+	}
+    }
+
+  /* Fast path.  A worker thread is available.  Remove it from the
+     free list and use it.  */
+  worker_item = bucket->free_worker;
+  _hurd_cap_list_item_remove (worker_item);
+  pthread_mutex_unlock (&bucket->lock);
+
+  *worker_thread = worker_item->thread;
+  return worker_item->tid;
+}
+
+
+/* Start managing RPCs on the bucket BUCKET.  The ROOT capability
    object, which must be unlocked and have one reference throughout
    the whole time this function runs, is used for bootstrapping client
-   connections.  The BOOTSTRAP capability object must be a capability
-   object in one of the classes that have been added to the bucket.
-   The GLOBAL_TIMEOUT parameter specifies the number of seconds until
-   the manager times out (if there are no active users of capability
-   objects in precious classes).  The WORKER_TIMEOUT parameter
-   specifies the number of seconds until each worker thread times out
-   (if there are no RPCs processed by the worker thread).
+   connections.  The GLOBAL_TIMEOUT parameter specifies the number of
+   seconds until the manager times out (if there are no active users).
+   The WORKER_TIMEOUT parameter specifies the number of seconds until
+   each worker thread times out (if there are no RPCs processed by the
+   worker thread).
 
    If this returns ECANCELED, then hurd_cap_bucket_end was called with
-   the force flag being true while there were still active users of
-   capability objects in precious classes.  If this returns without
-   any error, then the timeout expired, or hurd_cap_bucket_end was
-   called without active users of capability objects in precious
-   classes.  */
+   the force flag being true while there were still active users.  If
+   this returns without any error, then the timeout expired, or
+   hurd_cap_bucket_end was called without active users.  */
 error_t
 hurd_cap_bucket_manage_mt (hurd_cap_bucket_t bucket,
-			   hurd_cap_obj_t bootstrap,
-			   unsigned int global_timeout,
-			   unsigned int worker_timeout)
+			   hurd_cap_obj_t root,
+			   unsigned int global_timeout_sec,
+			   unsigned int worker_timeout_sec)
 {
-  /* FIXME: Implement me, of course.  */
-  return ENOSYS;
+  error_t err;
+  l4_time_t global_timeout;
+  pthread_t worker_thread;
+  l4_thread_id_t worker;
+  struct worker_info info;
+  _hurd_cap_list_item_t item;
+
+  global_timeout = (global_timeout_sec == 0) ? L4_NEVER
+    : l4_time_period (UINT64_C (1000000) * global_timeout_sec);
+
+  info.bucket = bucket;
+  info.manager_tid = l4_myself ();
+  info.timeout = (worker_timeout_sec == 0) ? L4_NEVER
+    : l4_time_period (UINT64_C (1000000) * worker_timeout_sec);
+
+  /* We never accept any map or grant items.  FIXME: For now, we also
+     do not accept any string buffer items.  */
+  l4_accept (L4_UNTYPED_WORDS_ACCEPTOR);
+
+  /* Because we do not accept any string items, we do not actually
+     need to set the Xfer timeouts.  But this is what we want to set
+     them to when we eventually do support string items.  */
+  l4_set_xfer_timeouts (l4_timeouts (L4_ZERO_TIME, L4_NEVER));
+
+  /* We create the first worker thread ourselves, to catch any
+     possible error at this stage and bail out properly if needed.  */
+  worker = pthread_pool_get_np ();
+  if (worker == l4_nilthread)
+    return EAGAIN;
+  err = pthread_create_from_l4_tid_np (&worker_thread, NULL,
+				       worker, manage_mt_worker, &info);
+  if (err)
+    return err;
+  pthread_detach (worker_thread);
+
+  pthread_mutex_lock (&bucket->lock);
+  bucket->manager = pthread_self ();
+  bucket->is_managed = true;
+  bucket->is_manager_waiting = false;
+  pthread_mutex_unlock (&bucket->lock);
+  
+  while (1)
+    {
+      l4_thread_id_t from = l4_anythread;
+      l4_msg_tag_t msg_tag;
+
+      /* FIXME: Make sure we have enabled deferred cancellation, and
+	 use an L4 ipc() stub that supports that.  In fact, this must
+	 be true for most of the IPC operations in this file.  */
+      msg_tag = l4_wait_timeout (global_timeout, &from);
+
+      if (__builtin_expect (l4_ipc_failed (msg_tag), 0))
+	{
+	  l4_word_t err_code = l4_error_code ();
+
+	  /* FIXME: We need a macro or inline function for that.  */
+	  l4_word_t ipc_err = (err_code >> 1) & 0x7;
+
+	  /* There are two possible errors, cancellation or timeout.
+	     Any other error indicates a bug in the code.  */
+	  if (ipc_err == L4_IPC_CANCELED || ipc_err == L4_IPC_ABORTED)
+	    {
+	      /* If we are canceled, then this means that our state is
+		 now _HURD_CAP_STATE_BLACK and we should end managing
+		 RPCs even if there are still active users.  */
+
+	      pthread_mutex_lock (&bucket->lock);
+	      assert (bucket->state == _HURD_CAP_STATE_BLACK);
+	      err = ECANCELED;
+	      break;
+	    }
+	  else
+	    {
+	      assert (((err_code >> 1) & 0x7) == L4_IPC_TIMEOUT);
+
+	      pthread_mutex_lock (&bucket->lock);
+	      /* Check if we can time out safely.  */
+	      if (bucket->state == _HURD_CAP_STATE_GREEN
+		  && !bucket->nr_caps && !bucket->pending_rpcs
+		  && !bucket->waiting_rpcs)
+		{
+		  err = 0;
+		  break;
+		}
+	      pthread_mutex_unlock (&bucket->lock);
+	    }
+	}
+      else
+	{
+	  /* Propagate the message to the worker thread.  */
+	  l4_set_propagation (&msg_tag);
+	  l4_set_virtual_sender (from);
+	  l4_set_msg_tag (msg_tag);
+
+	  /* FIXME: Make sure to use a non-cancellable l4_lcall that
+	     does preserve any pending cancellation flag for this
+	     thread.  Alternatively, we can handle cancellation here
+	     (reply ECANCEL to user, and enter shutdown sequence.  */
+	  msg_tag = l4_lcall (worker);
+	  assert (l4_ipc_succeeded (msg_tag));
+
+	  if (__builtin_expect
+	      (l4_label (msg_tag) == _HURD_CAP_MSG_WORKER_ACCEPTED, 1))
+	    {
+	      worker = manage_mt_get_next_worker (&info, &worker_thread);
+	      if (worker == l4_nilthread)
+		{
+		  /* The manage_mt_get_next_worker thread was
+		     canceled.  In this case we have to terminate
+		     ourselves.  */
+		  err = hurd_cap_bucket_inhibit (bucket);
+		  assert (!err);
+		  hurd_cap_bucket_end (bucket, true);
+
+		  pthread_mutex_lock (&bucket->lock);
+		  err = ECANCELED;
+		  break;
+		}
+	    }
+	}
+    }
+
+  /* At this point, bucket->lock is held.  Start the shutdown
+     sequence.  */
+  assert (!bucket->pending_rpcs);
+
+  /* First force all the waiting rpcs onto the free list.  They will
+     have noticed the state change to _HURD_CAP_STATE_BLACK already,
+     we just have to block until the last one wakes us up.  */
+  while (bucket->waiting_rpcs)
+    hurd_cond_wait (&bucket->cond, &bucket->lock);
+      
+  /* Cancel the free workers.  */
+  item = bucket->free_worker;
+  while (item)
+    {
+      pthread_cancel (item->thread);
+      item = item->next;
+    }
+
+  bucket->is_manager_waiting = true;
+
+  while (bucket->free_worker)
+    {
+      /* We ignore cancellations at this point, because we are already
+	 shutting down.  */
+      hurd_cond_wait (&bucket->cond, &bucket->lock);
+    }
+
+  /* Now cancel the current worker, except if we were canceled while
+     trying to get a new one (in which case there is no current
+     worker).  */
+  if (worker != l4_nilthread)
+    {
+      pthread_cancel (worker_thread);
+      while (bucket->free_worker)
+	{
+	  /* We ignore cancellations at this point, because we are already
+	     shutting down.  */
+	  hurd_cond_wait (&bucket->cond, &bucket->lock);
+	}
+    }
+
+  bucket->is_managed = false;
+  pthread_mutex_unlock (&bucket->lock);
+
+  return err;
 }
diff --git a/libhurd-cap-server/cap-server-intern.h b/libhurd-cap-server/cap-server-intern.h
index c58067c..b9dd284 100644
--- a/libhurd-cap-server/cap-server-intern.h
+++ b/libhurd-cap-server/cap-server-intern.h
@@ -34,9 +34,6 @@
 #define hurd_cond_wait pthread_cond_wait
 
 
-/* Forward declaration.  */
-struct hurd_cap_client;
-
 /* This is a simple list item, used to maintain lists of pending RPC
    worker threads in a class, client or capability object.  */
 struct _hurd_cap_list_item
@@ -51,23 +48,59 @@ struct _hurd_cap_list_item
     struct
     {
       /* This location pointer is used for fast removal from the
-	 CAP_CLASS->client_thread hash.  Unused for classes and
-	 capability objects.  */
+	 BUCKET->senders.  */
       hurd_ihash_locp_t locp;
 
       /* The worker thread processing the RPC.  */
       pthread_t thread;
+
+      /* The worker thread L4 thread ID.  */
+      l4_thread_id_t tid;
     };
 
     /* Used for reverse lookup of capability clients using a
        capability object.  */
     struct
     {
-      struct hurd_cap_client *client;
+      _hurd_cap_client_t client;
     };
   };
 };
 
+/* Add the list item ITEM to the list LIST.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_list_item_add (_hurd_cap_list_item_t *list,
+			 _hurd_cap_list_item_t item)
+{
+  if (*list)
+    (*list)->prevp = &item->next;
+  item->prevp = list;
+  item->next = *list;
+  *list = item;
+}
+
+
+/* Remove the list item ITEM from the list.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_list_item_remove (_hurd_cap_list_item_t item)
+{
+  if (item->next)
+    item->next->prevp = item->prevp;
+  *(item->prevp) = item->next;
+  item->prevp = NULL;
+}
+
+
+/* Check if the item ITEM is dequeued or not.  */
+static inline bool
+__attribute__((always_inline))
+_hurd_cap_list_item_dequeued (_hurd_cap_list_item_t item)
+{
+  return item->prevp == NULL;
+}
+
 
 /* Deallocate the capability object OBJ, which must be locked and have
    no more references.  */
@@ -104,7 +137,7 @@ struct _hurd_cap_obj_entry
   hurd_cap_obj_t cap_obj;
 
   /* The index in the capability table.  */
-  hurd_cap_id_t idx;
+  hurd_cap_id_t id;
   
   /* A list item that is used for reverse lookup from the capability
      object to the client.  Protected by the lock of the capability
@@ -137,6 +170,13 @@ typedef struct _hurd_cap_obj_entry *_hurd_cap_obj_entry_t;
 extern struct hurd_slab_space _hurd_cap_obj_entry_space
 	__attribute__((visibility("hidden")));
   
+/* Copy out a capability for the capability OBJ to the user CLIENT.
+   Returns the capability ID (valid only for this user) in *R_ID, or
+   an error.  OBJ must be locked.  */
+error_t _hurd_cap_obj_copy_out (hurd_cap_obj_t obj, hurd_cap_bucket_t bucket,
+				_hurd_cap_client_t client, hurd_cap_id_t *r_id)
+     __attribute__((visibility("hidden")));
+
 
 /* Client connections. */
 
@@ -188,7 +228,6 @@ struct _hurd_cap_client
   /* Reverse lookup from hurd_cap_obj_t to _hurd_cap_obj_entry_t.  */
   struct hurd_ihash caps_reverse;
 };
-typedef struct _hurd_cap_client *_hurd_cap_client_t;
 
 
 /* The global slab space for all capability clients.  */
@@ -201,13 +240,14 @@ extern struct hurd_slab_space _hurd_cap_client_space
    found, create it.  */
 error_t _hurd_cap_client_create (hurd_cap_bucket_t bucket,
 				 hurd_task_id_t task_id,
-				 hurd_cap_id_t *r_idx,
 				 _hurd_cap_client_t *r_client)
      __attribute__((visibility("hidden")));
 
 
 /* Deallocate the connection client CLIENT.  */
-void _hurd_cap_client_dealloc (_hurd_cap_client_t client);
+void _hurd_cap_client_dealloc (hurd_cap_bucket_t bucket,
+			       _hurd_cap_client_t client)
+     __attribute__((visibility("hidden")));
 
 
 /* Release a reference for the client with the ID IDX in class
@@ -218,22 +258,22 @@ void _hurd_cap_client_release (hurd_cap_bucket_t bucket,
 
 
 /* Inhibit all RPCs on the capability client CLIENT (which must not be
-   locked) in the capability class CAP_CLASS.  You _must_ follow up
-   with a hurd_cap_client_resume operation, and hold at least one
-   reference to the object continuously until you did so.  */
+   locked) in the bucket BUCKET.  You _must_ follow up with a
+   hurd_cap_client_resume operation, and hold at least one reference
+   to the object continuously until you did so.  */
 error_t _hurd_cap_client_inhibit (hurd_cap_bucket_t bucket,
 				  _hurd_cap_client_t client)
      __attribute__((visibility("hidden")));
 
 
-/* Resume RPCs on the capability client CLIENT in the class CAP_CLASS
+/* Resume RPCs on the capability client CLIENT in the bucket BUCKET
    and wake-up all waiters.  */
 void _hurd_cap_client_resume (hurd_cap_bucket_t bucket,
 			      _hurd_cap_client_t client)
      __attribute__((visibility("hidden")));
 
 
-/* End RPCs on the capability client CLIENT in the class CAP_CLASS and
+/* End RPCs on the capability client CLIENT in the bucket BUCKET and
    wake-up all waiters.  */
 void _hurd_cap_client_end (hurd_cap_bucket_t bucket,
 			   _hurd_cap_client_t client)
@@ -307,18 +347,41 @@ struct _hurd_cap_bucket
   /* The following members are protected by this lock.  */
   pthread_mutex_t lock;
 
+  /* The manager thread for this capability class.  */
+  pthread_t manager;
+
+  /* True if MANAGER is valid and the bucket is managed.  */
+  bool is_managed;
+
+  /* If this is true, then the manager is waiting for the free worker
+     list to become empty or filled (whatever it is not right now).
+     The first worker thread to notice that the condition is fulfilled
+     now should broadcast the condition.  */
+  bool is_manager_waiting;
+
   /* The state of the bucket.  */
   _hurd_cap_state_t state;
 
-  /* The condition used to wait on state changes.  */
+  /* The condition used to wait on state changes and changes in the
+     worker thread list.  */
   pthread_cond_t cond;
 
   /* The thread waiting for the RPCs to be inhibited.  */
   pthread_t cond_waiter;
 
+  /* The number of capabilities.  If this is not 0, then there are
+     active users.  */
+  unsigned int nr_caps;
+
   /* The pending RPCs in this bucket.  */
   _hurd_cap_list_item_t pending_rpcs;
 
+  /* The waiting RPCs in this bucket.  */
+  _hurd_cap_list_item_t waiting_rpcs;
+
+  /* The free worker threads in this bucket.  */
+  _hurd_cap_list_item_t free_worker;
+
   /* A hash from l4_thread_id_t numbers to the list items in
      PENDING_RPCs.  This is used to limit each client thread to just
      one RPC at one time.  */
@@ -332,4 +395,134 @@ struct _hurd_cap_bucket
 };
 
 
+/* Return true if there are still outstanding RPCs in this bucket
+   BUCKET, and fails if not.  This is only valid if
+   hurd_cap_bucket_inhibit is in progress (ie, if bucket->state is
+   _HURD_CAP_STATE_YELLOW).  BUCKET must be locked.  */
+static inline int
+__attribute__((always_inline))
+_hurd_cap_bucket_cond_busy (hurd_cap_bucket_t bucket)
+{
+  /* We have to remain in the state yellow until there are no pending
+     RPC threads except maybe the waiter.  */
+  return bucket->pending_rpcs
+    && (bucket->pending_rpcs->thread != bucket->cond_waiter
+	|| bucket->pending_rpcs->next);
+}
+
+
+/* Check if the inhibition state of the capability bucket BUCKET has
+   to be changed.  BUCKET must be locked.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_bucket_cond_check (hurd_cap_bucket_t bucket)
+{
+  if (bucket->state == _HURD_CAP_STATE_YELLOW && !_hurd_cap_bucket_cond_busy)
+    {
+      bucket->state == _HURD_CAP_STATE_RED;
+      pthread_cond_broadcast (&bucket->cond);
+    }
+}
+
+
+/* Capability clients.  */
+
+/* Return true if there are still outstanding RPCs in this capability
+   client, and fails if not.  CLIENT must be locked.  This is only
+   valid if hurd_cap_client_inhibit is in progress (ie, if
+   client->state is _HURD_CAP_STATE_YELLOW).  */
+static inline int
+__attribute__((always_inline))
+_hurd_cap_client_cond_busy (_hurd_cap_client_t client)
+{
+  /* We have to remain in the state yellow until there are no pending
+     RPC threads except maybe the waiter.  */
+  return client->pending_rpcs
+    && (client->pending_rpcs->thread != client->cond_waiter
+	|| client->pending_rpcs->next);
+}
+
+
+/* Check if the inhibition state of the capability client CLIENT has
+   to be changed.  CLIENT must be locked.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_client_cond_check (hurd_cap_bucket_t bucket,
+			     _hurd_cap_client_t client)
+{
+  if (client->state == _HURD_CAP_STATE_YELLOW
+      && !_hurd_cap_client_cond_busy (client))
+    {
+      client->state = _HURD_CAP_STATE_RED;
+      pthread_cond_broadcast (&bucket->client_cond);
+    }
+}
+
+
+/* Capability classes.  */
+
+/* Return true if there are still outstanding RPCs in this class, and
+   fails if not.  CAP_CLASS must be locked.  This is only valid if
+   hurd_cap_class_inhibit is in progress (ie, if cap_class->state is
+   _HURD_CAP_STATE_YELLOW).  */
+static inline int
+__attribute__((always_inline))
+_hurd_cap_class_cond_busy (hurd_cap_class_t cap_class)
+{
+  /* We have to remain in the state yellow until there are no pending
+     RPC threads except maybe the waiter.  */
+  return cap_class->pending_rpcs
+    && (cap_class->pending_rpcs->thread != cap_class->cond_waiter
+	|| cap_class->pending_rpcs->next);
+}
+
+
+/* Check if the inhibition state of the capability class CAP_CLASS has
+   to be changed.  CAP_CLASS must be locked.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_class_cond_check (hurd_cap_class_t cap_class)
+{
+  if (cap_class->state == _HURD_CAP_STATE_YELLOW
+      && !_hurd_cap_class_cond_busy (cap_class))
+    {
+      cap_class->state = _HURD_CAP_STATE_RED;
+      pthread_cond_broadcast (&cap_class->cond);
+    }
+}
+
+
+/* Capability objects.  */
+
+/* Return true if there are still outstanding RPCs in this capability
+   object, and fails if not.  OBJ must be locked.  This is only valid
+   if hurd_cap_obj_inhibit is in progress (ie, if cap_obj->state is
+   _HURD_CAP_STATE_YELLOW).  */
+static inline int
+__attribute__((always_inline))
+_hurd_cap_obj_cond_busy (hurd_cap_obj_t obj)
+{
+  /* We have to remain in the state yellow until there are no pending
+     RPC threads except maybe the waiter.  */
+  return obj->pending_rpcs
+    && (obj->pending_rpcs->thread != obj->cond_waiter
+	|| obj->pending_rpcs->next);
+}
+
+
+/* Check if the inhibition state of the capability class CAP_CLASS has
+   to be changed.  CAP_CLASS must be locked.  */
+static inline void
+__attribute__((always_inline))
+_hurd_cap_obj_cond_check (hurd_cap_obj_t obj)
+{
+  if (obj->state == _HURD_CAP_STATE_YELLOW
+      && !_hurd_cap_obj_cond_busy (obj))
+    {
+      obj->state = _HURD_CAP_STATE_RED;
+      pthread_cond_broadcast (&obj->cap_class->cond);
+    }
+}
+
+
 #endif	/* _HURD_CAP_SERVER_INTERN_H */
diff --git a/libhurd-cap-server/cap-server.h b/libhurd-cap-server/cap-server.h
index ad0d643..afc19f8 100644
--- a/libhurd-cap-server/cap-server.h
+++ b/libhurd-cap-server/cap-server.h
@@ -27,6 +27,7 @@
 #include <pthread.h>
 
 #include <hurd/slab.h>
+#include <hurd/types.h>
 
 
 /* Internal declarations.  */
@@ -50,7 +51,11 @@ _hurd_cap_state_t;
 
 /* Public interface.  */
 
-/* Forward declaration.  */
+/* Forward declarations.  */
+struct _hurd_cap_bucket;
+typedef struct _hurd_cap_bucket *hurd_cap_bucket_t;
+struct _hurd_cap_client;
+typedef struct _hurd_cap_client *_hurd_cap_client_t;
 struct hurd_cap_class;
 typedef struct hurd_cap_class *hurd_cap_class_t;
 struct hurd_cap_obj;
@@ -67,11 +72,39 @@ typedef void (*hurd_cap_obj_destroy_t) (hurd_cap_class_t cap_class,
 					hurd_cap_obj_t obj);
 
 
-/* FIXME: Whatever this type should be.  */
-struct hurd_cap_rpc_context;
-typedef void (*hurd_cap_class_demux_t) (struct hurd_cap_rpc_context *ctx);
+/* The RPC context contains various information for the RPC handler
+   and the support functions.  */
+struct hurd_cap_rpc_context
+{
+  /* Public members.  */
+
+  /* The task which contained the sender of the message.  */
+  hurd_task_id_t sender;
+
+  /* The bucket through which the message was received.  */
+  hurd_cap_bucket_t bucket;
 
+  /* The capability object on which the RPC was invoked.  */
+  hurd_cap_obj_t obj;
 
+
+  /* Private members.  */
+
+  /* The sender of the message.  */
+  l4_thread_id_t from;
+
+  /* The client corresponding to FROM.  */
+  _hurd_cap_client_t client;
+
+  /* The message.  */
+  l4_msg_t msg;
+};
+typedef struct hurd_cap_rpc_context *hurd_cap_rpc_context_t;
+
+/* FIXME: Add documentation.  */
+typedef error_t (*hurd_cap_class_demuxer_t) (hurd_cap_rpc_context_t ctx);
+
+
 /* A capability class is a group of capability objects of the same
    type.  */
 struct hurd_cap_class
@@ -152,7 +185,7 @@ struct hurd_cap_class
   /* The class management.  */
 
   /* The demuxer for this class.  */
-  hurd_cap_class_demux_t demuxer;
+  hurd_cap_class_demuxer_t demuxer;
 
   /* The lock protecting all the following members.  */
   pthread_mutex_t lock;
@@ -247,7 +280,7 @@ error_t hurd_cap_class_create (size_t size, size_t alignment,
 			       hurd_cap_obj_alloc_t obj_alloc,
 			       hurd_cap_obj_reinit_t obj_reinit,
 			       hurd_cap_obj_destroy_t obj_destroy,
-			       hurd_cap_class_demux_t demuxer,
+			       hurd_cap_class_demuxer_t demuxer,
 			       hurd_cap_class_t *r_class);
 
 
@@ -267,7 +300,7 @@ error_t hurd_cap_class_init (hurd_cap_class_t cap_class,
 			     hurd_cap_obj_alloc_t obj_alloc,
 			     hurd_cap_obj_reinit_t obj_reinit,
 			     hurd_cap_obj_destroy_t obj_destroy,
-			     hurd_cap_class_demux_t demuxer);
+			     hurd_cap_class_demuxer_t demuxer);
 
 
 /* Destroy the capability class CAP_CLASS and release all associated
@@ -361,10 +394,6 @@ void hurd_cap_obj_resume (hurd_cap_obj_t obj);
 /* Buckets are a set of capabilities, on which RPCs are managed
    collectively.  */
 
-struct _hurd_cap_bucket;
-typedef struct _hurd_cap_bucket *hurd_cap_bucket_t;
-
-
 /* Create a new bucket and return it in R_BUCKET.  */
 error_t hurd_cap_bucket_create (hurd_cap_bucket_t *r_bucket);
 
@@ -373,25 +402,30 @@ error_t hurd_cap_bucket_create (hurd_cap_bucket_t *r_bucket);
 void hurd_cap_bucket_free (hurd_cap_bucket_t bucket);
 
 
-/* Start managing RPCs on the bucket BUCKET.  The BOOTSTRAP capability
+/* Copy out a capability for the capability OBJ to the client with the
+   task ID TASK_ID.  Returns the capability (valid only for this user)
+   in *R_CAP, or an error.  It is not safe to call this from outside
+   an RPC on OBJ while the manager is running.  */
+error_t hurd_cap_bucket_inject (hurd_cap_bucket_t bucket, hurd_cap_obj_t obj,
+				hurd_task_id_t task_id, hurd_cap_t *r_cap);
+
+
+/* Start managing RPCs on the bucket BUCKET.  The ROOT capability
    object, which must be unlocked and have one reference throughout
    the whole time this function runs, is used for bootstrapping client
-   connections.  The BOOTSTRAP capability object must be a capability
-   object in one of the classes that have been added to the bucket.
-   The GLOBAL_TIMEOUT parameter specifies the number of seconds until
-   the manager times out (if there are no active users of capability
-   objects in precious classes).  The WORKER_TIMEOUT parameter
-   specifies the number of seconds until each worker thread times out
-   (if there are no RPCs processed by the worker thread).
+   connections.  The GLOBAL_TIMEOUT parameter specifies the number of
+   seconds until the manager times out (if there are no active users
+   of capability objects in precious classes).  The WORKER_TIMEOUT
+   parameter specifies the number of seconds until each worker thread
+   times out (if there are no RPCs processed by the worker thread).
 
    If this returns ECANCELED, then hurd_cap_bucket_end was called with
-   the force flag being true while there were still active users of
-   capability objects in precious classes.  If this returns without
-   any error, then the timeout expired, or hurd_cap_bucket_end was
-   called without active users of capability objects in precious
-   classes.  */
+   the force flag being true while there were still active users.  If
+   this returns without any error, then the timeout expired, or
+   hurd_cap_bucket_end was called without active users of capability
+   objects in precious classes.  */
 error_t hurd_cap_bucket_manage_mt (hurd_cap_bucket_t bucket,
-				   hurd_cap_obj_t bootstrap,
+				   hurd_cap_obj_t root,
 				   unsigned int global_timeout,
 				   unsigned int worker_timeout);
 
@@ -407,9 +441,9 @@ void hurd_cap_bucket_resume (hurd_cap_bucket_t bucket);
 
 
 /* Exit from the server loop of the managed capability bucket BUCKET.
-   This will only succeed if there are no active users of capability
-   objects in precious classes, or if the FORCE flag is set (otherwise
-   it will fail with EBUSY).  The bucket must be inhibited.  */
+   This will only succeed if there are no active users, or if the
+   FORCE flag is set (otherwise it will fail with EBUSY).  The bucket
+   must be inhibited.  */
 error_t hurd_cap_bucket_end (hurd_cap_bucket_t bucket, bool force);
 
 
diff --git a/libhurd-cap-server/class-create.c b/libhurd-cap-server/class-create.c
index e0c7865..2039f59 100644
--- a/libhurd-cap-server/class-create.c
+++ b/libhurd-cap-server/class-create.c
@@ -51,7 +51,7 @@ hurd_cap_class_create (size_t size, size_t alignment,
 		       hurd_cap_obj_alloc_t obj_alloc,
 		       hurd_cap_obj_reinit_t obj_reinit,
 		       hurd_cap_obj_destroy_t obj_destroy,
-		       hurd_cap_class_demux_t demuxer,
+		       hurd_cap_class_demuxer_t demuxer,
 		       hurd_cap_class_t *r_class)
 {
   error_t err;
@@ -68,5 +68,6 @@ hurd_cap_class_create (size_t size, size_t alignment,
       return err;
     }
 
+  *r_class = cap_class;
   return 0;
 }
diff --git a/libhurd-cap-server/class-inhibit.c b/libhurd-cap-server/class-inhibit.c
index ec8c842..ea2a663 100644
--- a/libhurd-cap-server/class-inhibit.c
+++ b/libhurd-cap-server/class-inhibit.c
@@ -30,23 +30,6 @@
 #include "cap-server-intern.h"
 
 
-/* Return true if there are still outstanding RPCs in this class, and
-   fails if not.  This is only valid if hurd_cap_class_inhibit is in
-   progress (ie, if cap_class->state is _HURD_CAP_STATE_YELLOW).
-   FIXME: We will need this in the RPC worker thread code, where the
-   last worker will get false as return value and then has to change
-   the state to RED and signal (broadcast?) the condition.  */
-static inline int
-_hurd_cap_class_cond_busy (hurd_cap_class_t cap_class)
-{
-  /* We have to remain in the state yellow until there are no pending
-     RPC threads except maybe the waiter.  */
-  return cap_class->pending_rpcs
-    && (cap_class->pending_rpcs->thread != cap_class->cond_waiter
-	|| cap_class->pending_rpcs->next);
-}
-
-
 /* Inhibit all RPCs on the capability class CAP_CLASS (which must not
    be locked).  You _must_ follow up with a hurd_cap_class_resume
    operation, and hold at least one reference to the object
diff --git a/libhurd-cap-server/class-init.c b/libhurd-cap-server/class-init.c
index 28976f1..e4468d3 100644
--- a/libhurd-cap-server/class-init.c
+++ b/libhurd-cap-server/class-init.c
@@ -93,7 +93,7 @@ hurd_cap_class_init (hurd_cap_class_t cap_class,
 		     hurd_cap_obj_alloc_t obj_alloc,
 		     hurd_cap_obj_reinit_t obj_reinit,
 		     hurd_cap_obj_destroy_t obj_destroy,
-		     hurd_cap_class_demux_t demuxer)
+		     hurd_cap_class_demuxer_t demuxer)
 {
   error_t err;
 
diff --git a/libhurd-cap-server/client-create.c b/libhurd-cap-server/client-create.c
index bb3a897..fc4e30c 100644
--- a/libhurd-cap-server/client-create.c
+++ b/libhurd-cap-server/client-create.c
@@ -135,7 +135,6 @@ error_t
 __attribute__((visibility("hidden")))
 _hurd_cap_client_create (hurd_cap_bucket_t bucket,
 			 hurd_task_id_t task_id,
-			 hurd_cap_id_t *r_idx,
 			 _hurd_cap_client_t *r_client)
 {
   error_t err = 0;
@@ -155,7 +154,6 @@ _hurd_cap_client_create (hurd_cap_bucket_t bucket,
       else
 	{
 	  entry->refs++;
-	  *r_idx = client->id;
 	  *r_client = entry->client;
 	}
       pthread_mutex_unlock (&bucket->lock);
@@ -189,11 +187,10 @@ _hurd_cap_client_create (hurd_cap_bucket_t bucket,
 	{
 	  /* Somebody else was indeed faster.  Use the existing entry.  */
 	  entry->refs++;
-	  *r_idx = client->id;
 	  *r_client = entry->client;
 	}
       pthread_mutex_unlock (&bucket->lock);
-      _hurd_cap_client_dealloc (*r_client);
+      _hurd_cap_client_dealloc (bucket, *r_client);
       return err;
     }
 
@@ -217,13 +214,13 @@ _hurd_cap_client_create (hurd_cap_bucket_t bucket,
       pthread_mutex_unlock (&bucket->lock);
       hurd_task_death_notify_resume ();
       
-      _hurd_cap_client_dealloc (client);
+      _hurd_cap_client_dealloc (bucket, client);
       return err;
     }
   pthread_mutex_unlock (&bucket->lock);
   hurd_task_death_notify_resume ();
 
-  *r_idx = client->id;
+  *r_client = client;
 
   return 0;
 }
diff --git a/libhurd-cap-server/client-inhibit.c b/libhurd-cap-server/client-inhibit.c
index b3ce6b9..f2a02d8 100644
--- a/libhurd-cap-server/client-inhibit.c
+++ b/libhurd-cap-server/client-inhibit.c
@@ -30,24 +30,6 @@
 #include "cap-server-intern.h"
 
 
-/* Return true if there are still outstanding RPCs in this capability
-   client, and fails if not.  This is only valid if
-   hurd_cap_client_inhibit is in progress (ie, if client->state is
-   _HURD_CAP_STATE_YELLOW).  FIXME: We will need this in the RPC
-   worker thread code, where the last worker will get false as return
-   value and then has to change the state to RED and broadcast the
-   condition.  */
-static inline int
-_hurd_cap_client_cond_busy (_hurd_cap_client_t client)
-{
-  /* We have to remain in the state yellow until there are no pending
-     RPC threads except maybe the waiter.  */
-  return client->pending_rpcs
-    && (client->pending_rpcs->thread != client->cond_waiter
-	|| client->pending_rpcs->next);
-}
-
-
 /* Inhibit all RPCs on the capability client CLIENT (which must not be
    locked) in the capability bucket BUCKET.  You _must_ follow up
    with a hurd_cap_client_resume operation, and hold at least one
diff --git a/libhurd-cap-server/client-release.c b/libhurd-cap-server/client-release.c
index 8b6cd7e..5e3c6af 100644
--- a/libhurd-cap-server/client-release.c
+++ b/libhurd-cap-server/client-release.c
@@ -32,10 +32,11 @@
 
 /* Deallocate the connection client CLIENT.  */
 void
-_hurd_cap_client_dealloc (_hurd_cap_client_t client)
+_hurd_cap_client_dealloc (hurd_cap_bucket_t bucket, _hurd_cap_client_t client)
 {
   unsigned int done;
   unsigned int current_idx;
+  unsigned int nr_caps = 0;
 
   /* This function is only invoked if the reference count for the
      client entry in the client table of the class drops to 0, and
@@ -81,6 +82,8 @@ _hurd_cap_client_dealloc (_hurd_cap_client_t client)
 	 reference would have been released before we get here.  */
       assert (entry->external_refs);
 
+      nr_caps++;
+
       /* The number of internal references is either one or zero.  If
 	 it is one, then the capability is not revoked yet, so we have
 	 to do it.  If it is zero, then the capability is revoked
@@ -155,6 +158,10 @@ _hurd_cap_client_dealloc (_hurd_cap_client_t client)
      enforce a per-client quota.  */
   pthread_mutex_unlock (&client->lock);
 
+  pthread_mutex_lock (&bucket->lock);
+  bucket->nr_caps -= nr_caps;
+  pthread_mutex_unlock (&bucket->lock);
+
   hurd_slab_dealloc (&_hurd_cap_client_space, client);
 }
 
@@ -184,6 +191,6 @@ _hurd_cap_client_release (hurd_cap_bucket_t bucket, hurd_cap_client_id_t idx)
       hurd_ihash_locp_remove (&bucket->clients_reverse, client->locp);
 
       pthread_mutex_unlock (&bucket->lock);
-      _hurd_cap_client_dealloc (client);
+      _hurd_cap_client_dealloc (bucket, client);
     }
 }
diff --git a/libhurd-cap-server/obj-copy-out.c b/libhurd-cap-server/obj-copy-out.c
new file mode 100644
index 0000000..1edfe7d
--- /dev/null
+++ b/libhurd-cap-server/obj-copy-out.c
@@ -0,0 +1,118 @@
+/* obj-copy-out.c - Copy out a capability to a client.
+   Copyright (C) 2004 Free Software Foundation, Inc.
+   Written by Marcus Brinkmann <marcus@gnu.org>
+
+   This file is part of the GNU Hurd.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2.1 of the License, or (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this program; if not, write to the Free
+   Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
+   02111-1307 USA.  */
+
+#if HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <errno.h>
+#include <pthread.h>
+#include <stdlib.h>
+
+#include "cap-server-intern.h"
+
+
+/* Copy out a capability for the capability OBJ to the user CLIENT.
+   Returns the capability ID (valid only for this user) in *R_ID, or
+   an error.  OBJ must be locked.  Note: No internal reference for
+   this capability is allocated for the caller.  */
+error_t
+_hurd_cap_obj_copy_out (hurd_cap_obj_t obj, hurd_cap_bucket_t bucket,
+			_hurd_cap_client_t client, hurd_cap_id_t *r_id)
+{
+  _hurd_cap_obj_entry_t entry;
+
+  pthread_mutex_lock (&client->lock);
+  entry = (_hurd_cap_obj_entry_t) hurd_ihash_find (&client->caps_reverse,
+						   (hurd_ihash_key_t) obj);
+
+  if (entry)
+    {
+      entry->external_refs++;
+      *r_id = entry->id;
+      pthread_mutex_unlock (&client->lock);
+      return 0;
+    }
+  else
+    {
+      _hurd_cap_obj_entry_t entry_check;
+      error_t err;
+
+      pthread_mutex_unlock (&client->lock);
+      err = hurd_slab_alloc (&_hurd_cap_obj_entry_space, (void **) &entry);
+      if (err)
+	return err;
+
+      entry->cap_obj = obj;
+      /* ID is filled in when adding the object to the table.  */
+      /* CLIENT_ITEM is filled after the object has been entered.  */
+      /* DEAD is 0 for initialized objects.  */
+      /* INTERNAL_REFS is 1 for initialized objects.  */
+      /* EXTERNAL_REFS is 1 for initialized objects.  */
+
+      pthread_mutex_lock (&client->lock);
+      entry_check = hurd_ihash_find (&client->caps_reverse,
+				     (hurd_ihash_key_t) obj);
+      if (entry_check)
+	{
+	  /* Somebody else was faster.  */
+	  entry_check->external_refs++;
+	  *r_id = entry_check->id;
+	  pthread_mutex_unlock (&client->lock);
+	  hurd_slab_dealloc (&_hurd_cap_obj_entry_space, entry);
+	  return 0;
+	}
+
+      /* Add the entry to the caps table of the client.  */
+      err = hurd_table_enter (&client->caps, &entry, &entry->id);
+      if (err)
+	{
+	  pthread_mutex_unlock (&client->lock);
+	  hurd_slab_dealloc (&_hurd_cap_obj_entry_space, entry);
+	  return err;
+	}
+      err = hurd_ihash_add (&client->caps_reverse,
+			    (hurd_ihash_key_t) obj, entry);
+      if (err)
+	{
+	  hurd_table_remove (&client->caps, entry->id);
+	  pthread_mutex_unlock (&client->lock);
+	  hurd_slab_dealloc (&_hurd_cap_obj_entry_space, entry);
+	  return err;
+	}
+
+      pthread_mutex_unlock (&client->lock);
+
+      /* Add the object to the list.  */
+      _hurd_cap_list_item_add (&obj->clients, &entry->client_item);
+
+      /* Add a reference for the internal reference of the capability
+	 entry to the capability object.  */
+      obj->refs++;
+
+      /* FIXME: Should probably use spin lock here, or so.  */
+      pthread_mutex_lock (&bucket->lock);
+      bucket->nr_caps++;
+      pthread_mutex_unlock (&bucket->lock);
+
+      return 0;
+    }
+}
diff --git a/libhurd-cap-server/obj-inhibit.c b/libhurd-cap-server/obj-inhibit.c
index 5cee758..4d11e9d 100644
--- a/libhurd-cap-server/obj-inhibit.c
+++ b/libhurd-cap-server/obj-inhibit.c
@@ -30,24 +30,6 @@
 #include "cap-server-intern.h"
 
 
-/* Return true if there are still outstanding RPCs in this capability
-   object, and fails if not.  This is only valid if
-   hurd_cap_obj_inhibit is in progress (ie, if cap_obj->state is
-   _HURD_CAP_STATE_YELLOW).  FIXME: We will need this in the RPC
-   worker thread code, where the last worker will get false as return
-   value and then has to change the state to RED and broadcast the
-   condition.  */
-static inline int
-_hurd_cap_obj_cond_busy (hurd_cap_obj_t obj)
-{
-  /* We have to remain in the state yellow until there are no pending
-     RPC threads except maybe the waiter.  */
-  return obj->pending_rpcs
-    && (obj->pending_rpcs->thread != obj->cond_waiter
-	|| obj->pending_rpcs->next);
-}
-
-
 /* Inhibit all RPCs on the capability object CAP_OBJ (which must not
    be locked).  You _must_ follow up with a hurd_cap_obj_resume
    operation, and hold at least one reference to the object
diff --git a/libhurd-cap-server/task-death.c b/libhurd-cap-server/task-death.c
index 17e68e2..855ed7a 100644
--- a/libhurd-cap-server/task-death.c
+++ b/libhurd-cap-server/task-death.c
@@ -26,7 +26,7 @@
 #include <pthread.h>
 
 #include <hurd/types.h>
-#include <hurd/task-death.h>
+#include "task-death.h"
 
 
 /* A lock that protects the linked list.  It also is held when