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

    22

struct memcache_conn *memcache_conn_open (struct memcache_server *server) {

    23

    struct memcache_conn *conn = NULL;

    24

    25

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

    26

        ERROR("calloc");

    27

    28

    // remember the server

    29

    conn->server = server;

    30

    31

    // attempt connect

    32

    if (memcache_conn_connect(conn))

    33

        ERROR("failed to connect to server");

    34

    35

    // success

    36

    return conn;

    37

    38

error:

    39

    free(conn);

    40

    41

    return NULL;

    42

}

    43

    44

int memcache_conn_connect (struct memcache_conn *conn) {

    45

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

    46

    47

    // begin connect

    48

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

    49

        goto error;

    50

    51

    // fd 0 should be stdin...

    52

    assert(conn->fd > 0);

    53

    54

    // set up the connect event

    55

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

    56

    57

    // add it

    58

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

    59

        PERROR("event_add");

    60

    61

    // success

    62

    return 0;

    63

    64

error:

    65

    if (conn->fd > 0) {

    66

        if (close(conn->fd))

    67

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

    68

    69

        conn->fd = -1;

    70

    }

    71

    72

    return -1;

    73

}

    74

    75

int memcache_conn_is_available (struct memcache_conn *conn) {

    76

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

    77

}

    78

    79

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

    80

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

    81

    assert(conn->req == NULL);

    82

    83

    // store the req

    84

    conn->req = req;

    85

    86

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

    87

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

    88

        ERROR("failed to init the cmd");

    89

    90

    // tell our bufferevent to send it

    91

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

    92

        PERROR("bufferevent_enable");

    93

    97

    // wait for that to complete

    98

    return 0;

    99

   100

error:

   101

    return -1;

   102

}

   103

   104

/*

   105

 * Start writing out the request data

   106

 */

   107

void memcache_conn_send_req_data (struct memcache_conn *conn) {

   108

    // just fake a call to the event handler

   109

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

   110

}

   111

   112

/*

   113

 * Start reading a reply from the connection

   114

 */

   115

void memcache_conn_handle_reply (struct memcache_conn *conn) {

   116

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

   117

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

   118

   119

    // start reading on the bufferevent

   120

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

   121

        PERROR("bufferevent_enable");

   122

   123

    // ok, wait for the reply

   124

    return;

   125

   126

error:

   127

    // XXX: error handling

   128

    assert(0);

   129

}

   130

   131

/*

   140

 * The connect() has finished

   141

 */

   142

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

   143

    struct memcache_conn *conn = arg;

   144

    int error;

   145

   146

    if (socket_check_error(fd, &error))

   147

        goto error;

   148

   149

    if (error)

   150

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

   151

   152

    // set up the bufferevent

   153

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

   154

        &_memcache_conn_bev_read,

   155

        &_memcache_conn_bev_write,

   156

        &_memcache_conn_bev_error,

   157

        conn

   158

    )) == NULL)

   159

        ERROR("bufferevent_new");

   160

   161

    // mark us as succesfully connected

   162

    conn->is_connected = 1;

   163

   164

    // notify the server

   165

    memcache_server_conn_ready(conn->server, conn);

   166

   167

    // good

   168

    return;

   169

   170

error:

   171

    // notify the server

   172

    memcache_server_conn_dead(conn->server, conn);

   173

}

   174

   175

/*

   176

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

   177

 */

   178

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

   179

    struct memcache_conn *conn = arg;

   180

   181

    // the command header has been sent

   182

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

   183

   184

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

   185

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

   186

        // we need to send the request data next

   187

        memcache_conn_send_req_data(conn);

   188

   189

    } else {

   190

        // wait for a reply

   191

        memcache_conn_handle_reply(conn);

   192

    }

   193

}

   194

   195

/*

   196

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

   197

 */

   198

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

   199

    struct memcache_conn *conn = arg;

   200

    struct evbuffer *in_buf = bufferevent_get_input(bev);

   202

    char *header_data;

   203

    enum memcache_reply reply_type;

   204

    int has_data;

   205

   206

    // ensure that we do indeed have some data

   207

    assert(evbuffer_get_length(in_buf) > 0);

   208

   209

    // attempt to parse the response header

   211

        ERROR("memcache_cmd_parse_header");

   217

   239

   240

error:

   241

    // XXX: error handling

   242

    return;

   243

}

   244

   245

   246

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

   247

    // XXX: error handling

   248

    assert(0);

   249

}

   250

   251

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

   252

    struct memcache_conn *conn = arg;

   253

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

   254

    int ret;

   255

   256

    // correct event

   257

    assert(event == EV_WRITE);

   258

   259

    // we do indeed have data to send

   260

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

   261

   262

    // do the actual write()

   263

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

   264

        PERROR("write");

   265

   266

    // should never be the case... ?

   267

    if (ret == 0)

   268

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

   269

   270

    // did we manage to write some data?

   271

    if (ret > 0) {

   272

        // update offset

   273

        buf->offset += ret;

   274

    }

   275

   276

    // data left to write?

   277

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

   278

        // reschedule

   279

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

   280

            PERROR("event_add");

   281

   282

    } else {

   283

        // done! We can handle the reply now

   284

        memcache_conn_handle_reply(conn);

   285

    }

   286

   287

    // success

   288

    return;

   289

   290

error:

   291

    // XXX: error handling

   292

    assert(0);

   293

}

   294

   310

void memcache_conn_free (struct memcache_conn *conn) {

   313

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

   314

    assert(!conn->is_connected);

   315

   316

    // close the fd if needed

   317

    if (conn->fd > 0) {

   318

        if (close(conn->fd))

   319

            PWARNING("close");

   320

   321

        conn->fd = 0;

   322

    }

   323

   324

    // ensure that the event is not pending anymore

   325

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

   326

   327

    // free it

   328

    free(conn);

   329

}

   330