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 <event2/event.h>

     9

#include <event2/event_struct.h>

    10

#include <event2/bufferevent.h>

    11

    13

#include "render_remote.h"

    14

#include "common.h"

    15

    16

struct render_remote {

    17

    // the socket fd

    18

    int sock;

    19

    20

    // the event/bufferevent

    21

    struct event ev_connect, ev_data;

    22

    struct bufferevent *bev_data;

    23

    24

    #pragma pack(push)

    25

    #pragma pack(1)

    26

    27

    struct {

    28

        u_int8_t    mode;

    29

    30

        u_int32_t   img_w;

    31

        u_int32_t   img_h;

    32

    33

        double      x1;

    34

        double      y1;

    35

        double      x2;

    36

        double      y2;

    37

    } render_cmd;

    38

    39

    #pragma pack(pop)

    40

    41

    // have we sent the command yet?

    42

    int cmd_sent;

    43

    44

    // have we received the EOF?

    45

    int have_eof;

    46

    47

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

    48

    int alive;

    49

    50

    void (*cb_sent)(void *arg);

    51

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

    52

    void (*cb_done)(void *arg);

    53

    void (*cb_fail)(void *arg);

    54

    55

    void *cb_arg;

    56

};

    57

    58

// internal prototypes

    59

static void _render_remote_free (struct render_remote *ctx);

    60

static void _render_remote_do_data (struct render_remote *ctx);

    61

static void _render_remote_do_done (struct render_remote *ctx);

    62

static void _render_remote_do_fail (struct render_remote *ctx);

    63

    64

static void _render_remote_free (struct render_remote *ctx) {

    65

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

    66

    67

    // free the bev_data

    68

    if (ctx->bev_data)

    69

        bufferevent_free(ctx->bev_data);

    70

    71

    // and the events

    72

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

    73

        event_del(&ctx->ev_connect);

    74

    75

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

    76

        event_del(&ctx->ev_data);

    77

    78

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

    79

    if (ctx->sock)

    80

        close(ctx->sock);

    81

    82

    // free the context structure

    83

    free(ctx);

    84

}

    85

    86

static void _render_remote_do_data (struct render_remote *ctx) {

    87

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

    88

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

    89

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

    90

    }

    91

    92

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

    93

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

    94

        _render_remote_do_done(ctx);

    95

    }

    96

}

    97

    98

static void _render_remote_do_done (struct render_remote *ctx) {

    99

    assert(ctx->alive);

   100

   101

    ctx->alive = 0;

   102

   103

    ctx->cb_done(ctx->cb_arg);

   104

}

   105

   106

static void _render_remote_do_fail (struct render_remote *ctx) {

   107

    assert(ctx->alive);

   108

   109

    ctx->alive = 0;

   110

   111

    ctx->cb_fail(ctx->cb_arg);

   112

}

   113

   114

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

   115

    struct render_remote *ctx = arg;

   116

   117

    if (!ctx->cmd_sent) {

   118

        // write the render command

   119

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

   120

            ERROR("bufferevent_write");

   121

   122

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

   123

        ctx->cmd_sent = 1;

   124

   125

    } else {

   126

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

   127

        assert(ctx->cb_sent);

   128

        ctx->cb_sent(ctx->cb_arg);

   129

        ctx->cb_sent = NULL;

   130

   131

        // we don't care about EV_WRITE anymore

   132

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

   133

            ERROR("bufferevent_disable");

   134

   135

        // are we buffered or raw?

   136

        if (ctx->cb_data) {

   137

            // start receiving data into our buffer

   138

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

   139

                ERROR("bufferevent_enable");

   140

   141

        } else {

   142

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

   143

   144

            // enable the raw read event

   145

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

   146

                ERROR("event_add");

   147

        }

   148

    }

   149

   150

    return;

   151

   152

error:

   153

    _render_remote_do_fail(ctx);

   154

}

   155

   156

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

   157

    struct render_remote *ctx = arg;

   158

   159

    _render_remote_do_data(ctx);

   160

}

   161

   162

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

   163

    struct render_remote *ctx = arg;

   164

   165

    // OH NOES; WHAT DO WE DO!?

   166

   167

    if (what & EVBUFFER_EOF) {

   168

        // great!

   169

        ctx->have_eof = 1;

   170

   171

        // flush any remaining data/call cb_send as needed

   172

        _render_remote_do_data(ctx);

   173

   174

        return;

   175

   176

    } else if (what & EVBUFFER_ERROR) {

   177

        // crap.

   178

        PWARNING("EVBUFFER_ERROR");

   179

   180

    } else if (what & EVBUFFER_TIMEOUT) {

   181

        // ah well

   182

        WARNING("render_remote: timeout");

   183

   184

    } else {

   185

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

   186

    }

   187

   188

    // cb_fail + free

   189

    _render_remote_do_fail(ctx);

   190

}

   191

   192

/*

   193

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

   194

 */

   195

static struct render_remote *_render_remote_init (struct render *render, struct remote_node *remote_node) {

   196

    struct render_remote *ctx;

   197

   198

    printf("remote_node render load: %d/%d\n", remote_node->current_load, remote_node->parallel_renders);

   199

   200

    // alloc the remote render ctx

   201

    if (!(ctx = calloc(1, sizeof(struct render_remote))))

   202

        ERROR("calloc");

   203

   204

    // copy the relevant stuff from the render_ctx

   205

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

   206

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

   207

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

   208

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

   209

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

   210

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

   211

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

   212

   213

    // create the socket

   214

    if ((ctx->sock = socket(remote_node->addr.ss_family, SOCK_STREAM, 0)) < 0)

   215

        PERROR("socket");

   216

   217

    // mark it as nonblocking

   218

    if (fcntl(ctx->sock, F_SETFL, O_NONBLOCK) == -1)

   219

        PERROR("fcntl");

   220

   221

    // initiate the connect

   222

    int err = connect(ctx->sock, (struct sockaddr *) &remote_node->addr, sizeof(remote_node->addr));

   223

   224

    if (err != -1 || errno != EINPROGRESS)

   225

        PERROR("connect");

   226

   227

    // return the raw ctx

   228

    return ctx;

   229

   230

error:

   231

    _render_remote_free(ctx);

   232

    return NULL;

   233

}

   234

   235

/*

   236

 * Raw unbuffered I/O mode

   237

 */

   238

struct render_remote *render_remote_rawio (

   239

        struct render *render,

   240

        struct remote_node *remote_node,

   241

        void (*cb_sent)(void *arg),

   242

        void (*cb_fail)(void *arg),

   243

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

   244

        void *cb_arg

   245

) {

   246

    struct render_remote *ctx;

   247

   248

    // short-circuit error handling

   249

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

   250

        return NULL;

   251

   252

    // store the provided callback functions

   253

    ctx->cb_sent = cb_sent;

   254

    ctx->cb_fail = cb_fail;

   255

    ctx->cb_arg = cb_arg;

   256

   257

    // set up the custom EV_READ callback

   258

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

   259

   260

    // set up the write bufferevent

   261

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

   262

        ERROR("bufferevent_new");

   263

   264

    // wait for it to connect

   265

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

   266

        ERROR("bufferevent_enable");

   267

   268

    // we are now alive

   269

    ctx->alive = 1;

   270

   271

    // success

   272

    return ctx;

   273

   274

error:

   275

    _render_remote_free(ctx);

   276

    return NULL;

   277

}   

   278

   279

/*

   280

 * Old buffered mode

   281

 */

   282

struct render_remote *render_remote (

   283

        struct render *render,

   284

        struct remote_node *remote_node,

   285

        void (*cb_sent)(void *arg),

   286

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

   287

        void (*cb_done)(void *arg),

   288

        void (*cb_fail)(void *arg),

   289

        void *cb_arg

   290

) {    

   291

    struct render_remote *ctx;

   292

   293

    // short-circuit error handling

   294

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

   295

        return NULL;

   296

   297

    // store the provided callback functions

   298

    ctx->cb_sent = cb_sent;

   299

    ctx->cb_data = cb_data;

   300

    ctx->cb_done = cb_done;

   301

    ctx->cb_fail = cb_fail;

   302

    ctx->cb_arg = cb_arg;

   303

   304

    // set up the read/write bufferevent

   305

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

   306

        ERROR("bufferevent_new");

   307

   308

    // wait for it to connect

   309

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

   310

        ERROR("bufferevent_enable");

   311

   312

    // we are now alive

   313

    ctx->alive = 1;

   314

   315

    // success

   316

    return ctx;

   317

   318

error:

   319

    _render_remote_free(ctx);

   320

    return NULL;

   321

}

   322

   323

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

   324

    assert(ctx->bev_data);

   325

   326

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

   327

}

   328

   329

void render_remote_flush (struct render_remote *ctx) {

   330

    assert(ctx->bev_data);

   331

   332

    // call cb_data/cb_done as appropriate

   333

    _render_remote_do_data(ctx);

   334

}

   335

   336

int render_remote_reschedule (struct render_remote *ctx) {

   337

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

   338

   339

    // just reschedule it

   340

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

   341

        ERROR("event_add");

   342

   343

    // ok

   344

    return 0;

   345

   346

error:

   347

    return -1;

   348

}

   349

   350

void render_remote_cancel (struct render_remote *ctx) {

   351

    // we must be alive for this..

   352

    assert(ctx->alive);

   353

   354

    ctx->alive = 0;

   355

   356

    _render_remote_free(ctx);

   357

}

   358

   359

void render_remote_done (struct render_remote *ctx) {

   360

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

   361

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

   362

   363

    ctx->alive = 0;

   364

   365

    _render_remote_free(ctx);

   366

}

   367

   368

void render_remote_free (struct render_remote *ctx) {

   369

    // must be dead already

   370

    assert(!ctx->alive);

   371

   372

    _render_remote_free(ctx);

   373

}

   374