terom@2: #include <stdlib.h>
terom@2: #include <arpa/inet.h>
terom@2: #include <unistd.h>
terom@2: #include <fcntl.h>
terom@2: #include <errno.h>
terom@2: 
terom@4: #include <event2/event.h>
terom@4: #include <event2/bufferevent.h>
terom@4: 
terom@2: #include "render_remote.h"
terom@2: #include "common.h"
terom@2: 
terom@2: struct remote_render_ctx {
terom@4:     struct event *ev_conn;
terom@2:     struct bufferevent *data_bev;
terom@2: 
terom@8:     struct remote_node *remote_node;
terom@8: 
terom@2:     #pragma pack(push)
terom@2:     #pragma pack(1)
terom@2: 
terom@2:     struct {
terom@2:         u_int8_t    mode;
terom@2: 
terom@2:         u_int32_t   img_w;
terom@2:         u_int32_t   img_h;
terom@2: 
terom@2:         double      x1;
terom@2:         double      y1;
terom@2:         double      x2;
terom@2:         double      y2;
terom@2:     } render_cmd;
terom@2: 
terom@2:     #pragma pack(pop)
terom@2: 
terom@2:     void (*cb_sent)(void *arg);
terom@2:     void (*cb_data)(struct evbuffer *buf, void *arg);
terom@2:     void (*cb_done)(void *arg);
terom@2:     void (*cb_fail)(void *arg);
terom@2: 
terom@2:     void *cb_arg;
terom@2: };
terom@2: 
terom@8: void _remote_render_ctx_free (struct remote_render_ctx *ctx) {
terom@8:     // close the socket (ctx->ev_conn remains valid even after we're done with it...)
terom@8:     close(event_get_fd(ctx->ev_conn));
terom@8:     
terom@8:     // free the connect event
terom@8:     event_free(ctx->ev_conn);
terom@8: 
terom@2:     // free the data_bev
terom@8:     if (ctx->data_bev) {
terom@8:         bufferevent_free(ctx->data_bev);
terom@8:         ctx->data_bev = NULL;
terom@2:     }
terom@4: 
terom@8:     // update remote_node load
terom@8:     ctx->remote_node->current_load--;
terom@2:     
terom@2:     // free the context structure
terom@8:     free(ctx);
terom@2:     
terom@8:     ctx = NULL;
terom@2: }
terom@2: 
terom@2: #define RENDER_FAILED(ctx, desc) \
terom@2:     do {                                        \
terom@2:         perror(desc);                           \
terom@2:         ctx->cb_fail(ctx->cb_arg);              \
terom@8:         _remote_render_ctx_free(ctx);          \
terom@3:         return;                                 \
terom@2:     } while (0)
terom@2: 
terom@2: void _remote_write (struct bufferevent *bev, void *arg) {
terom@2:     struct remote_render_ctx *ctx = arg;
terom@2: 
terom@2:     // the write buffer was drained, so the render command was sent
terom@2:     ctx->cb_sent(ctx->cb_arg);
terom@2:     
terom@2:     // we don't care about EV_WRITE anymore
terom@2:     if (bufferevent_disable(ctx->data_bev, EV_WRITE))
terom@2:         RENDER_FAILED(ctx, "render_remote: bufferevent_disable");
terom@2: 
terom@2:     // start receiving data
terom@2:     if (bufferevent_enable(ctx->data_bev, EV_READ))
terom@2:         RENDER_FAILED(ctx, "render_remote: bufferevent_enable");
terom@2: }
terom@2: 
terom@2: void _remote_read (struct bufferevent *bev, void *arg) {
terom@2:     struct remote_render_ctx *ctx = arg;
terom@2:     
terom@2:     // pass the bufferevent's input buffer to our callback - libevent doesn't provide any function to access this, but hopefully this works correctly
terom@3:     ctx->cb_data(EVBUFFER_INPUT(bev), ctx->cb_arg);
terom@2: }
terom@2: 
terom@2: void _remote_error (struct bufferevent *bev, short what, void *arg) {
terom@2:     struct remote_render_ctx *ctx = arg;
terom@2: 
terom@2:     // OH NOES; WHAT DO WE DO!?
terom@2:     
terom@2:     if (what & EVBUFFER_EOF) {
terom@2:         // great!
terom@3:         
terom@3:         // send any remaining-chunk data
terom@3:         if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) > 0)
terom@3:             ctx->cb_data(EVBUFFER_INPUT(bev), ctx->cb_arg);
terom@3: 
terom@3:         // signal completion
terom@2:         ctx->cb_done(ctx->cb_arg);
terom@2: 
terom@2:     } else if (what & EVBUFFER_ERROR) {
terom@2:         // crap.
terom@2:         perr("render_remote");
terom@2: 
terom@2:         ctx->cb_fail(ctx->cb_arg);
terom@2: 
terom@2:     } else if (what & EVBUFFER_TIMEOUT) {
terom@2:         // ah well
terom@2:         error("render_remote: timeout");
terom@2: 
terom@2:         ctx->cb_fail(ctx->cb_arg);
terom@2: 
terom@2:     } else {
terom@2:         err_exit("weird bufferevent error code: 0x%02X", what);
terom@2:     }
terom@2: 
terom@2:     // free resources
terom@8:     _remote_render_ctx_free(ctx);
terom@2: }
terom@2: 
terom@2: void _remote_connected (int fd, short event, void *arg) {
terom@2:     struct remote_render_ctx *ctx = arg;
terom@2: 
terom@2:     // set up the read/write bufferevent
terom@2:     if ((ctx->data_bev = bufferevent_new(fd, &_remote_read, &_remote_write, &_remote_error, ctx)) == NULL)
terom@2:         RENDER_FAILED(ctx, "render_remote: bufferevent_new");
terom@2: 
terom@2:     // write the render command
terom@2:     if (bufferevent_write(ctx->data_bev, &ctx->render_cmd, sizeof(ctx->render_cmd)))
terom@2:         RENDER_FAILED(ctx, "render_remote: bufferevent_write");
terom@2: 
terom@2:     // wait for it to be written out
terom@2:     if (bufferevent_enable(ctx->data_bev, EV_WRITE))
terom@2:         RENDER_FAILED(ctx, "render_remote: bufferevent_enable");
terom@2: }
terom@2: 
terom@2: void render_cmd_build (render_t *rctx, struct remote_render_ctx *rrctx) {
terom@2:     // just copy over the render params to the render_cmd
terom@2:     rrctx->render_cmd.mode = rctx->mode;
terom@2:     rrctx->render_cmd.img_w = htonl(rctx->img_w);
terom@2:     rrctx->render_cmd.img_h = htonl(rctx->img_h);
terom@2:     rrctx->render_cmd.x1 = rctx->x1;
terom@2:     rrctx->render_cmd.y1 = rctx->y1;
terom@2:     rrctx->render_cmd.x2 = rctx->x2;
terom@2:     rrctx->render_cmd.y2 = rctx->y2;
terom@2: }
terom@2: 
terom@3: struct remote_render_ctx *render_remote (
terom@2:         render_t *render_ctx,
terom@8:         struct remote_node *remote_node,
terom@2:         void (*cb_sent)(void *arg),
terom@2:         void (*cb_data)(struct evbuffer *buf, void *arg),
terom@2:         void (*cb_done)(void *arg),
terom@2:         void (*cb_fail)(void *arg),
terom@2:         void *cb_arg
terom@2: ) {    
terom@8:     printf("render_remote: remote render load: %d/%d\n", remote_node->current_load, remote_node->parallel_renders);
terom@8: 
terom@2:     // alloc the remote render ctx
terom@2:     struct remote_render_ctx *ctx = malloc(sizeof(struct remote_render_ctx));
terom@2: 
terom@2:     if (!ctx) {
terom@2:         error("render_remote: malloc");
terom@3:         return NULL;
terom@2:     }
terom@2:     
terom@2:     // store the provided callback functions
terom@2:     ctx->cb_sent = cb_sent;
terom@2:     ctx->cb_data = cb_data;
terom@2:     ctx->cb_done = cb_done;
terom@2:     ctx->cb_fail = cb_fail;
terom@2:     ctx->cb_arg = cb_arg;
terom@8: 
terom@8:     // keep a reference to remote_node so we can decr it's load
terom@8:     ctx->remote_node = remote_node;
terom@2:     
terom@2:     // copy the relevant stuff from the render_ctx
terom@2:     render_cmd_build(render_ctx, ctx);
terom@2:     
terom@2:     // create the socket
terom@8:     int sock = socket(remote_node->addr.ss_family, SOCK_STREAM, 0);
terom@2: 
terom@2:     if (sock < 0) {
terom@2:         perror("render_remote: socket");
terom@4:         goto error;
terom@2:     }
terom@2: 
terom@2:     // mark it as nonblocking
terom@2:     if (fcntl(sock, F_SETFL, O_NONBLOCK) == -1) {
terom@2:         perror("render_remote: fcntl");
terom@4:         goto error;
terom@2:     }
terom@2:     
terom@2:     // initiate the connect
terom@8:     int err = connect(sock, (struct sockaddr *) &remote_node->addr, sizeof(remote_node->addr));
terom@2: 
terom@2:     if (err != -1 || errno != EINPROGRESS) {
terom@2:         perror("render_remote: connect");
terom@4:         goto error;
terom@2:     }
terom@2: 
terom@2:     // do the libevent dance
terom@4:     ctx->ev_conn = event_new(NULL, sock, EV_WRITE, &_remote_connected, ctx);
terom@2: 
terom@4:     if (!ctx->ev_conn) {
terom@4:         error("render_remote: event_new");
terom@4:         goto error;
terom@4:     }
terom@4: 
terom@4:     if (event_add(ctx->ev_conn, NULL)) {
terom@2:         error("render_remote: event_add");
terom@4:         goto error;
terom@2:     }
terom@8: 
terom@8:     // update remote_node load
terom@8:     remote_node->current_load++;
terom@2:     
terom@2:     // success
terom@3:     return ctx;
terom@4: 
terom@4: error:
terom@4:     free(ctx);
terom@4: 
terom@4:     if (sock > 0)
terom@4:         close(sock);
terom@4: 
terom@4:     return NULL;
terom@3: }
terom@3: 
terom@6: int render_remote_set_recv (struct remote_render_ctx *ctx, size_t recv_threshold, size_t unread_buffer) {
terom@3:     if (ctx->data_bev == NULL)
terom@3:         return -1;
terom@3: 
terom@6:     bufferevent_setwatermark(ctx->data_bev, EV_READ, recv_threshold, recv_threshold + unread_buffer);
terom@6: 
terom@6:     return 0;
terom@6: }
terom@6: 
terom@6: int render_remote_shake (struct remote_render_ctx *ctx) {
terom@6:     if (ctx->data_bev == NULL)
terom@6:         return -1;
terom@6: 
terom@6:     ctx->cb_data(EVBUFFER_INPUT(ctx->data_bev), ctx->cb_arg);
terom@3: 
terom@2:     return 0;
terom@2: }
terom@3: 
terom@3: void render_remote_cancel (struct remote_render_ctx *ctx) {
terom@3:     // if it's still just connecting, cancel that
terom@4:     if (event_pending(ctx->ev_conn, EV_WRITE, NULL)) {
terom@4:         event_del(ctx->ev_conn);
terom@3:     }
terom@3:     
terom@3:     // this takes care of the rest
terom@8:     _remote_render_ctx_free (ctx);
terom@3: }
terom@3: