src/core/koa.c

Go to the documentation of this file.

#include <sys/types.h>
#include <sys/stat.h>
#include <libc/syscalls.h>
#include <misc/malloc.h>
#include <misc/result.h>
#include <misc/debug.h>
#include <misc/pool.h>
#include <misc/hash_table.h>
#include <misc/mutex.h>
#include <core/config.h>
#include <core/koa.h>
#include <core/sentinel.h>
#include <core/buffer.h>
#include <stdlib.h>
#include <pthread.h>
#include <stddef.h>


typedef struct txc_koa_file_s txc_koa_file_t;
typedef struct txc_koa_sock_dgram_s txc_koa_sock_dgram_t;
typedef struct txc_koa_pipe_read_end_s txc_koa_pipe_read_end_t;


struct txc_koa_file_s {
        ino_t st_ino;   
        dev_t st_dev;   
        dev_t st_rdev;  
};      


struct txc_koa_sock_dgram_s {
        txc_buffer_circular_t   *buffer_circular_input; 
};


struct txc_koa_pipe_read_end_s {
        txc_buffer_circular_t   *buffer_circular_input; 
};


struct txc_koa_s {
        txc_koamgr_t                *manager;                   
        txc_mutex_t                 mutex;                      
        txc_sentinel_t              *sentinel;                  
        struct {
                int                     fd[TXC_MAX_NUM_FDREFS_PER_KOA]; 
                int                     refcnt;                         
        } fdref;
        int                         refcnt;                     
        int                         type;                       
        union {
                txc_koa_file_t          file;                       
                txc_koa_sock_dgram_t    sock_dgram;                 
                txc_koa_pipe_read_end_t pipe_read_end;              
        };      
}; 

typedef struct txc_fd2koa_s txc_fd2koa_t;


struct txc_fd2koa_s {
        txc_mutex_t mutex;
        txc_koa_t   *koa;
};

typedef struct txc_alias_cache_s txc_alias_cache_t;


struct txc_alias_cache_s {
        txc_mutex_t      mutex;                  /* mutex */
        txc_hash_table_t *hash_tbl;              /* hash table */
};


struct txc_koamgr_s {
        txc_alias_cache_t alias_cache;            
        txc_fd2koa_t      map[TXC_KOA_MAP_SIZE];  
        txc_sentinelmgr_t *sentinelmgr;           
        txc_buffermgr_t   *buffermgr;             
        txc_pool_t        *pool_koa_obj;          
};


txc_koamgr_t *txc_g_koamgr;


txc_result_t
txc_koamgr_create(txc_koamgr_t **koamgrp, 
                  txc_sentinelmgr_t *sentinelmgr, 
                  txc_buffermgr_t *buffermgr) 
{
        int          i;
        txc_result_t result;
        txc_koa_t    *koa;

        *koamgrp = (txc_koamgr_t *) MALLOC(sizeof(txc_koamgr_t));
        if (*koamgrp == NULL) {
                return TXC_R_NOMEMORY;
        }
        if ((result = txc_pool_create(&((*koamgrp)->pool_koa_obj), 
                                      sizeof(txc_koa_t),
                                      TXC_KOA_NUM, 
                                      NULL)) != TXC_R_SUCCESS) 
        {
                FREE(*koamgrp);
                return result;
        }

        if ((result = txc_hash_table_create(&((*koamgrp)->alias_cache.hash_tbl), 
                                            TXC_KOA_CACHE_HASHTBL_SIZE,
                                            TXC_BOOL_FALSE)) != TXC_R_SUCCESS) 
        {
                txc_pool_destroy(&((*koamgrp)->pool_koa_obj));
                FREE(*koamgrp);
                return result;
        }

        for (i=0; i<TXC_KOA_MAP_SIZE; i++) {
                TXC_MUTEX_INIT(&((*koamgrp)->map[i].mutex), NULL);
                (*koamgrp)->map[i].koa = NULL;
        }

        (*koamgrp)->sentinelmgr = sentinelmgr;
        (*koamgrp)->buffermgr = buffermgr;

        /* Precreate standard KOAs: input, output, error */
        txc_koa_create(*koamgrp, &koa, TXC_KOA_IS_PIPE_READ_END, NULL); 
        txc_koa_attach_fd(koa, 0, 0);
        txc_koa_create(*koamgrp, &koa, TXC_KOA_IS_PIPE_WRITE_END, NULL);
        txc_koa_attach_fd(koa, 1, 0);
        txc_koa_create(*koamgrp, &koa, TXC_KOA_IS_PIPE_WRITE_END, NULL);
        txc_koa_attach_fd(koa, 2, 0);

        return TXC_R_SUCCESS;
}


void
txc_koamgr_destroy(txc_koamgr_t **koamgrp) 
{
        txc_pool_destroy(&((*koamgrp)->pool_koa_obj));
        txc_hash_table_destroy(&((*koamgrp)->alias_cache.hash_tbl));
        FREE(*koamgrp);
        *koamgrp = NULL;
}


txc_result_t
txc_koa_lock_alias_cache(txc_koamgr_t *koamgr) 
{
        TXC_MUTEX_LOCK(&(koamgr->alias_cache.mutex));
        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_unlock_alias_cache(txc_koamgr_t *koamgr) 
{
        TXC_MUTEX_UNLOCK(&(koamgr->alias_cache.mutex));
        return TXC_R_SUCCESS;
}


static inline
txc_result_t
alias_cache_insert(txc_koamgr_t *koamgr, txc_koa_t *koa)
{
        txc_hash_table_t *h = koamgr->alias_cache.hash_tbl;
        txc_result_t     ret;
        ino_t            inode_number = koa->file.st_ino;

        TXC_ASSERT(koa->type == TXC_KOA_IS_FILE);

        if ((ret = txc_hash_table_add(h, (unsigned int) inode_number, (void *) koa)) 
            != TXC_R_SUCCESS)
        {
                return ret;
        }
        return TXC_R_SUCCESS;
}


static inline
txc_result_t
alias_cache_remove(txc_koamgr_t *koamgr, txc_koa_t *koa)
{
        txc_hash_table_t *h = koamgr->alias_cache.hash_tbl;
        txc_result_t     ret;
        ino_t            inode_number = koa->file.st_ino;

        if ((ret = txc_hash_table_remove(h, (unsigned int) inode_number, NULL)) 
            != TXC_R_SUCCESS)
        {
                return ret;
        }
        return TXC_R_SUCCESS;
}


txc_result_t 
txc_koa_alias_cache_lookup_inode(txc_koamgr_t *koamgr, ino_t inode_number, txc_koa_t **koap)
{
        txc_koa_t *koa;

        if (txc_hash_table_lookup(koamgr->alias_cache.hash_tbl, 
                                  (unsigned int) inode_number, (void **) &koa)
            != TXC_R_SUCCESS) 
        {
                return TXC_R_NOTEXISTS;
        } else {
                *koap = koa;
                return TXC_R_SUCCESS;
        }
        
}


txc_result_t
txc_koa_create(txc_koamgr_t *koamgr, txc_koa_t **koap, int type, void *args) 
{
        txc_koa_t      *koa;
        txc_sentinel_t *sentinel;
        txc_result_t   result;

        if ((result = txc_pool_object_alloc(koamgr->pool_koa_obj,(void **) &koa, 1)) 
            != TXC_R_SUCCESS) 
        {
                TXC_INTERNALERROR("Could not create KOA object\n");
                return result;
        }
        if ((result = txc_sentinel_create(koamgr->sentinelmgr, &sentinel)) 
            != TXC_R_SUCCESS)
        {       
                TXC_INTERNALERROR("Could not create sentinel for KOA object\n");
                return result;
        }       
        TXC_MUTEX_INIT(&(koa->mutex), NULL); 
        koa->manager = koamgr;
        koa->sentinel = sentinel;
        koa->type = type;
        koa->refcnt = 0;
        koa->fdref.refcnt = 0;

        switch(type) {
                case TXC_KOA_IS_FILE:
                        koa->file.st_ino = (ino_t) args;
                        break;
                case TXC_KOA_IS_SOCK_DGRAM:
                        txc_buffer_circular_create(koa->manager->buffermgr, 
                                                   &(koa->sock_dgram.buffer_circular_input));
                        break;
                case TXC_KOA_IS_PIPE_READ_END:
                        txc_buffer_circular_create(koa->manager->buffermgr, 
                                                   &(koa->pipe_read_end.buffer_circular_input));
                        break;
                default:
                        break; /* do nothing */
        }

        *koap = koa;
        return TXC_R_SUCCESS;
}


void
txc_koa_destroy(txc_koa_t **koap) 
{
        txc_koa_t *koa = *koap;

        /* 
         * txc_sentinel_detach will destroy the sentinel if after the
         * detach operation is (sentinel->refcnt == 0) 
         */
        txc_sentinel_detach((*koap)->sentinel);

        switch((*koap)->type) {
                case TXC_KOA_IS_FILE:
                        break;
                case TXC_KOA_IS_SOCK_DGRAM:
                        txc_buffer_circular_destroy(&((*koap)->sock_dgram.buffer_circular_input));
                        break;
                case TXC_KOA_IS_PIPE_READ_END:
                        txc_buffer_circular_destroy(&((*koap)->pipe_read_end.buffer_circular_input));
                        break;
                default:
                        break; /* do nothing */
        }
        txc_pool_object_free((*koap)->manager->pool_koa_obj, 
                             (void **) &koa, 1);
}


txc_result_t
txc_koa_attach_fd(txc_koa_t *koa, int fd, int lock)
{
        txc_koamgr_t *koamgr;
        int          first_attach;

        TXC_ASSERT(koa != NULL);
        TXC_ASSERT(fd < TXC_KOA_MAP_SIZE);

        koamgr = koa->manager;

        if (lock) {
                TXC_MUTEX_LOCK(&(koamgr->map[fd].mutex));
        }       
        TXC_ASSERT(koamgr->map[fd].koa == NULL);
        koamgr->map[fd].koa = koa;
        koa->refcnt++;
        first_attach = (koa->refcnt == 1) ? 1 : 0;
        if (first_attach) {
                if (koa->type == TXC_KOA_IS_FILE) {
                        alias_cache_insert(koamgr, koa);
                }       
        }

        if (koa->fdref.refcnt >= TXC_MAX_NUM_FDREFS_PER_KOA) {
                TXC_INTERNALERROR("Reached maximum number of file descriptors per KOA.\n");
        }       
        koa->fdref.fd[koa->fdref.refcnt++] = fd;

        TXC_DEBUG_PRINT(TXC_DEBUG_KOA, 
                        "txc_koa_attach_fd: koa = %p, fd = %d, refcnt = %d, sentinel = %p\n", 
                        koa, fd, koa->refcnt, koa->sentinel);

        if (lock) {
                TXC_MUTEX_UNLOCK(&(koamgr->map[fd].mutex));
        }       

        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_detach_fd(txc_koa_t *koa, int fd, int lock)
{
        int          i;
        int          index = -1;
        int          last_detach;
        txc_koamgr_t *koamgr;

        TXC_ASSERT(koa != NULL);
        TXC_ASSERT(fd < TXC_KOA_MAP_SIZE);

        koamgr = koa->manager;

        if (lock) {
                TXC_MUTEX_LOCK(&(koamgr->map[fd].mutex));
        }       
        if (koamgr->map[fd].koa == NULL) {
                return TXC_R_FAILURE;
        }
        koamgr->map[fd].koa = NULL;
        /* Remove backward pointer from KOA to file descriptor */
        for (i=0; i<koa->fdref.refcnt; i++) {
                if (koa->fdref.fd[i] == fd) {
                        index = i;
                        break;
                }
        }
        if (index < 0) {
                if (lock) {
                        TXC_MUTEX_UNLOCK(&(koamgr->map[fd].mutex));
                }       
                return TXC_R_FAILURE;
        }
        koa->fdref.fd[index] = koa->fdref.fd[koa->fdref.refcnt];
        koa->fdref.refcnt--;

        koa->refcnt--;
        last_detach = (koa->refcnt == 0) ? 1 : 0;
        if (last_detach) {
                /* Last detach -- remove it from the alias cache if aliasable KOA. */
                if (koa->type == TXC_KOA_IS_FILE) {
                        alias_cache_remove(koamgr, koa);
                }
                txc_koa_destroy(&koa);
        }       


        TXC_DEBUG_PRINT(TXC_DEBUG_KOA, 
                        "txc_koa_detach_fd: koa = %p, fd = %d, refcnt = %d, sentinel = %p\n", 
                        koa, fd, koa->refcnt, koa->sentinel);

        if (lock) {
                TXC_MUTEX_UNLOCK(&(koamgr->map[fd].mutex));
        }       

        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_lookup_fd2koa(txc_koamgr_t *koamgr, int fd, txc_koa_t **koap) 
{
        TXC_ASSERT(fd < TXC_KOA_MAP_SIZE);
        *koap = koamgr->map[fd].koa;
        if (*koap == NULL) {
                return TXC_R_FAILURE;
        }
        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_lock_fd(txc_koamgr_t *koamgr, int fd)
{
        if (0 == TXC_MUTEX_LOCK(&(koamgr->map[fd].mutex))) {
                return TXC_R_SUCCESS;
        } else {
                return TXC_R_FAILURE;
        }       
}


txc_result_t
txc_koa_unlock_fd(txc_koamgr_t *koamgr, int fd)
{
        if (0 == TXC_MUTEX_UNLOCK(&(koamgr->map[fd].mutex))) {
                return TXC_R_SUCCESS;
        } else {
                return TXC_R_FAILURE;
        }       
}


txc_result_t
txc_koa_lock_fds_refby_koa(txc_koa_t *koa)
{
        int          i;
        int          fd;
        txc_fd2koa_t *entry; 
        txc_koamgr_t *koamgr = koa->manager;

    /*
         * FIXME: Is the following still true ???
         *
         * This is the only point where we acquire multiple FD locks.
         * Therefore we don't need to acquire locks in some total order. 
         */
        for (i=0; i<koa->fdref.refcnt; i++) {
                fd = koa->fdref.fd[i];
                entry = &koamgr->map[fd];
                TXC_MUTEX_LOCK(&(entry->mutex));
        }
        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_unlock_fds_refby_koa(txc_koa_t *koa)
{
        int          i;
        int          fd;
        txc_fd2koa_t *entry; 
        txc_koamgr_t *koamgr = koa->manager;

        for (i=0; i<koa->fdref.refcnt; i++) {
                fd = koa->fdref.fd[i];
                entry = &koamgr->map[fd];
                TXC_MUTEX_UNLOCK(&(entry->mutex));
        }
        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_path2inode(const char *path, ino_t *inode_number) 
{
        int         ret;
        struct stat stat_buf;

        ret = txc_libc_stat(path, &stat_buf);
        if (ret==0) {
                *inode_number = stat_buf.st_ino;
        } else {
                *inode_number = (ino_t) 0;
        }

        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_attach(txc_koa_t *koa) 
{
        int          first_attach;
        txc_koamgr_t *koamgr;

        TXC_ASSERT(koa != NULL);

        koamgr = koa->manager;
        TXC_MUTEX_LOCK(&(koa->mutex));
        koa->refcnt++;
        first_attach = (koa->refcnt == 1) ? 1 : 0;
        if (first_attach) {
                if (koa->type == TXC_KOA_IS_FILE) {
                        alias_cache_insert(koamgr, koa);
                }       
        }

        TXC_DEBUG_PRINT(TXC_DEBUG_KOA, 
                        "txc_koa_attach: koa = %p, refcnt = %d, sentinel = %p\n", 
                        koa, koa->refcnt, koa->sentinel);
        TXC_MUTEX_UNLOCK(&(koa->mutex));

        return TXC_R_SUCCESS;
}


txc_result_t
txc_koa_detach(txc_koa_t *koa) 
{
        int          last_detach;
        txc_koamgr_t *koamgr;

        TXC_ASSERT(koa != NULL);

        koamgr = koa->manager;
        TXC_MUTEX_LOCK(&(koa->mutex));
        koa->refcnt--;
        last_detach = (koa->refcnt == 0) ? 1 : 0;
        if (last_detach) {
                if (koa->type == TXC_KOA_IS_FILE) {
                        alias_cache_remove(koamgr, koa);
                }       
                txc_koa_destroy(&koa);
        }       

        TXC_DEBUG_PRINT(TXC_DEBUG_KOA, 
                        "txc_koa_detach: koa = %p, refcnt = %d, sentinel = %p\n",
                        koa, koa->refcnt, koa->sentinel);
        TXC_MUTEX_UNLOCK(&(koa->mutex));

        return TXC_R_SUCCESS;
}


txc_sentinel_t *
txc_koa_get_sentinel(txc_koa_t *koa)
{
        return koa->sentinel;
}


txc_koamgr_t *
txc_koa_get_koamgr(txc_koa_t *koa)
{
        return koa->manager;
}


void *
txc_koa_get_buffer(txc_koa_t *koa)
{
        void *ret;

        switch (koa->type) {
                case TXC_KOA_IS_SOCK_DGRAM:
                        ret = (void *) koa->sock_dgram.buffer_circular_input;
                case TXC_KOA_IS_PIPE_READ_END:
                        ret = (void *) koa->pipe_read_end.buffer_circular_input;
        }
        return ret;
}

---