1

#define _GNU_SOURCE

     2

#include <stdlib.h>

     3

#include <assert.h>

     4

#include <string.h>

     5

     6

#include "evsql.h"

     7

#include "../lib/log.h"

     8

#include "../lib/error.h"

     9

#include "../lib/misc.h"

    10

    11

/*

    12

 * A couple function prototypes

    13

 */ 

    14

static void _evsql_pump (struct evsql *evsql, struct evsql_conn *conn);

    15

    16

/*

    17

 * Actually execute the given query.

    18

 *

    19

 * The backend should be able to accept the query at this time.

    20

 *

    21

 * You should assume that if trying to execute a query fails, then the connection should also be considred as failed.

    22

 */

    23

static int _evsql_query_exec (struct evsql_conn *conn, struct evsql_query *query, const char *command) {

    24

    int err;

    25

    26

    DEBUG("evsql.%p: exec query=%p on trans=%p on conn=%p:", conn->evsql, query, conn->trans, conn);

    27

    28

    switch (conn->evsql->type) {

    29

        case EVSQL_EVPQ:

    30

            // got params?

    31

            if (query->params.count) {

    32

                err = evpq_query_params(conn->engine.evpq, command,

    33

                    query->params.count, 

    34

                    query->params.types, 

    35

                    query->params.values, 

    36

                    query->params.lengths, 

    37

                    query->params.formats, 

    38

                    query->params.result_format

    39

                );

    40

    41

            } else {

    42

                // plain 'ole query

    43

                err = evpq_query(conn->engine.evpq, command);

    44

            }

    45

    46

            if (err) {

    47

                if (PQstatus(evpq_pgconn(conn->engine.evpq)) != CONNECTION_OK)

    48

                    WARNING("conn failed");

    49

                else

    50

                    WARNING("query failed, dropping conn as well");

    51

            }

    52

    53

            break;

    54

    55

        default:

    56

            FATAL("evsql->type");

    57

    }

    58

    59

    if (!err)

    60

        // assign the query

    61

        conn->query = query;

    62

    63

    return err;

    64

}

    65

    66

void _evsql_query_free (struct evsql_query *query) {

    67

    if (!query)

    68

        return;

    69

    70

    assert(query->command == NULL);

    71

    72

    // free params if present

    73

    free(query->params.types);

    74

    free(query->params.values);

    75

    free(query->params.lengths);

    76

    free(query->params.formats);

    77

    78

    // free the query itself

    79

    free(query);

    80

}

    81

    82

/*

    83

 * Execute the callback if res is given, and free the query.

    84

 *

    85

 * The query has been aborted, it will simply be freed

    86

 */

    87

static void _evsql_query_done (struct evsql_query *query, struct evsql_result *res) {

    88

    if (res) {

    89

        if (query->cb_fn)

    90

            // call the callback

    91

            query->cb_fn(res, query->cb_arg);

    92

        else {

    93

            WARNING("supressing cb_fn because query was aborted");

    94

    95

            // free the results

    96

            evsql_result_free(res);

    97

        }

    98

    }

    99

   100

    // free

   101

    _evsql_query_free(query);

   102

}

   103

   104

/*

   105

 * XXX:

   106

 * /

   107

static void _evsql_destroy (struct evsql *evsql, const struct evsql_result *res) {

   108

    struct evsql_query *query;

   109

   110

    // clear the queue

   111

    while ((query = TAILQ_FIRST(&evsql->query_queue)) != NULL) {

   112

        _evsql_query_done(query, res);

   113

   114

        TAILQ_REMOVE(&evsql->query_queue, query, entry);

   115

    }

   116

   117

    // free

   118

    free(evsql);

   119

}

   120

*/

   121

   122

/*

   123

 * Free the transaction, it should already be deassociated from the query and conn.

   124

 */

   125

static void _evsql_trans_free (struct evsql_trans *trans) {

   126

    // ensure we don't leak anything

   127

    assert(trans->query == NULL);

   128

    assert(trans->conn == NULL);

   129

   130

    // free

   131

    free(trans);

   132

}

   133

   134

/*

   135

 * Release a connection. It should already be deassociated from the trans and query.

   136

 *

   137

 * Releases the engine, removes from the conn_list and frees this.

   138

 */

   139

static void _evsql_conn_release (struct evsql_conn *conn) {

   140

    // ensure we don't leak anything

   141

    assert(conn->trans == NULL);

   142

    assert(conn->query == NULL);

   143

   144

    // release the engine

   145

    switch (conn->evsql->type) {

   146

        case EVSQL_EVPQ:

   147

            evpq_release(conn->engine.evpq);

   148

            break;

   149

   150

        default:

   151

            FATAL("evsql->type");

   152

    }

   153

   154

    // remove from list

   155

    LIST_REMOVE(conn, entry);

   156

   157

    // catch deadlocks

   158

    assert(!LIST_EMPTY(&conn->evsql->conn_list) || TAILQ_EMPTY(&conn->evsql->query_queue));

   159

   160

    // free

   161

    free(conn);

   162

}

   163

   164

/*

   165

 * Release a transaction, it should already be deassociated from the query.

   166

 *

   167

 * Perform a two-way-deassociation with the conn, and then free the trans.

   168

 */

   169

static void _evsql_trans_release (struct evsql_trans *trans) {

   170

    assert(trans->query == NULL);

   171

    assert(trans->conn != NULL);

   172

   173

    // deassociate the conn

   174

    trans->conn->trans = NULL; trans->conn = NULL;

   175

   176

    // free the trans

   177

    _evsql_trans_free(trans);

   178

}

   179

   180

/*

   181

 * Fail a single query, this will trigger the callback and free it.

   182

 *

   183

 * NOTE: Only for *TRANSACTIONLESS* queries.

   184

 */

   185

static void _evsql_query_fail (struct evsql* evsql, struct evsql_query *query) {

   186

    struct evsql_result res; ZINIT(res);

   187

   188

    // set up the result_info

   189

    res.evsql = evsql;

   190

    res.error = 1;

   191

   192

    // finish off the query

   193

    _evsql_query_done(query, &res);

   194

}

   195

   196

/*

   197

 * Fail a transaction, this will silently drop any query, trigger the error callback, two-way-deassociate/release the

   198

 * conn, and then free the trans.

   199

 */ 

   200

static void _evsql_trans_fail (struct evsql_trans *trans) {

   201

    if (trans->query) {

   202

        // free the query silently

   203

        _evsql_query_free(trans->query); trans->query = NULL;

   204

   205

        // also deassociate it from the conn!

   206

        trans->conn->query = NULL;

   207

    }

   208

   209

    // tell the user

   210

    // XXX: trans is in a bad state during this call

   211

    if (trans->error_fn)

   212

        trans->error_fn(trans, trans->cb_arg);

   213

    else

   214

        WARNING("supressing error because error_fn was NULL");

   215

   216

    // deassociate and release the conn

   217

    trans->conn->trans = NULL; _evsql_conn_release(trans->conn); trans->conn = NULL;

   218

   219

    // pump the queue for requests that were waiting for this connection

   220

    _evsql_pump(trans->evsql, NULL);

   221

   222

    // free the trans

   223

    _evsql_trans_free(trans);

   224

}

   225

   226

/*

   227

 * Fail a connection. If the connection is transactional, this will just call _evsql_trans_fail, but otherwise it will

   228

 * fail any ongoing query, and then release the connection.

   229

 */

   230

static void _evsql_conn_fail (struct evsql_conn *conn) {

   231

    if (conn->trans) {

   232

        // let transactions handle their connection failures

   233

        _evsql_trans_fail(conn->trans);

   234

   235

    } else {

   236

        if (conn->query) {

   237

            // fail the in-progress query

   238

            _evsql_query_fail(conn->evsql, conn->query); conn->query = NULL;

   239

        }

   240

   241

        // finish off the whole connection

   242

        _evsql_conn_release(conn);

   243

    }

   244

}

   245

   246

/*

   247

 * Processes enqueued non-transactional queries until the queue is empty, or we managed to exec a query.

   248

 *

   249

 * If execing a query on a connection fails, both the query and the connection are failed (in that order).

   250

 *

   251

 * Any further queries will then also be failed, because there's no reconnection/retry logic yet.

   252

 *

   253

 * This means that if conn is NULL, all queries are failed.

   254

 */

   255

static void _evsql_pump (struct evsql *evsql, struct evsql_conn *conn) {

   256

    struct evsql_query *query;

   257

    int err;

   258

   259

    // look for waiting queries

   260

    while ((query = TAILQ_FIRST(&evsql->query_queue)) != NULL) {

   261

        // dequeue

   262

        TAILQ_REMOVE(&evsql->query_queue, query, entry);

   263

   264

        if (conn) {

   265

            // try and execute it

   266

            err = _evsql_query_exec(conn, query, query->command);

   267

        }

   268

   269

        // free the command buf

   270

        free(query->command); query->command = NULL;

   271

   272

        if (err || !conn) {

   273

            if (!conn) {

   274

                // warn when dropping queries

   275

                WARNING("failing query becuse there are no conns");

   276

            }

   277

   278

            // fail the query

   279

            _evsql_query_fail(evsql, query);

   280

   281

            if (conn) {

   282

                // fail the connection

   283

                WARNING("failing the connection because a query-exec failed");

   284

   285

                _evsql_conn_fail(conn); conn = NULL;

   286

            }

   287

   288

        } else {

   289

            // we have succesfully enqueued a query, and we can wait for this connection to complete

   290

            break;

   291

   292

        }

   293

   294

        // handle the rest of the queue

   295

    }

   296

   297

    // ok

   298

    return;

   299

}

   300

   301

/*

   302

 * Callback for a trans's 'BEGIN' query, which means the transaction is now ready for use.

   303

 */

   304

static void _evsql_trans_ready (struct evsql_result *res, void *arg) {

   305

    struct evsql_trans *trans = arg;

   306

   307

    assert(trans != NULL);

   308

   309

    // check for errors

   310

    if (res->error)

   311

        ERROR("transaction 'BEGIN' failed: %s", evsql_result_error(res));

   312

   313

    // transaction is now ready for use

   314

    trans->ready_fn(trans, trans->cb_arg);

   315

   316

    // good

   317

    return;

   318

   319

error:

   320

    _evsql_trans_fail(trans);

   321

}

   322

   323

/*

   324

 * The transaction's connection is ready, send the 'BEGIN' query.

   325

 *

   326

 * If anything fails, calls _evsql_trans_fail and returns nonzero, zero on success

   327

 */

   328

static int _evsql_trans_conn_ready (struct evsql *evsql, struct evsql_trans *trans) {

   329

    char trans_sql[EVSQL_QUERY_BEGIN_BUF];

   330

    const char *isolation_level;

   331

    int ret;

   332

   333

    // determine the isolation_level to use

   334

    switch (trans->type) {

   335

        case EVSQL_TRANS_DEFAULT:

   336

            isolation_level = NULL; break;

   337

   338

        case EVSQL_TRANS_SERIALIZABLE:

   339

            isolation_level = "SERIALIZABLE"; break;

   340

   341

        case EVSQL_TRANS_REPEATABLE_READ:

   342

            isolation_level = "REPEATABLE READ"; break;

   343

   344

        case EVSQL_TRANS_READ_COMMITTED:

   345

            isolation_level = "READ COMMITTED"; break;

   346

   347

        case EVSQL_TRANS_READ_UNCOMMITTED:

   348

            isolation_level = "READ UNCOMMITTED"; break;

   349

   350

        default:

   351

            FATAL("trans->type: %d", trans->type);

   352

    }

   353

   354

    // build the trans_sql

   355

    if (isolation_level)

   356

        ret = snprintf(trans_sql, EVSQL_QUERY_BEGIN_BUF, "BEGIN TRANSACTION ISOLATION LEVEL %s", isolation_level);

   357

    else

   358

        ret = snprintf(trans_sql, EVSQL_QUERY_BEGIN_BUF, "BEGIN TRANSACTION");

   359

   360

    // make sure it wasn't truncated

   361

    if (ret >= EVSQL_QUERY_BEGIN_BUF)

   362

        ERROR("trans_sql overflow: %d >= %d", ret, EVSQL_QUERY_BEGIN_BUF);

   363

   364

    // execute the query

   365

    if (evsql_query(evsql, trans, trans_sql, _evsql_trans_ready, trans) == NULL)

   366

        ERROR("evsql_query");

   367

   368

    // success

   369

    return 0;

   370

   371

error:

   372

    // fail the transaction

   373

    _evsql_trans_fail(trans);

   374

   375

    return -1;

   376

}

   377

   378

/*

   379

 * The evpq connection was succesfully established.

   380

 */ 

   381

static void _evsql_evpq_connected (struct evpq_conn *_conn, void *arg) {

   382

    struct evsql_conn *conn = arg;

   383

   384

    if (conn->trans)

   385

        // notify the transaction

   386

        // don't care about errors

   387

        (void) _evsql_trans_conn_ready(conn->evsql, conn->trans);

   388

   389

    else

   390

        // pump any waiting transactionless queries

   391

        _evsql_pump(conn->evsql, conn);

   392

}

   393

   394

/*

   395

 * Got one result on this evpq connection.

   396

 */

   397

static void _evsql_evpq_result (struct evpq_conn *_conn, PGresult *result, void *arg) {

   398

    struct evsql_conn *conn = arg;

   399

    struct evsql_query *query = conn->query;

   400

   401

    assert(query != NULL);

   402

   403

    // if we get multiple results, only return the first one

   404

    if (query->result.pq) {

   405

        WARNING("[evsql] evpq query returned multiple results, discarding previous one");

   406

   407

        PQclear(query->result.pq); query->result.pq = NULL;

   408

    }

   409

   410

    // remember the result

   411

    query->result.pq = result;

   412

}

   413

   414

/*

   415

 * No more results for this query.

   416

 */

   417

static void _evsql_evpq_done (struct evpq_conn *_conn, void *arg) {

   418

    struct evsql_conn *conn = arg;

   419

    struct evsql_query *query = conn->query;

   420

    struct evsql_result res; ZINIT(res);

   421

   422

    assert(query != NULL);

   423

   424

    // set up the result_info

   425

    res.evsql = conn->evsql;

   426

    res.result = query->result;

   427

   428

    if (query->result.pq == NULL) {

   429

        // if a query didn't return any results (bug?), warn and fail the query

   430

        WARNING("[evsql] evpq query didn't return any results");

   431

   432

        res.error = 1;

   433

   434

    } else if (strcmp(PQresultErrorMessage(query->result.pq), "") != 0) {

   435

        // the query failed with some error

   436

        res.error = 1;

   437

   438

    } else {

   439

        // the query succeeded \o/

   440

        res.error = 0;

   441

   442

    }

   443

   444

    // de-associate the query from the connection

   445

    conn->query = NULL;

   446

   447

    // how we handle query completion depends on if we're a transaction or not

   448

    if (conn->trans) {

   449

        // we can deassign the trans's query

   450

        conn->trans->query = NULL;

   451

   452

        // was an abort?

   453

        if (!query->cb_fn)

   454

            // notify the user that the transaction query has been aborted

   455

            conn->trans->ready_fn(conn->trans, conn->trans->cb_arg);

   456

   457

        // then hand the query to the user

   458

        _evsql_query_done(query, &res);

   459

   460

    } else {

   461

        // a transactionless query, so just finish it off and pump any other waiting ones

   462

        _evsql_query_done(query, &res);

   463

   464

        // pump the next one

   465

        _evsql_pump(conn->evsql, conn);

   466

    }

   467

}

   468

   469

/*

   470

 * The connection failed.

   471

 */

   472

static void _evsql_evpq_failure (struct evpq_conn *_conn, void *arg) {

   473

    struct evsql_conn *conn = arg;

   474

   475

    // just fail the conn

   476

    _evsql_conn_fail(conn);

   477

}

   478

   479

/*

   480

 * Our evpq behaviour

   481

 */

   482

static struct evpq_callback_info _evsql_evpq_cb_info = {

   483

    .fn_connected       = _evsql_evpq_connected,

   484

    .fn_result          = _evsql_evpq_result,

   485

    .fn_done            = _evsql_evpq_done,

   486

    .fn_failure         = _evsql_evpq_failure,

   487

};

   488

   489

/*

   490

 * Allocate the generic evsql context.

   491

 */

   492

static struct evsql *_evsql_new_base (struct event_base *ev_base, evsql_error_cb error_fn, void *cb_arg) {

   493

    struct evsql *evsql = NULL;

   494

   495

    // allocate it

   496

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

   497

        ERROR("calloc");

   498

   499

    // store

   500

    evsql->ev_base = ev_base;

   501

    evsql->error_fn = error_fn;

   502

    evsql->cb_arg = cb_arg;

   503

   504

    // init

   505

    LIST_INIT(&evsql->conn_list);

   506

    TAILQ_INIT(&evsql->query_queue);

   507

   508

    // done

   509

    return evsql;

   510

   511

error:

   512

    return NULL;

   513

}

   514

   515

/*

   516

 * Start a new connection and add it to the list, it won't be ready until _evsql_evpq_connected is called

   517

 */

   518

static struct evsql_conn *_evsql_conn_new (struct evsql *evsql) {

   519

    struct evsql_conn *conn = NULL;

   520

   521

    // allocate

   522

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

   523

        ERROR("calloc");

   524

   525

    // init