June 2023 - gerrit-log - lists.osmocom.org

[M] Change in libosmo-netif[master]: stream: Move some code in preparation of next commit
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33195 ) Change subject: stream: Move some code in preparation of next commit ...................................................................... stream: Move some code in preparation of next commit This avoids unnecessary static declarations. Change-Id: Ic3f21967fe929b17327eb95739ace16c9734a151 --- M src/stream.c 1 file changed, 67 insertions(+), 57 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/95/33195/1 diff --git a/src/stream.c b/src/stream.c index 6388928..c27e1aa 100644 --- a/src/stream.c +++ b/src/stream.c @@ -1373,6 +1373,62 @@ link->flags |= OSMO_STREAM_SRV_F_RECONF; } +#define OSMO_STREAM_SRV_F_FLUSH_DESTROY (1 << 0) + +struct osmo_stream_srv { + struct osmo_stream_srv_link *srv; + enum osmo_stream_mode mode; + union { + struct osmo_fd ofd; + struct osmo_io_fd *iofd; + }; + struct llist_head tx_queue; + int (*closed_cb)(struct osmo_stream_srv *peer); + int (*read_cb)(struct osmo_stream_srv *peer); + int (*iofd_read_cb)(struct osmo_stream_srv *peer, struct msgb *msg); + void *data; + int flags; +}; + +static void stream_srv_iofd_read_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); + LOGP(DLINP, LOGL_DEBUG, "message received (res=%d)\n", res); + + if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) { + LOGP(DLINP, LOGL_DEBUG, "Connection is being flushed and closed; ignoring received message\n"); + msgb_free(msg); + return; + } + + if (res <= 0) { + osmo_stream_srv_set_flush_and_destroy(conn); + if (osmo_iofd_txqueue_len(iofd) == 0) + osmo_stream_srv_destroy(conn); + } else if (conn->iofd_read_cb) { + conn->iofd_read_cb(conn, msg); // TODO: Handle return value? + } + + return; +} + +static void stream_srv_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); + LOGP(DLINP, LOGL_DEBUG, "connected write\n"); + + if (res == -1) + LOGP(DLINP, LOGL_ERROR, "error to send: %s\n", strerror(errno)); + + if (osmo_iofd_txqueue_len(iofd) == 0) + if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) + osmo_stream_srv_destroy(conn); +} + +static struct osmo_io_ops srv_ioops = { + .read_cb = stream_srv_iofd_read_cb, + .write_cb = stream_srv_iofd_write_cb, +}; /*! \brief Set the socket type for the stream server link * \param[in] link Stream Server Link to modify @@ -1557,23 +1613,6 @@ link->ofd.fd = -1; } -#define OSMO_STREAM_SRV_F_FLUSH_DESTROY (1 << 0) - -struct osmo_stream_srv { - struct osmo_stream_srv_link *srv; - enum osmo_stream_mode mode; - union { - struct osmo_fd ofd; - struct osmo_io_fd *iofd; - }; - struct llist_head tx_queue; - int (*closed_cb)(struct osmo_stream_srv *peer); - int (*read_cb)(struct osmo_stream_srv *peer); - int (*iofd_read_cb)(struct osmo_stream_srv *peer, struct msgb *msg); - void *data; - int flags; -}; - static int osmo_stream_srv_read(struct osmo_stream_srv *conn) { int rc = 0; @@ -1706,46 +1745,6 @@ return conn; } -static void stream_srv_iofd_read_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) -{ - struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); - LOGP(DLINP, LOGL_DEBUG, "message received (res=%d)\n", res); - - if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) { - LOGP(DLINP, LOGL_DEBUG, "Connection is being flushed and closed; ignoring received message\n"); - msgb_free(msg); - return; - } - - if (res <= 0) { - osmo_stream_srv_set_flush_and_destroy(conn); - if (osmo_iofd_txqueue_len(iofd) == 0) - osmo_stream_srv_destroy(conn); - } else if (conn->iofd_read_cb) { - conn->iofd_read_cb(conn, msg); // TODO: Handle return value? - } - - return; -} - -static void stream_srv_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) -{ - struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); - LOGP(DLINP, LOGL_DEBUG, "connected write\n"); - - if (res == -1) - LOGP(DLINP, LOGL_ERROR, "error to send: %s\n", strerror(errno)); - - if (osmo_iofd_txqueue_len(iofd) == 0) - if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) - osmo_stream_srv_destroy(conn); -} - -static struct osmo_io_ops srv_ioops = { - .read_cb = stream_srv_iofd_read_cb, - .write_cb = stream_srv_iofd_write_cb, -}; - struct osmo_stream_srv * osmo_stream_srv_create_iofd(void *ctx, const char *name, struct osmo_stream_srv_link *link, int fd, -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33195 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: Ic3f21967fe929b17327eb95739ace16c9734a151 Gerrit-Change-Number: 33195 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[M] Change in libosmo-netif[master]: examples: Use osmo_io in {ipa-,}stream-{client,server}
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33194 ) Change subject: examples: Use osmo_io in {ipa-,}stream-{client,server} ...................................................................... examples: Use osmo_io in {ipa-,}stream-{client,server} Change-Id: I97a9979199c816686b32080534627f6f033e009e --- M examples/ipa-stream-client.c M examples/ipa-stream-server.c M examples/stream-client.c M examples/stream-server.c 4 files changed, 28 insertions(+), 76 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/94/33194/1 diff --git a/examples/ipa-stream-client.c b/examples/ipa-stream-client.c index 0c9c589..fc48a50 100644 --- a/examples/ipa-stream-client.c +++ b/examples/ipa-stream-client.c @@ -103,21 +103,10 @@ return 0; } -static int read_cb(struct osmo_stream_cli *conn) +static int read_cb(struct osmo_stream_cli *conn, struct msgb *msg) { - struct msgb *msg; + LOGP(DIPATEST, LOGL_DEBUG, "received message from stream (len=%d)\n", msgb_length(msg)); - LOGP(DIPATEST, LOGL_DEBUG, "received message from stream\n"); - - msg = osmo_ipa_msg_alloc(0); - if (msg == NULL) { - LOGP(DIPATEST, LOGL_ERROR, "cannot allocate message\n"); - return 0; - } - if (osmo_stream_cli_recv(conn, msg) <= 0) { - LOGP(DIPATEST, LOGL_ERROR, "cannot receive message\n"); - return 0; - } if (osmo_ipa_process_msg(msg) < 0) { LOGP(DIPATEST, LOGL_ERROR, "bad IPA message\n"); return 0; @@ -169,13 +158,13 @@ tall_test = talloc_named_const(NULL, 1, "osmo_stream_client_test"); msgb_talloc_ctx_init(tall_test, 0); osmo_init_logging2(tall_test, &osmo_stream_client_test_log_info); - log_set_log_level(osmo_stderr_target, LOGL_NOTICE); + log_set_log_level(osmo_stderr_target, LOGL_DEBUG); /* * initialize stream client. */ - conn = osmo_stream_cli_create(tall_test); + conn = osmo_stream_cli_create_iofd(tall_test, "ipa test client"); if (conn == NULL) { fprintf(stderr, "cannot create client\n"); exit(EXIT_FAILURE); @@ -184,7 +173,7 @@ osmo_stream_cli_set_port(conn, 10000); osmo_stream_cli_set_connect_cb(conn, connect_cb); osmo_stream_cli_set_disconnect_cb(conn, disconnect_cb); - osmo_stream_cli_set_read_cb(conn, read_cb); + osmo_stream_cli_set_iofd_read_cb(conn, read_cb); osmo_stream_cli_set_data(conn, &num_msgs); osmo_stream_cli_set_nodelay(conn, true); diff --git a/examples/ipa-stream-server.c b/examples/ipa-stream-server.c index 1ca1aaf..9cc56db 100644 --- a/examples/ipa-stream-server.c +++ b/examples/ipa-stream-server.c @@ -47,23 +47,10 @@ exit(EXIT_SUCCESS); } -int read_cb(struct osmo_stream_srv *conn) +int read_cb(struct osmo_stream_srv *conn, struct msgb *msg) { - struct msgb *msg; + LOGP(DSTREAMTEST, LOGL_DEBUG, "received message from stream (len=%d)\n", msgb_length(msg)); - LOGP(DSTREAMTEST, LOGL_DEBUG, "received message from stream\n"); - - msg = osmo_ipa_msg_alloc(0); - if (msg == NULL) { - LOGP(DSTREAMTEST, LOGL_ERROR, "cannot allocate message\n"); - return 0; - } - if (osmo_stream_srv_recv(conn, msg) <= 0) { - LOGP(DSTREAMTEST, LOGL_ERROR, "cannot receive message\n"); - osmo_stream_srv_destroy(conn); - msgb_free(msg); - return -EBADF; - } if (osmo_ipa_process_msg(msg) < 0) { LOGP(DSTREAMTEST, LOGL_ERROR, "Bad IPA message\n"); msgb_free(msg); @@ -84,11 +71,11 @@ { if (conn != NULL) { LOGP(DSTREAMTEST, LOGL_ERROR, "Sorry, this example only " - "support one client simultaneously\n"); + "supports one client simultaneously\n"); return -1; } - conn = osmo_stream_srv_create(tall_test, srv, fd, + conn = osmo_stream_srv_create_iofd(tall_test, "srv link", srv, fd, read_cb, close_cb, NULL); if (conn == NULL) { LOGP(DSTREAMTEST, LOGL_ERROR, @@ -104,7 +91,7 @@ tall_test = talloc_named_const(NULL, 1, "osmo_stream_srv_test"); msgb_talloc_ctx_init(tall_test, 0); osmo_init_logging2(tall_test, &osmo_stream_srv_test_log_info); - log_set_log_level(osmo_stderr_target, LOGL_NOTICE); + log_set_log_level(osmo_stderr_target, LOGL_DEBUG); /* * initialize stream srv. diff --git a/examples/stream-client.c b/examples/stream-client.c index 428402e..7b8e092 100644 --- a/examples/stream-client.c +++ b/examples/stream-client.c @@ -50,27 +50,11 @@ return 0; } -static int read_cb(struct osmo_stream_cli *conn) +static int read_cb(struct osmo_stream_cli *conn, struct msgb* msg) { - int bytes; - struct msgb *msg; - LOGP(DSTREAMTEST, LOGL_NOTICE, "receiving message from stream... "); - msg = msgb_alloc(1024, "STREAMCLIENT/test"); - if (msg == NULL) { - LOGPC(DSTREAMTEST, LOGL_ERROR, "cannot allocate message\n"); - return 0; - } - - bytes = osmo_stream_cli_recv(conn, msg); - - if (bytes < 0) { - LOGPC(DSTREAMTEST, LOGL_ERROR, "cannot receive message\n"); - return 0; - } - - LOGPC(DSTREAMTEST, LOGL_NOTICE, "got %d (%d) bytes: %s\n", bytes, msg->len, msgb_hexdump(msg)); + LOGPC(DSTREAMTEST, LOGL_NOTICE, "got %d bytes: %s\n", msg->len, msgb_hexdump(msg)); msgb_free(msg); return 0; @@ -120,7 +104,7 @@ * initialize stream cli. */ - conn = osmo_stream_cli_create(tall_test); + conn = osmo_stream_cli_create_iofd(tall_test, "stream client"); if (conn == NULL) { fprintf(stderr, "cannot create cli\n"); exit(EXIT_FAILURE); @@ -129,7 +113,7 @@ osmo_stream_cli_set_port(conn, 10000); osmo_stream_cli_set_connect_cb(conn, connect_cb); osmo_stream_cli_set_disconnect_cb(conn, disconnect_cb); - osmo_stream_cli_set_read_cb(conn, read_cb); + osmo_stream_cli_set_iofd_read_cb(conn, read_cb); if (osmo_stream_cli_open(conn) < 0) { fprintf(stderr, "cannot open cli\n"); diff --git a/examples/stream-server.c b/examples/stream-server.c index e4ca480..f6edf42 100644 --- a/examples/stream-server.c +++ b/examples/stream-server.c @@ -42,29 +42,11 @@ exit(EXIT_SUCCESS); } -int read_cb(struct osmo_stream_srv *conn) +int read_cb(struct osmo_stream_srv *conn, struct msgb* msg) { - int bytes; - struct msgb *msg; - LOGP(DSTREAMTEST, LOGL_NOTICE, "receiving message from stream... "); - msg = msgb_alloc(1024, "STREAMSERVER/test"); - if (msg == NULL) { - LOGPC(DSTREAMTEST, LOGL_ERROR, "cannot allocate message\n"); - return 0; - } - - bytes = osmo_stream_srv_recv(conn, msg); - - if (bytes <= 0) { - if (bytes < 0) - LOGPC(DSTREAMTEST, LOGL_ERROR, "cannot receive message: %s\n", strerror(-bytes)); - else - LOGPC(DSTREAMTEST, LOGL_ERROR, "client closed connection\n"); - osmo_stream_srv_destroy(conn); - } else - LOGPC(DSTREAMTEST, LOGL_NOTICE, "got %d (%d) bytes: %s\n", bytes, msg->len, msgb_hexdump(msg)); + LOGPC(DSTREAMTEST, LOGL_NOTICE, "got %d bytes: %s\n", msg->len, msgb_hexdump(msg)); msgb_free(msg); return 0; @@ -72,6 +54,7 @@ static int close_cb(struct osmo_stream_srv *dummy) { + LOGPC(DSTREAMTEST, LOGL_ERROR, "client closed connection\n"); conn = NULL; return 0; } @@ -86,8 +69,8 @@ return -1; } - conn = osmo_stream_srv_create(tall_test, srv, fd, read_cb, - close_cb, NULL); + conn = osmo_stream_srv_create_iofd(tall_test, "stream server", srv, fd, read_cb, + close_cb, NULL); if (conn == NULL) { LOGP(DSTREAMTEST, LOGL_ERROR, "error while creating connection\n"); -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33194 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I97a9979199c816686b32080534627f6f033e009e Gerrit-Change-Number: 33194 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-CC: daniel <dwillmann(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[L] Change in libosmo-netif[master]: Add osmo_io support to osmo_stream_cli and osmo_stream_srv
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33193 ) Change subject: Add osmo_io support to osmo_stream_cli and osmo_stream_srv ...................................................................... Add osmo_io support to osmo_stream_cli and osmo_stream_srv Change-Id: I2f52c7107c392b6f4b0bf2a84f8c873c084a200c --- M include/osmocom/netif/stream.h M src/stream.c 2 files changed, 403 insertions(+), 36 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/93/33193/1 diff --git a/include/osmocom/netif/stream.h b/include/osmocom/netif/stream.h index a52e7c4..0e983a4 100644 --- a/include/osmocom/netif/stream.h +++ b/include/osmocom/netif/stream.h @@ -45,6 +45,7 @@ struct osmo_stream_srv; struct osmo_stream_srv *osmo_stream_srv_create(void *ctx, struct osmo_stream_srv_link *link, int fd, int (*cb)(struct osmo_stream_srv *conn), int (*closed_cb)(struct osmo_stream_srv *conn), void *data); +struct osmo_stream_srv *osmo_stream_srv_create_iofd(void *ctx, const char *name, struct osmo_stream_srv_link *link, int fd, int (*read_cb)(struct osmo_stream_srv *conn, struct msgb *msg), int (*closed_cb)(struct osmo_stream_srv *conn), void *data); void *osmo_stream_srv_get_data(struct osmo_stream_srv *conn); struct osmo_stream_srv_link *osmo_stream_srv_get_master(struct osmo_stream_srv *conn); struct osmo_fd *osmo_stream_srv_get_ofd(struct osmo_stream_srv *srv); @@ -79,10 +80,12 @@ void osmo_stream_cli_set_connect_cb(struct osmo_stream_cli *cli, int (*connect_cb)(struct osmo_stream_cli *cli)); void osmo_stream_cli_set_disconnect_cb(struct osmo_stream_cli *cli, int (*disconnect_cb)(struct osmo_stream_cli *cli)); void osmo_stream_cli_set_read_cb(struct osmo_stream_cli *cli, int (*read_cb)(struct osmo_stream_cli *cli)); +void osmo_stream_cli_set_iofd_read_cb(struct osmo_stream_cli *cli, int (*read_cb)(struct osmo_stream_cli *cli, struct msgb* msg)); void osmo_stream_cli_reconnect(struct osmo_stream_cli *cli); bool osmo_stream_cli_is_connected(struct osmo_stream_cli *cli); struct osmo_stream_cli *osmo_stream_cli_create(void *ctx); +struct osmo_stream_cli *osmo_stream_cli_create_iofd(void *ctx, const char *name); void osmo_stream_cli_destroy(struct osmo_stream_cli *cli); int osmo_stream_cli_open(struct osmo_stream_cli *cli); diff --git a/src/stream.c b/src/stream.c index 6afb7ca..6388928 100644 --- a/src/stream.c +++ b/src/stream.c @@ -37,6 +37,8 @@ #include <osmocom/core/utils.h> #include <osmocom/gsm/tlv.h> #include <osmocom/core/msgb.h> +#include <osmocom/core/osmo_io.h> +#include <osmocom/core/panic.h> #include <osmocom/core/logging.h> #include <osmocom/core/talloc.h> #include <osmocom/core/socket.h> @@ -247,8 +249,17 @@ #define OSMO_STREAM_MAX_ADDRS 1 #endif +enum osmo_stream_mode { + OSMO_STREAM_MODE_OSMO_FD, + OSMO_STREAM_MODE_OSMO_IO, +}; + struct osmo_stream_cli { - struct osmo_fd ofd; + enum osmo_stream_mode mode; + union { + struct osmo_fd ofd; + struct osmo_io_fd *iofd; + }; struct llist_head tx_queue; struct osmo_timer_list timer; enum osmo_stream_cli_state state; @@ -264,6 +275,7 @@ int (*connect_cb)(struct osmo_stream_cli *cli); int (*disconnect_cb)(struct osmo_stream_cli *cli); int (*read_cb)(struct osmo_stream_cli *cli); + int (*iofd_read_cb)(struct osmo_stream_cli *cli, struct msgb *msg); int (*write_cb)(struct osmo_stream_cli *cli); void *data; int flags; @@ -300,17 +312,40 @@ return cli->state == STREAM_CLI_STATE_CONNECTED; } -/*! \brief Close an Osmocom Stream Client - * \param[in] cli Osmocom Stream Client to be closed - * We unregister the socket fd from the osmocom select() loop - * abstraction and close the socket */ -void osmo_stream_cli_close(struct osmo_stream_cli *cli) +static void osmo_stream_cli_close_iofd(struct osmo_stream_cli *cli) +{ + if (!cli->iofd) + return; + + osmo_iofd_close(cli->iofd); +} + +static void osmo_stream_cli_close_ofd(struct osmo_stream_cli *cli) { if (cli->ofd.fd == -1) return; osmo_fd_unregister(&cli->ofd); close(cli->ofd.fd); cli->ofd.fd = -1; +} + +/*! \brief Close an Osmocom Stream Client + * \param[in] cli Osmocom Stream Client to be closed + * We unregister the socket fd from the osmocom select() loop + * abstraction and close the socket */ +void osmo_stream_cli_close(struct osmo_stream_cli *cli) +{ + if (cli->state == STREAM_CLI_STATE_CLOSED) + return; + + switch (cli->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + osmo_stream_cli_close_ofd(cli); + break; + case OSMO_STREAM_MODE_OSMO_IO: + osmo_stream_cli_close_iofd(cli); + break; + } if (cli->state == STREAM_CLI_STATE_CONNECTED) { LOGSCLI(cli, LOGL_DEBUG, "connection closed\n"); @@ -323,7 +358,14 @@ static inline int osmo_stream_cli_fd(const struct osmo_stream_cli *cli) { - return cli->ofd.fd; + switch (cli->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + return cli->ofd.fd; + case OSMO_STREAM_MODE_OSMO_IO: + return osmo_iofd_get_fd(cli->iofd); + default: + return -ENOTSUP; + } } static void osmo_stream_cli_read(struct osmo_stream_cli *cli) @@ -331,7 +373,7 @@ LOGSCLI(cli, LOGL_DEBUG, "message received\n"); if (cli->read_cb) - cli->read_cb(cli); + cli->read_cb(cli); // TODO: Return value? } static int osmo_stream_cli_write(struct osmo_stream_cli *cli) @@ -457,12 +499,12 @@ break; } if (cli->connect_cb) - cli->connect_cb(cli); + cli->connect_cb(cli); // TODO: Check return value? break; case STREAM_CLI_STATE_CONNECTED: if (what & OSMO_FD_READ) { LOGSCLI(cli, LOGL_DEBUG, "connected read\n"); - osmo_stream_cli_read(cli); + osmo_stream_cli_read(cli); // TODO: Check return value and return early? } if (what & OSMO_FD_WRITE) { LOGSCLI(cli, LOGL_DEBUG, "connected write\n"); @@ -492,6 +534,7 @@ if (!cli) return NULL; + cli->mode = OSMO_STREAM_MODE_OSMO_FD; cli->sk_domain = AF_UNSPEC; cli->sk_type = SOCK_STREAM; cli->proto = IPPROTO_TCP; @@ -507,6 +550,146 @@ return cli; } +static inline void osmo_stream_cli_handle_connecting(struct osmo_stream_cli *cli, int fd) +{ + int ret, error; + socklen_t len = sizeof(error); + + ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len); + if (ret >= 0 && error > 0) { + osmo_stream_cli_reconnect(cli); + return; + } + + LOGSCLI(cli, LOGL_DEBUG, "connection done.\n"); + cli->state = STREAM_CLI_STATE_CONNECTED; + switch (cli->sk_domain) { + case AF_UNIX: + _setsockopt_nosigpipe(cli); + break; + case AF_UNSPEC: + case AF_INET: + case AF_INET6: + if (cli->proto == IPPROTO_SCTP) { + _setsockopt_nosigpipe(cli); + sctp_sock_activate_events(fd); + } + break; + default: + break; + } + if (cli->connect_cb) + cli->connect_cb(cli); +} + +static void handle_connecting(struct osmo_stream_cli *cli, int res) +{ + int error, ret = res; + socklen_t len = sizeof(error); + + int fd = osmo_stream_cli_fd(cli); + + if (ret < 0) { + osmo_stream_cli_reconnect(cli); + return; + } + ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len); + if (ret >= 0 && error > 0) { + osmo_stream_cli_reconnect(cli); + return; + } + + LOGSCLI(cli, LOGL_DEBUG, "connection done.\n"); + cli->state = STREAM_CLI_STATE_CONNECTED; + switch (cli->sk_domain) { + case AF_UNIX: + _setsockopt_nosigpipe(cli); + break; + case AF_UNSPEC: + case AF_INET: + case AF_INET6: + if (cli->proto == IPPROTO_SCTP) { + //FIXME!! + _setsockopt_nosigpipe(cli); + sctp_sock_activate_events(fd); + } + break; + default: + break; + } + if (cli->connect_cb) + cli->connect_cb(cli); +} + +static void stream_cli_iofd_read_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_cli *cli = osmo_iofd_get_data(iofd); + + switch (cli->state) { + case STREAM_CLI_STATE_CONNECTING: + handle_connecting(cli, res); + break; + case STREAM_CLI_STATE_CONNECTED: + if (res == 0) + osmo_stream_cli_reconnect(cli); + else if (cli->iofd_read_cb) + cli->iofd_read_cb(cli, msg); + break; + default: + osmo_panic("osmo_stream_cli_write_cb() called with unexpected state %d\n", cli->state); + } +} + +static void stream_cli_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_cli *cli = osmo_iofd_get_data(iofd); + + switch (cli->state) { + case STREAM_CLI_STATE_CONNECTING: + handle_connecting(cli, res); + break; + case STREAM_CLI_STATE_CONNECTED: + if (msg && res <= 0) { + osmo_stream_cli_reconnect(cli); + LOGSCLI(cli, LOGL_ERROR, "error %d to send\n", res); + } + break; + default: + osmo_panic("osmo_stream_cli_write_cb() called with unexpected state %d\n", cli->state); + } +} + +static struct osmo_io_ops osmo_stream_cli_ioops = { + .read_cb = stream_cli_iofd_read_cb, + .write_cb = stream_cli_iofd_write_cb, + + .segmentation_cb = NULL, +}; + + +struct osmo_stream_cli *osmo_stream_cli_create_iofd(void *ctx, const char *name) +{ + struct osmo_stream_cli *cli; + + cli = talloc_zero(ctx, struct osmo_stream_cli); + if (!cli) + return NULL; + + cli->mode = OSMO_STREAM_MODE_OSMO_IO; + cli->sk_domain = AF_UNSPEC; + cli->sk_type = SOCK_STREAM; + cli->proto = IPPROTO_TCP; + + cli->iofd = osmo_iofd_setup(ctx, -1, name, OSMO_IO_FD_MODE_READ_WRITE, &osmo_stream_cli_ioops, cli); + + cli->state = STREAM_CLI_STATE_CLOSED; + osmo_timer_setup(&cli->timer, cli_timer_cb, cli); + cli->reconnect_timeout = 5; /* default is 5 seconds. */ + INIT_LLIST_HEAD(&cli->tx_queue); + + return cli; +} + /*! \brief Set the remote address to which we connect * \param[in] cli Stream Client to modify * \param[in] addr Remote IP address @@ -694,9 +877,20 @@ struct osmo_fd * osmo_stream_cli_get_ofd(struct osmo_stream_cli *cli) { + OSMO_ASSERT(cli->mode == OSMO_STREAM_MODE_OSMO_FD); return &cli->ofd; } +/*! \brief Get OsmoIO File Descriptor of the stream client socket + * \param[in] cli Stream Client to modify + * \returns Pointer to \ref osmo_iofd */ +struct osmo_io_fd * +osmo_stream_cli_get_iofd(struct osmo_stream_cli *cli) +{ + OSMO_ASSERT(cli->mode == OSMO_STREAM_MODE_OSMO_IO); + return cli->iofd; +} + /*! \brief Set the call-back function called on connect of the stream client socket * \param[in] cli Stream Client to modify * \param[in] connect_cb Call-back function to be called upon connect */ @@ -717,15 +911,29 @@ } /*! \brief Set the call-back function called to read from the stream client socket + * Only for osmo_stream_cli created with osmo_stream_cli_create() * \param[in] cli Stream Client to modify * \param[in] read_cb Call-back function to be called when we want to read */ void osmo_stream_cli_set_read_cb(struct osmo_stream_cli *cli, int (*read_cb)(struct osmo_stream_cli *cli)) { + OSMO_ASSERT(cli->mode == OSMO_STREAM_MODE_OSMO_FD); cli->read_cb = read_cb; } +/*! \brief Set the call-back function called to read from the stream client socket + * Only for oamo_stream_cli created with osmo_stream_cli_create_iofd() + * \param[in] cli Stream Client to modify + * \param[in] read_cb Call-back function to be called when we want to read */ +void +osmo_stream_cli_set_iofd_read_cb(struct osmo_stream_cli *cli, + int (*read_cb)(struct osmo_stream_cli *cli, struct msgb* msg)) +{ + OSMO_ASSERT(cli->mode == OSMO_STREAM_MODE_OSMO_IO); + cli->iofd_read_cb = read_cb; +} + /*! \brief Destroy a Osmocom stream client (includes close) * \param[in] cli Stream Client to destroy */ void osmo_stream_cli_destroy(struct osmo_stream_cli *cli) @@ -864,9 +1072,21 @@ goto error_close_socket; } - osmo_fd_setup(&cli->ofd, fd, OSMO_FD_READ | OSMO_FD_WRITE, cli->ofd.cb, cli->ofd.data, cli->ofd.priv_nr); - if (osmo_fd_register(&cli->ofd) < 0) - goto error_close_socket; + switch (cli->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + osmo_fd_setup(&cli->ofd, fd, OSMO_FD_READ | OSMO_FD_WRITE, cli->ofd.cb, cli->ofd.data, cli->ofd.priv_nr); + if (osmo_fd_register(&cli->ofd) < 0) + goto error_close_socket; + break; + case OSMO_STREAM_MODE_OSMO_IO: + if (osmo_iofd_register(cli->iofd, fd) < 0) + goto error_close_socket; + osmo_iofd_read_enable(cli->iofd); + osmo_iofd_write_enable(cli->iofd); + break; + default: + OSMO_ASSERT(false); + } cli->state = STREAM_CLI_STATE_CONNECTING; return 0; @@ -874,7 +1094,7 @@ error_close_socket: cli->state = STREAM_CLI_STATE_CLOSED; close(cli->ofd.fd); - cli->ofd.fd = -1; + cli->ofd.fd = -1; // XXX return -EIO; } @@ -893,8 +1113,17 @@ { OSMO_ASSERT(cli); OSMO_ASSERT(msg); - msgb_enqueue(&cli->tx_queue, msg); - osmo_fd_write_enable(&cli->ofd); + + switch (cli->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + msgb_enqueue(&cli->tx_queue, msg); + osmo_fd_write_enable(&cli->ofd); + break; + case OSMO_STREAM_MODE_OSMO_IO: + osmo_iofd_write_msgb(cli->iofd, msg); + osmo_iofd_write_enable(cli->iofd); + break; + } } /*! \brief Receive data via an Osmocom stream client @@ -926,11 +1155,19 @@ void osmo_stream_cli_clear_tx_queue(struct osmo_stream_cli *cli) { - msgb_queue_free(&cli->tx_queue); - /* If in state 'connecting', keep WRITE flag up to receive - * socket connection signal and then transition to STATE_CONNECTED: */ - if (cli->state == STREAM_CLI_STATE_CONNECTED) - osmo_fd_write_disable(&cli->ofd); + switch (cli->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + msgb_queue_free(&cli->tx_queue); + /* If in state 'connecting', keep WRITE flag up to receive + * socket connection signal and then transition to STATE_CONNECTED: */ + if (cli->state == STREAM_CLI_STATE_CONNECTED) + osmo_fd_write_disable(&cli->ofd); + break; + case OSMO_STREAM_MODE_OSMO_IO: + osmo_iofd_txqueue_clear(cli->iofd); + osmo_iofd_write_disable(cli->iofd); + break; + } } /* @@ -941,7 +1178,11 @@ #define OSMO_STREAM_SRV_F_NODELAY (1 << 1) struct osmo_stream_srv_link { - struct osmo_fd ofd; + enum osmo_stream_mode mode; + union { + struct osmo_fd ofd; + struct osmo_io_fd *iofd; + }; char *addr[OSMO_STREAM_MAX_ADDRS]; uint8_t addrcnt; uint16_t port; @@ -953,7 +1194,7 @@ int flags; }; -static int osmo_stream_srv_fd_cb(struct osmo_fd *ofd, unsigned int what) +static int osmo_stream_srv_ofd_cb(struct osmo_fd *ofd, unsigned int what) { int ret; int sock_fd; @@ -1038,7 +1279,23 @@ link->sk_domain = AF_UNSPEC; link->sk_type = SOCK_STREAM; link->proto = IPPROTO_TCP; - osmo_fd_setup(&link->ofd, -1, OSMO_FD_READ | OSMO_FD_WRITE, osmo_stream_srv_fd_cb, link, 0); + osmo_fd_setup(&link->ofd, -1, OSMO_FD_READ | OSMO_FD_WRITE, osmo_stream_srv_ofd_cb, link, 0); + + return link; +} + +struct osmo_stream_srv_link *osmo_stream_srv_link_create_iofd(void *ctx, const char *name) +{ + struct osmo_stream_srv_link *link; + + link = talloc_zero(ctx, struct osmo_stream_srv_link); + if (!link) + return NULL; + + link->sk_domain = AF_UNSPEC; + link->sk_type = SOCK_STREAM; + link->proto = IPPROTO_TCP; + osmo_fd_setup(&link->ofd, -1, OSMO_FD_READ | OSMO_FD_WRITE, osmo_stream_srv_ofd_cb, link, 0); return link; } @@ -1304,10 +1561,15 @@ struct osmo_stream_srv { struct osmo_stream_srv_link *srv; - struct osmo_fd ofd; + enum osmo_stream_mode mode; + union { + struct osmo_fd ofd; + struct osmo_io_fd *iofd; + }; struct llist_head tx_queue; int (*closed_cb)(struct osmo_stream_srv *peer); - int (*cb)(struct osmo_stream_srv *peer); + int (*read_cb)(struct osmo_stream_srv *peer); + int (*iofd_read_cb)(struct osmo_stream_srv *peer, struct msgb *msg); void *data; int flags; }; @@ -1323,8 +1585,8 @@ return 0; } - if (conn->cb) - rc = conn->cb(conn); + if (conn->read_cb) + rc = conn->read_cb(conn); return rc; } @@ -1415,12 +1677,13 @@ struct osmo_stream_srv * osmo_stream_srv_create(void *ctx, struct osmo_stream_srv_link *link, int fd, - int (*cb)(struct osmo_stream_srv *conn), + int (*read_cb)(struct osmo_stream_srv *conn), int (*closed_cb)(struct osmo_stream_srv *conn), void *data) { struct osmo_stream_srv *conn; OSMO_ASSERT(link); + OSMO_ASSERT(link->mode == OSMO_STREAM_MODE_OSMO_FD); conn = talloc_zero(ctx, struct osmo_stream_srv); if (conn == NULL) { @@ -1430,7 +1693,7 @@ } conn->srv = link; osmo_fd_setup(&conn->ofd, fd, OSMO_FD_READ, osmo_stream_srv_cb, conn, 0); - conn->cb = cb; + conn->read_cb = read_cb; conn->closed_cb = closed_cb; conn->data = data; INIT_LLIST_HEAD(&conn->tx_queue); @@ -1443,6 +1706,79 @@ return conn; } +static void stream_srv_iofd_read_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); + LOGP(DLINP, LOGL_DEBUG, "message received (res=%d)\n", res); + + if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) { + LOGP(DLINP, LOGL_DEBUG, "Connection is being flushed and closed; ignoring received message\n"); + msgb_free(msg); + return; + } + + if (res <= 0) { + osmo_stream_srv_set_flush_and_destroy(conn); + if (osmo_iofd_txqueue_len(iofd) == 0) + osmo_stream_srv_destroy(conn); + } else if (conn->iofd_read_cb) { + conn->iofd_read_cb(conn, msg); // TODO: Handle return value? + } + + return; +} + +static void stream_srv_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +{ + struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); + LOGP(DLINP, LOGL_DEBUG, "connected write\n"); + + if (res == -1) + LOGP(DLINP, LOGL_ERROR, "error to send: %s\n", strerror(errno)); + + if (osmo_iofd_txqueue_len(iofd) == 0) + if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) + osmo_stream_srv_destroy(conn); +} + +static struct osmo_io_ops srv_ioops = { + .read_cb = stream_srv_iofd_read_cb, + .write_cb = stream_srv_iofd_write_cb, +}; + +struct osmo_stream_srv * +osmo_stream_srv_create_iofd(void *ctx, const char *name, + struct osmo_stream_srv_link *link, int fd, + int (*read_cb)(struct osmo_stream_srv *conn, struct msgb *msg), + int (*closed_cb)(struct osmo_stream_srv *conn), void *data) +{ + struct osmo_stream_srv *conn; + + OSMO_ASSERT(link); + //OSMO_ASSERT(link->mode == OSMO_STREAM_MODE_OSMO_IO); + + conn = talloc_zero(ctx, struct osmo_stream_srv); + if (conn == NULL) { + LOGP(DLINP, LOGL_ERROR, "cannot allocate new peer in srv, " + "reason=`%s'\n", strerror(errno)); + return NULL; + } + conn->mode = OSMO_STREAM_MODE_OSMO_IO; + conn->srv = link; + conn->iofd = osmo_iofd_setup(conn, fd, name, OSMO_IO_FD_MODE_READ_WRITE, &srv_ioops, conn); + conn->iofd_read_cb = read_cb; + conn->closed_cb = closed_cb; + conn->data = data; + + if (osmo_iofd_register(conn->iofd, fd) < 0) { + LOGP(DLINP, LOGL_ERROR, "could not register FD\n"); + talloc_free(conn); + return NULL; + } + osmo_iofd_read_enable(conn->iofd); + return conn; +} + /*! \brief Prepare to send out all pending messages on the connection's Tx queue * and then automatically destroy the stream with osmo_stream_srv_destroy(). * This function disables queuing of new messages on the connection and also @@ -1480,6 +1816,11 @@ return &conn->ofd; } +struct osmo_io_fd * +osmo_stream_srv_get_iofd(struct osmo_stream_srv *conn) +{ + return conn->iofd; +} /*! \brief Get the master (Link) from a Stream Server * \param[in] conn Stream Server of which we want to know the Link * \returns Link through which the given Stream Server is established */ @@ -1496,12 +1837,19 @@ * \param[in] conn Stream Server to be destroyed */ void osmo_stream_srv_destroy(struct osmo_stream_srv *conn) { - osmo_fd_unregister(&conn->ofd); - close(conn->ofd.fd); - conn->ofd.fd = -1; + switch (conn->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + osmo_fd_unregister(&conn->ofd); + close(conn->ofd.fd); + msgb_queue_free(&conn->tx_queue); + conn->ofd.fd = -1; + break; + case OSMO_STREAM_MODE_OSMO_IO: + osmo_iofd_free(conn->iofd); + break; + } if (conn->closed_cb) conn->closed_cb(conn); - msgb_queue_free(&conn->tx_queue); talloc_free(conn); } @@ -1518,8 +1866,15 @@ return; } - msgb_enqueue(&conn->tx_queue, msg); - osmo_fd_write_enable(&conn->ofd); + switch (conn->mode) { + case OSMO_STREAM_MODE_OSMO_FD: + msgb_enqueue(&conn->tx_queue, msg); + osmo_fd_write_enable(&conn->ofd); + break; + case OSMO_STREAM_MODE_OSMO_IO: + osmo_iofd_write_msgb(conn->iofd, msg); + break; + } } #ifdef HAVE_LIBSCTP -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33193 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I2f52c7107c392b6f4b0bf2a84f8c873c084a200c Gerrit-Change-Number: 33193 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-CC: daniel <dwillmann(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[S] Change in libosmo-netif[master]: stream: Fix endless loop on server on client disconnect
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33196 ) Change subject: stream: Fix endless loop on server on client disconnect ...................................................................... stream: Fix endless loop on server on client disconnect Not sure why, but select() keeps reporting read events it seems, even though the client doesn't send any packets anymore. Without this line, the server gets stuck in an endless read loop with zero bytes read because of that. Change-Id: I012ddf14ae17642a52d34026d85ab6958cf488a1 --- M src/stream.c 1 file changed, 15 insertions(+), 0 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/96/33196/1 diff --git a/src/stream.c b/src/stream.c index c27e1aa..d6b16d3 100644 --- a/src/stream.c +++ b/src/stream.c @@ -1398,6 +1398,7 @@ if (conn->flags & OSMO_STREAM_SRV_F_FLUSH_DESTROY) { LOGP(DLINP, LOGL_DEBUG, "Connection is being flushed and closed; ignoring received message\n"); msgb_free(msg); + osmo_stream_srv_destroy(conn); return; } -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33196 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I012ddf14ae17642a52d34026d85ab6958cf488a1 Gerrit-Change-Number: 33196 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[S] Change in libosmo-netif[master]: stream: Add client-side (segmentation) support for IPAC
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33198 ) Change subject: stream: Add client-side (segmentation) support for IPAC ...................................................................... stream: Add client-side (segmentation) support for IPAC With this commit, IPAC segmentation is taken care of by setting the protocol being streamed Depends on change I3a639e6896cc3b3fc8e9b2e1a58254710efa0d3f Related: OS#5753, OS#5751 Change-Id: I822abf52c6ae396c90b5c50228a0a39c848d3de6 --- M include/osmocom/netif/stream.h M src/stream.c 2 files changed, 36 insertions(+), 0 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/98/33198/1 diff --git a/include/osmocom/netif/stream.h b/include/osmocom/netif/stream.h index c9b7a88..fa76d96 100644 --- a/include/osmocom/netif/stream.h +++ b/include/osmocom/netif/stream.h @@ -82,6 +82,7 @@ int osmo_stream_cli_set_type(struct osmo_stream_cli *cli, int type); int osmo_stream_cli_set_domain(struct osmo_stream_cli *cli, int domain); void osmo_stream_cli_set_proto(struct osmo_stream_cli *cli, uint16_t proto); +void osmo_stream_cli_set_stream_proto(struct osmo_stream_cli *cli, enum osmo_stream_proto osp); void osmo_stream_cli_set_local_addr(struct osmo_stream_cli *cli, const char *addr); int osmo_stream_cli_set_local_addrs(struct osmo_stream_cli *cli, const char **addr, size_t addrcnt); void osmo_stream_cli_set_local_port(struct osmo_stream_cli *cli, uint16_t port); diff --git a/src/stream.c b/src/stream.c index a033455..ea7e9b2 100644 --- a/src/stream.c +++ b/src/stream.c @@ -278,6 +278,7 @@ int sk_domain; int sk_type; uint16_t proto; + enum osmo_stream_proto stream_proto; int (*connect_cb)(struct osmo_stream_cli *cli); int (*disconnect_cb)(struct osmo_stream_cli *cli); int (*read_cb)(struct osmo_stream_cli *cli); @@ -685,6 +686,7 @@ cli->sk_domain = AF_UNSPEC; cli->sk_type = SOCK_STREAM; cli->proto = IPPROTO_TCP; + cli->stream_proto = OSMO_STREAM_UNSPECIFIED; cli->iofd = osmo_iofd_setup(ctx, -1, name, OSMO_IO_FD_MODE_READ_WRITE, &osmo_stream_cli_ioops, cli); @@ -799,6 +801,25 @@ cli->flags |= OSMO_STREAM_CLI_F_RECONF; } +/*! \brief Set the protocol streamed on the client socket + * \param[in] cli Stream Client to modify + * \param[in] osp Protocol being streamed + */ +void osmo_stream_cli_set_stream_proto(struct osmo_stream_cli *cli, enum osmo_stream_proto osp) +{ + if (!(OSMO_STREAM_UNSPECIFIED <= osp && osp < _NUM_OSMO_STREAM_PROTOS)) { + LOGSCLI(cli, LOGL_ERROR, "Unexpected value (%d) for variable of type " + "'enum osmo_stream_proto'\n", osp); + return; + } + cli->stream_proto = osp; + struct osmo_io_ops client_ops = osmo_stream_cli_ioops; + client_ops.segmentation_cb = segmentation_cbs[osp]; + osmo_iofd_set_ioops(cli->iofd, &client_ops); + cli->flags |= OSMO_STREAM_CLI_F_RECONF; + osmo_iofd_set_ioops(cli->iofd, &osmo_stream_cli_ioops); +} + /*! \brief Set the socket type for the stream server link * \param[in] cli Stream Client to modify * \param[in] type Socket Type (like SOCK_STREAM (default), SOCK_SEQPACKET, ...) -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33198 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I822abf52c6ae396c90b5c50228a0a39c848d3de6 Gerrit-Change-Number: 33198 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[S] Change in libosmo-netif[master]: stream: Adapt to new changes
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33199 ) Change subject: stream: Adapt to new changes ...................................................................... stream: Adapt to new changes - adapt to def. of struct osmo_io_ops - eliminate compiler warnings due to change in callback signature Requires change I6a0eebb8d4490f09a3cc6eb97d4ff47b4a8fd377 Related: OS#5753 Change-Id: I5f9a5ce85efb4ed512ff6ca1f3e170d2ffb2ba22 --- M src/stream.c 1 file changed, 17 insertions(+), 2 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/99/33199/1 diff --git a/src/stream.c b/src/stream.c index ea7e9b2..fbce7b4 100644 --- a/src/stream.c +++ b/src/stream.c @@ -669,7 +669,6 @@ static struct osmo_io_ops osmo_stream_cli_ioops = { .read_cb = stream_cli_iofd_read_cb, .write_cb = stream_cli_iofd_write_cb, - .segmentation_cb = NULL, }; @@ -1426,7 +1425,7 @@ return; } -static void stream_srv_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +static void stream_srv_iofd_write_cb(struct osmo_io_fd *iofd, int res, const struct msgb *msg) { struct osmo_stream_srv *conn = osmo_iofd_get_data(iofd); LOGP(DLINP, LOGL_DEBUG, "connected write\n"); -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33199 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I5f9a5ce85efb4ed512ff6ca1f3e170d2ffb2ba22 Gerrit-Change-Number: 33199 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[S] Change in libosmo-netif[master]: examples: Add extension header octet to example
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33202 ) Change subject: examples: Add extension header octet to example ...................................................................... examples: Add extension header octet to example Mainly to make data/packets look (pseudo-)correct in Wireshark Change-Id: I962b9edcba65cdc98da00d2f8753dc5acd481502 --- M examples/ipa-stream-client.c M src/stream.c 2 files changed, 16 insertions(+), 3 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/02/33202/1 diff --git a/examples/ipa-stream-client.c b/examples/ipa-stream-client.c index 48421b7..c3bb2a3 100644 --- a/examples/ipa-stream-client.c +++ b/examples/ipa-stream-client.c @@ -74,7 +74,7 @@ char *ptr; int x; - msg = osmo_ipa_msg_alloc(0); + msg = osmo_ipa_msg_alloc(sizeof(struct ipa_head_ext)); if (msg == NULL) { LOGP(DLINP, LOGL_ERROR, "cannot alloc msg\n"); return -1; @@ -93,6 +93,7 @@ msg_sent->num = i; llist_add(&msg_sent->head, &msg_sent_list); + ipa_prepend_header_ext(msg, IPAC_PROTO_EXT_MGCP); osmo_ipa_msg_push_header(msg, IPAC_PROTO_OSMO); osmo_stream_cli_send(conn, msg); @@ -115,7 +116,7 @@ int num; struct msg_sent *cur, *tmp, *found = NULL; - num = ntohl(*((int *)(msg->data + sizeof(struct ipa_head)))); + num = ntohl(*((int *)(msg->data + sizeof(struct ipa_head) + sizeof(struct ipa_head_ext)))); LOGP(DLINP, LOGL_DEBUG, "received msg number %d\n", num); llist_for_each_entry_safe(cur, tmp, &msg_sent_list, head) { diff --git a/src/stream.c b/src/stream.c index 396f344..35a1fc7 100644 --- a/src/stream.c +++ b/src/stream.c @@ -679,8 +679,9 @@ case STREAM_CLI_STATE_CONNECTED: if (res == 0) osmo_stream_cli_reconnect(cli); - else if (cli->iofd_read_cb) + else if (cli->iofd_read_cb) { cli->iofd_read_cb(cli, msg); + } break; default: osmo_panic("osmo_stream_cli_write_cb() called with unexpected state %d\n", cli->state); -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33202 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I962b9edcba65cdc98da00d2f8753dc5acd481502 Gerrit-Change-Number: 33202 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[S] Change in libosmo-netif[master]: stream: Add osmo_io based segmentation to example
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33200 ) Change subject: stream: Add osmo_io based segmentation to example ...................................................................... stream: Add osmo_io based segmentation to example Related: OS#5751, OS#5753 Change-Id: Ibf5fd3a606da7888f23e6f75cc30fe433897a97b --- M examples/ipa-stream-client.c M examples/ipa-stream-server.c 2 files changed, 12 insertions(+), 0 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/00/33200/1 diff --git a/examples/ipa-stream-client.c b/examples/ipa-stream-client.c index fc48a50..48421b7 100644 --- a/examples/ipa-stream-client.c +++ b/examples/ipa-stream-client.c @@ -176,6 +176,7 @@ osmo_stream_cli_set_iofd_read_cb(conn, read_cb); osmo_stream_cli_set_data(conn, &num_msgs); osmo_stream_cli_set_nodelay(conn, true); + osmo_stream_cli_set_stream_proto(conn, OSMO_STREAM_IPAC); if (osmo_stream_cli_open(conn) < 0) { fprintf(stderr, "cannot open client\n"); diff --git a/examples/ipa-stream-server.c b/examples/ipa-stream-server.c index 9cc56db..df5c26c 100644 --- a/examples/ipa-stream-server.c +++ b/examples/ipa-stream-server.c @@ -106,6 +106,7 @@ osmo_stream_srv_link_set_port(srv, 10000); osmo_stream_srv_link_set_accept_cb(srv, accept_cb); osmo_stream_srv_link_set_nodelay(srv, true); + osmo_stream_srv_link_set_stream_proto(srv, OSMO_STREAM_IPAC); if (osmo_stream_srv_link_open(srv) < 0) { fprintf(stderr, "cannot open client\n"); -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33200 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: Ibf5fd3a606da7888f23e6f75cc30fe433897a97b Gerrit-Change-Number: 33200 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[L] Change in libosmo-netif[master]: stream: Add server-side (segmentation) support for IPAC
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33197 ) Change subject: stream: Add server-side (segmentation) support for IPAC ...................................................................... stream: Add server-side (segmentation) support for IPAC With this commit, IPAC segmentation is taken care of by setting the protocol being streamed Depends on change I3a639e6896cc3b3fc8e9b2e1a58254710efa0d3f Related: OS#5753, OS#5751 Change-Id: I6c91ff385cb5f36ab6b6c96d0e44997995d0d24c --- M include/osmocom/netif/stream.h M src/stream.c M tests/stream/stream_test.c M tests/stream/stream_test.ok 4 files changed, 334 insertions(+), 21 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/97/33197/1 diff --git a/include/osmocom/netif/stream.h b/include/osmocom/netif/stream.h index 0e983a4..c9b7a88 100644 --- a/include/osmocom/netif/stream.h +++ b/include/osmocom/netif/stream.h @@ -21,6 +21,18 @@ /*! \brief Osmocom Stream Server Link: A server socket listening/accepting */ struct osmo_stream_srv_link; +/*! \brief Type of protocol transported by the data stream */ +enum osmo_stream_proto { + OSMO_STREAM_UNSPECIFIED = 0, + OSMO_STREAM_IPAC, + /* TODO: Add protocols for which libosmo-netif should be able to handle segmentation */ + _NUM_OSMO_STREAM_PROTOS +}; + +/*! \brief Shortcut for unsetting the stream protocol (gets rid of segmentation pertaining to stream protocol) */ +#define osmo_stream_srv_link_unset_stream_proto(struct_osmo_stream_srv_link_ptr)\ + osmo_stream_srv_link_set_stream_proto(struct_osmo_stream_srv_link_ptr, OSMO_STREAM_UNSPECIFIED) + struct osmo_stream_srv_link *osmo_stream_srv_link_create(void *ctx); void osmo_stream_srv_link_destroy(struct osmo_stream_srv_link *link); @@ -29,6 +41,7 @@ int osmo_stream_srv_link_set_addrs(struct osmo_stream_srv_link *link, const char **addr, size_t addrcnt); void osmo_stream_srv_link_set_port(struct osmo_stream_srv_link *link, uint16_t port); void osmo_stream_srv_link_set_proto(struct osmo_stream_srv_link *link, uint16_t proto); +void osmo_stream_srv_link_set_stream_proto(struct osmo_stream_srv_link *link, enum osmo_stream_proto osp); int osmo_stream_srv_link_set_type(struct osmo_stream_srv_link *link, int type); int osmo_stream_srv_link_set_domain(struct osmo_stream_srv_link *link, int domain); void osmo_stream_srv_link_set_accept_cb(struct osmo_stream_srv_link *link, int (*accept_cb)(struct osmo_stream_srv_link *link, int fd)); diff --git a/src/stream.c b/src/stream.c index d6b16d3..a033455 100644 --- a/src/stream.c +++ b/src/stream.c @@ -36,6 +36,7 @@ #include <osmocom/core/select.h> #include <osmocom/core/utils.h> #include <osmocom/gsm/tlv.h> +#include <osmocom/gsm/ipa.h> #include <osmocom/core/msgb.h> #include <osmocom/core/osmo_io.h> #include <osmocom/core/panic.h> @@ -78,6 +79,11 @@ #define MSG_NOSIGNAL 0 #endif +static int (*segmentation_cbs[_NUM_OSMO_STREAM_PROTOS])(struct msgb *) = { + [OSMO_STREAM_UNSPECIFIED] = NULL, + [OSMO_STREAM_IPAC] = ipa_segmentation_cb, +}; + /* is any of the bytes from offset .. u8_size in 'u8' non-zero? return offset or -1 if all zero */ static int byte_nonzero(const uint8_t *u8, unsigned int offset, unsigned int u8_size) { @@ -640,7 +646,7 @@ } } -static void stream_cli_iofd_write_cb(struct osmo_io_fd *iofd, int res, struct msgb *msg) +static void stream_cli_iofd_write_cb(struct osmo_io_fd *iofd, int res, const struct msgb *msg) { struct osmo_stream_cli *cli = osmo_iofd_get_data(iofd); @@ -1189,6 +1195,7 @@ int sk_domain; int sk_type; uint16_t proto; + enum osmo_stream_proto stream_proto; int (*accept_cb)(struct osmo_stream_srv_link *srv, int fd); void *data; int flags; @@ -1279,22 +1286,7 @@ link->sk_domain = AF_UNSPEC; link->sk_type = SOCK_STREAM; link->proto = IPPROTO_TCP; - osmo_fd_setup(&link->ofd, -1, OSMO_FD_READ | OSMO_FD_WRITE, osmo_stream_srv_ofd_cb, link, 0); - - return link; -} - -struct osmo_stream_srv_link *osmo_stream_srv_link_create_iofd(void *ctx, const char *name) -{ - struct osmo_stream_srv_link *link; - - link = talloc_zero(ctx, struct osmo_stream_srv_link); - if (!link) - return NULL; - - link->sk_domain = AF_UNSPEC; - link->sk_type = SOCK_STREAM; - link->proto = IPPROTO_TCP; + link->stream_proto = OSMO_STREAM_UNSPECIFIED; osmo_fd_setup(&link->ofd, -1, OSMO_FD_READ | OSMO_FD_WRITE, osmo_stream_srv_ofd_cb, link, 0); return link; @@ -1431,6 +1423,23 @@ .write_cb = stream_srv_iofd_write_cb, }; +/*! \brief Set the protocol transported by the stream for the stream server link + * \param[in] link Stream Server Link to modify + * \param[in] proto Protocol (like IPPROTO_TCP (default), IPPROTO_SCTP, ...). + */ +void osmo_stream_srv_link_set_stream_proto(struct osmo_stream_srv_link *link, + enum osmo_stream_proto osp) +{ + if (!(OSMO_STREAM_UNSPECIFIED <= osp && osp < _NUM_OSMO_STREAM_PROTOS)) { + LOGP(DLINP, LOGL_ERROR, "Unexpected value (%d) for variable of type " + "'enum osmo_stream_proto'\n", osp); + return; + } + link->stream_proto = osp; + srv_ioops.segmentation_cb = segmentation_cbs[osp]; + link->flags |= OSMO_STREAM_SRV_F_RECONF; +} + /*! \brief Set the socket type for the stream server link * \param[in] link Stream Server Link to modify * \param[in] type Socket Type (like SOCK_STREAM (default), SOCK_SEQPACKET, ...) @@ -1765,7 +1774,8 @@ } conn->mode = OSMO_STREAM_MODE_OSMO_IO; conn->srv = link; - conn->iofd = osmo_iofd_setup(conn, fd, name, OSMO_IO_FD_MODE_READ_WRITE, &srv_ioops, conn); + conn->iofd = osmo_iofd_setup(conn, fd, name, OSMO_IO_FD_MODE_READ_WRITE, + &srv_ioops, conn); conn->iofd_read_cb = read_cb; conn->closed_cb = closed_cb; conn->data = data; diff --git a/tests/stream/stream_test.c b/tests/stream/stream_test.c index 36a222e..b537d5e 100644 --- a/tests/stream/stream_test.c +++ b/tests/stream/stream_test.c @@ -8,18 +8,23 @@ * (at your option) any later version. */ +#include <inttypes.h> #include <stdio.h> #include <stdlib.h> #include <string.h> +#include <sys/socket.h> +#include <sys/un.h> #include <unistd.h> #include <errno.h> +#include <osmocom/core/byteswap.h> #include <osmocom/core/select.h> #include <osmocom/core/talloc.h> #include <osmocom/core/msgb.h> #include <osmocom/core/logging.h> #include <osmocom/core/application.h> #include <osmocom/core/timer.h> +#include <osmocom/gsm/protocol/ipaccess.h> #include <osmocom/netif/stream.h> @@ -363,10 +368,233 @@ printf("{%lu.%06lu} %s test complete.\n\n", tv.tv_sec, tv.tv_usec, ASTR(autoreconnect)); } +/* Segmentation test code (using IPA) */ +struct ipa_head { + uint16_t len; + uint8_t proto; + uint8_t data[0]; +} __attribute__ ((packed)); + +#define IPAC_MSG_PING_LEN 0x01 +static const uint8_t ipac_msg_ping[] = { + 0x00, IPAC_MSG_PING_LEN, + IPAC_PROTO_IPACCESS, + IPAC_MSGT_PING +}; +#define IPAC_MSG_PONG_LEN 0x01 +static const uint8_t ipac_msg_pong[] = { + 0x00, IPAC_MSG_PONG_LEN, + IPAC_PROTO_IPACCESS, + IPAC_MSGT_PONG +}; +#define IPAC_MSG_ID_REQ_LEN 0x03 +static const uint8_t ipac_msg_idreq[] = { + 0x00, IPAC_MSG_ID_REQ_LEN, + IPAC_PROTO_IPACCESS, + IPAC_MSGT_ID_GET, + 0x01, IPAC_IDTAG_UNITNAME +}; +#define ipac_msg_idreq_half (sizeof (ipac_msg_idreq)/2) +#define ipac_msg_idreq_other_half (sizeof (ipac_msg_idreq) - ipac_msg_idreq_half) +#define IPAC_MSG_ID_RESP_LEN 0x07 +static const uint8_t ipac_msg_idresp[] = { + 0x00, IPAC_MSG_ID_RESP_LEN, + IPAC_PROTO_IPACCESS, + IPAC_MSGT_ID_RESP, + 0x01, IPAC_IDTAG_UNITNAME, 0xde, 0xad, 0xbe, 0xef +}; + +#define put_ipa_msg(unsigned_char_ptr, struct_msgb_ptr, byte_array) do {\ + (unsigned_char_ptr) = msgb_put(struct_msgb_ptr, sizeof (byte_array));\ + memcpy(unsigned_char_ptr, byte_array, sizeof (byte_array));\ +} while (0) + +/* Array indices correspond to enum values stringified on the right */ +static const char * const IPAC_MSG_TYPES[] = { + [0] = "IPAC_MSGT_PING", + [1] = "IPAC_MSGT_PONG", + [2] = "UNEXPECTED VALUE", + [3] = "UNEXPECTED VALUE", + [4] = "IPAC_MSGT_ID_GET", + [5] = "IPAC_MSGT_ID_RESP", +}; + +#define IPAC_MSGT_OFFSET 3 +/* Append a message to UCHAR_PTR_DST. SRC_IPAC_MSG_BUF is expected to be a + * message buffer containing an * IPA message of type IPAC_PROTO_ACCESS that is + * syntactically correct up to offset 3 (IPAC_MSGT_OFFSET). + * Uses a counter so that appended messages can be distinguished easily in the logs */ +#define CLI_APPEND_MSG(OSMO_STREAM_CLI_PTR, UCHAR_PTR_DST, STRUCT_MSGB_PTR, SRC_IPAC_MSG_BUF) do {\ + LOGCLI(OSMO_STREAM_CLI_PTR, "[0%u-cli] Appending msg of type %s into buffer\n",\ + ++msglognum_cli, IPAC_MSG_TYPES[SRC_IPAC_MSG_BUF[IPAC_MSGT_OFFSET]]);\ + LOGCLI(OSMO_STREAM_CLI_PTR, "\t(msg dump: %s)\n", osmo_hexdump(SRC_IPAC_MSG_BUF,\ + sizeof(SRC_IPAC_MSG_BUF)));\ + put_ipa_msg(UCHAR_PTR_DST, STRUCT_MSGB_PTR, SRC_IPAC_MSG_BUF);\ +} while (0) + +static unsigned msglognum_cli = 0; +static int test_segm_cli_connect_cb(struct osmo_stream_cli *cli) +{ + printf("Connect callback triggered (segmentation test)\n"); + unsigned char *data; + struct msgb *m = msgb_alloc_headroom(128, 0, "IPA messages"); + if (m == NULL) { + fprintf(stderr, "Cannot allocate message\n"); + return -ENOMEM; + } + + /* Send 4 and 1/2 messages */ + CLI_APPEND_MSG(cli, data, m, ipac_msg_ping); + CLI_APPEND_MSG(cli, data, m, ipac_msg_pong); + CLI_APPEND_MSG(cli, data, m, ipac_msg_ping); + CLI_APPEND_MSG(cli, data, m, ipac_msg_idresp); + + LOGCLI(cli, "[0%u.5-cli] Appending *half* of msg of type %s into buffer\n", + msglognum_cli, IPAC_MSG_TYPES[ipac_msg_idreq[3]]); + LOGCLI(cli, "\t(dump: %s)\n", osmo_hexdump(ipac_msg_idreq, ipac_msg_idreq_half)); + data = msgb_put(m, ipac_msg_idreq_half); + memcpy(data, ipac_msg_idreq, ipac_msg_idreq_half); + + LOGCLI(cli, "Sending 4 1/2 messages as one:\n"); + LOGCLI(cli, "\t(msg dump: %s)\n\n", osmo_hexdump(m->data, m->len)); + osmo_stream_cli_send(cli, m); + return 0; +} + +static bool all_msgs_processed = false; + +static int test_segm_stream_cli_read_cb(struct osmo_stream_cli *osc, struct msgb *msg) +{ + unsigned char *data; + struct ipa_head *h = (struct ipa_head *) msg->data; + uint8_t ipa_msg_type = ((uint8_t *)h)[sizeof(struct ipa_head)]; + LOGCLI(osc, "Received message from stream (total len = %" PRIu16 ")\n", msgb_length(msg)); + if (ipa_msg_type < 0 || 5 < ipa_msg_type) { + fprintf(stderr, "Received unexpected IPAC message type %"PRIu8"\n", ipa_msg_type); + return -ENOMSG; + } + LOGCLI(osc, "\tType: %s\n", IPAC_MSG_TYPES[ipa_msg_type]); + if (ipa_msg_type == IPAC_MSGT_ID_GET) { + LOGCLI(osc, "Got IPAC_MSGT_ID_GET from server\n"); + LOGCLI(osc, "[%u-cli] Sending second half of IPAC_MSGT_ID_GET\n", ++msglognum_cli); + data = msgb_put(msg, ipac_msg_idreq_other_half); + memcpy(data, ipac_msg_idreq + ipac_msg_idreq_half, + ipac_msg_idreq_other_half); + LOGCLI(osc, "\t(msg dump: %s)\n", osmo_hexdump(data, ipac_msg_idreq_other_half)); + osmo_stream_cli_send(osc, msg); + } else if (ipa_msg_type == IPAC_MSGT_ID_RESP) { + LOGCLI(osc, "\tresult= %s\n", osmo_hexdump(msg->data, msg->len)); + LOGCLI(osc, "\texpected=%s\n", + osmo_hexdump(ipac_msg_idresp, sizeof(ipac_msg_idresp))); + } + printf("\n"); + return 0; +} + +static void *test_segm_run_client(void) +{ + struct osmo_stream_cli *osc; + void *ctx = talloc_named_const(NULL, 0, "test_segm_run_client"); + + (void) msgb_talloc_ctx_init(ctx, 0); + osc = osmo_stream_cli_create_iofd(ctx, "IPA test client"); + if (osc == NULL) { + fprintf(stderr, "osmo_stream_cli_create_iofd()\n"); + return NULL; + } + osmo_stream_cli_set_addr(osc, "127.0.0.11"); + osmo_stream_cli_set_port(osc, 1111); + osmo_stream_cli_set_connect_cb(osc, test_segm_cli_connect_cb); + osmo_stream_cli_set_data(osc, ctx); + osmo_stream_cli_set_iofd_read_cb(osc, test_segm_stream_cli_read_cb); + osmo_stream_cli_set_nodelay(osc, true); + if (osmo_stream_cli_open(osc) < 0) { + fprintf(stderr, "Cannot open stream client\n"); + return NULL; + } + + return NULL; +} + +static bool test_segm_ipa(void *ctx, const char *host, unsigned port, + struct osmo_stream_srv_link *srv) +{ + int rc; + const int max_test_duration = 2; + struct timespec start, now; + test_segm_run_client(); + + rc = clock_gettime(CLOCK_MONOTONIC, &start); + if (rc < 0) { + fprintf(stderr, "clock_gettime(): %s\n", strerror(errno)); + return EXIT_FAILURE; + } + while (!all_msgs_processed) { + osmo_gettimeofday_override_add(0, 1); /* small increment to easily spot iterations */ + osmo_select_main(1); + rc = clock_gettime(CLOCK_MONOTONIC, &now); + if (rc < 0) { + fprintf(stderr, "clock_gettime(): %s\n", strerror(errno)); + return EXIT_FAILURE; + } + if (now.tv_sec - start.tv_sec > max_test_duration) { + fprintf(stderr, "test_segm_ipa(): Timeout\n"); + osmo_stream_srv_link_unset_stream_proto(srv); + return EXIT_FAILURE; + } + } + + osmo_stream_srv_link_unset_stream_proto(srv); + return EXIT_SUCCESS; +} + +static unsigned msgnum_srv = 0; +int test_segm_stream_srv_read_cb(struct osmo_stream_srv *conn, struct msgb *msg) +{ + unsigned char *data; + struct msgb *m; + uint8_t *msgt = msg->data + sizeof(struct ipa_head); + LOGSRV(conn, "[%u-srv] Received message from stream (len = %" PRIu16 ")\n", + ++msgnum_srv, msgb_length(msg)); + msgb_pull(msg, sizeof(struct ipa_head)); + LOGSRV(conn, "\tIPA data: %s\n", osmo_hexdump(msg->data, msg->len)); + LOGSRV(conn, "\tType: %s\n", IPAC_MSG_TYPES[*msgt]); + LOGSRV(conn, "\t(msg dump: %s)\n", osmo_hexdump(msg->data - sizeof(struct ipa_head), + msg->len + sizeof(struct ipa_head))); + if (*msgt == IPAC_MSGT_ID_RESP) { /* */ + LOGSRV(conn, "Send IPAC_MSGT_ID_GET to trigger client to send 2nd half\n\n"); + m = msgb_alloc_headroom(128, 0, "IPA messages"); + if (m == NULL) { + fprintf(stderr, "Cannot allocate message\n"); + return -ENOMEM; + } + put_ipa_msg(data, m, ipac_msg_idreq); + osmo_stream_srv_send(conn, m); + } else if (*msgt == IPAC_MSGT_ID_GET) { + all_msgs_processed = true; + } + return 0; +} + +static int test_segm_stream_srv_accept_cb(struct osmo_stream_srv_link *srv, int fd) +{ + void *ctx = talloc_named_const(NULL, 0, "test_segm_stream_srv_accept_cb"); + struct osmo_stream_srv *oss = + osmo_stream_srv_create_iofd(ctx, "srv link", srv, fd, + test_segm_stream_srv_read_cb, + close_cb_srv, NULL); + if (oss == NULL) { + fprintf(stderr, "Error while creating connection\n"); + return -1; + } + + return 0; +} int main(void) { struct osmo_stream_srv_link *srv; + bool rc; char *host = "127.0.0.11"; unsigned port = 1111; void *tall_test = talloc_named_const(NULL, 1, "osmo_stream_test"); @@ -402,8 +630,14 @@ test_recon(tall_test, host, port, 12, srv, true); test_recon(tall_test, host, port, 8, srv, false); - osmo_stream_srv_link_destroy(srv); + osmo_stream_srv_link_set_accept_cb(srv, + test_segm_stream_srv_accept_cb); + osmo_stream_srv_link_set_stream_proto(srv, OSMO_STREAM_IPAC); + if (osmo_stream_srv_link_open(srv) < 0) { + printf("Unable to open server\n"); + return EXIT_FAILURE; + } + rc = test_segm_ipa(tall_test, host, port, srv); printf("Stream tests completed\n"); - - return EXIT_SUCCESS; + return rc; } diff --git a/tests/stream/stream_test.ok b/tests/stream/stream_test.ok index 2106295..7763212 100644 --- a/tests/stream/stream_test.ok +++ b/tests/stream/stream_test.ok @@ -48,4 +48,46 @@ {11.000019} [OK] Client's read_cb_cli(): 0-byte read, auto-reconnect will be triggered if enabled {20.000019} non-reconnecting test complete. +Connect callback triggered (segmentation test) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): [01-cli] Appending msg of type IPAC_MSGT_PING into buffer +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (msg dump: 00 01 fe 00 ) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): [02-cli] Appending msg of type IPAC_MSGT_PONG into buffer +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (msg dump: 00 01 fe 01 ) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): [03-cli] Appending msg of type IPAC_MSGT_PING into buffer +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (msg dump: 00 01 fe 00 ) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): [04-cli] Appending msg of type IPAC_MSGT_ID_RESP into buffer +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (msg dump: 00 07 fe 05 01 01 de ad be ef ) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): [04.5-cli] Appending *half* of msg of type IPAC_MSGT_ID_GET into buffer +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (dump: 00 03 fe ) +{20.000020} [OK] Client's test_segm_cli_connect_cb(): Sending 4 1/2 messages as one: +{20.000020} [OK] Client's test_segm_cli_connect_cb(): (msg dump: 00 01 fe 00 00 01 fe 01 00 01 fe 00 00 07 fe 05 01 01 de ad be ef 00 03 fe ) + +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): [1-srv] Received message from stream (len = 4) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): IPA data: 00 +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): Type: IPAC_MSGT_PING +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): (msg dump: 00 01 fe 00 ) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): [2-srv] Received message from stream (len = 4) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): IPA data: 01 +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): Type: IPAC_MSGT_PONG +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): (msg dump: 00 01 fe 01 ) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): [3-srv] Received message from stream (len = 4) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): IPA data: 00 +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): Type: IPAC_MSGT_PING +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): (msg dump: 00 01 fe 00 ) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): [4-srv] Received message from stream (len = 10) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): IPA data: 05 01 01 de ad be ef +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): Type: IPAC_MSGT_ID_RESP +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): (msg dump: 00 07 fe 05 01 01 de ad be ef ) +{20.000022} [NA|OK] Server's test_segm_stream_srv_read_cb(): Send IPAC_MSGT_ID_GET to trigger client to send 2nd half + +{20.000024} [OK] Client's test_segm_stream_cli_read_cb(): Received message from stream (total len = 6) +{20.000024} [OK] Client's test_segm_stream_cli_read_cb(): Type: IPAC_MSGT_ID_GET +{20.000024} [OK] Client's test_segm_stream_cli_read_cb(): Got IPAC_MSGT_ID_GET from server +{20.000024} [OK] Client's test_segm_stream_cli_read_cb(): [5-cli] Sending second half of IPAC_MSGT_ID_GET +{20.000024} [OK] Client's test_segm_stream_cli_read_cb(): (msg dump: 04 01 01 ) + +{20.000026} [NA|OK] Server's test_segm_stream_srv_read_cb(): [5-srv] Received message from stream (len = 6) +{20.000026} [NA|OK] Server's test_segm_stream_srv_read_cb(): IPA data: 04 01 01 +{20.000026} [NA|OK] Server's test_segm_stream_srv_read_cb(): Type: IPAC_MSGT_ID_GET +{20.000026} [NA|OK] Server's test_segm_stream_srv_read_cb(): (msg dump: 00 03 fe 04 01 01 ) Stream tests completed -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33197 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I6c91ff385cb5f36ab6b6c96d0e44997995d0d24c Gerrit-Change-Number: 33197 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0

[M] Change in libosmo-netif[master]: stream: Move helper functions
by arehbein 06 Jun '23

06 Jun '23

arehbein has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmo-netif/+/33203 ) Change subject: stream: Move helper functions ...................................................................... stream: Move helper functions Prepare for next commits Change-Id: I318965538e5329c44d0910694621b5e1f1db0626 --- M src/stream.c 1 file changed, 32 insertions(+), 21 deletions(-) git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/03/33203/1 diff --git a/src/stream.c b/src/stream.c index 35a1fc7..30639e2 100644 --- a/src/stream.c +++ b/src/stream.c @@ -1173,6 +1173,27 @@ osmo_stream_cli_open(cli); } +/* msgb_l1(msg) is expected to be set */ +static inline enum ipaccess_proto msg_get_ipa_proto(struct msgb *msg) +{ + struct ipa_head *ih = msgb_l1(msg); + OSMO_ASSERT(ih); + return ih->proto; +} + +/* msgb->l1h is expected to be set, msgb->l2h is expected to be set if + * we have IPAC_PROTO_OSMO specified in the header. + * Returns the protocol extension (enum ipaccess_proto) or -ENOPROTOOPT if + * we don't have IPAC_PROTO_OSMO specified in the IPA header */ +static inline int msg_get_ipa_proto_ext(struct msgb *msg) +{ + if (msg_get_ipa_proto(msg) != IPAC_PROTO_OSMO) + return -ENOPROTOOPT; + struct ipa_head_ext *ihe = msgb_l2(msg); + OSMO_ASSERT(ihe); + return ihe->proto; +} + /*! \brief Enqueue data to be sent via an Osmocom stream client * \param[in] cli Stream Client through which we want to send * \param[in] msg Message buffer to enqueue in transmit queue */ @@ -1926,27 +1947,6 @@ talloc_free(conn); } -/* msgb_l1(msg) is expected to be set */ -static inline enum ipaccess_proto msg_get_ipa_proto(struct msgb *msg) -{ - struct ipa_head *ih = msgb_l1(msg); - OSMO_ASSERT(ih); - return ih->proto; -} - -/* msgb->l1h is expected to be set, msgb->l2h is expected to be set if - * we have IPAC_PROTO_OSMO specified in the header. - * Returns the protocol extension (enum ipaccess_proto) or -ENOPROTOOPT if - * we don't have IPAC_PROTO_OSMO specified in the IPA header */ -static inline int msg_get_ipa_proto_ext(struct msgb *msg) -{ - if (msg_get_ipa_proto(msg) != IPAC_PROTO_OSMO) - return -ENOPROTOOPT; - struct ipa_head_ext *ihe = msgb_l2(msg); - OSMO_ASSERT(ihe); - return ihe->proto; -} - /*! \brief Enqueue IPA data to be sent via an Osmocom stream server * \param[in] conn Stream Server through which we want to send * \param[in] p Protocol transported by IPA. When set to IPAC_PROTO_UNSPECIFIED, the protocol will be -- To view, visit https://gerrit.osmocom.org/c/libosmo-netif/+/33203 To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings Gerrit-Project: libosmo-netif Gerrit-Branch: master Gerrit-Change-Id: I318965538e5329c44d0910694621b5e1f1db0626 Gerrit-Change-Number: 33203 Gerrit-PatchSet: 1 Gerrit-Owner: arehbein <arehbein(a)sysmocom.de> Gerrit-MessageType: newchange

1 0