1

#define _GNU_SOURCE

     2

#include <stdlib.h>

     3

#include <assert.h>

     4

#include <string.h>

     5

     6

#include "internal.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

        // zero err

   262

        err = 0;

   263

   264

        // dequeue

   265

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

   266

   267

        if (conn) {

   268

            // try and execute it

   269

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

   270

        }

   271

   272

        // free the command buf

   273

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

   274

   275

        if (err || !conn) {

   276

            if (!conn) {

   277

                // warn when dropping queries

   278

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

   279

            }

   280

   281

            // fail the query

   282

            _evsql_query_fail(evsql, query);

   283

   284

            if (conn) {

   285

                // fail the connection

   286

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

   287

   288

                _evsql_conn_fail(conn); conn = NULL;

   289

            }

   290

   291

        } else {

   292

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

   293

            break;

   294

   295

        }

   296

   297

        // handle the rest of the queue

   298

    }

   299

   300

    // ok

   301

    return;

   302

}

   303

   304

/*

   305

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

   306

 */

   307

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

   308

    struct evsql_trans *trans = arg;

   309

   310

    assert(trans != NULL);

   311

   312

    // check for errors

   313

    if (res->error)

   314

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

   315

   316

    // transaction is now ready for use

   317

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

   318

   319

    // good

   320

    return;

   321

   322

error:

   323

    _evsql_trans_fail(trans);

   324

}

   325

   326

/*

   327

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

   328

 *

   329

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

   330

 */

   331

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

   332

    char trans_sql[EVSQL_QUERY_BEGIN_BUF];

   333

    const char *isolation_level;

   334

    int ret;

   335

   336

    // determine the isolation_level to use

   337

    switch (trans->type) {

   338

        case EVSQL_TRANS_DEFAULT:

   339

            isolation_level = NULL; break;

   340

   341

        case EVSQL_TRANS_SERIALIZABLE:

   342

            isolation_level = "SERIALIZABLE"; break;

   343

   344

        case EVSQL_TRANS_REPEATABLE_READ:

   345

            isolation_level = "REPEATABLE READ"; break;

   346

   347

        case EVSQL_TRANS_READ_COMMITTED:

   348

            isolation_level = "READ COMMITTED"; break;

   349

   350

        case EVSQL_TRANS_READ_UNCOMMITTED:

   351

            isolation_level = "READ UNCOMMITTED"; break;

   352

   353

        default:

   354

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

   355

    }

   356

   357

    // build the trans_sql

   358

    if (isolation_level)

   359

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

   360

    else

   361

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

   362

   363

    // make sure it wasn't truncated

   364

    if (ret >= EVSQL_QUERY_BEGIN_BUF)

   365

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

   366

   367

    // execute the query

   368

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

   369

        ERROR("evsql_query");

   370

   371

    // success

   372

    return 0;

   373

   374

error:

   375

    // fail the transaction

   376

    _evsql_trans_fail(trans);

   377

   378

    return -1;

   379

}

   380

   381

/*

   382

 * The evpq connection was succesfully established.

   383

 */ 

   384

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

   385

    struct evsql_conn *conn = arg;

   386

   387

    if (conn->trans)

   388

        // notify the transaction

   389

        // don't care about errors

   390

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

   391

   392

    else

   393

        // pump any waiting transactionless queries

   394

        _evsql_pump(conn->evsql, conn);

   395

}

   396

   397

/*

   398

 * Got one result on this evpq connection.

   399

 */

   400

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

   401

    struct evsql_conn *conn = arg;

   402

    struct evsql_query *query = conn->query;

   403

   404

    assert(query != NULL);

   405

   406

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

   407

    if (query->result.pq) {

   408

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

   409

   410

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

   411

    }

   412

   413

    // remember the result

   414

    query->result.pq = result;

   415

}

   416

   417

/*

   418

 * No more results for this query.

   419

 */

   420

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

   421

    struct evsql_conn *conn = arg;

   422

    struct evsql_query *query = conn->query;

   423

    struct evsql_result res; ZINIT(res);

   424

   425

    assert(query != NULL);

   426

   427

    // set up the result_info

   428

    res.evsql = conn->evsql;

   429

    res.result = query->result;

   430

   431

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

   432

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

   433

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

   434

   435

        res.error = 1;

   436

   437

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

   438

        // the query failed with some error

   439

        res.error = 1;

   440

   441

    } else {

   442

        // the query succeeded \o/

   443

        res.error = 0;

   444

   445

    }

   446

   447

    // de-associate the query from the connection

   448

    conn->query = NULL;

   449

   450

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

   451

    if (conn->trans) {

   452

        // we can deassign the trans's query

   453

        conn->trans->query = NULL;

   454

   455

        // was an abort?

   456

        if (!query->cb_fn)

   457

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

   458

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

   459

   460

        // then hand the query to the user

   461

        _evsql_query_done(query, &res);

   462

   463

    } else {

   464

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

   465

        _evsql_query_done(query, &res);

   466

   467

        // pump the next one

   468

        _evsql_pump(conn->evsql, conn);

   469

    }

   470

}

   471

   472

/*

   473

 * The connection failed.

   474

 */

   475

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

   476

    struct evsql_conn *conn = arg;

   477

   478

    // just fail the conn

   479

    _evsql_conn_fail(conn);

   480

}

   481

   482

/*

   483

 * Our evpq behaviour

   484

 */

   485

static struct evpq_callback_info _evsql_evpq_cb_info = {

   486

    .fn_connected       = _evsql_evpq_connected,

   487

    .fn_result          = _evsql_evpq_result,

   488

    .fn_done            = _evsql_evpq_done,

   489

    .fn_failure         = _evsql_evpq_failure,

   490

};

   491

   492

/*

   493

 * Allocate the generic evsql context.

   494

 */

   495

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

   496

    struct evsql *evsql = NULL;

   497

   498

    // allocate it

   499

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

   500

        ERROR("calloc");

   501

   502

    // store

   503

    evsql->ev_base = ev_base;

   504

    evsql->error_fn = error_fn;

   505

    evsql->cb_arg = cb_arg;

   506

   507

    // init

   508

    LIST_INIT(&evsql->conn_list);

   509

    TAILQ_INIT(&evsql->query_queue);

   510

   511

    // done

   512

    return evsql;

   513

   514

error:

   515

    return NULL;

   516

}

   517

   518

/*

   519

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

   520

 */

   521

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

   522

    struct evsql_conn *conn = NULL;

   523

   524

    // allocate