/** @file async.c * @brief Asynchronous input and output. * * Class uses non-blocking input and output and has internal input and output * buffers. Class has associated file descriptor to work with it. * * For async writing user uses faux_async_write() function. It writes all * given data to internal buffer and then tries to really write it to file * descriptor. If not all data was written in non-blocking mode then function * executes special callback "stall" function to inform us about non-empty * output buffer. "Stall" callback function can make programm to inspect fd * for write possibility. Then programm must call faux_async_out() to really * write the rest of the data to fd. Function also can execute "stall" callback. * * For async reading user can call faux_sync_in(). For example this function * can be called after select() or poll() when data is available on interested * fd. Function reads data in non-blocking mode and stores to internal buffer. * User can specify read "limits" - min and max. When amount of reded data is * greater or equal to "min" limit then "read" callback will be executed. * The "read" callback will get allocated buffer with received data. The * length of the data is greater or equal to "min" limit and less or equal to * "max" limit. */ #ifdef HAVE_CONFIG_H #include "config.h" #endif /* HAVE_CONFIG_H */ #include #include #include #include #include #include #include #include #include #include #include #include "faux/faux.h" #include "faux/str.h" #include "faux/buf.h" #include "faux/net.h" #include "faux/async.h" #include "private.h" /** @brief Create new async I/O object. * * Constructor gets associated file descriptor to operate on it. File * descriptor must be nonblocked. If not so then constructor will set * nonblock flag itself. * * @param [in] fd File descriptor. * @return Allocated object or NULL on error. */ faux_async_t *faux_async_new(int fd) { faux_async_t *async = NULL; int fflags = 0; // Prepare FD if (fd < 0) // Illegal fd return NULL; if ((fflags = fcntl(fd, F_GETFL)) == -1) return NULL; if (fcntl(fd, F_SETFL, fflags | O_NONBLOCK) == -1) return NULL; async = faux_zmalloc(sizeof(*async)); assert(async); if (!async) return NULL; // Init async->fd = fd; // Read (Input) async->read_cb = NULL; async->read_udata = NULL; async->min = 1; async->max = FAUX_ASYNC_UNLIMITED; async->ibuf = faux_buf_new(DATA_CHUNK); faux_buf_set_limit(async->ibuf, FAUX_ASYNC_IN_OVERFLOW); // Write (Output) async->stall_cb = NULL; async->stall_udata = NULL; async->obuf = faux_buf_new(DATA_CHUNK); faux_buf_set_limit(async->obuf, FAUX_ASYNC_OUT_OVERFLOW); return async; } /** @brief Free async I/O object. * * @param [in] Async I/O object. */ void faux_async_free(faux_async_t *async) { if (!async) return; faux_buf_free(async->ibuf); faux_buf_free(async->obuf); faux_free(async); } /** @brief Get file descriptor from async I/O object. * * @param [in] async Allocated and initialized async I/O object. * @return Serviced file descriptor. */ int faux_async_fd(const faux_async_t *async) { assert(async); if (!async) return -1; return async->fd; } /** @brief Get input buffer from async I/O object. * * @param [in] async Allocated and initialized async I/O object. * @return faux_buf_t object. */ faux_buf_t *faux_async_ibuf(const faux_async_t *async) { assert(async); if (!async) return NULL; return async->ibuf; } /** @brief Get output buffer from async I/O object. * * @param [in] async Allocated and initialized async I/O object. * @return faux_buf_t object. */ faux_buf_t *faux_async_obuf(const faux_async_t *async) { assert(async); if (!async) return NULL; return async->obuf; } /** @brief Set read callback and associated user data. * * If callback function pointer is NULL then class will drop all readed data. * * @param [in] async Allocated and initialized async I/O object. * @param [in] read_cb Read callback. * @param [in] user_data Associated user data. */ void faux_async_set_read_cb(faux_async_t *async, faux_async_read_cb_fn read_cb, void *user_data) { assert(async); if (!async) return; async->read_cb = read_cb; async->read_udata = user_data; } /** @brief Set read limits. * * Read limits define conditions when the read callback will be executed. * Buffer must contain data amount greater or equal to "min" value. Callback * will not get data amount greater than "max" value. If min == max then * callback will be executed with fixed data size. The "max" value can be "0". * It means indefinite i.e. data transferred to callback can be really large. * * @param [in] async Allocated and initialized async I/O object. * @param [in] min Minimal data amount. * @param [in] max Maximal data amount. The "0" means indefinite. * @return BOOL_TRUE - success, BOOL_FALSE - error. */ bool_t faux_async_set_read_limits(faux_async_t *async, size_t min, size_t max) { assert(async); if (!async) return BOOL_FALSE; if (min < 1) return BOOL_FALSE; if ((min > max) && (max != 0)) return BOOL_FALSE; async->min = min; async->max = max; return BOOL_TRUE; } /** @brief Set stall callback and associated user data. * * @param [in] async Allocated and initialized async I/O object. * @param [in] stall_cb Stall callback. * @param [in] user_data Associated user data. */ void faux_async_set_stall_cb(faux_async_t *async, faux_async_stall_cb_fn stall_cb, void *user_data) { assert(async); if (!async) return; async->stall_cb = stall_cb; async->stall_udata = user_data; } /** @brief Set write overflow value. * * "Overflow" is a value when engine consider data consumer as a stalled. * Data gets into the async I/O object buffer but object can't write it to * serviced fd for too long time. So it accumulates great amount of data. * * @param [in] async Allocated and initialized async I/O object. * @param [in] overflow Overflow value. */ void faux_async_set_write_overflow(faux_async_t *async, size_t overflow) { assert(async); if (!async) return; faux_buf_set_limit(async->obuf, overflow); } /** @brief Set read overflow value. * * "Overflow" is a value when engine consider data consumer as a stalled. * Data gets into the async I/O object buffer but object can't write it to * serviced fd for too long time. So it accumulates great amount of data. * * @param [in] async Allocated and initialized async I/O object. * @param [in] overflow Overflow value. */ void faux_async_set_read_overflow(faux_async_t *async, size_t overflow) { assert(async); if (!async) return; faux_buf_set_limit(async->ibuf, overflow); } /** @brief Async data write. * * All given data will be stored to internal buffer (list of data chunks). * Then function will try to write stored data to file descriptor in * non-blocking mode. Note some data can be left within buffer. In this case * the "stall" callback will be executed to inform about it. To try to write * the rest of the data user can be call faux_async_out() function. Both * functions will not block. * * @param [in] async Allocated and initialized async I/O object. * @param [in] data Data buffer to write. * @param [in] len Data length to write. * @return Length of stored/writed data or < 0 on error. */ ssize_t faux_async_write(faux_async_t *async, void *data, size_t len) { ssize_t data_written = len; assert(async); if (!async) return -1; assert(data); if (!data) return -1; data_written = faux_buf_write(async->obuf, data, len); if (data_written < 0) return -1; // Try to real write data to fd in nonblocked mode faux_async_out(async); return len; } /** @brief Async "struct iovec" write. * * This function is like a faux_async_write() function but uses scatter/gather. * * @see faux_async_write(). * @param [in] async Allocated and initialized async I/O object. * @param [in] iov Array of "struct iovec" structures. * @param [in] iovcnt Number of iov array members. * @return Length of stored/writed data or < 0 on error. */ ssize_t faux_async_writev(faux_async_t *async, const struct iovec *iov, int iovcnt) { size_t total_written = 0; int i = 0; assert(async); if (!async) return -1; if (!iov) return -1; if (iovcnt == 0) return 0; for (i = 0; i < iovcnt; i++) { ssize_t bytes_written = 0; if (iov[i].iov_len == 0) continue; bytes_written = faux_buf_write(async->obuf, iov[i].iov_base, iov[i].iov_len); if (bytes_written < 0) { // Error if (total_written != 0) break; return -1; } if (0 == bytes_written) // Insufficient space break; total_written += bytes_written; } // Try to real write data to fd in nonblocked mode if (total_written > 0) faux_async_out(async); return total_written; } /** @brief Write output buffer to fd in non-blocking mode. * * Previously data must be written to internal buffer by faux_async_write() * function. But some data can be left within internal buffer because can't be * written to fd in non-blocking mode. This function tries to write the rest of * data to fd in non-blocking mode. So function doesn't block. It can be called * after select() or poll() if fd is ready to be written to. If function can't * to write all buffer to fd it executes "stall" callback to inform about it. * * @param [in] async Allocated and initialized async I/O object. * @return Length of data actually written or < 0 on error. */ static ssize_t faux_async_out_internal(faux_async_t *async, bool_t process_all_data) { ssize_t total_written = 0; assert(async); if (!async) return -1; while (faux_buf_len(async->obuf) > 0) { ssize_t data_to_write = 0; ssize_t bytes_written = 0; bool_t postpone = BOOL_FALSE; void *data = NULL; data_to_write = faux_buf_dread_lock_easy(async->obuf, &data); if (data_to_write <= 0) return -1; bytes_written = write(async->fd, data, data_to_write); if (bytes_written > 0) { total_written += bytes_written; faux_buf_dread_unlock_easy(async->obuf, bytes_written); } else { faux_buf_dread_unlock_easy(async->obuf, 0); } if (bytes_written < 0) { if ( // Something went wrong (errno != EINTR) && (errno != EAGAIN) && (errno != EWOULDBLOCK) ) return -1; // Postpone next read postpone = BOOL_TRUE; // Not whole data block was written } else if (bytes_written != data_to_write) { // Postpone next read postpone = BOOL_TRUE; // Write only one data block and buffer is not empty // Programm can be more responsive if to write only one data // block and then allow other events to be processed } else if (!process_all_data && (faux_buf_len(async->obuf) > 0)) { postpone = BOOL_TRUE; } // Postponed if (postpone) { // Execute callback if (async->stall_cb) async->stall_cb(async, faux_buf_len(async->obuf), async->stall_udata); break; } } return total_written; } ssize_t faux_async_out(faux_async_t *async) { return faux_async_out_internal(async, BOOL_TRUE); } ssize_t faux_async_out_easy(faux_async_t *async) { return faux_async_out_internal(async, BOOL_FALSE); } /** @brief Read data and store it to internal buffer in non-blocking mode. * * Reads fd and puts data to internal buffer. It can't be blocked. If length of * data stored within internal buffer is greater or equal than "min" limit then * function will execute "read" callback. It gives faux_buf_t object to callback. * If "max" limit is "0" * (it means indefinite) then function will pass all available data to callback. * * @param [in] async Allocated and initialized async I/O object. * @return Length of data actually readed or < 0 on error. */ static ssize_t faux_async_in_internal(faux_async_t *async, bool_t process_all_data) { ssize_t total_readed = 0; ssize_t bytes_readed = 0; ssize_t locked_len = 0; assert(async); if (!async) return -1; do { void *data = NULL; size_t bytes_stored = 0; locked_len = faux_buf_dwrite_lock_easy(async->ibuf, &data); if (locked_len <= 0) break; // May be buffer is full // Read data bytes_readed = read(async->fd, data, locked_len); if (bytes_readed < 0) { faux_buf_dwrite_unlock_easy(async->ibuf, 0); if ( // Something went wrong (errno != EINTR) && (errno != EAGAIN) && (errno != EWOULDBLOCK) ) return -1; break; } faux_buf_dwrite_unlock_easy(async->ibuf, bytes_readed); total_readed += bytes_readed; if (!async->read_cb) // No read callback continue; // Check for amount of stored data while ((bytes_stored = faux_buf_len(async->ibuf)) >= async->min) { size_t copy_len = 0; // Calculate length of user-requested block if (FAUX_ASYNC_UNLIMITED == async->max) { // Indefinite copy_len = bytes_stored; // Take all data } else { copy_len = (bytes_stored < async->max) ? bytes_stored : async->max; } // Execute callback async->read_cb(async, async->ibuf, copy_len, async->read_udata); } } while ((bytes_readed == locked_len) && process_all_data); return total_readed; } ssize_t faux_async_in(faux_async_t *async) { return faux_async_in_internal(async, BOOL_TRUE); } ssize_t faux_async_in_easy(faux_async_t *async) { return faux_async_in_internal(async, BOOL_FALSE); }