server: use the new reds_stream_{read,write}

https://bugs.freedesktop.org/show_bug.cgi?id=34795

5 files changed, 53 insertions, 61 deletions

diff --git a/server/red_channel.c b/server/red_channel.c
index f0640549..60eb443e 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -42,7 +42,8 @@ static int red_peer_receive(RedsStream *peer, uint8_t *buf, uint32_t size)
         if (peer->shutdown) {
             return -1;
         }
-        if ((now = peer->cb_read(peer->ctx, pos, size)) <= 0) {
+        now = reds_stream_read(peer, pos, size);
+        if (now <= 0) {
             if (now == 0) {
                 return -1;
             }
@@ -140,7 +141,8 @@ static struct iovec *__iovec_skip(struct iovec vec[], int skip, int *vec_size)
 
 static void red_peer_handle_outgoing(RedsStream *peer, OutgoingHandler *handler)
 {
-    int n;
+    ssize_t n;
+
     if (handler->size == 0) {
         handler->vec = handler->vec_buf;
         handler->size = handler->get_msg_size(handler->opaque);
@@ -149,8 +151,10 @@ static void red_peer_handle_outgoing(RedsStream *peer, OutgoingHandler *handler)
         }
         handler->prepare(handler->opaque, handler->vec, &handler->vec_size);
     }
+
     for (;;) {
-        if ((n = peer->cb_writev(peer->ctx, handler->vec, handler->vec_size)) == -1) {
+        n = reds_stream_writev(peer, handler->vec, handler->vec_size);
+        if (n == -1) {
             switch (errno) {
             case EAGAIN:
                 handler->on_block(handler->opaque);
diff --git a/server/red_worker.c b/server/red_worker.c
index 84697b3d..a1c18b6b 100644
--- a/server/red_worker.c
+++ b/server/red_worker.c
@@ -7316,9 +7316,9 @@ static void inline channel_release_res(RedChannel *channel)
 static void red_send_data(RedChannel *channel, void *item)
 {
     for (;;) {
-        uint32_t n = channel->send_data.size - channel->send_data.pos;
+        ssize_t n = channel->send_data.size - channel->send_data.pos;
         struct iovec vec[MAX_SEND_VEC];
-        int vec_size;
+        size_t vec_size;
 
         if (!n) {
             channel->send_data.blocked = FALSE;
@@ -7331,7 +7331,8 @@ static void red_send_data(RedChannel *channel, void *item)
         vec_size = spice_marshaller_fill_iovec(channel->send_data.marshaller,
                                                vec, MAX_SEND_VEC, channel->send_data.pos);
         ASSERT(channel->peer);
-        if ((n = channel->peer->cb_writev(channel->peer->ctx, vec, vec_size)) == -1) {
+        n = reds_stream_writev(channel->peer, vec, vec_size);
+        if (n == -1) {
             switch (errno) {
             case EAGAIN:
                 channel->send_data.blocked = TRUE;
@@ -9239,7 +9240,8 @@ static void red_receive(RedChannel *channel)
         n = channel->recive_data.end - channel->recive_data.now;
         ASSERT(n);
         ASSERT(channel->peer);
-        if ((n = channel->peer->cb_read(channel->peer->ctx, channel->recive_data.now, n)) <= 0) {
+        n = reds_stream_read(channel->peer, channel->recive_data.now, n);
+        if (n <= 0) {
             if (n == 0) {
                 channel->disconnect(channel);
                 return;
diff --git a/server/reds.c b/server/reds.c
index 19da2ae8..beda0e3d 100644
--- a/server/reds.c
+++ b/server/reds.c
@@ -391,73 +391,60 @@ static void reds_channel_event(RedsStream *peer, int event)
     core->channel_event(event, &peer->info);
 }
 
-static int reds_write(void *ctx, void *buf, size_t size)
+static ssize_t stream_write_cb(RedsStream *s, const void *buf, size_t size)
 {
-    int return_code;
-    int sock = (long)ctx;
-    size_t count = size;
-
-    return_code = write(sock, buf, count);
-
-    return (return_code);
+    return write(s->socket, buf, size);
 }
 
-static int reds_read(void *ctx, void *buf, size_t size)
+static ssize_t stream_writev_cb(RedsStream *s, const struct iovec *iov, int iovcnt)
 {
-    int return_code;
-    int sock = (long)ctx;
-    size_t count = size;
-
-    return_code = read(sock, buf, count);
+    return writev(s->socket, iov, iovcnt);
+}
 
-    return (return_code);
+static ssize_t stream_read_cb(RedsStream *s, void *buf, size_t size)
+{
+    return read(s->socket, buf, size);
 }
 
-static int reds_ssl_write(void *ctx, void *buf, size_t size)
+static ssize_t stream_ssl_write_cb(RedsStream *s, const void *buf, size_t size)
 {
     int return_code;
-    int ssl_error;
-    SSL *ssl = ctx;
+    SPICE_GNUC_UNUSED int ssl_error;
 
-    return_code = SSL_write(ssl, buf, size);
+    return_code = SSL_write(s->ssl, buf, size);
 
     if (return_code < 0) {
-        ssl_error = SSL_get_error(ssl, return_code);
-        (void)ssl_error;
+        ssl_error = SSL_get_error(s->ssl, return_code);
     }
 
-    return (return_code);
+    return return_code;
 }
 
-static int reds_ssl_read(void *ctx, void *buf, size_t size)
+static ssize_t stream_ssl_read_cb(RedsStream *s, void *buf, size_t size)
 {
     int return_code;
-    int ssl_error;
-    SSL *ssl = ctx;
+    SPICE_GNUC_UNUSED int ssl_error;
 
-    return_code = SSL_read(ssl, buf, size);
+    return_code = SSL_read(s->ssl, buf, size);
 
     if (return_code < 0) {
-        ssl_error = SSL_get_error(ssl, return_code);
-        (void)ssl_error;
+        ssl_error = SSL_get_error(s->ssl, return_code);
     }
 
-    return (return_code);
+    return return_code;
 }
 
-static int reds_ssl_writev(void *ctx, const struct iovec *vector, int count)
+static ssize_t stream_ssl_writev_cb(RedsStream *s, const struct iovec *vector, int count)
 {
     int i;
     int n;
-    int return_code = 0;
+    ssize_t return_code = 0;
     int ssl_error;
-    SSL *ssl = ctx;
 
     for (i = 0; i < count; ++i) {
-        n = SSL_write(ssl, vector[i].iov_base, vector[i].iov_len);
+        n = SSL_write(s->ssl, vector[i].iov_base, vector[i].iov_len);
         if (n <= 0) {
-            ssl_error = SSL_get_error(ssl, n);
-            (void)ssl_error;
+            ssl_error = SSL_get_error(s->ssl, n);
             if (return_code <= 0) {
                 return n;
             } else {
@@ -786,7 +773,7 @@ static int handle_incoming(RedsStream *peer, IncomingHandler *handler)
         uint8_t *end = buf + pos;
         SpiceDataHeader *header;
         int n;
-        n = peer->cb_read(peer->ctx, buf + pos, RECIVE_BUF_SIZE - pos);
+        n = reds_stream_read(peer, buf + pos, RECIVE_BUF_SIZE - pos);
         if (n <= 0) {
             if (n == 0) {
                 return -1;
@@ -840,7 +827,8 @@ static int handle_outgoing(RedsStream *peer, OutgoingHandler *handler)
     while (handler->length) {
         int n;
 
-        n = peer->cb_write(peer->ctx, handler->now, handler->length);
+        n = reds_stream_write(peer, handler->now, handler->length);
+
         if (n <= 0) {
             if (n == 0) {
                 return -1;
@@ -880,7 +868,7 @@ static int outgoing_write(RedsStream *peer, OutgoingHandler *handler, void *in_d
     }
 
     while (length) {
-        int n = peer->cb_write(peer->ctx, data, length);
+        int n = reds_stream_write(peer, data, length);
         if (n < 0) {
             switch (errno) {
             case EAGAIN:
@@ -1831,7 +1819,7 @@ static int reds_send_data()
 
     ASSERT(outgoing->vec_size);
     for (;;) {
-        if ((n = reds->peer->cb_writev(reds->peer->ctx, outgoing->vec, outgoing->vec_size)) == -1) {
+        if ((n = reds_stream_writev(reds->peer, outgoing->vec, outgoing->vec_size)) == -1) {
             switch (errno) {
             case EAGAIN:
                 core->watch_update_mask(reds->peer->watch,
@@ -1905,7 +1893,7 @@ static int sync_write(RedsStream *peer, void *in_buf, size_t n)
 {
     uint8_t *buf = (uint8_t *)in_buf;
     while (n) {
-        int now = peer->cb_write(peer->ctx, buf, n);
+        int now = reds_stream_write(peer, buf, n);
         if (now <= 0) {
             if (now == -1 && (errno == EINTR || errno == EAGAIN)) {
                 continue;
@@ -2633,7 +2621,8 @@ static void async_read_handler(int fd, int event, void *data)
         int n = obj->end - obj->now;
 
         ASSERT(n > 0);
-        if ((n = obj->peer->cb_read(obj->peer->ctx, obj->now, n)) <= 0) {
+        n = reds_stream_read(obj->peer, obj->now, n);
+        if (n <= 0) {
             if (n < 0) {
                 switch (errno) {
                 case EAGAIN:
@@ -2859,9 +2848,9 @@ static RedLinkInfo *reds_accept_connection(int listen_socket)
     }
     peer = link->peer;
     peer->ctx = (void *)((unsigned long)link->peer->socket);
-    peer->cb_read = (int (*)(void *, void *, int))reds_read;
-    peer->cb_write = (int (*)(void *, void *, int))reds_write;
-    peer->cb_writev = (int (*)(void *, const struct iovec *vector, int count))writev;
+    peer->read = stream_read_cb;
+    peer->write = stream_write_cb;
+    peer->writev = stream_writev_cb;
 
     return link;
 }
@@ -2894,9 +2883,9 @@ static void reds_accept_ssl_connection(int fd, int event, void *data)
     SSL_set_bio(link->peer->ssl, sbio, sbio);
 
     link->peer->ctx = (void *)(link->peer->ssl);
-    link->peer->cb_write = (int (*)(void *, void *, int))reds_ssl_write;
-    link->peer->cb_read = (int (*)(void *, void *, int))reds_ssl_read;
-    link->peer->cb_writev = reds_ssl_writev;
+    link->peer->write = stream_ssl_write_cb;
+    link->peer->read = stream_ssl_read_cb;
+    link->peer->writev = stream_ssl_writev_cb;
 
     return_code = SSL_accept(link->peer->ssl);
     if (return_code == 1) {
diff --git a/server/reds.h b/server/reds.h
index 2a9a0def..8913e83e 100644
--- a/server/reds.h
+++ b/server/reds.h
@@ -39,11 +39,6 @@ struct RedsStream {
 
     SpiceChannelEventInfo info;
 
-    int (*cb_write)(void *, void *, int);
-    int (*cb_read)(void *, void *, int);
-
-    int (*cb_writev)(void *, const struct iovec *vector, int count);
-
     /* private */
     ssize_t (*read)(RedsStream *s, void *buf, size_t nbyte);
     ssize_t (*write)(RedsStream *s, const void *buf, size_t nbyte);
diff --git a/server/snd_worker.c b/server/snd_worker.c
index fcfaf978..80a0f788 100644
--- a/server/snd_worker.c
+++ b/server/snd_worker.c
@@ -243,7 +243,8 @@ static int snd_send_data(SndChannel *channel)
 
         vec_size = spice_marshaller_fill_iovec(channel->send_data.marshaller,
                                                vec, MAX_SEND_VEC, channel->send_data.pos);
-        if ((n = channel->peer->cb_writev(channel->peer->ctx, vec, vec_size)) == -1) {
+        n = reds_stream_writev(channel->peer, vec, vec_size);
+        if (n == -1) {
             switch (errno) {
             case EAGAIN:
                 channel->blocked = TRUE;
@@ -389,7 +390,8 @@ static void snd_receive(void* data)
         ssize_t n;
         n = channel->recive_data.end - channel->recive_data.now;
         ASSERT(n);
-        if ((n = channel->peer->cb_read(channel->peer->ctx, channel->recive_data.now, n)) <= 0) {
+        n = reds_stream_read(channel->peer, channel->recive_data.now, n);
+        if (n <= 0) {
             if (n == 0) {
                 snd_disconnect_channel(channel);
                 return;