fiber.h

Go to the documentation of this file.

/********************************************************************

  Copyright 2012 Konstantin Olkhovskiy <lupus@oxnull.net>

  This file is part of libevfibers.

  libevfibers 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 3 of the
  License, or any later version.

  libevfibers 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 libevfibers.  If not, see
  <http://www.gnu.org/licenses/>.

 ********************************************************************/

#ifndef _FBR_FIBER_H_
#define _FBR_FIBER_H_

#include <unistd.h>
#include <stdarg.h>
#include <stddef.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <assert.h>
#include <ev.h>

#define FBR_CALL_STACK_SIZE 16

#define FBR_STACK_SIZE (64 * 1024) /* 64 KB */

#ifdef NDEBUG
#define fbr_assert(context, expr)           ((void)(0))
#else
#define fbr_assert(context, expr)                                                             \
        do {                                                                                  \
                __typeof__(expr) ex = (expr);                                                 \
                if (ex)                                                                       \
                        (void)(0);                                                            \
                else {                                                                        \
                        fbr_dump_stack(context, fbr_log_e);                                   \
                        __assert_fail(__STRING(expr), __FILE__, __LINE__, __ASSERT_FUNCTION); \
                }                                                                             \
        } while (0)
#endif

#define fbr_container_of(ptr, type, member) ({                       \
                const typeof( ((type *)0)->member ) *__mptr = (ptr); \
                (type *)( (char *)__mptr - offsetof(type,member) );  \
                })

struct fbr_context_private;
struct fbr_logger;
typedef __uint128_t fbr_id_t;

enum fbr_error_code {
        FBR_SUCCESS = 0,
        FBR_EINVAL,
        FBR_ENOFIBER,
        FBR_ESYSTEM,
        FBR_EBUFFERMMAP,
        FBR_ENOKEY,
};

struct fbr_context {
        struct fbr_context_private *__p; 
        enum fbr_error_code f_errno; 
        struct fbr_logger *logger; 
};

#define FBR_P struct fbr_context *fctx

#define FBR_PU __attribute__((unused)) FBR_P

#define FBR_P_ FBR_P,

#define FBR_PU_ __attribute__((unused)) FBR_P_

#define FBR_A fctx

#define FBR_A_ FBR_A,

typedef void (*fbr_fiber_func_t)(FBR_P_ void *_arg);

typedef void (*fbr_alloc_destructor_func_t)(FBR_P_ void *ptr, void *context);

enum fbr_log_level {
        FBR_LOG_ERROR = 0,
        FBR_LOG_WARNING,
        FBR_LOG_NOTICE,
        FBR_LOG_INFO,
        FBR_LOG_DEBUG
};

struct fbr_logger;

typedef void (*fbr_log_func_t)(FBR_P_ struct fbr_logger *logger,
                enum fbr_log_level level, const char *format, va_list ap);

typedef void (*fbr_logutil_func_t)(FBR_P_ const char *format, ...);

struct fbr_logger {
        fbr_log_func_t logv; 
        enum fbr_log_level level; 
        void *data; 
};

static inline int fbr_need_log(FBR_P_ enum fbr_log_level level)
{
        return level <= fctx->logger->level;
}

static inline void fbr_set_log_level(FBR_P_ enum fbr_log_level desired_level)
{
        fctx->logger->level = desired_level;
}

enum fbr_ev_type {
        FBR_EV_WATCHER = 1, 
        FBR_EV_MUTEX, 
        FBR_EV_COND_VAR, 
};

struct fbr_ev_base;

typedef void (*fbr_destructor_func_t)(FBR_P_ void *arg);

struct fbr_destructor {
        fbr_destructor_func_t func; 
        void *arg; 
        TAILQ_ENTRY(fbr_destructor) entries; //Private
        int active; //Private
};

#define FBR_DESTRUCTOR_INITIALIZER { \
        .func   = NULL,              \
        .arg    = NULL,              \
        .active = 0,                 \
};

struct fbr_id_tailq;

struct fbr_id_tailq_i {
        /* Private structure */
        fbr_id_t id;
        struct fbr_ev_base *ev;
        TAILQ_ENTRY(fbr_id_tailq_i) entries;
        struct fbr_destructor dtor;
        struct fbr_id_tailq *head;
};

TAILQ_HEAD(fbr_id_tailq, fbr_id_tailq_i);

struct fbr_ev_base {
        enum fbr_ev_type type; 
        fbr_id_t id; 
        int arrived; 
        struct fbr_context *fctx; //Private
        void *data; //Private
        struct fbr_id_tailq_i item; //Private
};

#define fbr_ev_upcast(ptr, type_no_struct) \
       fbr_container_of(ptr, struct type_no_struct, ev_base)

struct fbr_ev_watcher {
        ev_watcher *w; 
        struct fbr_ev_base ev_base;
};

struct fbr_ev_mutex {
        struct fbr_mutex *mutex; 
        struct fbr_ev_base ev_base;
};

struct fbr_ev_cond_var {
        struct fbr_cond_var *cond; 
        struct fbr_mutex *mutex; 
        struct fbr_ev_base ev_base;
};

struct fbr_mutex {
        fbr_id_t locked_by;
        struct fbr_id_tailq pending;
        TAILQ_ENTRY(fbr_mutex) entries;
};

struct fbr_cond_var {
        struct fbr_mutex *mutex;
        struct fbr_id_tailq waiting;
};

struct vrb;
struct fbr_buffer {
        struct vrb *vrb;
        size_t prepared_bytes;
        size_t waiting_bytes;
        struct fbr_cond_var committed_cond;
        struct fbr_mutex write_mutex;
        struct fbr_cond_var bytes_freed_cond;
        struct fbr_mutex read_mutex;
};

typedef unsigned int fbr_key_t;


#define FBR_MAX_KEY 64

void fbr_destructor_add(FBR_P_ struct fbr_destructor *dtor);

void fbr_destructor_remove(FBR_P_ struct fbr_destructor *dtor,
                int call);

static inline void fbr_destructor_init(struct fbr_destructor *dtor)
{
        memset(dtor, 0x00, sizeof(*dtor));
}

void fbr_ev_watcher_init(FBR_P_ struct fbr_ev_watcher *ev, ev_watcher *w);

void fbr_ev_mutex_init(FBR_P_ struct fbr_ev_mutex *ev,
                struct fbr_mutex *mutex);

void fbr_ev_cond_var_init(FBR_P_ struct fbr_ev_cond_var *ev,
                struct fbr_cond_var *cond, struct fbr_mutex *mutex);

int fbr_ev_wait_one(FBR_P_ struct fbr_ev_base *one);

int fbr_ev_wait(FBR_P_ struct fbr_ev_base *events[]);

int fbr_transfer(FBR_P_ fbr_id_t to);

void fbr_init(struct fbr_context *fctx, struct ev_loop *loop);

void fbr_destroy(FBR_P);

void fbr_enable_backtraces(FBR_P, int enabled);

const char *fbr_strerror(FBR_P_ enum fbr_error_code code);

void fbr_log_e(FBR_P_ const char *format, ...)
        __attribute__ ((format (printf, 2, 3)));

void fbr_log_w(FBR_P_ const char *format, ...)
        __attribute__ ((format (printf, 2, 3)));

void fbr_log_n(FBR_P_ const char *format, ...)
        __attribute__ ((format (printf, 2, 3)));

void fbr_log_i(FBR_P_ const char *format, ...)
        __attribute__ ((format (printf, 2, 3)));

void fbr_log_d(FBR_P_ const char *format, ...)
        __attribute__ ((format (printf, 2, 3)));

#define FBR_MAX_FIBER_NAME 64

fbr_id_t fbr_create(FBR_P_ const char *name, fbr_fiber_func_t func, void *arg,
                size_t stack_size);

const char *fbr_get_name(FBR_P_ fbr_id_t id);

int fbr_set_name(FBR_P_ fbr_id_t id, const char *name);

int fbr_disown(FBR_P_ fbr_id_t parent);

fbr_id_t fbr_parent(FBR_P);

int fbr_reclaim(FBR_P_ fbr_id_t fiber);

int fbr_is_reclaimed(FBR_P_ fbr_id_t fiber);

fbr_id_t fbr_self(FBR_P);

int fbr_key_create(FBR_P_ fbr_key_t *key);

int fbr_key_delete(FBR_P_ fbr_key_t key);

int fbr_key_set(FBR_P_ fbr_id_t id, fbr_key_t key, void *value);

void *fbr_key_get(FBR_P_ fbr_id_t id, fbr_key_t key);

void fbr_yield(FBR_P);

void *fbr_alloc(FBR_P_ size_t size);

void fbr_alloc_set_destructor(FBR_P_ void *ptr, fbr_alloc_destructor_func_t func,
                void *context);

void *fbr_calloc(FBR_P_ unsigned int nmemb, size_t size);

void fbr_free(FBR_P_ void *ptr);

void fbr_free_nd(FBR_P_ void *ptr);

int fbr_fd_nonblock(FBR_P_ int fd);

ssize_t fbr_read(FBR_P_ int fd, void *buf, size_t count);

ssize_t fbr_read_all(FBR_P_ int fd, void *buf, size_t count);

ssize_t fbr_readline(FBR_P_ int fd, void *buffer, size_t n);

ssize_t fbr_write(FBR_P_ int fd, const void *buf, size_t count);

ssize_t fbr_write_all(FBR_P_ int fd, const void *buf, size_t count);

ssize_t fbr_recvfrom(FBR_P_ int sockfd, void *buf, size_t len, int flags,
                struct sockaddr *src_addr, socklen_t *addrlen);

ssize_t fbr_sendto(FBR_P_ int sockfd, const void *buf, size_t len, int flags, const
                struct sockaddr *dest_addr, socklen_t addrlen);

int fbr_accept(FBR_P_ int sockfd, struct sockaddr *addr, socklen_t *addrlen);

ev_tstamp fbr_sleep(FBR_P_ ev_tstamp seconds);

void fbr_dump_stack(FBR_P_ fbr_logutil_func_t log);

void fbr_mutex_init(FBR_P_ struct fbr_mutex *mutex);

void fbr_mutex_lock(FBR_P_ struct fbr_mutex *mutex);

int fbr_mutex_trylock(FBR_P_ struct fbr_mutex *mutex);

void fbr_mutex_unlock(FBR_P_ struct fbr_mutex *mutex);

void fbr_mutex_destroy(FBR_P_ struct fbr_mutex *mutex);

void fbr_cond_init(FBR_P_ struct fbr_cond_var *cond);

void fbr_cond_destroy(FBR_P_ struct fbr_cond_var *cond);

int fbr_cond_wait(FBR_P_ struct fbr_cond_var *cond, struct fbr_mutex *mutex);

void fbr_cond_broadcast(FBR_P_ struct fbr_cond_var *cond);

void fbr_cond_signal(FBR_P_ struct fbr_cond_var *cond);

int fbr_buffer_init(FBR_P_ struct fbr_buffer *buffer, size_t size);

void fbr_buffer_destroy(FBR_P_ struct fbr_buffer *buffer);

void *fbr_buffer_alloc_prepare(FBR_P_ struct fbr_buffer *buffer, size_t size);

void fbr_buffer_alloc_commit(FBR_P_ struct fbr_buffer *buffer);

void fbr_buffer_alloc_abort(FBR_P_ struct fbr_buffer *buffer);

void *fbr_buffer_read_address(FBR_P_ struct fbr_buffer *buffer, size_t size);

void fbr_buffer_read_advance(FBR_P_ struct fbr_buffer *buffer);

void fbr_buffer_read_discard(FBR_P_ struct fbr_buffer *buffer);

size_t fbr_buffer_bytes(FBR_P_ struct fbr_buffer *buffer);

size_t fbr_buffer_free_bytes(FBR_P_ struct fbr_buffer *buffer);

static inline struct fbr_cond_var *fbr_buffer_cond_read(FBR_PU_
                struct fbr_buffer *buffer)
{
        return &buffer->committed_cond;
}

static inline struct fbr_cond_var *fbr_buffer_cond_write(FBR_PU_
                struct fbr_buffer *buffer)
{
        return &buffer->bytes_freed_cond;
}

static inline int fbr_buffer_wait_read(FBR_P_ struct fbr_buffer *buffer,
                size_t size)
{
        struct fbr_mutex mutex;
        int retval;
        fbr_mutex_init(FBR_A_ &mutex);
        fbr_mutex_lock(FBR_A_ &mutex);
        while (fbr_buffer_bytes(FBR_A_ buffer) < size) {
                retval = fbr_cond_wait(FBR_A_ &buffer->committed_cond, &mutex);
                assert(0 == retval);
        }
        fbr_mutex_unlock(FBR_A_ &mutex);
        fbr_mutex_destroy(FBR_A_ &mutex);
        return 1;
}

static inline int fbr_buffer_can_read(FBR_P_ struct fbr_buffer *buffer,
                size_t size)
{
        return fbr_buffer_bytes(FBR_A_ buffer) >= size;
}

static inline int fbr_buffer_wait_write(FBR_P_ struct fbr_buffer *buffer,
                size_t size)
{
        struct fbr_mutex mutex;
        int retval;
        fbr_mutex_init(FBR_A_ &mutex);
        fbr_mutex_lock(FBR_A_ &mutex);
        while (fbr_buffer_free_bytes(FBR_A_ buffer) < size) {
                retval = fbr_cond_wait(FBR_A_ &buffer->bytes_freed_cond,
                                &mutex);
                assert(0 == retval);
        }
        fbr_mutex_unlock(FBR_A_ &mutex);
        fbr_mutex_destroy(FBR_A_ &mutex);
        return 1;
}

static inline int fbr_buffer_can_write(FBR_P_ struct fbr_buffer *buffer,
                size_t size)
{
        return fbr_buffer_free_bytes(FBR_A_ buffer) >= size;
}

void *fbr_get_user_data(FBR_P_ fbr_id_t id);

int fbr_set_user_data(FBR_P_ fbr_id_t id, void *data);

#endif