Re: RFC: localcount_hadref() or localcount_trydarin()

[Taylor R Campbell <campbell+netbsd-tech-kern%mumble.net@localhost>]

> Date: Mon, 12 Jun 2017 09:09:14 -0400
> From: Thor Lancelot Simon <tls%panix.com@localhost>
> 
> On Mon, Jun 12, 2017 at 12:51:29PM +0000, Taylor R Campbell wrote:
> > Why don't you just use a global reference count first?  Is the latency
> > and scalability of crypto_newsession and crypto_freesession critical?
> 
> For many workloads, it will be, yes.

OK.  But we still won't really get any benefit of localcount(9) until
the global locks go away in the first place.

Now suppose
(i)   you get rid of those locks;
(ii)  you adapt the opencrypto driver lookup to use pserialize(9);
(iii) you still want to allow a kind of `trydetach' operation that
  returns EBUSY if there's anything in flight, instead of the `detach'
  operation that localcont(9) is currently designed for, which blocks
  until the operations are completed; and
(iv)  psref(9) won't do it (which may be obvious; I just haven't
  thought about using psref(9) for this).

Here's how I would adapt localcount(9) to do this:

1. Unhook the resource so that
   (a) new references cannot be acquired (e.g., attempts to look it up
       will fail with ENOENT), but
   (b) there is still a sentinel in place, so nobody can put another
       resource in the same place, because we will need to be able to
       put the resource back in place if the operation fails.
   You might do this by leaving a pointer to an object in a table, but
   setting a flag indicating that the object cannot be used now.

        struct driver *drv;

        /* detach */
        mutex_enter(&drivers.lock);
        drv = drivers.table[idx];
        if (drv == NULL || drv->flags & DYING)
                goto enoent;
        drv->flags |= DYING;
        ...

        /* lookup */
        struct driver *drv;
        pserialize_read_enter();
        if ((drv = drivers.table[idx]) == NULL)
                goto enoent;
        membar_datadep_consumer();
        if (drv->flags & DYING)
                goto enoent;
        localcount_acquire(&drv->localcount);
        ...

2. Use a new operation localcount_trydrain which does a cross-call to
   count up the references and check whether they sum to zero.  It is
   critical that at this point, new references on any CPU must be
   prohibited, so that if the total count changes after this function
   answers, it can only go down toward zero.

bool
localcount_trydrain(struct localcount *lc)
{
        volatile int64_t total = 0;

        KASSERT(lc->lc_totalp == NULL);

        /* Count up all the references.  */
        xc_wait(xc_broadcast(0, drained_xc, lc, __UNVOLATILE(&total)));

        /* If there were any, stop here with no permanent effects.  */
        if (total != 0)
                return false;

        /*
         * All references have drained.
         *
         * Paranoia: Cause any further use of lc->lc_totalp to crash.
         */
        lc->lc_totalp = (void *)(uintptr_t)1;

        return true;
}

static void
drained_xc(void *cookie0, void *cookie1)
{
        struct localcount *lc = cookie0;
        volatile int64_t *totalp = cookie1;
        int64_t *localp;

        localp = percpu_getref(lc->lc_percpu);
        atomic_add_64(totalp, *localp);
        percpu_putref(lc->lc_percpu);
}

   (This sketch may need to be adapted for 32-bit machines, e.g. to
   use a local kmutex_t instead of 64-bit atomics.  Exercise for the
   reader!)

3. If localcount_trydrain returns false, you must put the resource
   back as it was to allow new references again, and stop here.

        /* detach */
        ...
        if (!localcount_trydrain(&drv->localcount)) {
                drv->flags &= ~DYING;
                goto ebusy;
        }

4. Otherwise, if localcount_trydrain returns true, then not only are
   there no references now, but there can be no new references made
   either because you guaranteed that in (1) and (2).  Now you can
   unhook the sentinel (and perhaps wait for that to be visible on all
   CPUs, e.g., with pserialize_perform, if the sentinel uses allocated
   memory that you need to free) and destroy the original resource.

        /* detach */
        ...
        drivers.table[idx] = NULL;
        pserialize_perform(drivers.psz);
        pool_put(&drivers.pool, drv);

Both localcount_drain and successful localcount_trydrain are final:
the only thing you can, and must, do with a localcount afterward is
localcount_fini.