}
#ifdef VCL_TEST
- client_fd = vppcom_session_accept (ssm->listen_fd, &conn->endpt,
+ client_fd = vppcom_session_accept (ssm->listen_fd, &conn->endpt, 0,
-1.0 /* wait forever */ );
if (client_fd < 0)
errno = -client_fd;
* RETURN: 0 on success -1 on failure
* */
static inline int
-vcom_init ()
+vcom_init (void)
{
pid_t pid = getpid ();
return libc_send (__fd, __buf, __n, __flags);
}
+ssize_t
+sendfile (int __out_fd, int __in_fd, off_t * __offset, size_t __len)
+{
+ ssize_t size;
+
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] __out_fd %d, __in_fd %d, __offset %p, __len %ld",
+ getpid (), __out_fd, __in_fd, __offset, __len);
+
+ if (is_vcom_socket_fd (__out_fd))
+ {
+ /* TBD: refactor this check to be part of is_vcom_socket_fd() */
+ if (vcom_init () != 0)
+ return -1;
+
+ size = vcom_socket_sendfile (__out_fd, __in_fd, __offset, __len);
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] vcom_socket_sendfile (out_fd %d, in_fd %d, "
+ "offset %p (%ld), len %lu) returned %ld",
+ getpid (), __out_fd, __in_fd, __offset,
+ __offset ? *__offset : -1, __len, size);
+ if (size < 0)
+ {
+ errno = -size;
+ return -1;
+ }
+ return size;
+ }
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] calling libc_sendfile!", getpid ());
+ return libc_sendfile (__out_fd, __in_fd, __offset, __len);
+}
+
+ssize_t
+sendfile64 (int __out_fd, int __in_fd, off_t * __offset, size_t __len)
+{
+ return sendfile (__out_fd, __in_fd, __offset, __len);
+}
+
+
/*
* Read N bytes into BUF from socket FD.
* Returns the number read or -1 for errors.
rv =
vcom_socket_epoll_pwait (__epfd, __events, __maxevents, __timeout, NULL);
- if (VCOM_DEBUG > 1)
+ if (VCOM_DEBUG > 2)
fprintf (stderr,
"[%d] epoll_wait: "
"'%04d'='%04d', '%p', "
#define VCOM_DEBUG 0
#endif
+#include <vppinfra/error.h>
+#include <vppinfra/types.h>
#include <vcl/vcom_glibc_socket.h>
#define MAX_VCOM_APP_NAME 256
socklen_t * __restrict __len);
extern ssize_t
-vcom_send (int __fd, const void *__buf, size_t __n, int __flags);
+vcom_sendfile (int __out_fd, int __in_fd, off_t * __offset, int __len);
extern ssize_t vcom_recv (int __fd, void *__buf, size_t __n, int __flags);
__THROW. */
extern ssize_t send (int __fd, const void *__buf, size_t __n, int __flags);
+extern ssize_t sendfile (int __out_fd, int __in_fd, off_t * __offset,
+ size_t __len);
+
/* Read N bytes into BUF from socket FD.
Returns the number read or -1 for errors.
/* Hash table - key:fd, value:vec of epitemidx */
uword *epitemidxs_by_fd;
+ u8 *io_buffer;
} vcom_socket_main_t;
vcom_socket_main_t vcom_socket_main;
return vcom_socket_sendto (__fd, __buf, __n, __flags, NULL, 0);
}
+/* NOTE: this function is not thread safe or 32-bit friendly */
+ssize_t
+vcom_socket_sendfile (int __out_fd, int __in_fd, off_t * __offset,
+ size_t __len)
+{
+ vcom_socket_main_t *vsm = &vcom_socket_main;
+ uword *p;
+ vcom_socket_t *vsock;
+ size_t n_bytes_left = __len;
+ u32 out_sockidx, out_sid = ~0;
+ size_t bytes_to_read;
+ int nbytes;
+ int rv, errno_val;
+ ssize_t results = 0;
+ u8 eagain = 0;
+
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] __out_fd %d, __in_fd %d, __offset %p, __len %lu",
+ getpid (), __out_fd, __in_fd, __offset, __len);
+
+ p = hash_get (vsm->sockidx_by_fd, __out_fd);
+ if (!p)
+ {
+ clib_warning ("[%d] ERROR: invalid __out_fd (%d), fd lookup failed!",
+ getpid (), __len);
+ return -EBADF;
+ }
+ out_sockidx = p[0];
+ vsock = pool_elt_at_index (vsm->vsockets, out_sockidx);
+ if (!vsock)
+ {
+ clib_warning ("[%d] ERROR: invalid __out_fd (%d) / out_sockidx %u, "
+ "missing vsock pool element!",
+ getpid (), __len, out_sockidx);
+ return -ENOTSOCK;
+ }
+ out_sid = vsock->sid;
+ if (vsock->type != SOCKET_TYPE_VPPCOM_BOUND)
+ {
+ clib_warning ("[%d] ERROR: __out_fd (%d), socket (sid %u) "
+ "is not VCL bound!", getpid (), __out_fd, out_sid);
+ return -EINVAL;
+ }
+
+ if (__offset)
+ {
+ off_t offset = lseek (__in_fd, *__offset, SEEK_SET);
+ if (offset == -1)
+ {
+ errno_val = errno;
+ perror ("lseek()");
+ clib_warning ("[%d] ERROR: lseek SEEK_SET failed: "
+ "in_fd %d, offset %p (%ld), rv %ld, errno %d",
+ getpid (), __in_fd, __offset, *__offset, offset,
+ errno_val);
+ return -errno_val;
+ }
+
+ ASSERT (offset == *__offset);
+ }
+
+ do
+ {
+ bytes_to_read = vppcom_session_attr (out_sid,
+ VPPCOM_ATTR_GET_NWRITE, 0, 0);
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] results %ld, n_bytes_left %lu, "
+ "bytes_to_read %lu", getpid (), results,
+ n_bytes_left, bytes_to_read);
+ if (bytes_to_read == 0)
+ {
+ u32 flags, flags_len = sizeof (flags);
+ rv = vppcom_session_attr (out_sid, VPPCOM_ATTR_GET_FLAGS, &flags,
+ &flags_len);
+ ASSERT (rv == VPPCOM_OK);
+
+ if (flags & O_NONBLOCK)
+ {
+ if (!results)
+ {
+ if (VCOM_DEBUG > 2)
+ clib_warning ("[%d] EAGAIN", getpid ());
+ eagain = 1;
+ }
+ goto update_offset;
+ }
+ else
+ continue;
+ }
+ bytes_to_read = clib_min (n_bytes_left, bytes_to_read);
+ vec_validate (vsm->io_buffer, bytes_to_read);
+ nbytes = libc_read (__in_fd, vsm->io_buffer, bytes_to_read);
+ if (nbytes < 0)
+ {
+ errno_val = errno;
+ perror ("read()");
+ clib_warning ("[%d] ERROR: libc_read (__in_fd (%d), "
+ "io_buffer %p, bytes_to_read %lu) returned "
+ "errno %d",
+ getpid (), __in_fd, vsm->io_buffer,
+ bytes_to_read, errno_val);
+ if (results == 0)
+ {
+ vec_reset_length (vsm->io_buffer);
+ return -errno_val;
+ }
+ goto update_offset;
+ }
+ rv = vppcom_session_write (out_sid, vsm->io_buffer, nbytes);
+ if (rv < 0)
+ {
+ clib_warning ("[%d] ERROR: vppcom_session_write ("
+ "out_sid %u, io_buffer %p, nbytes %d) returned %d",
+ getpid (), out_sid, vsm->io_buffer, nbytes, rv);
+ if (results == 0)
+ {
+ vec_reset_length (vsm->io_buffer);
+ return rv;
+ }
+ goto update_offset;
+ }
+
+ results += nbytes;
+ ASSERT (n_bytes_left >= nbytes);
+ n_bytes_left = n_bytes_left - nbytes;
+ }
+ while (n_bytes_left > 0);
+
+update_offset:
+ if (__offset)
+ {
+ off_t offset = lseek (__in_fd, *__offset, SEEK_SET);
+ if (offset == -1)
+ {
+ errno_val = errno;
+ perror ("lseek()");
+ clib_warning ("[%d] ERROR: lseek (__in_fd %d, __offset %p "
+ "(%ld), SEEK_SET) returned errno %d",
+ getpid (), __in_fd, __offset, *__offset, errno_val);
+ vec_reset_length (vsm->io_buffer);
+ return -errno_val;
+ }
+
+ *__offset += results + 1;
+ }
+
+ vec_reset_length (vsm->io_buffer);
+ return eagain ? -EAGAIN : results;
+}
+
ssize_t
vcom_socket_recv (int __fd, void *__buf, size_t __n, int __flags)
{
* on the queue, accept () blocks the caller
* until a connection is present.
*/
- rv = vppcom_session_accept (vsock->sid, &ep,
+ rv = vppcom_session_accept (vsock->sid, &ep, flags,
-1.0 /* wait forever */ );
}
else
* block.
* */
/* is non blocking */
- rv = vppcom_session_accept (vsock->sid, &ep, 0);
+ rv = vppcom_session_accept (vsock->sid, &ep, flags, 0);
/* If the socket is marked nonblocking and
* no pending connections are present on the
* queue, accept fails with the error
ssize_t vcom_socket_recv (int __fd, void *__buf, size_t __n, int __flags);
+ssize_t
+vcom_socket_sendfile (int __out_fd, int __in_fd, off_t * __offset,
+ size_t __len);
+
/*
* RETURN 1 if __fd is (SOCK_STREAM, SOCK_SEQPACKET),
* 0 otherwise
int flags);
typedef int (*__libc_send) (int sockfd, const void *buf, size_t len,
int flags);
+typedef ssize_t (*__libc_sendfile) (int out_fd, int in_fd, off_t * offset,
+ size_t len);
typedef int (*__libc_sendmsg) (int sockfd, const struct msghdr * msg,
int flags);
typedef int (*__libc_sendto) (int sockfd, const void *buf, size_t len,
SWRAP_SYMBOL_ENTRY (recvfrom);
SWRAP_SYMBOL_ENTRY (recvmsg);
SWRAP_SYMBOL_ENTRY (send);
+ SWRAP_SYMBOL_ENTRY (sendfile);
SWRAP_SYMBOL_ENTRY (sendmsg);
SWRAP_SYMBOL_ENTRY (sendto);
SWRAP_SYMBOL_ENTRY (setsockopt);
return swrap.libc.symbols._libc_listen.f (sockfd, backlog);
}
+/* TBD: libc_read() should return ssize_t not an int */
int
libc_read (int fd, void *buf, size_t count)
{
return swrap.libc.symbols._libc_send.f (sockfd, buf, len, flags);
}
+ssize_t
+libc_sendfile (int out_fd, int in_fd, off_t * offset, size_t len)
+{
+ swrap_bind_symbol_libc (sendfile);
+
+ return swrap.libc.symbols._libc_sendfile.f (out_fd, in_fd, offset, len);
+}
+
int
libc_sendmsg (int sockfd, const struct msghdr *msg, int flags)
{
int libc_send (int sockfd, const void *buf, size_t len, int flags);
+ssize_t libc_sendfile (int out_fd, int in_fd, off_t * offset, size_t len);
+
int libc_sendmsg (int sockfd, const struct msghdr *msg, int flags);
int
}
else
{
- clib_warning ("[%d] couldn't find session: unknonwn vpp handle 0x%llx",
+ clib_warning ("[%d] couldn't find session: unknown vpp handle 0x%llx",
getpid (), mp->handle);
rv = -11;
}
svm_fifo_t *rx_fifo, *tx_fifo;
session_t *session;
u32 session_index;
+ uword *p;
clib_spinlock_lock (&vcm->sessions_lockp);
if (!clib_fifo_free_elts (vcm->client_session_index_fifo))
/* Add it to lookup table */
hash_set (vcm->session_index_by_vpp_handles, mp->handle, session_index);
+ p = hash_get (vcm->session_index_by_vpp_handles, mp->listener_handle);
+ if (p)
+ {
+ int rval;
+ session_t *listen_session;
+
+ rval = vppcom_session_at_index (p[0], &listen_session);
+ if (PREDICT_FALSE (rval))
+ {
+ if (VPPCOM_DEBUG > 1)
+ clib_warning ("[%d] invalid listen session, sid (%u) "
+ "has been closed!", getpid (), p[0]);
+ }
+ else
+ {
+ session->lcl_port = listen_session->lcl_port;
+ session->lcl_addr = listen_session->lcl_addr;
+ }
+ clib_spinlock_unlock (&vcm->sessions_lockp);
+ hash_unset (vcm->session_index_by_vpp_handles, mp->handle);
+ }
+ else
+ {
+ clib_warning ("[%d] couldn't find listen session: unknown vpp "
+ "listener handle %llx", getpid (), mp->listener_handle);
+ }
+
+ /* TBD: move client_session_index_fifo into listener session */
clib_fifo_add1 (vcm->client_session_index_fifo, session_index);
clib_spinlock_unlock (&vcm->sessions_lockp);
}
static void
-vppcom_send_connect_session_reply (session_t * session, u32 context,
- int retval, int handle)
+vppcom_send_connect_session_reply (session_t * session, int retval)
{
vl_api_connect_session_reply_t *rmp;
u32 len;
memset (rmp, 0, sizeof (*rmp));
rmp->_vl_msg_id = ntohs (VL_API_CONNECT_SESSION_REPLY);
- rmp->context = session ? session->client_context : context;
+ rmp->context = session->client_context;
rmp->retval = htonl (retval);
- rmp->handle = session ? session->vpp_handle : handle;
-
- if (session)
- {
- rmp->server_rx_fifo = pointer_to_uword (session->server_rx_fifo);
- rmp->server_tx_fifo = pointer_to_uword (session->server_tx_fifo);
- rmp->vpp_event_queue_address =
- pointer_to_uword (session->vpp_event_queue);
- rmp->segment_size = vcm->cfg.segment_size;
- len = vec_len (session->segment_name);
- rmp->segment_name_length = clib_min (len, sizeof (rmp->segment_name));
- clib_memcpy (rmp->segment_name, session->segment_name,
- rmp->segment_name_length - 1);
- clib_memcpy (rmp->lcl_ip, session->lcl_addr.ip46.as_u8,
- sizeof (rmp->lcl_ip));
- rmp->is_ip4 = session->lcl_addr.is_ip4;
- }
-
+ rmp->handle = session->vpp_handle;
+ rmp->server_rx_fifo = pointer_to_uword (session->server_rx_fifo);
+ rmp->server_tx_fifo = pointer_to_uword (session->server_tx_fifo);
+ rmp->vpp_event_queue_address = pointer_to_uword (session->vpp_event_queue);
+ rmp->segment_size = vcm->cfg.segment_size;
+ len = vec_len (session->segment_name);
+ rmp->segment_name_length = clib_min (len, sizeof (rmp->segment_name));
+ clib_memcpy (rmp->segment_name, session->segment_name,
+ rmp->segment_name_length - 1);
+ clib_memcpy (rmp->lcl_ip, session->lcl_addr.ip46.as_u8,
+ sizeof (rmp->lcl_ip));
+ rmp->is_ip4 = session->lcl_addr.is_ip4;
client_q = uword_to_pointer (session->client_queue_address,
unix_shared_memory_queue_t *);
ASSERT (client_q);
rv = vppcom_wait_for_session_state_change (session_index,
STATE_DISCONNECT, 1.0);
+ /* TBD: Force clean up on error/timeout since there is no other
+ * way to recover from a failed disconnect.
+ */
if ((VPPCOM_DEBUG > 0) && (rv < 0))
clib_warning ("[%d] disconnect (session %d) failed, rv = %s (%d)",
getpid (), session_index, vppcom_retval_str (rv), rv);
}
else
- clib_spinlock_unlock (&vcm->sessions_lockp);
+ {
+ /* TBD: Handle cut-thru disconnect */
+ clib_spinlock_unlock (&vcm->sessions_lockp);
+ }
return VPPCOM_OK;
}
int
vppcom_session_accept (uint32_t listen_session_index, vppcom_endpt_t * ep,
- double wait_for_time)
+ uint32_t flags, double wait_for_time)
{
session_t *listen_session = 0;
session_t *client_session = 0;
- u32 client_session_index;
+ u32 client_session_index = ~0;
int rv;
f64 wait_for;
char *cut_thru_str;
ASSERT (client_session->peer_addr.is_ip4 ==
listen_session->lcl_addr.is_ip4);
+ client_session->is_nonblocking = (flags & O_NONBLOCK) ? 1 : 0;
if (VPPCOM_DEBUG > 0)
- clib_warning ("[%d] Got a request: client sid %d", getpid (),
- client_session_index);
+ clib_warning ("[%d] Got a request: client sid %d, flags %d, "
+ " is_nonblocking %u", getpid (), client_session_index,
+ flags, client_session->is_nonblocking);
ep->vrf = client_session->vrf;
ep->is_cut_thru = client_session->is_cut_thru;
getpid (), a->segment_name);
vec_reset_length (a->new_segment_indices);
rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
- vppcom_send_connect_session_reply (client_session, 0, rv, 0);
+ vppcom_send_connect_session_reply (client_session, rv);
clib_spinlock_unlock (&vcm->sessions_lockp);
return VPPCOM_ENOMEM;
}
getpid (), vcm->cfg.rx_fifo_size,
vcm->cfg.rx_fifo_size);
rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
- vppcom_send_connect_session_reply (client_session, 0, rv, 0);
+ vppcom_send_connect_session_reply (client_session, rv);
clib_spinlock_unlock (&vcm->sessions_lockp);
return VPPCOM_ENOMEM;
}
getpid (), vcm->cfg.tx_fifo_size,
vcm->cfg.tx_fifo_size);
rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
- vppcom_send_connect_session_reply (client_session, 0, rv, 0);
+ vppcom_send_connect_session_reply (client_session, rv);
clib_spinlock_unlock (&vcm->sessions_lockp);
return VPPCOM_ENOMEM;
}
ssvm_unlock_non_recursive (sh);
}
#endif
- vppcom_send_connect_session_reply (client_session, 0, 0, 0);
+ vppcom_send_connect_session_reply (client_session, 0);
}
else
{
{
case VPPCOM_ATTR_GET_NREAD:
rv = vppcom_session_read_ready (session, session_index);
- if (VPPCOM_DEBUG > 1)
- clib_warning ("[%d] VPPCOM_ATTR_GET_NREAD: nread = %d",
+ if (VPPCOM_DEBUG > 2)
+ clib_warning ("[%d] VPPCOM_ATTR_GET_NREAD: sid %u, nread = %d",
getpid (), rv);
break;
- case VPPCOM_ATTR_PEEK_NREAD:
- /* TBD */
+ case VPPCOM_ATTR_GET_NWRITE:
+ rv = vppcom_session_write_ready (session, session_index);
+ if (VPPCOM_DEBUG > 2)
+ clib_warning ("[%d] VPPCOM_ATTR_GET_NWRITE: sid %u, nwrite = %d",
+ getpid (), session_index, rv);
+
break;
case VPPCOM_ATTR_GET_FLAGS:
{
*flags = O_RDWR | ((session->is_nonblocking) ? O_NONBLOCK : 0);
*buflen = sizeof (*flags);
- if (VPPCOM_DEBUG > 1)
- clib_warning ("[%d] VPPCOM_ATTR_GET_FLAGS: flags = 0x%08x, "
- "is_nonblocking = %u", getpid (), *flags,
- session->is_nonblocking);
+ if (VPPCOM_DEBUG > 2)
+ clib_warning ("[%d] VPPCOM_ATTR_GET_FLAGS: sid %u, "
+ "flags = 0x%08x, is_nonblocking = %u", getpid (),
+ session_index, *flags, session->is_nonblocking);
}
else
rv = VPPCOM_EINVAL;
if (buffer && buflen && (*buflen >= sizeof (*flags)))
{
session->is_nonblocking = (*flags & O_NONBLOCK) ? 1 : 0;
- if (VPPCOM_DEBUG > 1)
- clib_warning ("[%d] VPPCOM_ATTR_SET_FLAGS: flags = 0x%08x, "
- "is_nonblocking = %u", getpid (), *flags,
- session->is_nonblocking);
+ if (VPPCOM_DEBUG > 2)
+ clib_warning ("[%d] VPPCOM_ATTR_SET_FLAGS: sid %u, "
+ "flags = 0x%08x, is_nonblocking = %u",
+ getpid (), *flags, session->is_nonblocking);
}
else
rv = VPPCOM_EINVAL;
sizeof (ip6_address_t));
*buflen = sizeof (*ep);
if (VPPCOM_DEBUG > 1)
- clib_warning ("[%d] VPPCOM_ATTR_GET_PEER_ADDR: sid %u is_ip4 = "
+ clib_warning ("[%d] VPPCOM_ATTR_GET_PEER_ADDR: sid %u, is_ip4 = "
"%u, addr = %U, port %u", getpid (),
session_index, ep->is_ip4, format_ip46_address,
&session->peer_addr.ip46, ep->is_ip4,
sizeof (ip6_address_t));
*buflen = sizeof (*ep);
if (VPPCOM_DEBUG > 1)
- clib_warning ("[%d] VPPCOM_ATTR_GET_LCL_ADDR: sid %u is_ip4 = "
+ clib_warning ("[%d] VPPCOM_ATTR_GET_LCL_ADDR: sid %u, is_ip4 = "
"%u, addr = %U port %d", getpid (),
session_index, ep->is_ip4, format_ip46_address,
&session->lcl_addr.ip46, ep->is_ip4,
typedef enum
{
VPPCOM_ATTR_GET_NREAD,
- VPPCOM_ATTR_PEEK_NREAD,
+ VPPCOM_ATTR_GET_NWRITE,
VPPCOM_ATTR_GET_FLAGS,
VPPCOM_ATTR_SET_FLAGS,
VPPCOM_ATTR_GET_LCL_ADDR,
extern int vppcom_session_listen (uint32_t session_index, uint32_t q_len);
extern int vppcom_session_accept (uint32_t session_index,
vppcom_endpt_t * client_ep,
- double wait_for_time);
+ uint32_t flags, double wait_for_time);
extern int vppcom_session_connect (uint32_t session_index,
vppcom_endpt_t * server_ep);
Code Review / vpp.git / commitdiff