Compare commits

...
3 Commits
2 changed files with 115 additions and 20 deletions
+20
View File
@@ -457,6 +457,18 @@ struct hio_dev_sck_qxmsg_t
};
typedef struct hio_dev_sck_qxmsg_t hio_dev_sck_qxmsg_t;
/* the qx side channel is a socketpair. on most systems it is a datagram pair,
* where one write is one message and the reader is handed whole messages. where
* a datagram pair cannot carry messages reliably it is a stream pair instead,
* and the reader has to recover the message boundaries itself.
*
* the socket type, the writing side and the reader must agree on which of the
* two it is, so all three ask this one macro rather than testing for the system
* separately. */
#if defined(__BEOS__) || defined(__HAIKU__)
# define HIO_DEV_SCK_QX_STREAM
#endif
struct hio_dev_sck_t
{
HIO_DEV_HEADER;
@@ -512,8 +524,16 @@ struct hio_dev_sck_t
struct
{
hio_syshnd_t side_chan; /* side-channel for HIO_DEV_SCK_QX */
#if defined(HIO_DEV_SCK_QX_STREAM)
/* the message being reassembled, and how much of it has arrived.
* a stream side channel splits and merges writes freely, so a read
* can end in the middle of a message and the remainder has to be
* held until the rest comes. only the stream variant needs it, and
* the hio_skad_t inside the message makes it large, so it is not
* carried on the systems that never look at it. */
hio_dev_sck_qxmsg_t qxacc;
hio_oow_t qxacc_len;
#endif
} qx;
struct
+95 -20
View File
@@ -209,7 +209,11 @@ open_socket:
if (hio_makesyshndasync(hio, sck) <= -1 ||
hio_makesyshndcloexec(hio, sck) <= -1) goto oops;
/* the label is only jumped to from the branch that asks socket() for the flags
* directly, so it exists only where that branch does */
#if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
done:
#endif
return sck;
oops:
@@ -221,13 +225,11 @@ oops:
static hio_syshnd_t open_async_qx (hio_t* hio, hio_syshnd_t* side_chan)
{
int fd[2];
int type = SOCK_DGRAM;
#if defined(__BEOS__) || defined(__HAIKU__)
#if defined(HIO_DEV_SCK_QX_STREAM)
/* on haiku os r6beta, SOCK_DGRAM isn't reliable. more than one write() on it causes SIGPIPE */
type = SOCK_STREAM;
int type = SOCK_STREAM;
#else
type = SOCK_DGRAM;
int type = SOCK_DGRAM;
#endif
#if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
@@ -265,7 +267,11 @@ open_socket:
return HIO_SYSHND_INVALID;
}
/* the label is only jumped to from the branch that asks socketpair() for the
* flags directly, so it exists only where that branch does */
#if defined(SOCK_NONBLOCK) && defined(SOCK_CLOEXEC) && !(defined(__BEOS__) || defined(__HAIKU__))
done:
#endif
*side_chan = fd[1]; /* write end of the pipe */
return fd[0]; /* read end of the pipe */
}
@@ -340,7 +346,7 @@ static int dev_sck_read_bpf (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_de
{
hio_t* hio = dev->hio;
hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
struct bpf_hdr* bh;
hio_oow_t caplen;
@@ -400,7 +406,7 @@ static int dev_sck_read_bpf (hio_dev_t* dev, void* buf, hio_iolen_t* len, hio_de
static int dev_sck_readpending_bpf (hio_dev_t* dev)
{
hio_dev_sck_t* rdev = (hio_dev_sck_t*)dev;
bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
return st && st->pos < st->len;
}
@@ -928,9 +934,9 @@ static int dev_sck_kill (hio_dev_t* dev, int force)
}
#if defined(USE_BPF)
if (rdev->type == HIO_DEV_SCK_PACKET && rdev->u.bpf.state)
if (sck_type_map[rdev->type].domain == __AF_BPF && rdev->u.bpf.state)
{
bpf_state_t* st = (bpf_state_t*)rdev->bpf_state;
bpf_state_t* st = (bpf_state_t*)rdev->u.bpf.state;
if (st->buf) hio_freemem(hio, st->buf);
hio_freemem(hio, st);
rdev->u.bpf.state = HIO_NULL;
@@ -2114,7 +2120,7 @@ static int dev_sck_ioctl (hio_dev_t* dev, int cmd, void* arg)
}
#if defined(USE_BPF)
if (rdev->bpf_state)
if (sck_type_map[rdev->type].domain == __AF_BPF)
{
/* a bpf device attaches to an interface by name, and what the
* caller gave is the ifindex hio_skad_init_for_eth() puts in an
@@ -3165,9 +3171,27 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
if (rdev->type == HIO_DEV_SCK_QX)
{
#if defined(__BEOS__) || defined(__HAIKU__)
const hio_uint8_t* p = (const hio_uint8_t*)data;
hio_oow_t rem = (hio_oow_t)dlen;
#if defined(HIO_DEV_SCK_QX_STREAM)
const hio_uint8_t* p;
hio_oow_t rem;
#else
hio_dev_sck_qxmsg_t* qxmsg;
#endif
if (dlen <= 0)
{
/* the writing end of the side channel is gone. a qx device carries
* no stream capability bit, so the loop has nothing to recognise
* end-of-file by and would keep reading a handle that stays
* readable with nothing on it. report it as a read failure and let
* the device be halted. */
hio_seterrbfmt(hio, HIO_EINVAL, "qx side channel closed");
return -1;
}
#if defined(HIO_DEV_SCK_QX_STREAM)
p = (const hio_uint8_t*)data;
rem = (hio_oow_t)dlen;
while (rem > 0)
{
@@ -3179,7 +3203,6 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
p += take;
rem -= take;
if (rdev->u.qx.qxacc_len < HIO_SIZEOF(rdev->u.qx.qxacc)) break; /* wait for more */
rdev->u.qx.qxacc_len = 0;
@@ -3188,21 +3211,24 @@ static int dev_evcb_sck_on_read_qx (hio_dev_t* dev, const void* data, hio_iolen_
if (make_accepted_client_connection(rdev, rdev->u.qx.qxacc.syshnd, &rdev->u.qx.qxacc.remoteaddr, rdev->u.qx.qxacc.scktype) <= -1)
{
/*printf ("unable to accept new client connection %d\n", qxmsg->syshnd);*/
return (rdev->state & HIO_DEV_SCK_LENIENT)? 0: -1;
if (!(rdev->state & HIO_DEV_SCK_LENIENT)) return -1;
/* one read can carry several messages. the accumulator is
* empty again, so the framing is intact and the messages
* queued behind this one are still deliverable - returning
* here would throw them away along with their handles. */
}
}
else
{
/* the stream is framed by size alone, so a bad command means
* the two ends disagree about the message layout and nothing
* after this point can be trusted to be a message boundary. */
* after this point can be trusted to be a message boundary.
* there is no resynchronising from that, so the device goes. */
hio_seterrbfmt(hio, HIO_EINVAL, "wrong qx command code");
return 0;
return -1;
}
}
#else
hio_dev_sck_qxmsg_t* qxmsg;
if (dlen != HIO_SIZEOF(*qxmsg))
{
hio_seterrbfmt(hio, HIO_EINVAL, "wrong qx packet size");
@@ -3981,12 +4007,61 @@ int hio_dev_sck_sendfileok (hio_dev_sck_t* dev)
int hio_dev_sck_writetosidechan (hio_dev_sck_t* dev, const void* dptr, hio_oow_t dlen)
{
if (dev->type == HIO_DEV_SCK_QX && write(dev->u.qx.side_chan, dptr, dlen) <= -1)
if (dev->type != HIO_DEV_SCK_QX)
{
errno = EINVAL;
return -1;
}
#if defined(HIO_DEV_SCK_QX_STREAM)
/* the reader recovers message boundaries by counting octets, so a message
* left half-written would shift every message behind it and there is no
* recovering from that. a channel that has taken nothing yet is reported as
* a failure, the way a datagram channel reports a full socket buffer; once
* part of a message is out the channel is committed and the remainder is
* pushed out here. the reader drains this channel from the event loop, so
* the room it needs does come back. */
{
const hio_uint8_t* p = (const hio_uint8_t*)dptr;
hio_oow_t rem = dlen;
while (rem > 0)
{
ssize_t n = write(dev->u.qx.side_chan, p, rem);
if (n > 0)
{
p += n;
rem -= n;
}
else if (n == 0)
{
/* neither progress nor an error to wait on. it cannot happen on
* a socket, but looping on it would hang the caller. */
errno = EIO;
return -1;
}
else if (errno == EINTR ||
((errno == EAGAIN || errno == EWOULDBLOCK) && rem < dlen))
{
/* retry: interrupted, or out of room part-way through a message */
}
else
{
/* this doesn't set the error information on the main socket. if you may check errno, though */
/* TODO: make hio_seterrbfmt() thread safe and set the error information properly. still the caller may be in the thread-unsafe context */
return -1;
}
}
}
#else
if (write(dev->u.qx.side_chan, dptr, dlen) <= -1)
{
/* this doesn't set the error information on the main socket. if you may check errno, though */
/* TODO: make hio_seterrbfmt() thread safe and set the error information properly. still the caller may be in the thread-unsafe context */
return -1;
}
#endif
return 0;
}