ovs/lib/seq.c

/*
 * Copyright (c) 2013, 2014 Nicira, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <config.h>

#include "seq.h"

#include <stdbool.h>

#include "coverage.h"
#include "hash.h"
#include "openvswitch/hmap.h"
#include "latch.h"
#include "openvswitch/list.h"
#include "ovs-thread.h"
#include "openvswitch/poll-loop.h"

COVERAGE_DEFINE(seq_change);

/* A sequence number object. */
struct seq {
    atomic_uint64_t value;
    struct hmap waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
};

/* A thread waiting on a particular seq. */
struct seq_waiter {
    struct hmap_node hmap_node OVS_GUARDED; /* In 'seq->waiters'. */
    struct seq *seq OVS_GUARDED;            /* Seq being waited for. */
    unsigned int ovsthread_id OVS_GUARDED;  /* Key in 'waiters' hmap. */

    struct seq_thread *thread OVS_GUARDED;  /* Thread preparing to wait. */
    struct ovs_list list_node OVS_GUARDED;  /* In 'thread->waiters'. */

    uint64_t value OVS_GUARDED; /* seq->value we're waiting to change. */
};

/* A thread that might be waiting on one or more seqs. */
struct seq_thread {
    struct ovs_list waiters OVS_GUARDED; /* Contains 'struct seq_waiter's. */
    struct latch latch OVS_GUARDED;  /* Wakeup latch for this thread. */
    bool waiting OVS_GUARDED;        /* True if latch_wait() already called. */
};

static struct ovs_mutex seq_mutex = OVS_MUTEX_INITIALIZER;

static uint64_t seq_next OVS_GUARDED_BY(seq_mutex) = 1;

static pthread_key_t seq_thread_key;

static void seq_init(void);
static struct seq_thread *seq_thread_get(void) OVS_REQUIRES(seq_mutex);
static void seq_thread_exit(void *thread_) OVS_EXCLUDED(seq_mutex);
static void seq_thread_woke(struct seq_thread *) OVS_REQUIRES(seq_mutex);
static void seq_waiter_destroy(struct seq_waiter *) OVS_REQUIRES(seq_mutex);
static void seq_wake_waiters(struct seq *) OVS_REQUIRES(seq_mutex);

/* Creates and returns a new 'seq' object. */
struct seq * OVS_EXCLUDED(seq_mutex)
seq_create(void)
{
    uint64_t seq_value;
    struct seq *seq;

    seq_init();

    seq = xmalloc(sizeof *seq);

    COVERAGE_INC(seq_change);

    ovs_mutex_lock(&seq_mutex);
    seq_value = seq_next++;
    atomic_store_relaxed(&seq->value, seq_value);
    hmap_init(&seq->waiters);
    ovs_mutex_unlock(&seq_mutex);

    return seq;
}

/* Destroys 'seq', waking up threads that were waiting on it, if any. */
void
seq_destroy(struct seq *seq)
     OVS_EXCLUDED(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
    seq_wake_waiters(seq);
    hmap_destroy(&seq->waiters);
    free(seq);
    ovs_mutex_unlock(&seq_mutex);
}

int
seq_try_lock(void)
{
    return ovs_mutex_trylock(&seq_mutex);
}

void
seq_lock(void)
    OVS_ACQUIRES(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
}

void
seq_unlock(void)
    OVS_RELEASES(seq_mutex)
{
    ovs_mutex_unlock(&seq_mutex);
}

/* Increments 'seq''s sequence number, waking up any threads that are waiting
 * on 'seq'. */
void
seq_change_protected(struct seq *seq)
    OVS_REQUIRES(seq_mutex)
{
    uint64_t seq_value = seq_next++;

    COVERAGE_INC(seq_change);

    atomic_store_explicit(&seq->value, seq_value, memory_order_release);
    seq_wake_waiters(seq);
}

/* Increments 'seq''s sequence number, waking up any threads that are waiting
 * on 'seq'. */
void
seq_change(struct seq *seq)
    OVS_EXCLUDED(seq_mutex)
{
    ovs_mutex_lock(&seq_mutex);
    seq_change_protected(seq);
    ovs_mutex_unlock(&seq_mutex);
}

/* Returns 'seq''s current sequence number (which could change immediately).
 *
 * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 * when an object changes, even without an ability to lock the object.  See
 * Usage in seq.h for details. */
uint64_t
seq_read(const struct seq *seq)
{
    uint64_t value;

    /* Note that the odd CONST_CAST() is here to keep sparse happy. */
    atomic_read_explicit(&CONST_CAST(struct seq *, seq)->value, &value,
                         memory_order_acquire);
    return value;
}

/* poll_immediate_wake_at() must not be called while holding seq_mutex
 * in order to avoid potential deadlock with time_init() that calls
 * seq_create() if the timeval module is not initialized yet.  */
void poll_immediate_wake_at(const char *where) OVS_EXCLUDED(seq_mutex);

static void
seq_wait__(struct seq *seq, uint64_t value, const char *where)
    OVS_RELEASES(seq_mutex)
{
    unsigned int id = ovsthread_id_self();
    uint32_t hash = hash_int(id, 0);
    struct seq_waiter *waiter;

    HMAP_FOR_EACH_IN_BUCKET (waiter, hmap_node, hash, &seq->waiters) {
        if (waiter->ovsthread_id == id) {
            if (waiter->value != value) {
                /* The current value is different from the value we've already
                 * waited for, */
                ovs_mutex_unlock(&seq_mutex);
                poll_immediate_wake_at(where);
            } else {
                /* Already waiting on 'value', nothing more to do. */
                ovs_mutex_unlock(&seq_mutex);
            }
            return;
        }
    }

    waiter = xmalloc(sizeof *waiter);
    waiter->seq = seq;
    hmap_insert(&seq->waiters, &waiter->hmap_node, hash);
    waiter->ovsthread_id = id;
    waiter->value = value;
    waiter->thread = seq_thread_get();
    ovs_list_push_back(&waiter->thread->waiters, &waiter->list_node);

    if (!waiter->thread->waiting) {
        latch_wait_at(&waiter->thread->latch, where);
        waiter->thread->waiting = true;
    }
    ovs_mutex_unlock(&seq_mutex);
}

/* Causes the following poll_block() to wake up when 'seq''s sequence number
 * changes from 'value'.  (If 'seq''s sequence number isn't 'value', then
 * poll_block() won't block at all.)
 *
 * seq_read() and seq_wait() can be used together to yield a race-free wakeup
 * when an object changes, even without an ability to lock the object.  See
 * Usage in seq.h for details.
 *
 * ('where' is used in debug logging.  Commonly one would use seq_wait() to
 * automatically provide the caller's source file and line number for
 * 'where'.) */
void
seq_wait_at(const struct seq *seq_, uint64_t value, const char *where)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq *seq = CONST_CAST(struct seq *, seq_);

    ovs_mutex_lock(&seq_mutex);
    if (value == seq_read(seq_)) {
        seq_wait__(seq, value, where);
    } else {
        ovs_mutex_unlock(&seq_mutex);
        poll_immediate_wake_at(where);
    }
}

/* Called by poll_block() just before it returns, this function destroys any
 * seq_waiter objects associated with the current thread. */
void
seq_woke(void)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq_thread *thread;

    seq_init();

    thread = pthread_getspecific(seq_thread_key);
    if (thread) {
        ovs_mutex_lock(&seq_mutex);
        seq_thread_woke(thread);
        thread->waiting = false;
        ovs_mutex_unlock(&seq_mutex);
    }
}

static void
seq_init(void)
{
    static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;

    if (ovsthread_once_start(&once)) {
        xpthread_key_create(&seq_thread_key, seq_thread_exit);
        ovsthread_once_done(&once);
    }
}

static struct seq_thread *
seq_thread_get(void)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_thread *thread = pthread_getspecific(seq_thread_key);
    if (!thread) {
        thread = xmalloc(sizeof *thread);
        ovs_list_init(&thread->waiters);
        latch_init(&thread->latch);
        thread->waiting = false;

        xpthread_setspecific(seq_thread_key, thread);
    }
    return thread;
}

static void
seq_thread_exit(void *thread_)
    OVS_EXCLUDED(seq_mutex)
{
    struct seq_thread *thread = thread_;

    ovs_mutex_lock(&seq_mutex);
    seq_thread_woke(thread);
    latch_destroy(&thread->latch);
    free(thread);
    ovs_mutex_unlock(&seq_mutex);
}

static void
seq_thread_woke(struct seq_thread *thread)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_waiter *waiter;

    LIST_FOR_EACH_SAFE (waiter, list_node, &thread->waiters) {
        ovs_assert(waiter->thread == thread);
        seq_waiter_destroy(waiter);
    }
    latch_poll(&thread->latch);
}

static void
seq_waiter_destroy(struct seq_waiter *waiter)
    OVS_REQUIRES(seq_mutex)
{
    hmap_remove(&waiter->seq->waiters, &waiter->hmap_node);
    ovs_list_remove(&waiter->list_node);
    free(waiter);
}

static void
seq_wake_waiters(struct seq *seq)
    OVS_REQUIRES(seq_mutex)
{
    struct seq_waiter *waiter;

    HMAP_FOR_EACH_SAFE (waiter, hmap_node, &seq->waiters) {
        latch_set(&waiter->thread->latch);
        seq_waiter_destroy(waiter);
    }
}