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

    free(query->params.item_vals);

    78

    79

    // free the query itself

    80

    free(query);

    81

}

    82

    83

/*

    84

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

    85

 *

    86

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

    87

 */

    88

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

    89

    if (res) {

    90

        if (query->cb_fn) {

    91

            // call the callback

    92

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

    93

    94

        } else {

    95

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

    96

    97

            // free the results

    98

            evsql_result_free(res);

    99

        }

   100

    }

   101

   102

    // free

   103

    _evsql_query_free(query);

   104

}

   105

   106

/*

   107

 * XXX:

   108

 * /

   109

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

   110

    struct evsql_query *query;

   111

   112

    // clear the queue

   113

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

   114

        _evsql_query_done(query, res);

   115

   116

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

   117

    }

   118

   119

    // free

   120

    free(evsql);

   121

}

   122

*/

   123

   124

/*

   125

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

   126

 */

   127

static void _evsql_trans_free (struct evsql_trans *trans) {

   128

    // ensure we don't leak anything

   129

    assert(trans->query == NULL);

   130

    assert(trans->conn == NULL);

   131

   132

    // free

   133

    free(trans);

   134

}

   135

   136

/*

   137

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

   138

 *

   139

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

   140

 */

   141

static void _evsql_conn_release (struct evsql_conn *conn) {

   142

    // ensure we don't leak anything

   143

    assert(conn->trans == NULL);

   144

    assert(conn->query == NULL);

   145

   146

    // release the engine

   147

    switch (conn->evsql->type) {

   148

        case EVSQL_EVPQ:

   149

            evpq_release(conn->engine.evpq);

   150

            break;

   151

   152

        default:

   153

            FATAL("evsql->type");

   154

    }

   155

   156

    // remove from list

   157

    LIST_REMOVE(conn, entry);

   158

   159

    // catch deadlocks

   160

    // XXX: still just assert?

   161

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

   162

   163

    // free

   164

    free(conn);

   165

}

   166

   167

/*

   168

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

   169

 *

   170

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

   171

 */

   172

static void _evsql_trans_release (struct evsql_trans *trans) {

   173

    assert(trans->query == NULL);

   174

    assert(trans->conn != NULL);

   175

   176

    // deassociate the conn

   177

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

   178

   179

    // free the trans

   180

    _evsql_trans_free(trans);

   181

}

   182

   183

/*

   184

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

   185

 *

   186

 * NOTE: Only for *TRANSACTIONLESS* queries.

   187

 */

   188

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

   189

    struct evsql_result res; ZINIT(res);

   190

   191

    // set up the result_info

   192

    res.evsql = evsql;

   193

    res.error = 1;

   194

   195

    // finish off the query

   196

    _evsql_query_done(query, &res);

   197

}

   198

   199

/*

   200

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

   201

 * conn, and then free the trans.

   202

 */ 

   203

static void _evsql_trans_fail (struct evsql_trans *trans) {

   204

    if (trans->query) {

   205

        // free the query silently

   206

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

   207

   208

        // also deassociate it from the conn!

   209

        trans->conn->query = NULL;

   210

    }

   211

   212

    // tell the user

   213

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

   214

    if (trans->error_fn)

   215

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

   216

    else

   217

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

   218

   219

    // deassociate and release the conn

   220

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

   221

   222

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

   223

    _evsql_pump(trans->evsql, NULL);

   224

   225

    // free the trans

   226

    _evsql_trans_free(trans);

   227

}

   228

   229

/*

   230

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

   231

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

   232

 */

   233

static void _evsql_conn_fail (struct evsql_conn *conn) {

   234

    if (conn->trans) {

   235

        // let transactions handle their connection failures

   236

        _evsql_trans_fail(conn->trans);

   237

   238

    } else {

   239

        if (conn->query) {

   240

            // fail the in-progress query

   241

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

   242

        }

   243

   244

        // finish off the whole connection

   245

        _evsql_conn_release(conn);

   246

    }

   247

}

   248

   249

/*

   250

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

   251

 *

   252

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

   253

 *

   254

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

   255

 *

   256

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

   257

 */

   258

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

   259

    struct evsql_query *query;

   260

    int err;

   261

   262

    // look for waiting queries

   263

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

   264

        // zero err

   265

        err = 0;

   266

   267

        // dequeue

   268

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

   269

   270

        if (conn) {

   271

            // try and execute it

   272

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

   273

        }

   274

   275

        // free the command buf

   276

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

   277

   278

        if (err || !conn) {

   279

            if (!conn) {

   280

                // warn when dropping queries

   281

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

   282

            }

   283

   284

            // fail the query

   285

            _evsql_query_fail(evsql, query);

   286

   287

            if (conn) {

   288

                // fail the connection

   289

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

   290

   291

                _evsql_conn_fail(conn); conn = NULL;

   292

            }

   293

   294

        } else {

   295

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

   296

            break;

   297

   298

        }

   299

   300

        // handle the rest of the queue

   301

    }

   302

   303

    // ok

   304

    return;

   305

}

   306

   307

/*

   308

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

   309

 */

   310

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

   311

    struct evsql_trans *trans = arg;

   312

   313

    assert(trans != NULL);

   314

   315

    // check for errors

   316

    if (res->error)

   317

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

   318

   319

    // transaction is now ready for use

   320

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

   321

   322

    // good

   323

    return;

   324

   325

error:

   326

    _evsql_trans_fail(trans);

   327

}

   328

   329

/*

   330

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

   331

 *

   332

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

   333

 */

   334

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

   335

    char trans_sql[EVSQL_QUERY_BEGIN_BUF];

   336

    const char *isolation_level;

   337

    int ret;

   338

   339

    // determine the isolation_level to use

   340

    switch (trans->type) {

   341

        case EVSQL_TRANS_DEFAULT:

   342

            isolation_level = NULL; break;

   343

   344

        case EVSQL_TRANS_SERIALIZABLE:

   345

            isolation_level = "SERIALIZABLE"; break;

   346

   347

        case EVSQL_TRANS_REPEATABLE_READ:

   348

            isolation_level = "REPEATABLE READ"; break;

   349

   350

        case EVSQL_TRANS_READ_COMMITTED:

   351

            isolation_level = "READ COMMITTED"; break;

   352

   353

        case EVSQL_TRANS_READ_UNCOMMITTED:

   354

            isolation_level = "READ UNCOMMITTED"; break;

   355

   356

        default:

   357

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

   358

    }

   359

   360

    // build the trans_sql

   361

    if (isolation_level)

   362

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

   363

    else

   364

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

   365

   366

    // make sure it wasn't truncated

   367

    if (ret >= EVSQL_QUERY_BEGIN_BUF)

   368

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

   369

   370

    // execute the query

   371

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

   372

        ERROR("evsql_query");

   373

   374

    // success

   375

    return 0;

   376

   377

error:

   378

    // fail the transaction

   379

    _evsql_trans_fail(trans);

   380

   381

    return -1;

   382

}

   383

   384

/*

   385

 * The evpq connection was succesfully established.

   386

 */ 

   387

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

   388

    struct evsql_conn *conn = arg;

   389

   390

    if (conn->trans)

   391

        // notify the transaction

   392

        // don't care about errors

   393

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

   394

   395

    else

   396

        // pump any waiting transactionless queries

   397

        _evsql_pump(conn->evsql, conn);

   398

}

   399

   400

/*

   401

 * Got one result on this evpq connection.

   402

 */

   403

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

   404

    struct evsql_conn *conn = arg;

   405

    struct evsql_query *query = conn->query;

   406

   407

    assert(query != NULL);

   408

   409

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

   410

    if (query->result.pq) {

   411

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

   412

   413

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

   414

    }

   415

   416

    // remember the result

   417

    query->result.pq = result;

   418

}

   419

   420

/*

   421

 * No more results for this query.

   422

 */

   423

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

   424

    struct evsql_conn *conn = arg;

   425

    struct evsql_query *query = conn->query;

   426

    struct evsql_result res; ZINIT(res);

   427

   428

    assert(query != NULL);

   429

   430

    // set up the result_info

   431

    res.evsql = conn->evsql;

   432

    res.result = query->result;

   433

   434

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

   435

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

   436

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

   437

   438

        res.error = 1;

   439

   440

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

   441

        // the query failed with some error

   442

        res.error = 1;

   443

   444

    } else {

   445

        // the query succeeded \o/

   446

        res.error = 0;

   447

   448

    }

   449

   450

    // de-associate the query from the connection

   451

    conn->query = NULL;

   452

   453

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

   454

    if (conn->trans) {

   455

        // we can deassign the trans's query

   456

        conn->trans->query = NULL;

   457

   458

        // was an abort?

   459

        if (!query->cb_fn)

   460

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

   461

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

   462

   463

        // then hand the query to the user

   464

        _evsql_query_done(query, &res);

   465

   466

    } else {

   467

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

   468

        _evsql_query_done(query, &res);

   469

   470

        // pump the next one

   471

        _evsql_pump(conn->evsql, conn);

   472

    }

   473

}

   474

   475

/*

   476

 * The connection failed.

   477

 */

   478

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

   479

    struct evsql_conn *conn = arg;

   480

   481

    // just fail the conn

   482

    _evsql_conn_fail(conn);

   483

}

   484

   485

/*

   486

 * Our evpq behaviour

   487

 */

   488

static struct evpq_callback_info _evsql_evpq_cb_info = {

   489

    .fn_connected       = _evsql_evpq_connected,

   490

    .fn_result          = _evsql_evpq_result,

   491

    .fn_done            = _evsql_evpq_done,

   492

    .fn_failure         = _evsql_evpq_failure,

   493

};

   494

   495

/*

   496

 * Allocate the generic evsql context.

   497

 */

   498

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

   499

    struct evsql *evsql = NULL;

   500

   501

    // allocate it

   502

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

   503

        ERROR("calloc");

   504

   505

    // store

   506

    evsql->ev_base = ev_base;

   507

    evsql->error_fn = error_fn;

   508

    evsql->cb_arg = cb_arg;

   509

   510

    // init

   511

    LIST_INIT(&evsql->conn_list);

   512

    TAILQ_INIT(&evsql->query_queue);

   513

   514

    // done

   515

    return evsql;

   516

   517

error:

   518

    return NULL;

   519

}

   520

   521

/*

   522

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

   523

 */

   524

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