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

    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

/*

    67

 * Free the query and related resources, doesn't trigger any callbacks or remove from any queues.

    68

 *

    69

 * The command should already be taken care of (NULL).

    70

 */

    71

static void _evsql_query_free (struct evsql_query *query) {

    72

    if (!query)

    73

        return;

    74

    75

    assert(query->command == NULL);

    76

    77

    // free params if present

    78

    free(query->params.types);

    79

    free(query->params.values);

    80

    free(query->params.lengths);

    81

    free(query->params.formats);

    82

    83

    // free the query itself

    84

    free(query);

    85

}

    86

    87

/*

    88

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

    89

 *

    90

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

    91

 */

    92

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

    93

    if (res) {

    94

        if (query->cb_fn)

    95

            // call the callback

    96

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

    97

        else

    98

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

    99

    }

   100

   101

    // free

   102

    _evsql_query_free(query);

   103

}

   104

   105

/*

   106

 * XXX:

   107

 * /

   108

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

   109

    struct evsql_query *query;

   110

   111

    // clear the queue

   112

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

   113

        _evsql_query_done(query, res);

   114

   115

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

   116

    }

   117

   118

    // free

   119

    free(evsql);

   120

}

   121

*/

   122

   123

/*

   124

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

   125

 */

   126

static void _evsql_trans_free (struct evsql_trans *trans) {

   127

    // ensure we don't leak anything

   128

    assert(trans->query == NULL);

   129

    assert(trans->conn == NULL);

   130

   131

    // free

   132

    free(trans);

   133

}

   134

   135

/*

   136

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

   137

 *

   138

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

   139

 */

   140

static void _evsql_conn_release (struct evsql_conn *conn) {

   141

    // ensure we don't leak anything

   142

    assert(conn->trans == NULL);

   143

    assert(conn->query == NULL);

   144

   145

    // release the engine

   146

    switch (conn->evsql->type) {

   147

        case EVSQL_EVPQ:

   148

            evpq_release(conn->engine.evpq);

   149

            break;

   150

   151

        default:

   152

            FATAL("evsql->type");

   153

    }

   154

   155

    // remove from list

   156

    LIST_REMOVE(conn, entry);

   157

   158

    // catch deadlocks

   159

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

   160

   161

    // free

   162

    free(conn);

   163

}

   164

   165

/*

   166

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

   167

 *

   168

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

   169

 */

   170

static void _evsql_trans_release (struct evsql_trans *trans) {

   171

    assert(trans->query == NULL);

   172

    assert(trans->conn != NULL);

   173

   174

    // deassociate the conn

   175

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

   176

   177

    // free the trans

   178

    _evsql_trans_free(trans);

   179

}

   180

   181

/*

   182

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

   183

 *

   184

 * NOTE: Only for *TRANSACTIONLESS* queries.

   185

 */

   186

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

   187

    struct evsql_result_info res; ZINIT(res);

   188

   189

    // set up the result_info

   190

    res.evsql = evsql;

   191

    res.trans = NULL;

   192

    res.error = 1;

   193

   194

    // finish off the query

   195

    _evsql_query_done(query, &res);

   196

}

   197

   198

/*

   199

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

   200

 * conn, and then free the trans.

   201

 */ 

   202

static void _evsql_trans_fail (struct evsql_trans *trans) {

   203

    if (trans->query) {

   204

        // free the query silently

   205

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

   206

   207

        // also deassociate it from the conn!

   208

        trans->conn->query = NULL;

   209

    }

   210

   211

    // tell the user

   212

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

   213

    if (trans->error_fn)

   214

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

   215

    else

   216

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

   217

   218

    // deassociate and release the conn

   219

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

   220

   221

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

   222

    _evsql_pump(trans->evsql, NULL);

   223

   224

    // free the trans

   225

    _evsql_trans_free(trans);

   226

}

   227

   228

/*

   229

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

   230

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

   231

 */

   232

static void _evsql_conn_fail (struct evsql_conn *conn) {

   233

    if (conn->trans) {

   234

        // let transactions handle their connection failures

   235

        _evsql_trans_fail(conn->trans);

   236

   237

    } else {

   238

        if (conn->query) {

   239

            // fail the in-progress query

   240

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

   241

        }

   242

   243

        // finish off the whole connection

   244

        _evsql_conn_release(conn);

   245

    }

   246

}

   247

   248

/*

   249

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

   250

 *

   251

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

   252

 *

   253

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

   254

 *

   255

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

   256

 */

   257

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

   258

    struct evsql_query *query;

   259

    int err;

   260

   261

    // look for waiting queries

   262

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

   263

        // dequeue

   264

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

   265

   266

        if (conn) {

   267

            // try and execute it

   268

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

   269

        }

   270

   271

        // free the command buf

   272

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

   273

   274

        if (err || !conn) {

   275

            if (!conn) {

   276

                // warn when dropping queries

   277

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

   278

            }

   279

   280

            // fail the query

   281

            _evsql_query_fail(evsql, query);

   282

   283

            if (conn) {

   284

                // fail the connection

   285

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

   286

   287

                _evsql_conn_fail(conn); conn = NULL;

   288

            }

   289

   290

        } else {

   291

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

   292

            break;

   293

   294

        }

   295

   296

        // handle the rest of the queue

   297

    }

   298

   299

    // ok

   300

    return;

   301

}

   302

   303

/*

   304

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

   305

 */

   306

static void _evsql_trans_ready (const struct evsql_result_info *res, void *arg) {

   307

    (void) arg;

   308

   309

    assert(res->trans);

   310

   311

    // check for errors

   312

    if (res->error)

   313

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

   314

   315

    // transaction is now ready for use

   316

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

   317

   318

    // good

   319

    return;

   320

   321

error:

   322

    _evsql_trans_fail(res->trans);

   323

}

   324

   325

/*

   326

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

   327

 *

   328

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

   329

 */

   330

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

   331

    char trans_sql[EVSQL_QUERY_BEGIN_BUF];

   332

    const char *isolation_level;

   333

    int ret;

   334

   335

    // determine the isolation_level to use

   336

    switch (trans->type) {

   337

        case EVSQL_TRANS_DEFAULT:

   338

            isolation_level = NULL; break;

   339

   340

        case EVSQL_TRANS_SERIALIZABLE:

   341

            isolation_level = "SERIALIZABLE"; break;

   342

   343

        case EVSQL_TRANS_REPEATABLE_READ:

   344

            isolation_level = "REPEATABLE READ"; break;

   345

   346

        case EVSQL_TRANS_READ_COMMITTED:

   347

            isolation_level = "READ COMMITTED"; break;

   348

   349

        case EVSQL_TRANS_READ_UNCOMMITTED:

   350

            isolation_level = "READ UNCOMMITTED"; break;

   351

   352

        default:

   353

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

   354

    }

   355

   356

    // build the trans_sql

   357

    if (isolation_level)

   358

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

   359

    else

   360

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

   361

   362

    // make sure it wasn't truncated

   363

    if (ret >= EVSQL_QUERY_BEGIN_BUF)

   364

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

   365

   366

    // execute the query

   367

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

   368

        ERROR("evsql_query");

   369

   370

    // success

   371

    return 0;

   372

   373

error:

   374

    // fail the transaction

   375

    _evsql_trans_fail(trans);

   376

   377

    return -1;

   378

}

   379

   380

/*

   381

 * The evpq connection was succesfully established.

   382

 */ 

   383

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

   384

    struct evsql_conn *conn = arg;

   385

   386

    if (conn->trans)

   387

        // notify the transaction

   388

        // don't care about errors

   389

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

   390

   391

    else

   392

        // pump any waiting transactionless queries

   393

        _evsql_pump(conn->evsql, conn);

   394

}

   395

   396

/*

   397

 * Got one result on this evpq connection.

   398

 */

   399

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

   400

    struct evsql_conn *conn = arg;

   401

    struct evsql_query *query = conn->query;

   402

   403

    assert(query != NULL);

   404

   405

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

   406

    if (query->result.evpq) {

   407

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

   408

   409

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

   410

    }

   411

   412

    // remember the result

   413

    query->result.evpq = result;

   414

}

   415

   416

/*

   417

 * No more results for this query.

   418

 */

   419

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

   420

    struct evsql_conn *conn = arg;

   421

    struct evsql_query *query = conn->query;

   422

    struct evsql_result_info res; ZINIT(res);

   423

   424

    assert(query != NULL);

   425

   426

    // set up the result_info

   427

    res.evsql = conn->evsql;

   428

    res.trans = conn->trans;

   429

   430

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

   431

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

   432

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

   433

   434

        res.error = 1;

   435

   436

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

   437

        // the query failed with some error

   438

        res.error = 1;

   439

        res.result.pq = query->result.evpq;

   440

   441

    } else {

   442

        res.error = 0;

   443

        res.result.pq = query->result.evpq;

   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

   525

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

   526

        ERROR("calloc");

   527

   528

    // init

   529

    conn->evsql = evsql;