1

     2

#include <stdlib.h>

     3

#include <unistd.h>

     4

#include <string.h>

     5

#include <errno.h>

     6

#include <assert.h>

     7

     8

#include "connection.h"

    10

#include "command.h"

    11

#include "request.h"

    12

#include "../socket.h"

    13

#include "../common.h"

    14

    23

    24

static void _memcache_conn_ev_connect (evutil_socket_t fd, short what, void *arg);

    25

static void _memcache_conn_bev_write (struct bufferevent *bev, void *arg);

    26

static void _memcache_conn_bev_read (struct bufferevent *bev, void *arg);

    27

static void _memcache_conn_bev_error (struct bufferevent *bev, short what, void *arg);

    28

static void _memcache_conn_ev_write (evutil_socket_t fd, short event, void *arg);

    29

static void _memcache_conn_ev_read (evutil_socket_t fd, short event, void *arg);

    30

    31

static void memcache_conn_error (struct memcache_conn *conn);

    33

    34

struct memcache_conn *memcache_conn_open (struct memcache_server *server) {

    35

    struct memcache_conn *conn = NULL;

    36

    37

    if ((conn = calloc(1, sizeof(*conn))) == NULL)

    38

        ERROR("calloc");

    39

    40

    // remember the server

    41

    conn->server = server;

    42

    46

    // attempt connect

    47

    if (memcache_conn_connect(conn))

    48

        ERROR("failed to connect to server");

    49

    50

    // success

    51

    return conn;

    52

    53

error:

    54

    free(conn);

    55

    56

    return NULL;

    57

}

    58

    59

int memcache_conn_connect (struct memcache_conn *conn) {

    60

    assert(conn->fd <= 0 && !conn->is_connected);

    61

    62

    // begin connect

    63

    if ((conn->fd = socket_connect_async(conn->server->endpoint, SOCK_STREAM)) == -1)

    64

        goto error;

    65

    66

    // fd 0 should be stdin...

    67

    assert(conn->fd > 0);

    68

    69

    // set up the connect event

    70

    event_set(&conn->ev_connect, conn->fd, EV_WRITE, &_memcache_conn_ev_connect, conn);

    71

    72

    // add it

    73

    if (event_add(&conn->ev_connect, NULL))

    74

        PERROR("event_add");

    75

    76

    // success

    77

    return 0;

    78

    79

error:

    80

    if (conn->fd > 0) {

    81

        if (close(conn->fd))

    82

            PWARNING("close %d", conn->fd);

    83

    84

        conn->fd = -1;

    85

    }

    86

    87

    return -1;

    88

}

    89

    90

int memcache_conn_is_available (struct memcache_conn *conn) {

    92

}

    93

    94

void memcache_conn_do_req (struct memcache_conn *conn, struct memcache_req *req) {

    98

   163

    }

   164

   166

    return;

   167

   168

error:

   170

}

   171

   172

/*

   173

 * Start writing out the request data

   174

 */

   177

        // set up the ev_write

   178

        event_set(&conn->ev_write, conn->fd, EV_WRITE, &_memcache_conn_ev_write, conn);

   179

   180

        // just fake a call to the event handler

   181

        _memcache_conn_ev_write(conn->fd, EV_WRITE, conn);

   182

   183

    } else {

   184

        // just send the \r\n

   186

    }

   187

}

   188

   189

/*

   190

 * Write out the final \r\n to terminate the request data

   191

 */

   194

    if (bufferevent_write(conn->bev, "\r\n", 2))

   195

        PERROR("bufferevent_write");

   196

   197

    // ok

   198

    return;

   199

   200

error:

   201

    memcache_conn_error(conn);

   202

}

   203

   204

/*

   219

 * Start reading a reply from the connection

   220

 */

   222

    // start/continue reading on the bufferevent

   223

    if (bufferevent_enable(conn->bev, EV_READ))

   224

        PERROR("bufferevent_enable");

   225

   226

    // Note: we don't need to recurse into the callback ourselves in case there is data in it, since the read callback

   227

    // will consume all available data iteratively.

   228

   229

    // ok, wait for the reply

   230

    return;

   231

   232

error:

   233

    memcache_conn_error(conn);

   234

}

   235

   236

/*

   237

 * Start reading reply data from the connection

   238

 */

   241

    int ret;

   242

   243

    // check that the buf doesn't contain any data

   245

   246

    // bytes *may* be zero if we have an empty cache entry

   248

        // XXX: memcache_req_make_buffer?

   249

   250

        // allocate a buffer for the reply data

   252

            ERROR("malloc");

   253

   254

        // update the length

   256

   257

        // set offset to zero

   259

   260

        // and note that it is present, and is ours

   263

   264

        // do we have any data in the buf that we need to copy?

   265

        if (evbuffer_get_length(buf) > 0) {

   266

            // read the data into the memcache_buf

   268

   269

            // sanity check...

   271

   272

            // update offset

   274

        }

   275

   276

        // still need to receive more data?

   278

   279

            // disable the bufferevent while we read the data

   280

            if (bufferevent_disable(conn->bev, EV_READ))

   281

                PERROR("bufferevent_disable");

   282

   283

            // set up the ev_read

   284

            event_set(&conn->ev_read, conn->fd, EV_READ, &_memcache_conn_ev_read, conn);

   285

   286

            // then receive what data is left to receive

   287

            _memcache_conn_ev_read(conn->fd, EV_READ, conn);

   288

   289

            // wait for the data to arrive

   290

            return;

   291

   292

        } else {

   293

            // the buffer already contained the cache data, no need to read any more

   294

   295

        }

   296

    } else {

   297

        // there is no data to receive for this item, so we can ignore this

   298

   299

    }

   300

   303

   304

    // ok

   305

    return;

   306

   307

error:

   308

    memcache_conn_error(conn);

   309

}

   310

   311

/*

   344

 * The connect() has finished

   345

 */

   346

static void _memcache_conn_ev_connect (evutil_socket_t fd, short what, void *arg) {

   347

    struct memcache_conn *conn = arg;

   348

    int error;

   349

   350

    if (socket_check_error(fd, &error))

   351

        goto error;

   352

   353

    if (error)

   354

        ERROR("connect failed: %s", strerror(error));

   355

   356

    // set up the bufferevent

   357

    if ((conn->bev = bufferevent_new(fd, 

   358

        &_memcache_conn_bev_read,

   359

        &_memcache_conn_bev_write,

   360

        &_memcache_conn_bev_error,

   361

        conn

   362

    )) == NULL)

   363

        ERROR("bufferevent_new");

   364

   365

    // mark us as succesfully connected

   366

    conn->is_connected = 1;

   370

   371

    // notify the server

   372

    memcache_server_conn_ready(conn->server, conn);

   373

   374

    // good

   375

    return;

   376

   377

error:

   378

    memcache_conn_error(conn);

   379

}

   380

   381

/*

   382

 * The write buffer is empty, which means that we have written out a command header

   383

 */

   384

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

   385

    struct memcache_conn *conn = arg;

   386

   389

    // the command header has been sent

   390

    assert(evbuffer_get_length(bufferevent_get_output(bev)) == 0);

   391

   392

    // does this request have some data to be included in the request?

   393

    // if the data has already been sent (we handle the final \r\n as well), then skip this.

   395

        // we need to send the request data next

   397

   398

    } else {

   401

    }

   402

}

   403

   404

/*

   405

 * We have received some reply data, which should include the complete reply line at some point

   406

 */

   407

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

   408

    struct memcache_conn *conn = arg;

   409

    struct evbuffer *in_buf = bufferevent_get_input(bev);

   411

    struct memcache_key key;

   412

    char *header_data;

   413

    enum memcache_reply reply_type;

   414

    int has_data;

   415

   416

    // ensure that we do indeed have some data

   417

    assert(evbuffer_get_length(in_buf) > 0);

   418

   419

    // consume as much data as possible

   420

    do {

   424

        // attempt to parse the response header

   426

            ERROR("memcache_cmd_parse_header");

   427

   428

        if (!header_data) {

   429

            // no complete header received yet

   430

            return;

   431

        }

   432

   437

        // if the reply contains data, check that they key is the same

   444

   445

        // notify the request (no reply data is ready for reading yet, though)

   447

   448

        // does the reply include data?

   449

        if (has_data) {

   450

            // start reading the data (including whatever might be left over in the bufferevent buffer...)

   452

   453

        } else {

   454

            // the request is done with

   456

        }

   457

   460

   461

    } while (evbuffer_get_length(in_buf) > 0);

   462

   463

    // done

   464

    return;

   465

   466

error:

   467

    // free the header data read from the buf

   468

    free(header_data);

   469

   470

    memcache_conn_error(conn);

   471

}

   472

   473

   474

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

   475

    struct memcache_conn *conn = arg;

   476

   477

    // fail the entire connection

   478

    memcache_conn_error(conn);

   479

}

   480

   481

static void _memcache_conn_ev_write (evutil_socket_t fd, short event, void *arg) {

   482

    struct memcache_conn *conn = arg;

   484

    int ret;

   485

   486

    // correct event

   487

    assert(event == EV_WRITE);

   488

   489

    // we do indeed have data to send

   490

    assert(buf->len > 0 && buf->data != NULL && buf->offset < buf->len);

   491

   492

    // do the actual write()