1

#include <stdlib.h>

     2

#include <arpa/inet.h>

     3

#include <unistd.h>

     4

#include <fcntl.h>

     5

#include <errno.h>

     6

#include <assert.h>

     7

     8

#include <sys/un.h> /* XXX: for SUN_LEN */

     9

    10

#include <event2/event.h>

    11

#include <event2/event_struct.h>

    12

#include <event2/bufferevent.h>

    13

    14

#include "render_remote.h"

    15

#include "common.h"

    16

#include "render_struct.h"

    17

#include "render_net.h"

    18

#include "socket.h"

    19

    20

struct render_remote {

    21

    // the socket fd

    22

    int sock;

    23

    24

    // the event/bufferevent

    25

    struct event ev_connect, ev_data;

    26

    struct bufferevent *bev_data;

    27

    28

    // the command

    29

    struct render_cmd render_cmd;

    30

    31

    // have we sent the command yet?

    32

    int cmd_sent;

    33

    34

    // have we received the EOF?

    35

    int have_eof;

    36

    37

    // has cb_done/cb_fail/cancel already been called?

    38

    int alive;

    39

    40

    void (*cb_sent)(void *arg);

    41

    void (*cb_data)(struct evbuffer *buf, void *arg);

    42

    void (*cb_done)(void *arg);

    43

    void (*cb_fail)(void *arg);

    44

    45

    void *cb_arg;

    46

};

    47

    48

// internal prototypes

    49

static void _render_remote_free (struct render_remote *ctx);

    50

static void _render_remote_do_data (struct render_remote *ctx);

    51

static void _render_remote_do_done (struct render_remote *ctx);

    52

static void _render_remote_do_fail (struct render_remote *ctx);

    53

    54

static void _render_remote_free (struct render_remote *ctx) {

    55

    assert(ctx && !ctx->alive);

    56

    57

    // free the bev_data

    58

    if (ctx->bev_data)

    59

        bufferevent_free(ctx->bev_data);

    60

    61

    // and the events

    62

    if (event_pending(&ctx->ev_connect, EV_WRITE, NULL))

    63

        event_del(&ctx->ev_connect);

    64

    65

    if (event_pending(&ctx->ev_data, EV_READ, NULL))

    66

        event_del(&ctx->ev_data);

    67

    68

    // close the socket (ctx->ev_connect remains valid even after we're done with it...)

    69

    if (ctx->sock)

    70

        close(ctx->sock);

    71

    72

    // free the context structure

    73

    free(ctx);

    74

}

    75

    76

static void _render_remote_do_data (struct render_remote *ctx) {

    77

    // if there's data in the buffer, call cb_data

    78

    if (evbuffer_get_length(bufferevent_get_input(ctx->bev_data))) {

    79

        ctx->cb_data(EVBUFFER_INPUT(ctx->bev_data), ctx->cb_arg);

    80

    }

    81

    82

    // if we got EOF on the connection and there's no data left in the buffer, call cb_done

    83

    if (ctx->have_eof && evbuffer_get_length(bufferevent_get_input(ctx->bev_data)) == 0) {

    84

        _render_remote_do_done(ctx);

    85

    }

    86

}

    87

    88

static void _render_remote_do_done (struct render_remote *ctx) {

    89

    assert(ctx->alive);

    90

    91

    ctx->alive = 0;

    92

    93

    ctx->cb_done(ctx->cb_arg);

    94

}

    95

    96

static void _render_remote_do_fail (struct render_remote *ctx) {

    97

    assert(ctx->alive);

    98

    99

    ctx->alive = 0;

   100

   101

    ctx->cb_fail(ctx->cb_arg);

   102

}

   103

   104

static void _remote_write (struct bufferevent *bev, void *arg) {

   105

    struct render_remote *ctx = arg;

   106

   107

    if (!ctx->cmd_sent) {

   108

        // write the render command

   109

        if (bufferevent_write(ctx->bev_data, &ctx->render_cmd, sizeof(ctx->render_cmd)))

   110

            ERROR("bufferevent_write");

   111

   112

        // wait for it to be written out (we get called a second time)

   113

        ctx->cmd_sent = 1;

   114

   115

    } else {

   116

        // the write buffer was drained, so the render command was write():n

   117

        assert(ctx->cb_sent);

   118

        ctx->cb_sent(ctx->cb_arg);

   119

        ctx->cb_sent = NULL;

   120

   121

        // we don't care about EV_WRITE anymore

   122

        if (bufferevent_disable(ctx->bev_data, EV_WRITE))

   123

            ERROR("bufferevent_disable");

   124

   125

        // are we buffered or raw?

   126

        if (ctx->cb_data) {

   127

            // start receiving data into our buffer

   128

            if (bufferevent_enable(ctx->bev_data, EV_READ))

   129

                ERROR("bufferevent_enable");

   130

   131

        } else {

   132

            assert(event_initialized(&ctx->ev_data));

   133

   134

            // enable the raw read event

   135

            if (event_add(&ctx->ev_data, NULL))

   136

                ERROR("event_add");

   137

        }

   138

    }

   139

   140

    return;

   141

   142

error:

   143

    _render_remote_do_fail(ctx);

   144

}

   145

   146

static void _remote_read (struct bufferevent *bev, void *arg) {

   147

    struct render_remote *ctx = arg;

   148

   149

    _render_remote_do_data(ctx);

   150

}

   151

   152

static void _remote_error (struct bufferevent *bev, short what, void *arg) {

   153

    struct render_remote *ctx = arg;

   154

   155

    // OH NOES; WHAT DO WE DO!?

   156

   157

    if (what & EVBUFFER_EOF) {

   158

        // great!

   159

        ctx->have_eof = 1;

   160

   161

        // flush any remaining data/call cb_send as needed

   162

        _render_remote_do_data(ctx);

   163

   164

        return;

   165

   166

    } else if (what & EVBUFFER_ERROR) {

   167

        // crap.

   168

        PWARNING("EVBUFFER_ERROR");

   169

   170

    } else if (what & EVBUFFER_TIMEOUT) {

   171

        // ah well

   172

        WARNING("render_remote: timeout");

   173

   174

    } else {

   175

        FATAL("weird bufferevent error code: 0x%02X", what);

   176

    }

   177

   178

    // cb_fail + free

   179

    _render_remote_do_fail(ctx);

   180

}

   181

   182

/*

   183

 * Do the initial IO-agnostic work to initialize the rendering process

   184

 */

   185

static struct render_remote *_render_remote_init (struct render *render, struct remote_pool *pool_info) {

   186

    struct render_remote *ctx;

   187

    struct remote_node *node_info;

   188

   189

    // get a node from the pool

   190

    if (!(node_info = remote_pool_get(pool_info)))

   191

        ERROR("remote_pool_get");

   192

   193

    // alloc the remote render ctx

   194

    if (!(ctx = calloc(1, sizeof(*ctx))))

   195

        ERROR("calloc");

   196

   197

    // copy the relevant stuff from the render_ctx

   198

    ctx->render_cmd.mode = render->mode;

   199

    ctx->render_cmd.img_w = htonl(render->img_w);

   200

    ctx->render_cmd.img_h = htonl(render->img_h);

   201

    ctx->render_cmd.x1 = render->x1;

   202

    ctx->render_cmd.y1 = render->y1;

   203

    ctx->render_cmd.x2 = render->x2;

   204

    ctx->render_cmd.y2 = render->y2;

   205

   206

    // create the socket

   207

    if ((ctx->sock = socket_connect_async(&node_info->endpoint, SOCK_STREAM)) == -1)

   208

        goto error;

   209

   210

    // return the raw ctx

   211

    return ctx;

   212

   213

error:

   214

    _render_remote_free(ctx);

   215

    return NULL;

   216

}

   217

   218

/*

   219

 * Raw unbuffered I/O mode

   220

 */

   221

struct render_remote *render_remote_rawio (

   222

        struct render *render,

   223

        struct remote_pool *pool_info,

   224

        void (*cb_sent)(void *arg),

   225

        void (*cb_fail)(void *arg),

   226

        void (*cb_io_data)(evutil_socket_t, short, void*),

   227

        void *cb_arg

   228

) {

   229

    struct render_remote *ctx;

   230

   231

    // short-circuit error handling

   232

    if (!(ctx = _render_remote_init(render, pool_info)))

   233

        return NULL;

   234

   235

    // store the provided callback functions

   236

    ctx->cb_sent = cb_sent;

   237

    ctx->cb_fail = cb_fail;

   238

    ctx->cb_arg = cb_arg;

   239

   240

    // set up the custom EV_READ callback

   241

    event_set(&ctx->ev_data, ctx->sock, EV_READ, cb_io_data, cb_arg);

   242

   243

    // set up the write bufferevent

   244

    if ((ctx->bev_data = bufferevent_new(ctx->sock, NULL, &_remote_write, &_remote_error, ctx)) == NULL)

   245

        ERROR("bufferevent_new");

   246

   247

    // wait for it to connect

   248

    if (bufferevent_enable(ctx->bev_data, EV_WRITE))

   249

        ERROR("bufferevent_enable");

   250

   251

    // we are now alive

   252

    ctx->alive = 1;

   253

   254

    // success

   255

    return ctx;

   256

   257

error:

   258

    _render_remote_free(ctx);

   259

    return NULL;

   260

}   

   261

   262

/*

   263

 * Old buffered mode

   264

 */

   265

struct render_remote *render_remote (

   266

        struct render *render,

   267

        struct remote_pool *pool_info,

   268

        void (*cb_sent)(void *arg),

   269

        void (*cb_data)(struct evbuffer *buf, void *arg),

   270

        void (*cb_done)(void *arg),

   271

        void (*cb_fail)(void *arg),

   272

        void *cb_arg

   273

) {    

   274

    struct render_remote *ctx;

   275

   276

    // short-circuit error handling

   277

    if (!(ctx = _render_remote_init(render, pool_info)))

   278

        return NULL;

   279

   280

    // store the provided callback functions

   281

    ctx->cb_sent = cb_sent;

   282

    ctx->cb_data = cb_data;

   283

    ctx->cb_done = cb_done;

   284

    ctx->cb_fail = cb_fail;

   285

    ctx->cb_arg = cb_arg;

   286

   287

    // set up the read/write bufferevent

   288

    if ((ctx->bev_data = bufferevent_new(ctx->sock, &_remote_read, &_remote_write, &_remote_error, ctx)) == NULL)

   289

        ERROR("bufferevent_new");

   290

   291

    // wait for it to connect

   292

    if (bufferevent_enable(ctx->bev_data, EV_WRITE))

   293

        ERROR("bufferevent_enable");

   294

   295

    // we are now alive

   296

    ctx->alive = 1;

   297

   298

    // success

   299

    return ctx;

   300

   301

error:

   302

    _render_remote_free(ctx);

   303

    return NULL;

   304

}

   305

   306

void render_remote_set_recv (struct render_remote *ctx, size_t recv_threshold, size_t unread_buffer) {

   307

    assert(ctx->bev_data);

   308

   309

    bufferevent_setwatermark(ctx->bev_data, EV_READ, recv_threshold, recv_threshold + unread_buffer);

   310

}

   311

   312

void render_remote_flush (struct render_remote *ctx) {

   313

    assert(ctx->bev_data);

   314

   315

    // call cb_data/cb_done as appropriate

   316

    _render_remote_do_data(ctx);

   317

}

   318

   319

int render_remote_reschedule (struct render_remote *ctx) {

   320

    assert(event_initialized(&ctx->ev_data));

   321

   322

    // just reschedule it

   323

    if (event_add(&ctx->ev_data, NULL))

   324

        ERROR("event_add");

   325

   326

    // ok

   327

    return 0;

   328

   329

error:

   330

    return -1;

   331

}

   332

   333

void render_remote_cancel (struct render_remote *ctx) {

   334

    // we must be alive for this..

   335

    assert(ctx->alive);

   336

   337

    ctx->alive = 0;

   338

   339

    _render_remote_free(ctx);

   340

}

   341

   342

void render_remote_done (struct render_remote *ctx) {

   343

    // we must be alive and non-buffered for this..

   344

    assert(ctx->alive && event_initialized(&ctx->ev_data));

   345

   346

    ctx->alive = 0;

   347

   348

    _render_remote_free(ctx);

   349

}

   350

   351

void render_remote_free (struct render_remote *ctx) {

   352

    // must be dead already

   353

    assert(!ctx->alive);

   354

   355

    _render_remote_free(ctx);

   356

}

   357