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

added stio_ruindev() and improved half-open state handling for a tcp device

Browse Source
This commit is contained in:
hyung-hwan 2016-02-02 07:51:17 +00:00
parent cccb7bee65
commit 7f47a92573
5 changed files with 263 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
stio/lib/main.c
View File

@ -107,7 +107,14 @@ static int tcp_on_sent (stio_dev_tcp_t* tcp, void* sendctx)
static int tcp_on_recv (stio_dev_tcp_t* tcp, const void* buf, stio_len_t len) static int tcp_on_recv (stio_dev_tcp_t* tcp, const void* buf, stio_len_t len)
{ {
return stio_dev_tcp_send (tcp, "HELLO", 5, STIO_NULL); int n;
static char a ='A';
char* xxx = malloc (1000000);
memset (xxx, a++ ,1000000);
//return stio_dev_tcp_send (tcp, "HELLO", 5, STIO_NULL);
n = stio_dev_tcp_send (tcp, xxx, 1000000, STIO_NULL);
free (xxx);
return n;
} }
static stio_t* g_stio; static stio_t* g_stio;
@ -143,13 +150,13 @@ int main ()
sigact.sa_handler = handle_signal; sigact.sa_handler = handle_signal;
sigaction (SIGINT, &sigact, STIO_NULL); sigaction (SIGINT, &sigact, STIO_NULL);
//STIO_MEMSET (&sigact, 0, STIO_SIZEOF(sigact)); memset (&sigact, 0, STIO_SIZEOF(sigact));
//sigact.sa_handler = SIG_IGN; sigact.sa_handler = SIG_IGN;
//sigaction (SIGPIPE, &sigact, STIO_NULL); sigaction (SIGPIPE, &sigact, STIO_NULL);
memset (&sin, 0, STIO_SIZEOF(sin)); /*memset (&sin, 0, STIO_SIZEOF(sin));
sin.sin_family = AF_INET; sin.sin_family = AF_INET;
sin.sin_port = htons(1234); sin.sin_port = htons(1234); */
/* /*
udp = (stio_dev_udp_t*)stio_makedev (stio, STIO_SIZEOF(*udp), &udp_mth, &udp_evcb, &sin); udp = (stio_dev_udp_t*)stio_makedev (stio, STIO_SIZEOF(*udp), &udp_mth, &udp_evcb, &sin);
if (!udp) if (!udp)

4
stio/lib/stio-prv.h
View File

@ -68,7 +68,9 @@ struct stio_t
{ {
stio_dev_t* head; stio_dev_t* head;
stio_dev_t* tail; stio_dev_t* tail;
} dev; } dev; /* normal devices */
stio_dev_t* rdev; /* ruined device list - singly linked list */
stio_uint8_t bigbuf[65535]; /* TODO: make this dynamic depending on devices added. device may indicate a buffer size required??? */ stio_uint8_t bigbuf[65535]; /* TODO: make this dynamic depending on devices added. device may indicate a buffer size required??? */

15
stio/lib/stio-tcp.c
View File

@ -165,10 +165,6 @@ static void tmr_connect_handle (stio_t* stio, const stio_ntime_t* now, stio_tmrj
* doesn't need to be deleted when it gets connected for this check * doesn't need to be deleted when it gets connected for this check
* here. this libarary, however, deletes the job when it gets * here. this libarary, however, deletes the job when it gets
* connected. */ * connected. */
/*
if (tcp->on_disconnected) tcp->on_disconnected (tcp);
tcp->state &= ~STIO_DEV_TCP_CONNECTING;
*/
stio_dev_tcp_kill (tcp); stio_dev_tcp_kill (tcp);
} }
} }
@ -241,7 +237,7 @@ static int tcp_ioctl (stio_dev_t* dev, int cmd, void* arg)
{ {
if (errno == EINPROGRESS || errno == EWOULDBLOCK) if (errno == EINPROGRESS || errno == EWOULDBLOCK)
{ {
if (stio_dev_event ((stio_dev_t*)tcp, STIO_DEV_EVENT_UPD, STIO_DEV_EVENT_IN | STIO_DEV_EVENT_OUT) >= 0) if (stio_dev_watch ((stio_dev_t*)tcp, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_IN | STIO_DEV_EVENT_OUT) >= 0)
{ {
stio_tmrjob_t tmrjob; stio_tmrjob_t tmrjob;
@ -262,7 +258,7 @@ static int tcp_ioctl (stio_dev_t* dev, int cmd, void* arg)
tcp->tmridx_connect = stio_instmrjob (tcp->stio, &tmrjob); tcp->tmridx_connect = stio_instmrjob (tcp->stio, &tmrjob);
if (tcp->tmridx_connect == STIO_TMRIDX_INVALID) if (tcp->tmridx_connect == STIO_TMRIDX_INVALID)
{ {
stio_dev_event ((stio_dev_t*)tcp, STIO_DEV_EVENT_UPD, STIO_DEV_EVENT_IN); stio_dev_watch ((stio_dev_t*)tcp, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_IN);
/* event manipulation failure can't be handled properly. so ignore it */ /* event manipulation failure can't be handled properly. so ignore it */
return -1; return -1;
} }
@ -386,7 +382,6 @@ printf ("TCP READY...%p\n", dev);
STIO_ASSERT (!(tcp->state & STIO_DEV_TCP_CONNECTED)); STIO_ASSERT (!(tcp->state & STIO_DEV_TCP_CONNECTED));
printf ("XXXXXXXXXXXXXXX CONNECTED...\n");
len = STIO_SIZEOF(errcode); len = STIO_SIZEOF(errcode);
if (getsockopt (tcp->sck, SOL_SOCKET, SO_ERROR, (char*)&errcode, &len) == -1) if (getsockopt (tcp->sck, SOL_SOCKET, SO_ERROR, (char*)&errcode, &len) == -1)
{ {
@ -398,9 +393,9 @@ printf ("XXXXXXXXXXXXXXX CONNECTED...\n");
tcp->state &= ~STIO_DEV_TCP_CONNECTING; tcp->state &= ~STIO_DEV_TCP_CONNECTING;
tcp->state |= STIO_DEV_TCP_CONNECTED; tcp->state |= STIO_DEV_TCP_CONNECTED;
if (stio_dev_event ((stio_dev_t*)tcp, STIO_DEV_EVENT_UPD, STIO_DEV_EVENT_IN) <= -1) if (stio_dev_watch ((stio_dev_t*)tcp, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_IN) <= -1)
{ {
printf ("CAANOT MANIPULTE EVENT ...\n"); printf ("CAANOT MANIPULTE WATCHER ...\n");
return -1; return -1;
} }
@ -464,7 +459,7 @@ printf ("CAANOT MANIPULTE EVENT ...\n");
/* addr is the address of the peer */ /* addr is the address of the peer */
/* local addresss is inherited from the server */ /* local addresss is inherited from the server */
clitcp = (stio_dev_tcp_t*)stio_makedev (tcp->stio, STIO_SIZEOF(*tcp), &tcp_acc_mth, tcp->evcb, &clisck); clitcp = (stio_dev_tcp_t*)stio_makedev (tcp->stio, STIO_SIZEOF(*tcp), &tcp_acc_mth, tcp->dev_evcb, &clisck);
if (!clitcp) if (!clitcp)
{ {
close (clisck); close (clisck);

274
stio/lib/stio.c
View File

@ -123,43 +123,41 @@ stio_dev_t* stio_makedev (stio_t* stio, stio_size_t dev_size, stio_dev_mth_t* de
STIO_MEMSET (dev, 0, dev_size); STIO_MEMSET (dev, 0, dev_size);
dev->stio = stio; dev->stio = stio;
dev->mth = dev_mth; /* default capability. dev->dev_mth->make() can change this.
dev->evcb = dev_evcb; * stio_dev_watch() is affected by the capability change. */
dev->dev_capa = STIO_DEV_CAPA_IN | STIO_DEV_CAPA_OUT;
dev->dev_mth = dev_mth;
dev->dev_evcb = dev_evcb;
STIO_WQ_INIT(&dev->wq); STIO_WQ_INIT(&dev->wq);
/* call the callback function first */ /* call the callback function first */
stio->errnum = STIO_ENOERR; stio->errnum = STIO_ENOERR;
if (dev->mth->make (dev, make_ctx) <= -1) if (dev->dev_mth->make (dev, make_ctx) <= -1)
{ {
if (stio->errnum == STIO_ENOERR) stio->errnum = STIO_EDEVMAKE; if (stio->errnum == STIO_ENOERR) stio->errnum = STIO_EDEVMAKE;
goto oops; goto oops;
} }
/* ------------------------------------ */ #if defined(_WIN32)
{ if (CreateIoCompletionPort ((HANDLE)dev->dev_mth->getsyshnd(dev), stio->iocp, STIO_IOCP_KEY, 0) == NULL)
#if defined(_WIN32)
if (CreateIoCompletionPort ((HANDLE)dev->mth->getsyshnd(dev), stio->iocp, STIO_IOCP_KEY, 0) == NULL)
{ {
/* TODO: set errnum from GetLastError()... */ /* TODO: set errnum from GetLastError()... */
goto oops_after_make; goto oops_after_make;
} }
#else
#else if (stio_dev_watch (dev, STIO_DEV_WATCH_START, STIO_DEV_EVENT_IN) <= -1) goto oops_after_make;
if (stio_dev_event (dev, STIO_DEV_EVENT_ADD, STIO_DEV_EVENT_IN) <= -1) goto oops_after_make; #endif
#endif
}
/* ------------------------------------ */
/* and place the new dev at the back */ /* and place the new dev at the back */
if (stio->dev.tail) stio->dev.tail->next = dev; if (stio->dev.tail) stio->dev.tail->dev_next = dev;
else stio->dev.head = dev; else stio->dev.head = dev;
dev->prev = stio->dev.tail; dev->dev_prev = stio->dev.tail;
stio->dev.tail = dev; stio->dev.tail = dev;
return dev; return dev;
oops_after_make: oops_after_make:
dev->mth->kill (dev); dev->dev_mth->kill (dev);
oops: oops:
STIO_MMGR_FREE (stio->mmgr, dev); STIO_MMGR_FREE (stio->mmgr, dev);
return STIO_NULL; return STIO_NULL;
@ -175,31 +173,56 @@ void stio_killdev (stio_t* stio, stio_dev_t* dev)
stio_wq_t* wq; stio_wq_t* wq;
wq = STIO_WQ_HEAD(&dev->wq); wq = STIO_WQ_HEAD(&dev->wq);
printf ("DELETING UNSENT REQUETS...%p\n", wq);
STIO_WQ_DEQ (&dev->wq); STIO_WQ_DEQ (&dev->wq);
STIO_MMGR_FREE (stio->mmgr, wq); STIO_MMGR_FREE (stio->mmgr, wq);
} }
/* delink the dev object */ /* delink the dev object */
if (dev->prev) if (!(dev->dev_capa & STIO_DEV_CAPA_RUINED))
dev->prev->next = dev->next; {
if (dev->dev_prev)
dev->dev_prev->dev_next = dev->dev_next;
else else
stio->dev.head = dev->next; stio->dev.head = dev->dev_next;
if (dev->next) if (dev->dev_next)
dev->next->prev = dev->prev; dev->dev_next->dev_prev = dev->dev_prev;
else else
stio->dev.tail = dev->prev; stio->dev.tail = dev->dev_prev;
}
stio_dev_event (dev, STIO_DEV_EVENT_DEL, 0); stio_dev_watch (dev, STIO_DEV_WATCH_STOP, 0);
/* and call the callback function */ /* and call the callback function */
dev->mth->kill (dev); dev->dev_mth->kill (dev);
STIO_MMGR_FREE (stio->mmgr, dev); STIO_MMGR_FREE (stio->mmgr, dev);
} }
void stio_ruindev (stio_t* stio, stio_dev_t* dev)
{
if (!(dev->dev_capa & STIO_DEV_CAPA_RUINED))
{
/* delink the dev object from the device list */
if (dev->dev_prev)
dev->dev_prev->dev_next = dev->dev_next;
else
stio->dev.head = dev->dev_next;
if (dev->dev_next)
dev->dev_next->dev_prev = dev->dev_prev;
else
stio->dev.tail = dev->dev_prev;
/* place it at the beginning of the ruined device list */
dev->dev_prev = STIO_NULL;
dev->dev_next = stio->rdev;
stio->rdev = dev;
dev->dev_capa |= STIO_DEV_CAPA_RUINED;
}
}
int stio_prologue (stio_t* stio) int stio_prologue (stio_t* stio)
{ {
/* TODO: */ /* TODO: */
@ -215,6 +238,7 @@ int stio_exec (stio_t* stio)
{ {
stio_ntime_t tmout; stio_ntime_t tmout;
#if defined(_WIN32) #if defined(_WIN32)
ULONG nentries, i; ULONG nentries, i;
#else #else
@ -260,31 +284,42 @@ int stio_exec (stio_t* stio)
/* TODO: merge events??? for the same descriptor */ /* TODO: merge events??? for the same descriptor */
for (i = 0; i < nentries; i++) for (i = 0; i < nentries; i++)
{ {
stio_dev_t* dev = stio->revs[i].data.ptr; stio_dev_t* dev;
if (dev->evcb->ready) dev = stio->revs[i].data.ptr;
if (dev->dev_evcb->ready)
{ {
int x, events = 0; int x, events = 0;
if (stio->revs[i].events & EPOLLERR) events |= STIO_DEV_EVENT_ERR; if (stio->revs[i].events & EPOLLERR) events |= STIO_DEV_EVENT_ERR;
if (stio->revs[i].events & EPOLLHUP) events |= STIO_DEV_EVENT_HUP; if (stio->revs[i].events & EPOLLHUP) events |= STIO_DEV_EVENT_HUP;
#if defined(EPOLLRDHUP)
/* treat it the same way as EPOLLHUP */
if (stio->revs[i].events & EPOLLRDHUP) events |= STIO_DEV_EVENT_HUP;
#endif
if (stio->revs[i].events & EPOLLIN) events |= STIO_DEV_EVENT_IN; if (stio->revs[i].events & EPOLLIN) events |= STIO_DEV_EVENT_IN;
if (stio->revs[i].events & EPOLLOUT) events |= STIO_DEV_EVENT_OUT; if (stio->revs[i].events & EPOLLOUT) events |= STIO_DEV_EVENT_OUT;
if (stio->revs[i].events & EPOLLPRI) events |= STIO_DEV_EVENT_PRI; if (stio->revs[i].events & EPOLLPRI) events |= STIO_DEV_EVENT_PRI;
#if defined(EPOLLRDHUP)
/* interprete EPOLLRDHUP the same way as EPOLLHUP.
*
* when EPOLLRDHUP is set, EPOLLIN or EPOLLPRI or both are
* assumed to be set as EPOLLRDHUP is requested only if
* STIO_DEV_WATCH_IN is set for stio_dev_watch().
* in linux, when EPOLLRDHUP is set, EPOLLIN is set together
* if it's requested together. it seems to be safe to have
* the following assertion. however, let me commect it out
* in case the assumption above is not right.
* STIO_ASSERT (events & (STIO_DEV_EVENT_IN | STIO_DEV_EVENT_PRI));
*/
if (stio->revs[i].events & EPOLLRDHUP) events |= STIO_DEV_EVENT_HUP;
#endif
/* return value of ready() /* return value of ready()
* <= -1 - failure. kill the device. * <= -1 - failure. kill the device.
* == 0 - ok. but don't invoke recv() or send(). * == 0 - ok. but don't invoke recv() or send().
* >= 1 - everything is ok. */ * >= 1 - everything is ok. */
/* TODO: can the revs array contain the same file descriptor again??? */ if ((x = dev->dev_evcb->ready (dev, events)) <= -1)
if ((x = dev->evcb->ready (dev, events)) <= -1)
{ {
stio_killdev (stio, dev); stio_ruindev (stio, dev);
dev = STIO_NULL; dev = STIO_NULL;
} }
else if (x >= 1) else if (x >= 1)
@ -294,14 +329,12 @@ int stio_exec (stio_t* stio)
} }
else else
{ {
int dev_eof;
invoke_evcb: invoke_evcb:
dev_eof = 0;
if (dev && stio->revs[i].events & EPOLLPRI) if (dev && stio->revs[i].events & EPOLLPRI)
{ {
/* urgent data */ /* urgent data */
/* TODO: urgent data.... */
/*x = dev->dev_mth->urgrecv (dev, stio->bugbuf, &len);*/
printf ("has urgent data...\n"); printf ("has urgent data...\n");
} }
@ -315,35 +348,55 @@ int stio_exec (stio_t* stio)
while (1) while (1)
{ {
len = STIO_COUNTOF(stio->bigbuf); len = STIO_COUNTOF(stio->bigbuf);
x = dev->mth->recv (dev, stio->bigbuf, &len); x = dev->dev_mth->recv (dev, stio->bigbuf, &len);
printf ("DATA...recv %d length %d\n", (int)x, len);
if (x <= -1) if (x <= -1)
{ {
/*TODO: what to do? killdev? how to indicate an error? call on_recv? with error indicator?? */ stio_ruindev (stio, dev);
stio_killdev (stio, dev);
dev = STIO_NULL; dev = STIO_NULL;
break; break;
} }
else if (x == 0) else if (x == 0)
{ {
/* no data is available */ /* no data is available - EWOULDBLOCK or something similar */
break; break;
} }
else if (x >= 1) else if (x >= 1)
{ {
if (len <= 0) if (len <= 0)
{ {
/* EOF received. delay killing until output has been handled */ /* EOF received. delay killing until output has been handled. */
dev_eof = 1; if (STIO_WQ_ISEMPTY(&dev->wq))
{
/* no pending writes - ruin the device to kill it eventually */
stio_ruindev (stio, dev);
/* don't set 'dev' to STIO_NULL so that
* output handling can kick in if EPOLLOUT is set */
}
else
{
/* it might be in a half-open state */
dev->dev_capa &= ~(STIO_DEV_CAPA_IN | STIO_DEV_CAPA_PRI);
/* disable the input watching */
if (stio_dev_watch (dev, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_OUT) <= -1)
{
stio_ruindev (stio, dev);
dev = STIO_NULL;
}
}
break; break;
} }
else else
{ {
/* TODO: for a stream device, merge received data if bigbuf isn't full and fire the on_recv callback
* when x == 0 or <= -1. you can */
/* data available */ /* data available */
if (dev->evcb->on_recv (dev, stio->bigbuf, len) <= -1) if (dev->dev_evcb->on_recv (dev, stio->bigbuf, len) <= -1)
{ {
stio_killdev (stio, dev); stio_ruindev (stio, dev);
dev = STIO_NULL; dev = STIO_NULL;
break; break;
} }
@ -368,7 +421,7 @@ int stio_exec (stio_t* stio)
send_leftover: send_leftover:
ulen = urem; ulen = urem;
x = dev->mth->send (dev, uptr, &ulen); x = dev->dev_mth->send (dev, uptr, &ulen);
if (x <= -1) if (x <= -1)
{ {
/* TODO: error handling? call callback? or what? */ /* TODO: error handling? call callback? or what? */
@ -393,7 +446,7 @@ int stio_exec (stio_t* stio)
int y; int y;
STIO_WQ_UNLINK (q); /* STIO_WQ_DEQ(&dev->wq); */ STIO_WQ_UNLINK (q); /* STIO_WQ_DEQ(&dev->wq); */
y = dev->evcb->on_sent (dev, q->ctx); y = dev->dev_evcb->on_sent (dev, q->ctx);
STIO_MMGR_FREE (dev->stio->mmgr, q); STIO_MMGR_FREE (dev->stio->mmgr, q);
if (y <= -1) if (y <= -1)
@ -411,22 +464,33 @@ int stio_exec (stio_t* stio)
{ {
/* no pending request to write. /* no pending request to write.
* watch input only. disable output watching */ * watch input only. disable output watching */
if (stio_dev_event (dev, STIO_DEV_EVENT_UPD, STIO_DEV_EVENT_IN) <= -1) if (dev->dev_capa & STIO_DEV_CAPA_IN)
{ {
/* TODO: call an error handler??? */ if (stio_dev_watch (dev, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_IN) <= -1)
stio_killdev (stio, dev); {
stio_ruindev (stio, dev);
dev = STIO_NULL; dev = STIO_NULL;
} }
} }
else
{
/* the device is not capable of reading.
* finish the device */
stio_ruindev (stio, dev);
dev = STIO_NULL;
}
}
}
}
} }
if (dev && dev_eof) /* kill all ruined devices */
while (stio->rdev)
{ {
/* handled delayed device killing */ stio_dev_t* next;
stio_killdev (stio, dev); next = stio->rdev->dev_next;
dev = STIO_NULL; stio_killdev (stio, stio->rdev);
} stio->rdev = next;
}
} }
#endif #endif
@ -456,44 +520,58 @@ int stio_loop (stio_t* stio)
return 0; return 0;
} }
int stio_dev_ioctl (stio_dev_t* dev, int cmd, void* arg) int stio_dev_ioctl (stio_dev_t* dev, int cmd, void* arg)
{ {
if (dev->mth->ioctl) return dev->mth->ioctl (dev, cmd, arg); if (dev->dev_mth->ioctl) return dev->dev_mth->ioctl (dev, cmd, arg);
dev->stio->errnum = STIO_ENOSUP; /* TODO: different error code ? */ dev->stio->errnum = STIO_ENOSUP; /* TODO: different error code ? */
return -1; return -1;
} }
int stio_dev_event (stio_dev_t* dev, stio_dev_event_cmd_t cmd, int flags) int stio_dev_watch (stio_dev_t* dev, stio_dev_watch_cmd_t cmd, int events)
{ {
#if defined(_WIN32)
/* TODO */
#else
struct epoll_event ev; struct epoll_event ev;
int epoll_op; int epoll_op;
/* this function honors STIO_DEV_EVENT_IN and STIO_DEV_EVENT_OUT only
* as valid input event bits. it intends to provide simple abstraction
* by reducing the variety of event bits that the caller has to handle. */
ev.events = EPOLLHUP | EPOLLERR; ev.events = EPOLLHUP | EPOLLERR;
#if defined(EPOLLRDHUP)
if (events & STIO_DEV_EVENT_IN)
{
if (dev->dev_capa & STIO_DEV_CAPA_IN)
{
ev.events |= EPOLLIN;
#if defined(EPOLLRDHUP)
ev.events |= EPOLLRDHUP; ev.events |= EPOLLRDHUP;
#endif #endif
}
if (dev->dev_capa & STIO_DEV_CAPA_PRI)
{
ev.events |= EPOLLPRI;
#if defined(EPOLLRDHUP)
ev.events |= EPOLLRDHUP;
#endif
}
}
if (events & STIO_DEV_EVENT_OUT)
{
if (dev->dev_capa & STIO_DEV_CAPA_OUT) ev.events |= EPOLLOUT;
}
if (flags & STIO_DEV_EVENT_IN) ev.events |= EPOLLIN;
if (flags & STIO_DEV_EVENT_OUT) ev.events |= EPOLLOUT;
if (flags & STIO_DEV_EVENT_PRI) ev.events |= EPOLLPRI;
ev.data.ptr = dev; ev.data.ptr = dev;
switch (cmd) switch (cmd)
{ {
case STIO_DEV_EVENT_ADD: case STIO_DEV_WATCH_START:
epoll_op = EPOLL_CTL_ADD; epoll_op = EPOLL_CTL_ADD;
break; break;
case STIO_DEV_EVENT_UPD: case STIO_DEV_WATCH_UPDATE:
epoll_op = EPOLL_CTL_MOD; epoll_op = EPOLL_CTL_MOD;
break; break;
case STIO_DEV_EVENT_DEL: case STIO_DEV_WATCH_STOP:
epoll_op = EPOLL_CTL_DEL; epoll_op = EPOLL_CTL_DEL;
break; break;
@ -502,41 +580,52 @@ int stio_dev_event (stio_dev_t* dev, stio_dev_event_cmd_t cmd, int flags)
return -1; return -1;
} }
if (epoll_ctl (dev->stio->mux, epoll_op, dev->mth->getsyshnd(dev), &ev) == -1) if (epoll_ctl (dev->stio->mux, epoll_op, dev->dev_mth->getsyshnd(dev), &ev) == -1)
{ {
dev->stio->errnum = stio_syserrtoerrnum(errno); dev->stio->errnum = stio_syserrtoerrnum(errno);
return -1; return -1;
} }
return 0; return 0;
#endif
} }
int stio_dev_send (stio_dev_t* dev, const void* data, stio_len_t len, void* sendctx) int stio_dev_send (stio_dev_t* dev, const void* data, stio_len_t len, void* sendctx)
{ {
const stio_uint8_t* uptr; const stio_uint8_t* uptr;
stio_len_t urem, ulen; stio_len_t urem, ulen;
int x; stio_wq_t* q;
int x, wq_empty;
if (!(dev->dev_capa & STIO_DEV_CAPA_OUT))
{
dev->stio->errnum = STIO_ENOCAPA;
return -1;
}
uptr = data; uptr = data;
urem = len; urem = len;
wq_empty = STIO_WQ_ISEMPTY(&dev->wq);
if (!wq_empty) goto enqueue_data;
while (urem > 0) while (urem > 0)
{ {
ulen = urem; ulen = urem;
x = dev->mth->send (dev, data, &ulen); x = dev->dev_mth->send (dev, data, &ulen);
if (x <= -1) if (x <= -1) return -1;
else if (x == 0) goto enqueue_data; /* enqueue remaining data */
else
{ {
return -1; urem -= ulen;
uptr += ulen;
}
} }
else if (x == 0)
{
stio_wq_t* q;
int wq_empty;
/* queue the uremaining data*/ dev->dev_evcb->on_sent (dev, sendctx);
wq_empty = STIO_WQ_ISEMPTY(&dev->wq); return 0;
enqueue_data:
/* queue the remaining data*/
q = (stio_wq_t*)STIO_MMGR_ALLOC (dev->stio->mmgr, STIO_SIZEOF(*q) + urem); q = (stio_wq_t*)STIO_MMGR_ALLOC (dev->stio->mmgr, STIO_SIZEOF(*q) + urem);
if (!q) if (!q)
{ {
@ -552,7 +641,7 @@ int stio_dev_send (stio_dev_t* dev, const void* data, stio_len_t len, void* send
STIO_WQ_ENQ (&dev->wq, q); STIO_WQ_ENQ (&dev->wq, q);
if (wq_empty) if (wq_empty)
{ {
if (stio_dev_event (dev, STIO_DEV_EVENT_UPD, STIO_DEV_EVENT_IN | STIO_DEV_EVENT_OUT) <= -1) if (stio_dev_watch (dev, STIO_DEV_WATCH_UPDATE, STIO_DEV_EVENT_IN | STIO_DEV_EVENT_OUT) <= -1)
{ {
STIO_WQ_UNLINK (q); /* unlink the ENQed item */ STIO_WQ_UNLINK (q); /* unlink the ENQed item */
STIO_MMGR_FREE (dev->stio->mmgr, q); STIO_MMGR_FREE (dev->stio->mmgr, q);
@ -560,20 +649,9 @@ int stio_dev_send (stio_dev_t* dev, const void* data, stio_len_t len, void* send
} }
} }
return 0;
}
else
{
urem -= ulen;
uptr += ulen;
}
}
dev->evcb->on_sent (dev, sendctx);
return 0; return 0;
} }
stio_errnum_t stio_syserrtoerrnum (int no) stio_errnum_t stio_syserrtoerrnum (int no)
{ {
switch (no) switch (no)

57
stio/lib/stio.h
View File

@ -107,6 +107,7 @@ enum stio_errnum_t
STIO_ENFILE, STIO_ENFILE,
STIO_ECONRF, /* connection refused */ STIO_ECONRF, /* connection refused */
STIO_ECONRS, /* connection reset */ STIO_ECONRS, /* connection reset */
STIO_ENOCAPA, /* no capability */
STIO_EDEVMAKE, STIO_EDEVMAKE,
STIO_EDEVERR, STIO_EDEVERR,
@ -127,12 +128,6 @@ struct stio_dev_mth_t
void (*kill) (stio_dev_t* dev); /* mandatory. called in stio_killdev(). called in stio_makedev() upon failure after make() success */ void (*kill) (stio_dev_t* dev); /* mandatory. called in stio_killdev(). called in stio_makedev() upon failure after make() success */
/* ------------------------------------------------------------------ */ /* ------------------------------------------------------------------ */
#if 0
/* TODO: countsyshnds() if the device has multiple handles.
* getsyshnd() to accept the handle id between 0 and countsysnhnds() - 1
*/
int (*countsyshnds) (stio_dev_t* dev); /* optional */
#endif
stio_syshnd_t (*getsyshnd) (stio_dev_t* dev); /* mandatory. called in stio_makedev() after successful make() */ stio_syshnd_t (*getsyshnd) (stio_dev_t* dev); /* mandatory. called in stio_makedev() after successful make() */
/* ------------------------------------------------------------------ */ /* ------------------------------------------------------------------ */
@ -217,34 +212,48 @@ struct stio_wq_t
#define STIO_DEV_HEADERS \ #define STIO_DEV_HEADERS \
stio_t* stio; \ stio_t* stio; \
stio_dev_mth_t* mth; \ int dev_capa; \
stio_dev_evcb_t* evcb; \ stio_dev_mth_t* dev_mth; \
stio_dev_evcb_t* dev_evcb; \
stio_wq_t wq; \ stio_wq_t wq; \
stio_dev_t* prev; \ stio_dev_t* dev_prev; \
stio_dev_t* next stio_dev_t* dev_next
struct stio_dev_t struct stio_dev_t
{ {
STIO_DEV_HEADERS; STIO_DEV_HEADERS;
}; };
enum stio_dev_event_cmd_t enum stio_dev_capa_t
{ {
STIO_DEV_EVENT_ADD, STIO_DEV_CAPA_IN = (1 << 0),
STIO_DEV_EVENT_UPD, STIO_DEV_CAPA_OUT = (1 << 1),
STIO_DEV_EVENT_DEL STIO_DEV_CAPA_PRI = (1 << 2),
}; STIO_DEV_CAPA_STREAM = (1 << 3),
typedef enum stio_dev_event_cmd_t stio_dev_event_cmd_t;
enum stio_dev_event_flag_t /* internal use only. never set this bit to the dev_capa field */
STIO_DEV_CAPA_RUINED = (1 << 15)
};
typedef enum stio_dev_capa_t stio_dev_capa_t;
enum stio_dev_watch_cmd_t
{
STIO_DEV_WATCH_START,
STIO_DEV_WATCH_UPDATE,
STIO_DEV_WATCH_STOP
};
typedef enum stio_dev_watch_cmd_t stio_dev_watch_cmd_t;
enum stio_dev_event_t
{ {
STIO_DEV_EVENT_IN = (1 << 0), STIO_DEV_EVENT_IN = (1 << 0),
STIO_DEV_EVENT_OUT = (1 << 1), STIO_DEV_EVENT_OUT = (1 << 1),
STIO_DEV_EVENT_PRI = (1 << 2), STIO_DEV_EVENT_PRI = (1 << 2),
STIO_DEV_EVENT_HUP = (1 << 3), STIO_DEV_EVENT_HUP = (1 << 3),
STIO_DEV_EVENT_ERR = (1 << 4) STIO_DEV_EVENT_ERR = (1 << 4)
}; };
typedef enum stio_dev_event_flag_t stio_dev_event_flag_t; typedef enum stio_dev_event_t stio_dev_event_t;
typedef struct stio_tmrjob_t stio_tmrjob_t; typedef struct stio_tmrjob_t stio_tmrjob_t;
@ -313,16 +322,22 @@ STIO_EXPORT void stio_killdev (
stio_dev_t* dev stio_dev_t* dev
); );
STIO_EXPORT void stio_ruindev (
stio_t* stio,
stio_dev_t* dev
);
STIO_EXPORT int stio_dev_ioctl ( STIO_EXPORT int stio_dev_ioctl (
stio_dev_t* dev, stio_dev_t* dev,
int cmd, int cmd,
void* arg void* arg
); );
STIO_EXPORT int stio_dev_event ( STIO_EXPORT int stio_dev_watch (
stio_dev_t* dev, stio_dev_t* dev,
stio_dev_event_cmd_t cmd, stio_dev_watch_cmd_t cmd,
int flags /** 0 or bitwise-ORed of #STIO_DEV_EVENT_IN and #STIO_DEV_EVENT_OUT */
int events
); );
STIO_EXPORT int stio_dev_send ( STIO_EXPORT int stio_dev_send (