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

    14

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

    15

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

    16

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

    17

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

    18

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

    19

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

    20

    21

static void memcache_conn_error (struct memcache_conn *conn);

    22

static void memcache_conn_req_error (struct memcache_conn *conn);

    23

static void memcache_conn_req_done (struct memcache_conn *conn);

    24

    25

struct memcache_conn *memcache_conn_open (struct memcache_server *server) {

    26

    struct memcache_conn *conn = NULL;

    27

    28

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

    29

        ERROR("calloc");

    30

    31

    // remember the server

    32

    conn->server = server;

    33

    34

    // attempt connect

    35

    if (memcache_conn_connect(conn))

    36

        ERROR("failed to connect to server");

    37

    38

    // success

    39

    return conn;

    40

    41

error:

    42

    free(conn);

    43

    44

    return NULL;

    45

}

    46

    47

int memcache_conn_connect (struct memcache_conn *conn) {

    48

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

    49

    50

    // begin connect

    51

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

    52

        goto error;

    53

    54

    // fd 0 should be stdin...

    55

    assert(conn->fd > 0);

    56

    57

    // set up the connect event

    58

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

    59

    60

    // add it

    61

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

    62

        PERROR("event_add");

    63

    64

    // success

    65

    return 0;

    66

    67

error:

    68

    if (conn->fd > 0) {

    69

        if (close(conn->fd))

    70

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

    71

    72

        conn->fd = -1;

    73

    }

    74

    75

    return -1;

    76

}

    77

    78

int memcache_conn_is_available (struct memcache_conn *conn) {

    79

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

    80

}

    81

    82

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

    83

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

    84

    assert(conn->req == NULL);

    85

    86

    // store the req

    87

    conn->req = req;

    88

    89

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

    90

    if (memcache_cmd_format_header(bufferevent_get_output(conn->bev), req->cmd_type, &req->key, &req->obj)) {

    91

        // just fail the request

    92

        memcache_conn_req_error(conn);

    93

    94

        ERROR("failed to init the cmd");

    95

    }

    96

    97

    // tell our bufferevent to send it

    98

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

    99

        // fail the entire connection

   100

        memcache_conn_error(conn);

   101

   102

        PERROR("bufferevent_enable");

   103

    }

   104

   105

    // tell the req that it is underway

   106

    memcache_req_send(req);

   107

   108

error:

   109

   110

    return;

   111

}

   112

   113

/*

   114

 * Start writing out the request data

   115

 */

   116

void memcache_conn_send_req_data (struct memcache_conn *conn) {

   117

    // set up the ev_write

   118

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

   119

   120

    // just fake a call to the event handler

   121

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

   122

}

   123

   124

/*

   125

 * Start reading a reply from the connection

   126

 */

   127

void memcache_conn_handle_reply (struct memcache_conn *conn) {

   128

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

   129

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

   130

   131

    // start/continue reading on the bufferevent

   132

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

   133

        PERROR("bufferevent_enable");

   134

   135

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

   136

    // will consume all available data iteratively.

   137

   138

    // ok, wait for the reply

   139

    return;

   140

   141

error:

   142

    memcache_conn_req_error(conn);

   143

}

   144

   145

/*

   146

 * Start reading reply data from the connection

   147

 */

   148

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

   149

    int ret;

   150

   151

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

   152

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

   153

   154

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

   155

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

   156

        // allocate a buffer for the reply data

   157

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

   158

            ERROR("malloc");

   159

   160

        // update the length

   161

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

   162

   163

        // set offset to zero

   164

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

   165

   166

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

   167

        if (evbuffer_get_length(buf) > 0) {

   168

            // read the data into the memcache_buf

   169

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

   170

   171

            // sanity check...

   172

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

   173

   174

            // update offset

   175

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

   176

        }

   177

   178

        // still need to receive more data?

   179

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

   180

   181

            // disable the bufferevent while we read the data

   182

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

   183

                PERROR("bufferevent_disable");

   184

   185

            // set up the ev_read

   186

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

   187

   188

            // then receive what data is left to receive

   189

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

   190

   191

            // wait for the data to arrive

   192

            return;

   193

   194

        } else {

   195

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

   196

   197

        }

   198

    } else {

   199

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

   200

   201

    }

   202

   203

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

   204

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

   205

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

   206

    // Elegant!

   207

    memcache_conn_handle_reply(conn);

   208

   209

    // ok

   210

    return;

   211

   212

error:

   213

    memcache_conn_req_error(conn);

   214

}

   215

   216

/*

   217

 * The connect() has finished

   218

 */

   219

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

   220

    struct memcache_conn *conn = arg;

   221

    int error;

   222

   223

    if (socket_check_error(fd, &error))

   224

        goto error;

   225

   226

    if (error)

   227

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

   228

   229

    // set up the bufferevent

   230

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

   231

        &_memcache_conn_bev_read,

   232

        &_memcache_conn_bev_write,

   233

        &_memcache_conn_bev_error,

   234

        conn

   235

    )) == NULL)

   236

        ERROR("bufferevent_new");

   237

   238

    // mark us as succesfully connected

   239

    conn->is_connected = 1;

   240

   241

    // notify the server

   242

    memcache_server_conn_ready(conn->server, conn);

   243

   244

    // good

   245

    return;

   246

   247

error:

   248

    memcache_conn_error(conn);

   249

}

   250

   251

/*

   252

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

   253

 */

   254

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

   255

    struct memcache_conn *conn = arg;

   256

   257

    // the command header has been sent

   258

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

   259

   260

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

   261

    if (conn->req->buf.data > 0) {

   262

        // we need to send the request data next

   263

        memcache_conn_send_req_data(conn);

   264

   265

    } else {

   266

        // wait for a reply

   267

        memcache_conn_handle_reply(conn);

   268

    }

   269

}

   270

   271

/*

   272

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

   273

 */

   274

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

   275

    struct memcache_conn *conn = arg;

   276

    struct evbuffer *in_buf = bufferevent_get_input(bev);

   277

    struct memcache_key key;

   278

    char *header_data;

   279

    enum memcache_reply reply_type;

   280

    int has_data;

   281

   282

    // ensure that we do indeed have some data

   283

    assert(evbuffer_get_length(in_buf) > 0);

   284

   285

    // consume as much data as possible

   286

    do {

   287

        // attempt to parse the response header

   288

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

   289

            ERROR("memcache_cmd_parse_header");

   290

   291

        if (!header_data) {

   292

            // no complete header received yet

   293

            return;

   294

        }

   295

   296

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

   297

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

   298

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

   299

   300

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

   302

   303

        // does the reply include data?

   304

        if (has_data) {

   305

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

   306

            memcache_conn_handle_reply_data(conn, in_buf);

   307

   308

        } else {

   309

            // the request is done with

   310

            memcache_conn_req_done(conn);

   311

        }

   312

   313

        // free the header data

   314

        free(header_data);

   315

   316

    } while (evbuffer_get_length(in_buf) > 0);

   317

   318

    // done

   319

    return;

   320

   321

error:

   322

    // free the header data read from the buf

   323

    free(header_data);

   324

   325

    memcache_conn_req_error(conn);

   326

}

   327

   328

   329

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

   330

    struct memcache_conn *conn = arg;

   331

   332

    // fail the entire connection

   333

    memcache_conn_error(conn);

   334

}

   335

   336

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

   337

    struct memcache_conn *conn = arg;

   338

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

   339

    int ret;

   340

   341

    // correct event

   342

    assert(event == EV_WRITE);

   343

   344

    // we do indeed have data to send

   345

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

   346

   347

    // do the actual write()

   348

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

   349

        PERROR("write");

   350

   351

    // should never be the case... ?

   352

    if (ret == 0)

   353

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

   354

   355

    // did we manage to write some data?

   356

    if (ret > 0) {

   357

        // update offset

   358

        buf->offset += ret;

   359

    }

   360

   361

    // data left to write?

   362

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

   363

        // reschedule

   364

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

   365

            PERROR("event_add");

   366

   367

    } else {

   368

        // done! We can handle the reply now

   369

        memcache_conn_handle_reply(conn);

   370

    }

   371

   372

    // success

   373

    return;

   374

   375

error:

   376

    // fail the entire connection

   377

    memcache_conn_error(conn);

   378

}

   379

   380

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

   381

    struct memcache_conn *conn = arg;

   382

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

   383

    int ret;

   384

   385

    // correct event

   386

    assert(event == EV_READ);

   387

   388

    // we do indeed expect to receive data

   389

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

   390

   391

    // do the actual read()

   392

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

   393

        PERROR("read");

   394

   395

    // should never be the case... ?

   396

    if (ret == 0)

   397

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

   398

   399

    // did we manage to read some data?

   400

    if (ret > 0) {

   401

        // update offset

   402

        buf->offset += ret;

   403

    }

   404

   405

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

   406

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

   407

        // notify the req

   408

        memcache_req_data(conn->req);

   409

    }

   410

   411

    // data left to read?

   412

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

   413

        // reschedule

   414

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

   415

            PERROR("event_add");

   416

   417

   418

    } else {

   419

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

   420

        memcache_conn_handle_reply(conn);

   421

    }

   422

   423

    // success

   424

    return;

   425

   426

error:

   427

    // fail the entire connection

   428

    memcache_conn_error(conn);

   429

}

   430

   431

/*

   432

 * The entire connection failed

   433

 */

   434

static void memcache_conn_error (struct memcache_conn *conn) {

   435

    // fail the request, if we have one

   436

    if (conn->req)

   437

        memcache_conn_req_error(conn);

   438

   439

    // tell the server we failed

   440

    memcache_server_conn_dead(conn->server, conn);

   441

}

   442

   443

/*

   444

 * Request failed somehow

   445

 */

   446

static void memcache_conn_req_error (struct memcache_conn *conn) {

   447

    // ensure that we do currently have a req

   448

    assert(conn->req);

   449

   450

    // error out the req

   451

    memcache_req_error(conn->req);

   452

    assert(conn->req->conn == NULL);

   453

   454

    // we are now available again

   455

    conn->req = NULL;

   456

   457

}

   458

   459

/*

   460

 * Detach the request

   461

 */

   462

static void memcache_conn_req_done (struct memcache_conn *conn) {

   463

    // ensure that we do currently have a req

   464

    assert(conn->req);

   465

   466

    // have the req detach and check it did so

   467

    memcache_req_done(conn->req);

   468

    assert(conn->req->conn == NULL);

   469

   470

    // we are now available again

   471

    conn->req = NULL;

   472

}

   473

   474

void memcache_conn_free (struct memcache_conn *conn) {

   475

    // ensure we don't have a req bound to us

   476

    assert(conn->req == NULL);

   477

   478

    // ensure that the connection is not considered to be connected anymore

   479

    assert(!conn->is_connected);

   480

   481

    // close the fd if needed

   482

    if (conn->fd > 0) {

   483

        if (close(conn->fd))

   484

            PWARNING("close");

   485

   486

        conn->fd = 0;

   487

    }

   488

   489

    // ensure that the events are not pending anymore

   490

    assert(event_pending(&conn->ev_connect, EV_WRITE|EV_TIMEOUT, NULL) == 0);

   491

    assert(event_pending(&conn->ev_read, EV_READ|EV_TIMEOUT, NULL) == 0);

   492

    assert(event_pending(&conn->ev_write, EV_WRITE|EV_TIMEOUT, NULL) == 0);

   493

   494

    // free the bufferevent

   495

    if (conn->bev)