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"

     9

#include "command.h"

    10

#include "request.h"

    11

#include "../socket.h"

    12

#include "../common.h"

    13

    20

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

    21

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

    22

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

    23

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

    24

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

    25

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

    26

    27

static void memcache_conn_error (struct memcache_conn *conn);

    28

static void memcache_conn_req_done (struct memcache_conn *conn);

    29

    30

void memcache_conn_close (struct memcache_conn *conn);

    31

    32

struct memcache_conn *memcache_conn_open (struct memcache_server *server) {

    33

    struct memcache_conn *conn = NULL;

    34

    35

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

    36

        ERROR("calloc");

    37

    38

    // remember the server

    39

    conn->server = server;

    40

    41

    // attempt connect

    42

    if (memcache_conn_connect(conn))

    43

        ERROR("failed to connect to server");

    44

    45

    // success

    46

    return conn;

    47

    48

error:

    49

    free(conn);

    50

    51

    return NULL;

    52

}

    53

    54

int memcache_conn_connect (struct memcache_conn *conn) {

    55

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

    56

    57

    // begin connect

    58

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

    59

        goto error;

    60

    61

    // fd 0 should be stdin...

    62

    assert(conn->fd > 0);

    63

    64

    // set up the connect event

    65

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

    66

    67

    // add it

    68

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

    69

        PERROR("event_add");

    70

    71

    // success

    72

    return 0;

    73

    74

error:

    75

    if (conn->fd > 0) {

    76

        if (close(conn->fd))

    77

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

    78

    79

        conn->fd = -1;

    80

    }

    81

    82

    return -1;

    83

}

    84

    85

int memcache_conn_is_available (struct memcache_conn *conn) {

    86

    return (conn->fd > 0 && conn->is_connected && conn->req == NULL);

    87

}

    88

    89

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

    90

    assert(conn->fd > 0 && conn->is_connected);

    91

    assert(conn->req == NULL);

    92

    93

    // write the request header into our bufferevent's output buffer

    99

        ERROR("failed to init the cmd");

   100

    }

   101

   102

    // store the req

   103

    conn->req = req;

   104

   105

    // tell our bufferevent to send it

   106

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

   107

        PERROR("bufferevent_enable");

   108

   109

    // tell the req that it is underway

   110

    memcache_req_send(req);

   111

   112

    // success

   113

    return;

   114

   115

error:

   116

    if (conn->req)

   117

        memcache_conn_error(conn);

   118

   119

    else

   120

        memcache_req_error(req);

   121

}

   122

   123

/*

   124

 * Start writing out the request data

   125

 */

   126

void memcache_conn_send_req_data (struct memcache_conn *conn) {

   130

   138

}

   139

   140

/*

   141

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

   142

 */

   143

void memcache_conn_finish_req_data (struct memcache_conn *conn) {

   144

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

   145

        PERROR("bufferevent_write");

   146

   147

    // ok

   148

    return;

   149

   150

error:

   151

    memcache_conn_error(conn);

   152

}

   153

   154

/*

   155

 * Start reading a reply from the connection

   156

 */

   157

void memcache_conn_handle_reply (struct memcache_conn *conn) {

   158

    // ensure that we either didn't have a command, or it has been sent

   159

    assert(conn->req->buf.data == NULL || conn->req->buf.offset == conn->req->buf.len);

   160

   161

    // start/continue reading on the bufferevent

   162

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

   163

        PERROR("bufferevent_enable");

   164

   165

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

   166

    // will consume all available data iteratively.

   167

   168

    // ok, wait for the reply

   169

    return;

   170

   171

error:

   172

    memcache_conn_error(conn);

   173

}

   174

   175

/*

   176

 * Start reading reply data from the connection

   177

 */

   178

void memcache_conn_handle_reply_data (struct memcache_conn *conn, struct evbuffer *buf) {

   179

    int ret;

   180

   181

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

   182

    assert(conn->req->buf.data == NULL);

   183

   184

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

   185

    if (conn->req->obj.bytes > 0) {

   188

        // allocate a buffer for the reply data

   189

        if ((conn->req->buf.data = malloc(conn->req->obj.bytes)) == NULL)

   190

            ERROR("malloc");

   191

   192

        // update the length

   193

        conn->req->buf.len = conn->req->obj.bytes;

   194

   195

        // set offset to zero

   196

        conn->req->buf.offset = 0;

   201

   202

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

   203

        if (evbuffer_get_length(buf) > 0) {

   204

            // read the data into the memcache_buf

   205

            ret = evbuffer_remove(buf, conn->req->buf.data, conn->req->buf.len);

   206

   207

            // sanity check...

   208

            assert(ret > 0 && ret <= conn->req->buf.len);

   209

   210

            // update offset

   211

            conn->req->buf.offset += ret;

   212

        }

   213

   214

        // still need to receive more data?

   215

        if (conn->req->buf.offset < conn->req->buf.len) {

   216

   217

            // disable the bufferevent while we read the data

   218

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

   219

                PERROR("bufferevent_disable");

   220

   221

            // set up the ev_read

   222

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

   223

   224

            // then receive what data is left to receive

   225

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

   226

   227

            // wait for the data to arrive

   228

            return;

   229

   230

        } else {

   231

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

   232

   233

        }

   234

    } else {

   235

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

   236

   237

    }

   238

   239

    // we kind of "recurse" to handle the MEMCACHE_RPL_END reply, that is, we activate the bufferevent for EV_READ

   240

    // again, use memcache_cmd_parse_header to parse the data (it will skip the "empty line" after the data and then

   241

    // return the MEMCACHE_RPL_END line). This will then have has_data=0, which will cause req_done to be called.

   242

    // Elegant!

   243

    memcache_conn_handle_reply(conn);

   244

   245

    // ok

   246

    return;

   247

   248

error:

   249

    memcache_conn_error(conn);

   250

}

   251

   252

/*

   253

 * The connect() has finished

   254

 */

   255

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

   256

    struct memcache_conn *conn = arg;

   257

    int error;

   258

   259

    if (socket_check_error(fd, &error))

   260

        goto error;

   261

   262

    if (error)

   263

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

   264

   265

    // set up the bufferevent

   266

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

   267

        &_memcache_conn_bev_read,

   268

        &_memcache_conn_bev_write,

   269

        &_memcache_conn_bev_error,

   270

        conn

   271

    )) == NULL)

   272

        ERROR("bufferevent_new");

   273

   274

    // mark us as succesfully connected

   275

    conn->is_connected = 1;

   276

   277

    // notify the server

   278

    memcache_server_conn_ready(conn->server, conn);

   279

   280

    // good

   281

    return;

   282

   283

error:

   284

    memcache_conn_error(conn);

   285

}

   286

   287

/*

   288

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

   289

 */

   290

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

   291

    struct memcache_conn *conn = arg;

   292

   293

    // the command header has been sent

   294

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

   295

   296

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

   297

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

   298

    if (conn->req->have_buf && conn->req->buf.offset == 0) {

   299

        // we need to send the request data next

   300

        memcache_conn_send_req_data(conn);

   301

   302

    } else {

   303

        // wait for a reply

   304

        memcache_conn_handle_reply(conn);

   305

    }

   306

}

   307

   308

/*

   309

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

   310

 */

   311

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

   312

    struct memcache_conn *conn = arg;

   313

    struct evbuffer *in_buf = bufferevent_get_input(bev);

   314

    struct memcache_key key;

   315

    char *header_data;

   316

    enum memcache_reply reply_type;

   317

    int has_data;

   318

   319

    // ensure that we do indeed have some data

   320

    assert(evbuffer_get_length(in_buf) > 0);

   321

   322

    // consume as much data as possible

   323

    do {

   324

        // attempt to parse the response header

   325

        if (memcache_cmd_parse_header(in_buf, &header_data, &reply_type, &key, &conn->req->obj, &has_data))

   326

            ERROR("memcache_cmd_parse_header");

   327

   328

        if (!header_data) {

   329

            // no complete header received yet

   330

            return;

   331

        }

   332

   333

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

   334

        if (has_data && (key.len != conn->req->key.len || memcmp(key.buf, conn->req->key.buf, key.len) != 0))

   335

            ERROR("got reply with wrong key !?!");

   336

   337

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

   338

        memcache_req_recv(conn->req, reply_type);

   339

   340

        // does the reply include data?

   341

        if (has_data) {

   342

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

   343

            memcache_conn_handle_reply_data(conn, in_buf);

   344

   345

        } else {

   346

            // the request is done with

   347

            memcache_conn_req_done(conn);

   348

        }

   349

   350

        // free the header data

   351

        free(header_data);

   352

   353

    } while (evbuffer_get_length(in_buf) > 0);

   354

   355

    // done

   356

    return;

   357

   358

error:

   359

    // free the header data read from the buf

   360

    free(header_data);

   361

   362

    memcache_conn_error(conn);

   363

}

   364

   365

   366

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

   367

    struct memcache_conn *conn = arg;

   368

   369

    // fail the entire connection

   370

    memcache_conn_error(conn);

   371

}

   372

   373

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

   374

    struct memcache_conn *conn = arg;

   375

    struct memcache_buf *buf = &conn->req->buf;

   376

    int ret;

   377

   378

    // correct event

   379

    assert(event == EV_WRITE);

   380

   381

    // we do indeed have data to send

   382

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

   383

   384

    // do the actual write()

   385

    if ((ret = write(fd, buf->data + buf->offset, buf->len - buf->offset)) == -1 && errno != EAGAIN)

   386

        PERROR("write");

   387

   388

    // should never be the case... ?

   389

    if (ret == 0)

   390

        ERROR("write returned EOF !?!");

   391

   392

    // did we manage to write some data?

   393

    if (ret > 0) {

   394

        // update offset

   395

        buf->offset += ret;

   396

    }

   397

   398

    // data left to write?

   399

    if (buf->offset < buf->len) {

   400

        // reschedule

   401

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

   402

            PERROR("event_add");

   403

   404

    } else {

   405

        // done! Send the terminating \r\n next

   406

        memcache_conn_finish_req_data(conn);

   407

    }

   408

   409

    // success

   410

    return;

   411

   412

error:

   413

    // fail the entire connection

   414

    memcache_conn_error(conn);

   415

}

   416

   417

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

   418

    struct memcache_conn *conn = arg;

   419

    struct memcache_buf *buf = &conn->req->buf;

   420

    int ret;

   421

   422

    // correct event

   423

    assert(event == EV_READ);

   424

   425

    // we do indeed expect to receive data

   426

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

   427

   428

    // do the actual read()

   429

    if ((ret = read(fd, buf->data + buf->offset, buf->len - buf->offset)) == -1 && errno != EAGAIN)

   430

        PERROR("read");

   431

   432

    // should never be the case... ?

   433

    if (ret == 0)

   434

        ERROR("read returned EOF !?!");

   435

   436

    // did we manage to read some data?

   437

    if (ret > 0) {

   438

        // update offset

   439

        buf->offset += ret;

   440

    }

   441

   442

    // only notify the req if new data was received, and we won't be calling req_done next.

   443

    if (ret > 0 && buf->offset < buf->len) {

   444

        // notify the req

   445

        memcache_req_data(conn->req);

   446

    }

   447

   448

    // data left to read?

   449

    if (buf->offset < buf->len) {

   450

        // reschedule

   451

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

   452

            PERROR("event_add");

   453

   454

   455

    } else {

   456

        // done! We can let the bufferenvet handle the rest of the reply now

   457

        memcache_conn_handle_reply(conn);

   458

    }

   459

   460

    // success

   461

    return;

   462

   463

error:

   464

    // fail the entire connection

   465

    memcache_conn_error(conn);

   466

}

   467

   468

// XXX: need to flush/disable buffers/events on errors

   469

   470

/*

   471

 * The entire connection failed

   472

 */

   473

static void memcache_conn_error (struct memcache_conn *conn) {

   474

    // fail the request, if we have one

   475

    if (conn->req) {

   476

        // error out the req

   477

        memcache_req_error(conn->req);

   478

   479

        // we are now available again

   480

        conn->req = NULL;

   481

    }

   482

   483

    // close the connection

   484

    memcache_conn_close(conn);

   485

   486

    // tell the server we failed

   487

    memcache_server_conn_dead(conn->server, conn);

   488

}

   489

   490

/*

   491

 * Detach the request

   492

 */

   493

static void memcache_conn_req_done (struct memcache_conn *conn) {

   494

    // ensure that we do currently have a req