src/core/sentinel.c

Go to the documentation of this file.

#include <sys/types.h>
#include <sys/stat.h>
#include <pthread.h>
#include <unistd.h>
#include <stddef.h>
#include <misc/result.h>
#include <misc/pool.h>
#include <misc/malloc.h>
#include <misc/debug.h>
#include <misc/mutex.h>
#include <core/config.h>
#include <core/sentinel.h>
#include <core/tx.h>
#include <core/txdesc.h>

#define TXC_SENTINEL_NOOWNER                0x0

struct txc_sentinel_s {
        txc_mutex_t       sentinel_mutex;  
        txc_mutex_t       synch_mutex;     
        int               id;              
        int               refcnt;          
        txc_tx_t          *owner;          
        txc_sentinelmgr_t *manager;        
};

struct txc_sentinel_list_entry_s {
        txc_sentinel_t *sentinel;          
        txc_result_t   status;             
};

struct txc_sentinel_list_s {
        txc_sentinel_list_entry_t *entries;      
        unsigned int              num_entries;   
        unsigned int              size;          
        txc_sentinelmgr_t         *manager;      
};


struct txc_sentinelmgr_s {
        txc_mutex_t mutex;
        txc_pool_t  *pool_sentinel;
};


static inline void sentinel_list_print(txc_sentinel_list_t *sentinel_list, char *heading);
static inline txc_result_t enlist_sentinel(txc_sentinel_list_t *sentinel_list, txc_sentinel_t *sentinel, txc_result_t status);
static inline void backoff(txc_tx_t *txd);


txc_sentinelmgr_t *txc_g_sentinelmgr;


txc_result_t
txc_sentinelmgr_create(txc_sentinelmgr_t **sentinelmgrp) 
{
        txc_result_t      result;
        txc_pool_object_t *pool_object;
        txc_sentinel_t    *sentinel;
        int               i;

        *sentinelmgrp = (txc_sentinelmgr_t *) MALLOC(sizeof(txc_sentinelmgr_t));
        if (*sentinelmgrp == NULL) {
                return TXC_R_NOMEMORY;
        }

        if ((result = txc_pool_create(&((*sentinelmgrp)->pool_sentinel), 
                                      sizeof(txc_sentinel_t),
                                      TXC_SENTINEL_NUM, 
                                  NULL)) != TXC_R_SUCCESS) 
        {
                FREE(*sentinelmgrp);
                return result;
        }
        TXC_MUTEX_INIT(&((*sentinelmgrp)->mutex), NULL);
        for (i = 0,
             pool_object = txc_pool_object_first((*sentinelmgrp)->pool_sentinel,
                                                 TXC_POOL_OBJECT_ALLOCATED & 
                                                 TXC_POOL_OBJECT_FREE);
             pool_object != NULL;
                 i++, pool_object = txc_pool_object_next(pool_object))
        {
                sentinel = (txc_sentinel_t *) txc_pool_object_of(pool_object);
                TXC_MUTEX_INIT(&sentinel->synch_mutex, NULL);
                TXC_MUTEX_INIT(&sentinel->sentinel_mutex, NULL);
                sentinel->id = i;
                sentinel->manager = *sentinelmgrp;
                sentinel->owner = TXC_SENTINEL_NOOWNER;  
        }
        
        return TXC_R_SUCCESS;
}


void
txc_sentinelmgr_destroy(txc_sentinelmgr_t **sentinelmgrp)
{
        txc_pool_destroy(&((*sentinelmgrp)->pool_sentinel));
        FREE(*sentinelmgrp);
        *sentinelmgrp = NULL;
}


txc_result_t
txc_sentinel_create(txc_sentinelmgr_t *sentinelmgr, txc_sentinel_t **sentinelp) 
{
        txc_result_t   result;
        txc_sentinel_t *sentinel;

        if ((result = txc_pool_object_alloc(sentinelmgr->pool_sentinel, 
                                            (void **) &sentinel, 1)) 
            != TXC_R_SUCCESS) 
        {
                TXC_INTERNALERROR("Could not create sentinel object\n");
                return result;
        }

        sentinel->owner = TXC_SENTINEL_NOOWNER;
        sentinel->refcnt = 1;
        *sentinelp = sentinel;

        return TXC_R_SUCCESS;
}


static inline
void 
sentinel_destroy(txc_sentinel_t *sentinel) 
{
        txc_sentinel_t **sentinelp = &sentinel;

        txc_pool_object_free(sentinel->manager->pool_sentinel, 
                             (void **) sentinelp, 1);
}

void 
txc_sentinel_destroy(txc_sentinel_t *sentinel) 
{
        TXC_ASSERT(sentinel != NULL);
        TXC_MUTEX_LOCK(&sentinel->synch_mutex);
        sentinel_destroy(sentinel);
        TXC_MUTEX_UNLOCK(&sentinel->synch_mutex);
}


static 
void
sentinel_attach(txc_sentinel_t *sentinel)
{
        sentinel->refcnt++;
}


/*
 * Assumes lock sentinel->synch_mutex is held by the caller.
 */
static
txc_result_t 
sentinel_detach(txc_sentinel_t *sentinel)
{
        if (--sentinel->refcnt==0) {
                sentinel_destroy(sentinel);     
        }
        return TXC_R_SUCCESS;
}


txc_result_t
txc_sentinel_detach(txc_sentinel_t *sentinel)
{
        txc_result_t ret;

        TXC_MUTEX_LOCK(&sentinel->synch_mutex);
        ret = sentinel_detach(sentinel);
        TXC_MUTEX_UNLOCK(&sentinel->synch_mutex);
        return ret;
}


static inline
void
sentinel_list_acquire(txc_tx_t *txd, txc_sentinel_list_t *list_acquire)
{
        txc_sentinel_list_entry_t *entry;
        int                       i;

        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "SENTINEL LIST: ACQUIRE SENTINELS\n");
        sentinel_list_print(list_acquire, "SENTINEL_LIST_ACQUIRE");
        for (i=0; i<list_acquire->num_entries; i++) {
                entry = &list_acquire->entries[i];
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                "ACQUIRE SENTINEL %3d: MAY BLOCK\n",
                                entry->sentinel->id);
                TXC_MUTEX_LOCK(&(entry->sentinel->sentinel_mutex));
                entry->sentinel->owner = txd;
                enlist_sentinel(txd->sentinel_list, entry->sentinel, 
                                TXC_SENTINEL_ACQUIRED | 
                                                TXC_SENTINEL_ACQUIREONRETRY);

                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                "ACQUIRE SENTINEL %3d: SUCCESS\n",
                                entry->sentinel->id);
        }
}


static
void
sentinel_list_release(txc_sentinel_list_t *sentinel_list, int transaction_completion)
{
        txc_sentinel_list_entry_t *entry;
        int                       i;

        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "SENTINEL LIST: RELEASE SENTINELS\n");
        for (i=0; i<sentinel_list->num_entries; i++) {
                entry = &sentinel_list->entries[i];
                TXC_MUTEX_LOCK(&(entry->sentinel->synch_mutex));
                if (entry->status & TXC_SENTINEL_ACQUIRED) {
                        entry->sentinel->owner = TXC_SENTINEL_NOOWNER;
                        TXC_MUTEX_UNLOCK(&(entry->sentinel->sentinel_mutex));
                        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                        "RELEASE SENTINEL %3d: SUCCESS\n",
                                        entry->sentinel->id);
                }
                if (transaction_completion) {
                        /* 
                         * sentinel_detach will destroy the sentinel if after the
                         * detach operation (sentinel->refcnt == 0) 
                         */
                        sentinel_detach(entry->sentinel);
                } else {
                        if (!(entry->status & TXC_SENTINEL_ACQUIREONRETRY)) {
                                sentinel_detach(entry->sentinel);
                        }
                }
                TXC_MUTEX_UNLOCK(&(entry->sentinel->synch_mutex));
        }
}


static 
void
sentinel_list_sort(txc_sentinel_list_t *sentinel_list) 
{
        int                       i;
        int                       n = sentinel_list->num_entries;
        txc_sentinel_list_entry_t entry_temp;
        unsigned int              swapped;

        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "SENTINEL LIST: SORT\n");
        do {
                swapped = 0;
                for (i = 0; i<n-1; i++) {
                        if (sentinel_list->entries[i].sentinel->id >
                            sentinel_list->entries[i+1].sentinel->id) 
                        {
                                entry_temp = sentinel_list->entries[i+1];
                                sentinel_list->entries[i+1] = sentinel_list->entries[i];
                                sentinel_list->entries[i] = entry_temp;
                                swapped = 1;
                        }
                }
                n = n-1;
        } while (swapped);
}



static inline
void
sentinelmgr_transaction_on_complete(txc_tx_t *txd)
{
#ifdef _TXC_DEBUG_BUILD
        sentinel_list_print(txd->sentinel_list, "TRANSACTION ON COMPLETE");
        sentinel_list_print(txd->sentinel_list, "SENTINEL LIST (MAIN LIST)");
#endif
        sentinel_list_release(txd->sentinel_list, 1);
        txc_sentinel_list_init(txd->sentinel_list);
        txd->sentinelmgr_undo_action_registered = 0;                                                      
        txd->sentinelmgr_commit_action_registered = 0;                                                    
}


static 
void
sentinelmgr_transaction_before_retry(txc_tx_t *txd)
{
        int                       i;
        txc_sentinel_list_entry_t *entry;

#ifdef _TXC_DEBUG_BUILD
        sentinel_list_print(txd->sentinel_list, "TRANSACTION BEFORE RETRY");
        sentinel_list_print(txd->sentinel_list, "SENTINEL_LIST (MAIN LIST)");
#endif
        txd->sentinel_list_preacquire->num_entries = 0;
        for (i=0; i<txd->sentinel_list->num_entries; i++) {
                entry = &txd->sentinel_list->entries[i];
                if (entry->status & TXC_SENTINEL_ACQUIREONRETRY) {
                        enlist_sentinel(txd->sentinel_list_preacquire, 
                                        entry->sentinel, 
                                        0 & ~TXC_SENTINEL_ACQUIRED);
                }
        }
        sentinel_list_sort(txd->sentinel_list_preacquire);
        /* 
         * Ask sentinel_list_release to release all the sentinels but only drop 
         * the references to sentinels that we won't reacquire after transaction 
         * restarts. This is because we continue logically referencing sentinels 
         * we'll acquire after transaction restarts. 
         */
        sentinel_list_release(txd->sentinel_list, 0);
#ifdef _TXC_DEBUG_BUILD
        sentinel_list_print(txd->sentinel_list_preacquire, 
                            "SENTINEL LIST (PREACQUIRE LIST)");
#endif  
        txc_sentinel_list_init(txd->sentinel_list);
        txd->sentinelmgr_undo_action_registered = 0;                                                      
        txd->sentinelmgr_commit_action_registered = 0;
        backoff(txd);
        sentinel_list_acquire(txd, txd->sentinel_list_preacquire);
}


static
void 
sentinelmgr_commit_action(void *args, int *result)
{
        txc_tx_t *txd = (txc_tx_t *) args;
        sentinelmgr_transaction_on_complete(txd);

        if (result) {
                *result = 0;
        }
}


static 
void 
sentinelmgr_undo_action(void *args, int *result)
{
        txc_tx_t *txd = (txc_tx_t *) args;

        switch (txd->abort_reason) {
                case TXC_ABORTREASON_USERABORT:
                        sentinelmgr_transaction_on_complete(txd);
                        break;
                case TXC_ABORTREASON_TMCONFLICT:
                case TXC_ABORTREASON_USERRETRY:
                case TXC_ABORTREASON_INCONSISTENCY:
                case TXC_ABORTREASON_BUSYSENTINEL:
                case TXC_ABORTREASON_BUSYTXLOCK:
                        sentinelmgr_transaction_before_retry(txd);
                        break;
                default:
                        TXC_INTERNALERROR("Unknown abort reason\n");
                        /* never returns here */
        }

        if (result) {
                *result = 0;
        }
}


static inline
void
register_sentinelmgr_commit_action(txc_tx_t *txd)
{
        if (txd->sentinelmgr_commit_action_registered == 0) {
                txc_tx_register_commit_action(txd, 
                                              sentinelmgr_commit_action, 
                                              (void *) txd,
                                              NULL,
                                          TXC_SENTINELMGR_COMMIT_ACTION_ORDER);
                txd->sentinelmgr_commit_action_registered = 1;                                                    
        }
}


static inline
void
register_sentinelmgr_undo_action(txc_tx_t *txd)
{
        if (txd->sentinelmgr_undo_action_registered == 0) {
                txc_tx_register_undo_action(txd, 
                                              sentinelmgr_undo_action,
                                              (void *) txd,
                                              NULL,
                                          TXC_SENTINELMGR_UNDO_ACTION_ORDER);
                txd->sentinelmgr_undo_action_registered = 1;                                                      
        }
}


void
txc_sentinel_register_sentinelmgr_commit_action(txc_tx_t *txd)
{
        register_sentinelmgr_commit_action(txd);
}


void
txc_sentinel_register_sentinelmgr_undo_action(txc_tx_t *txd)
{
        register_sentinelmgr_undo_action(txd);
}


void
txc_sentinel_transaction_postbegin(txc_tx_t *txd)
{
        /* 
         * If we have preacquired sentinels then we need to register commit and
         * undo actions so that we release the sentinels when the transaction
         * commits/aborts/retries.
         */
        if (txd->sentinel_list->num_entries) {  
                register_sentinelmgr_commit_action(txd);
                register_sentinelmgr_undo_action(txd);
        }       
}


static inline
txc_result_t
allocate_sentinel_list_entries(txc_sentinel_list_t *la, int extend)
{
        if (extend) {
                /* Extend action list */
                la->size *= 2;
                if ((la->entries = 
                     (txc_sentinel_list_entry_t *) REALLOC (la->entries, 
                                                             la->size * sizeof(txc_sentinel_list_entry_t)))
                    == NULL)
                {
                        return TXC_R_NOMEMORY;
                }
        } else {
                /* Allocate action list */
                if ((la->entries = 
                     (txc_sentinel_list_entry_t *) MALLOC (la->size * sizeof(txc_sentinel_list_entry_t)))
                    == NULL)
                {
                        return TXC_R_NOMEMORY;
                }       
        }
        return TXC_R_SUCCESS;
}


static inline
txc_result_t
deallocate_sentinel_list_entries(txc_sentinel_list_t *la)
{
        FREE(la->entries);
        return TXC_R_SUCCESS;
}


txc_result_t
txc_sentinel_list_create(txc_sentinelmgr_t *sentinelmgr, 
                         txc_sentinel_list_t **sentinel_list)
{
        txc_result_t result;

        if ((*sentinel_list = (txc_sentinel_list_t *) MALLOC(sizeof(txc_sentinel_list_t)))
            == NULL) 
        {
                return TXC_R_NOMEMORY;
        }

        (*sentinel_list)->manager = sentinelmgr;
        (*sentinel_list)->num_entries = 0;
        (*sentinel_list)->size = TXC_SENTINEL_LIST_SIZE;

        if ((result = allocate_sentinel_list_entries(*sentinel_list, 0)) 
            != TXC_R_SUCCESS) 
        {
                FREE(*sentinel_list);
                return result;
        }

        return TXC_R_SUCCESS;
}

txc_result_t
txc_sentinel_list_init(txc_sentinel_list_t *sentinel_list)
{
        sentinel_list->num_entries = 0;

        return TXC_R_SUCCESS;
}

txc_result_t
txc_sentinel_list_destroy(txc_sentinel_list_t **sentinel_list) 
{
        txc_result_t result;

        if ((result = deallocate_sentinel_list_entries(*sentinel_list))
            != TXC_R_SUCCESS)
        {
                return result;
        }       
        FREE(*sentinel_list);
        *sentinel_list = NULL;
        return TXC_R_SUCCESS;
}


static inline 
txc_result_t
enlist_sentinel(txc_sentinel_list_t *sentinel_list, 
                txc_sentinel_t *sentinel, 
                txc_result_t status)
{
        txc_result_t ret;

        if (sentinel_list->num_entries == sentinel_list->size) {
                if ((ret = allocate_sentinel_list_entries(sentinel_list, 1)) 
                    != TXC_R_SUCCESS) 
                {
                        return ret;
                }       
        }

        sentinel_list->entries[sentinel_list->num_entries].sentinel = sentinel;
        sentinel_list->entries[sentinel_list->num_entries].status = status;
        sentinel_list->num_entries++; 

        return TXC_R_SUCCESS;
}


txc_result_t 
txc_sentinel_enlist(txc_tx_t *txd, txc_sentinel_t *sentinel, int flags) 
{
        int acquire_on_retry;

        acquire_on_retry = (flags & TXC_SENTINEL_ACQUIREONRETRY) > 0 ? 
                           TXC_SENTINEL_ACQUIREONRETRY : 0;

        TXC_MUTEX_LOCK(&sentinel->synch_mutex);
        sentinel_attach(sentinel);
        TXC_MUTEX_UNLOCK(&sentinel->synch_mutex);
        enlist_sentinel(txd->sentinel_list, sentinel, acquire_on_retry);
        return TXC_R_SUCCESS;
}


txc_result_t 
txc_sentinel_tryacquire(txc_tx_t *txd, txc_sentinel_t *sentinel, int flags) 
{
        int          ret;
        int          num_spin_retries;
        int          acquire_on_retry;
        txc_result_t result;

        TXC_ASSERT(sentinel != NULL);
        TXC_ASSERT(txd != NULL);
        acquire_on_retry = (flags & TXC_SENTINEL_ACQUIREONRETRY) > 0 ? 
                           TXC_SENTINEL_ACQUIREONRETRY : 0;


        if (sentinel->owner != txd) {
                switch (txc_tx_get_xactstate(txd)) {
                        case TXC_XACTSTATE_NONTRANSACTIONAL:
                                TXC_INTERNALERROR("Acquiring a sentinel outside of a transaction not supported\n");
                                /* never gets here */
                        case TXC_XACTSTATE_TRANSACTIONAL_IRREVOCABLE:
                                /* 
                                 * Do not need to acquire any sentinels. 
                                 * Irrevocable global lock already provides atomicity & isolation. 
                                 */
                                result = TXC_R_SUCCESS;
                                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                                "ACQUIRE SENTINEL %3d: SUCCESS (GLOCK)\n",
                                                sentinel->id);
                                break;
                        case TXC_XACTSTATE_TRANSACTIONAL_RETRYABLE:
                                register_sentinelmgr_undo_action(txd);
                                register_sentinelmgr_commit_action(txd);
                                num_spin_retries = txc_runtime_settings.sentinel_max_spin_retries;
                                do {
                                        ret = TXC_MUTEX_TRYLOCK(&sentinel->sentinel_mutex);
                                } while (--num_spin_retries >= 0 && ret != 0);
                                if (ret == 0) {
                                        TXC_MUTEX_LOCK(&sentinel->synch_mutex);
                                        sentinel_attach(sentinel);
                                        TXC_MUTEX_UNLOCK(&sentinel->synch_mutex);
                                        sentinel->owner = txd;
                                        enlist_sentinel(txd->sentinel_list, sentinel, 
                                                        TXC_SENTINEL_ACQUIRED | 
                                                        acquire_on_retry);
                                        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                                        "ACQUIRE SENTINEL %3d: SUCCESS\n",
                                                        sentinel->id);
                                        result = TXC_R_SUCCESS;
                                } else {
                                        TXC_MUTEX_LOCK(&sentinel->synch_mutex);
                                        sentinel_attach(sentinel);
                                        TXC_MUTEX_UNLOCK(&sentinel->synch_mutex);
                                        enlist_sentinel(txd->sentinel_list, sentinel, 
                                                        acquire_on_retry);
                                        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                                        "ACQUIRE SENTINEL %3d: BUSY\n",
                                                        sentinel->id);
                                        result = TXC_R_BUSYSENTINEL;
                                }
                                break;
                        default:
                                TXC_INTERNALERROR("Unknown transaction state\n");
                }
        } else {
#ifdef _TXC_DEBUG_BUILD
                TXC_ASSERT(txc_sentinel_owner(sentinel) == txd);
                TXC_ASSERT(txc_sentinel_is_enlisted(txd, sentinel) == TXC_R_SUCCESS);
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                "ACQUIRE SENTINEL %3d: SUCCESS (HAVE IT)\n",
                                sentinel->id);  
#endif                                                                                                  
        }
        return result;
}


static inline
void
backoff(txc_tx_t *txd)
{
        useconds_t wait_time;

        if (txc_runtime_settings.sentinel_max_backoff_time>0) {
                wait_time = (useconds_t) txc_tx_get_forced_retries(txd) * 32;
                if (wait_time < txc_runtime_settings.sentinel_max_backoff_time) {
                        usleep(wait_time);
                } else {
                        usleep(txc_runtime_settings.sentinel_max_backoff_time);
                }
        }
}


/*
 ****************************************************************************
 ***                      DEBUGGING SUPPORT ROUTINES                      ***
 ****************************************************************************
 */


static inline
void
sentinel_print(void *addr)
{
        txc_sentinel_t *sentinel = (txc_sentinel_t *) addr;

        fprintf(TXC_DEBUG_OUT, "SENTINEL %3d: owner = %p (TID = %u)\n", 
                sentinel->id, 
                sentinel->owner,
                sentinel->owner->tid);

}

static inline
void
sentinel_list_print(txc_sentinel_list_t *sentinel_list, char *heading)
{
        int                       i;
        txc_sentinel_list_entry_t *entry;
        txc_sentinel_t            *sentinel;

        if (TXC_DEBUG_SENTINEL) {
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "%s\n", heading);
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "====================================\n");
                for (i=0; i<sentinel_list->num_entries; i++) {
                        entry = &sentinel_list->entries[i];
                        sentinel = entry->sentinel;
                        TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, 
                                        "SENTINEL %3d: owner = %p (TID = %d) entry_status = %x\n", 
                                        sentinel->id, 
                                        sentinel->owner,
                                        (sentinel->owner == NULL) ? -1 : sentinel->owner->tid,
                                        entry->status);
                }
        }
}


void 
txc_sentinelmgr_print_pools(txc_sentinelmgr_t *sentinelmgr) {
        if (TXC_DEBUG_SENTINEL) {
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "SENTINEL POOL\n");
                TXC_DEBUG_PRINT(TXC_DEBUG_SENTINEL, "====================================\n");
                txc_pool_print(sentinelmgr->pool_sentinel, sentinel_print, 0);
        }
}


txc_tx_t * 
txc_sentinel_owner(txc_sentinel_t *sentinel)
{
        return sentinel->owner;
}


txc_result_t 
txc_sentinel_is_enlisted(txc_tx_t *txd, txc_sentinel_t *sentinel)
{
        int            i;
        txc_sentinel_t *sentinel_iter;

        /* 
         * Sentinel must be found in the list of acquired sentinels and the
         * sentinel's owner field be the descriptor txd.
         */
        for (i=0; i<txd->sentinel_list->num_entries; i++) {
                sentinel_iter = txd->sentinel_list->entries[i].sentinel;
                if (sentinel_iter == sentinel) {
                        return TXC_R_SUCCESS;
                }
        }
        return TXC_R_FAILURE;
}

---