hyung-hwan/hio
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added more sendfile related code

Browse Source
This commit is contained in:
hyung-hwan 2020-07-19 06:23:43 +00:00
parent e439c29cb8
commit 8655269e07
7 changed files with 300 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
mio/lib/http-fil.c
View File

@ -119,6 +119,25 @@ static int file_state_write_to_client (file_state_t* file_state, const void* dat
return 0;
}
static int file_state_sendfile_to_client (file_state_t* file_state, mio_foff_t foff, mio_iolen_t len)
{
file_state->ever_attempted_to_write_to_client = 1;
file_state->num_pending_writes_to_client++;
if (mio_dev_sck_sendfile(file_state->client->sck, file_state->peer, foff, len, MIO_NULL) <= -1)
{
file_state->num_pending_writes_to_client--;
return -1;
}
if (file_state->num_pending_writes_to_client > FILE_STATE_PENDING_IO_THRESHOLD)
{
/* STOP READING */
/*if (mio_dev_pro_read(file_state->peer, MIO_DEV_PRO_OUT, 0) <= -1) return -1;*/
}
return 0;
}
static int file_state_send_final_status_to_client (file_state_t* file_state, int status_code, int force_close)
{
mio_svc_htts_cli_t* cli = file_state->client;
@ -137,20 +156,6 @@ static int file_state_send_final_status_to_client (file_state_t* file_state, int
(force_close && file_state_write_to_client(file_state, MIO_NULL, 0) <= -1))? -1: 0;
}
#if 0
static int file_state_write_last_chunk_to_client (file_state_t* file_state)
{
if (!file_state->ever_attempted_to_write_to_client)
{
if (file_state_send_final_status_to_client(file_state, 500, 0) <= -1) return -1;
}
if (!file_state->keep_alive && file_state_write_to_client(file_state, MIO_NULL, 0) <= -1) return -1;
return 0;
}
#endif
static void file_state_close_peer (file_state_t* file_state)
{
mio_t* mio = file_state->htts->mio;
@ -473,7 +478,6 @@ static int file_state_send_contents_to_client (file_state_t* file_state)
*/
mio_t* mio = file_state->htts->mio;
mio_foff_t lim;
ssize_t n;
if (file_state->cur_offset > file_state->end_offset)
{
@ -483,37 +487,49 @@ static int file_state_send_contents_to_client (file_state_t* file_state)
}
lim = file_state->end_offset - file_state->cur_offset + 1;
n = read(file_state->peer, file_state->peer_buf, (lim < MIO_SIZEOF(file_state->peer_buf)? lim: MIO_SIZEOF(file_state->peer_buf)));
if (n == -1)
if (1 /*mio_dev_sck_sendfileok(file_state->client->sck)*/)
{
if ((errno == EAGAIN || errno == EINTR) && file_state->peer_tmridx == MIO_TMRIDX_INVALID)
if (lim > 0xFFFF) lim = 0xFFFF; /* TODO: change this... */
if (file_state_sendfile_to_client(file_state, file_state->cur_offset, lim) <= -1) return -1;
file_state->cur_offset += lim;
}
else
{
ssize_t n;
n = read(file_state->peer, file_state->peer_buf, (lim < MIO_SIZEOF(file_state->peer_buf)? lim: MIO_SIZEOF(file_state->peer_buf)));
if (n == -1)
{
mio_tmrjob_t tmrjob;
/* use a timer job for a new sending attempt */
MIO_MEMSET (&tmrjob, 0, MIO_SIZEOF(tmrjob));
tmrjob.ctx = file_state;
/*tmrjob.when = leave it at 0 for immediate firing.*/
tmrjob.handler = send_contents_to_client_later;
tmrjob.idxptr = &file_state->peer_tmridx;
return mio_instmrjob(mio, &tmrjob) == MIO_TMRIDX_INVALID? -1: 0;
if ((errno == EAGAIN || errno == EINTR) && file_state->peer_tmridx == MIO_TMRIDX_INVALID)
{
mio_tmrjob_t tmrjob;
/* use a timer job for a new sending attempt */
MIO_MEMSET (&tmrjob, 0, MIO_SIZEOF(tmrjob));
tmrjob.ctx = file_state;
/*tmrjob.when = leave it at 0 for immediate firing.*/
tmrjob.handler = send_contents_to_client_later;
tmrjob.idxptr = &file_state->peer_tmridx;
return mio_instmrjob(mio, &tmrjob) == MIO_TMRIDX_INVALID? -1: 0;
}
return -1;
}
else if (n == 0)
{
/* no more data to read - this must not happend unless file size changed while the file is open. */
/* TODO: I probably must close the connection by force??? */
file_state_mark_over (file_state, FILE_STATE_OVER_READ_FROM_PEER);
return -1;
}
return -1;
if (file_state_write_to_client(file_state, file_state->peer_buf, n) <= -1) return -1;
file_state->cur_offset += n;
/* if (file_state->cur_offset > file_state->end_offset) should i check this or wait until this function is invoked?
file_state_mark_over (file_state, FILE_STATE_OVER_READ_FROM_PEER);*/
}
else if (n == 0)
{
/* no more data to read - this must not happend unless file size changed while the file is open. */
/* TODO: I probably must close the connection by force??? */
file_state_mark_over (file_state, FILE_STATE_OVER_READ_FROM_PEER);
return -1;
}
if (file_state_write_to_client(file_state, file_state->peer_buf, n) <= -1) return -1;
file_state->cur_offset += n;
/* if (file_state->cur_offset > file_state->end_offset) should i check this or wait until this function is invoked?
file_state_mark_over (file_state, FILE_STATE_OVER_READ_FROM_PEER);*/
return 0;
}

1
mio/lib/mar.c
View File

@ -329,6 +329,7 @@ static mio_dev_mth_t dev_mar_methods =
MIO_NULL,
MIO_NULL,
MIO_NULL,
MIO_NULL, /* sendfile */
dev_mar_ioctl
};

15
mio/lib/mio-sck.h
View File

@ -475,6 +475,7 @@ MIO_EXPORT int mio_dev_sck_writev (
const mio_skad_t* dstaddr
);
MIO_EXPORT int mio_dev_sck_timedwrite (
mio_dev_sck_t* dev,
const void* data,
@ -516,13 +517,27 @@ static MIO_INLINE int mio_dev_sck_timedread (mio_dev_sck_t* sck, int enabled, mi
{
return mio_dev_timedread((mio_dev_t*)sck, enabled, tmout);
}
static MIO_INLINE int mio_dev_sck_sendfile (mio_dev_sck_t* sck, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t len, void* wrctx)
{
return mio_dev_sendfile((mio_dev_t*)sck, in_fd, foff, len, wrctx);
}
static MIO_INLINE int mio_dev_sck_timedsendfile (mio_dev_sck_t* sck, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t len, mio_ntime_t* tmout, void* wrctx)
{
return mio_dev_timedsendfile((mio_dev_t*)sck, in_fd, foff, len, tmout, wrctx);
}
#else
#define mio_dev_sck_kill(sck) mio_dev_kill((mio_dev_t*)sck)
#define mio_dev_sck_halt(sck) mio_dev_halt((mio_dev_t*)sck)
#define mio_dev_sck_read(sck,enabled) mio_dev_read((mio_dev_t*)sck, enabled)
#define mio_dev_sck_timedread(sck,enabled,tmout) mio_dev_timedread((mio_dev_t*)sck, enabled, tmout)
#define mio_dev_sck_sendfile(sck,in_fd,foff,len,wrctx) mio_dev_sendfile((mio_dev_t*)sck, in_fd, foff, len, wrctx)
#define mio_dev_sck_timedsendfile(sck,in_fd,foff,len,tmout,wrctx) mio_dev_timedsendfile((mio_dev_t*)sck, in_fd, foff, len, tmout, wrctx)
#endif

132
mio/lib/mio.c
View File

@ -37,6 +37,13 @@ static int kill_and_free_device (mio_dev_t* dev, int force);
static void on_read_timeout (mio_t* mio, const mio_ntime_t* now, mio_tmrjob_t* job);
static void on_write_timeout (mio_t* mio, const mio_ntime_t* now, mio_tmrjob_t* job);
struct wq_sendfile_data_t
{
mio_syshnd_t in_fd;
mio_foff_t foff;
};
typedef struct wq_sendfile_data_t wq_sendfile_data_t;
/* ========================================================================= */
static void* mmgr_alloc (mio_mmgr_t* mmgr, mio_oow_t size)
@ -444,7 +451,14 @@ static MIO_INLINE void handle_event (mio_t* mio, mio_dev_t* dev, int events, int
send_leftover:
ulen = urem;
x = dev->dev_mth->write(dev, uptr, &ulen, &q->dstaddr);
if (q->sendfile)
{
x = dev->dev_mth->sendfile(dev, ((wq_sendfile_data_t*)uptr)->in_fd, ((wq_sendfile_data_t*)uptr)->foff, &ulen);
}
else
{
x = dev->dev_mth->write(dev, uptr, &ulen, &q->dstaddr);
}
if (x <= -1)
{
mio_dev_halt (dev);
@ -1291,6 +1305,7 @@ static MIO_INLINE int __enqueue_pending_write (mio_dev_t* dev, mio_iolen_t olen,
q = (mio_wq_t*)mio_allocmem(mio, MIO_SIZEOF(*q) + (dstaddr? dstaddr->len: 0) + urem);
if (MIO_UNLIKELY(!q)) return -1;
q->sendfile = 0;
q->tmridx = MIO_TMRIDX_INVALID;
q->dev = dev;
q->ctx = wrctx;
@ -1349,7 +1364,81 @@ static MIO_INLINE int __enqueue_pending_write (mio_dev_t* dev, mio_iolen_t olen,
return 0; /* request pused to a write queue. */
}
static int __dev_write (mio_dev_t* dev, const void* data, mio_iolen_t len, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
static MIO_INLINE int __enqueue_pending_sendfile (mio_dev_t* dev, mio_iolen_t olen, mio_iolen_t urem, mio_foff_t foff, mio_syshnd_t in_fd, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
{
mio_t* mio = dev->mio;
mio_wq_t* q;
if (dev->dev_cap & MIO_DEV_CAP_OUT_UNQUEUEABLE)
{
/* writing queuing is not requested. so return failure */
mio_seterrbfmt (mio, MIO_ENOCAPA, "device incapable of queuing");
return -1;
}
/* queue the remaining data*/
q = (mio_wq_t*)mio_allocmem(mio, MIO_SIZEOF(*q) + (dstaddr? dstaddr->len: 0) + MIO_SIZEOF(wq_sendfile_data_t));
if (MIO_UNLIKELY(!q)) return -1;
q->sendfile = 1;
q->tmridx = MIO_TMRIDX_INVALID;
q->dev = dev;
q->ctx = wrctx;
if (dstaddr)
{
q->dstaddr.ptr = (mio_uint8_t*)(q + 1);
q->dstaddr.len = dstaddr->len;
MIO_MEMCPY (q->dstaddr.ptr, dstaddr->ptr, dstaddr->len);
}
else
{
q->dstaddr.len = 0;
}
q->ptr = (mio_uint8_t*)(q + 1) + q->dstaddr.len;
q->len = urem;
q->olen = olen; /* original length to use when invoking on_write() */
((wq_sendfile_data_t*)q->ptr)->in_fd = in_fd;
((wq_sendfile_data_t*)q->ptr)->foff = foff;
if (tmout && MIO_IS_POS_NTIME(tmout))
{
mio_tmrjob_t tmrjob;
MIO_MEMSET (&tmrjob, 0, MIO_SIZEOF(tmrjob));
tmrjob.ctx = q;
mio_gettime (mio, &tmrjob.when);
MIO_ADD_NTIME (&tmrjob.when, &tmrjob.when, tmout);
tmrjob.handler = on_write_timeout;
tmrjob.idxptr = &q->tmridx;
q->tmridx = mio_instmrjob(mio, &tmrjob);
if (q->tmridx == MIO_TMRIDX_INVALID)
{
mio_freemem (mio, q);
return -1;
}
}
MIO_WQ_ENQ (&dev->wq, q);
if (!(dev->dev_cap & MIO_DEV_CAP_OUT_WATCHED))
{
/* if output is not being watched, arrange to do so */
if (mio_dev_watch(dev, MIO_DEV_WATCH_RENEW, MIO_DEV_EVENT_IN) <= -1)
{
unlink_wq (mio, q);
mio_freemem (mio, q);
return -1;
}
}
return 0; /* request pused to a write queue. */
}
static MIO_INLINE int __dev_write (mio_dev_t* dev, const void* data, mio_iolen_t len, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
{
mio_t* mio = dev->mio;
const mio_uint8_t* uptr;
@ -1444,7 +1533,7 @@ enqueue_completed_write:
return __enqueue_completed_write(dev, len, wrctx, dstaddr);
}
static int __dev_writev (mio_dev_t* dev, mio_iovec_t* iov, mio_iolen_t iovcnt, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
static MIO_INLINE int __dev_writev (mio_dev_t* dev, mio_iovec_t* iov, mio_iolen_t iovcnt, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
{
mio_t* mio = dev->mio;
mio_iolen_t urem, len;
@ -1551,18 +1640,23 @@ enqueue_completed_write:
static int __dev_sendfile (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t len, const mio_ntime_t* tmout, void* wrctx)
{
#if 1
mio_t* mio = dev->mio;
mio_foff_t uoff;
mio_iolen_t urem, ulen;
int x;
if (dev->dev_cap & MIO_DEV_CAP_OUT_CLOSED)
if (MIO_UNLIKELY(dev->dev_cap & MIO_DEV_CAP_OUT_CLOSED))
{
mio_seterrbfmt (mio, MIO_ENOCAPA, "unable to sendfile to closed device");
return -1;
}
if (MIO_UNLIKELY(!dev->dev_mth->sendfile))
{
mio_seterrbfmt (mio, MIO_ENOCAPA, "unable to senfile over unsupported device");
return -1;
}
uoff = foff;
urem = len;
@ -1579,7 +1673,7 @@ static int __dev_sendfile (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff,
do
{
ulen = urem;
x = dev->dev_mth->sendfile(dev, in_fd, foff, &ulen);
x = dev->dev_mth->sendfile(dev, in_fd, uoff, &ulen);
if (x <= -1) return -1;
else if (x == 0)
{
@ -1626,21 +1720,11 @@ static int __dev_sendfile (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff,
return 1; /* written immediately and called on_write callback. but this line will never be reached */
enqueue_data:
#if 0
iov.iov_ptr = (void*)uptr;
iov.iov_len = urem;
return __enqueue_pending_write(dev, len, urem, &iov, 1, 0, tmout, wrctx, dstaddr);
#else
/* TODO: */
return -1;
#endif
printf ("queueing sendfiel...\n");
return __enqueue_pending_sendfile(dev, len, urem, uoff, in_fd, tmout, wrctx, MIO_NULL);
enqueue_completed_write:
return __enqueue_completed_write(dev, len, wrctx, dstaddr);
#else
return 0;
#endif
return __enqueue_completed_write(dev, len, wrctx, MIO_NULL);
}
int mio_dev_write (mio_dev_t* dev, const void* data, mio_iolen_t len, void* wrctx, const mio_devaddr_t* dstaddr)
@ -1653,6 +1737,11 @@ int mio_dev_writev (mio_dev_t* dev, mio_iovec_t* iov, mio_iolen_t iovcnt, void*
return __dev_writev(dev, iov, iovcnt, MIO_NULL, wrctx, dstaddr);
}
int mio_dev_sendfile (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t len, void* wrctx)
{
return __dev_sendfile(dev,in_fd, foff, len, MIO_NULL, wrctx);
}
int mio_dev_timedwrite (mio_dev_t* dev, const void* data, mio_iolen_t len, const mio_ntime_t* tmout, void* wrctx, const mio_devaddr_t* dstaddr)
{
return __dev_write(dev, data, len, tmout, wrctx, dstaddr);
@ -1663,6 +1752,11 @@ int mio_dev_timedwritev (mio_dev_t* dev, mio_iovec_t* iov, mio_iolen_t iovcnt, c
return __dev_writev(dev, iov, iovcnt, tmout, wrctx, dstaddr);
}
int mio_dev_timedsendfile (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t len, const mio_ntime_t* tmout, void* wrctx)
{
return __dev_sendfile(dev,in_fd, foff, len, tmout, wrctx);
}
/* -------------------------------------------------------------------------- */
void mio_gettime (mio_t* mio, mio_ntime_t* now)

17
mio/lib/mio.h
View File

@ -299,6 +299,7 @@ struct mio_wq_t
mio_wq_t* q_next;
mio_wq_t* q_prev;
int sendfile;
mio_iolen_t olen; /* original data length */
mio_uint8_t* ptr; /* pointer to data */
mio_iolen_t len; /* remaining data length */
@ -899,6 +900,14 @@ MIO_EXPORT int mio_dev_writev (
const mio_devaddr_t* dstaddr
);
MIO_EXPORT int mio_dev_sendfile (
mio_dev_t* dev,
mio_syshnd_t in_fd,
mio_foff_t foff,
mio_iolen_t len,
void* wrctx
);
MIO_EXPORT int mio_dev_timedwrite (
mio_dev_t* dev,
const void* data,
@ -918,6 +927,14 @@ MIO_EXPORT int mio_dev_timedwritev (
const mio_devaddr_t* dstaddr
);
MIO_EXPORT int mio_dev_timedsendfile (
mio_dev_t* dev,
mio_syshnd_t in_fd,
mio_foff_t foff,
mio_iolen_t len,
const mio_ntime_t* tmout,
void* wrctx
);
/* =========================================================================
* SERVICE
* ========================================================================= */

8
mio/lib/pro.c
View File

@ -732,9 +732,10 @@ static mio_dev_mth_t dev_pro_methods =
dev_pro_kill_master,
dev_pro_getsyshnd,
MIO_NULL,
MIO_NULL,
MIO_NULL,
MIO_NULL, /* read */
MIO_NULL, /* write */
MIO_NULL, /* writev */
MIO_NULL, /* sendfile */
dev_pro_ioctl
};
@ -747,6 +748,7 @@ static mio_dev_mth_t dev_pro_methods_slave =
dev_pro_read_slave,
dev_pro_write_slave,
dev_pro_writev_slave,
MIO_NULL, /* sendfile */
dev_pro_ioctl
};

93
mio/lib/sck.c
View File

@ -44,6 +44,9 @@
# include <net/if_dl.h>
#endif
#if defined(HAVE_SYS_SENDFILE_H)
# include <sys/sendfile.h>
#endif
#if defined(__linux__)
# include <limits.h>
@ -409,6 +412,7 @@ static mio_syshnd_t dev_sck_getsyshnd (mio_dev_t* dev)
mio_dev_sck_t* rdev = (mio_dev_sck_t*)dev;
return (mio_syshnd_t)rdev->hnd;
}
/* ------------------------------------------------------------------------------ */
static int dev_sck_read_stateful (mio_dev_t* dev, void* buf, mio_iolen_t* len, mio_devaddr_t* srcaddr)
{
@ -476,6 +480,7 @@ static int dev_sck_read_stateless (mio_dev_t* dev, void* buf, mio_iolen_t* len,
return 1;
}
/* ------------------------------------------------------------------------------ */
static int dev_sck_write_stateful (mio_dev_t* dev, const void* data, mio_iolen_t* len, const mio_devaddr_t* dstaddr)
{
@ -643,6 +648,8 @@ static int dev_sck_writev_stateful (mio_dev_t* dev, const mio_iovec_t* iov, mio_
return 1;
}
/* ------------------------------------------------------------------------------ */
static int dev_sck_write_stateless (mio_dev_t* dev, const void* data, mio_iolen_t* len, const mio_devaddr_t* dstaddr)
{
mio_t* mio = dev->mio;
@ -691,6 +698,88 @@ static int dev_sck_writev_stateless (mio_dev_t* dev, const mio_iovec_t* iov, mio
return 1;
}
/* ------------------------------------------------------------------------------ */
static int dev_sck_sendfile_stateful (mio_dev_t* dev, mio_syshnd_t in_fd, mio_foff_t foff, mio_iolen_t* len)
{
mio_t* mio = dev->mio;
mio_dev_sck_t* rdev = (mio_dev_sck_t*)dev;
#if 0 && defined(USE_SSL)
/* TODO: ssl needs to read from the file... and send... */
if (rdev->ssl)
{
int x;
if (*len <= 0)
{
/* it's a writing finish indicator. close the writing end of
* the socket, probably leaving it in the half-closed state */
if ((x = SSL_shutdown((SSL*)rdev->ssl)) == -1)
{
set_ssl_error (mio, SSL_get_error((SSL*)rdev->ssl, x));
return -1;
}
return 1;
}
x = SSL_write((SSL*)rdev->ssl, data, *len);
if (x <= -1)
{
int err = SSL_get_error ((SSL*)rdev->ssl, x);
if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) return 0;
set_ssl_error (mio, err);
return -1;
}
*len = x;
}
else
{
#endif
ssize_t x;
if (*len <= 0)
{
/* the write handler for a stream device must handle a zero-length
* writing request specially. it's a writing finish indicator. close
* the writing end of the socket, probably leaving it in the half-closed state */
if (shutdown(rdev->hnd, SHUT_WR) == -1)
{
mio_seterrwithsyserr (mio, 0, errno);
return -1;
}
/* it must return a non-zero positive value. if it returns 0, this request
* gets enqueued by the core. we must aovid it */
return 1;
}
#if defined(HAVE_SENDFILE)
/* TODO: cater for other systems */
x = sendfile(rdev->hnd, in_fd, &foff, *len);
if (x == -1)
{
if (errno == EINPROGRESS || errno == EWOULDBLOCK || errno == EAGAIN) return 0; /* no data can be written */
if (errno == EINTR) return 0;
mio_seterrwithsyserr (mio, 0, errno);
return -1;
}
*len = x;
#else
mio_seterrnum (mio, MIO_ENOIMPL);
return -1;
#endif
#if 0 && defined(USE_SSL)
}
#endif
return 1;
}
/* ------------------------------------------------------------------------------ */
#if defined(USE_SSL)
static int do_ssl (mio_dev_sck_t* dev, int (*ssl_func)(SSL*))
@ -1124,6 +1213,7 @@ static mio_dev_mth_t dev_sck_methods_stateless =
dev_sck_read_stateless,
dev_sck_write_stateless,
dev_sck_writev_stateless,
MIO_NULL, /* sendfile */
dev_sck_ioctl, /* ioctl */
};
@ -1137,6 +1227,7 @@ static mio_dev_mth_t dev_sck_methods_stateful =
dev_sck_read_stateful,
dev_sck_write_stateful,
dev_sck_writev_stateful,
dev_sck_sendfile_stateful,
dev_sck_ioctl, /* ioctl */
};
@ -1149,6 +1240,7 @@ static mio_dev_mth_t dev_mth_clisck =
dev_sck_read_stateful,
dev_sck_write_stateful,
dev_sck_writev_stateful,
dev_sck_sendfile_stateful,
dev_sck_ioctl
};
/* ========================================================================= */
@ -1710,7 +1802,6 @@ int mio_dev_sck_timedwritev (mio_dev_sck_t* dev, mio_iovec_t* iov, mio_iolen_t i
return mio_dev_timedwritev((mio_dev_t*)dev, iov, iovcnt, tmout, wrctx, skad_to_devaddr(dev, dstaddr, &devaddr));
}
/* ========================================================================= */
int mio_dev_sck_setsockopt (mio_dev_sck_t* dev, int level, int optname, void* optval, mio_scklen_t optlen)
{