fiber.h

Go to the documentation of this file.

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

   Copyright 2013 Konstantin Olkhovskiy <lupus@oxnull.net>

   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.

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

#ifndef _FBR_FIBER_H_
#define _FBR_FIBER_H_

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

#include <evfibers/config.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;

struct fbr_id_s {
        uint64_t g;
        void *p;
} __attribute__((packed));
typedef struct fbr_id_s fbr_id_t;

extern const fbr_id_t FBR_ID_NULL;

static inline int fbr_id_eq(fbr_id_t a, fbr_id_t b)
{
        return a.p == b.p && a.g == b.g;
}

static inline int fbr_id_isnull(fbr_id_t a)
{
        return fbr_id_eq(a, FBR_ID_NULL);
}

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

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, 
        FBR_EV_EIO, 
};

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 fbr_vrb {
        void *mem_ptr;
        size_t mem_ptr_size;
        void *lower_ptr;
        void *upper_ptr;
        size_t ptr_size;
        void *data_ptr;
        void *space_ptr;
};

struct fbr_buffer {
        struct fbr_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_ev_wait_to(FBR_P_ struct fbr_ev_base *events[], ev_tstamp timeout);

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_set_reclaim(FBR_P_ fbr_id_t fiber);
int fbr_set_noreclaim(FBR_P_ fbr_id_t fiber);
int fbr_want_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_cooperate(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);

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

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_recv(FBR_P_ int sockfd, void *buf, size_t len, int flags);

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

ssize_t fbr_send(FBR_P_ int sockfd, const void *buf, size_t len, int flags);

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_vrb_init(struct fbr_vrb *vrb, size_t size, const char *file_pattern);


void fbr_vrb_destroy(struct fbr_vrb *vrb);

static inline size_t fbr_vrb_data_len(struct fbr_vrb *vrb)
{
        return vrb->space_ptr - vrb->data_ptr;
}

static inline size_t fbr_vrb_space_len(struct fbr_vrb *vrb)
{
        return vrb->data_ptr + vrb->ptr_size - vrb->space_ptr;
}

static inline size_t fbr_vrb_capacity(struct fbr_vrb *vrb)
{
        return vrb->ptr_size;
}

static inline void *fbr_vrb_space_ptr(struct fbr_vrb *vrb)
{
        return vrb->space_ptr;
}

static inline void *fbr_vrb_data_ptr(struct fbr_vrb *vrb)
{
        return vrb->data_ptr;
}

static inline int fbr_vrb_give(struct fbr_vrb *vrb, size_t size)
{
        if (size > fbr_vrb_space_len(vrb))
                return -1;
        vrb->space_ptr += size;
        return 0;
}

static inline int fbr_vrb_take(struct fbr_vrb *vrb, size_t size)
{
        if (size > fbr_vrb_data_len(vrb))
                return -1;
        vrb->data_ptr += size;

        if (vrb->data_ptr >= vrb->upper_ptr) {
                vrb->data_ptr -= vrb->ptr_size;
                vrb->space_ptr -= vrb->ptr_size;
        }

        return 0;
}

static inline void fbr_vrb_reset(struct fbr_vrb *vrb)
{
        vrb->data_ptr = vrb->lower_ptr;
        vrb->space_ptr = vrb->lower_ptr;
}

/* Private function */
int fbr_vrb_resize_do(struct fbr_vrb *vrb, size_t new_size,
                const char *file_pattern);

static inline int fbr_vrb_resize(struct fbr_vrb *vrb, size_t new_size,
                const char *file_pattern)
{
        struct fbr_vrb tmp;
        int rv;
        if (fbr_vrb_capacity(vrb) >= new_size)
                return 0;
        memcpy(&tmp, vrb, sizeof(tmp));
        rv = fbr_vrb_init(vrb, new_size, file_pattern);
        if (rv) {
                memcpy(vrb, &tmp, sizeof(tmp));
                return rv;
        }
        memcpy(fbr_vrb_space_ptr(vrb), fbr_vrb_data_ptr(&tmp),
                        fbr_vrb_data_len(&tmp));
        fbr_vrb_give(vrb, fbr_vrb_data_len(&tmp));
        fbr_vrb_destroy(&tmp);
        return 0;
}


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

static inline size_t fbr_buffer_bytes(FBR_PU_ struct fbr_buffer *buffer)
{
        return fbr_vrb_data_len(&buffer->vrb);
}

static inline size_t fbr_buffer_free_bytes(FBR_PU_ struct fbr_buffer *buffer)
{
        return fbr_vrb_space_len(&buffer->vrb);
}

static inline size_t fbr_buffer_size(FBR_PU_ struct fbr_buffer *buffer)
{
        return fbr_vrb_capacity(&buffer->vrb);
}

static inline void *fbr_buffer_space_ptr(FBR_PU_ struct fbr_buffer *buffer)
{
        return fbr_vrb_space_ptr(&buffer->vrb);
}

static inline void *fbr_buffer_data_ptr(FBR_PU_ struct fbr_buffer *buffer)
{
        return fbr_vrb_data_ptr(&buffer->vrb);
}

static inline void fbr_buffer_reset(FBR_PU_ struct fbr_buffer *buffer)
{
        fbr_vrb_reset(&buffer->vrb);
}


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);

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

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);
                (void)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);
                (void)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);

pid_t fbr_popen3(FBR_P_ const char *filename, char *const argv[],
                char *const envp[], const char *working_dir,
                int *stdin_w_ptr, int *stdout_r_ptr, int *stderr_r_ptr);

static inline pid_t fbr_popen2(FBR_P_ const char *filename, char *const argv[],
                char *const envp[], const char *working_dir, int *stdin_w_ptr,
                int *stdout_r_ptr)
{
        return fbr_popen3(FBR_A_ filename, argv, envp, working_dir, stdin_w_ptr,
                        stdout_r_ptr, NULL);
}

static inline pid_t fbr_popen0(FBR_P_ const char *filename, char *const argv[],
                char *const envp[], const char *working_dir)
{
        return fbr_popen3(FBR_A_ filename, argv, envp, working_dir, NULL, NULL,
                        NULL);
}

int fbr_waitpid(FBR_P_ pid_t pid);

int fbr_system(FBR_P_ const char *filename, char *const argv[],
                char *const envp[], const char *working_dir);

#endif