123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510 |
- /** @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 <stdlib.h>
- #include <stdint.h>
- #include <stdio.h>
- #include <string.h>
- #include <assert.h>
- #include <unistd.h>
- #include <errno.h>
- #include <sys/types.h>
- #include <sys/stat.h>
- #include <fcntl.h>
- #include <syslog.h>
- #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)
- return -1;
- // 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);
- }
|