ipa: Change ipa_msg_recv() to support partial receive
Currently ipa_msg_recv() fails, when messages are received partially. This patch provides a new function ipa_msg_recv_buffered() that uses an additional ** to a message buffer to store partial data. When this happens, -EAGAIN is returned. If NULL is used, the function behaves similar to ipa_msg_recv() and fails on partial read. In addition in case of errors the return value is now always -EXXX and the contents of errno is undefined. Note that this feature needs support by the calling code insofar that *tmp_msg must be set to NULL initially and it must be freed and set to NULL manually when the socket is closed. Note also that ipa_msg_recv() is then a wrapper around ipa_msg_recv_buffered() which mimics the old error behaviour by setting errno explicitely to -rc and returning -1 when an error has happened. Ticket: OW#728 Sponsored-by: On-Waves ehf
This commit is contained in:
parent
36106ae244
commit
98af3c3a6d
|
@ -1 +1,2 @@
|
|||
#library what description / commit summary line
|
||||
libosmoabis abi-change ipa: Change ipa_msg_recv() to support partial receive
|
||||
|
|
|
@ -109,6 +109,8 @@ struct e1inp_ts {
|
|||
struct osmo_fd fd;
|
||||
} rs232;
|
||||
} driver;
|
||||
|
||||
struct msgb *pending_msg;
|
||||
};
|
||||
|
||||
struct gsm_e1_subslot {
|
||||
|
|
|
@ -27,6 +27,7 @@ struct ipa_server_conn {
|
|||
int (*closed_cb)(struct ipa_server_conn *peer);
|
||||
int (*cb)(struct ipa_server_conn *peer, struct msgb *msg);
|
||||
void *data;
|
||||
struct msgb *pending_msg;
|
||||
};
|
||||
|
||||
struct ipa_server_conn *ipa_server_conn_create(void *ctx, struct ipa_server_link *link, int fd, int (*cb)(struct ipa_server_conn *peer, struct msgb *msg), int (*closed_cb)(struct ipa_server_conn *peer), void *data);
|
||||
|
@ -53,6 +54,7 @@ struct ipa_client_conn {
|
|||
int (*read_cb)(struct ipa_client_conn *link, struct msgb *msg);
|
||||
int (*write_cb)(struct ipa_client_conn *link);
|
||||
void *data;
|
||||
struct msgb *pending_msg;
|
||||
};
|
||||
|
||||
struct ipa_client_conn *ipa_client_conn_create(void *ctx, struct e1inp_ts *ts, int priv_nr, const char *addr, uint16_t port, void (*updown)(struct ipa_client_conn *link, int), int (*read_cb)(struct ipa_client_conn *link, struct msgb *msgb), int (*write_cb)(struct ipa_client_conn *link), void *data);
|
||||
|
@ -64,6 +66,7 @@ void ipa_client_conn_close(struct ipa_client_conn *link);
|
|||
void ipa_client_conn_send(struct ipa_client_conn *link, struct msgb *msg);
|
||||
|
||||
int ipa_msg_recv(int fd, struct msgb **rmsg);
|
||||
int ipa_msg_recv_buffered(int fd, struct msgb **rmsg, struct msgb **tmp_msg);
|
||||
|
||||
int ipaccess_rcvmsg_base(struct msgb *msg, struct osmo_fd *bfd);
|
||||
|
||||
|
|
151
src/input/ipa.c
151
src/input/ipa.c
|
@ -49,50 +49,130 @@ void ipa_msg_push_header(struct msgb *msg, uint8_t proto)
|
|||
|
||||
int ipa_msg_recv(int fd, struct msgb **rmsg)
|
||||
{
|
||||
struct msgb *msg;
|
||||
int rc = ipa_msg_recv_buffered(fd, rmsg, NULL);
|
||||
if (rc < 0) {
|
||||
errno = -rc;
|
||||
rc = -1;
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
int ipa_msg_recv_buffered(int fd, struct msgb **rmsg, struct msgb **tmp_msg)
|
||||
{
|
||||
struct msgb *msg = tmp_msg ? *tmp_msg : NULL;
|
||||
struct ipaccess_head *hh;
|
||||
int len, ret;
|
||||
int needed;
|
||||
|
||||
msg = ipa_msg_alloc(0);
|
||||
if (msg == NULL)
|
||||
return -ENOMEM;
|
||||
msg = ipa_msg_alloc(0);
|
||||
|
||||
/* first read our 3-byte header */
|
||||
hh = (struct ipaccess_head *) msg->data;
|
||||
ret = recv(fd, msg->data, sizeof(*hh), 0);
|
||||
if (ret <= 0) {
|
||||
msgb_free(msg);
|
||||
return ret;
|
||||
} else if (ret != sizeof(*hh)) {
|
||||
LOGP(DLINP, LOGL_ERROR, "too small message received\n");
|
||||
msgb_free(msg);
|
||||
return -EIO;
|
||||
if (msg == NULL) {
|
||||
ret = -ENOMEM;
|
||||
goto discard_msg;
|
||||
}
|
||||
msgb_put(msg, ret);
|
||||
|
||||
if (msg->l2h == NULL) {
|
||||
/* first read our 3-byte header */
|
||||
needed = sizeof(*hh) - msg->len;
|
||||
ret = recv(fd, msg->tail, needed, 0);
|
||||
if (ret == 0)
|
||||
goto discard_msg;
|
||||
|
||||
if (ret < 0) {
|
||||
if (errno == EAGAIN || errno == EINTR)
|
||||
ret = 0;
|
||||
else {
|
||||
ret = -errno;
|
||||
goto discard_msg;
|
||||
}
|
||||
}
|
||||
|
||||
msgb_put(msg, ret);
|
||||
|
||||
if (ret < needed) {
|
||||
if (msg->len == 0) {
|
||||
ret = -EAGAIN;
|
||||
goto discard_msg;
|
||||
}
|
||||
|
||||
LOGP(DLINP, LOGL_INFO,
|
||||
"Received part of IPA message header (%d/%d)\n",
|
||||
msg->len, sizeof(*hh));
|
||||
if (!tmp_msg) {
|
||||
ret = -EIO;
|
||||
goto discard_msg;
|
||||
}
|
||||
*tmp_msg = msg;
|
||||
return -EAGAIN;
|
||||
}
|
||||
|
||||
msg->l2h = msg->tail;
|
||||
}
|
||||
|
||||
hh = (struct ipaccess_head *) msg->data;
|
||||
|
||||
/* then read the length as specified in header */
|
||||
msg->l2h = msg->data + sizeof(*hh);
|
||||
len = ntohs(hh->len);
|
||||
|
||||
if (len < 0 || IPA_ALLOC_SIZE < len + sizeof(*hh)) {
|
||||
LOGP(DLINP, LOGL_ERROR, "bad message length of %d bytes, "
|
||||
"received %d bytes\n", len, ret);
|
||||
msgb_free(msg);
|
||||
return -EIO;
|
||||
"received %d bytes\n", len, msg->len);
|
||||
ret = -EIO;
|
||||
goto discard_msg;
|
||||
}
|
||||
|
||||
ret = recv(fd, msg->l2h, len, 0);
|
||||
if (ret <= 0) {
|
||||
msgb_free(msg);
|
||||
return ret;
|
||||
} else if (ret < len) {
|
||||
LOGP(DLINP, LOGL_ERROR, "truncated message received\n");
|
||||
msgb_free(msg);
|
||||
return -EIO;
|
||||
needed = len - msgb_l2len(msg);
|
||||
|
||||
if (needed > 0) {
|
||||
ret = recv(fd, msg->tail, needed, 0);
|
||||
|
||||
if (ret == 0)
|
||||
goto discard_msg;
|
||||
|
||||
if (ret < 0) {
|
||||
if (errno == EAGAIN || errno == EINTR)
|
||||
ret = 0;
|
||||
else {
|
||||
ret = -errno;
|
||||
goto discard_msg;
|
||||
}
|
||||
}
|
||||
|
||||
msgb_put(msg, ret);
|
||||
|
||||
if (ret < needed) {
|
||||
LOGP(DLINP, LOGL_INFO,
|
||||
"Received part of IPA message L2 data (%d/%d)\n",
|
||||
msgb_l2len(msg), len);
|
||||
if (!tmp_msg) {
|
||||
ret = -EIO;
|
||||
goto discard_msg;
|
||||
}
|
||||
*tmp_msg = msg;
|
||||
return -EAGAIN;
|
||||
}
|
||||
}
|
||||
msgb_put(msg, ret);
|
||||
|
||||
ret = msgb_l2len(msg);
|
||||
|
||||
if (ret == 0) {
|
||||
LOGP(DLINP, LOGL_INFO,
|
||||
"Discarding IPA message without payload\n");
|
||||
ret = -EAGAIN;
|
||||
goto discard_msg;
|
||||
}
|
||||
|
||||
if (tmp_msg)
|
||||
*tmp_msg = NULL;
|
||||
*rmsg = msg;
|
||||
return ret;
|
||||
|
||||
discard_msg:
|
||||
if (tmp_msg)
|
||||
*tmp_msg = NULL;
|
||||
msgb_free(msg);
|
||||
return ret;
|
||||
}
|
||||
|
||||
void ipa_client_conn_close(struct ipa_client_conn *link)
|
||||
|
@ -103,6 +183,8 @@ void ipa_client_conn_close(struct ipa_client_conn *link)
|
|||
close(link->ofd->fd);
|
||||
link->ofd->fd = -1;
|
||||
}
|
||||
msgb_free(link->pending_msg);
|
||||
link->pending_msg = NULL;
|
||||
}
|
||||
|
||||
static void ipa_client_read(struct ipa_client_conn *link)
|
||||
|
@ -113,11 +195,12 @@ static void ipa_client_read(struct ipa_client_conn *link)
|
|||
|
||||
LOGP(DLINP, LOGL_DEBUG, "message received\n");
|
||||
|
||||
ret = ipa_msg_recv(ofd->fd, &msg);
|
||||
ret = ipa_msg_recv_buffered(ofd->fd, &msg, &link->pending_msg);
|
||||
if (ret < 0) {
|
||||
if (errno == EPIPE || errno == ECONNRESET) {
|
||||
if (ret == -EAGAIN)
|
||||
return;
|
||||
if (ret == -EPIPE || ret == -ECONNRESET)
|
||||
LOGP(DLINP, LOGL_ERROR, "lost connection with server\n");
|
||||
}
|
||||
ipa_client_conn_close(link);
|
||||
if (link->updown_cb)
|
||||
link->updown_cb(link, 0);
|
||||
|
@ -382,11 +465,12 @@ static void ipa_server_conn_read(struct ipa_server_conn *conn)
|
|||
|
||||
LOGP(DLINP, LOGL_DEBUG, "message received\n");
|
||||
|
||||
ret = ipa_msg_recv(ofd->fd, &msg);
|
||||
ret = ipa_msg_recv_buffered(ofd->fd, &msg, &conn->pending_msg);
|
||||
if (ret < 0) {
|
||||
if (errno == EPIPE || errno == ECONNRESET) {
|
||||
if (ret == -EAGAIN)
|
||||
return;
|
||||
if (ret == -EPIPE || ret == -ECONNRESET)
|
||||
LOGP(DLINP, LOGL_ERROR, "lost connection with server\n");
|
||||
}
|
||||
ipa_server_conn_destroy(conn);
|
||||
return;
|
||||
} else if (ret == 0) {
|
||||
|
@ -471,6 +555,7 @@ ipa_server_conn_create(void *ctx, struct ipa_server_link *link, int fd,
|
|||
void ipa_server_conn_destroy(struct ipa_server_conn *conn)
|
||||
{
|
||||
close(conn->ofd.fd);
|
||||
msgb_free(conn->pending_msg);
|
||||
osmo_fd_unregister(&conn->ofd);
|
||||
if (conn->closed_cb)
|
||||
conn->closed_cb(conn);
|
||||
|
|
|
@ -258,6 +258,8 @@ int ipaccess_rcvmsg_bts_base(struct msgb *msg,
|
|||
static int ipaccess_drop(struct osmo_fd *bfd, struct e1inp_line *line)
|
||||
{
|
||||
int ret = 1;
|
||||
unsigned int ts_nr = bfd->priv_nr;
|
||||
struct e1inp_ts *e1i_ts = &line->ts[ts_nr-1];
|
||||
|
||||
/* Error case: we did not see any ID_RESP yet for this socket. */
|
||||
if (bfd->fd != -1) {
|
||||
|
@ -269,6 +271,9 @@ static int ipaccess_drop(struct osmo_fd *bfd, struct e1inp_line *line)
|
|||
ret = -ENOENT;
|
||||
}
|
||||
|
||||
msgb_free(e1i_ts->pending_msg);
|
||||
e1i_ts->pending_msg = NULL;
|
||||
|
||||
/* e1inp_sign_link_destroy releases the socket descriptors for us. */
|
||||
line->ops->sign_link_down(line);
|
||||
|
||||
|
@ -415,13 +420,15 @@ static int handle_ts1_read(struct osmo_fd *bfd)
|
|||
struct e1inp_ts *e1i_ts = &line->ts[ts_nr-1];
|
||||
struct e1inp_sign_link *link;
|
||||
struct ipaccess_head *hh;
|
||||
struct msgb *msg;
|
||||
struct msgb *msg = NULL;
|
||||
int ret;
|
||||
|
||||
ret = ipa_msg_recv(bfd->fd, &msg);
|
||||
ret = ipa_msg_recv_buffered(bfd->fd, &msg, &e1i_ts->pending_msg);
|
||||
if (ret < 0) {
|
||||
if (ret == -EAGAIN)
|
||||
return 0;
|
||||
LOGP(DLINP, LOGL_NOTICE, "Sign link problems, "
|
||||
"closing socket. Reason: %s\n", strerror(errno));
|
||||
"closing socket. Reason: %s\n", strerror(-ret));
|
||||
goto err;
|
||||
} else if (ret == 0) {
|
||||
LOGP(DLINP, LOGL_NOTICE, "Sign link vanished, dead socket\n");
|
||||
|
|
|
@ -65,7 +65,7 @@ static void append_ipa_message(struct msgb *msg, int proto, const char *text)
|
|||
strcpy((char *)l2, text);
|
||||
}
|
||||
|
||||
static int receive_messages(int fd)
|
||||
static int receive_messages(int fd, struct msgb **pending_msg)
|
||||
{
|
||||
struct msgb *msg;
|
||||
char dummy;
|
||||
|
@ -76,13 +76,22 @@ static int receive_messages(int fd)
|
|||
break;
|
||||
}
|
||||
msg = NULL;
|
||||
rc = ipa_msg_recv(fd, &msg);
|
||||
if (rc == -1)
|
||||
rc = -errno;
|
||||
rc = ipa_msg_recv_buffered(fd, &msg, pending_msg);
|
||||
|
||||
fprintf(stderr,
|
||||
"ipa_msg_recv: %d, msg %s NULL\n",
|
||||
rc, msg ? "!=" : "==");
|
||||
if (rc == -EAGAIN)
|
||||
"ipa_msg_recv_buffered: %d, msg %s NULL, "
|
||||
"pending_msg %s NULL\n",
|
||||
rc, msg ? "!=" : "==",
|
||||
!pending_msg ? "??" : *pending_msg ? "!=" : "==");
|
||||
if (pending_msg && !!msg == !!*pending_msg)
|
||||
printf( "got msg %s NULL, pending_msg %s NULL, "
|
||||
"returned: %s\n",
|
||||
msg ? "!=" : "==",
|
||||
*pending_msg ? "!=" : "==",
|
||||
rc == 0 ? "EOF" :
|
||||
rc > 0 ? "OK" :
|
||||
strerror(-rc));
|
||||
else if (!pending_msg && rc == -EAGAIN)
|
||||
printf( "got msg %s NULL, "
|
||||
"returned: %s\n",
|
||||
msg ? "!=" : "==",
|
||||
|
@ -94,7 +103,8 @@ static int receive_messages(int fd)
|
|||
if (rc == -EAGAIN)
|
||||
break;
|
||||
if (rc < 0) {
|
||||
printf("ipa_msg_recv failed with: %s\n", strerror(-rc));
|
||||
printf("ipa_msg_recv_buffered failed with: %s\n",
|
||||
strerror(-rc));
|
||||
return rc;
|
||||
}
|
||||
printf("got IPA message, size=%d, proto=%d, text=\"%s\"\n",
|
||||
|
@ -121,13 +131,15 @@ static int slurp_data(int fd) {
|
|||
return count;
|
||||
};
|
||||
|
||||
static void test_complete_recv(void)
|
||||
static void test_complete_recv(int do_not_assemble)
|
||||
{
|
||||
int sv[2];
|
||||
struct msgb *msg_out = msgb_alloc(4096, "msg_out");
|
||||
struct msgb *pending_msg = NULL;
|
||||
int rc, i;
|
||||
|
||||
printf("Testing IPA recv with complete messages.\n");
|
||||
printf("Testing IPA recv with complete messages%s.\n",
|
||||
do_not_assemble ? "" : " with assembling enabled");
|
||||
|
||||
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1)
|
||||
err(1, "socketpair");
|
||||
|
@ -145,7 +157,11 @@ static void test_complete_recv(void)
|
|||
}
|
||||
|
||||
for (i=0; i < ARRAY_SIZE(ipa_test_messages); i++) {
|
||||
rc = receive_messages(sv[0]);
|
||||
rc = receive_messages(sv[0],
|
||||
do_not_assemble ? NULL : &pending_msg);
|
||||
if (pending_msg)
|
||||
printf("Unexpected partial message: size=%d\n",
|
||||
pending_msg->len);
|
||||
if (rc == 0)
|
||||
break;
|
||||
|
||||
|
@ -160,16 +176,19 @@ static void test_complete_recv(void)
|
|||
close(sv[0]);
|
||||
|
||||
msgb_free(msg_out);
|
||||
msgb_free(pending_msg);
|
||||
}
|
||||
|
||||
|
||||
static void test_partial_recv(void)
|
||||
static void test_partial_recv(int do_not_assemble)
|
||||
{
|
||||
int sv[2];
|
||||
struct msgb *msg_out = msgb_alloc(4096, "msg_out");
|
||||
struct msgb *pending_msg = NULL;
|
||||
int rc, i;
|
||||
|
||||
printf("Testing IPA recv with partitioned messages.\n");
|
||||
printf("Testing IPA recv with partitioned messages%s.\n",
|
||||
do_not_assemble ? "" : " with assembling enabled");
|
||||
|
||||
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == -1)
|
||||
err(1, "socketpair");
|
||||
|
@ -190,7 +209,8 @@ static void test_partial_recv(void)
|
|||
if (msg_out->len == 0)
|
||||
shutdown(sv[1], SHUT_WR);
|
||||
|
||||
rc = receive_messages(sv[0]);
|
||||
rc = receive_messages(sv[0],
|
||||
do_not_assemble ? NULL : &pending_msg);
|
||||
|
||||
if (rc == 0)
|
||||
break;
|
||||
|
@ -205,6 +225,7 @@ static void test_partial_recv(void)
|
|||
close(sv[0]);
|
||||
|
||||
msgb_free(msg_out);
|
||||
msgb_free(pending_msg);
|
||||
}
|
||||
|
||||
static struct log_info info = {};
|
||||
|
@ -218,8 +239,10 @@ int main(int argc, char **argv)
|
|||
printf("Testing the IPA layer.\n");
|
||||
|
||||
/* run the tests */
|
||||
test_complete_recv();
|
||||
test_partial_recv();
|
||||
test_complete_recv(1);
|
||||
test_partial_recv(1);
|
||||
test_complete_recv(0);
|
||||
test_partial_recv(0);
|
||||
|
||||
printf("No crashes.\n");
|
||||
return 0;
|
||||
|
|
|
@ -5,8 +5,31 @@ got IPA message, size=86, proto=200, text="A longer test message. ABCDEFGHIJKLMN
|
|||
got IPA message, size=16, proto=200, text="Hello again IPA"
|
||||
got IPA message, size=1, proto=200, text=""
|
||||
got IPA message, size=14, proto=200, text="Next is empty"
|
||||
done: unread 14, unsent 0
|
||||
got msg == NULL, returned: Resource temporarily unavailable
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
done: unread 0, unsent 0
|
||||
Testing IPA recv with partitioned messages.
|
||||
ipa_msg_recv failed with: Input/output error
|
||||
ipa_msg_recv_buffered failed with: Input/output error
|
||||
done: unread 0, unsent 154
|
||||
Testing IPA recv with complete messages with assembling enabled.
|
||||
got IPA message, size=10, proto=200, text="Hello IPA"
|
||||
got IPA message, size=86, proto=200, text="A longer test message. ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyz"
|
||||
got IPA message, size=16, proto=200, text="Hello again IPA"
|
||||
got IPA message, size=1, proto=200, text=""
|
||||
got IPA message, size=14, proto=200, text="Next is empty"
|
||||
got msg == NULL, pending_msg == NULL, returned: Resource temporarily unavailable
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
done: unread 0, unsent 0
|
||||
Testing IPA recv with partitioned messages with assembling enabled.
|
||||
got IPA message, size=10, proto=200, text="Hello IPA"
|
||||
got IPA message, size=86, proto=200, text="A longer test message. ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyz"
|
||||
got IPA message, size=16, proto=200, text="Hello again IPA"
|
||||
got IPA message, size=1, proto=200, text=""
|
||||
got IPA message, size=14, proto=200, text="Next is empty"
|
||||
got msg == NULL, pending_msg == NULL, returned: Resource temporarily unavailable
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
got IPA message, size=4, proto=200, text="Bye"
|
||||
done: unread 0, unsent 0
|
||||
No crashes.
|
||||
|
|
Loading…
Reference in New Issue